Merge branch 'refact/tsdb_last' of github.com:taosdata/tdengine into refact/tsdb_last

cmake/rust-bindings_CMakeLists.txt.in

-
-# rust-bindings
-ExternalProject_Add(rust-bindings
-        GIT_REPOSITORY https://github.com/songtianyi/tdengine-rust-bindings.git
-        GIT_TAG 7ed7a97
-        SOURCE_DIR "${TD_SOURCE_DIR}/examples/rust"
-        BINARY_DIR "${TD_SOURCE_DIR}/examples/rust"
-        CONFIGURE_COMMAND "" 
-        BUILD_COMMAND ""  
-        INSTALL_COMMAND ""
-        TEST_COMMAND ""
-        )

contrib/CMakeLists.txt

@@ -105,11 +105,6 @@ if(${BUILD_WITH_SQLITE})
     cat("${TD_SUPPORT_DIR}/sqlite_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
 endif(${BUILD_WITH_SQLITE})
 
-# rust-bindings
-if(${RUST_BINDINGS})
-    cat("${TD_SUPPORT_DIR}/rust-bindings_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
-endif(${RUST_BINDINGS})
-
 # lucene
 if(${BUILD_WITH_LUCENE})
     cat("${TD_SUPPORT_DIR}/lucene_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})

docs/examples/c/tmq_example.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include <assert.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <time.h>
+#include "taos.h"
+
+static int  running = 1;
+static char dbName[64] = "tmqdb";
+static char stbName[64] = "stb";
+static char topicName[64] = "topicname";
+
+static int32_t msg_process(TAOS_RES* msg) {
+  char    buf[1024];
+  int32_t rows = 0;
+
+  const char* topicName = tmq_get_topic_name(msg);
+  const char* dbName = tmq_get_db_name(msg);
+  int32_t     vgroupId = tmq_get_vgroup_id(msg);
+
+  printf("topic: %s\n", topicName);
+  printf("db: %s\n", dbName);
+  printf("vgroup id: %d\n", vgroupId);
+
+  while (1) {
+    TAOS_ROW row = taos_fetch_row(msg);
+    if (row == NULL) break;
+
+    TAOS_FIELD* fields = taos_fetch_fields(msg);
+    int32_t     numOfFields = taos_field_count(msg);
+    int32_t*    length = taos_fetch_lengths(msg);
+    int32_t     precision = taos_result_precision(msg);
+    rows++;
+    taos_print_row(buf, row, fields, numOfFields);
+    printf("row content: %s\n", buf);
+  }
+
+  return rows;
+}
+
+static int32_t init_env() {
+  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  if (pConn == NULL) {
+    return -1;
+  }
+
+  TAOS_RES* pRes;
+  // drop database if exists
+  printf("create database\n");
+  pRes = taos_query(pConn, "drop database if exists tmqdb");
+  if (taos_errno(pRes) != 0) {
+    printf("error in drop tmqdb, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  // create database
+  pRes = taos_query(pConn, "create database tmqdb");
+  if (taos_errno(pRes) != 0) {
+    printf("error in create tmqdb, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  // create super table
+  printf("create super table\n");
+  pRes = taos_query(
+      pConn, "create table tmqdb.stb (ts timestamp, c1 int, c2 float, c3 varchar(16)) tags(t1 int, t3 varchar(16))");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create super table stb, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  // create sub tables
+  printf("create sub tables\n");
+  pRes = taos_query(pConn, "create table tmqdb.ctb0 using tmqdb.stb tags(0, 'subtable0')");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create super table ctb0, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "create table tmqdb.ctb1 using tmqdb.stb tags(1, 'subtable1')");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create super table ctb1, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "create table tmqdb.ctb2 using tmqdb.stb tags(2, 'subtable2')");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create super table ctb2, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "create table tmqdb.ctb3 using tmqdb.stb tags(3, 'subtable3')");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create super table ctb3, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  // insert data
+  printf("insert data into sub tables\n");
+  pRes = taos_query(pConn, "insert into tmqdb.ctb0 values(now, 0, 0, 'a0')(now+1s, 0, 0, 'a00')");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to insert into ctb0, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "insert into tmqdb.ctb1 values(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11')");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to insert into ctb0, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "insert into tmqdb.ctb2 values(now, 2, 2, 'a1')(now+1s, 22, 22, 'a22')");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to insert into ctb0, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "insert into tmqdb.ctb3 values(now, 3, 3, 'a1')(now+1s, 33, 33, 'a33')");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to insert into ctb0, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  taos_close(pConn);
+  return 0;
+}
+
+int32_t create_topic() {
+  printf("create topic\n");
+  TAOS_RES* pRes;
+  TAOS*     pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  if (pConn == NULL) {
+    return -1;
+  }
+
+  pRes = taos_query(pConn, "use tmqdb");
+  if (taos_errno(pRes) != 0) {
+    printf("error in use tmqdb, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "create topic topicname as select ts, c1, c2, c3, tbname from tmqdb.stb where c1 > 1");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create topic topicname, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  taos_close(pConn);
+  return 0;
+}
+
+void tmq_commit_cb_print(tmq_t* tmq, int32_t code, void* param) {
+  printf("tmq_commit_cb_print() code: %d, tmq: %p, param: %p\n", code, tmq, param);
+}
+
+tmq_t* build_consumer() {
+  tmq_conf_res_t code;
+  tmq_conf_t*    conf = tmq_conf_new();
+  code = tmq_conf_set(conf, "enable.auto.commit", "true");
+  if (TMQ_CONF_OK != code) return NULL;
+  code = tmq_conf_set(conf, "auto.commit.interval.ms", "1000");
+  if (TMQ_CONF_OK != code) return NULL;
+  code = tmq_conf_set(conf, "group.id", "cgrpName");
+  if (TMQ_CONF_OK != code) return NULL;
+  code = tmq_conf_set(conf, "client.id", "user defined name");
+  if (TMQ_CONF_OK != code) return NULL;
+  code = tmq_conf_set(conf, "td.connect.user", "root");
+  if (TMQ_CONF_OK != code) return NULL;
+  code = tmq_conf_set(conf, "td.connect.pass", "taosdata");
+  if (TMQ_CONF_OK != code) return NULL;
+  code = tmq_conf_set(conf, "auto.offset.reset", "earliest");
+  if (TMQ_CONF_OK != code) return NULL;
+  code = tmq_conf_set(conf, "experimental.snapshot.enable", "false");
+  if (TMQ_CONF_OK != code) return NULL;
+
+  tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL);
+
+  tmq_t* tmq = tmq_consumer_new(conf, NULL, 0);
+  tmq_conf_destroy(conf);
+  return tmq;
+}
+
+tmq_list_t* build_topic_list() {
+  tmq_list_t* topicList = tmq_list_new();
+  int32_t     code = tmq_list_append(topicList, "topicname");
+  if (code) {
+    return NULL;
+  }
+  return topicList;
+}
+
+void basic_consume_loop(tmq_t* tmq) {
+  int32_t totalRows = 0;
+  int32_t msgCnt = 0;
+  int32_t timeout = 5000;
+  while (running) {
+    TAOS_RES* tmqmsg = tmq_consumer_poll(tmq, timeout);
+    if (tmqmsg) {
+      msgCnt++;
+      totalRows += msg_process(tmqmsg);
+      taos_free_result(tmqmsg);
+    } else {
+      break;
+    }
+  }
+
+  fprintf(stderr, "%d msg consumed, include %d rows\n", msgCnt, totalRows);
+}
+
+int main(int argc, char* argv[]) {
+  int32_t code;
+
+  if (init_env() < 0) {
+    return -1;
+  }
+
+  if (create_topic() < 0) {
+    return -1;
+  }
+
+  tmq_t* tmq = build_consumer();
+  if (NULL == tmq) {
+    fprintf(stderr, "%% build_consumer() fail!\n");
+    return -1;
+  }
+
+  tmq_list_t* topic_list = build_topic_list();
+  if (NULL == topic_list) {
+    return -1;
+  }
+
+  if ((code = tmq_subscribe(tmq, topic_list))) {
+    fprintf(stderr, "%% Failed to tmq_subscribe(): %s\n", tmq_err2str(code));
+  }
+  tmq_list_destroy(topic_list);
+
+  basic_consume_loop(tmq);
+
+  code = tmq_consumer_close(tmq);
+  if (code) {
+    fprintf(stderr, "%% Failed to close consumer: %s\n", tmq_err2str(code));
+  } else {
+    fprintf(stderr, "%% Consumer closed\n");
+  }
+
+  return 0;
+}

docs/zh/05-get-started/03-package.md

@@ -7,12 +7,13 @@ import Tabs from "@theme/Tabs";
 import TabItem from "@theme/TabItem";
 import PkgListV3 from "/components/PkgListV3";
 
-TDengine 完整的软件包包括服务端（taosd）、用于与第三方系统对接并提供 RESTful 接口的 taosAdapter、应用驱动（taosc）、命令行程序 (CLI，taos) 和一些工具软件。目前 taosAdapter 仅在 Linux 系统上安装和运行，后续将支持 Windows、macOS 等系统。TDengine 除了提供多种语言的连接器之外，还通过 [taosAdapter](../../reference/taosadapter/) 提供 [RESTful 接口](../../reference/rest-api/)。
+您可以[用 Docker 立即体验](../../get-started/docker/) TDengine。如果您希望对 TDengine 贡献代码或对内部实现感兴趣，请参考我们的 [TDengine GitHub 主页](https://github.com/taosdata/TDengine) 下载源码构建和安装.
 
-为方便使用，标准的服务端安装包包含了 taos、taosd、taosAdapter、taosdump、taosBenchmark、TDinsight 安装脚本和示例代码；如果您只需要用到服务端程序和客户端连接的 C/C++ 语言支持，也可以仅下载 lite 版本的安装包。
+TDengine 完整的软件包包括服务端（taosd）、用于与第三方系统对接并提供 RESTful 接口的 taosAdapter、应用驱动（taosc）、命令行程序 (CLI，taos) 和一些工具软件。目前 taosAdapter 仅在 Linux 系统上安装和运行，后续将支持 Windows、macOS 等系统。TDengine 除了提供多种语言的连接器之外，还通过 [taosAdapter](../../reference/taosadapter/) 提供 [RESTful 接口](../../reference/rest-api/)。
 
-在 Linux 系统上，TDengine 开源版本提供 deb 和 rpm 格式安装包，用户可以根据自己的运行环境选择合适的安装包。其中 deb 支持 Debian/Ubuntu 及衍生系统，rpm 支持 CentOS/RHEL/SUSE 及衍生系统。同时我们也为企业用户提供 tar.gz 格式安装包，也支持通过 `apt-get` 工具从线上进行安装。TDengine 也提供 Windows x64 平台的安装包。您也可以[用 Docker 立即体验](../../get-started/docker/)。需要注意的是，rpm 和 deb 包不含 taosdump 和 TDinsight 安装脚本，这些工具需要通过安装 taosTool 包获得。如果您希望对 TDengine 贡献代码或对内部实现感兴趣，请参考我们的 [TDengine GitHub 主页](https://github.com/taosdata/TDengine) 下载源码构建和安装.
+为方便使用，标准的服务端安装包包含了 taosd、taosAdapter、taosc、taos、taosdump、taosBenchmark、TDinsight 安装脚本和示例代码；如果您只需要用到服务端程序和客户端连接的 C/C++ 语言支持，也可以仅下载 lite 版本的安装包。
 
+在 Linux 系统上，TDengine 开源版本提供 deb 和 rpm 格式安装包，用户可以根据自己的运行环境选择合适的安装包。其中 deb 支持 Debian/Ubuntu 及衍生系统，rpm 支持 CentOS/RHEL/SUSE 及衍生系统。同时我们也为企业用户提供 tar.gz 格式安装包，也支持通过 `apt-get` 工具从线上进行安装。需要注意的是，rpm 和 deb 包不含 taosdump 和 TDinsight 安装脚本，这些工具需要通过安装 taosTool 包获得。TDengine 也提供 Windows x64 平台的安装包。
 
 ## 安装
 

docs/zh/07-develop/07-tmq.mdx

@@ -724,7 +724,6 @@ consumer.close();
 
 </TabItem>
 
-
 <TabItem value="Go" label="Go">
 
 ```go
@@ -769,6 +768,7 @@ consumer.Unsubscribe();
 // 关闭消费
 consumer.Close();
 ```
+
 </TabItem>
 
 </Tabs>
@@ -809,7 +809,7 @@ SHOW SUBSCRIPTIONS;
 <Tabs defaultValue="java" groupId="lang">
 
 <TabItem label="C" value="c">
-  <CDemo>
+  <CDemo />
 </TabItem>
 
 <TabItem label="Java" value="java">

docs/zh/07-develop/_sub_c.mdx

 ```c
-{{#include docs/examples/c/tmq-example.c}}
+{{#include docs/examples/c/tmq_example.c}}
 ```

examples/rust/.gitignore

+/target
+Cargo.lock

examples/rust/Cargo.toml

+[package]
+name = "rust"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+taos = "*"
+
+[dev-dependencies]
+chrono = "0.4"
+itertools = "0.10.3"
+pretty_env_logger = "0.4.0"
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+tokio = { version = "1", features = ["full"] }
+anyhow = "1"

examples/rust/examples/bind-tags.rs

+use anyhow::Result;
+use serde::Deserialize;
+use taos::*;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let taos = TaosBuilder::from_dsn("taos://")?.build()?;
+    taos.exec_many([
+        "drop database if exists test",
+        "create database test keep 36500",
+        "use test",
+        "create table tb1 (ts timestamp, c1 bool, c2 tinyint, c3 smallint, c4 int, c5 bigint,
+            c6 tinyint unsigned, c7 smallint unsigned, c8 int unsigned, c9 bigint unsigned,
+            c10 float, c11 double, c12 varchar(100), c13 nchar(100)) tags(t1 varchar(100))",
+    ])
+    .await?;
+    let mut stmt = Stmt::init(&taos)?;
+    stmt.prepare(
+        "insert into ? using tb1 tags(?) values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+    )?;
+    stmt.set_tbname("d0")?;
+    stmt.set_tags(&[Value::VarChar("涛思".to_string())])?;
+
+    let params = vec![
+        ColumnView::from_millis_timestamp(vec![164000000000]),
+        ColumnView::from_bools(vec![true]),
+        ColumnView::from_tiny_ints(vec![i8::MAX]),
+        ColumnView::from_small_ints(vec![i16::MAX]),
+        ColumnView::from_ints(vec![i32::MAX]),
+        ColumnView::from_big_ints(vec![i64::MAX]),
+        ColumnView::from_unsigned_tiny_ints(vec![u8::MAX]),
+        ColumnView::from_unsigned_small_ints(vec![u16::MAX]),
+        ColumnView::from_unsigned_ints(vec![u32::MAX]),
+        ColumnView::from_unsigned_big_ints(vec![u64::MAX]),
+        ColumnView::from_floats(vec![f32::MAX]),
+        ColumnView::from_doubles(vec![f64::MAX]),
+        ColumnView::from_varchar(vec!["ABC"]),
+        ColumnView::from_nchar(vec!["涛思数据"]),
+    ];
+    let rows = stmt.bind(&params)?.add_batch()?.execute()?;
+    assert_eq!(rows, 1);
+
+    #[derive(Debug, Deserialize)]
+    #[allow(dead_code)]
+    struct Row {
+        ts: String,
+        c1: bool,
+        c2: i8,
+        c3: i16,
+        c4: i32,
+        c5: i64,
+        c6: u8,
+        c7: u16,
+        c8: u32,
+        c9: u64,
+        c10: Option<f32>,
+        c11: f64,
+        c12: String,
+        c13: String,
+        t1: serde_json::Value,
+    }
+
+    let rows: Vec<Row> = taos
+        .query("select * from tb1")
+        .await?
+        .deserialize()
+        .try_collect()
+        .await?;
+    let row = &rows[0];
+    dbg!(&row);
+    assert_eq!(row.c5, i64::MAX);
+    assert_eq!(row.c8, u32::MAX);
+    assert_eq!(row.c9, u64::MAX);
+    assert_eq!(row.c10.unwrap(), f32::MAX);
+    // assert_eq!(row.c11, f64::MAX);
+    assert_eq!(row.c12, "ABC");
+    assert_eq!(row.c13, "涛思数据");
+
+    Ok(())
+}

examples/rust/examples/bind.rs

+use anyhow::Result;
+use serde::Deserialize;
+use taos::*;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let taos = TaosBuilder::from_dsn("taos://")?.build()?;
+    taos.exec_many([
+        "drop database if exists test_bindable",
+        "create database test_bindable keep 36500",
+        "use test_bindable",
+        "create table tb1 (ts timestamp, c1 bool, c2 tinyint, c3 smallint, c4 int, c5 bigint,
+            c6 tinyint unsigned, c7 smallint unsigned, c8 int unsigned, c9 bigint unsigned,
+            c10 float, c11 double, c12 varchar(100), c13 nchar(100))",
+    ])
+    .await?;
+    let mut stmt = Stmt::init(&taos)?;
+    stmt.prepare("insert into tb1 values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")?;
+    let params = vec![
+        ColumnView::from_millis_timestamp(vec![0]),
+        ColumnView::from_bools(vec![true]),
+        ColumnView::from_tiny_ints(vec![i8::MAX]),
+        ColumnView::from_small_ints(vec![i16::MAX]),
+        ColumnView::from_ints(vec![i32::MAX]),
+        ColumnView::from_big_ints(vec![i64::MAX]),
+        ColumnView::from_unsigned_tiny_ints(vec![u8::MAX]),
+        ColumnView::from_unsigned_small_ints(vec![u16::MAX]),
+        ColumnView::from_unsigned_ints(vec![u32::MAX]),
+        ColumnView::from_unsigned_big_ints(vec![u64::MAX]),
+        ColumnView::from_floats(vec![f32::MAX]),
+        ColumnView::from_doubles(vec![f64::MAX]),
+        ColumnView::from_varchar(vec!["ABC"]),
+        ColumnView::from_nchar(vec!["涛思数据"]),
+    ];
+    let rows = stmt.bind(&params)?.add_batch()?.execute()?;
+    assert_eq!(rows, 1);
+
+    #[derive(Debug, Deserialize)]
+    #[allow(dead_code)]
+    struct Row {
+        ts: String,
+        c1: bool,
+        c2: i8,
+        c3: i16,
+        c4: i32,
+        c5: i64,
+        c6: u8,
+        c7: u16,
+        c8: u32,
+        c9: u64,
+        c10: Option<f32>,
+        c11: f64,
+        c12: String,
+        c13: String,
+    }
+
+    let rows: Vec<Row> = taos
+        .query("select * from tb1")
+        .await?
+        .deserialize()
+        .try_collect()
+        .await?;
+    let row = &rows[0];
+    dbg!(&row);
+    assert_eq!(row.c5, i64::MAX);
+    assert_eq!(row.c8, u32::MAX);
+    assert_eq!(row.c9, u64::MAX);
+    assert_eq!(row.c10.unwrap(), f32::MAX);
+    // assert_eq!(row.c11, f64::MAX);
+    assert_eq!(row.c12, "ABC");
+    assert_eq!(row.c13, "涛思数据");
+
+    Ok(())
+}

examples/rust/examples/query.rs

+use std::time::Duration;
+
+use chrono::{DateTime, Local};
+use taos::*;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let dsn = "taos://";
+
+    let opts = PoolBuilder::new()
+        .max_size(5000) // max connections
+        .max_lifetime(Some(Duration::from_secs(60 * 60))) // lifetime of each connection
+        .min_idle(Some(1000)) // minimal idle connections
+        .connection_timeout(Duration::from_secs(2));
+
+    let pool = TaosBuilder::from_dsn(dsn)?.with_pool_builder(opts)?;
+
+    let taos = pool.get()?;
+
+    let db = "query";
+
+    // prepare database
+    taos.exec_many([
+        format!("DROP DATABASE IF EXISTS `{db}`"),
+        format!("CREATE DATABASE `{db}`"),
+        format!("USE `{db}`"),
+    ])
+    .await?;
+
+    let inserted = taos.exec_many([
+        // create super table
+        "CREATE TABLE `meters` (`ts` TIMESTAMP, `current` FLOAT, `voltage` INT, `phase` FLOAT) TAGS (`groupid` INT, `location` BINARY(16))",
+        // create child table
+        "CREATE TABLE `d0` USING `meters` TAGS(0, 'Los Angles')",
+        // insert into child table
+        "INSERT INTO `d0` values(now - 10s, 10, 116, 0.32)",
+        // insert with NULL values
+        "INSERT INTO `d0` values(now - 8s, NULL, NULL, NULL)",
+        // insert and automatically create table with tags if not exists
+        "INSERT INTO `d1` USING `meters` TAGS(1, 'San Francisco') values(now - 9s, 10.1, 119, 0.33)",
+        // insert many records in a single sql
+        "INSERT INTO `d1` values (now-8s, 10, 120, 0.33) (now - 6s, 10, 119, 0.34) (now - 4s, 11.2, 118, 0.322)",
+    ]).await?;
+
+    assert_eq!(inserted, 6);
+    loop {
+        let count: usize = taos
+            .query_one("select count(*) from `meters`")
+            .await?
+            .unwrap_or_default();
+
+        if count >= 6 {
+            break;
+        } else {
+            println!("waiting for data");
+        }
+    }
+
+    let mut result = taos.query("select tbname, * from `meters`").await?;
+
+    for field in result.fields() {
+        println!("got field: {}", field.name());
+    }
+
+    // Query option 1, use rows stream.
+    let mut rows = result.rows();
+    let mut nrows = 0;
+    while let Some(row) = rows.try_next().await? {
+        for (col, (name, value)) in row.enumerate() {
+            println!(
+                "[{}] got value in col {} (named `{:>8}`): {}",
+                nrows, col, name, value
+            );
+        }
+        nrows += 1;
+    }
+
+    // Query options 2, use deserialization with serde.
+    #[derive(Debug, serde::Deserialize)]
+    #[allow(dead_code)]
+    struct Record {
+        tbname: String,
+        // deserialize timestamp to chrono::DateTime<Local>
+        ts: DateTime<Local>,
+        // float to f32
+        current: Option<f32>,
+        // int to i32
+        voltage: Option<i32>,
+        phase: Option<f32>,
+        groupid: i32,
+        // binary/varchar to String
+        location: String,
+    }
+
+    let records: Vec<Record> = taos
+        .query("select tbname, * from `meters`")
+        .await?
+        .deserialize()
+        .try_collect()
+        .await?;
+
+    dbg!(result.summary());
+    assert_eq!(records.len(), 6);
+    dbg!(records);
+    Ok(())
+}

examples/rust/examples/subscribe.rs

+use std::time::Duration;
+
+use chrono::{DateTime, Local};
+use taos::*;
+
+// Query options 2, use deserialization with serde.
+#[derive(Debug, serde::Deserialize)]
+#[allow(dead_code)]
+struct Record {
+    // deserialize timestamp to chrono::DateTime<Local>
+    ts: DateTime<Local>,
+    // float to f32
+    current: Option<f32>,
+    // int to i32
+    voltage: Option<i32>,
+    phase: Option<f32>,
+}
+
+async fn prepare(taos: Taos) -> anyhow::Result<()> {
+    let inserted = taos.exec_many([
+        // create child table
+        "CREATE TABLE `d0` USING `meters` TAGS(0, 'Los Angles')",
+        // insert into child table
+        "INSERT INTO `d0` values(now - 10s, 10, 116, 0.32)",
+        // insert with NULL values
+        "INSERT INTO `d0` values(now - 8s, NULL, NULL, NULL)",
+        // insert and automatically create table with tags if not exists
+        "INSERT INTO `d1` USING `meters` TAGS(1, 'San Francisco') values(now - 9s, 10.1, 119, 0.33)",
+        // insert many records in a single sql
+        "INSERT INTO `d1` values (now-8s, 10, 120, 0.33) (now - 6s, 10, 119, 0.34) (now - 4s, 11.2, 118, 0.322)",
+    ]).await?;
+    assert_eq!(inserted, 6);
+    Ok(())
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    // std::env::set_var("RUST_LOG", "debug");
+    pretty_env_logger::init();
+    let dsn = "taos://localhost:6030";
+    let builder = TaosBuilder::from_dsn(dsn)?;
+
+    let taos = builder.build()?;
+    let db = "tmq";
+
+    // prepare database
+    taos.exec_many([
+        "DROP TOPIC IF EXISTS tmq_meters".to_string(),
+        format!("DROP DATABASE IF EXISTS `{db}`"),
+        format!("CREATE DATABASE `{db}`"),
+        format!("USE `{db}`"),
+        // create super table
+        "CREATE TABLE `meters` (`ts` TIMESTAMP, `current` FLOAT, `voltage` INT, `phase` FLOAT) TAGS (`groupid` INT, `location` BINARY(16))".to_string(),
+        // create topic for subscription
+        format!("CREATE TOPIC tmq_meters with META AS DATABASE {db}")
+    ])
+    .await?;
+
+    let task = tokio::spawn(prepare(taos));
+
+    tokio::time::sleep(Duration::from_secs(1)).await;
+
+    // subscribe
+    let tmq = TmqBuilder::from_dsn("taos://localhost:6030/?group.id=test")?;
+
+    let mut consumer = tmq.build()?;
+    consumer.subscribe(["tmq_meters"]).await?;
+
+    {
+        let mut stream = consumer.stream();
+
+        while let Some((offset, message)) = stream.try_next().await? {
+            // get information from offset
+
+            // the topic
+            let topic = offset.topic();
+            // the vgroup id, like partition id in kafka.
+            let vgroup_id = offset.vgroup_id();
+            println!("* in vgroup id {vgroup_id} of topic {topic}\n");
+
+            if let Some(data) = message.into_data() {
+                while let Some(block) = data.fetch_raw_block().await? {
+                    // one block for one table, get table name if needed
+                    let name = block.table_name();
+                    let records: Vec<Record> = block.deserialize().try_collect()?;
+                    println!(
+                        "** table: {}, got {} records: {:#?}\n",
+                        name.unwrap(),
+                        records.len(),
+                        records
+                    );
+                }
+            }
+            consumer.commit(offset).await?;
+        }
+    }
+
+    consumer.unsubscribe().await;
+
+    task.await??;
+
+    Ok(())
+}

examples/rust/src/main.rs

+fn main() {
+    println!("Hello, world!");
+}

include/common/tmsg.h

@@ -2555,10 +2555,14 @@ typedef struct {
   char    topic[TSDB_TOPIC_FNAME_LEN];
   int64_t ntbUid;
   SArray* colIdList;  // SArray<int16_t>
-} SCheckAlterInfo;
+} STqCheckInfo;
 
-int32_t tEncodeSCheckAlterInfo(SEncoder* pEncoder, const SCheckAlterInfo* pInfo);
-int32_t tDecodeSCheckAlterInfo(SDecoder* pDecoder, SCheckAlterInfo* pInfo);
+int32_t tEncodeSTqCheckInfo(SEncoder* pEncoder, const STqCheckInfo* pInfo);
+int32_t tDecodeSTqCheckInfo(SDecoder* pDecoder, STqCheckInfo* pInfo);
+
+typedef struct {
+  char topic[TSDB_TOPIC_FNAME_LEN];
+} STqDelCheckInfoReq;
 
 typedef struct {
   int32_t vgId;

include/common/tmsgdef.h

@@ -188,7 +188,8 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_VND_MQ_VG_CHANGE, "vnode-mq-vg-change", SMqRebVgReq, SMqRebVgRsp)
   TD_DEF_MSG_TYPE(TDMT_VND_MQ_VG_DELETE, "vnode-mq-vg-delete", SMqVDeleteReq, SMqVDeleteRsp)
   TD_DEF_MSG_TYPE(TDMT_VND_MQ_COMMIT_OFFSET, "vnode-commit-offset", STqOffset, STqOffset)
-  TD_DEF_MSG_TYPE(TDMT_VND_CHECK_ALTER_INFO, "vnode-alter-check-info", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_ADD_CHECK_INFO, "vnode-add-check-info", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_DELETE_CHECK_INFO, "vnode-delete-check-info", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_CREATE_TOPIC, "vnode-create-topic", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_ALTER_TOPIC, "vnode-alter-topic", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_DROP_TOPIC, "vnode-drop-topic", NULL, NULL)

include/libs/stream/tstream.h

@@ -515,7 +515,7 @@ SStreamMeta* streamMetaOpen(const char* path, void* ahandle, FTaskExpand expandF
 void         streamMetaClose(SStreamMeta* streamMeta);
 
 int32_t      streamMetaAddTask(SStreamMeta* pMeta, SStreamTask* pTask);
-int32_t      streamMetaAddSerializedTask(SStreamMeta* pMeta, char* msg, int32_t msgLen);
+int32_t      streamMetaAddSerializedTask(SStreamMeta* pMeta, int64_t startVer, char* msg, int32_t msgLen);
 int32_t      streamMetaRemoveTask(SStreamMeta* pMeta, int32_t taskId);
 SStreamTask* streamMetaGetTask(SStreamMeta* pMeta, int32_t taskId);
 

include/libs/sync/sync.h

@@ -26,11 +26,14 @@ extern "C" {
 
 extern bool gRaftDetailLog;
 
-#define SYNC_RESP_TTL_MS       10000000
-#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_RESP_TTL_MS             10000000
+#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          2
+#define SYNC_MAX_PROGRESS_WAIT_MS    4000
+#define SYNC_MAX_START_TIME_RANGE_MS (1000 * 20)
+#define SYNC_MAX_RECV_TIME_RANGE_MS  1000
 
 #define SYNC_MAX_BATCH_SIZE 1
 #define SYNC_INDEX_BEGIN    0

include/libs/sync/syncTools.h

@@ -423,6 +423,7 @@ typedef struct SyncAppendEntriesReply {
   SyncTerm  privateTerm;
   bool      success;
   SyncIndex matchIndex;
+  int64_t   startTime;
 } SyncAppendEntriesReply;
 
 SyncAppendEntriesReply* syncAppendEntriesReplyBuild(int32_t vgId);

source/common/src/tmsg.c

@@ -4262,7 +4262,6 @@ int32_t tDeserializeSServerStatusRsp(void *buf, int32_t bufLen, SServerStatusRsp
   tDecoderClear(&decoder);
   return 0;
 }
-
 int32_t tEncodeSMqOffset(SEncoder *encoder, const SMqOffset *pOffset) {
   if (tEncodeI32(encoder, pOffset->vgId) < 0) return -1;
   if (tEncodeI64(encoder, pOffset->offset) < 0) return -1;
@@ -4300,7 +4299,6 @@ int32_t tDecodeSMqCMCommitOffsetReq(SDecoder *decoder, SMqCMCommitOffsetReq *pRe
   tEndDecode(decoder);
   return 0;
 }
-
 int32_t tSerializeSExplainRsp(void *buf, int32_t bufLen, SExplainRsp *pRsp) {
   SEncoder encoder = {0};
   tEncoderInit(&encoder, buf, bufLen);
@@ -5590,7 +5588,6 @@ int32_t tDecodeSTqOffsetVal(SDecoder *pDecoder, STqOffsetVal *pOffsetVal) {
   return 0;
 }
 
-#if 1
 int32_t tFormatOffset(char *buf, int32_t maxLen, const STqOffsetVal *pVal) {
   if (pVal->type == TMQ_OFFSET__RESET_NONE) {
     snprintf(buf, maxLen, "offset(reset to none)");
@@ -5609,7 +5606,6 @@ int32_t tFormatOffset(char *buf, int32_t maxLen, const STqOffsetVal *pVal) {
   }
   return 0;
 }
-#endif
 
 bool tOffsetEqual(const STqOffsetVal *pLeft, const STqOffsetVal *pRight) {
   if (pLeft->type == pRight->type) {
@@ -5643,7 +5639,7 @@ int32_t tDecodeSTqOffset(SDecoder *pDecoder, STqOffset *pOffset) {
   return 0;
 }
 
-int32_t tEncodeSCheckAlterInfo(SEncoder *pEncoder, const SCheckAlterInfo *pInfo) {
+int32_t tEncodeSTqCheckInfo(SEncoder *pEncoder, const STqCheckInfo *pInfo) {
   if (tEncodeCStr(pEncoder, pInfo->topic) < 0) return -1;
   if (tEncodeI64(pEncoder, pInfo->ntbUid) < 0) return -1;
   int32_t sz = taosArrayGetSize(pInfo->colIdList);
@@ -5655,7 +5651,7 @@ int32_t tEncodeSCheckAlterInfo(SEncoder *pEncoder, const SCheckAlterInfo *pInfo)
   return pEncoder->pos;
 }
 
-int32_t tDecodeSCheckAlterInfo(SDecoder *pDecoder, SCheckAlterInfo *pInfo) {
+int32_t tDecodeSTqCheckInfo(SDecoder *pDecoder, STqCheckInfo *pInfo) {
   if (tDecodeCStrTo(pDecoder, pInfo->topic) < 0) return -1;
   if (tDecodeI64(pDecoder, &pInfo->ntbUid) < 0) return -1;
   int32_t sz;

source/dnode/mgmt/mgmt_mnode/src/mmHandle.c

@@ -225,7 +225,8 @@ SArray *mmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_SMA_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_CHANGE_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_DELETE_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_CHECK_ALTER_INFO_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_ADD_CHECK_INFO_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_DELETE_CHECK_INFO_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_SCH_DROP_TASK, mmPutMsgToFetchQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DEPLOY_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DROP_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;

source/dnode/mgmt/mgmt_vnode/src/vmHandle.c

@@ -361,7 +361,8 @@ SArray *vmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_CHANGE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_DELETE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_COMMIT_OFFSET, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_CHECK_ALTER_INFO, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_ADD_CHECK_INFO, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_DELETE_CHECK_INFO, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_CONSUME, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_DELETE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_BATCH_DEL, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;

source/dnode/mnode/impl/src/mndOffset.c

@@ -15,10 +15,10 @@
 
 #define _DEFAULT_SOURCE
 #include "mndOffset.h"
-#include "mndPrivilege.h"
 #include "mndDb.h"
 #include "mndDnode.h"
 #include "mndMnode.h"
+#include "mndPrivilege.h"
 #include "mndShow.h"
 #include "mndStb.h"
 #include "mndTopic.h"
@@ -305,7 +305,7 @@ int32_t mndDropOffsetByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) {
     sdbRelease(pSdb, pOffset);
   }
 
-   return code;
+  return code;
 }
 
 int32_t mndDropOffsetByTopic(SMnode *pMnode, STrans *pTrans, const char *topic) {

source/dnode/mnode/impl/src/mndStb.c

@@ -1145,7 +1145,7 @@ static int32_t mndAddSuperTableTag(const SStbObj *pOld, SStbObj *pNew, SArray *p
   return 0;
 }
 
-int32_t mndCheckColAndTagModifiable(SMnode *pMnode, const char *stbname, int64_t suid, col_id_t colId) {
+static int32_t mndCheckAlterColForTopic(SMnode *pMnode, const char *stbFullName, int64_t suid, col_id_t colId) {
   SSdb *pSdb = pMnode->pSdb;
   void *pIter = NULL;
   while (1) {
@@ -1154,7 +1154,7 @@ int32_t mndCheckColAndTagModifiable(SMnode *pMnode, const char *stbname, int64_t
     if (pIter == NULL) break;
 
     mDebug("topic:%s, check tag and column modifiable, stb:%s suid:%" PRId64 " colId:%d, subType:%d sql:%s",
-           pTopic->name, stbname, suid, colId, pTopic->subType, pTopic->sql);
+           pTopic->name, stbFullName, suid, colId, pTopic->subType, pTopic->sql);
     if (pTopic->subType != TOPIC_SUB_TYPE__COLUMN) {
       sdbRelease(pSdb, pTopic);
       continue;
@@ -1192,20 +1192,66 @@ int32_t mndCheckColAndTagModifiable(SMnode *pMnode, const char *stbname, int64_t
     sdbRelease(pSdb, pTopic);
     nodesDestroyNode(pAst);
   }
+  return 0;
+}
+
+static int32_t mndCheckAlterColForStream(SMnode *pMnode, const char *stbFullName, int64_t suid, col_id_t colId) {
+  SSdb *pSdb = pMnode->pSdb;
+  void *pIter = NULL;
+  while (1) {
+    SStreamObj *pStream = NULL;
+    pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&pStream);
+    if (pIter == NULL) break;
+
+    SNode *pAst = NULL;
+    if (nodesStringToNode(pStream->ast, &pAst) != 0) {
+      ASSERT(0);
+      return -1;
+    }
+
+    SNodeList *pNodeList = NULL;
+    nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList);
+    SNode *pNode = NULL;
+    FOREACH(pNode, pNodeList) {
+      SColumnNode *pCol = (SColumnNode *)pNode;
+
+      if (pCol->tableId != suid) {
+        mDebug("stream:%s, check colId:%d passed", pStream->name, pCol->colId);
+        goto NEXT;
+      }
+      if (pCol->colId > 0 && pCol->colId == colId) {
+        sdbRelease(pSdb, pStream);
+        nodesDestroyNode(pAst);
+        terrno = TSDB_CODE_MND_STREAM_MUST_BE_DELETED;
+        mError("stream:%s, check colId:%d conflicted", pStream->name, pCol->colId);
+        return -1;
+      }
+      mDebug("stream:%s, check colId:%d passed", pStream->name, pCol->colId);
+    }
+
+  NEXT:
+    sdbRelease(pSdb, pStream);
+    nodesDestroyNode(pAst);
+  }
+  return 0;
+}
 
+static int32_t mndCheckAlterColForTSma(SMnode *pMnode, const char *stbFullName, int64_t suid, col_id_t colId) {
+  SSdb *pSdb = pMnode->pSdb;
+  void *pIter = NULL;
   while (1) {
     SSmaObj *pSma = NULL;
     pIter = sdbFetch(pSdb, SDB_SMA, pIter, (void **)&pSma);
     if (pIter == NULL) break;
 
-    mDebug("tsma:%s, check tag and column modifiable, stb:%s suid:%" PRId64 " colId:%d, sql:%s", pSma->name, stbname,
-           suid, colId, pSma->sql);
+    mDebug("tsma:%s, check tag and column modifiable, stb:%s suid:%" PRId64 " colId:%d, sql:%s", pSma->name,
+           stbFullName, suid, colId, pSma->sql);
 
     SNode *pAst = NULL;
     if (nodesStringToNode(pSma->ast, &pAst) != 0) {
       terrno = TSDB_CODE_SDB_INVALID_DATA_CONTENT;
       mError("tsma:%s, check tag and column modifiable, stb:%s suid:%" PRId64 " colId:%d failed since parse AST err",
-             pSma->name, stbname, suid, colId);
+             pSma->name, stbFullName, suid, colId);
       return -1;
     }
 
@@ -1218,7 +1264,7 @@ int32_t mndCheckColAndTagModifiable(SMnode *pMnode, const char *stbname, int64_t
 
       if ((pCol->tableId != suid) && (pSma->stbUid != suid)) {
         mDebug("tsma:%s, check colId:%d passed", pSma->name, pCol->colId);
-        goto NEXT2;
+        goto NEXT;
       }
       if ((pCol->colId) > 0 && (pCol->colId == colId)) {
         sdbRelease(pSdb, pSma);
@@ -1230,11 +1276,24 @@ int32_t mndCheckColAndTagModifiable(SMnode *pMnode, const char *stbname, int64_t
       mDebug("tsma:%s, check colId:%d passed", pSma->name, pCol->colId);
     }
 
-  NEXT2:
+  NEXT:
     sdbRelease(pSdb, pSma);
     nodesDestroyNode(pAst);
   }
+  return 0;
+}
+
+int32_t mndCheckColAndTagModifiable(SMnode *pMnode, const char *stbFullName, int64_t suid, col_id_t colId) {
+  if (mndCheckAlterColForTopic(pMnode, stbFullName, suid, colId) < 0) {
+    return -1;
+  }
+  if (mndCheckAlterColForStream(pMnode, stbFullName, suid, colId) < 0) {
+    return -1;
+  }
 
+  if (mndCheckAlterColForTSma(pMnode, stbFullName, suid, colId) < 0) {
+    return -1;
+  }
   return 0;
 }
 
@@ -1930,6 +1989,90 @@ _OVER:
   return code;
 }
 
+static int32_t mndCheckDropStbForTopic(SMnode *pMnode, const char *stbFullName, int64_t suid) {
+  SSdb *pSdb = pMnode->pSdb;
+  void *pIter = NULL;
+  while (1) {
+    SMqTopicObj *pTopic = NULL;
+    pIter = sdbFetch(pSdb, SDB_TOPIC, pIter, (void **)&pTopic);
+    if (pIter == NULL) break;
+
+    if (pTopic->subType == TOPIC_SUB_TYPE__TABLE) {
+      if (pTopic->stbUid == suid) {
+        sdbRelease(pSdb, pTopic);
+        return -1;
+      }
+    }
+
+    if (pTopic->subType != TOPIC_SUB_TYPE__COLUMN) {
+      sdbRelease(pSdb, pTopic);
+      continue;
+    }
+
+    SNode *pAst = NULL;
+    if (nodesStringToNode(pTopic->ast, &pAst) != 0) {
+      ASSERT(0);
+      return -1;
+    }
+
+    SNodeList *pNodeList = NULL;
+    nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList);
+    SNode *pNode = NULL;
+    FOREACH(pNode, pNodeList) {
+      SColumnNode *pCol = (SColumnNode *)pNode;
+
+      if (pCol->tableId != suid) {
+        mDebug("topic:%s, check colId:%d passed", pTopic->name, pCol->colId);
+        sdbRelease(pSdb, pTopic);
+        nodesDestroyNode(pAst);
+        return -1;
+      } else {
+        goto NEXT;
+      }
+    }
+  NEXT:
+    sdbRelease(pSdb, pTopic);
+    nodesDestroyNode(pAst);
+  }
+  return 0;
+}
+
+static int32_t mndCheckDropStbForStream(SMnode *pMnode, const char *stbFullName, int64_t suid) {
+  SSdb *pSdb = pMnode->pSdb;
+  void *pIter = NULL;
+  while (1) {
+    SStreamObj *pStream = NULL;
+    pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&pStream);
+    if (pIter == NULL) break;
+
+    SNode *pAst = NULL;
+    if (nodesStringToNode(pStream->ast, &pAst) != 0) {
+      ASSERT(0);
+      return -1;
+    }
+
+    SNodeList *pNodeList = NULL;
+    nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList);
+    SNode *pNode = NULL;
+    FOREACH(pNode, pNodeList) {
+      SColumnNode *pCol = (SColumnNode *)pNode;
+
+      if (pCol->tableId != suid) {
+        mDebug("stream:%s, check colId:%d passed", pStream->name, pCol->colId);
+        sdbRelease(pSdb, pStream);
+        nodesDestroyNode(pAst);
+        return -1;
+      } else {
+        goto NEXT;
+      }
+    }
+  NEXT:
+    sdbRelease(pSdb, pStream);
+    nodesDestroyNode(pAst);
+  }
+  return 0;
+}
+
 static int32_t mndProcessDropStbReq(SRpcMsg *pReq) {
   SMnode      *pMnode = pReq->info.node;
   int32_t      code = -1;
@@ -1971,6 +2114,16 @@ static int32_t mndProcessDropStbReq(SRpcMsg *pReq) {
     goto _OVER;
   }
 
+  if (mndCheckDropStbForTopic(pMnode, dropReq.name, pStb->uid) < 0) {
+    terrno = TSDB_CODE_MND_TOPIC_MUST_BE_DELETED;
+    goto _OVER;
+  }
+
+  if (mndCheckDropStbForStream(pMnode, dropReq.name, pStb->uid) < 0) {
+    terrno = TSDB_CODE_MND_STREAM_MUST_BE_DELETED;
+    goto _OVER;
+  }
+
   code = mndDropStb(pMnode, pReq, pDb, pStb);
   if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
 

source/dnode/mnode/impl/src/mndTopic.c

@@ -57,7 +57,8 @@ int32_t mndInitTopic(SMnode *pMnode) {
   mndSetMsgHandle(pMnode, TDMT_MND_CREATE_TOPIC, mndProcessCreateTopicReq);
   mndSetMsgHandle(pMnode, TDMT_MND_DROP_TOPIC, mndProcessDropTopicReq);
   mndSetMsgHandle(pMnode, TDMT_VND_DROP_TOPIC_RSP, mndTransProcessRsp);
-  mndSetMsgHandle(pMnode, TDMT_VND_CHECK_ALTER_INFO_RSP, mndTransProcessRsp);
+  mndSetMsgHandle(pMnode, TDMT_VND_ADD_CHECK_INFO_RSP, mndTransProcessRsp);
+  mndSetMsgHandle(pMnode, TDMT_VND_DELETE_CHECK_INFO_RSP, mndTransProcessRsp);
 
   mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_TOPICS, mndRetrieveTopic);
   mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_TOPICS, mndCancelGetNextTopic);
@@ -450,7 +451,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
   sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
 
   if (topicObj.ntbUid != 0) {
-    SCheckAlterInfo info;
+    STqCheckInfo info;
     memcpy(info.topic, topicObj.name, TSDB_TOPIC_FNAME_LEN);
     info.ntbUid = topicObj.ntbUid;
     info.colIdList = topicObj.ntbColIds;
@@ -470,7 +471,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
       // encoder check alter info
       int32_t len;
       int32_t code;
-      tEncodeSize(tEncodeSCheckAlterInfo, &info, len, code);
+      tEncodeSize(tEncodeSTqCheckInfo, &info, len, code);
       if (code < 0) {
         sdbRelease(pSdb, pVgroup);
         mndTransDrop(pTrans);
@@ -481,7 +482,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
       void    *abuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
       SEncoder encoder;
       tEncoderInit(&encoder, abuf, len);
-      if (tEncodeSCheckAlterInfo(&encoder, &info) < 0) {
+      if (tEncodeSTqCheckInfo(&encoder, &info) < 0) {
         sdbRelease(pSdb, pVgroup);
         mndTransDrop(pTrans);
         return -1;
@@ -493,7 +494,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
       action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
       action.pCont = buf;
       action.contLen = sizeof(SMsgHead) + len;
-      action.msgType = TDMT_VND_CHECK_ALTER_INFO;
+      action.msgType = TDMT_VND_ADD_CHECK_INFO;
       if (mndTransAppendRedoAction(pTrans, &action) != 0) {
         taosMemoryFree(buf);
         sdbRelease(pSdb, pVgroup);
@@ -659,12 +660,14 @@ static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {
 
   mDebug("trans:%d, used to drop topic:%s", pTrans->id, pTopic->name);
 
+#if 0
   if (mndDropOffsetByTopic(pMnode, pTrans, dropReq.name) < 0) {
     ASSERT(0);
     mndTransDrop(pTrans);
     mndReleaseTopic(pMnode, pTopic);
     return -1;
   }
+#endif
 
   // TODO check if rebalancing
   if (mndDropSubByTopic(pMnode, pTrans, dropReq.name) < 0) {
@@ -675,6 +678,37 @@ static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {
     return -1;
   }
 
+  if (pTopic->ntbUid != 0) {
+    // broadcast to all vnode
+    void   *pIter = NULL;
+    SVgObj *pVgroup = NULL;
+    while (1) {
+      pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
+      if (pIter == NULL) break;
+      if (!mndVgroupInDb(pVgroup, pTopic->dbUid)) {
+        sdbRelease(pSdb, pVgroup);
+        continue;
+      }
+
+      void *buf = taosMemoryCalloc(1, sizeof(SMsgHead) + TSDB_TOPIC_FNAME_LEN);
+      void *abuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
+      ((SMsgHead *)buf)->vgId = htonl(pVgroup->vgId);
+      memcpy(abuf, pTopic->name, TSDB_TOPIC_FNAME_LEN);
+
+      STransAction action = {0};
+      action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
+      action.pCont = buf;
+      action.contLen = sizeof(SMsgHead) + TSDB_TOPIC_FNAME_LEN;
+      action.msgType = TDMT_VND_DELETE_CHECK_INFO;
+      if (mndTransAppendRedoAction(pTrans, &action) != 0) {
+        taosMemoryFree(buf);
+        sdbRelease(pSdb, pVgroup);
+        mndTransDrop(pTrans);
+        return -1;
+      }
+    }
+  }
+
   int32_t code = mndDropTopic(pMnode, pTrans, pReq, pTopic);
   mndReleaseTopic(pMnode, pTopic);
 

source/dnode/vnode/src/inc/tq.h

@@ -117,16 +117,15 @@ typedef struct {
 struct STQ {
   SVnode*   pVnode;
   char*     path;
-  SHashObj* pushMgr;     // consumerId -> STqHandle*
-  SHashObj* handles;     // subKey -> STqHandle
-  SHashObj* pAlterInfo;  // topic -> SAlterCheckInfo
+  SHashObj* pPushMgr;    // consumerId -> STqHandle*
+  SHashObj* pHandle;     // subKey -> STqHandle
+  SHashObj* pCheckInfo;  // topic -> SAlterCheckInfo
 
   STqOffsetStore* pOffsetStore;
 
-  TDB* pMetaStore;
+  TDB* pMetaDB;
   TTB* pExecStore;
-
-  TTB* pAlterInfoStore;
+  TTB* pCheckStore;
 
   SStreamMeta* pStreamMeta;
 };
@@ -155,6 +154,9 @@ int32_t tqMetaClose(STQ* pTq);
 int32_t tqMetaSaveHandle(STQ* pTq, const char* key, const STqHandle* pHandle);
 int32_t tqMetaDeleteHandle(STQ* pTq, const char* key);
 int32_t tqMetaRestoreHandle(STQ* pTq);
+int32_t tqMetaSaveCheckInfo(STQ* pTq, const char* key, const void* value, int32_t vLen);
+int32_t tqMetaDeleteCheckInfo(STQ* pTq, const char* key);
+int32_t tqMetaRestoreCheckInfo(STQ* pTq);
 
 typedef struct {
   int32_t size;

source/dnode/vnode/src/inc/tsdb.h

@@ -123,6 +123,7 @@ int32_t tGetBlockL(uint8_t *p, void *ph);
 int32_t tPutBlockIdx(uint8_t *p, void *ph);
 int32_t tGetBlockIdx(uint8_t *p, void *ph);
 int32_t tCmprBlockIdx(void const *lhs, void const *rhs);
+int32_t tCmprBlockL(void const *lhs, void const *rhs);
 // SColdata
 void    tColDataInit(SColData *pColData, int16_t cid, int8_t type, int8_t smaOn);
 void    tColDataReset(SColData *pColData);
@@ -342,9 +343,6 @@ struct STbData {
   tb_uid_t     uid;
   TSKEY        minKey;
   TSKEY        maxKey;
-  int64_t      minVersion;
-  int64_t      maxVersion;
-  int32_t      maxSkmVer;
   SDelData    *pHead;
   SDelData    *pTail;
   SMemSkipList sl;
@@ -358,8 +356,6 @@ struct SMemTable {
   volatile int32_t nRef;
   TSKEY            minKey;
   TSKEY            maxKey;
-  int64_t          minVersion;
-  int64_t          maxVersion;
   int64_t          nRow;
   int64_t          nDel;
   struct {

source/dnode/vnode/src/inc/vnodeInt.h

@@ -163,13 +163,16 @@ int     tqPushMsg(STQ*, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver);
 int     tqCommit(STQ*);
 int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, bool isAdd);
 int32_t tqCheckColModifiable(STQ* pTq, int64_t tbUid, int32_t colId);
-int32_t tqProcessCheckAlterInfoReq(STQ* pTq, char* msg, int32_t msgLen);
-int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen);
-int32_t tqProcessVgDeleteReq(STQ* pTq, char* msg, int32_t msgLen);
-int32_t tqProcessOffsetCommitReq(STQ* pTq, char* msg, int32_t msgLen, int64_t ver);
+// tq-mq
+int32_t tqProcessAddCheckInfoReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
+int32_t tqProcessDelCheckInfoReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
+int32_t tqProcessVgChangeReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
+int32_t tqProcessVgDeleteReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
+int32_t tqProcessOffsetCommitReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
 int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg);
-int32_t tqProcessTaskDeployReq(STQ* pTq, char* msg, int32_t msgLen);
-int32_t tqProcessTaskDropReq(STQ* pTq, char* msg, int32_t msgLen);
+// tq-stream
+int32_t tqProcessTaskDeployReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
+int32_t tqProcessTaskDropReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
 int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* data, int64_t ver);
 int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskDispatchReq(STQ* pTq, SRpcMsg* pMsg, bool exec);

source/dnode/vnode/src/meta/metaQuery.c

@@ -429,18 +429,65 @@ STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver) {
 }
 
 int32_t metaGetTbTSchemaEx(SMeta *pMeta, tb_uid_t suid, tb_uid_t uid, int32_t sver, STSchema **ppTSchema) {
-  int32_t   code = 0;
-  STSchema *pTSchema = NULL;
-  SSkmDbKey skmDbKey = {.uid = suid ? suid : uid, .sver = sver};
+  int32_t code = 0;
+
   void     *pData = NULL;
   int       nData = 0;
+  SSkmDbKey skmDbKey;
+  if (sver <= 0) {
+    SMetaInfo info;
+    if (metaGetInfo(pMeta, suid ? suid : uid, &info) == 0) {
+      sver = info.skmVer;
+    } else {
+      TBC *pSkmDbC = NULL;
+      int  c;
+
+      skmDbKey.uid = suid ? suid : uid;
+      skmDbKey.sver = INT32_MAX;
+
+      tdbTbcOpen(pMeta->pSkmDb, &pSkmDbC, NULL);
+      metaRLock(pMeta);
+
+      if (tdbTbcMoveTo(pSkmDbC, &skmDbKey, sizeof(skmDbKey), &c) < 0) {
+        metaULock(pMeta);
+        tdbTbcClose(pSkmDbC);
+        code = TSDB_CODE_NOT_FOUND;
+        goto _exit;
+      }
+
+      ASSERT(c);
+
+      if (c < 0) {
+        tdbTbcMoveToPrev(pSkmDbC);
+      }
+
+      const void *pKey = NULL;
+      int32_t     nKey = 0;
+      tdbTbcGet(pSkmDbC, &pKey, &nKey, NULL, NULL);
+
+      if (((SSkmDbKey *)pKey)->uid != uid) {
+        metaULock(pMeta);
+        tdbTbcClose(pSkmDbC);
+        code = TSDB_CODE_NOT_FOUND;
+        goto _exit;
+      }
 
-  // query
+      sver = ((SSkmDbKey *)pKey)->sver;
+
+      metaULock(pMeta);
+      tdbTbcClose(pSkmDbC);
+    }
+  }
+
+  ASSERT(sver > 0);
+
+  skmDbKey.uid = suid ? suid : uid;
+  skmDbKey.sver = sver;
   metaRLock(pMeta);
-  if (tdbTbGet(pMeta->pSkmDb, &skmDbKey, sizeof(skmDbKey), &pData, &nData) < 0) {
-    code = TSDB_CODE_NOT_FOUND;
+  if (tdbTbGet(pMeta->pSkmDb, &skmDbKey, sizeof(SSkmDbKey), &pData, &nData) < 0) {
     metaULock(pMeta);
-    goto _err;
+    code = TSDB_CODE_NOT_FOUND;
+    goto _exit;
   }
   metaULock(pMeta);
 
@@ -462,15 +509,13 @@ int32_t metaGetTbTSchemaEx(SMeta *pMeta, tb_uid_t suid, tb_uid_t uid, int32_t sv
     SSchema *pSchema = pSchemaWrapper->pSchema + i;
     tdAddColToSchema(&sb, pSchema->type, pSchema->flags, pSchema->colId, pSchema->bytes);
   }
-  pTSchema = tdGetSchemaFromBuilder(&sb);
+  STSchema *pTSchema = tdGetSchemaFromBuilder(&sb);
   tdDestroyTSchemaBuilder(&sb);
 
   *ppTSchema = pTSchema;
   taosMemoryFree(pSchemaWrapper->pSchema);
-  return code;
 
-_err:
-  *ppTSchema = NULL;
+_exit:
   return code;
 }
 

source/dnode/vnode/src/tq/tq.c

@@ -60,11 +60,11 @@ STQ* tqOpen(const char* path, SVnode* pVnode) {
   pTq->path = strdup(path);
   pTq->pVnode = pVnode;
 
-  pTq->handles = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK);
+  pTq->pHandle = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK);
 
-  pTq->pushMgr = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
+  pTq->pPushMgr = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
 
-  pTq->pAlterInfo = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK);
+  pTq->pCheckInfo = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK);
 
   if (tqMetaOpen(pTq) < 0) {
     ASSERT(0);
@@ -85,9 +85,9 @@ STQ* tqOpen(const char* path, SVnode* pVnode) {
 void tqClose(STQ* pTq) {
   if (pTq) {
     tqOffsetClose(pTq->pOffsetStore);
-    taosHashCleanup(pTq->handles);
-    taosHashCleanup(pTq->pushMgr);
-    taosHashCleanup(pTq->pAlterInfo);
+    taosHashCleanup(pTq->pHandle);
+    taosHashCleanup(pTq->pPushMgr);
+    taosHashCleanup(pTq->pCheckInfo);
     taosMemoryFree(pTq->path);
     tqMetaClose(pTq);
     streamMetaClose(pTq->pStreamMeta);
@@ -183,7 +183,12 @@ int32_t tqSendDataRsp(STQ* pTq, const SRpcMsg* pMsg, const SMqPollReq* pReq, con
   return 0;
 }
 
-int32_t tqProcessOffsetCommitReq(STQ* pTq, char* msg, int32_t msgLen, int64_t ver) {
+static FORCE_INLINE bool tqOffsetLessOrEqual(const STqOffset* pLeft, const STqOffset* pRight) {
+  return pLeft->val.type == TMQ_OFFSET__LOG && pRight->val.type == TMQ_OFFSET__LOG &&
+         pLeft->val.version <= pRight->val.version;
+}
+
+int32_t tqProcessOffsetCommitReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen) {
   STqOffset offset = {0};
   SDecoder  decoder;
   tDecoderInit(&decoder, msg, msgLen);
@@ -199,19 +204,24 @@ int32_t tqProcessOffsetCommitReq(STQ* pTq, char* msg, int32_t msgLen, int64_t ve
   } else if (offset.val.type == TMQ_OFFSET__LOG) {
     tqDebug("receive offset commit msg to %s on vgId:%d, offset(type:log) version:%" PRId64, offset.subKey,
             TD_VID(pTq->pVnode), offset.val.version);
+    if (offset.val.version + 1 == version) {
+      offset.val.version += 1;
+    }
   } else {
     ASSERT(0);
   }
-  /*STqOffset* pOffset = tqOffsetRead(pTq->pOffsetStore, offset.subKey);*/
-  /*if (pOffset != NULL) {*/
-  /*if (pOffset->val.type == TMQ_OFFSET__LOG && pOffset->val.version < offset.val.version) {*/
+  STqOffset* pOffset = tqOffsetRead(pTq->pOffsetStore, offset.subKey);
+  if (pOffset != NULL && tqOffsetLessOrEqual(&offset, pOffset)) {
+    return 0;
+  }
+
   if (tqOffsetWrite(pTq->pOffsetStore, &offset) < 0) {
     ASSERT(0);
     return -1;
   }
 
   if (offset.val.type == TMQ_OFFSET__LOG) {
-    STqHandle* pHandle = taosHashGet(pTq->handles, offset.subKey, strlen(offset.subKey));
+    STqHandle* pHandle = taosHashGet(pTq->pHandle, offset.subKey, strlen(offset.subKey));
     if (pHandle) {
       if (walRefVer(pHandle->pRef, offset.val.version) < 0) {
         ASSERT(0);
@@ -220,6 +230,8 @@ int32_t tqProcessOffsetCommitReq(STQ* pTq, char* msg, int32_t msgLen, int64_t ve
     }
   }
 
+  // rsp
+
   /*}*/
   /*}*/
 
@@ -229,15 +241,15 @@ int32_t tqProcessOffsetCommitReq(STQ* pTq, char* msg, int32_t msgLen, int64_t ve
 int32_t tqCheckColModifiable(STQ* pTq, int64_t tbUid, int32_t colId) {
   void* pIter = NULL;
   while (1) {
-    pIter = taosHashIterate(pTq->pAlterInfo, pIter);
+    pIter = taosHashIterate(pTq->pCheckInfo, pIter);
     if (pIter == NULL) break;
-    SCheckAlterInfo* pCheck = (SCheckAlterInfo*)pIter;
+    STqCheckInfo* pCheck = (STqCheckInfo*)pIter;
     if (pCheck->ntbUid == tbUid) {
       int32_t sz = taosArrayGetSize(pCheck->colIdList);
       for (int32_t i = 0; i < sz; i++) {
         int16_t forbidColId = *(int16_t*)taosArrayGet(pCheck->colIdList, i);
         if (forbidColId == colId) {
-          taosHashCancelIterate(pTq->pAlterInfo, pIter);
+          taosHashCancelIterate(pTq->pCheckInfo, pIter);
           return -1;
         }
       }
@@ -289,7 +301,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
   SWalCkHead*  pCkHead = NULL;
 
   // 1.find handle
-  STqHandle* pHandle = taosHashGet(pTq->handles, pReq->subKey, strlen(pReq->subKey));
+  STqHandle* pHandle = taosHashGet(pTq->pHandle, pReq->subKey, strlen(pReq->subKey));
   /*ASSERT(pHandle);*/
   if (pHandle == NULL) {
     tqError("tmq poll: no consumer handle for consumer:%" PRId64 ", in vgId:%d, subkey %s", consumerId,
@@ -478,10 +490,10 @@ OVER:
   return code;
 }
 
-int32_t tqProcessVgDeleteReq(STQ* pTq, char* msg, int32_t msgLen) {
+int32_t tqProcessVgDeleteReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen) {
   SMqVDeleteReq* pReq = (SMqVDeleteReq*)msg;
 
-  int32_t code = taosHashRemove(pTq->handles, pReq->subKey, strlen(pReq->subKey));
+  int32_t code = taosHashRemove(pTq->pHandle, pReq->subKey, strlen(pReq->subKey));
   ASSERT(code == 0);
 
   tqOffsetDelete(pTq->pOffsetStore, pReq->subKey);
@@ -492,27 +504,43 @@ int32_t tqProcessVgDeleteReq(STQ* pTq, char* msg, int32_t msgLen) {
   return 0;
 }
 
-int32_t tqProcessCheckAlterInfoReq(STQ* pTq, char* msg, int32_t msgLen) {
-  SCheckAlterInfo info = {0};
-  SDecoder        decoder;
+int32_t tqProcessAddCheckInfoReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen) {
+  STqCheckInfo info = {0};
+  SDecoder     decoder;
   tDecoderInit(&decoder, msg, msgLen);
-  if (tDecodeSCheckAlterInfo(&decoder, &info) < 0) {
+  if (tDecodeSTqCheckInfo(&decoder, &info) < 0) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
   }
   tDecoderClear(&decoder);
-  if (taosHashPut(pTq->pAlterInfo, info.topic, strlen(info.topic), &info, sizeof(SCheckAlterInfo)) < 0) {
+  if (taosHashPut(pTq->pCheckInfo, info.topic, strlen(info.topic), &info, sizeof(STqCheckInfo)) < 0) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return -1;
+  }
+  if (tqMetaSaveCheckInfo(pTq, info.topic, msg, msgLen) < 0) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
   }
   return 0;
 }
 
-int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
+int32_t tqProcessDelCheckInfoReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen) {
+  if (taosHashRemove(pTq->pCheckInfo, msg, strlen(msg)) < 0) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return -1;
+  }
+  if (tqMetaDeleteCheckInfo(pTq, msg) < 0) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return -1;
+  }
+  return 0;
+}
+
+int32_t tqProcessVgChangeReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen) {
   SMqRebVgReq req = {0};
   tDecodeSMqRebVgReq(msg, &req);
   // todo lock
-  STqHandle* pHandle = taosHashGet(pTq->handles, req.subKey, strlen(req.subKey));
+  STqHandle* pHandle = taosHashGet(pTq->pHandle, req.subKey, strlen(req.subKey));
   if (pHandle == NULL) {
     if (req.oldConsumerId != -1) {
       tqError("vgId:%d, build new consumer handle %s for consumer %d, but old consumerId is %ld", req.vgId, req.subKey,
@@ -579,7 +607,7 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
       tqReaderSetTbUidList(pHandle->execHandle.pExecReader, tbUidList);
       taosArrayDestroy(tbUidList);
     }
-    taosHashPut(pTq->handles, req.subKey, strlen(req.subKey), pHandle, sizeof(STqHandle));
+    taosHashPut(pTq->pHandle, req.subKey, strlen(req.subKey), pHandle, sizeof(STqHandle));
     tqDebug("try to persist handle %s consumer %" PRId64, req.subKey, pHandle->consumerId);
     if (tqMetaSaveHandle(pTq, req.subKey, pHandle) < 0) {
       // TODO
@@ -668,34 +696,9 @@ FAIL:
   return code;
 }
 
-int32_t tqProcessTaskDeployReq(STQ* pTq, char* msg, int32_t msgLen) {
+int32_t tqProcessTaskDeployReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen) {
   //
-  return streamMetaAddSerializedTask(pTq->pStreamMeta, msg, msgLen);
-#if 0
-  SStreamTask* pTask = taosMemoryCalloc(1, sizeof(SStreamTask));
-  if (pTask == NULL) {
-    return -1;
-  }
-  SDecoder decoder;
-  tDecoderInit(&decoder, (uint8_t*)msg, msgLen);
-  if (tDecodeSStreamTask(&decoder, pTask) < 0) {
-    ASSERT(0);
-    goto FAIL;
-  }
-  tDecoderClear(&decoder);
-
-  if (tqExpandTask(pTq, pTask) < 0) {
-    goto FAIL;
-  }
-
-  taosHashPut(pTq->pStreamTasks, &pTask->taskId, sizeof(int32_t), &pTask, sizeof(void*));
-
-  return 0;
-
-FAIL:
-  if (pTask) taosMemoryFree(pTask);
-  return -1;
-#endif
+  return streamMetaAddSerializedTask(pTq->pStreamMeta, version, msg, msgLen);
 }
 
 int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* pReq, int64_t ver) {
@@ -817,7 +820,7 @@ int32_t tqProcessTaskDispatchRsp(STQ* pTq, SRpcMsg* pMsg) {
   }
 }
 
-int32_t tqProcessTaskDropReq(STQ* pTq, char* msg, int32_t msgLen) {
+int32_t tqProcessTaskDropReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen) {
   SVDropStreamTaskReq* pReq = (SVDropStreamTaskReq*)msg;
 
   return streamMetaRemoveTask(pTq->pStreamMeta, pReq->taskId);

source/dnode/vnode/src/tq/tqMeta.c

@@ -43,86 +43,116 @@ int32_t tDecodeSTqHandle(SDecoder* pDecoder, STqHandle* pHandle) {
   return 0;
 }
 
-int32_t tqMetaRestoreHandle(STQ* pTq) {
-  TBC* pCur = NULL;
-  if (tdbTbcOpen(pTq->pExecStore, &pCur, NULL) < 0) {
+int32_t tqMetaOpen(STQ* pTq) {
+  if (tdbOpen(pTq->path, 16 * 1024, 1, &pTq->pMetaDB) < 0) {
     ASSERT(0);
     return -1;
   }
 
-  void*    pKey = NULL;
-  int      kLen = 0;
-  void*    pVal = NULL;
-  int      vLen = 0;
-  SDecoder decoder;
+  if (tdbTbOpen("tq.db", -1, -1, NULL, pTq->pMetaDB, &pTq->pExecStore) < 0) {
+    ASSERT(0);
+    return -1;
+  }
 
-  tdbTbcMoveToFirst(pCur);
+  if (tdbTbOpen("tq.check.db", -1, -1, NULL, pTq->pMetaDB, &pTq->pCheckStore) < 0) {
+    ASSERT(0);
+    return -1;
+  }
 
-  while (tdbTbcNext(pCur, &pKey, &kLen, &pVal, &vLen) == 0) {
-    STqHandle handle;
-    tDecoderInit(&decoder, (uint8_t*)pVal, vLen);
-    tDecodeSTqHandle(&decoder, &handle);
+  if (tqMetaRestoreHandle(pTq) < 0) {
+    return -1;
+  }
 
-    handle.pRef = walOpenRef(pTq->pVnode->pWal);
-    if (handle.pRef == NULL) {
-      ASSERT(0);
-      return -1;
-    }
-    walRefVer(handle.pRef, handle.snapshotVer);
+  if (tqMetaRestoreCheckInfo(pTq) < 0) {
+    return -1;
+  }
 
-    if (handle.execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
-      SReadHandle reader = {
-          .meta = pTq->pVnode->pMeta,
-          .vnode = pTq->pVnode,
-          .initTableReader = true,
-          .initTqReader = true,
-          .version = handle.snapshotVer,
-      };
+  return 0;
+}
 
-      handle.execHandle.execCol.task = qCreateQueueExecTaskInfo(
-          handle.execHandle.execCol.qmsg, &reader, &handle.execHandle.numOfCols, &handle.execHandle.pSchemaWrapper);
-      ASSERT(handle.execHandle.execCol.task);
-      void* scanner = NULL;
-      qExtractStreamScanner(handle.execHandle.execCol.task, &scanner);
-      ASSERT(scanner);
-      handle.execHandle.pExecReader = qExtractReaderFromStreamScanner(scanner);
-      ASSERT(handle.execHandle.pExecReader);
-    } else {
-      handle.pWalReader = walOpenReader(pTq->pVnode->pWal, NULL);
-      handle.execHandle.execDb.pFilterOutTbUid =
-          taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
-    }
-    tqDebug("tq restore %s consumer %" PRId64 " vgId:%d", handle.subKey, handle.consumerId, TD_VID(pTq->pVnode));
-    taosHashPut(pTq->handles, pKey, kLen, &handle, sizeof(STqHandle));
+int32_t tqMetaClose(STQ* pTq) {
+  if (pTq->pExecStore) {
+    tdbTbClose(pTq->pExecStore);
   }
-
-  tdbTbcClose(pCur);
+  if (pTq->pCheckStore) {
+    tdbTbClose(pTq->pCheckStore);
+  }
+  tdbClose(pTq->pMetaDB);
   return 0;
 }
 
-int32_t tqMetaOpen(STQ* pTq) {
-  if (tdbOpen(pTq->path, 16 * 1024, 1, &pTq->pMetaStore) < 0) {
-    ASSERT(0);
+int32_t tqMetaSaveCheckInfo(STQ* pTq, const char* key, const void* value, int32_t vLen) {
+  TXN txn;
+  if (tdbTxnOpen(&txn, 0, tdbDefaultMalloc, tdbDefaultFree, NULL, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) < 0) {
     return -1;
   }
 
-  if (tdbTbOpen("tq.db", -1, -1, NULL, pTq->pMetaStore, &pTq->pExecStore) < 0) {
-    ASSERT(0);
+  if (tdbBegin(pTq->pMetaDB, &txn) < 0) {
     return -1;
   }
 
-  if (tqMetaRestoreHandle(pTq) < 0) {
+  if (tdbTbUpsert(pTq->pExecStore, key, strlen(key), value, vLen, &txn) < 0) {
+    return -1;
+  }
+
+  if (tdbCommit(pTq->pMetaDB, &txn) < 0) {
     return -1;
   }
 
   return 0;
 }
 
-int32_t tqMetaClose(STQ* pTq) {
-  if (pTq->pExecStore) {
-    tdbTbClose(pTq->pExecStore);
+int32_t tqMetaDeleteCheckInfo(STQ* pTq, const char* key) {
+  TXN txn;
+
+  if (tdbTxnOpen(&txn, 0, tdbDefaultMalloc, tdbDefaultFree, NULL, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) < 0) {
+    ASSERT(0);
+  }
+
+  if (tdbBegin(pTq->pMetaDB, &txn) < 0) {
+    ASSERT(0);
+  }
+
+  if (tdbTbDelete(pTq->pCheckStore, key, (int)strlen(key), &txn) < 0) {
+    /*ASSERT(0);*/
+  }
+
+  if (tdbCommit(pTq->pMetaDB, &txn) < 0) {
+    ASSERT(0);
+  }
+
+  return 0;
+}
+
+int32_t tqMetaRestoreCheckInfo(STQ* pTq) {
+  TBC* pCur = NULL;
+  if (tdbTbcOpen(pTq->pCheckStore, &pCur, NULL) < 0) {
+    ASSERT(0);
+    return -1;
+  }
+
+  void*    pKey = NULL;
+  int      kLen = 0;
+  void*    pVal = NULL;
+  int      vLen = 0;
+  SDecoder decoder;
+
+  tdbTbcMoveToFirst(pCur);
+
+  while (tdbTbcNext(pCur, &pKey, &kLen, &pVal, &vLen) == 0) {
+    STqCheckInfo info;
+    tDecoderInit(&decoder, (uint8_t*)pVal, vLen);
+    if (tDecodeSTqCheckInfo(&decoder, &info) < 0) {
+      terrno = TSDB_CODE_OUT_OF_MEMORY;
+      return -1;
+    }
+    tDecoderClear(&decoder);
+    if (taosHashPut(pTq->pCheckInfo, info.topic, strlen(info.topic), &info, sizeof(STqCheckInfo)) < 0) {
+      terrno = TSDB_CODE_OUT_OF_MEMORY;
+      return -1;
+    }
   }
-  tdbClose(pTq->pMetaStore);
+  tdbTbcClose(pCur);
   return 0;
 }
 
@@ -153,7 +183,7 @@ int32_t tqMetaSaveHandle(STQ* pTq, const char* key, const STqHandle* pHandle) {
     ASSERT(0);
   }
 
-  if (tdbBegin(pTq->pMetaStore, &txn) < 0) {
+  if (tdbBegin(pTq->pMetaDB, &txn) < 0) {
     ASSERT(0);
   }
 
@@ -161,7 +191,7 @@ int32_t tqMetaSaveHandle(STQ* pTq, const char* key, const STqHandle* pHandle) {
     ASSERT(0);
   }
 
-  if (tdbCommit(pTq->pMetaStore, &txn) < 0) {
+  if (tdbCommit(pTq->pMetaDB, &txn) < 0) {
     ASSERT(0);
   }
 
@@ -177,7 +207,7 @@ int32_t tqMetaDeleteHandle(STQ* pTq, const char* key) {
     ASSERT(0);
   }
 
-  if (tdbBegin(pTq->pMetaStore, &txn) < 0) {
+  if (tdbBegin(pTq->pMetaDB, &txn) < 0) {
     ASSERT(0);
   }
 
@@ -185,9 +215,67 @@ int32_t tqMetaDeleteHandle(STQ* pTq, const char* key) {
     /*ASSERT(0);*/
   }
 
-  if (tdbCommit(pTq->pMetaStore, &txn) < 0) {
+  if (tdbCommit(pTq->pMetaDB, &txn) < 0) {
     ASSERT(0);
   }
 
   return 0;
 }
+
+int32_t tqMetaRestoreHandle(STQ* pTq) {
+  TBC* pCur = NULL;
+  if (tdbTbcOpen(pTq->pExecStore, &pCur, NULL) < 0) {
+    ASSERT(0);
+    return -1;
+  }
+
+  void*    pKey = NULL;
+  int      kLen = 0;
+  void*    pVal = NULL;
+  int      vLen = 0;
+  SDecoder decoder;
+
+  tdbTbcMoveToFirst(pCur);
+
+  while (tdbTbcNext(pCur, &pKey, &kLen, &pVal, &vLen) == 0) {
+    STqHandle handle;
+    tDecoderInit(&decoder, (uint8_t*)pVal, vLen);
+    tDecodeSTqHandle(&decoder, &handle);
+
+    handle.pRef = walOpenRef(pTq->pVnode->pWal);
+    if (handle.pRef == NULL) {
+      ASSERT(0);
+      return -1;
+    }
+    walRefVer(handle.pRef, handle.snapshotVer);
+
+    if (handle.execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
+      SReadHandle reader = {
+          .meta = pTq->pVnode->pMeta,
+          .vnode = pTq->pVnode,
+          .initTableReader = true,
+          .initTqReader = true,
+          .version = handle.snapshotVer,
+      };
+
+      handle.execHandle.execCol.task = qCreateQueueExecTaskInfo(
+          handle.execHandle.execCol.qmsg, &reader, &handle.execHandle.numOfCols, &handle.execHandle.pSchemaWrapper);
+      ASSERT(handle.execHandle.execCol.task);
+      void* scanner = NULL;
+      qExtractStreamScanner(handle.execHandle.execCol.task, &scanner);
+      ASSERT(scanner);
+      handle.execHandle.pExecReader = qExtractReaderFromStreamScanner(scanner);
+      ASSERT(handle.execHandle.pExecReader);
+    } else {
+      handle.pWalReader = walOpenReader(pTq->pVnode->pWal, NULL);
+      handle.execHandle.execDb.pFilterOutTbUid =
+          taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
+    }
+    tqDebug("tq restore %s consumer %" PRId64 " vgId:%d", handle.subKey, handle.consumerId, TD_VID(pTq->pVnode));
+    taosHashPut(pTq->pHandle, pKey, kLen, &handle, sizeof(STqHandle));
+  }
+
+  tdbTbcClose(pCur);
+  return 0;
+}
+

source/dnode/vnode/src/tq/tqRead.c

@@ -394,7 +394,7 @@ int tqReaderRemoveTbUidList(STqReader* pReader, const SArray* tbUidList) {
 int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, bool isAdd) {
   void* pIter = NULL;
   while (1) {
-    pIter = taosHashIterate(pTq->handles, pIter);
+    pIter = taosHashIterate(pTq->pHandle, pIter);
     if (pIter == NULL) break;
     STqHandle* pExec = (STqHandle*)pIter;
     if (pExec->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {

source/dnode/vnode/src/tq/tqSnapshot.c

@@ -165,9 +165,9 @@ int32_t tqSnapWriterClose(STqSnapWriter** ppWriter, int8_t rollback) {
   STQ*           pTq = pWriter->pTq;
 
   if (rollback) {
-    ASSERT(0);
+    tdbAbort(pWriter->pTq->pMetaDB, &pWriter->txn);
   } else {
-    code = tdbCommit(pWriter->pTq->pMetaStore, &pWriter->txn);
+    code = tdbCommit(pWriter->pTq->pMetaDB, &pWriter->txn);
     if (code) goto _err;
   }
 

source/dnode/vnode/src/tsdb/tsdbCache.c

@@ -279,6 +279,7 @@ int32_t tsdbCacheInsertLast(SLRUCache *pCache, tb_uid_t uid, STSRow *row, STsdb
       }
     }
 
+  _invalidate:
     taosMemoryFreeClear(pTSchema);
 
     taosLRUCacheRelease(pCache, h, invalidate);
@@ -404,6 +405,178 @@ _err:
   return code;
 }
 
+typedef enum {
+  SFSLASTNEXTROW_FS,
+  SFSLASTNEXTROW_FILESET,
+  SFSLASTNEXTROW_BLOCKDATA,
+  SFSLASTNEXTROW_BLOCKROW
+} SFSLASTNEXTROWSTATES;
+
+typedef struct {
+  SFSLASTNEXTROWSTATES state;         // [input]
+  STsdb               *pTsdb;         // [input]
+  SBlockIdx           *pBlockIdxExp;  // [input]
+  STSchema            *pTSchema;      // [input]
+  int32_t              nFileSet;
+  int32_t              iFileSet;
+  SArray              *aDFileSet;
+  SDataFReader        *pDataFReader;
+  SArray              *aBlockL;
+  SBlockL             *pBlockL;
+  SBlockData          *pBlockDataL;
+  SBlockData           blockDataL;
+  int32_t              nRow;
+  int32_t              iRow;
+  TSDBROW              row;
+  /*
+  SArray    *aBlockIdx;
+  SBlockIdx *pBlockIdx;
+  SMapData   blockMap;
+  int32_t    nBlock;
+  int32_t    iBlock;
+  SBlock     block;
+  */
+} SFSLastNextRowIter;
+
+static int32_t getNextRowFromFSLast(void *iter, TSDBROW **ppRow) {
+  SFSLastNextRowIter *state = (SFSLastNextRowIter *)iter;
+  int32_t             code = 0;
+
+  switch (state->state) {
+    case SFSLASTNEXTROW_FS:
+      // state->aDFileSet = state->pTsdb->pFS->cState->aDFileSet;
+      state->nFileSet = taosArrayGetSize(state->aDFileSet);
+      state->iFileSet = state->nFileSet;
+
+      state->pBlockDataL = NULL;
+
+    case SFSLASTNEXTROW_FILESET: {
+      SDFileSet *pFileSet = NULL;
+    _next_fileset:
+      if (--state->iFileSet >= 0) {
+        pFileSet = (SDFileSet *)taosArrayGet(state->aDFileSet, state->iFileSet);
+      } else {
+        if (state->pBlockDataL) {
+          tBlockDataDestroy(state->pBlockDataL, 1);
+          state->pBlockDataL = NULL;
+        }
+
+        *ppRow = NULL;
+        return code;
+      }
+
+      code = tsdbDataFReaderOpen(&state->pDataFReader, state->pTsdb, pFileSet);
+      if (code) goto _err;
+
+      if (!state->aBlockL) {
+        state->aBlockL = taosArrayInit(0, sizeof(SBlockIdx));
+      } else {
+        taosArrayClear(state->aBlockL);
+      }
+
+      code = tsdbReadBlockL(state->pDataFReader, state->aBlockL);
+      if (code) goto _err;
+
+      // SBlockL *pBlockL = (SBlockL *)taosArrayGet(state->aBlockL, state->iBlockL);
+
+      state->pBlockL = taosArraySearch(state->aBlockL, state->pBlockIdxExp, tCmprBlockL, TD_EQ);
+      if (!state->pBlockL) {
+        goto _next_fileset;
+      }
+
+      int64_t suid = state->pBlockL->suid;
+      int64_t uid = state->pBlockL->maxUid;
+
+      if (!state->pBlockDataL) {
+        state->pBlockDataL = &state->blockDataL;
+      }
+      code = tBlockDataInit(state->pBlockDataL, suid, suid ? 0 : uid, state->pTSchema);
+      if (code) goto _err;
+    }
+    case SFSLASTNEXTROW_BLOCKDATA:
+      code = tsdbReadLastBlock(state->pDataFReader, state->pBlockL, state->pBlockDataL);
+      if (code) goto _err;
+
+      state->nRow = state->blockDataL.nRow;
+      state->iRow = state->nRow - 1;
+
+      if (!state->pBlockDataL->uid) {
+        while (state->pBlockIdxExp->uid != state->pBlockDataL->aUid[state->iRow]) {
+          --state->iRow;
+        }
+      }
+
+      state->state = SFSLASTNEXTROW_BLOCKROW;
+    case SFSLASTNEXTROW_BLOCKROW:
+      if (state->pBlockDataL->uid) {
+        if (state->iRow >= 0) {
+          state->row = tsdbRowFromBlockData(state->pBlockDataL, state->iRow);
+          *ppRow = &state->row;
+
+          if (--state->iRow < 0) {
+            state->state = SFSLASTNEXTROW_FILESET;
+          }
+        }
+      } else {
+        if (state->iRow >= 0 && state->pBlockIdxExp->uid == state->pBlockDataL->aUid[state->iRow]) {
+          state->row = tsdbRowFromBlockData(state->pBlockDataL, state->iRow);
+          *ppRow = &state->row;
+
+          if (--state->iRow < 0 || state->pBlockIdxExp->uid != state->pBlockDataL->aUid[state->iRow]) {
+            state->state = SFSLASTNEXTROW_FILESET;
+          }
+        }
+      }
+
+      return code;
+    default:
+      ASSERT(0);
+      break;
+  }
+
+_err:
+  if (state->pDataFReader) {
+    tsdbDataFReaderClose(&state->pDataFReader);
+    state->pDataFReader = NULL;
+  }
+  if (state->aBlockL) {
+    taosArrayDestroy(state->aBlockL);
+    state->aBlockL = NULL;
+  }
+  if (state->pBlockDataL) {
+    tBlockDataDestroy(state->pBlockDataL, 1);
+    state->pBlockDataL = NULL;
+  }
+
+  *ppRow = NULL;
+
+  return code;
+}
+
+int32_t clearNextRowFromFSLast(void *iter) {
+  SFSLastNextRowIter *state = (SFSLastNextRowIter *)iter;
+  int32_t             code = 0;
+
+  if (!state) {
+    return code;
+  }
+
+  if (state->pDataFReader) {
+    tsdbDataFReaderClose(&state->pDataFReader);
+    state->pDataFReader = NULL;
+  }
+  if (state->aBlockL) {
+    taosArrayDestroy(state->aBlockL);
+    state->aBlockL = NULL;
+  }
+  if (state->pBlockDataL) {
+    tBlockDataDestroy(state->pBlockDataL, 1);
+    state->pBlockDataL = NULL;
+  }
+
+  return code;
+}
+
 typedef enum SFSNEXTROWSTATES {
   SFSNEXTROW_FS,
   SFSNEXTROW_FILESET,
@@ -722,18 +895,19 @@ typedef struct {
   SArray *pSkyline;
   int64_t iSkyline;
 
-  SBlockIdx       idx;
-  SMemNextRowIter memState;
-  SMemNextRowIter imemState;
-  SFSNextRowIter  fsState;
-  TSDBROW         memRow, imemRow, fsRow;
+  SBlockIdx          idx;
+  SMemNextRowIter    memState;
+  SMemNextRowIter    imemState;
+  SFSLastNextRowIter fsLastState;
+  SFSNextRowIter     fsState;
+  TSDBROW            memRow, imemRow, fsLastRow, fsRow;
 
-  TsdbNextRowState input[3];
+  TsdbNextRowState input[4];
   STsdbReadSnap   *pReadSnap;
   STsdb           *pTsdb;
 } CacheNextRowIter;
 
-static int32_t nextRowIterOpen(CacheNextRowIter *pIter, tb_uid_t uid, STsdb *pTsdb) {
+static int32_t nextRowIterOpen(CacheNextRowIter *pIter, tb_uid_t uid, STsdb *pTsdb, STSchema *pTSchema) {
   int code = 0;
 
   tb_uid_t suid = getTableSuidByUid(uid, pTsdb);
@@ -779,6 +953,12 @@ static int32_t nextRowIterOpen(CacheNextRowIter *pIter, tb_uid_t uid, STsdb *pTs
 
   pIter->idx = (SBlockIdx){.suid = suid, .uid = uid};
 
+  pIter->fsLastState.state = SFSNEXTROW_FS;
+  pIter->fsLastState.pTsdb = pTsdb;
+  pIter->fsLastState.aDFileSet = pIter->pReadSnap->fs.aDFileSet;
+  pIter->fsLastState.pBlockIdxExp = &pIter->idx;
+  pIter->fsLastState.pTSchema = pTSchema;
+
   pIter->fsState.state = SFSNEXTROW_FS;
   pIter->fsState.pTsdb = pTsdb;
   pIter->fsState.aDFileSet = pIter->pReadSnap->fs.aDFileSet;
@@ -786,7 +966,9 @@ static int32_t nextRowIterOpen(CacheNextRowIter *pIter, tb_uid_t uid, STsdb *pTs
 
   pIter->input[0] = (TsdbNextRowState){&pIter->memRow, true, false, &pIter->memState, getNextRowFromMem, NULL};
   pIter->input[1] = (TsdbNextRowState){&pIter->imemRow, true, false, &pIter->imemState, getNextRowFromMem, NULL};
-  pIter->input[2] =
+  pIter->input[2] = (TsdbNextRowState){&pIter->fsLastRow,     false, true, &pIter->fsLastState, getNextRowFromFSLast,
+                                       clearNextRowFromFSLast};
+  pIter->input[3] =
       (TsdbNextRowState){&pIter->fsRow, false, true, &pIter->fsState, getNextRowFromFS, clearNextRowFromFS};
 
   if (pMem) {
@@ -811,7 +993,7 @@ _err:
 static int32_t nextRowIterClose(CacheNextRowIter *pIter) {
   int code = 0;
 
-  for (int i = 0; i < 3; ++i) {
+  for (int i = 0; i < 4; ++i) {
     if (pIter->input[i].nextRowClearFn) {
       pIter->input[i].nextRowClearFn(pIter->input[i].iter);
     }
@@ -823,7 +1005,6 @@ static int32_t nextRowIterClose(CacheNextRowIter *pIter) {
 
   tsdbUntakeReadSnap(pIter->pTsdb, pIter->pReadSnap);
 
-  return code;
 _err:
   return code;
 }
@@ -832,7 +1013,7 @@ _err:
 static int32_t nextRowIterGet(CacheNextRowIter *pIter, TSDBROW **ppRow) {
   int code = 0;
 
-  for (int i = 0; i < 3; ++i) {
+  for (int i = 0; i < 4; ++i) {
     if (pIter->input[i].next && !pIter->input[i].stop) {
       code = pIter->input[i].nextRowFn(pIter->input[i].iter, &pIter->input[i].pRow);
       if (code) goto _err;
@@ -844,18 +1025,18 @@ static int32_t nextRowIterGet(CacheNextRowIter *pIter, TSDBROW **ppRow) {
     }
   }
 
-  if (pIter->input[0].stop && pIter->input[1].stop && pIter->input[2].stop) {
+  if (pIter->input[0].stop && pIter->input[1].stop && pIter->input[2].stop && pIter->input[3].stop) {
     *ppRow = NULL;
     return code;
   }
 
-  // select maxpoint(s) from mem, imem, fs
-  TSDBROW *max[3] = {0};
-  int      iMax[3] = {-1, -1, -1};
+  // select maxpoint(s) from mem, imem, fs and last
+  TSDBROW *max[4] = {0};
+  int      iMax[4] = {-1, -1, -1, -1};
   int      nMax = 0;
   TSKEY    maxKey = TSKEY_MIN;
 
-  for (int i = 0; i < 3; ++i) {
+  for (int i = 0; i < 4; ++i) {
     if (!pIter->input[i].stop && pIter->input[i].pRow != NULL) {
       TSDBKEY key = TSDBROW_KEY(pIter->input[i].pRow);
 
@@ -873,8 +1054,8 @@ static int32_t nextRowIterGet(CacheNextRowIter *pIter, TSDBROW **ppRow) {
   }
 
   // delete detection
-  TSDBROW *merge[3] = {0};
-  int      iMerge[3] = {-1, -1, -1};
+  TSDBROW *merge[4] = {0};
+  int      iMerge[4] = {-1, -1, -1, -1};
   int      nMerge = 0;
   for (int i = 0; i < nMax; ++i) {
     TSDBKEY maxKey = TSDBROW_KEY(max[i]);
@@ -915,7 +1096,7 @@ static int32_t mergeLastRow(tb_uid_t uid, STsdb *pTsdb, bool *dup, STSRow **ppRo
   TSKEY lastRowTs = TSKEY_MAX;
 
   CacheNextRowIter iter = {0};
-  nextRowIterOpen(&iter, uid, pTsdb);
+  nextRowIterOpen(&iter, uid, pTsdb, pTSchema);
 
   do {
     TSDBROW *pRow = NULL;
@@ -1012,7 +1193,7 @@ static int32_t mergeLast(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray) {
   TSKEY lastRowTs = TSKEY_MAX;
 
   CacheNextRowIter iter = {0};
-  nextRowIterOpen(&iter, uid, pTsdb);
+  nextRowIterOpen(&iter, uid, pTsdb, pTSchema);
 
   do {
     TSDBROW *pRow = NULL;

source/dnode/vnode/src/tsdb/tsdbCommit.c

@@ -290,23 +290,19 @@ _err:
   return code;
 }
 
-static int32_t tsdbCommitterUpdateTableSchema(SCommitter *pCommitter, int64_t suid, int64_t uid, int32_t sver) {
+static int32_t tsdbCommitterUpdateTableSchema(SCommitter *pCommitter, int64_t suid, int64_t uid) {
   int32_t code = 0;
 
-  if (pCommitter->skmTable.pTSchema) {
-    if (pCommitter->skmTable.suid == suid) {
-      if (suid == 0) {
-        if (pCommitter->skmTable.uid == uid && sver == pCommitter->skmTable.pTSchema->version) goto _exit;
-      } else {
-        if (sver == pCommitter->skmTable.pTSchema->version) goto _exit;
-      }
-    }
+  if (suid) {
+    if (pCommitter->skmTable.suid == suid) goto _exit;
+  } else {
+    if (pCommitter->skmTable.uid == uid) goto _exit;
   }
 
   pCommitter->skmTable.suid = suid;
   pCommitter->skmTable.uid = uid;
   tTSchemaDestroy(pCommitter->skmTable.pTSchema);
-  code = metaGetTbTSchemaEx(pCommitter->pTsdb->pVnode->pMeta, suid, uid, sver, &pCommitter->skmTable.pTSchema);
+  code = metaGetTbTSchemaEx(pCommitter->pTsdb->pVnode->pMeta, suid, uid, -1, &pCommitter->skmTable.pTSchema);
   if (code) goto _exit;
 
 _exit:
@@ -360,7 +356,7 @@ static int32_t tsdbCommitterNextLastRow(SCommitter *pCommitter) {
       int64_t  suid = pBlockL->suid;
       int64_t  uid = pBlockL->maxUid;
 
-      code = tsdbCommitterUpdateTableSchema(pCommitter, suid, uid, 1 /*TODO*/);
+      code = tsdbCommitterUpdateTableSchema(pCommitter, suid, uid);
       if (code) goto _exit;
 
       code = tBlockDataInit(pBlockDatal, suid, suid ? 0 : uid, pCommitter->skmTable.pTSchema);
@@ -978,7 +974,7 @@ static int32_t tsdbCommitTableData(SCommitter *pCommitter, STbData *pTbData) {
     pBlock = NULL;
   }
 
-  code = tsdbCommitterUpdateTableSchema(pCommitter, pTbData->suid, pTbData->uid, pTbData->maxSkmVer /*TODO*/);
+  code = tsdbCommitterUpdateTableSchema(pCommitter, pTbData->suid, pTbData->uid);
   if (code) goto _err;
 
   tMapDataReset(&pCommitter->dWriter.mBlock);
@@ -1160,7 +1156,7 @@ static int32_t tsdbMoveCommitData(SCommitter *pCommitter, TABLEID toTable) {
 
     // set block data schema if need
     if (pBlockDataW->suid == 0 && pBlockDataW->uid == 0) {
-      code = tsdbCommitterUpdateTableSchema(pCommitter, suid, uid, 1 /*TOOD*/);
+      code = tsdbCommitterUpdateTableSchema(pCommitter, suid, uid);
       if (code) goto _err;
 
       code = tBlockDataInit(pBlockDataW, suid, suid ? 0 : uid, pCommitter->skmTable.pTSchema);

source/dnode/vnode/src/tsdb/tsdbMemTable.c

@@ -46,8 +46,6 @@ int32_t tsdbMemTableCreate(STsdb *pTsdb, SMemTable **ppMemTable) {
   pMemTable->nRef = 1;
   pMemTable->minKey = TSKEY_MAX;
   pMemTable->maxKey = TSKEY_MIN;
-  pMemTable->minVersion = VERSION_MAX;
-  pMemTable->maxVersion = VERSION_MIN;
   pMemTable->nRow = 0;
   pMemTable->nDel = 0;
   pMemTable->nTbData = 0;
@@ -180,10 +178,6 @@ int32_t tsdbDeleteTableData(STsdb *pTsdb, int64_t version, tb_uid_t suid, tb_uid
     pTbData->pTail = pDelData;
   }
 
-  // update the state of pMemTable and other (todo)
-
-  pMemTable->minVersion = TMIN(pMemTable->minVersion, version);
-  pMemTable->maxVersion = TMAX(pMemTable->maxVersion, version);
   pMemTable->nDel++;
 
   if (TSDB_CACHE_LAST_ROW(pMemTable->pTsdb->pVnode->config) && tsdbKeyCmprFn(&lastKey, &pTbData->maxKey) >= 0) {
@@ -368,9 +362,6 @@ static int32_t tsdbGetOrCreateTbData(SMemTable *pMemTable, tb_uid_t suid, tb_uid
   pTbData->uid = uid;
   pTbData->minKey = TSKEY_MAX;
   pTbData->maxKey = TSKEY_MIN;
-  pTbData->minVersion = VERSION_MAX;
-  pTbData->maxVersion = VERSION_MIN;
-  pTbData->maxSkmVer = -1;
   pTbData->pHead = NULL;
   pTbData->pTail = NULL;
   pTbData->sl.seed = taosRand();
@@ -615,15 +606,9 @@ static int32_t tsdbInsertTableDataImpl(SMemTable *pMemTable, STbData *pTbData, i
     tsdbCacheInsertLast(pMemTable->pTsdb->lruCache, pTbData->uid, pLastRow, pMemTable->pTsdb);
   }
 
-  pTbData->minVersion = TMIN(pTbData->minVersion, version);
-  pTbData->maxVersion = TMAX(pTbData->maxVersion, version);
-  pTbData->maxSkmVer = TMAX(pTbData->maxSkmVer, pMsgIter->sversion);
-
   // SMemTable
   pMemTable->minKey = TMIN(pMemTable->minKey, pTbData->minKey);
   pMemTable->maxKey = TMAX(pMemTable->maxKey, pTbData->maxKey);
-  pMemTable->minVersion = TMIN(pMemTable->minVersion, pTbData->minVersion);
-  pMemTable->maxVersion = TMAX(pMemTable->maxVersion, pTbData->maxVersion);
   pMemTable->nRow += nRow;
 
   pRsp->numOfRows = nRow;

source/dnode/vnode/src/tsdb/tsdbUtil.c

@@ -196,6 +196,25 @@ int32_t tCmprBlockIdx(void const *lhs, void const *rhs) {
   return 0;
 }
 
+int32_t tCmprBlockL(void const *lhs, void const *rhs) {
+  SBlockIdx *lBlockIdx = (SBlockIdx *)lhs;
+  SBlockL   *rBlockL = (SBlockL *)rhs;
+
+  if (lBlockIdx->suid < rBlockL->suid) {
+    return -1;
+  } else if (lBlockIdx->suid > rBlockL->suid) {
+    return 1;
+  }
+
+  if (lBlockIdx->uid < rBlockL->minUid) {
+    return -1;
+  } else if (lBlockIdx->uid > rBlockL->maxUid) {
+    return 1;
+  }
+
+  return 0;
+}
+
 // SBlock ======================================================
 void tBlockReset(SBlock *pBlock) {
   *pBlock = (SBlock){.minKey = TSDBKEY_MAX, .maxKey = TSDBKEY_MIN, .minVer = VERSION_MAX, .maxVer = VERSION_MIN};
@@ -1944,4 +1963,4 @@ int32_t tsdbReadAndCheck(TdFilePtr pFD, int64_t offset, uint8_t **ppOut, int32_t
 
 _exit:
   return code;
-}
+}
\ No newline at end of file

source/dnode/vnode/src/vnd/vnodeSvr.c

@@ -196,36 +196,42 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRp
       break;
     /* TQ */
     case TDMT_VND_MQ_VG_CHANGE:
-      if (tqProcessVgChangeReq(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
+      if (tqProcessVgChangeReq(pVnode->pTq, version, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
                                pMsg->contLen - sizeof(SMsgHead)) < 0) {
         goto _err;
       }
       break;
     case TDMT_VND_MQ_VG_DELETE:
-      if (tqProcessVgDeleteReq(pVnode->pTq, pMsg->pCont, pMsg->contLen) < 0) {
+      if (tqProcessVgDeleteReq(pVnode->pTq, version, pMsg->pCont, pMsg->contLen) < 0) {
         goto _err;
       }
       break;
     case TDMT_VND_MQ_COMMIT_OFFSET:
-      if (tqProcessOffsetCommitReq(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
-                                   pMsg->contLen - sizeof(SMsgHead), version) < 0) {
+      if (tqProcessOffsetCommitReq(pVnode->pTq, version, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
+                                   pMsg->contLen - sizeof(SMsgHead)) < 0) {
         goto _err;
       }
       break;
-    case TDMT_VND_CHECK_ALTER_INFO:
-      if (tqProcessCheckAlterInfoReq(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
-                                     pMsg->contLen - sizeof(SMsgHead)) < 0) {
+    case TDMT_VND_ADD_CHECK_INFO:
+      if (tqProcessAddCheckInfoReq(pVnode->pTq, version, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
+                                   pMsg->contLen - sizeof(SMsgHead)) < 0) {
+        goto _err;
+      }
+      break;
+    case TDMT_VND_DELETE_CHECK_INFO:
+      if (tqProcessDelCheckInfoReq(pVnode->pTq, version, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
+                                   pMsg->contLen - sizeof(SMsgHead)) < 0) {
         goto _err;
       }
       break;
     case TDMT_STREAM_TASK_DEPLOY: {
-      if (tqProcessTaskDeployReq(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
+      if (tqProcessTaskDeployReq(pVnode->pTq, version, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
                                  pMsg->contLen - sizeof(SMsgHead)) < 0) {
         goto _err;
       }
     } break;
     case TDMT_STREAM_TASK_DROP: {
-      if (tqProcessTaskDropReq(pVnode->pTq, pMsg->pCont, pMsg->contLen) < 0) {
+      if (tqProcessTaskDropReq(pVnode->pTq, version, pMsg->pCont, pMsg->contLen) < 0) {
         goto _err;
       }
     } break;

source/libs/function/src/builtins.c

@@ -192,6 +192,24 @@ static bool validateTimezoneFormat(const SValueNode* pVal) {
   return true;
 }
 
+static int32_t countTrailingSpaces(const SValueNode* pVal, bool isLtrim) {
+  int32_t numOfSpaces = 0;
+  int32_t len = varDataLen(pVal->datum.p);
+  char*   str = varDataVal(pVal->datum.p);
+
+  int32_t startPos = isLtrim ? 0 : len - 1;
+  int32_t step = isLtrim ? 1 : -1;
+  for (int32_t i = startPos; i < len || i >= 0; i += step) {
+    if (!isspace(str[i])) {
+      break;
+    }
+    numOfSpaces++;
+  }
+
+  return numOfSpaces;
+
+}
+
 void static addTimezoneParam(SNodeList* pList) {
   char      buf[6] = {0};
   time_t    t = taosTime(NULL);
@@ -293,6 +311,40 @@ static int32_t translateInOutStr(SFunctionNode* pFunc, char* pErrBuf, int32_t le
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t translateTrimStr(SFunctionNode* pFunc, char* pErrBuf, int32_t len, bool isLtrim) {
+  if (1 != LIST_LENGTH(pFunc->pParameterList)) {
+    return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
+  }
+
+  SExprNode* pPara1 = (SExprNode*)nodesListGetNode(pFunc->pParameterList, 0);
+  if (!IS_VAR_DATA_TYPE(pPara1->resType.type)) {
+    return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
+  }
+
+  int32_t numOfSpaces = 0;
+  SNode* pParamNode1 = nodesListGetNode(pFunc->pParameterList, 0);
+  // for select trim functions with constant value from table,
+  // need to set the proper result result schema bytes to avoid
+  // trailing garbage characters
+  if (nodeType(pParamNode1) == QUERY_NODE_VALUE) {
+    SValueNode* pValue = (SValueNode*)pParamNode1;
+    numOfSpaces = countTrailingSpaces(pValue, isLtrim);
+  }
+
+
+  int32_t resBytes = pPara1->resType.bytes - numOfSpaces;
+  pFunc->node.resType = (SDataType){.bytes = resBytes, .type = pPara1->resType.type};
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t translateLtrim(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
+  return translateTrimStr(pFunc, pErrBuf, len, true);
+}
+
+static int32_t translateRtrim(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
+  return translateTrimStr(pFunc, pErrBuf, len, false);
+}
+
 static int32_t translateLogarithm(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
   int32_t numOfParams = LIST_LENGTH(pFunc->pParameterList);
   if (1 != numOfParams && 2 != numOfParams) {
@@ -2827,7 +2879,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .name = "ltrim",
     .type = FUNCTION_TYPE_LTRIM,
     .classification = FUNC_MGT_SCALAR_FUNC | FUNC_MGT_STRING_FUNC,
-    .translateFunc = translateInOutStr,
+    .translateFunc = translateLtrim,
     .getEnvFunc   = NULL,
     .initFunc     = NULL,
     .sprocessFunc = ltrimFunction,
@@ -2837,7 +2889,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .name = "rtrim",
     .type = FUNCTION_TYPE_RTRIM,
     .classification = FUNC_MGT_SCALAR_FUNC | FUNC_MGT_STRING_FUNC,
-    .translateFunc = translateInOutStr,
+    .translateFunc = translateRtrim,
     .getEnvFunc   = NULL,
     .initFunc     = NULL,
     .sprocessFunc = rtrimFunction,

source/libs/scalar/src/scalar.c

@@ -758,7 +758,9 @@ EDealRes sclRewriteFunction(SNode** pNode, SScalarCtx *ctx) {
       res->datum.p = taosMemoryCalloc(len, 1);
       memcpy(res->datum.p, output.columnData->pData, len);
     } else if (IS_VAR_DATA_TYPE(type)) {
-      res->datum.p = taosMemoryCalloc(res->node.resType.bytes + VARSTR_HEADER_SIZE + 1, 1);
+      //res->datum.p = taosMemoryCalloc(res->node.resType.bytes + VARSTR_HEADER_SIZE + 1, 1);
+      res->datum.p = taosMemoryCalloc(varDataTLen(output.columnData->pData), 1);
+      res->node.resType.bytes = varDataTLen(output.columnData->pData);
       memcpy(res->datum.p, output.columnData->pData, varDataTLen(output.columnData->pData));
     } else {
       nodesSetValueNodeValue(res, output.columnData->pData);

source/libs/stream/src/streamMeta.c

@@ -81,7 +81,7 @@ void streamMetaClose(SStreamMeta* pMeta) {
   taosMemoryFree(pMeta);
 }
 
-int32_t streamMetaAddSerializedTask(SStreamMeta* pMeta, char* msg, int32_t msgLen) {
+int32_t streamMetaAddSerializedTask(SStreamMeta* pMeta, int64_t startVer, char* msg, int32_t msgLen) {
   SStreamTask* pTask = taosMemoryCalloc(1, sizeof(SStreamTask));
   if (pTask == NULL) {
     return -1;

source/libs/sync/inc/syncIndexMgr.h

@@ -29,8 +29,12 @@ extern "C" {
 // SIndexMgr -----------------------------
 typedef struct SSyncIndexMgr {
   SRaftId (*replicas)[TSDB_MAX_REPLICA];
-  SyncIndex  index[TSDB_MAX_REPLICA];
-  SyncTerm   privateTerm[TSDB_MAX_REPLICA];  // for advanced function
+  SyncIndex index[TSDB_MAX_REPLICA];
+  SyncTerm  privateTerm[TSDB_MAX_REPLICA];  // for advanced function
+
+  int64_t startTimeArr[TSDB_MAX_REPLICA];
+  int64_t recvTimeArr[TSDB_MAX_REPLICA];
+
   int32_t    replicaNum;
   SSyncNode *pSyncNode;
 } SSyncIndexMgr;
@@ -41,8 +45,13 @@ void           syncIndexMgrDestroy(SSyncIndexMgr *pSyncIndexMgr);
 void           syncIndexMgrClear(SSyncIndexMgr *pSyncIndexMgr);
 void           syncIndexMgrSetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, SyncIndex index);
 SyncIndex      syncIndexMgrGetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId);
-cJSON *        syncIndexMgr2Json(SSyncIndexMgr *pSyncIndexMgr);
-char *         syncIndexMgr2Str(SSyncIndexMgr *pSyncIndexMgr);
+cJSON         *syncIndexMgr2Json(SSyncIndexMgr *pSyncIndexMgr);
+char          *syncIndexMgr2Str(SSyncIndexMgr *pSyncIndexMgr);
+
+void    syncIndexMgrSetStartTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, int64_t startTime);
+int64_t syncIndexMgrGetStartTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId);
+void    syncIndexMgrSetRecvTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, int64_t recvTime);
+int64_t syncIndexMgrGetRecvTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId);
 
 // void     syncIndexMgrSetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, SyncTerm term);
 // SyncTerm syncIndexMgrGetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId);

source/libs/sync/inc/syncInt.h

@@ -269,6 +269,8 @@ int32_t syncNodeLeaderTransfer(SSyncNode* pSyncNode);
 int32_t syncNodeLeaderTransferTo(SSyncNode* pSyncNode, SNodeInfo newLeader);
 int32_t syncDoLeaderTransfer(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftEntry* pEntry);
 
+int32_t syncNodeDynamicQuorum(const SSyncNode* pSyncNode);
+
 // trace log
 void syncLogSendRequestVote(SSyncNode* pSyncNode, const SyncRequestVote* pMsg, const char* s);
 void syncLogRecvRequestVote(SSyncNode* pSyncNode, const SyncRequestVote* pMsg, const char* s);

source/libs/sync/inc/syncReplication.h

@@ -55,6 +55,8 @@ int32_t syncNodeAppendEntriesPeers(SSyncNode* pSyncNode);
 int32_t syncNodeAppendEntriesPeersSnapshot(SSyncNode* pSyncNode);
 int32_t syncNodeAppendEntriesPeersSnapshot2(SSyncNode* pSyncNode);
 
+int32_t syncNodeAppendEntriesOnePeer(SSyncNode* pSyncNode, SRaftId* pDestId, SyncIndex nextIndex);
+
 int32_t syncNodeReplicate(SSyncNode* pSyncNode, bool isTimer);
 int32_t syncNodeAppendEntries(SSyncNode* pSyncNode, const SRaftId* destRaftId, const SyncAppendEntries* pMsg);
 int32_t syncNodeAppendEntriesBatch(SSyncNode* pSyncNode, const SRaftId* destRaftId, const SyncAppendEntriesBatch* pMsg);

source/libs/sync/src/syncAppendEntries.c

@@ -148,6 +148,7 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
     pReply->term = ths->pRaftStore->currentTerm;
     pReply->success = false;
     pReply->matchIndex = SYNC_INDEX_INVALID;
+    pReply->startTime = ths->startTime;
 
     // msg event log
     syncLogSendAppendEntriesReply(ths, pReply, "");
@@ -290,6 +291,8 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
       pReply->matchIndex = pMsg->prevLogIndex;
     }
 
+    pReply->startTime = ths->startTime;
+
     // msg event log
     syncLogSendAppendEntriesReply(ths, pReply, "");
 
@@ -603,6 +606,7 @@ int32_t syncNodeOnAppendEntriesSnapshot2Cb(SSyncNode* ths, SyncAppendEntriesBatc
       pReply->privateTerm = ths->pNewNodeReceiver->privateTerm;
       pReply->success = true;
       pReply->matchIndex = matchIndex;
+      pReply->startTime = ths->startTime;
 
       // msg event log
       syncLogSendAppendEntriesReply(ths, pReply, "");
@@ -651,6 +655,7 @@ int32_t syncNodeOnAppendEntriesSnapshot2Cb(SSyncNode* ths, SyncAppendEntriesBatc
       pReply->privateTerm = ths->pNewNodeReceiver->privateTerm;
       pReply->success = false;
       pReply->matchIndex = ths->commitIndex;
+      pReply->startTime = ths->startTime;
 
       // msg event log
       syncLogSendAppendEntriesReply(ths, pReply, "");
@@ -729,6 +734,7 @@ int32_t syncNodeOnAppendEntriesSnapshot2Cb(SSyncNode* ths, SyncAppendEntriesBatc
       pReply->privateTerm = ths->pNewNodeReceiver->privateTerm;
       pReply->success = true;
       pReply->matchIndex = hasAppendEntries ? pMsg->prevLogIndex + pMsg->dataCount : pMsg->prevLogIndex;
+      pReply->startTime = ths->startTime;
 
       // msg event log
       syncLogSendAppendEntriesReply(ths, pReply, "");
@@ -874,6 +880,7 @@ int32_t syncNodeOnAppendEntriesSnapshotCb(SSyncNode* ths, SyncAppendEntries* pMs
       pReply->privateTerm = ths->pNewNodeReceiver->privateTerm;
       pReply->success = true;
       pReply->matchIndex = matchIndex;
+      pReply->startTime = ths->startTime;
 
       // msg event log
       syncLogSendAppendEntriesReply(ths, pReply, "");
@@ -919,6 +926,7 @@ int32_t syncNodeOnAppendEntriesSnapshotCb(SSyncNode* ths, SyncAppendEntries* pMs
       pReply->privateTerm = ths->pNewNodeReceiver->privateTerm;
       pReply->success = false;
       pReply->matchIndex = SYNC_INDEX_INVALID;
+      pReply->startTime = ths->startTime;
 
       // msg event log
       syncLogSendAppendEntriesReply(ths, pReply, "");
@@ -984,6 +992,7 @@ int32_t syncNodeOnAppendEntriesSnapshotCb(SSyncNode* ths, SyncAppendEntries* pMs
       pReply->privateTerm = ths->pNewNodeReceiver->privateTerm;
       pReply->success = true;
       pReply->matchIndex = hasAppendEntries ? pMsg->prevLogIndex + 1 : pMsg->prevLogIndex;
+      pReply->startTime = ths->startTime;
 
       // msg event log
       syncLogSendAppendEntriesReply(ths, pReply, "");

source/libs/sync/src/syncAppendEntriesReply.c

@@ -64,6 +64,10 @@ int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* p
 
   ASSERT(pMsg->term == ths->pRaftStore->currentTerm);
 
+  // update time
+  syncIndexMgrSetStartTime(ths->pNextIndex, &(pMsg->srcId), pMsg->startTime);
+  syncIndexMgrSetRecvTime(ths->pNextIndex, &(pMsg->srcId), taosGetTimestampMs());
+
   SyncIndex beforeNextIndex = syncIndexMgrGetIndex(ths->pNextIndex, &(pMsg->srcId));
   SyncIndex beforeMatchIndex = syncIndexMgrGetIndex(ths->pMatchIndex, &(pMsg->srcId));
 
@@ -170,6 +174,10 @@ int32_t syncNodeOnAppendEntriesReplySnapshot2Cb(SSyncNode* ths, SyncAppendEntrie
 
   ASSERT(pMsg->term == ths->pRaftStore->currentTerm);
 
+  // update time
+  syncIndexMgrSetStartTime(ths->pNextIndex, &(pMsg->srcId), pMsg->startTime);
+  syncIndexMgrSetRecvTime(ths->pNextIndex, &(pMsg->srcId), taosGetTimestampMs());
+
   SyncIndex beforeNextIndex = syncIndexMgrGetIndex(ths->pNextIndex, &(pMsg->srcId));
   SyncIndex beforeMatchIndex = syncIndexMgrGetIndex(ths->pMatchIndex, &(pMsg->srcId));
 
@@ -330,6 +338,10 @@ int32_t syncNodeOnAppendEntriesReplySnapshotCb(SSyncNode* ths, SyncAppendEntries
 
   ASSERT(pMsg->term == ths->pRaftStore->currentTerm);
 
+  // update time
+  syncIndexMgrSetStartTime(ths->pNextIndex, &(pMsg->srcId), pMsg->startTime);
+  syncIndexMgrSetRecvTime(ths->pNextIndex, &(pMsg->srcId), taosGetTimestampMs());
+
   SyncIndex beforeNextIndex = syncIndexMgrGetIndex(ths->pNextIndex, &(pMsg->srcId));
   SyncIndex beforeMatchIndex = syncIndexMgrGetIndex(ths->pMatchIndex, &(pMsg->srcId));
 

source/libs/sync/src/syncCommit.c

@@ -133,6 +133,63 @@ bool syncAgreeIndex(SSyncNode* pSyncNode, SRaftId* pRaftId, SyncIndex index) {
   return false;
 }
 
+static inline int64_t syncNodeAbs64(int64_t a, int64_t b) {
+  ASSERT(a >= 0);
+  ASSERT(b >= 0);
+
+  int64_t c = a > b ? a - b : b - a;
+  return c;
+}
+
+int32_t syncNodeDynamicQuorum(const SSyncNode* pSyncNode) {
+  int32_t quorum = 1;  // self
+
+  int64_t timeNow = taosGetTimestampMs();
+  for (int i = 0; i < pSyncNode->peersNum; ++i) {
+    int64_t peerStartTime = syncIndexMgrGetStartTime(pSyncNode->pNextIndex, &(pSyncNode->peersId)[i]);
+    int64_t peerRecvTime = syncIndexMgrGetRecvTime(pSyncNode->pNextIndex, &(pSyncNode->peersId)[i]);
+
+    int64_t recvTimeDiff = syncNodeAbs64(peerRecvTime, timeNow);
+    int64_t startTimeDiff = syncNodeAbs64(peerStartTime, pSyncNode->startTime);
+
+    int32_t addQuorum = 0;
+
+    if (recvTimeDiff < SYNC_MAX_RECV_TIME_RANGE_MS) {
+      addQuorum = 1;
+    } else {
+      addQuorum = 0;
+    }
+
+    if (startTimeDiff > SYNC_MAX_START_TIME_RANGE_MS) {
+      addQuorum = 0;
+    }
+
+    quorum += addQuorum;
+  }
+
+  ASSERT(quorum <= pSyncNode->replicaNum);
+
+  if (quorum < pSyncNode->quorum) {
+    quorum = pSyncNode->quorum;
+  }
+
+  return quorum;
+}
+
+bool syncAgree(SSyncNode* pSyncNode, SyncIndex index) {
+  int agreeCount = 0;
+  for (int i = 0; i < pSyncNode->replicaNum; ++i) {
+    if (syncAgreeIndex(pSyncNode, &(pSyncNode->replicasId[i]), index)) {
+      ++agreeCount;
+    }
+    if (agreeCount >= syncNodeDynamicQuorum(pSyncNode)) {
+      return true;
+    }
+  }
+  return false;
+}
+
+/*
 bool syncAgree(SSyncNode* pSyncNode, SyncIndex index) {
   int agreeCount = 0;
   for (int i = 0; i < pSyncNode->replicaNum; ++i) {
@@ -145,3 +202,4 @@ bool syncAgree(SSyncNode* pSyncNode, SyncIndex index) {
   }
   return false;
 }
+*/

source/libs/sync/src/syncIndexMgr.c

@@ -47,6 +47,13 @@ void syncIndexMgrDestroy(SSyncIndexMgr *pSyncIndexMgr) {
 void syncIndexMgrClear(SSyncIndexMgr *pSyncIndexMgr) {
   memset(pSyncIndexMgr->index, 0, sizeof(pSyncIndexMgr->index));
   memset(pSyncIndexMgr->privateTerm, 0, sizeof(pSyncIndexMgr->privateTerm));
+
+  // int64_t timeNow = taosGetMonotonicMs();
+  for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
+    pSyncIndexMgr->startTimeArr[i] = 0;
+    pSyncIndexMgr->recvTimeArr[i] = 0;
+  }
+
   /*
   for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
     pSyncIndexMgr->index[i] = 0;
@@ -68,7 +75,8 @@ void syncIndexMgrSetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId,
   char     host[128];
   uint16_t port;
   syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port);
-  sError("vgId:%d, index mgr set for %s:%d, index:%" PRId64 " error", pSyncIndexMgr->pSyncNode->vgId, host, port, index);
+  sError("vgId:%d, index mgr set for %s:%d, index:%" PRId64 " error", pSyncIndexMgr->pSyncNode->vgId, host, port,
+         index);
 }
 
 SyncIndex syncIndexMgrGetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId) {
@@ -125,11 +133,65 @@ cJSON *syncIndexMgr2Json(SSyncIndexMgr *pSyncIndexMgr) {
 
 char *syncIndexMgr2Str(SSyncIndexMgr *pSyncIndexMgr) {
   cJSON *pJson = syncIndexMgr2Json(pSyncIndexMgr);
-  char * serialized = cJSON_Print(pJson);
+  char  *serialized = cJSON_Print(pJson);
   cJSON_Delete(pJson);
   return serialized;
 }
 
+void syncIndexMgrSetStartTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, int64_t startTime) {
+  for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
+    if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) {
+      (pSyncIndexMgr->startTimeArr)[i] = startTime;
+      return;
+    }
+  }
+
+  // maybe config change
+  // ASSERT(0);
+  char     host[128];
+  uint16_t port;
+  syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port);
+  sError("vgId:%d, index mgr set for %s:%d, start-time:%" PRId64 " error", pSyncIndexMgr->pSyncNode->vgId, host, port,
+         startTime);
+}
+
+int64_t syncIndexMgrGetStartTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId) {
+  for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
+    if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) {
+      int64_t startTime = (pSyncIndexMgr->startTimeArr)[i];
+      return startTime;
+    }
+  }
+  ASSERT(0);
+}
+
+void syncIndexMgrSetRecvTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, int64_t recvTime) {
+  for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
+    if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) {
+      (pSyncIndexMgr->recvTimeArr)[i] = recvTime;
+      return;
+    }
+  }
+
+  // maybe config change
+  // ASSERT(0);
+  char     host[128];
+  uint16_t port;
+  syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port);
+  sError("vgId:%d, index mgr set for %s:%d, recv-time:%" PRId64 " error", pSyncIndexMgr->pSyncNode->vgId, host, port,
+         recvTime);
+}
+
+int64_t syncIndexMgrGetRecvTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId) {
+  for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
+    if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) {
+      int64_t recvTime = (pSyncIndexMgr->recvTimeArr)[i];
+      return recvTime;
+    }
+  }
+  ASSERT(0);
+}
+
 // for debug -------------------
 void syncIndexMgrPrint(SSyncIndexMgr *pObj) {
   char *serialized = syncIndexMgr2Str(pObj);

source/libs/sync/src/syncMain.c

@@ -1682,13 +1682,13 @@ inline void syncNodeEventLog(const SSyncNode* pSyncNode, char* str) {
                ", sby:%d, "
                "stgy:%d, bch:%d, "
                "r-num:%d, "
-               "lcfg:%" PRId64 ", chging:%d, rsto:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s",
+               "lcfg:%" PRId64 ", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s",
                pSyncNode->vgId, syncUtilState2String(pSyncNode->state), str, pSyncNode->pRaftStore->currentTerm,
                pSyncNode->commitIndex, logBeginIndex, logLastIndex, snapshot.lastApplyIndex, snapshot.lastApplyTerm,
                pSyncNode->pRaftCfg->isStandBy, pSyncNode->pRaftCfg->snapshotStrategy, pSyncNode->pRaftCfg->batchSize,
                pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex, pSyncNode->changing,
-               pSyncNode->restoreFinish, pSyncNode->electTimerLogicClockUser, pSyncNode->heartbeatTimerLogicClockUser,
-               printStr);
+               pSyncNode->restoreFinish, syncNodeDynamicQuorum(pSyncNode), pSyncNode->electTimerLogicClockUser,
+               pSyncNode->heartbeatTimerLogicClockUser, printStr);
     } else {
       snprintf(logBuf, sizeof(logBuf), "%s", str);
     }
@@ -1706,12 +1706,13 @@ inline void syncNodeEventLog(const SSyncNode* pSyncNode, char* str) {
                ", sby:%d, "
                "stgy:%d, bch:%d, "
                "r-num:%d, "
-               "lcfg:%" PRId64 ", chging:%d, rsto:%d, %s",
+               "lcfg:%" PRId64 ", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s",
                pSyncNode->vgId, syncUtilState2String(pSyncNode->state), str, pSyncNode->pRaftStore->currentTerm,
                pSyncNode->commitIndex, logBeginIndex, logLastIndex, snapshot.lastApplyIndex, snapshot.lastApplyTerm,
                pSyncNode->pRaftCfg->isStandBy, pSyncNode->pRaftCfg->snapshotStrategy, pSyncNode->pRaftCfg->batchSize,
                pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex, pSyncNode->changing,
-               pSyncNode->restoreFinish, printStr);
+               pSyncNode->restoreFinish, syncNodeDynamicQuorum(pSyncNode), pSyncNode->electTimerLogicClockUser,
+               pSyncNode->heartbeatTimerLogicClockUser, printStr);
     } else {
       snprintf(s, len, "%s", str);
     }

source/libs/sync/src/syncMessage.c

@@ -1947,6 +1947,8 @@ cJSON* syncAppendEntriesReply2Json(const SyncAppendEntriesReply* pMsg) {
     cJSON_AddNumberToObject(pRoot, "success", pMsg->success);
     snprintf(u64buf, sizeof(u64buf), "%" PRId64, pMsg->matchIndex);
     cJSON_AddStringToObject(pRoot, "matchIndex", u64buf);
+    snprintf(u64buf, sizeof(u64buf), "%" PRId64, pMsg->startTime);
+    cJSON_AddStringToObject(pRoot, "startTime", u64buf);
   }
 
   cJSON* pJson = cJSON_CreateObject();

source/libs/sync/src/syncReplication.c

@@ -116,6 +116,120 @@ int32_t syncNodeAppendEntriesPeers(SSyncNode* pSyncNode) {
   return ret;
 }
 
+int32_t syncNodeAppendEntriesOnePeer(SSyncNode* pSyncNode, SRaftId* pDestId, SyncIndex nextIndex) {
+  int32_t ret = 0;
+
+  // pre index, pre term
+  SyncIndex preLogIndex = syncNodeGetPreIndex(pSyncNode, nextIndex);
+  SyncTerm  preLogTerm = syncNodeGetPreTerm(pSyncNode, nextIndex);
+  if (preLogTerm == SYNC_TERM_INVALID) {
+    SyncIndex newNextIndex = syncNodeGetLastIndex(pSyncNode) + 1;
+    // SyncIndex newNextIndex = nextIndex + 1;
+
+    syncIndexMgrSetIndex(pSyncNode->pNextIndex, pDestId, newNextIndex);
+    syncIndexMgrSetIndex(pSyncNode->pMatchIndex, pDestId, SYNC_INDEX_INVALID);
+    sError("vgId:%d, sync get pre term error, nextIndex:%" PRId64 ", update next-index:%" PRId64
+           ", match-index:%d, raftid:%" PRId64,
+           pSyncNode->vgId, nextIndex, newNextIndex, SYNC_INDEX_INVALID, pDestId->addr);
+    return -1;
+  }
+
+  // entry pointer array
+  SSyncRaftEntry* entryPArr[SYNC_MAX_BATCH_SIZE];
+  memset(entryPArr, 0, sizeof(entryPArr));
+
+  // get entry batch
+  int32_t   getCount = 0;
+  SyncIndex getEntryIndex = nextIndex;
+  for (int32_t i = 0; i < pSyncNode->pRaftCfg->batchSize; ++i) {
+    SSyncRaftEntry* pEntry = NULL;
+    int32_t         code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, getEntryIndex, &pEntry);
+    if (code == 0) {
+      ASSERT(pEntry != NULL);
+      entryPArr[i] = pEntry;
+      getCount++;
+      getEntryIndex++;
+
+    } else {
+      break;
+    }
+  }
+
+  // event log
+  do {
+    char     logBuf[128];
+    char     host[64];
+    uint16_t port;
+    syncUtilU642Addr(pDestId->addr, host, sizeof(host), &port);
+    snprintf(logBuf, sizeof(logBuf), "build batch:%d for %s:%d", getCount, host, port);
+    syncNodeEventLog(pSyncNode, logBuf);
+  } while (0);
+
+  // build msg
+  SyncAppendEntriesBatch* pMsg = syncAppendEntriesBatchBuild(entryPArr, getCount, pSyncNode->vgId);
+  ASSERT(pMsg != NULL);
+
+  // free entries
+  for (int32_t i = 0; i < pSyncNode->pRaftCfg->batchSize; ++i) {
+    SSyncRaftEntry* pEntry = entryPArr[i];
+    if (pEntry != NULL) {
+      syncEntryDestory(pEntry);
+      entryPArr[i] = NULL;
+    }
+  }
+
+  // prepare msg
+  pMsg->srcId = pSyncNode->myRaftId;
+  pMsg->destId = *pDestId;
+  pMsg->term = pSyncNode->pRaftStore->currentTerm;
+  pMsg->prevLogIndex = preLogIndex;
+  pMsg->prevLogTerm = preLogTerm;
+  pMsg->commitIndex = pSyncNode->commitIndex;
+  pMsg->privateTerm = 0;
+  pMsg->dataCount = getCount;
+
+  // send msg
+  syncNodeAppendEntriesBatch(pSyncNode, pDestId, pMsg);
+
+  // speed up
+  if (pMsg->dataCount > 0 && pSyncNode->commitIndex - pMsg->prevLogIndex > SYNC_SLOW_DOWN_RANGE) {
+    ret = 1;
+
+#if 0
+      do {
+        char     logBuf[128];
+        char     host[64];
+        uint16_t port;
+        syncUtilU642Addr(pDestId->addr, host, sizeof(host), &port);
+        snprintf(logBuf, sizeof(logBuf), "maybe speed up for %s:%d, pre-index:%ld", host, port, pMsg->prevLogIndex);
+        syncNodeEventLog(pSyncNode, logBuf);
+      } while (0);
+#endif
+  }
+
+  syncAppendEntriesBatchDestroy(pMsg);
+
+  return ret;
+}
+
+int32_t syncNodeAppendEntriesPeersSnapshot2(SSyncNode* pSyncNode) {
+  if (pSyncNode->state != TAOS_SYNC_STATE_LEADER) {
+    return -1;
+  }
+
+  int32_t ret = 0;
+  for (int i = 0; i < pSyncNode->peersNum; ++i) {
+    SRaftId* pDestId = &(pSyncNode->peersId[i]);
+
+    // next index
+    SyncIndex nextIndex = syncIndexMgrGetIndex(pSyncNode->pNextIndex, pDestId);
+    ret = syncNodeAppendEntriesOnePeer(pSyncNode, pDestId, nextIndex);
+  }
+
+  return ret;
+}
+
+#if 0
 int32_t syncNodeAppendEntriesPeersSnapshot2(SSyncNode* pSyncNode) {
   if (pSyncNode->state != TAOS_SYNC_STATE_LEADER) {
     return -1;
@@ -221,6 +335,7 @@ int32_t syncNodeAppendEntriesPeersSnapshot2(SSyncNode* pSyncNode) {
 
   return ret;
 }
+#endif
 
 int32_t syncNodeAppendEntriesPeersSnapshot(SSyncNode* pSyncNode) {
   ASSERT(pSyncNode->state == TAOS_SYNC_STATE_LEADER);

source/libs/sync/test/syncAppendEntriesReplyTest.cpp

@@ -24,6 +24,7 @@ SyncAppendEntriesReply *createMsg() {
   pMsg->matchIndex = 77;
   pMsg->term = 33;
   pMsg->privateTerm = 44;
+  pMsg->startTime = taosGetTimestampMs();
   return pMsg;
 }
 
@@ -89,6 +90,8 @@ void test5() {
 }
 
 int main() {
+  gRaftDetailLog = true;
+
   tsAsyncLog = 0;
   sDebugFlag = DEBUG_TRACE + DEBUG_SCREEN + DEBUG_FILE;
   logTest();

source/libs/transport/src/transCli.c

@@ -205,28 +205,6 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) {
 #define CONN_PERSIST_TIME(para)    ((para) <= 90000 ? 90000 : (para))
 #define CONN_GET_HOST_THREAD(conn) (conn ? ((SCliConn*)conn)->hostThrd : NULL)
 #define CONN_GET_INST_LABEL(conn)  (((STrans*)(((SCliThrd*)(conn)->hostThrd)->pTransInst))->label)
-#define CONN_SHOULD_RELEASE(conn, head)                                                                              \
-  do {                                                                                                               \
-    if ((head)->release == 1 && (head->msgLen) == sizeof(*head)) {                                                   \
-      uint64_t ahandle = head->ahandle;                                                                              \
-      CONN_GET_MSGCTX_BY_AHANDLE(conn, ahandle);                                                                     \
-      transClearBuffer(&conn->readBuf);                                                                              \
-      transFreeMsg(transContFromHead((char*)head));                                                                  \
-      if (transQueueSize(&conn->cliMsgs) > 0 && ahandle == 0) {                                                      \
-        SCliMsg* cliMsg = transQueueGet(&conn->cliMsgs, 0);                                                          \
-        if (cliMsg->type == Release) return;                                                                         \
-      }                                                                                                              \
-      tDebug("%s conn %p receive release request, refId:%" PRId64 "", CONN_GET_INST_LABEL(conn), conn, conn->refId); \
-      if (T_REF_VAL_GET(conn) > 1) {                                                                                 \
-        transUnrefCliHandle(conn);                                                                                   \
-      }                                                                                                              \
-      destroyCmsg(pMsg);                                                                                             \
-      cliReleaseUnfinishedMsg(conn);                                                                                 \
-      transQueueClear(&conn->cliMsgs);                                                                               \
-      addConnToPool(((SCliThrd*)conn->hostThrd)->pool, conn);                                                        \
-      return;                                                                                                        \
-    }                                                                                                                \
-  } while (0)
 
 #define CONN_GET_MSGCTX_BY_AHANDLE(conn, ahandle)                                         \
   do {                                                                                    \
@@ -358,7 +336,6 @@ void cliHandleResp(SCliConn* conn) {
   if (cliRecvReleaseReq(conn, pHead)) {
     return;
   }
-  CONN_SHOULD_RELEASE(conn, pHead);
 
   if (CONN_NO_PERSIST_BY_APP(conn)) {
     pMsg = transQueuePop(&conn->cliMsgs);
@@ -1418,7 +1395,7 @@ int transReleaseCliHandle(void* handle) {
   }
 
   STransMsg tmsg = {.info.handle = handle};
-  // TRACE_SET_MSGID(&tmsg.info.traceId, tGenIdPI64());
+  TRACE_SET_MSGID(&tmsg.info.traceId, tGenIdPI64());
 
   SCliMsg* cmsg = taosMemoryCalloc(1, sizeof(SCliMsg));
   cmsg->msg = tmsg;