diff --git a/docs/zh/05-get-started/03-package.md b/docs/zh/05-get-started/03-package.md index 846cd9e9cd..b35872e7ae 100644 --- a/docs/zh/05-get-started/03-package.md +++ b/docs/zh/05-get-started/03-package.md @@ -5,12 +5,69 @@ title: 使用安装包立即开始 import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; +import PkgListV3 from "/components/PkgListV3"; 在 Linux 系统上，TDengine 开源版本提供 deb 和 rpm 格式安装包，用户可以根据自己的运行环境选择合适的安装包。其中 deb 支持 Debian/Ubuntu 及衍生系统，rpm 支持 CentOS/RHEL/SUSE 及衍生系统。同时我们也为企业用户提供 tar.gz 格式安装包，也支持通过 `apt-get` 工具从线上进行安装。TDengine 也提供 Windows x64 平台的安装包。您也可以[用Docker立即体验](../../get-started/docker/)。如果您希望对 TDengine 贡献代码或对内部实现感兴趣，请参考我们的 [TDengine GitHub 主页](https://github.com/taosdata/TDengine) 下载源码构建和安装. ## 安装 +:::info +下载其他组件、最新 Beta 版及之前版本的安装包，请点击[发布历史页面](../../releases) +::: + + + +1. 从 [发布历史页面](../../releases) 下载获得 deb 安装包，例如 TDengine-server-3.0.0.0-Linux-x64.deb； +2. 进入到 TDengine-server-3.0.0.0-Linux-x64.deb 安装包所在目录，执行如下的安装命令： + +```bash +sudo dpkg -i TDengine-server-3.0.0.0-Linux-x64.deb +``` + + + + + +1. 从 [发布历史页面](../../releases) 下载获得 rpm 安装包，例如 TDengine-server-3.0.0.0-Linux-x64.rpm； +2. 进入到 TDengine-server-3.0.0.0-Linux-x64.rpm 安装包所在目录，执行如下的安装命令： + +```bash +sudo rpm -ivh TDengine-server-3.0.0.0-Linux-x64.rpm +``` + + + + + +1. 从列表中下载获得 tar.gz 安装包，例如 TDengine-server-3.0.0.0-Linux-x64.tar.gz； + +2. 进入到 TDengine-server-3.0.0.0-Linux-x64.tar.gz 安装包所在目录，先解压文件后，进入子目录，执行其中的 install.sh 安装脚本： + +```bash +tar -zxvf TDengine-server-3.0.0.0-Linux-x64.tar.gz +``` + +解压后进入相应路径，执行 + +```bash +sudo ./install.sh +``` + +:::info +install.sh 安装脚本在执行过程中，会通过命令行交互界面询问一些配置信息。如果希望采取无交互安装方式，那么可以用 -e no 参数来执行 install.sh 脚本。运行 `./install.sh -h` 指令可以查看所有参数的详细说明信息。 +::: + + + + + + + +1. 从 [发布历史页面](../../releases) 下载获得 exe 安装程序，例如 TDengine-server-3.0.0.0-Windows-x64.exe； +2. 运行 TDengine-server-3.0.0.0-Windows-x64.exe 来安装 TDengine。 + + 可以使用 apt-get 工具从官方仓库安装。 @@ -39,56 +96,6 @@ sudo apt-get install tdengine :::tip apt-get 方式只适用于 Debian 或 Ubuntu 系统 :::: - - - -1. 从 [发布历史页面](../../releases) 下载获得 deb 安装包，例如 TDengine-server-3.0.0.0-Linux-x64.deb； -2. 进入到 TDengine-server-3.0.0.0-Linux-x64.deb 安装包所在目录，执行如下的安装命令： - -```bash -sudo dpkg -i TDengine-server-3.0.0.0-Linux-x64.deb -``` - - - - - -1. 从 [发布历史页面](../../releases) 下载获得 rpm 安装包，例如 TDengine-server-3.0.0.0-Linux-x64.rpm； -2. 进入到 TDengine-server-3.0.0.0-Linux-x64.rpm 安装包所在目录，执行如下的安装命令： - -```bash -sudo rpm -ivh TDengine-server-3.0.0.0-Linux-x64.rpm -``` - - - - - -1. 从 [发布历史页面](../../releases) 下载获得 tar.gz 安装包，例如 TDengine-server-3.0.0.0-Linux-x64.tar.gz； -2. 进入到 TDengine-server-3.0.0.0-Linux-x64.tar.gz 安装包所在目录，先解压文件后，进入子目录，执行其中的 install.sh 安装脚本： - -```bash -tar -zxvf TDengine-server-3.0.0.0-Linux-x64.tar.gz -``` - -解压后进入相应路径，执行 - -```bash -sudo ./install.sh -``` - -:::info -install.sh 安装脚本在执行过程中，会通过命令行交互界面询问一些配置信息。如果希望采取无交互安装方式，那么可以用 -e no 参数来执行 install.sh 脚本。运行 `./install.sh -h` 指令可以查看所有参数的详细说明信息。 - -::: - - - - - -1. 从 [发布历史页面](../../releases) 下载获得 exe 安装程序，例如 TDengine-server-3.0.0.0-Windows-x64.exe； -2. 运行 TDengine-server-3.0.0.0-Windows-x64.exe 来安装 TDengine。 - diff --git a/docs/zh/07-develop/07-tmq.mdx b/docs/zh/07-develop/07-tmq.mdx index 957a8bbdda..985b6a6ff2 100644 --- a/docs/zh/07-develop/07-tmq.mdx +++ b/docs/zh/07-develop/07-tmq.mdx @@ -88,6 +88,7 @@ void close() throws SQLException; ``` + ```C# @@ -110,8 +111,26 @@ void Unsubscribe() void Commit(ConsumeResult consumerResult) void Close() -``` + + + +```go +func NewConsumer(conf *Config) (*Consumer, error) + +func (c *Consumer) Close() error + +func (c *Consumer) Commit(ctx context.Context, message unsafe.Pointer) error + +func (c *Consumer) FreeMessage(message unsafe.Pointer) + +func (c *Consumer) Poll(timeout time.Duration) (*Result, error) + +func (c *Consumer) Subscribe(topics []string) error + +func (c *Consumer) Unsubscribe() error + +``` @@ -255,6 +274,7 @@ public class MetersDeserializer extends ReferenceDeserializer { ``` + ```C# @@ -276,6 +296,55 @@ var cfg = new ConsumerConfig }; var consumer = new ConsumerBuilder(cfg).Build(); + + + + +```go +config := tmq.NewConfig() +defer config.Destroy() +err = config.SetGroupID("test") +if err != nil { + panic(err) +} +err = config.SetAutoOffsetReset("earliest") +if err != nil { + panic(err) +} +err = config.SetConnectIP("127.0.0.1") +if err != nil { + panic(err) +} +err = config.SetConnectUser("root") +if err != nil { + panic(err) +} +err = config.SetConnectPass("taosdata") +if err != nil { + panic(err) +} +err = config.SetConnectPort("6030") +if err != nil { + panic(err) +} +err = config.SetMsgWithTableName(true) +if err != nil { + panic(err) +} +err = config.EnableHeartBeat() +if err != nil { + panic(err) +} +err = config.EnableAutoCommit(func(result *wrapper.TMQCommitCallbackResult) { + if result.ErrCode != 0 { + errStr := wrapper.TMQErr2Str(result.ErrCode) + err := errors.NewError(int(result.ErrCode), errStr) + panic(err) + } +}) +if err != nil { + panic(err) +} ``` @@ -309,6 +378,20 @@ topics.add("tmq_topic"); consumer.subscribe(topics); ``` + + + +```go +consumer, err := tmq.NewConsumer(config) +if err != nil { + panic(err) +} +err = consumer.Subscribe([]string{"example_tmq_topic"}) +if err != nil { + panic(err) +} +``` + @@ -354,6 +437,21 @@ while(running){ } ``` + + + +```go +for { + result, err := consumer.Poll(time.Second) + if err != nil { + panic(err) + } + fmt.Println(result) + consumer.Commit(context.Background(), result.Message) + consumer.FreeMessage(result.Message) +} +``` + @@ -410,6 +508,13 @@ consumer.Unsubscribe(); // 关闭消费 consumer.Close(); + + + + +```go +consumer.Close() + ``` diff --git a/docs/zh/12-taos-sql/29-changes.md b/docs/zh/12-taos-sql/29-changes.md index 896f1ad5de..d653c59a5c 100644 --- a/docs/zh/12-taos-sql/29-changes.md +++ b/docs/zh/12-taos-sql/29-changes.md @@ -6,7 +6,7 @@ description: "TDengine 3.0 版本的语法变更说明" ## SQL 基本元素变更 -| # | **元素** | ****
差异性
**** | **说明** | +| # | **元素** | **
差异性
** | **说明** | | - | :------- | :-------- | :------- | | 1 | VARCHAR | 新增 | BINARY类型的别名。 | 2 | TIMESTAMP字面量 | 新增 | 新增支持 TIMESTAMP 'timestamp format' 语法。 @@ -22,7 +22,7 @@ description: "TDengine 3.0 版本的语法变更说明" 在 TDengine 中，普通表的数据模型中可使用以下数据类型。 -| # | **语句** | **
差异性
** | **说明** | +| # | **语句** | **
差异性
** | **说明** | | - | :------- | :-------- | :------- | | 1 | ALTER ACCOUNT | 废除 | 2.x中为企业版功能，3.0不再支持。语法暂时保留了，执行报“This statement is no longer supported”错误。 | 2 | ALTER ALL DNODES | 新增 | 修改所有DNODE的参数。 @@ -80,7 +80,7 @@ description: "TDengine 3.0 版本的语法变更说明" ## SQL 函数变更 -| # | **函数** | **
差异性
** | **说明** | +| # | **函数** | **
差异性
** | **说明** | | - | :------- | :-------- | :------- | | 1 | TWA | 增强 | 可以直接用于超级表了。没有PARTITION BY时，超级表的数据会被合并成一条时间线。 | 2 | IRATE | 增强 | 可以直接用于超级表了。没有PARTITION BY时，超级表的数据会被合并成一条时间线。 diff --git a/docs/zh/14-reference/03-connector/_linux_install.mdx b/docs/zh/14-reference/03-connector/_linux_install.mdx index eb7f683288..a4667caec9 100644 --- a/docs/zh/14-reference/03-connector/_linux_install.mdx +++ b/docs/zh/14-reference/03-connector/_linux_install.mdx @@ -1,4 +1,4 @@ -import PkgList from "/components/PkgList"; +import PkgListV3 from "/components/PkgListV3"; 1. 下载客户端安装包 diff --git a/docs/zh/14-reference/03-connector/_windows_install.mdx b/docs/zh/14-reference/03-connector/_windows_install.mdx index 755f96b2d7..10cf37a7b4 100644 --- a/docs/zh/14-reference/03-connector/_windows_install.mdx +++ b/docs/zh/14-reference/03-connector/_windows_install.mdx @@ -1,8 +1,8 @@ -import PkgList from "/components/PkgList"; +import PkgListV3 from "/components/PkgListV3"; 1. 下载客户端安装包 - + [所有下载](https://www.taosdata.com/cn/all-downloads/) diff --git a/docs/zh/14-reference/03-connector/java.mdx b/docs/zh/14-reference/03-connector/java.mdx index c0b83e7d4a..6a78902b1e 100644 --- a/docs/zh/14-reference/03-connector/java.mdx +++ b/docs/zh/14-reference/03-connector/java.mdx @@ -93,12 +93,12 @@ Maven 项目中，在 pom.xml 中添加以下依赖： 可以通过下载 TDengine 的源码，自己编译最新版本的 Java connector ```shell -git clone https://github.com/taosdata/taos-connector-jdbc.git --branch 2.0 +git clone https://github.com/taosdata/taos-connector-jdbc.git cd taos-connector-jdbc mvn clean install -Dmaven.test.skip=true ``` -编译后，在 target 目录下会产生 taos-jdbcdriver-2.0.XX-dist.jar 的 jar 包，并自动将编译的 jar 文件放在本地的 Maven 仓库中。 +编译后，在 target 目录下会产生 taos-jdbcdriver-3.0.*-dist.jar 的 jar 包，并自动将编译的 jar 文件放在本地的 Maven 仓库中。 @@ -198,7 +198,7 @@ url 中的配置参数如下： - user：登录 TDengine 用户名，默认值 'root'。 - password：用户登录密码，默认值 'taosdata'。 -- batchfetch: true：在执行查询时批量拉取结果集；false：逐行拉取结果集。默认值为：false。逐行拉取结果集使用 HTTP 方式进行数据传输。从 taos-jdbcdriver-2.0.38 开始，JDBC REST 连接增加批量拉取数据功能。taos-jdbcdriver 与 TDengine 之间通过 WebSocket 连接进行数据传输。相较于 HTTP，WebSocket 可以使 JDBC REST 连接支持大数据量查询，并提升查询性能。 +- batchfetch: true：在执行查询时批量拉取结果集；false：逐行拉取结果集。默认值为：false。逐行拉取结果集使用 HTTP 方式进行数据传输。JDBC REST 连接支持批量拉取数据功能。taos-jdbcdriver 与 TDengine 之间通过 WebSocket 连接进行数据传输。相较于 HTTP，WebSocket 可以使 JDBC REST 连接支持大数据量查询，并提升查询性能。 - charset: 当开启批量拉取数据时，指定解析字符串数据的字符集。 - batchErrorIgnore：true：在执行 Statement 的 executeBatch 时，如果中间有一条 SQL 执行失败，继续执行下面的 SQL 了。false：不再执行失败 SQL 后的任何语句。默认值为：false。 - httpConnectTimeout: 连接超时时间，单位 ms， 默认值为 5000。 @@ -216,7 +216,7 @@ url 中的配置参数如下： INSERT INTO test.t1 USING test.weather (ts, temperature) TAGS('California.SanFrancisco') VALUES(now, 24.6); ``` -- 从 taos-jdbcdriver-2.0.36 开始，如果在 url 中指定了 dbname，那么，JDBC REST 连接会默认使用/rest/sql/dbname 作为 restful 请求的 url，在 SQL 中不需要指定 dbname。例如：url 为 jdbc:TAOS-RS://127.0.0.1:6041/test，那么，可以执行 sql：insert into t1 using weather(ts, temperature) tags('California.SanFrancisco') values(now, 24.6); +- 如果在 url 中指定了 dbname，那么，JDBC REST 连接会默认使用/rest/sql/dbname 作为 restful 请求的 url，在 SQL 中不需要指定 dbname。例如：url 为 jdbc:TAOS-RS://127.0.0.1:6041/test，那么，可以执行 sql：insert into t1 using weather(ts, temperature) tags('California.SanFrancisco') values(now, 24.6); ::: @@ -230,7 +230,7 @@ INSERT INTO test.t1 USING test.weather (ts, temperature) TAGS('California.SanFra **注意**： - 应用中设置的 client parameter 为进程级别的，即如果要更新 client 的参数，需要重启应用。这是因为 client parameter 是全局参数，仅在应用程序的第一次设置生效。 -- 以下示例代码基于 taos-jdbcdriver-2.0.36。 +- 以下示例代码基于 taos-jdbcdriver-3.0.0。 ```java public Connection getConn() throws Exception{ @@ -367,7 +367,7 @@ TDengine 的 JDBC 原生连接实现大幅改进了参数绑定方式对数据 **注意**： - JDBC REST 连接目前不支持参数绑定 -- 以下示例代码基于 taos-jdbcdriver-2.0.36 +- 以下示例代码基于 taos-jdbcdriver-3.0.0 - binary 类型数据需要调用 setString 方法，nchar 类型数据需要调用 setNString 方法 - setString 和 setNString 都要求用户在 size 参数里声明表定义中对应列的列宽 @@ -635,7 +635,7 @@ TDengine 支持无模式写入功能。无模式写入兼容 InfluxDB 的 行协 **注意**： - JDBC REST 连接目前不支持无模式写入 -- 以下示例代码基于 taos-jdbcdriver-2.0.36 +- 以下示例代码基于 taos-jdbcdriver-3.0.0 ```java public class SchemalessInsertTest { @@ -666,7 +666,7 @@ public class SchemalessInsertTest { } ``` -### 订阅 +### 数据订阅 TDengine Java 连接器支持订阅功能，应用 API 如下： @@ -717,9 +717,14 @@ while(true) { #### 关闭订阅 ```java +// 取消订阅 +consumer.unsubscribe(); +// 关闭消费 consumer.close() ``` +详情请参考：[数据订阅](../../../develop/tmq) + ### 使用示例如下： ```java @@ -734,7 +739,7 @@ public abstract class ConsumerLoop { config.setProperty("msg.with.table.name", "true"); config.setProperty("enable.auto.commit", "true"); config.setProperty("group.id", "group1"); - config.setProperty("value.deserializer", "com.taosdata.jdbc.tmq.ConsumerTest.ResultDeserializer"); + config.setProperty("value.deserializer", "com.taosdata.jdbc.tmq.ConsumerTest.ConsumerLoop$ResultDeserializer"); this.consumer = new TaosConsumer<>(config); this.topics = Collections.singletonList("topic_speed"); @@ -754,8 +759,9 @@ public abstract class ConsumerLoop { process(record); } } + consumer.unsubscribe(); } finally { - consumer.close(); + consumer.close(); shutdownLatch.countDown(); } } @@ -875,6 +881,7 @@ public static void main(String[] args) throws Exception { | taos-jdbcdriver 版本 | 主要变化 | | :------------------: | :----------------------------: | +| 3.0.0 | 支持 TDengine 3.0 | | 2.0.39 - 2.0.40 | 增加 REST 连接/请求 超时设置 | | 2.0.38 | JDBC REST 连接增加批量拉取功能 | | 2.0.37 | 增加对 json tag 支持 | diff --git a/include/libs/sync/sync.h b/include/libs/sync/sync.h index aa563343f8..6d8895eb96 100644 --- a/include/libs/sync/sync.h +++ b/include/libs/sync/sync.h @@ -30,6 +30,7 @@ extern bool gRaftDetailLog; #define SYNC_SPEED_UP_HB_TIMER 400 #define SYNC_SPEED_UP_AFTER_MS (1000 * 20) #define SYNC_SLOW_DOWN_RANGE 100 +#define SYNC_MAX_READ_RANGE 10 #define SYNC_MAX_BATCH_SIZE 1 #define SYNC_INDEX_BEGIN 0 @@ -210,9 +211,12 @@ void syncStop(int64_t rid); int32_t syncSetStandby(int64_t rid); ESyncState syncGetMyRole(int64_t rid); bool syncIsReady(int64_t rid); +bool syncIsReadyForRead(int64_t rid); const char* syncGetMyRoleStr(int64_t rid); bool syncRestoreFinish(int64_t rid); SyncTerm syncGetMyTerm(int64_t rid); +SyncIndex syncGetLastIndex(int64_t rid); +SyncIndex syncGetCommitIndex(int64_t rid); SyncGroupId syncGetVgId(int64_t rid); void syncGetEpSet(int64_t rid, SEpSet* pEpSet); void syncGetRetryEpSet(int64_t rid, SEpSet* pEpSet); diff --git a/source/dnode/vnode/src/inc/vnd.h b/source/dnode/vnode/src/inc/vnd.h index 77b18b8c02..5164e22474 100644 --- a/source/dnode/vnode/src/inc/vnd.h +++ b/source/dnode/vnode/src/inc/vnd.h @@ -98,6 +98,7 @@ void vnodeSyncStart(SVnode* pVnode); void vnodeSyncClose(SVnode* pVnode); void vnodeRedirectRpcMsg(SVnode* pVnode, SRpcMsg* pMsg); bool vnodeIsLeader(SVnode* pVnode); +bool vnodeIsReadyForRead(SVnode* pVnode); bool vnodeIsRoleLeader(SVnode* pVnode); #ifdef __cplusplus diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index ee75e39c5f..43c9b4c09f 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -283,7 +283,7 @@ int32_t vnodePreprocessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) { int32_t vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) { vTrace("message in vnode query queue is processing"); - if ((pMsg->msgType == TDMT_SCH_QUERY) && !vnodeIsLeader(pVnode)) { + if ((pMsg->msgType == TDMT_SCH_QUERY) && !vnodeIsReadyForRead(pVnode)) { vnodeRedirectRpcMsg(pVnode, pMsg); return 0; } @@ -307,7 +307,7 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) { vTrace("vgId:%d, msg:%p in fetch queue is processing", pVnode->config.vgId, pMsg); if ((pMsg->msgType == TDMT_SCH_FETCH || pMsg->msgType == TDMT_VND_TABLE_META || pMsg->msgType == TDMT_VND_TABLE_CFG || pMsg->msgType == TDMT_VND_BATCH_META) && - !vnodeIsLeader(pVnode)) { + !vnodeIsReadyForRead(pVnode)) { vnodeRedirectRpcMsg(pVnode, pMsg); return 0; } diff --git a/source/dnode/vnode/src/vnd/vnodeSync.c b/source/dnode/vnode/src/vnd/vnodeSync.c index b695352c3f..65d4e9aaf1 100644 --- a/source/dnode/vnode/src/vnd/vnodeSync.c +++ b/source/dnode/vnode/src/vnd/vnodeSync.c @@ -781,3 +781,17 @@ bool vnodeIsLeader(SVnode *pVnode) { return true; } + +bool vnodeIsReadyForRead(SVnode *pVnode) { + if (syncIsReady(pVnode->sync)) { + return true; + } + + if (syncIsReadyForRead(pVnode->sync)) { + return true; + } + + vDebug("vgId:%d, vnode not ready for read, state:%s, last:%ld, cmt:%ld", pVnode->config.vgId, + syncGetMyRoleStr(pVnode->sync), syncGetLastIndex(pVnode->sync), syncGetCommitIndex(pVnode->sync)); + return false; +} diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c index ef985a3894..1c7446ad6f 100644 --- a/source/libs/parser/src/parTranslater.c +++ b/source/libs/parser/src/parTranslater.c @@ -1399,7 +1399,7 @@ static int32_t translateTimelineFunc(STranslateContext* pCxt, SFunctionNode* pFu "%s function must be used in select statements", pFunc->functionName); } SSelectStmt* pSelect = (SSelectStmt*)pCxt->pCurrStmt; - if (QUERY_NODE_TEMP_TABLE == nodeType(pSelect->pFromTable) && + if (NULL != pSelect->pFromTable && QUERY_NODE_TEMP_TABLE == nodeType(pSelect->pFromTable) && !isTimeLineQuery(((STempTableNode*)pSelect->pFromTable)->pSubquery)) { return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_NOT_ALLOWED_FUNC, "%s function requires valid time series input", pFunc->functionName); @@ -2037,16 +2037,13 @@ static int32_t setVnodeSysTableVgroupList(STranslateContext* pCxt, SName* pName, code = getDBVgInfoImpl(pCxt, pName, &vgroupList); } - if (TSDB_CODE_SUCCESS == code && - 0 == strcmp(pRealTable->table.dbName, TSDB_INFORMATION_SCHEMA_DB) && - 0 == strcmp(pRealTable->table.tableName, TSDB_INS_TABLE_TAGS) && - isSelectStmt(pCxt->pCurrStmt) && + if (TSDB_CODE_SUCCESS == code && 0 == strcmp(pRealTable->table.dbName, TSDB_INFORMATION_SCHEMA_DB) && + 0 == strcmp(pRealTable->table.tableName, TSDB_INS_TABLE_TAGS) && isSelectStmt(pCxt->pCurrStmt) && 0 == taosArrayGetSize(vgroupList)) { ((SSelectStmt*)pCxt->pCurrStmt)->isEmptyResult = true; } - if (TSDB_CODE_SUCCESS == code && - 0 == strcmp(pRealTable->table.dbName, TSDB_INFORMATION_SCHEMA_DB) && + if (TSDB_CODE_SUCCESS == code && 0 == strcmp(pRealTable->table.dbName, TSDB_INFORMATION_SCHEMA_DB) && 0 == strcmp(pRealTable->table.tableName, TSDB_INS_TABLE_TABLES)) { code = addMnodeToVgroupList(&pCxt->pParseCxt->mgmtEpSet, &vgroupList); } diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index c7784cd62e..1991560d42 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -392,6 +392,29 @@ bool syncIsReady(int64_t rid) { return b; } +bool syncIsReadyForRead(int64_t rid) { + SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); + if (pSyncNode == NULL) { + return false; + } + ASSERT(rid == pSyncNode->rid); + + // TODO: last not noop? + SyncIndex lastIndex = syncNodeGetLastIndex(pSyncNode); + bool b = (pSyncNode->state == TAOS_SYNC_STATE_LEADER) && (pSyncNode->commitIndex >= lastIndex - SYNC_MAX_READ_RANGE); + taosReleaseRef(tsNodeRefId, pSyncNode->rid); + + // if false, set error code + if (false == b) { + if (pSyncNode->state != TAOS_SYNC_STATE_LEADER) { + terrno = TSDB_CODE_SYN_NOT_LEADER; + } else { + terrno = TSDB_CODE_APP_NOT_READY; + } + } + return b; +} + bool syncIsRestoreFinish(int64_t rid) { SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); if (pSyncNode == NULL) { @@ -519,6 +542,30 @@ SyncTerm syncGetMyTerm(int64_t rid) { return term; } +SyncIndex syncGetLastIndex(int64_t rid) { + SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); + if (pSyncNode == NULL) { + return SYNC_INDEX_INVALID; + } + ASSERT(rid == pSyncNode->rid); + SyncIndex lastIndex = syncNodeGetLastIndex(pSyncNode); + + taosReleaseRef(tsNodeRefId, pSyncNode->rid); + return lastIndex; +} + +SyncIndex syncGetCommitIndex(int64_t rid) { + SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); + if (pSyncNode == NULL) { + return SYNC_INDEX_INVALID; + } + ASSERT(rid == pSyncNode->rid); + SyncIndex cmtIndex = pSyncNode->commitIndex; + + taosReleaseRef(tsNodeRefId, pSyncNode->rid); + return cmtIndex; +} + SyncGroupId syncGetVgId(int64_t rid) { SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); if (pSyncNode == NULL) { @@ -828,6 +875,15 @@ int32_t syncNodePropose(SSyncNode* pSyncNode, SRpcMsg* pMsg, bool isWeak) { pSyncNode->changing = true; } + // not restored, vnode enable + if (!pSyncNode->restoreFinish && pSyncNode->vgId != 1) { + ret = -1; + terrno = TSDB_CODE_SYN_PROPOSE_NOT_READY; + sError("vgId:%d, failed to sync propose since not ready, type:%s, last:%ld, cmt:%ld", pSyncNode->vgId, + TMSG_INFO(pMsg->msgType), syncNodeGetLastIndex(pSyncNode), pSyncNode->commitIndex); + goto _END; + } + SRespStub stub; stub.createTime = taosGetTimestampMs(); stub.rpcMsg = *pMsg;