Merge remote-tracking branch 'origin/3.0' into feat/TD-27337

docs/en/12-taos-sql/05-insert.md

@@ -1,7 +1,7 @@
 ---
 title: Insert
 sidebar_label: Insert
-description: This document describes how to insert data into TDengine.
+description: This document describes the SQL commands and syntax for inserting data into TDengine.
 ---
 
 ## Syntax

docs/en/12-taos-sql/13-tmq.md

@@ -1,5 +1,5 @@
 ---
-title: Data Subscription
+title: Data Subscription SQL Reference
 sidebar_label: Data Subscription
 description: This document describes the SQL statements related to the data subscription component of TDengine.
 ---

docs/en/12-taos-sql/14-stream.md

@@ -1,5 +1,5 @@
 ---
-title: Stream Processing
+title: Stream Processing SQL Reference
 sidebar_label: Stream Processing
 description: This document describes the SQL statements related to the stream processing component of TDengine.
 ---
@@ -148,7 +148,7 @@ T = latest event time - watermark
 
 The window closing time for each batch of data that arrives at the system is updated using the preceding formula, and all windows are closed whose closing time is less than T. If the triggering method is WINDOW_CLOSE or MAX_DELAY, the aggregate result for the window is pushed.
 
-Stream processing strategy for expired data
+## Stream processing strategy for expired data
 The data in expired windows is tagged as expired. TDengine stream processing provides two methods for handling such data:
 
 1. Drop the data. This is the default and often only handling method for most stream processing engines.
@@ -157,6 +157,14 @@ The data in expired windows is tagged as expired. TDengine stream processing pro
 
 In both of these methods, configuring the watermark is essential for obtaining accurate results (if expired data is dropped) and avoiding repeated triggers that affect system performance (if expired data is recalculated).
 
+## Stream processing strategy for modifying data
+
+TDengine provides two ways to handle modified data, which are specified by the IGNORE UPDATE option:
+
+1. Check whether the data has been modified, i.e. IGNORE UPDATE 0, and recalculate the corresponding window if the data has been modified.
+
+2. Do not check whether the data has been modified, and calculate all the data as incremental data, i.e. IGNORE UPDATE 1, the default configuration.
+
 ## Supported functions
 
 All [scalar functions](../function/#scalar-functions) are available in stream processing. All [Aggregate functions](../function/#aggregate-functions)  and  [Selection functions](../function/#selection-functions) are available in stream processing, except the followings:

docs/en/12-taos-sql/26-udf.md

@@ -1,5 +1,5 @@
 ---
-title: User-Defined Functions (UDF)
+title: User-Defined Functions (UDF) SQL Reference
 sidebar_label: User-Defined Functions
 description: This document describes the SQL statements related to user-defined functions (UDF) in TDengine.
 ---

docs/en/14-reference/07-tdinsight/index.md

@@ -1,5 +1,5 @@
 ---
-title: TDinsight - Grafana-based Zero-Dependency Monitoring Solution for TDengine
+title: TDinsight
 sidebar_label: TDinsight
 description: This document describes TDinsight, a monitoring solution for TDengine.
 ---

docs/en/25-application/01-telegraf.md

@@ -1,5 +1,5 @@
 ---
-title: Quickly Build IT DevOps Visualization System with TDengine + Telegraf + Grafana
+title: IT Visualization with TDengine + Telegraf + Grafana
 sidebar_label: TDengine + Telegraf + Grafana
 description: This document describes how to create an IT visualization system by integrating TDengine with Telegraf and Grafana.
 ---

docs/examples/rust/nativeexample/examples/query.rs

@@ -0,0 +1,66 @@
+use taos::*;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let dsn = "taos://localhost:6030";
+    let builder = TaosBuilder::from_dsn(dsn)?;
+
+    let taos = builder.build()?;
+
+    // ANCHOR: create_db_and_table
+    let db = "power";
+    // create database
+    taos.exec_many([
+        format!("DROP DATABASE IF EXISTS `{db}`"),
+        format!("CREATE DATABASE `{db}`"),
+        format!("USE `{db}`"),
+    ])
+    .await?;
+
+    // create table
+    taos.exec_many([
+        // create super table
+        "CREATE TABLE `meters` (`ts` TIMESTAMP, `current` FLOAT, `voltage` INT, `phase` FLOAT) \
+         TAGS (`groupid` INT, `location` BINARY(24))",
+    ]).await?;
+    // ANCHOR_END: create_db_and_table
+    
+    // ANCHOR: insert_data
+    let inserted = taos.exec("INSERT INTO " +
+    "power.d1001 USING power.meters TAGS(2,'California.SanFrancisco') " +
+    "VALUES " +
+    "(NOW + 1a, 10.30000, 219, 0.31000) " +
+    "(NOW + 2a, 12.60000, 218, 0.33000) " +
+    "(NOW + 3a, 12.30000, 221, 0.31000) " +
+    "power.d1002 USING power.meters TAGS(3, 'California.SanFrancisco') " +
+    "VALUES " +
+    "(NOW + 1a, 10.30000, 218, 0.25000) ").await?;
+
+    println!("inserted: {} rows", inserted);
+    // ANCHOR_END: insert_data
+
+    // ANCHOR: query_data
+    let mut result = taos.query("SELECT * FROM power.meters").await?;
+
+    for field in result.fields() {
+        println!("got field: {}", field.name());
+    }
+
+    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;
+    }
+    // ANCHOR_END: query_data
+
+    // ANCHOR: query_with_req_id
+    let result = taos.query_with_req_id("SELECT * FROM power.meters", 0).await?;
+    // ANCHOR_END: query_with_req_id
+
+}

docs/examples/rust/nativeexample/examples/schemaless.rs

@@ -0,0 +1,80 @@
+use taos_query::common::SchemalessPrecision;
+use taos_query::common::SchemalessProtocol;
+use taos_query::common::SmlDataBuilder;
+
+use crate::AsyncQueryable;
+use crate::AsyncTBuilder;
+use crate::TaosBuilder;
+
+async fn put() -> anyhow::Result<()> {
+    std::env::set_var("RUST_LOG", "taos=debug");
+    pretty_env_logger::init();
+    let dsn =
+        std::env::var("TDENGINE_ClOUD_DSN").unwrap_or("http://localhost:6041".to_string());
+    log::debug!("dsn: {:?}", &dsn);
+
+    let client = TaosBuilder::from_dsn(dsn)?.build().await?;
+
+    let db = "power";
+
+    client.exec(format!("drop database if exists {db}")).await?;
+
+    client
+        .exec(format!("create database if not exists {db}"))
+        .await?;
+
+    // should specify database before insert
+    client.exec(format!("use {db}")).await?;
+
+    // SchemalessProtocol::Line
+    let data = [
+        "meters,groupid=2,location=California.SanFrancisco current=10.3000002f64,voltage=219i32,phase=0.31f64 1626006833639000000",
+    ]
+    .map(String::from)
+    .to_vec();
+
+    let sml_data = SmlDataBuilder::default()
+        .protocol(SchemalessProtocol::Line)
+        .precision(SchemalessPrecision::Millisecond)
+        .data(data.clone())
+        .ttl(1000)
+        .req_id(100u64)
+        .build()?;
+    assert_eq!(client.put(&sml_data).await?, ());
+
+    // SchemalessProtocol::Telnet
+    let data = [
+        "meters.current 1648432611249 10.3 location=California.SanFrancisco group=2",
+    ]
+    .map(String::from)
+    .to_vec();
+
+    let sml_data = SmlDataBuilder::default()
+        .protocol(SchemalessProtocol::Telnet)
+        .precision(SchemalessPrecision::Millisecond)
+        .data(data.clone())
+        .ttl(1000)
+        .req_id(200u64)
+        .build()?;
+    assert_eq!(client.put(&sml_data).await?, ());
+
+    // SchemalessProtocol::Json
+    let data = [
+        r#"[{"metric": "meters.current", "timestamp": 1681345954000, "value": 10.3, "tags": {"location": "California.SanFrancisco", "groupid": 2}}, {"metric": "meters.voltage", "timestamp": 1648432611249, "value": 219, "tags": {"location": "California.LosAngeles", "groupid": 1}}, {"metric": "meters.current", "timestamp": 1648432611250, "value": 12.6, "tags": {"location": "California.SanFrancisco", "groupid": 2}}, {"metric": "meters.voltage", "timestamp": 1648432611250, "value": 221, "tags": {"location": "California.LosAngeles", "groupid": 1}}]"#
+    ]
+    .map(String::from)
+    .to_vec();
+
+    let sml_data = SmlDataBuilder::default()
+        .protocol(SchemalessProtocol::Json)
+        .precision(SchemalessPrecision::Millisecond)
+        .data(data.clone())
+        .ttl(1000)
+        .req_id(300u64)
+        .build()?;
+    assert_eq!(client.put(&sml_data).await?, ());
+
+    client.exec(format!("drop database if exists {db}")).await?;
+
+    Ok(())
+}

docs/examples/rust/nativeexample/examples/stmt.rs

@@ -0,0 +1,37 @@
+use taos::*;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let taos = TaosBuilder::from_dsn("taos://")?.build().await?;
+
+    taos.exec("DROP DATABASE IF EXISTS power").await?;
+    taos.create_database("power").await?;
+    taos.use_database("power").await?;
+    taos.exec("CREATE STABLE IF NOT EXISTS meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (location BINARY(64), groupId INT)").await?;
+
+    let mut stmt = Stmt::init(&taos).await?;
+    stmt.prepare("INSERT INTO ? USING meters TAGS(?, ?) VALUES(?, ?, ?, ?)").await?;
+
+    const NUM_TABLES: usize = 10;
+    const NUM_ROWS: usize = 10;
+    for i in 0..NUM_TABLES {
+        let table_name = format!("d{}", i);
+        let tags = vec![Value::VarChar("California.SanFransico".into()), Value::Int(2)];
+        stmt.set_tbname_tags(&table_name, &tags).await?;
+        for j in 0..NUM_ROWS {
+            let values = vec![
+                ColumnView::from_millis_timestamp(vec![1648432611249 + j as i64]),
+                ColumnView::from_floats(vec![10.3 + j as f32]),
+                ColumnView::from_ints(vec![219 + j as i32]),
+                ColumnView::from_floats(vec![0.31 + j as f32]),
+            ];
+            stmt.bind(&values).await?;
+        }
+        stmt.add_batch().await?;
+    }
+
+    // execute.
+    let rows = stmt.execute().await?;
+    assert_eq!(rows, NUM_TABLES * NUM_ROWS);
+    Ok(())
+}

docs/examples/rust/nativeexample/examples/tmq.rs

@@ -0,0 +1,166 @@
+use std::time::Duration;
+use std::str::FromStr;
+
+use taos::*;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    pretty_env_logger::formatted_timed_builder()
+        .filter_level(log::LevelFilter::Info)
+        .init();
+    use taos_query::prelude::*;
+    let dsn = "taos://localhost:6030".to_string();
+    log::info!("dsn: {}", dsn);
+    let mut dsn = Dsn::from_str(&dsn)?;
+
+    let taos = TaosBuilder::from_dsn(&dsn)?.build().await?;
+
+    // prepare database and table
+    taos.exec_many([
+        "drop topic if exists topic_meters",
+        "drop database if exists power",
+        "create database if not exists power WAL_RETENTION_PERIOD 86400",
+        "use power",
+
+        "CREATE STABLE IF NOT EXISTS power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))",
+
+        "create table if not exists power.d001 using power.meters tags(1,'location')",
+
+    ])
+    .await?;
+
+    taos.exec_many([
+        "drop database if exists db2",
+        "create database if not exists db2 wal_retention_period 3600",
+        "use db2",
+    ])
+    .await?;
+
+    // ANCHOR: create_topic
+    taos.exec_many([
+        "CREATE TOPIC IF NOT EXISTS topic_meters AS SELECT ts, current, voltage, phase, groupid, location FROM power.meters",
+    ])
+    .await?;
+    // ANCHOR_END: create_topic
+
+    // ANCHOR: create_consumer
+    dsn.params.insert("group.id".to_string(), "abc".to_string());
+    dsn.params.insert("auto.offset.reset".to_string(), "earliest".to_string());
+
+    let builder = TmqBuilder::from_dsn(&dsn)?;
+    let mut consumer = builder.build().await?;
+    // ANCHOR_END: create_consumer
+
+    // ANCHOR: subscribe
+    consumer.subscribe(["topic_meters"]).await?;
+    // ANCHOR_END: subscribe
+
+    // ANCHOR: consume
+    {
+        let mut stream = consumer.stream_with_timeout(Timeout::from_secs(1));
+
+        while let Some((offset, message)) = stream.try_next().await? {
+
+            let topic: &str = offset.topic();
+            let database = offset.database();
+            let vgroup_id = offset.vgroup_id();
+            log::debug!(
+                "topic: {}, database: {}, vgroup_id: {}",
+                topic,
+                database,
+                vgroup_id
+            );
+
+            match message {
+                MessageSet::Meta(meta) => {
+                    log::info!("Meta");
+                    let raw = meta.as_raw_meta().await?;
+                    taos.write_raw_meta(&raw).await?;
+
+                    let json = meta.as_json_meta().await?;
+                    let sql = json.to_string();
+                    if let Err(err) = taos.exec(sql).await {
+                        println!("maybe error: {}", err);
+                    }
+                }
+                MessageSet::Data(data) => {
+                    log::info!("Data");
+                    while let Some(data) = data.fetch_raw_block().await? {
+                        log::debug!("data: {:?}", data);
+                    }
+                }
+                MessageSet::MetaData(meta, data) => {
+                    log::info!("MetaData");
+                    let raw = meta.as_raw_meta().await?;
+                    taos.write_raw_meta(&raw).await?;
+
+                    let json = meta.as_json_meta().await?;
+                    let sql = json.to_string();
+                    if let Err(err) = taos.exec(sql).await {
+                        println!("maybe error: {}", err);
+                    }
+
+                    while let Some(data) = data.fetch_raw_block().await? {
+                        log::debug!("data: {:?}", data);
+                    }
+                }
+            }
+            consumer.commit(offset).await?;
+        }
+    }
+    // ANCHOR_END: consume
+
+    // ANCHOR: assignments
+    let assignments = consumer.assignments().await.unwrap();
+    log::info!("assignments: {:?}", assignments);
+    // ANCHOR_END: assignments
+
+    // seek offset
+    for topic_vec_assignment in assignments {
+        let topic = &topic_vec_assignment.0;
+        let vec_assignment = topic_vec_assignment.1;
+        for assignment in vec_assignment {
+            let vgroup_id = assignment.vgroup_id();
+            let current = assignment.current_offset();
+            let begin = assignment.begin();
+            let end = assignment.end();
+            log::debug!(
+                "topic: {}, vgroup_id: {}, current offset: {} begin {}, end: {}",
+                topic,
+                vgroup_id,
+                current,
+                begin,
+                end
+            );
+            // ANCHOR: seek_offset
+            let res = consumer.offset_seek(topic, vgroup_id, end).await;
+            if res.is_err() {
+                log::error!("seek offset error: {:?}", res);
+                let a = consumer.assignments().await.unwrap();
+                log::error!("assignments: {:?}", a);
+            }
+            // ANCHOR_END: seek_offset
+        }
+
+        let topic_assignment = consumer.topic_assignment(topic).await;
+        log::debug!("topic assignment: {:?}", topic_assignment);
+    }
+
+    // after seek offset
+    let assignments = consumer.assignments().await.unwrap();
+    log::info!("after seek offset assignments: {:?}", assignments);
+
+    // ANCHOR: unsubscribe
+    consumer.unsubscribe().await;
+    // ANCHOR_END: unsubscribe
+
+    tokio::time::sleep(Duration::from_secs(1)).await;
+
+    taos.exec_many([
+        "drop database db2",
+        "drop topic topic_meters",
+        "drop database power",
+    ])
+    .await?;
+    Ok(())
+}

docs/zh/12-taos-sql/14-stream.md

@@ -201,9 +201,9 @@ TDengine 对于过期数据提供两种处理方式，由 IGNORE EXPIRED 选项
 
 TDengine 对于修改数据提供两种处理方式，由 IGNORE UPDATE 选项指定：
 
-1. 检查数据是否被修改，即 IGNORE UPDATE 0：默认配置，如果被修改，则重新计算对应窗口。
+1. 检查数据是否被修改，即 IGNORE UPDATE 0，如果数据被修改，则重新计算对应窗口。
 
-2. 不检查数据是否被修改，全部按增量数据计算，即 IGNORE UPDATE 1。
+2. 不检查数据是否被修改，全部按增量数据计算，即 IGNORE UPDATE 1，默认配置。
 
 
 ## 写入已存在的超级表

include/common/tdatablock.h

@@ -275,7 +275,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq2** pReq, const SSDataBlock* pData
 
 bool  alreadyAddGroupId(char* ctbName);
 bool  isAutoTableName(char* ctbName);
-void  buildCtbNameAddGroupId(char* ctbName, uint64_t groupId);
+void  buildCtbNameAddGroupId(const char* stbName, char* ctbName, uint64_t groupId);
 char* buildCtbNameByGroupId(const char* stbName, uint64_t groupId);
 int32_t buildCtbNameByGroupIdImpl(const char* stbName, uint64_t groupId, char* pBuf);
 

include/common/ttokendef.h

@@ -386,6 +386,7 @@
 
 
 
+
 #define TK_NK_SPACE         600
 #define TK_NK_COMMENT       601
 #define TK_NK_ILLEGAL       602

include/libs/stream/tstream.h

@@ -61,7 +61,7 @@ typedef struct SStreamTask   SStreamTask;
 typedef struct SStreamQueue  SStreamQueue;
 typedef struct SStreamTaskSM SStreamTaskSM;
 
-#define SSTREAM_TASK_VER                  3
+#define SSTREAM_TASK_VER                  4
 #define SSTREAM_TASK_INCOMPATIBLE_VER     1
 #define SSTREAM_TASK_NEED_CONVERT_VER     2
 #define SSTREAM_TASK_SUBTABLE_CHANGED_VER 3
@@ -355,6 +355,8 @@ typedef struct SMetaHbInfo        SMetaHbInfo;
 typedef struct SDispatchMsgInfo {
   SStreamDispatchReq* pData;  // current dispatch data
   int8_t              dispatchMsgType;
+  int64_t             checkpointId;// checkpoint id msg
+  int32_t             transId;     // transId for current checkpoint
   int16_t             msgType;     // dispatch msg type
   int32_t             retryCount;  // retry send data count
   int64_t             startTs;     // dispatch start time, record total elapsed time for dispatch

source/client/src/clientTmq.c

@@ -1626,6 +1626,22 @@ void changeByteEndian(char* pData){
   }
 }
 
+static void tmqGetRawDataRowsPrecisionFromRes(void *pRetrieve, void** rawData, int64_t *rows, int32_t *precision){
+  if(*(int64_t*)pRetrieve == 0){
+    *rawData = ((SRetrieveTableRsp*)pRetrieve)->data;
+    *rows = htobe64(((SRetrieveTableRsp*)pRetrieve)->numOfRows);
+    if(precision != NULL){
+      *precision = ((SRetrieveTableRsp*)pRetrieve)->precision;
+    }
+  }else if(*(int64_t*)pRetrieve == 1){
+    *rawData = ((SRetrieveTableRspForTmq*)pRetrieve)->data;
+    *rows = htobe64(((SRetrieveTableRspForTmq*)pRetrieve)->numOfRows);
+    if(precision != NULL){
+      *precision = ((SRetrieveTableRspForTmq*)pRetrieve)->precision;
+    }
+  }
+}
+
 static void tmqBuildRspFromWrapperInner(SMqPollRspWrapper* pWrapper, SMqClientVg* pVg, int64_t* numOfRows, SMqRspObj* pRspObj) {
   (*numOfRows) = 0;
   tstrncpy(pRspObj->topic, pWrapper->topicHandle->topicName, TSDB_TOPIC_FNAME_LEN);
@@ -1648,13 +1664,7 @@ static void tmqBuildRspFromWrapperInner(SMqPollRspWrapper* pWrapper, SMqClientVg
     void* rawData = NULL;
     int64_t rows = 0;
     // deal with compatibility
-    if(*(int64_t*)pRetrieve == 0){
-      rawData = ((SRetrieveTableRsp*)pRetrieve)->data;
-      rows = htobe64(((SRetrieveTableRsp*)pRetrieve)->numOfRows);
-    }else if(*(int64_t*)pRetrieve == 1){
-      rawData = ((SRetrieveTableRspForTmq*)pRetrieve)->data;
-      rows = htobe64(((SRetrieveTableRspForTmq*)pRetrieve)->numOfRows);
-    }
+    tmqGetRawDataRowsPrecisionFromRes(pRetrieve, &rawData, &rows, NULL);
 
     pVg->numOfRows += rows;
     (*numOfRows) += rows;
@@ -2625,18 +2635,22 @@ SReqResultInfo* tmqGetNextResInfo(TAOS_RES* res, bool convertUcs4) {
   pRspObj->resIter++;
 
   if (pRspObj->resIter < pRspObj->rsp.blockNum) {
-    SRetrieveTableRspForTmq* pRetrieveTmq =
-        (SRetrieveTableRspForTmq*)taosArrayGetP(pRspObj->rsp.blockData, pRspObj->resIter);
     if (pRspObj->rsp.withSchema) {
       doFreeReqResultInfo(&pRspObj->resInfo);
       SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(pRspObj->rsp.blockSchema, pRspObj->resIter);
       setResSchemaInfo(&pRspObj->resInfo, pSW->pSchema, pSW->nCols);
     }
 
-    pRspObj->resInfo.pData = (void*)pRetrieveTmq->data;
-    pRspObj->resInfo.numOfRows = htobe64(pRetrieveTmq->numOfRows);
+    void* pRetrieve = taosArrayGetP(pRspObj->rsp.blockData, pRspObj->resIter);
+    void* rawData = NULL;
+    int64_t rows = 0;
+    int32_t precision = 0;
+    tmqGetRawDataRowsPrecisionFromRes(pRetrieve, &rawData, &rows, &precision);
+
+    pRspObj->resInfo.pData = rawData;
+    pRspObj->resInfo.numOfRows = rows;
     pRspObj->resInfo.current = 0;
-    pRspObj->resInfo.precision = pRetrieveTmq->precision;
+    pRspObj->resInfo.precision = precision;
 
     // TODO handle the compressed case
     pRspObj->resInfo.totalRows += pRspObj->resInfo.numOfRows;

source/common/src/tdatablock.c

@@ -2344,10 +2344,14 @@ _end:
   return TSDB_CODE_SUCCESS;
 }
 
-void  buildCtbNameAddGroupId(char* ctbName, uint64_t groupId){
+void  buildCtbNameAddGroupId(const char* stbName, char* ctbName, uint64_t groupId){
   char tmp[TSDB_TABLE_NAME_LEN] = {0};
+  if (stbName == NULL){
     snprintf(tmp, TSDB_TABLE_NAME_LEN, "_%"PRIu64, groupId);
-  ctbName[TSDB_TABLE_NAME_LEN - strlen(tmp) - 1] = 0;  // put groupId to the end
+  }else{
+    snprintf(tmp, TSDB_TABLE_NAME_LEN, "_%s_%"PRIu64, stbName, groupId);
+  }
+  ctbName[TSDB_TABLE_NAME_LEN - strlen(tmp) - 1] = 0;  // put stbname + groupId to the end
   strcat(ctbName, tmp);
 }
 
@@ -2357,6 +2361,7 @@ bool isAutoTableName(char* ctbName) { return (strlen(ctbName) == 34 && ctbName[0
 
 bool alreadyAddGroupId(char* ctbName) {
   size_t len = strlen(ctbName);
+  if (len == 0) return false;
   size_t _location = len - 1;
   while (_location > 0) {
     if (ctbName[_location] < '0' || ctbName[_location] > '9') {

source/common/src/tmsg.c

@@ -51,8 +51,8 @@
 
 #define ENCODESQL()                                                        \
   do {                                                                     \
-    if (pReq->sqlLen > 0 && pReq->sql != NULL) {                           \
     if (tEncodeI32(&encoder, pReq->sqlLen) < 0) return -1;                 \
+    if (pReq->sqlLen > 0 && pReq->sql != NULL) {                           \
       if (tEncodeBinary(&encoder, pReq->sql, pReq->sqlLen) < 0) return -1; \
     }                                                                      \
   } while (0)
@@ -3025,7 +3025,7 @@ int32_t tSerializeSCreateDbReq(void *buf, int32_t bufLen, SCreateDbReq *pReq) {
 
   ENCODESQL();
 
-  if (tEncodeI32(&encoder, pReq->withArbitrator) < 0) return -1;
+  if (tEncodeI8(&encoder, pReq->withArbitrator) < 0) return -1;
 
   tEndEncode(&encoder);
 
@@ -3140,7 +3140,7 @@ int32_t tSerializeSAlterDbReq(void *buf, int32_t bufLen, SAlterDbReq *pReq) {
   if (tEncodeI32(&encoder, pReq->walRetentionSize) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->keepTimeOffset) < 0) return -1;
   ENCODESQL();
-  if (tEncodeI32(&encoder, pReq->withArbitrator) < 0) return -1;
+  if (tEncodeI8(&encoder, pReq->withArbitrator) < 0) return -1;
   tEndEncode(&encoder);
 
   int32_t tlen = encoder.pos;

source/dnode/mnode/impl/inc/mndArbGroup.h

@@ -35,6 +35,7 @@ int32_t mndSetCreateArbGroupRedoLogs(STrans *pTrans, SArbGroup *pGroup);
 int32_t mndSetCreateArbGroupUndoLogs(STrans *pTrans, SArbGroup *pGroup);
 int32_t mndSetCreateArbGroupCommitLogs(STrans *pTrans, SArbGroup *pGroup);
 
+int32_t mndSetDropArbGroupPrepareLogs(STrans *pTrans, SArbGroup *pGroup);
 int32_t mndSetDropArbGroupCommitLogs(STrans *pTrans, SArbGroup *pGroup);
 
 bool mndUpdateArbGroupByHeartBeat(SArbGroup *pGroup, SVArbHbRspMember *pRspMember, int64_t nowMs, int32_t dnodeId,

source/dnode/mnode/impl/inc/mndDef.h

@@ -160,6 +160,7 @@ typedef struct {
   ETrnConflct   conflict;
   ETrnExec      exec;
   EOperType     oper;
+  bool          changeless;
   int32_t       code;
   int32_t       failedTimes;
   void*         rpcRsp;

source/dnode/mnode/impl/inc/mndTrans.h

@@ -81,6 +81,7 @@ void    mndTransSetDbName(STrans *pTrans, const char *dbname, const char *stbnam
 void    mndTransSetArbGroupId(STrans *pTrans, int32_t groupId);
 void    mndTransSetSerial(STrans *pTrans);
 void    mndTransSetParallel(STrans *pTrans);
+void    mndTransSetChangeless(STrans *pTrans);
 void    mndTransSetOper(STrans *pTrans, EOperType oper);
 int32_t mndTransCheckConflict(SMnode *pMnode, STrans *pTrans);
 #ifndef BUILD_NO_CALL

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

@@ -260,6 +260,14 @@ int32_t mndSetCreateArbGroupCommitLogs(STrans *pTrans, SArbGroup *pGroup) {
   return 0;
 }
 
+int32_t mndSetDropArbGroupPrepareLogs(STrans *pTrans, SArbGroup *pGroup) {
+  SSdbRaw *pRedoRaw = mndArbGroupActionEncode(pGroup);
+  if (pRedoRaw == NULL) return -1;
+  if (mndTransAppendPrepareLog(pTrans, pRedoRaw) != 0) return -1;
+  if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPING) != 0) return -1;
+  return 0;
+}
+
 static int32_t mndSetDropArbGroupRedoLogs(STrans *pTrans, SArbGroup *pGroup) {
   SSdbRaw *pRedoRaw = mndArbGroupActionEncode(pGroup);
   if (pRedoRaw == NULL) return -1;
@@ -668,8 +676,15 @@ static int32_t mndProcessArbUpdateGroupReq(SRpcMsg *pReq) {
   newGroup.version = req.version;
 
   SMnode    *pMnode = pReq->info.node;
+  SArbGroup *pOldGroup = sdbAcquire(pMnode->pSdb, SDB_ARBGROUP, &newGroup.vgId);
+  if (!pOldGroup) {
+    mInfo("vgId:%d, arb skip to update arbgroup, since no obj found", newGroup.vgId);
+    return 0;
+  }
+  sdbRelease(pMnode->pSdb, pOldGroup);
+
   if (mndArbGroupUpdateTrans(pMnode, &newGroup) != 0) {
-    mError("vgId:%d, arb failed to update arbgroup, since %s", req.vgId, terrstr());
+    mError("vgId:%d, arb failed to update arbgroup, since %s", newGroup.vgId, terrstr());
     ret = -1;
   }
 

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

@@ -292,16 +292,16 @@ static void storeOffsetRows(SMnode *pMnode, SMqHbReq *req, SMqConsumerObj *pCons
 static int32_t buildMqHbRsp(SRpcMsg *pMsg, SMqHbRsp *rsp){
   int32_t tlen = tSerializeSMqHbRsp(NULL, 0, rsp);
   if (tlen <= 0){
-    return TSDB_CODE_OUT_OF_MEMORY;
+    return TSDB_CODE_TMQ_INVALID_MSG;
   }
   void   *buf = rpcMallocCont(tlen);
   if (buf == NULL) {
     return TSDB_CODE_OUT_OF_MEMORY;
   }
 
-  if(tSerializeSMqHbRsp(buf, tlen, rsp) != 0){
+  if(tSerializeSMqHbRsp(buf, tlen, rsp) <= 0){
     rpcFreeCont(buf);
-    return TSDB_CODE_OUT_OF_MEMORY;
+    return TSDB_CODE_TMQ_INVALID_MSG;
   }
   pMsg->info.rsp = buf;
   pMsg->info.rspLen = tlen;
@@ -316,7 +316,7 @@ static int32_t mndProcessMqHbReq(SRpcMsg *pMsg) {
   SMqConsumerObj *pConsumer = NULL;
 
   if (tDeserializeSMqHbReq(pMsg->pCont, pMsg->contLen, &req) < 0) {
-    code = TSDB_CODE_OUT_OF_MEMORY;
+    code = TSDB_CODE_TMQ_INVALID_MSG;
     goto end;
   }
 

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

@@ -1209,6 +1209,25 @@ static int32_t mndSetDropDbPrepareLogs(SMnode *pMnode, STrans *pTrans, SDbObj *p
   if (mndTransAppendPrepareLog(pTrans, pRedoRaw) != 0) return -1;
   if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPING) != 0) return -1;
 
+  SSdb *pSdb = pMnode->pSdb;
+  void *pIter = NULL;
+
+  while (1) {
+    SArbGroup *pArbGroup = NULL;
+    pIter = sdbFetch(pSdb, SDB_ARBGROUP, pIter, (void **)&pArbGroup);
+    if (pIter == NULL) break;
+
+    if (pArbGroup->dbUid == pDb->uid) {
+      if (mndSetDropArbGroupPrepareLogs(pTrans,pArbGroup) != 0) {
+        sdbCancelFetch(pSdb, pIter);
+        sdbRelease(pSdb, pArbGroup);
+        return -1;
+      }
+    }
+
+    sdbRelease(pSdb, pArbGroup);
+  }
+
   return 0;
 }
 

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

@@ -72,7 +72,9 @@ int32_t tEncodeSStreamObj(SEncoder *pEncoder, const SStreamObj *pObj) {
     if (tEncodeI32(pEncoder, innerSz) < 0) return -1;
     for (int32_t j = 0; j < innerSz; j++) {
       SStreamTask *pTask = taosArrayGetP(pArray, j);
+      if (pTask->ver < SSTREAM_TASK_SUBTABLE_CHANGED_VER){
         pTask->ver = SSTREAM_TASK_VER;
+      }
       if (tEncodeStreamTask(pEncoder, pTask) < 0) return -1;
     }
   }

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

@@ -1104,14 +1104,16 @@ static int32_t mndProcessCreateStbReq(SRpcMsg *pReq) {
       } else if (createReq.tagVer > 0 || createReq.colVer > 0) {
         int32_t tagDelta = createReq.tagVer - pStb->tagVer;
         int32_t colDelta = createReq.colVer - pStb->colVer;
-        int32_t verDelta = tagDelta + colDelta;
         mInfo("stb:%s, already exist while create, input tagVer:%d colVer:%d, exist tagVer:%d colVer:%d",
               createReq.name, createReq.tagVer, createReq.colVer, pStb->tagVer, pStb->colVer);
         if (tagDelta <= 0 && colDelta <= 0) {
           mInfo("stb:%s, schema version is not incremented and nothing needs to be done", createReq.name);
           code = 0;
           goto _OVER;
-        } else if ((tagDelta == 1 || colDelta == 1) && (verDelta == 1)) {
+        } else if ((tagDelta == 1 && colDelta == 0) ||
+                   (tagDelta == 0 && colDelta == 1) ||
+                   (pStb->colVer == 1 && createReq.colVer > 1) ||
+                   (pStb->tagVer == 1 && createReq.tagVer > 1)) {
           isAlter = true;
           mInfo("stb:%s, schema version is only increased by 1 number, do alter operation", createReq.name);
         } else {

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

@@ -434,7 +434,9 @@ int32_t mndPersistTaskDeployReq(STrans *pTrans, SStreamTask *pTask) {
   SEncoder encoder;
   tEncoderInit(&encoder, NULL, 0);
 
+  if (pTask->ver < SSTREAM_TASK_SUBTABLE_CHANGED_VER){
     pTask->ver = SSTREAM_TASK_VER;
+  }
   tEncodeStreamTask(&encoder, pTask);
 
   int32_t size = encoder.pos;
@@ -2153,41 +2155,60 @@ int32_t mndProcessStreamReqCheckpoint(SRpcMsg *pReq) {
 
   SStreamObj *pStream = mndGetStreamObj(pMnode, req.streamId);
   if (pStream == NULL) {
-    mError("failed to find the stream:0x%" PRIx64 " not handle the checkpoint req", req.streamId);
+    mWarn("failed to find the stream:0x%" PRIx64 ", not handle the checkpoint req, try to acquire in buf", req.streamId);
+
+    // not in meta-store yet, try to acquire the task in exec buffer
+    // the checkpoint req arrives too soon before the completion of the create stream trans.
+    STaskId id = {.streamId = req.streamId, .taskId = req.taskId};
+    void* p = taosHashGet(execInfo.pTaskMap, &id, sizeof(id));
+    if (p == NULL) {
+      mError("failed to find the stream:0x%" PRIx64 " in buf, not handle the checkpoint req", req.streamId);
       terrno = TSDB_CODE_MND_STREAM_NOT_EXIST;
       taosThreadMutexUnlock(&execInfo.lock);
-
       return -1;
+    } else {
+      mDebug("s-task:0x%" PRIx64 "-0x%x in buf not in mnode/meta, create stream trans may not complete yet",
+             req.streamId, req.taskId);
+    }
   }
 
-  int32_t  numOfTasks = mndGetNumOfStreamTasks(pStream);
+  int32_t numOfTasks = (pStream == NULL)? 0: mndGetNumOfStreamTasks(pStream);
+
   SArray **pReqTaskList = (SArray **)taosHashGet(execInfo.pTransferStateStreams, &req.streamId, sizeof(req.streamId));
   if (pReqTaskList == NULL) {
     SArray *pList = taosArrayInit(4, sizeof(int32_t));
-    doAddTaskId(pList, req.taskId, pStream->uid, numOfTasks);
+    doAddTaskId(pList, req.taskId, req.streamId, numOfTasks);
     taosHashPut(execInfo.pTransferStateStreams, &req.streamId, sizeof(int64_t), &pList, sizeof(void *));
 
     pReqTaskList = (SArray **)taosHashGet(execInfo.pTransferStateStreams, &req.streamId, sizeof(req.streamId));
   } else {
-    doAddTaskId(*pReqTaskList, req.taskId, pStream->uid, numOfTasks);
+    doAddTaskId(*pReqTaskList, req.taskId, req.streamId, numOfTasks);
   }
 
   int32_t total = taosArrayGetSize(*pReqTaskList);
   if (total == numOfTasks) {  // all tasks has send the reqs
     int64_t checkpointId = mndStreamGenChkpId(pMnode);
-    mDebug("stream:0x%" PRIx64 " all tasks req, start checkpointId:%" PRId64, pStream->uid, checkpointId);
+    mInfo("stream:0x%" PRIx64 " all tasks req checkpoint, start checkpointId:%" PRId64, req.streamId, checkpointId);
 
-    // TODO:handle error
+    if (pStream != NULL) { // TODO:handle error
       int32_t code = mndProcessStreamCheckpointTrans(pMnode, pStream, checkpointId, 0, false);
+    } else {
+      // todo: wait for the create stream trans completed, and launch the checkpoint trans
+      // SStreamObj *pStream = mndGetStreamObj(pMnode, req.streamId);
+      // sleep(500ms)
+    }
 
     // remove this entry
     taosHashRemove(execInfo.pTransferStateStreams, &req.streamId, sizeof(int64_t));
 
     int32_t numOfStreams = taosHashGetSize(execInfo.pTransferStateStreams);
-    mDebug("stream:0x%" PRIx64 " removed, remain streams:%d fill-history not completed", pStream->uid, numOfStreams);
+    mDebug("stream:0x%" PRIx64 " removed, remain streams:%d fill-history not completed", req.streamId, numOfStreams);
   }
 
+  if (pStream != NULL) {
     mndReleaseStream(pMnode, pStream);
+  }
+
   taosThreadMutexUnlock(&execInfo.lock);
 
   {

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

@@ -739,6 +739,8 @@ void mndTransSetSerial(STrans *pTrans) { pTrans->exec = TRN_EXEC_SERIAL; }
 
 void mndTransSetParallel(STrans *pTrans) { pTrans->exec = TRN_EXEC_PARALLEL; }
 
+void mndTransSetChangeless(STrans *pTrans) { pTrans->changeless = true; }
+
 void mndTransSetOper(STrans *pTrans, EOperType oper) { pTrans->oper = oper; }
 
 static int32_t mndTransSync(SMnode *pMnode, STrans *pTrans) {
@@ -862,7 +864,7 @@ int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans) {
     return -1;
   }
 
-  if (taosArrayGetSize(pTrans->commitActions) <= 0) {
+  if (!pTrans->changeless && taosArrayGetSize(pTrans->commitActions) <= 0) {
     terrno = TSDB_CODE_MND_TRANS_CLOG_IS_NULL;
     mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
     return -1;

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

@@ -2342,24 +2342,7 @@ int32_t mndAddVgroupBalanceToTrans(SMnode *pMnode, SVgObj *pVgroup, STrans *pTra
       return -1;
     }
 
-    if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, pVgroup) != 0) {
-      mError("trans:%d, vgid:%d failed to be balanced to dnode:%d", pTrans->id, vgid, dnodeId);
-      return -1;
-    }
-
     mndReleaseDb(pMnode, pDb);
-
-    SSdbRaw *pRaw = mndVgroupActionEncode(pVgroup);
-    if (pRaw == NULL) {
-      mError("trans:%d, vgid:%d failed to encode action to dnode:%d", pTrans->id, vgid, dnodeId);
-      return -1;
-    }
-    if (mndTransAppendCommitlog(pTrans, pRaw) != 0) {
-      sdbFreeRaw(pRaw);
-      mError("trans:%d, vgid:%d failed to append commit log dnode:%d", pTrans->id, vgid, dnodeId);
-      return -1;
-    }
-    (void)sdbSetRawStatus(pRaw, SDB_STATUS_READY);
   } else {
     mInfo("trans:%d, vgid:%d cant be balanced to dnode:%d, exist:%d, online:%d", pTrans->id, vgid, dnodeId, exist,
           online);

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

@@ -79,6 +79,7 @@ int32_t metaSnapRead(SMetaSnapReader* pReader, uint8_t** ppData) {
   int32_t     nKey = 0;
   int32_t     nData = 0;
   STbDbKey    key;
+  SMetaInfo   info;
 
   *ppData = NULL;
   for (;;) {
@@ -91,7 +92,8 @@ int32_t metaSnapRead(SMetaSnapReader* pReader, uint8_t** ppData) {
       goto _exit;
     }
 
-    if (key.version < pReader->sver) {
+    if (key.version < pReader->sver  //
+        || metaGetInfo(pReader->pMeta, key.uid, &info, NULL) == TSDB_CODE_NOT_FOUND) {
       tdbTbcMoveToNext(pReader->pTbc);
       continue;
     }

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

@@ -101,10 +101,7 @@ int32_t tqInitialize(STQ* pTq) {
     return -1;
   }
 
-  if (streamMetaLoadAllTasks(pTq->pStreamMeta) < 0) {
-    return -1;
-  }
-
+  /*int32_t code = */streamMetaLoadAllTasks(pTq->pStreamMeta);
   return 0;
 }
 

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

@@ -71,8 +71,8 @@ int32_t tqBuildDeleteReq(STQ* pTq, const char* stbFullName, const SSDataBlock* p
     if (varTbName != NULL && varTbName != (void*)-1) {
       name = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN);
       memcpy(name, varDataVal(varTbName), varDataLen(varTbName));
-      if (newSubTableRule && !isAutoTableName(name) && !alreadyAddGroupId(name) && groupId != 0) {
-        buildCtbNameAddGroupId(name, groupId);
+      if (newSubTableRule && !isAutoTableName(name) && !alreadyAddGroupId(name) && groupId != 0 && stbFullName) {
+        buildCtbNameAddGroupId(stbFullName, name, groupId);
       }
     } else if (stbFullName) {
       name = buildCtbNameByGroupId(stbFullName, groupId);
@@ -182,10 +182,10 @@ void setCreateTableMsgTableName(SVCreateTbReq* pCreateTableReq, SSDataBlock* pDa
                                 int64_t gid, bool newSubTableRule) {
   if (pDataBlock->info.parTbName[0]) {
     if (newSubTableRule && !isAutoTableName(pDataBlock->info.parTbName) &&
-        !alreadyAddGroupId(pDataBlock->info.parTbName) && gid != 0) {
+        !alreadyAddGroupId(pDataBlock->info.parTbName) && gid != 0 && stbFullName) {
       pCreateTableReq->name = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN);
       strcpy(pCreateTableReq->name, pDataBlock->info.parTbName);
-      buildCtbNameAddGroupId(pCreateTableReq->name, gid);
+      buildCtbNameAddGroupId(stbFullName, pCreateTableReq->name, gid);
 //      tqDebug("gen name from:%s", pDataBlock->info.parTbName);
     } else {
       pCreateTableReq->name = taosStrdup(pDataBlock->info.parTbName);
@@ -671,10 +671,14 @@ int32_t setDstTableDataUid(SVnode* pVnode, SStreamTask* pTask, SSDataBlock* pDat
       memset(dstTableName, 0, TSDB_TABLE_NAME_LEN);
       buildCtbNameByGroupIdImpl(stbFullName, groupId, dstTableName);
     } else {
-      if (pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER && pTask->subtableWithoutMd5 != 1 &&
-          !isAutoTableName(dstTableName) && !alreadyAddGroupId(dstTableName) && groupId != 0) {
+      if (pTask->subtableWithoutMd5 != 1 && !isAutoTableName(dstTableName) &&
+          !alreadyAddGroupId(dstTableName) && groupId != 0) {
         tqDebug("s-task:%s append groupId:%" PRId64 " for generated dstTable:%s", id, groupId, dstTableName);
-        buildCtbNameAddGroupId(dstTableName, groupId);
+        if(pTask->ver == SSTREAM_TASK_SUBTABLE_CHANGED_VER){
+          buildCtbNameAddGroupId(NULL, dstTableName, groupId);
+        }else if(pTask->ver > SSTREAM_TASK_SUBTABLE_CHANGED_VER && stbFullName) {
+          buildCtbNameAddGroupId(stbFullName, dstTableName, groupId);
+        }
       }
     }
 

source/dnode/vnode/src/tqCommon/tqCommon.c

@@ -850,12 +850,18 @@ int32_t tqStreamTaskProcessTaskResetReq(SStreamMeta* pMeta, SRpcMsg* pMsg) {
 
   tqDebug("s-task:%s receive task-reset msg from mnode, reset status and ready for data processing", pTask->id.idStr);
 
+  taosThreadMutexLock(&pTask->lock);
+
   // clear flag set during do checkpoint, and open inputQ for all upstream tasks
   if (streamTaskGetStatus(pTask)->state == TASK_STATUS__CK) {
+    tqDebug("s-task:%s reset task status from checkpoint, current checkpointingId:%" PRId64 ", transId:%d",
+            pTask->id.idStr, pTask->chkInfo.checkpointingId, pTask->chkInfo.transId);
     streamTaskClearCheckInfo(pTask, true);
     streamTaskSetStatusReady(pTask);
   }
 
+  taosThreadMutexUnlock(&pTask->lock);
+
   streamMetaReleaseTask(pMeta, pTask);
   return TSDB_CODE_SUCCESS;
 }

source/libs/executor/src/executil.c

@@ -642,7 +642,8 @@ int32_t getColInfoResultForGroupby(void* pVnode, SNodeList* group, STableListInf
     info->groupId = calcGroupId(keyBuf, len);
     if (initRemainGroups) {
       // groupId ~ table uid
-      taosHashPut(pTableListInfo->remainGroups, &(info->groupId), sizeof(info->groupId), &(info->uid), sizeof(info->uid));
+      taosHashPut(pTableListInfo->remainGroups, &(info->groupId), sizeof(info->groupId), &(info->uid),
+                  sizeof(info->uid));
     }
   }
 
@@ -939,7 +940,8 @@ SSDataBlock* createTagValBlockForFilter(SArray* pColList, int32_t numOfTables, S
   return pResBlock;
 }
 
-static int32_t doSetQualifiedUid(STableListInfo* pListInfo, SArray* pUidList, const SArray* pUidTagList, bool* pResultList, bool addUid) {
+static int32_t doSetQualifiedUid(STableListInfo* pListInfo, SArray* pUidList, const SArray* pUidTagList,
+                                 bool* pResultList, bool addUid) {
   taosArrayClear(pUidList);
 
   STableKeyInfo info = {.uid = 0, .groupId = 0};
@@ -1165,7 +1167,8 @@ int32_t getTableList(void* pVnode, SScanPhysiNode* pScanNode, SNode* pTagCond, S
         memcpy(pPayload + sizeof(int32_t), taosArrayGet(pUidList, 0), numOfTables * sizeof(uint64_t));
       }
 
-      pStorageAPI->metaFn.putCachedTableList(pVnode, pScanNode->suid, context.digest, tListLen(context.digest), pPayload, size, 1);
+      pStorageAPI->metaFn.putCachedTableList(pVnode, pScanNode->suid, context.digest, tListLen(context.digest),
+                                             pPayload, size, 1);
       digest[0] = 1;
       memcpy(digest + 1, context.digest, tListLen(context.digest));
     }
@@ -1725,7 +1728,8 @@ SColumn extractColumnFromColumnNode(SColumnNode* pColNode) {
   return c;
 }
 
-int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode, const SReadHandle* readHandle) {
+int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode,
+                               const SReadHandle* readHandle) {
   pCond->order = pTableScanNode->scanSeq[0] > 0 ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
   pCond->numOfCols = LIST_LENGTH(pTableScanNode->scan.pScanCols);
 
@@ -1748,8 +1752,7 @@ int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysi
 
   // allowed read stt file optimization mode
   pCond->notLoadData = (pTableScanNode->dataRequired == FUNC_DATA_REQUIRED_NOT_LOAD) &&
-                       (pTableScanNode->scan.node.pConditions == NULL) &&
-                       (pTableScanNode->interval == 0);
+                       (pTableScanNode->scan.node.pConditions == NULL) && (pTableScanNode->interval == 0);
 
   int32_t j = 0;
   for (int32_t i = 0; i < pCond->numOfCols; ++i) {
@@ -1891,7 +1894,8 @@ void getNextTimeWindow(const SInterval* pInterval, STimeWindow* tw, int32_t orde
   int32_t factor = GET_FORWARD_DIRECTION_FACTOR(order);
   slidingStart = taosTimeAdd(slidingStart, factor * pInterval->sliding, pInterval->slidingUnit, pInterval->precision);
   tw->skey = taosTimeAdd(slidingStart, pInterval->offset, pInterval->offsetUnit, pInterval->precision);
-  int64_t slidingEnd = taosTimeAdd(slidingStart, pInterval->interval, pInterval->intervalUnit, pInterval->precision) - 1;
+  int64_t slidingEnd =
+      taosTimeAdd(slidingStart, pInterval->interval, pInterval->intervalUnit, pInterval->precision) - 1;
   tw->ekey = taosTimeAdd(slidingEnd, pInterval->offset, pInterval->offsetUnit, pInterval->precision);
 }
 
@@ -2147,7 +2151,8 @@ int32_t buildGroupIdMapForAllTables(STableListInfo* pTableListInfo, SReadHandle*
     bool initRemainGroups = false;
     if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == nodeType(pScanNode)) {
       STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pScanNode;
-      if (tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable && !(groupSort || pScanNode->groupOrderScan)) {
+      if (tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable &&
+          !(groupSort || pScanNode->groupOrderScan)) {
         initRemainGroups = true;
       }
     }
@@ -2289,7 +2294,8 @@ void updateTimeWindowInfo(SColumnInfoData* pColData, STimeWindow* pWin, int64_t
   ts[4] = pWin->ekey + delta;  // window end key
 }
 
-int32_t compKeys(const SArray* pSortGroupCols, const char* oldkeyBuf, int32_t oldKeysLen, const SSDataBlock* pBlock, int32_t rowIndex) {
+int32_t compKeys(const SArray* pSortGroupCols, const char* oldkeyBuf, int32_t oldKeysLen, const SSDataBlock* pBlock,
+                 int32_t rowIndex) {
   SColumnDataAgg* pColAgg = NULL;
   const char*     isNull = oldkeyBuf;
   const char*     p = oldkeyBuf + sizeof(int8_t) * pSortGroupCols->size;
@@ -2321,8 +2327,7 @@ int32_t compKeys(const SArray* pSortGroupCols, const char* oldkeyBuf, int32_t ol
   return 0;
 }
 
-int32_t buildKeys(char* keyBuf, const SArray* pSortGroupCols, const SSDataBlock* pBlock,
-                 int32_t rowIndex) {
+int32_t buildKeys(char* keyBuf, const SArray* pSortGroupCols, const SSDataBlock* pBlock, int32_t rowIndex) {
   uint32_t        colNum = pSortGroupCols->size;
   SColumnDataAgg* pColAgg = NULL;
   char*           isNull = keyBuf;

source/libs/executor/src/sysscanoperator.c

@@ -879,7 +879,7 @@ static int32_t sysTableGetGeomText(char* iGeom, int32_t nGeom, char** output, in
   char*   outputWKT = NULL;
 
   if (nGeom == 0) {
-    if (!(*output = strdup(""))) code = TSDB_CODE_OUT_OF_MEMORY;
+    if (!(*output = taosStrdup(""))) code = TSDB_CODE_OUT_OF_MEMORY;
     *nOutput = 0;
     return code;
   }

source/libs/parser/inc/sql.y

@@ -747,16 +747,52 @@ insert_query(A) ::= INSERT INTO full_table_name(C) query_or_subquery(B).
 
 /************************************************ tags_literal *************************************************************/
 tags_literal(A) ::= NK_INTEGER(B).                                                { A = createRawValueNode(pCxt, TSDB_DATA_TYPE_UBIGINT, &B, NULL); }
+tags_literal(A) ::= NK_INTEGER(B) NK_PLUS duration_literal(C).                    {
+                                                                                    SToken l = B;
+                                                                                    SToken r = getTokenFromRawExprNode(pCxt, C);
+                                                                                    l.n = (r.z + r.n) - l.z;  
+                                                                                    A = createRawValueNodeExt(pCxt, TSDB_DATA_TYPE_BINARY, &l, NULL, C);
+                                                                                  }
+tags_literal(A) ::= NK_INTEGER(B) NK_MINUS duration_literal(C).                   {
+                                                                                    SToken l = B;
+                                                                                    SToken r = getTokenFromRawExprNode(pCxt, C);
+                                                                                    l.n = (r.z + r.n) - l.z;  
+                                                                                    A = createRawValueNodeExt(pCxt, TSDB_DATA_TYPE_BINARY, &l, NULL, C);
+                                                                                  }
 tags_literal(A) ::= NK_PLUS(B) NK_INTEGER(C).                                     {
                                                                                     SToken t = B;
                                                                                     t.n = (C.z + C.n) - B.z;
                                                                                     A = createRawValueNode(pCxt, TSDB_DATA_TYPE_UBIGINT, &t, NULL);
                                                                                   }
+tags_literal(A) ::= NK_PLUS(B) NK_INTEGER NK_PLUS duration_literal(C).            {
+                                                                                    SToken l = B;
+                                                                                    SToken r = getTokenFromRawExprNode(pCxt, C);
+                                                                                    l.n = (r.z + r.n) - l.z;  
+                                                                                    A = createRawValueNodeExt(pCxt, TSDB_DATA_TYPE_BINARY, &l, NULL, C);
+                                                                                  }
+tags_literal(A) ::= NK_PLUS(B) NK_INTEGER NK_MINUS duration_literal(C).           {
+                                                                                    SToken l = B;
+                                                                                    SToken r = getTokenFromRawExprNode(pCxt, C);
+                                                                                    l.n = (r.z + r.n) - l.z;  
+                                                                                    A = createRawValueNodeExt(pCxt, TSDB_DATA_TYPE_BINARY, &l, NULL, C);
+                                                                                  }
 tags_literal(A) ::= NK_MINUS(B) NK_INTEGER(C).                                    {
                                                                                     SToken t = B;
                                                                                     t.n = (C.z + C.n) - B.z;
                                                                                     A = createRawValueNode(pCxt, TSDB_DATA_TYPE_UBIGINT, &t, NULL);
                                                                                   }
+tags_literal(A) ::= NK_MINUS(B) NK_INTEGER NK_PLUS duration_literal(C).           {
+                                                                                    SToken l = B;
+                                                                                    SToken r = getTokenFromRawExprNode(pCxt, C);
+                                                                                    l.n = (r.z + r.n) - l.z;  
+                                                                                    A = createRawValueNodeExt(pCxt, TSDB_DATA_TYPE_BINARY, &l, NULL, C);
+                                                                                  }
+tags_literal(A) ::= NK_MINUS(B) NK_INTEGER NK_MINUS duration_literal(C).          {
+                                                                                    SToken l = B;
+                                                                                    SToken r = getTokenFromRawExprNode(pCxt, C);
+                                                                                    l.n = (r.z + r.n) - l.z;  
+                                                                                    A = createRawValueNodeExt(pCxt, TSDB_DATA_TYPE_BINARY, &l, NULL, C);
+                                                                                  }
 tags_literal(A) ::= NK_FLOAT(B).                                                  { A = createRawValueNode(pCxt, TSDB_DATA_TYPE_DOUBLE, &B, NULL); }
 tags_literal(A) ::= NK_PLUS(B) NK_FLOAT(C).                                       {
                                                                                     SToken t = B;
@@ -770,29 +806,113 @@ tags_literal(A) ::= NK_MINUS(B) NK_FLOAT(C).
                                                                                   }
 
 tags_literal(A) ::= NK_BIN(B).                                                    { A = createRawValueNode(pCxt, TSDB_DATA_TYPE_UBIGINT, &B, NULL); }
+tags_literal(A) ::= NK_BIN(B) NK_PLUS duration_literal(C).                        {
+                                                                                    SToken l = B;
+                                                                                    SToken r = getTokenFromRawExprNode(pCxt, C);
+                                                                                    l.n = (r.z + r.n) - l.z;  
+                                                                                    A = createRawValueNodeExt(pCxt, TSDB_DATA_TYPE_BINARY, &l, NULL, C);
+                                                                                  }
+tags_literal(A) ::= NK_BIN(B) NK_MINUS duration_literal(C).                       {
+                                                                                    SToken l = B;
+                                                                                    SToken r = getTokenFromRawExprNode(pCxt, C);
+                                                                                    l.n = (r.z + r.n) - l.z;  
+                                                                                    A = createRawValueNodeExt(pCxt, TSDB_DATA_TYPE_BINARY, &l, NULL, C);
+                                                                                  }
 tags_literal(A) ::= NK_PLUS(B) NK_BIN(C).                                         {
                                                                                     SToken t = B;
                                                                                     t.n = (C.z + C.n) - B.z;
                                                                                     A = createRawValueNode(pCxt, TSDB_DATA_TYPE_UBIGINT, &t, NULL);
                                                                                   }
+tags_literal(A) ::= NK_PLUS(B) NK_BIN NK_PLUS duration_literal(C).                {
+                                                                                    SToken l = B;
+                                                                                    SToken r = getTokenFromRawExprNode(pCxt, C);
+                                                                                    l.n = (r.z + r.n) - l.z;  
+                                                                                    A = createRawValueNodeExt(pCxt, TSDB_DATA_TYPE_BINARY, &l, NULL, C);
+                                                                                  }
+tags_literal(A) ::= NK_PLUS(B) NK_BIN NK_MINUS duration_literal(C).               {
+                                                                                    SToken l = B;
+                                                                                    SToken r = getTokenFromRawExprNode(pCxt, C);
+                                                                                    l.n = (r.z + r.n) - l.z;  
+                                                                                    A = createRawValueNodeExt(pCxt, TSDB_DATA_TYPE_BINARY, &l, NULL, C);
+                                                                                  }
 tags_literal(A) ::= NK_MINUS(B) NK_BIN(C).                                        {
                                                                                     SToken t = B;
                                                                                     t.n = (C.z + C.n) - B.z;
                                                                                     A = createRawValueNode(pCxt, TSDB_DATA_TYPE_UBIGINT, &t, NULL);
                                                                                   }
+tags_literal(A) ::= NK_MINUS(B) NK_BIN NK_PLUS duration_literal(C).               {
+                                                                                    SToken l = B;
+                                                                                    SToken r = getTokenFromRawExprNode(pCxt, C);
+                                                                                    l.n = (r.z + r.n) - l.z;  
+                                                                                    A = createRawValueNodeExt(pCxt, TSDB_DATA_TYPE_BINARY, &l, NULL, C);
+                                                                                  }
+tags_literal(A) ::= NK_MINUS(B) NK_BIN NK_MINUS duration_literal(C).              {
+                                                                                    SToken l = B;
+                                                                                    SToken r = getTokenFromRawExprNode(pCxt, C);
+                                                                                    l.n = (r.z + r.n) - l.z;  
+                                                                                    A = createRawValueNodeExt(pCxt, TSDB_DATA_TYPE_BINARY, &l, NULL, C);
+                                                                                  }
 tags_literal(A) ::= NK_HEX(B).                                                    { A = createRawValueNode(pCxt, TSDB_DATA_TYPE_UBIGINT, &B, NULL); }
+tags_literal(A) ::= NK_HEX(B) NK_PLUS duration_literal(C).                        {
+                                                                                    SToken l = B;
+                                                                                    SToken r = getTokenFromRawExprNode(pCxt, C);
+                                                                                    l.n = (r.z + r.n) - l.z;  
+                                                                                    A = createRawValueNodeExt(pCxt, TSDB_DATA_TYPE_BINARY, &l, NULL, C);
+                                                                                  }
+tags_literal(A) ::= NK_HEX(B) NK_MINUS duration_literal(C).                       {
+                                                                                    SToken l = B;
+                                                                                    SToken r = getTokenFromRawExprNode(pCxt, C);
+                                                                                    l.n = (r.z + r.n) - l.z;  
+                                                                                    A = createRawValueNodeExt(pCxt, TSDB_DATA_TYPE_BINARY, &l, NULL, C);
+                                                                                  }
 tags_literal(A) ::= NK_PLUS(B) NK_HEX(C).                                         {
                                                                                     SToken t = B;
                                                                                     t.n = (C.z + C.n) - B.z;
                                                                                     A = createRawValueNode(pCxt, TSDB_DATA_TYPE_UBIGINT, &t, NULL);
                                                                                   }
+tags_literal(A) ::= NK_PLUS(B) NK_HEX NK_PLUS duration_literal(C).                {
+                                                                                    SToken l = B;
+                                                                                    SToken r = getTokenFromRawExprNode(pCxt, C);
+                                                                                    l.n = (r.z + r.n) - l.z;  
+                                                                                    A = createRawValueNodeExt(pCxt, TSDB_DATA_TYPE_BINARY, &l, NULL, C);
+                                                                                  }
+tags_literal(A) ::= NK_PLUS(B) NK_HEX NK_MINUS duration_literal(C).               {
+                                                                                    SToken l = B;
+                                                                                    SToken r = getTokenFromRawExprNode(pCxt, C);
+                                                                                    l.n = (r.z + r.n) - l.z;  
+                                                                                    A = createRawValueNodeExt(pCxt, TSDB_DATA_TYPE_BINARY, &l, NULL, C);
+                                                                                  }
 tags_literal(A) ::= NK_MINUS(B) NK_HEX(C).                                        {
                                                                                     SToken t = B;
                                                                                     t.n = (C.z + C.n) - B.z;
                                                                                     A = createRawValueNode(pCxt, TSDB_DATA_TYPE_UBIGINT, &t, NULL);
                                                                                   }
+tags_literal(A) ::= NK_MINUS(B) NK_HEX NK_PLUS duration_literal(C).               {
+                                                                                    SToken l = B;
+                                                                                    SToken r = getTokenFromRawExprNode(pCxt, C);
+                                                                                    l.n = (r.z + r.n) - l.z;  
+                                                                                    A = createRawValueNodeExt(pCxt, TSDB_DATA_TYPE_BINARY, &l, NULL, C);
+                                                                                  }
+tags_literal(A) ::= NK_MINUS(B) NK_HEX NK_MINUS duration_literal(C).              {
+                                                                                    SToken l = B;
+                                                                                    SToken r = getTokenFromRawExprNode(pCxt, C);
+                                                                                    l.n = (r.z + r.n) - l.z;  
+                                                                                    A = createRawValueNodeExt(pCxt, TSDB_DATA_TYPE_BINARY, &l, NULL, C);
+                                                                                  }
 
 tags_literal(A) ::= NK_STRING(B).                                                 { A = createRawValueNode(pCxt, TSDB_DATA_TYPE_BINARY, &B, NULL); }
+tags_literal(A) ::= NK_STRING(B) NK_PLUS duration_literal(C).                     {
+                                                                                    SToken l = B;
+                                                                                    SToken r = getTokenFromRawExprNode(pCxt, C);
+                                                                                    l.n = (r.z + r.n) - l.z;  
+                                                                                    A = createRawValueNodeExt(pCxt, TSDB_DATA_TYPE_BINARY, &l, NULL, C);
+                                                                                  }
+tags_literal(A) ::= NK_STRING(B) NK_MINUS duration_literal(C).                    {
+                                                                                    SToken l = B;
+                                                                                    SToken r = getTokenFromRawExprNode(pCxt, C);
+                                                                                    l.n = (r.z + r.n) - l.z;  
+                                                                                    A = createRawValueNodeExt(pCxt, TSDB_DATA_TYPE_BINARY, &l, NULL, C);
+                                                                                  }
 tags_literal(A) ::= NK_BOOL(B).                                                   { A = createRawValueNode(pCxt, TSDB_DATA_TYPE_BOOL, &B, NULL); }
 tags_literal(A) ::= NULL(B).                                                      { A = createRawValueNode(pCxt, TSDB_DATA_TYPE_NULL, &B, NULL); }
 

source/libs/scheduler/src/schRemote.c

@@ -486,6 +486,7 @@ int32_t schHandleNotifyCallback(void *param, SDataBuf *pMsg, int32_t code) {
   SSchTaskCallbackParam *pParam = (SSchTaskCallbackParam *)param;
   qDebug("QID:0x%" PRIx64 ",TID:0x%" PRIx64 " task notify rsp received, code:0x%x", pParam->queryId, pParam->taskId,
          code);
+  rpcReleaseHandle(pMsg->handle, TAOS_CONN_CLIENT);
   if (pMsg) {
     taosMemoryFree(pMsg->pData);
     taosMemoryFree(pMsg->pEpSet);
@@ -526,6 +527,7 @@ int32_t schHandleHbCallback(void *param, SDataBuf *pMsg, int32_t code) {
 
   if (code) {
     qError("hb rsp error:%s", tstrerror(code));
+    rpcReleaseHandle(pMsg->handle, TAOS_CONN_CLIENT);
     SCH_ERR_JRET(code);
   }
 
@@ -1181,7 +1183,7 @@ int32_t schBuildAndSendMsg(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *addr,
       qMsg.queryId = pJob->queryId;
       qMsg.taskId = pTask->taskId;
       qMsg.refId = pJob->refId;
-      qMsg.execId = pTask->execId;
+      qMsg.execId = *(int32_t*)param;
 
       msgSize = tSerializeSTaskDropReq(NULL, 0, &qMsg);
       if (msgSize < 0) {

source/libs/scheduler/src/schTask.c

@@ -371,7 +371,6 @@ int32_t schChkUpdateRedirectCtx(SSchJob *pJob, SSchTask *pTask, SEpSet *pEpSet,
     pCtx->roundTotal = pEpSet->numOfEps;
   }
 
-
   if (pCtx->roundTimes >= pCtx->roundTotal) {
     int64_t nowTs = taosGetTimestampMs();
     int64_t lastTime = nowTs - pCtx->startTs;
@@ -862,7 +861,9 @@ void schDropTaskOnExecNode(SSchJob *pJob, SSchTask *pTask) {
   while (nodeInfo) {
     if (nodeInfo->handle) {
       SCH_SET_TASK_HANDLE(pTask, nodeInfo->handle);
-      schBuildAndSendMsg(pJob, pTask, &nodeInfo->addr, TDMT_SCH_DROP_TASK, NULL);
+      void *pExecId = taosHashGetKey(nodeInfo, NULL);
+      schBuildAndSendMsg(pJob, pTask, &nodeInfo->addr, TDMT_SCH_DROP_TASK, pExecId);
+
       SCH_TASK_DLOG("start to drop task's %dth execNode", i);
     } else {
       SCH_TASK_DLOG("no need to drop task %dth execNode", i);
@@ -901,7 +902,6 @@ int32_t schNotifyTaskOnExecNode(SSchJob *pJob, SSchTask *pTask, ETaskNotifyType
   return TSDB_CODE_SUCCESS;
 }
 
-
 int32_t schProcessOnTaskStatusRsp(SQueryNodeEpId *pEpId, SArray *pStatusList) {
   int32_t   taskNum = (int32_t)taosArrayGetSize(pStatusList);
   SSchTask *pTask = NULL;
@@ -1289,7 +1289,6 @@ int32_t schNotifyTaskInHashList(SSchJob *pJob, SHashObj *list, ETaskNotifyType t
   SCH_RET(code);
 }
 
-
 int32_t schExecRemoteFetch(SSchJob *pJob, SSchTask *pTask) {
   SCH_RET(schBuildAndSendMsg(pJob, pJob->fetchTask, &pJob->resNode, SCH_FETCH_TYPE(pJob->fetchTask), NULL));
 }

source/libs/stream/src/streamCheckpoint.c

@@ -278,6 +278,7 @@ void streamTaskClearCheckInfo(SStreamTask* pTask, bool clearChkpReadyMsg) {
   pTask->chkInfo.numOfNotReady = 0;
   pTask->chkInfo.transId = 0;
   pTask->chkInfo.dispatchCheckpointTrigger = false;
+  pTask->chkInfo.downstreamAlignNum = 0;
 
   streamTaskOpenAllUpstreamInput(pTask);  // open inputQ for all upstream tasks
   if (clearChkpReadyMsg) {

source/libs/stream/src/streamDispatch.c

@@ -321,6 +321,8 @@ void clearBufferedDispatchMsg(SStreamTask* pTask) {
     destroyDispatchMsg(pMsgInfo->pData, getNumOfDispatchBranch(pTask));
   }
 
+  pMsgInfo->checkpointId = -1;
+  pMsgInfo->transId = -1;
   pMsgInfo->pData = NULL;
   pMsgInfo->dispatchMsgType = 0;
 }
@@ -332,6 +334,12 @@ static int32_t doBuildDispatchMsg(SStreamTask* pTask, const SStreamDataBlock* pD
 
   pTask->msgInfo.dispatchMsgType = pData->type;
 
+  if (pData->type == STREAM_INPUT__CHECKPOINT_TRIGGER) {
+    SSDataBlock* p = taosArrayGet(pData->blocks, 0);
+    pTask->msgInfo.checkpointId = p->info.version;
+    pTask->msgInfo.transId = p->info.window.ekey;
+  }
+
   if (pTask->outputInfo.type == TASK_OUTPUT__FIXED_DISPATCH) {
     SStreamDispatchReq* pReq = taosMemoryCalloc(1, sizeof(SStreamDispatchReq));
 
@@ -580,12 +588,15 @@ int32_t streamSearchAndAddBlock(SStreamTask* pTask, SStreamDispatchReq* pReqs, S
   } else {
     char ctbName[TSDB_TABLE_FNAME_LEN] = {0};
     if (pDataBlock->info.parTbName[0]) {
-      if(pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER &&
-          pTask->subtableWithoutMd5 != 1 &&
+      if(pTask->subtableWithoutMd5 != 1 &&
           !isAutoTableName(pDataBlock->info.parTbName) &&
           !alreadyAddGroupId(pDataBlock->info.parTbName) &&
           groupId != 0){
-        buildCtbNameAddGroupId(pDataBlock->info.parTbName, groupId);
+        if(pTask->ver == SSTREAM_TASK_SUBTABLE_CHANGED_VER){
+          buildCtbNameAddGroupId(NULL, pDataBlock->info.parTbName, groupId);
+        }else if(pTask->ver > SSTREAM_TASK_SUBTABLE_CHANGED_VER) {
+          buildCtbNameAddGroupId(pTask->outputInfo.shuffleDispatcher.stbFullName, pDataBlock->info.parTbName, groupId);
+        }
       }
     } else {
       buildCtbNameByGroupIdImpl(pTask->outputInfo.shuffleDispatcher.stbFullName, groupId, pDataBlock->info.parTbName);
@@ -947,9 +958,21 @@ void streamClearChkptReadyMsg(SStreamTask* pTask) {
 // this message has been sent successfully, let's try next one.
 static int32_t handleDispatchSuccessRsp(SStreamTask* pTask, int32_t downstreamId) {
   stDebug("s-task:%s destroy dispatch msg:%p", pTask->id.idStr, pTask->msgInfo.pData);
+
   bool delayDispatch = (pTask->msgInfo.dispatchMsgType == STREAM_INPUT__CHECKPOINT_TRIGGER);
   if (delayDispatch) {
+    taosThreadMutexLock(&pTask->lock);
+    // we only set the dispatch msg info for current checkpoint trans
+    if (streamTaskGetStatus(pTask)->state == TASK_STATUS__CK && pTask->chkInfo.checkpointingId == pTask->msgInfo.checkpointId) {
+      ASSERT(pTask->chkInfo.transId == pTask->msgInfo.transId);
       pTask->chkInfo.dispatchCheckpointTrigger = true;
+      stDebug("s-task:%s checkpoint-trigger msg rsp for checkpointId:%" PRId64 " transId:%d confirmed",
+             pTask->id.idStr, pTask->msgInfo.checkpointId, pTask->msgInfo.transId);
+    } else {
+      stWarn("s-task:%s checkpoint-trigger msg rsp for checkpointId:%" PRId64 " transId:%d discard, since expired",
+             pTask->id.idStr, pTask->msgInfo.checkpointId, pTask->msgInfo.transId);
+    }
+    taosThreadMutexUnlock(&pTask->lock);
   }
 
   clearBufferedDispatchMsg(pTask);

source/libs/stream/src/streamMeta.c

@@ -542,7 +542,6 @@ int32_t streamMetaSaveTask(SStreamMeta* pMeta, SStreamTask* pTask) {
   void*   buf = NULL;
   int32_t len;
   int32_t code;
-  pTask->ver = SSTREAM_TASK_VER;
   tEncodeSize(tEncodeStreamTask, pTask, len, code);
   if (code < 0) {
     return -1;
@@ -552,6 +551,9 @@ int32_t streamMetaSaveTask(SStreamMeta* pMeta, SStreamTask* pTask) {
     return -1;
   }
 
+  if (pTask->ver < SSTREAM_TASK_SUBTABLE_CHANGED_VER){
+    pTask->ver = SSTREAM_TASK_VER;
+  }
   SEncoder encoder = {0};
   tEncoderInit(&encoder, buf, len);
   tEncodeStreamTask(&encoder, pTask);
@@ -648,14 +650,17 @@ SStreamTask* streamMetaAcquireOneTask(SStreamTask* pTask) {
 }
 
 void streamMetaReleaseTask(SStreamMeta* UNUSED_PARAM(pMeta), SStreamTask* pTask) {
+  int32_t taskId = pTask->id.taskId;
   int32_t ref = atomic_sub_fetch_32(&pTask->refCnt, 1);
+
+  // not safe to use the pTask->id.idStr, since pTask may be released by other threads when print logs.
   if (ref > 0) {
-    stTrace("s-task:%s release task, ref:%d", pTask->id.idStr, ref);
+    stTrace("s-task:0x%x release task, ref:%d", taskId, ref);
   } else if (ref == 0) {
-    stTrace("s-task:%s all refs are gone, free it", pTask->id.idStr);
+    stTrace("s-task:0x%x all refs are gone, free it", taskId);
     tFreeStreamTask(pTask);
   } else if (ref < 0) {
-    stError("task ref is invalid, ref:%d, %s", ref, pTask->id.idStr);
+    stError("task ref is invalid, ref:%d, 0x%x", ref, taskId);
   }
 }
 
@@ -824,13 +829,6 @@ int64_t streamMetaGetLatestCheckpointId(SStreamMeta* pMeta) {
   return chkpId;
 }
 
-static void doClear(void* pKey, void* pVal, TBC* pCur, SArray* pRecycleList) {
-  tdbFree(pKey);
-  tdbFree(pVal);
-  tdbTbcClose(pCur);
-  taosArrayDestroy(pRecycleList);
-}
-
 int32_t streamMetaLoadAllTasks(SStreamMeta* pMeta) {
   TBC*     pCur = NULL;
   void*    pKey = NULL;
@@ -847,10 +845,11 @@ int32_t streamMetaLoadAllTasks(SStreamMeta* pMeta) {
   int32_t vgId = pMeta->vgId;
   stInfo("vgId:%d load stream tasks from meta files", vgId);
 
-  if (tdbTbcOpen(pMeta->pTaskDb, &pCur, NULL) < 0) {
-    stError("vgId:%d failed to open stream meta, code:%s", vgId, tstrerror(terrno));
+  int32_t code = tdbTbcOpen(pMeta->pTaskDb, &pCur, NULL);
+  if (code != TSDB_CODE_SUCCESS) {
+    stError("vgId:%d failed to open stream meta, code:%s, not load any stream tasks", vgId, tstrerror(terrno));
     taosArrayDestroy(pRecycleList);
-    return -1;
+    return TSDB_CODE_SUCCESS;
   }
 
   tdbTbcMoveToFirst(pCur);
@@ -859,20 +858,18 @@ int32_t streamMetaLoadAllTasks(SStreamMeta* pMeta) {
     if (pTask == NULL) {
       terrno = TSDB_CODE_OUT_OF_MEMORY;
       stError("vgId:%d failed to load stream task from meta-files, code:%s", vgId, tstrerror(terrno));
-      doClear(pKey, pVal, pCur, pRecycleList);
-      return -1;
+      break;
     }
 
     tDecoderInit(&decoder, (uint8_t*)pVal, vLen);
     if (tDecodeStreamTask(&decoder, pTask) < 0) {
       tDecoderClear(&decoder);
-      doClear(pKey, pVal, pCur, pRecycleList);
       tFreeStreamTask(pTask);
       stError(
           "vgId:%d stream read incompatible data, rm %s/vnode/vnode*/tq/stream if taosd cannot start, and rebuild "
           "stream manually",
           vgId, tsDataDir);
-      return -1;
+      break;
     }
     tDecoderClear(&decoder);
 
@@ -892,10 +889,11 @@ int32_t streamMetaLoadAllTasks(SStreamMeta* pMeta) {
     STaskId id = {.streamId = pTask->id.streamId, .taskId = pTask->id.taskId};
     void*   p = taosHashGet(pMeta->pTasksMap, &id, sizeof(id));
     if (p == NULL) {
-      if (pMeta->expandFunc(pMeta->ahandle, pTask, pTask->chkInfo.checkpointVer + 1) < 0) {
-        doClear(pKey, pVal, pCur, pRecycleList);
+      code = pMeta->expandFunc(pMeta->ahandle, pTask, pTask->chkInfo.checkpointVer + 1);
+      if (code < 0) {
+        stError("failed to expand s-task:0x%"PRIx64", code:%s, continue", id.taskId, tstrerror(terrno));
         tFreeStreamTask(pTask);
-        return -1;
+        continue;
       }
 
       taosArrayPush(pMeta->pTaskList, &pTask->id);
@@ -907,9 +905,10 @@ int32_t streamMetaLoadAllTasks(SStreamMeta* pMeta) {
     }
 
     if (taosHashPut(pMeta->pTasksMap, &id, sizeof(id), &pTask, POINTER_BYTES) < 0) {
-      doClear(pKey, pVal, pCur, pRecycleList);
+      stError("s-task:0x%x failed to put into hashTable, code:%s, continue", pTask->id.taskId, tstrerror(terrno));
+      taosArrayPop(pMeta->pTaskList);
       tFreeStreamTask(pTask);
-      return -1;
+      continue;
     }
 
     if (pTask->info.fillHistory == 0) {
@@ -925,10 +924,9 @@ int32_t streamMetaLoadAllTasks(SStreamMeta* pMeta) {
 
   tdbFree(pKey);
   tdbFree(pVal);
+
   if (tdbTbcClose(pCur) < 0) {
-    stError("vgId:%d failed to close meta-file cursor", vgId);
-    taosArrayDestroy(pRecycleList);
-    return -1;
+    stError("vgId:%d failed to close meta-file cursor, code:%s, continue", vgId, tstrerror(terrno));
   }
 
   if (taosArrayGetSize(pRecycleList) > 0) {
@@ -942,8 +940,9 @@ int32_t streamMetaLoadAllTasks(SStreamMeta* pMeta) {
   ASSERT(pMeta->numOfStreamTasks <= numOfTasks && pMeta->numOfPausedTasks <= numOfTasks);
   stDebug("vgId:%d load %d tasks into meta from disk completed, streamTask:%d, paused:%d", pMeta->vgId, numOfTasks,
           pMeta->numOfStreamTasks, pMeta->numOfPausedTasks);
+
   taosArrayDestroy(pRecycleList);
-  return 0;
+  return TSDB_CODE_SUCCESS;
 }
 
 int32_t tEncodeStreamHbMsg(SEncoder* pEncoder, const SStreamHbMsg* pReq) {

source/libs/stream/src/streamTask.c

@@ -216,7 +216,7 @@ int32_t tDecodeStreamTask(SDecoder* pDecoder, SStreamTask* pTask) {
 
   if (tStartDecode(pDecoder) < 0) return -1;
   if (tDecodeI64(pDecoder, &pTask->ver) < 0) return -1;
-  if (pTask->ver <= SSTREAM_TASK_INCOMPATIBLE_VER) return -1;
+  if (pTask->ver <= SSTREAM_TASK_INCOMPATIBLE_VER || pTask->ver > SSTREAM_TASK_VER) return -1;
 
   if (tDecodeI64(pDecoder, &pTask->id.streamId) < 0) return -1;
   if (tDecodeI32(pDecoder, &pTask->id.taskId) < 0) return -1;
@@ -287,7 +287,9 @@ int32_t tDecodeStreamTask(SDecoder* pDecoder, SStreamTask* pTask) {
     if (tDecodeCStrTo(pDecoder, pTask->outputInfo.shuffleDispatcher.stbFullName) < 0) return -1;
   }
   if (tDecodeI64(pDecoder, &pTask->info.triggerParam) < 0) return -1;
+  if (pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER){
     if (tDecodeI8(pDecoder, &pTask->subtableWithoutMd5) < 0) return -1;
+  }
   if (tDecodeCStrTo(pDecoder, pTask->reserve) < 0) return -1;
 
   tEndDecode(pDecoder);

source/libs/stream/src/streamTaskSm.c

@@ -543,8 +543,6 @@ void streamTaskSetStatusReady(SStreamTask* pTask) {
     return;
   }
 
-  taosThreadMutexLock(&pTask->lock);
-
   pSM->prev.state = pSM->current;
   pSM->prev.evt = 0;
 
@@ -552,8 +550,6 @@ void streamTaskSetStatusReady(SStreamTask* pTask) {
   pSM->startTs = taosGetTimestampMs();
   pSM->pActiveTrans = NULL;
   taosArrayClear(pSM->pWaitingEventList);
-
-  taosThreadMutexUnlock(&pTask->lock);
 }
 
 STaskStateTrans createStateTransform(ETaskStatus current, ETaskStatus next, EStreamTaskEvent event, __state_trans_fn fn,

source/libs/sync/src/syncPipeline.c

@@ -866,9 +866,14 @@ int32_t syncLogReplRecover(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncAppendEn
   SyncTerm  term = -1;
   SyncIndex firstVer = pNode->pLogStore->syncLogBeginIndex(pNode->pLogStore);
   SyncIndex index = TMIN(pMsg->matchIndex, pNode->pLogBuf->matchIndex);
+  errno = 0;
 
   if (pMsg->matchIndex < pNode->pLogBuf->matchIndex) {
     term = syncLogReplGetPrevLogTerm(pMgr, pNode, index + 1);
+    if (term < 0 && (errno == ENFILE || errno == EMFILE)) {
+      sError("vgId:%d, failed to get prev log term since %s. index:%" PRId64, pNode->vgId, terrstr(), index + 1);
+      return -1;
+    }
     if ((index + 1 < firstVer) || (term < 0) ||
         (term != pMsg->lastMatchTerm && (index + 1 == firstVer || index == firstVer))) {
       ASSERT(term >= 0 || terrno == TSDB_CODE_WAL_LOG_NOT_EXIST);

source/libs/transport/inc/transComm.h

@@ -19,16 +19,12 @@ extern "C" {
 #endif
 
 #include <uv.h>
-#include "os.h"
-#include "taoserror.h"
 #include "theap.h"
-#include "tmisce.h"
 #include "tmsg.h"
 #include "transLog.h"
 #include "transportInt.h"
 #include "trpc.h"
 #include "ttrace.h"
-#include "tutil.h"
 
 typedef bool (*FilteFunc)(void* arg);
 
@@ -118,6 +114,9 @@ typedef struct SExHandle {
   void*    handle;
   int64_t  refId;
   void*    pThrd;
+  queue    q;
+  int8_t   inited;
+  SRWLatch latch;
 } SExHandle;
 
 typedef struct {

source/libs/transport/src/transCli.c

@@ -92,6 +92,7 @@ typedef struct SCliMsg {
   int64_t  refId;
   uint64_t st;
   int      sent;  //(0: no send, 1: alread sent)
+  queue    seqq;
 } SCliMsg;
 
 typedef struct SCliThrd {
@@ -121,11 +122,7 @@ typedef struct SCliThrd {
   SHashObj* batchCache;
 
   SCliMsg* stopMsg;
-
   bool     quit;
-
-  int       newConnCount;
-  SHashObj* msgCount;
 } SCliThrd;
 
 typedef struct SCliObj {
@@ -262,10 +259,8 @@ static void cliWalkCb(uv_handle_t* handle, void* arg);
     }                                                                     \
     if (i == sz) {                                                        \
       pMsg = NULL;                                                        \
-      tDebug("msg not found, %" PRIu64 "", ahandle);                      \
     } else {                                                              \
       pMsg = transQueueRm(&conn->cliMsgs, i);                             \
-      tDebug("msg found, %" PRIu64 "", ahandle);                          \
     }                                                                     \
   } while (0)
 
@@ -343,6 +338,34 @@ bool cliMaySendCachedMsg(SCliConn* conn) {
 _RETURN:
   return false;
 }
+bool cliConnSendSeqMsg(int64_t refId, SCliConn* conn) {
+  if (refId == 0) return false;
+  SExHandle* exh = transAcquireExHandle(transGetRefMgt(), refId);
+  if (exh == NULL) {
+    tDebug("release conn %p, refId: %" PRId64 "", conn, refId);
+    return false;
+  }
+  taosWLockLatch(&exh->latch);
+  if (exh->handle == NULL) exh->handle = conn;
+  exh->inited = 1;
+  if (!QUEUE_IS_EMPTY(&exh->q)) {
+    queue* h = QUEUE_HEAD(&exh->q);
+    QUEUE_REMOVE(h);
+    taosWUnLockLatch(&exh->latch);
+    SCliMsg* t = QUEUE_DATA(h, SCliMsg, seqq);
+    transCtxMerge(&conn->ctx, &t->ctx->appCtx);
+    transQueuePush(&conn->cliMsgs, t);
+    tDebug("pop from conn %p, refId: %" PRId64 "", conn, refId);
+    transReleaseExHandle(transGetRefMgt(), refId);
+    cliSend(conn);
+    return true;
+  }
+  taosWUnLockLatch(&exh->latch);
+  tDebug("empty conn %p, refId: %" PRId64 "", conn, refId);
+  transReleaseExHandle(transGetRefMgt(), refId);
+  return false;
+}
+
 void cliHandleResp(SCliConn* conn) {
   SCliThrd* pThrd = conn->hostThrd;
   STrans*   pTransInst = pThrd->pTransInst;
@@ -439,8 +462,14 @@ void cliHandleResp(SCliConn* conn) {
       return;
     }
   }
+  int64_t refId = (pMsg == NULL ? 0 : (int64_t)(pMsg->msg.info.handle));
+  tDebug("conn %p msg refId: %" PRId64 "", conn, refId);
   destroyCmsg(pMsg);
 
+  if (cliConnSendSeqMsg(refId, conn)) {
+    return;
+  }
+
   if (cliMaySendCachedMsg(conn) == true) {
     return;
   }
@@ -451,6 +480,21 @@ void cliHandleResp(SCliConn* conn) {
 
   uv_read_start((uv_stream_t*)conn->stream, cliAllocRecvBufferCb, cliRecvCb);
 }
+static void cliDestroyMsgInExhandle(int64_t refId) {
+  if (refId == 0) return;
+  SExHandle* exh = transAcquireExHandle(transGetRefMgt(), refId);
+  if (exh) {
+    taosWLockLatch(&exh->latch);
+    while (!QUEUE_IS_EMPTY(&exh->q)) {
+      queue* h = QUEUE_HEAD(&exh->q);
+      QUEUE_REMOVE(h);
+      SCliMsg* t = QUEUE_DATA(h, SCliMsg, seqq);
+      destroyCmsg(t);
+    }
+    taosWUnLockLatch(&exh->latch);
+    transReleaseExHandle(transGetRefMgt(), refId);
+  }
+}
 
 void cliHandleExceptImpl(SCliConn* pConn, int32_t code) {
   if (transQueueEmpty(&pConn->cliMsgs)) {
@@ -510,6 +554,8 @@ void cliHandleExceptImpl(SCliConn* pConn, int32_t code) {
     }
 
     if (pMsg == NULL || (pMsg && pMsg->type != Release)) {
+      int64_t refId = (pMsg == NULL ? 0 : (int64_t)(pMsg->msg.info.handle));
+      cliDestroyMsgInExhandle(refId);
       if (cliAppCb(pConn, &transMsg, pMsg) != 0) {
         return;
       }
@@ -585,11 +631,12 @@ void* destroyConnPool(SCliThrd* pThrd) {
 static SCliConn* getConnFromPool(SCliThrd* pThrd, char* key, bool* exceed) {
   void*      pool = pThrd->pool;
   STrans*    pTranInst = pThrd->pTransInst;
-  SConnList* plist = taosHashGet((SHashObj*)pool, key, strlen(key) + 1);
+  size_t     klen = strlen(key);
+  SConnList* plist = taosHashGet((SHashObj*)pool, key, klen);
   if (plist == NULL) {
     SConnList list = {0};
-    taosHashPut((SHashObj*)pool, key, strlen(key) + 1, (void*)&list, sizeof(list));
-    plist = taosHashGet(pool, key, strlen(key) + 1);
+    taosHashPut((SHashObj*)pool, key, klen, (void*)&list, sizeof(list));
+    plist = taosHashGet(pool, key, klen);
 
     SMsgList* nList = taosMemoryCalloc(1, sizeof(SMsgList));
     QUEUE_INIT(&nList->msgQ);
@@ -624,11 +671,12 @@ static SCliConn* getConnFromPool(SCliThrd* pThrd, char* key, bool* exceed) {
 static SCliConn* getConnFromPool2(SCliThrd* pThrd, char* key, SCliMsg** pMsg) {
   void*      pool = pThrd->pool;
   STrans*    pTransInst = pThrd->pTransInst;
-  SConnList* plist = taosHashGet((SHashObj*)pool, key, strlen(key) + 1);
+  size_t     klen = strlen(key);
+  SConnList* plist = taosHashGet((SHashObj*)pool, key, klen);
   if (plist == NULL) {
     SConnList list = {0};
-    taosHashPut((SHashObj*)pool, key, strlen(key) + 1, (void*)&list, sizeof(list));
-    plist = taosHashGet(pool, key, strlen(key) + 1);
+    taosHashPut((SHashObj*)pool, key, klen, (void*)&list, sizeof(list));
+    plist = taosHashGet(pool, key, klen);
 
     SMsgList* nList = taosMemoryCalloc(1, sizeof(SMsgList));
     QUEUE_INIT(&nList->msgQ);
@@ -676,7 +724,7 @@ static SCliConn* getConnFromPool2(SCliThrd* pThrd, char* key, SCliMsg** pMsg) {
       }
       list->numOfConn++;
     }
-    tTrace("%s numOfConn: %d, limit: %d", pTransInst->label, list->numOfConn, pTransInst->connLimitNum);
+    tDebug("%s numOfConn: %d, limit: %d, dst:%s", pTransInst->label, list->numOfConn, pTransInst->connLimitNum, key);
     return NULL;
   }
 
@@ -714,7 +762,7 @@ static void addConnToPool(void* pool, SCliConn* conn) {
   cliDestroyConnMsgs(conn, false);
 
   if (conn->list == NULL) {
-    conn->list = taosHashGet((SHashObj*)pool, conn->dstAddr, strlen(conn->dstAddr) + 1);
+    conn->list = taosHashGet((SHashObj*)pool, conn->dstAddr, strlen(conn->dstAddr));
   }
 
   SConnList* pList = conn->list;
@@ -740,13 +788,13 @@ static void addConnToPool(void* pool, SCliConn* conn) {
   QUEUE_PUSH(&conn->list->conns, &conn->q);
   conn->list->size += 1;
 
-  if (conn->list->size >= 20) {
+  if (conn->list->size >= 10) {
     STaskArg* arg = taosMemoryCalloc(1, sizeof(STaskArg));
     arg->param1 = conn;
     arg->param2 = thrd;
 
     STrans* pTransInst = thrd->pTransInst;
-    conn->task = transDQSched(thrd->timeoutQueue, doCloseIdleConn, arg, CONN_PERSIST_TIME(pTransInst->idleTime));
+    conn->task = transDQSched(thrd->timeoutQueue, doCloseIdleConn, arg, 10 * CONN_PERSIST_TIME(pTransInst->idleTime));
   }
 }
 static int32_t allocConnRef(SCliConn* conn, bool update) {
@@ -759,8 +807,10 @@ static int32_t allocConnRef(SCliConn* conn, bool update) {
   exh->handle = conn;
   exh->pThrd = conn->hostThrd;
   exh->refId = transAddExHandle(transGetRefMgt(), exh);
-  conn->refId = exh->refId;
+  QUEUE_INIT(&exh->q);
+  taosInitRWLatch(&exh->latch);
 
+  conn->refId = exh->refId;
   if (conn->refId == -1) {
     taosMemoryFree(exh);
   }
@@ -777,9 +827,11 @@ static int32_t specifyConnRef(SCliConn* conn, bool update, int64_t handle) {
   if (exh == NULL) {
     return -1;
   }
+  taosWLockLatch(&exh->latch);
   exh->handle = conn;
   exh->pThrd = conn->hostThrd;
   conn->refId = exh->refId;
+  taosWUnLockLatch(&exh->latch);
 
   transReleaseExHandle(transGetRefMgt(), handle);
   return 0;
@@ -880,7 +932,6 @@ static void cliDestroyConn(SCliConn* conn, bool clear) {
   }
 
   conn->list = NULL;
-  pThrd->newConnCount--;
 
   transReleaseExHandle(transGetRefMgt(), conn->refId);
   transRemoveExHandle(transGetRefMgt(), conn->refId);
@@ -1188,7 +1239,6 @@ static void cliHandleBatchReq(SCliBatch* pBatch, SCliThrd* pThrd) {
     addr.sin_port = (uint16_t)htons(pList->port);
 
     tTrace("%s conn %p try to connect to %s", pTransInst->label, conn, pList->dst);
-    pThrd->newConnCount++;
     int32_t fd = taosCreateSocketWithTimeout(TRANS_CONN_TIMEOUT * 10);
     if (fd == -1) {
       tError("%s conn %p failed to create socket, reason:%s", transLabel(pTransInst), conn,
@@ -1279,7 +1329,7 @@ static void cliHandleFastFail(SCliConn* pConn, int status) {
 
     if (pMsg != NULL && REQUEST_NO_RESP(&pMsg->msg) &&
         (pTransInst->failFastFp != NULL && pTransInst->failFastFp(pMsg->msg.msgType))) {
-      SFailFastItem* item = taosHashGet(pThrd->failFastCache, pConn->dstAddr, strlen(pConn->dstAddr) + 1);
+      SFailFastItem* item = taosHashGet(pThrd->failFastCache, pConn->dstAddr, strlen(pConn->dstAddr));
       int64_t        cTimestamp = taosGetTimestampMs();
       if (item != NULL) {
         int32_t elapse = cTimestamp - item->timestamp;
@@ -1291,7 +1341,7 @@ static void cliHandleFastFail(SCliConn* pConn, int status) {
         }
       } else {
         SFailFastItem item = {.count = 1, .timestamp = cTimestamp};
-        taosHashPut(pThrd->failFastCache, pConn->dstAddr, strlen(pConn->dstAddr) + 1, &item, sizeof(SFailFastItem));
+        taosHashPut(pThrd->failFastCache, pConn->dstAddr, strlen(pConn->dstAddr), &item, sizeof(SFailFastItem));
       }
     }
   } else {
@@ -1390,7 +1440,10 @@ static void cliHandleRelease(SCliMsg* pMsg, SCliThrd* pThrd) {
     return;
   }
 
+  taosRLockLatch(&exh->latch);
   SCliConn* conn = exh->handle;
+  taosRUnLockLatch(&exh->latch);
+
   transReleaseExHandle(transGetRefMgt(), refId);
   tDebug("%s conn %p start to release to inst", CONN_GET_INST_LABEL(conn), conn);
 
@@ -1423,7 +1476,9 @@ SCliConn* cliGetConn(SCliMsg** pMsg, SCliThrd* pThrd, bool* ignore, char* addr)
       *ignore = true;
       return NULL;
     } else {
+      taosRLockLatch(&exh->latch);
       conn = exh->handle;
+      taosRUnLockLatch(&exh->latch);
       if (conn == NULL) {
         conn = getConnFromPool2(pThrd, addr, pMsg);
         if (conn != NULL) specifyConnRef(conn, true, refId);
@@ -1437,7 +1492,7 @@ SCliConn* cliGetConn(SCliMsg** pMsg, SCliThrd* pThrd, bool* ignore, char* addr)
   if (conn != NULL) {
     tTrace("%s conn %p get from conn pool:%p", CONN_GET_INST_LABEL(conn), conn, pThrd->pool);
   } else {
-    tTrace("%s not found conn in conn pool:%p", ((STrans*)pThrd->pTransInst)->label, pThrd->pool);
+    tTrace("%s not found conn in conn pool:%p, dst:%s", ((STrans*)pThrd->pTransInst)->label, pThrd->pool, addr);
   }
   return conn;
 }
@@ -1471,7 +1526,8 @@ FORCE_INLINE int32_t cliBuildExceptResp(SCliMsg* pMsg, STransMsg* pResp) {
 }
 static FORCE_INLINE uint32_t cliGetIpFromFqdnCache(SHashObj* cache, char* fqdn) {
   uint32_t  addr = 0;
-  uint32_t* v = taosHashGet(cache, fqdn, strlen(fqdn) + 1);
+  size_t    len = strlen(fqdn);
+  uint32_t* v = taosHashGet(cache, fqdn, len);
   if (v == NULL) {
     addr = taosGetIpv4FromFqdn(fqdn);
     if (addr == 0xffffffff) {
@@ -1480,7 +1536,7 @@ static FORCE_INLINE uint32_t cliGetIpFromFqdnCache(SHashObj* cache, char* fqdn)
       return addr;
     }
 
-    taosHashPut(cache, fqdn, strlen(fqdn) + 1, &addr, sizeof(addr));
+    taosHashPut(cache, fqdn, len, &addr, sizeof(addr));
   } else {
     addr = *v;
   }
@@ -1490,13 +1546,14 @@ static FORCE_INLINE void cliUpdateFqdnCache(SHashObj* cache, char* fqdn) {
   // impl later
   uint32_t addr = taosGetIpv4FromFqdn(fqdn);
   if (addr != 0xffffffff) {
-    uint32_t* v = taosHashGet(cache, fqdn, strlen(fqdn) + 1);
+    size_t    len = strlen(fqdn);
+    uint32_t* v = taosHashGet(cache, fqdn, len);
     if (addr != *v) {
       char old[64] = {0}, new[64] = {0};
       tinet_ntoa(old, *v);
       tinet_ntoa(new, addr);
       tWarn("update ip of fqdn:%s, old: %s, new: %s", fqdn, old, new);
-      taosHashPut(cache, fqdn, strlen(fqdn) + 1, &addr, sizeof(addr));
+      taosHashPut(cache, fqdn, len, &addr, sizeof(addr));
     }
   }
   return;
@@ -1537,21 +1594,6 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) {
     return;
   }
 
-  if (rpcDebugFlag & DEBUG_TRACE) {
-    if (tmsgIsValid(pMsg->msg.msgType)) {
-      char buf[128] = {0};
-      sprintf(buf, "%s", TMSG_INFO(pMsg->msg.msgType));
-      int* count = taosHashGet(pThrd->msgCount, buf, sizeof(buf));
-      if (NULL == 0) {
-        int localCount = 1;
-        taosHashPut(pThrd->msgCount, buf, sizeof(buf), &localCount, sizeof(localCount));
-      } else {
-        int localCount = *count + 1;
-        taosHashPut(pThrd->msgCount, buf, sizeof(buf), &localCount, sizeof(localCount));
-      }
-    }
-  }
-
   char*    fqdn = EPSET_GET_INUSE_IP(&pMsg->ctx->epSet);
   uint16_t port = EPSET_GET_INUSE_PORT(&pMsg->ctx->epSet);
   char     addr[TSDB_FQDN_LEN + 64] = {0};
@@ -1609,7 +1651,6 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) {
     addr.sin_port = (uint16_t)htons(port);
 
     tGTrace("%s conn %p try to connect to %s", pTransInst->label, conn, conn->dstAddr);
-    pThrd->newConnCount++;
     int32_t fd = taosCreateSocketWithTimeout(TRANS_CONN_TIMEOUT * 10);
     if (fd == -1) {
       tGError("%s conn %p failed to create socket, reason:%s", transLabel(pTransInst), conn,
@@ -1704,9 +1745,8 @@ static void cliBatchDealReq(queue* wq, SCliThrd* pThrd) {
       uint32_t port = EPSET_GET_INUSE_PORT(&pCtx->epSet);
       char     key[TSDB_FQDN_LEN + 64] = {0};
       CONN_CONSTRUCT_HASH_KEY(key, ip, port);
-
-      // SCliBatch** ppBatch = taosHashGet(pThrd->batchCache, key, sizeof(key));
-      SCliBatchList** ppBatchList = taosHashGet(pThrd->batchCache, key, sizeof(key));
+      size_t          klen = strlen(key);
+      SCliBatchList** ppBatchList = taosHashGet(pThrd->batchCache, key, klen);
       if (ppBatchList == NULL || *ppBatchList == NULL) {
         SCliBatchList* pBatchList = taosMemoryCalloc(1, sizeof(SCliBatchList));
         QUEUE_INIT(&pBatchList->wq);
@@ -1730,7 +1770,7 @@ static void cliBatchDealReq(queue* wq, SCliThrd* pThrd) {
 
         QUEUE_PUSH(&pBatchList->wq, &pBatch->listq);
 
-        taosHashPut(pThrd->batchCache, key, sizeof(key), &pBatchList, sizeof(void*));
+        taosHashPut(pThrd->batchCache, key, klen, &pBatchList, sizeof(void*));
       } else {
         if (QUEUE_IS_EMPTY(&(*ppBatchList)->wq)) {
           SCliBatch* pBatch = taosMemoryCalloc(1, sizeof(SCliBatch));
@@ -1800,21 +1840,6 @@ static void cliAsyncCb(uv_async_t* handle) {
   QUEUE_MOVE(&item->qmsg, &wq);
   taosThreadMutexUnlock(&item->mtx);
 
-  if (rpcDebugFlag & DEBUG_TRACE) {
-    void* pIter = taosHashIterate(pThrd->msgCount, NULL);
-    while (pIter != NULL) {
-      int*   count = pIter;
-      size_t len = 0;
-      char*  key = taosHashGetKey(pIter, &len);
-      if (*count != 0) {
-        tDebug("key: %s count: %d", key, *count);
-      }
-
-      pIter = taosHashIterate(pThrd->msgCount, pIter);
-    }
-    tDebug("all conn count: %d", pThrd->newConnCount);
-  }
-
   int8_t supportBatch = pTransInst->supportBatch;
   if (supportBatch == 0) {
     cliNoBatchDealReq(&wq, pThrd);
@@ -1885,9 +1910,10 @@ void cliIteraConnMsgs(SCliConn* conn) {
 bool cliRecvReleaseReq(SCliConn* conn, STransMsgHead* pHead) {
   if (pHead->release == 1 && (pHead->msgLen) == sizeof(*pHead)) {
     uint64_t ahandle = pHead->ahandle;
-    tDebug("ahandle = %" PRIu64 "", ahandle);
     SCliMsg* pMsg = NULL;
     CONN_GET_MSGCTX_BY_AHANDLE(conn, ahandle);
+    tDebug("%s conn %p receive release request, refId:%" PRId64 ", may ignore", CONN_GET_INST_LABEL(conn), conn,
+           conn->refId);
 
     transClearBuffer(&conn->readBuf);
     transFreeMsg(transContFromHead((char*)pHead));
@@ -1896,6 +1922,9 @@ bool cliRecvReleaseReq(SCliConn* conn, STransMsgHead* pHead) {
       SCliMsg* cliMsg = transQueueGet(&conn->cliMsgs, i);
       if (cliMsg->type == Release) {
         ASSERTS(pMsg == NULL, "trans-cli recv invaid release-req");
+        tDebug("%s conn %p receive release request, refId:%" PRId64 ", ignore msg", CONN_GET_INST_LABEL(conn), conn,
+               conn->refId);
+        cliDestroyConn(conn, true);
         return true;
       }
     }
@@ -1971,8 +2000,9 @@ static FORCE_INLINE void destroyCmsgWrapper(void* arg, void* param) {
   if (pMsg == NULL) {
     return;
   }
-  if (param != NULL) {
+
   SCliThrd* pThrd = param;
+  if (pMsg->msg.info.notFreeAhandle == 0 && pThrd != NULL) {
     if (pThrd->destroyAhandleFp) (*pThrd->destroyAhandleFp)(pMsg->msg.info.ahandle);
   }
   destroyCmsg(pMsg);
@@ -1984,12 +2014,9 @@ static FORCE_INLINE void destroyCmsgAndAhandle(void* param) {
   SCliMsg*  pMsg = arg->param1;
   SCliThrd* pThrd = arg->param2;
 
-  tDebug("destroy Ahandle A");
-  if (pThrd != NULL && pThrd->destroyAhandleFp != NULL) {
-    tDebug("destroy Ahandle B");
+  if (pMsg->msg.info.notFreeAhandle == 0 && pThrd != NULL && pThrd->destroyAhandleFp != NULL) {
     pThrd->destroyAhandleFp(pMsg->ctx->ahandle);
   }
-  tDebug("destroy Ahandle C");
 
   transDestroyConnCtx(pMsg->ctx);
   transFreeMsg(pMsg->msg.pCont);
@@ -2013,11 +2040,9 @@ static SCliThrd* createThrdObj(void* trans) {
     taosMemoryFree(pThrd);
     return NULL;
   }
-  if (pTransInst->supportBatch) {
-    pThrd->asyncPool = transAsyncPoolCreate(pThrd->loop, 4, pThrd, cliAsyncCb);
-  } else {
-    pThrd->asyncPool = transAsyncPoolCreate(pThrd->loop, 8, pThrd, cliAsyncCb);
-  }
+  int32_t nSync = pTransInst->supportBatch ? 4 : 8;
+  pThrd->asyncPool = transAsyncPoolCreate(pThrd->loop, nSync, pThrd, cliAsyncCb);
+
   if (pThrd->asyncPool == NULL) {
     tError("failed to init async pool");
     uv_loop_close(pThrd->loop);
@@ -2058,8 +2083,6 @@ static SCliThrd* createThrdObj(void* trans) {
 
   pThrd->quit = false;
 
-  pThrd->newConnCount = 0;
-  pThrd->msgCount = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
   return pThrd;
 }
 static void destroyThrdObj(SCliThrd* pThrd) {
@@ -2105,7 +2128,6 @@ static void destroyThrdObj(SCliThrd* pThrd) {
     pIter = (void**)taosHashIterate(pThrd->batchCache, pIter);
   }
   taosHashCleanup(pThrd->batchCache);
-  taosHashCleanup(pThrd->msgCount);
   taosMemoryFree(pThrd);
 }
 
@@ -2124,14 +2146,7 @@ void cliSendQuit(SCliThrd* thrd) {
 void cliWalkCb(uv_handle_t* handle, void* arg) {
   if (!uv_is_closing(handle)) {
     if (uv_handle_get_type(handle) == UV_TIMER) {
-      // SCliConn* pConn = handle->data;
-      //  if (pConn != NULL && pConn->timer != NULL) {
-      //    SCliThrd* pThrd = pConn->hostThrd;
-      //    uv_timer_stop((uv_timer_t*)handle);
-      //    handle->data = NULL;
-      //    taosArrayPush(pThrd->timerList, &pConn->timer);
-      //    pConn->timer = NULL;
-      //  }
+      // do nothing
     } else {
       uv_read_stop((uv_stream_t*)handle);
     }
@@ -2166,18 +2181,23 @@ static void doCloseIdleConn(void* param) {
   cliDestroyConn(conn, true);
   taosMemoryFree(arg);
 }
+static void cliSchedMsgToDebug(SCliMsg* pMsg, char* label) {
+  if (!(rpcDebugFlag & DEBUG_DEBUG)) {
+    return;
+  }
+  STransConnCtx* pCtx = pMsg->ctx;
+  STraceId*      trace = &pMsg->msg.info.traceId;
+  char           tbuf[512] = {0};
+  EPSET_TO_STR(&pCtx->epSet, tbuf);
+  tGDebug("%s retry on next node,use:%s, step: %d,timeout:%" PRId64 "", label, tbuf, pCtx->retryStep,
+          pCtx->retryNextInterval);
+  return;
+}
 
 static void cliSchedMsgToNextNode(SCliMsg* pMsg, SCliThrd* pThrd) {
   STrans*        pTransInst = pThrd->pTransInst;
   STransConnCtx* pCtx = pMsg->ctx;
-
-  if (rpcDebugFlag & DEBUG_DEBUG) {
-    STraceId* trace = &pMsg->msg.info.traceId;
-    char      tbuf[512] = {0};
-    EPSET_TO_STR(&pCtx->epSet, tbuf);
-    tGDebug("%s retry on next node,use:%s, step: %d,timeout:%" PRId64 "", transLabel(pThrd->pTransInst), tbuf,
-            pCtx->retryStep, pCtx->retryNextInterval);
-  }
+  cliSchedMsgToDebug(pMsg, transLabel(pThrd->pTransInst));
 
   STaskArg* arg = taosMemoryMalloc(sizeof(STaskArg));
   arg->param1 = pMsg;
@@ -2186,12 +2206,6 @@ static void cliSchedMsgToNextNode(SCliMsg* pMsg, SCliThrd* pThrd) {
   transDQSched(pThrd->delayQueue, doDelayTask, arg, pCtx->retryNextInterval);
 }
 
-FORCE_INLINE void cliCompareAndSwap(int8_t* val, int8_t exp, int8_t newVal) {
-  if (*val != exp) {
-    *val = newVal;
-  }
-}
-
 FORCE_INLINE bool cliTryExtractEpSet(STransMsg* pResp, SEpSet* dst) {
   if ((pResp == NULL || pResp->info.hasEpSet == 0)) {
     return false;
@@ -2411,20 +2425,6 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
       tGTrace("%s conn %p extract epset from msg", CONN_GET_INST_LABEL(pConn), pConn);
     }
   }
-  if (rpcDebugFlag & DEBUG_TRACE) {
-    if (tmsgIsValid(pResp->msgType - 1)) {
-      char buf[128] = {0};
-      sprintf(buf, "%s", TMSG_INFO(pResp->msgType - 1));
-      int* count = taosHashGet(pThrd->msgCount, buf, sizeof(buf));
-      if (NULL == 0) {
-        int localCount = 0;
-        taosHashPut(pThrd->msgCount, buf, sizeof(buf), &localCount, sizeof(localCount));
-      } else {
-        int localCount = *count - 1;
-        taosHashPut(pThrd->msgCount, buf, sizeof(buf), &localCount, sizeof(localCount));
-      }
-    }
-  }
   if (pCtx->pSem || pCtx->syncMsgRef != 0) {
     tGTrace("%s conn %p(sync) handle resp", CONN_GET_INST_LABEL(pConn), pConn);
     if (pCtx->pSem) {
@@ -2547,21 +2547,7 @@ int transReleaseCliHandle(void* handle) {
   }
   return 0;
 }
-
-int transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransCtx* ctx) {
-  STrans* pTransInst = (STrans*)transAcquireExHandle(transGetInstMgt(), (int64_t)shandle);
-  if (pTransInst == NULL) {
-    transFreeMsg(pReq->pCont);
-    return TSDB_CODE_RPC_BROKEN_LINK;
-  }
-
-  SCliThrd* pThrd = transGetWorkThrd(pTransInst, (int64_t)pReq->info.handle);
-  if (pThrd == NULL) {
-    transFreeMsg(pReq->pCont);
-    transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
-    return TSDB_CODE_RPC_BROKEN_LINK;
-  }
-
+static SCliMsg* transInitMsg(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransCtx* ctx) {
   TRACE_SET_MSGID(&pReq->info.traceId, tGenIdPI64());
   STransConnCtx* pCtx = taosMemoryCalloc(1, sizeof(STransConnCtx));
   epsetAssign(&pCtx->epSet, pEpSet);
@@ -2578,12 +2564,48 @@ int transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STran
   cliMsg->st = taosGetTimestampUs();
   cliMsg->type = Normal;
   cliMsg->refId = (int64_t)shandle;
+  QUEUE_INIT(&cliMsg->seqq);
+  return cliMsg;
+}
+
+int transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransCtx* ctx) {
+  STrans* pTransInst = (STrans*)transAcquireExHandle(transGetInstMgt(), (int64_t)shandle);
+  if (pTransInst == NULL) {
+    transFreeMsg(pReq->pCont);
+    return TSDB_CODE_RPC_BROKEN_LINK;
+  }
+
+  int64_t   handle = (int64_t)pReq->info.handle;
+  SCliThrd* pThrd = transGetWorkThrd(pTransInst, handle);
+  if (pThrd == NULL) {
+    transFreeMsg(pReq->pCont);
+    transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
+    return TSDB_CODE_RPC_BROKEN_LINK;
+  }
+  if (handle != 0) {
+    SExHandle* exh = transAcquireExHandle(transGetRefMgt(), handle);
+    if (exh != NULL) {
+      taosWLockLatch(&exh->latch);
+      if (exh->handle == NULL && exh->inited != 0) {
+        SCliMsg* pCliMsg = transInitMsg(shandle, pEpSet, pReq, ctx);
+        QUEUE_PUSH(&exh->q, &pCliMsg->seqq);
+        taosWUnLockLatch(&exh->latch);
+        tDebug("msg refId: %" PRId64 "", handle);
+        transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
+        return 0;
+      }
+      exh->inited = 1;
+      taosWUnLockLatch(&exh->latch);
+      transReleaseExHandle(transGetRefMgt(), handle);
+    }
+  }
+  SCliMsg* pCliMsg = transInitMsg(shandle, pEpSet, pReq, ctx);
 
   STraceId* trace = &pReq->info.traceId;
   tGDebug("%s send request at thread:%08" PRId64 ", dst:%s:%d, app:%p", transLabel(pTransInst), pThrd->pid,
-          EPSET_GET_INUSE_IP(&pCtx->epSet), EPSET_GET_INUSE_PORT(&pCtx->epSet), pReq->info.ahandle);
-  if (0 != transAsyncSend(pThrd->asyncPool, &(cliMsg->q))) {
-    destroyCmsg(cliMsg);
+          EPSET_GET_INUSE_IP(pEpSet), EPSET_GET_INUSE_PORT(pEpSet), pReq->info.ahandle);
+  if (0 != transAsyncSend(pThrd->asyncPool, &(pCliMsg->q))) {
+    destroyCmsg(pCliMsg);
     transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
     return TSDB_CODE_RPC_BROKEN_LINK;
   }
@@ -2769,6 +2791,8 @@ int transSetDefaultAddr(void* shandle, const char* ip, const char* fqdn) {
 int64_t transAllocHandle() {
   SExHandle* exh = taosMemoryCalloc(1, sizeof(SExHandle));
   exh->refId = transAddExHandle(transGetRefMgt(), exh);
+  QUEUE_INIT(&exh->q);
+  taosInitRWLatch(&exh->latch);
   tDebug("pre alloc refId %" PRId64 "", exh->refId);
 
   return exh->refId;

source/libs/transport/src/transSvr.c

@@ -761,9 +761,12 @@ static bool uvRecvReleaseReq(SSvrConn* pConn, STransMsgHead* pHead) {
     tTrace("conn %p received release request", pConn);
 
     STraceId traceId = pHead->traceId;
-    pConn->status = ConnRelease;
     transClearBuffer(&pConn->readBuf);
     transFreeMsg(transContFromHead((char*)pHead));
+    if (pConn->status != ConnAcquire) {
+      return true;
+    }
+    pConn->status = ConnRelease;
 
     STransMsg tmsg = {.code = 0, .info.handle = (void*)pConn, .info.traceId = traceId, .info.ahandle = (void*)0x9527};
     SSvrMsg*  srvMsg = taosMemoryCalloc(1, sizeof(SSvrMsg));
@@ -1090,6 +1093,7 @@ static FORCE_INLINE SSvrConn* createConn(void* hThrd) {
 
   STrans* pTransInst = pThrd->pTransInst;
   pConn->refId = exh->refId;
+  QUEUE_INIT(&exh->q);
   transRefSrvHandle(pConn);
   tTrace("%s handle %p, conn %p created, refId:%" PRId64, transLabel(pTransInst), exh, pConn, pConn->refId);
   return pConn;
@@ -1121,6 +1125,7 @@ static int reallocConnRef(SSvrConn* conn) {
   exh->handle = conn;
   exh->pThrd = conn->hostThrd;
   exh->refId = transAddExHandle(transGetRefMgt(), exh);
+  QUEUE_INIT(&exh->q);
   transAcquireExHandle(transGetRefMgt(), exh->refId);
   conn->refId = exh->refId;
 

source/util/src/tworker.c

@@ -19,7 +19,7 @@
 #include "tgeosctx.h"
 #include "tlog.h"
 
-#define QUEUE_THRESHOLD 1000 * 1000
+#define QUEUE_THRESHOLD (1000 * 1000)
 
 typedef void *(*ThreadFp)(void *param);
 

source/util/test/tbaseCodecTest.cpp

@@ -17,11 +17,6 @@ using namespace std;
 #pragma GCC diagnostic ignored "-Wunused-variable"
 #pragma GCC diagnostic ignored "-Wsign-compare"
 
-int main(int argc, char **argv) {
-  testing::InitGoogleTest(&argc, argv);
-  return RUN_ALL_TESTS();
-}
-
 static void checkBase58Codec(uint8_t *pRaw, int32_t rawLen, int32_t index) {
   int64_t start = taosGetTimestampUs();
   char   *pEnc = base58_encode((const uint8_t *)pRaw, rawLen);

tests/script/tsim/parser/columnValue_bool.sim

@@ -936,6 +936,9 @@ sql_error alter table st_bool_i1 set tag tagname="123abc"
 sql        alter table st_bool_i2 set tag tagname="123"
 sql_error alter table st_bool_i3 set tag tagname=abc
 sql_error alter table st_bool_i4 set tag tagname="abc"
+sql_error alter table st_bool_i4 set tag tagname=now
+sql_error alter table st_bool_i4 set tag tagname=now()+1d
+sql_error alter table st_bool_i4 set tag tagname=1+1d
 sql_error alter table st_bool_i5 set tag tagname=" "
 sql_error alter table st_bool_i6 set tag tagname=''
 

tests/script/tsim/parser/columnValue_int.sim

@@ -913,6 +913,8 @@ sql_error alter table st_int_e19 set tag tagname=123abc
 sql_error alter table st_int_e20 set tag tagname="123abc"
 sql_error alter table st_int_e22 set tag tagname=abc
 sql_error alter table st_int_e23 set tag tagname="abc"
+sql_error alter table st_int_e25 set tag tagname=1+1d
+sql_error alter table st_int_e25 set tag tagname="1"+1d
 sql_error alter table st_int_e24 set tag tagname=" "
 sql_error alter table st_int_e25 set tag tagname=''
 sql alter table st_int_e26_1 set tag tagname='123'

tests/script/tsim/parser/columnValue_timestamp.sim

@@ -132,6 +132,77 @@ sql show tags from  st_timestamp_22
 if $data05 != -1 then
   return -1
 endi
+sql create table st_timestamp_23 using mt_timestamp tags (1+ 1d )
+sql show tags from  st_timestamp_23
+if $data05 != 86400001 then
+  return -1
+endi
+sql create table st_timestamp_24 using mt_timestamp tags (-0 + 1d)
+sql show tags from  st_timestamp_24
+if $data05 != 86400000 then
+  return -1
+endi
+sql create table st_timestamp_25 using mt_timestamp tags ("-0" -1s)
+sql show tags from  st_timestamp_25
+if $data05 != -1000 then
+  return -1
+endi
+sql create table st_timestamp_26 using mt_timestamp tags (0b01 -1a)
+sql show tags from  st_timestamp_26
+if $data05 != 0 then
+  return -1
+endi
+sql create table st_timestamp_27 using mt_timestamp tags (0b01 -1s)
+sql show tags from  st_timestamp_27
+if $data05 != -999 then
+  return -1
+endi
+sql create table st_timestamp_28 using mt_timestamp tags ("0x01" +1u)
+sql show tags from  st_timestamp_28
+if $data05 != 1 then
+  return -1
+endi
+sql create table st_timestamp_29 using mt_timestamp tags (0x01 +1b)
+sql show tags from  st_timestamp_29
+if $data05 != 1 then
+  return -1
+endi
+sql create table st_timestamp_30 using mt_timestamp tags (-0b00 -0a)
+sql show tags from  st_timestamp_30
+if $data05 != 0 then
+  return -1
+endi
+sql create table st_timestamp_31 using mt_timestamp tags ("-0x00" +1u)
+sql show tags from  st_timestamp_31
+if $data05 != 0 then
+  return -1
+endi
+sql create table st_timestamp_32 using mt_timestamp tags (-0x00 +1b)
+sql show tags from  st_timestamp_32
+if $data05 != 0 then
+  return -1
+endi
+sql create table st_timestamp_33 using mt_timestamp tags (now +1b)
+sql show tags from  st_timestamp_33
+if $data05 < 1711883186000 then
+  return -1
+endi
+sql create table st_timestamp_34 using mt_timestamp tags ("now()" +1b)
+sql show tags from  st_timestamp_34
+if $data05 < 1711883186000 then
+  return -1
+endi
+sql create table st_timestamp_35 using mt_timestamp tags (today() +1d)
+sql show tags from  st_timestamp_35
+if $data05 < 1711883186000 then
+  return -1
+endi
+sql create table st_timestamp_36 using mt_timestamp tags ("today()" +1d)
+sql show tags from  st_timestamp_36
+if $data05 < 1711883186000 then
+  return -1
+endi
+
 
 ## case 01: insert values  for test column values
 sql insert into st_timestamp_0 values(now,NULL)
@@ -249,6 +320,76 @@ sql select ts, cast(c as bigint) from  st_timestamp_22
 if $data01 != -1 then
   return -1
 endi
+sql insert into st_timestamp_23 values(now,1+ 1d )
+sql select ts, cast(c as bigint) from st_timestamp_23
+if $data01 != 86400001 then 
+  return -1
+endi
+sql insert into st_timestamp_24 values(now,-0 + 1d)
+sql select ts, cast(c as bigint) from st_timestamp_24
+if $data01 != 86400000 then 
+  return -1
+endi
+sql insert into st_timestamp_25 values(now,"-0" -1s)
+sql select ts, cast(c as bigint) from st_timestamp_25
+if $data01 != -1000 then 
+  return -1
+endi
+sql insert into st_timestamp_26 values(now,0b01 -1a)
+sql select ts, cast(c as bigint) from st_timestamp_26
+if $data01 != 0 then 
+  return -1
+endi
+sql insert into st_timestamp_27 values(now,+0b01 -1s)
+sql select ts, cast(c as bigint) from st_timestamp_27
+if $data01 != -999 then 
+  return -1
+endi
+sql insert into st_timestamp_28 values(now,"+0x01" +1u)
+sql select ts, cast(c as bigint) from st_timestamp_28
+if $data01 != 1 then 
+  return -1
+endi
+sql insert into st_timestamp_29 values(now,0x01 +1b)
+sql select ts, cast(c as bigint) from st_timestamp_29
+if $data01 != 1 then 
+  return -1
+endi
+sql insert into  st_timestamp_30 values(now,-0b00 -0a)
+sql show tags from  st_timestamp_30
+if $data05 != 0 then
+  return -1
+endi
+sql insert into  st_timestamp_31 values(now,"-0x00" +1u)
+sql show tags from  st_timestamp_31
+if $data05 != 0 then
+  return -1
+endi
+sql insert into st_timestamp_32 values (now,-0x00 +1b)
+sql show tags from  st_timestamp_32
+if $data05 != 0 then
+  return -1
+endi
+sql insert into st_timestamp_33 values(now,now +1b)
+sql select ts, cast(c as bigint) from st_timestamp_33
+if $data01 < 1711883186000 then 
+  return -1
+endi
+sql insert into st_timestamp_34 values(now,"now()" +1b)
+sql select ts, cast(c as bigint) from st_timestamp_34
+if $data01 < 1711883186000 then 
+  return -1
+endi
+sql insert into st_timestamp_35 values(now,today() +1d)
+sql select ts, cast(c as bigint) from st_timestamp_35
+if $data01 < 1711883186000 then 
+  return -1
+endi
+sql insert into st_timestamp_36 values(now,"today()" +1d)
+sql select ts, cast(c as bigint) from st_timestamp_36
+if $data01 < 1711883186000 then 
+  return -1
+endi
 
 ## case 02: dynamic create table for test tag values
 sql insert into st_timestamp_100 using mt_timestamp tags(NULL) values(now, NULL)
@@ -450,6 +591,136 @@ sql select ts, cast(c as bigint) from st_timestamp_1022
 if $data01 != -1 then
   return -1
 endi
+sql insert into st_timestamp_1023 using mt_timestamp tags(+1+1d) values(now,+1+ 1d )
+sql show tags from  st_timestamp_1023
+if $data05 != 86400001 then
+  return -1
+endi
+sql select ts, cast(c as bigint) from st_timestamp_1023
+if $data01 != 86400001 then 
+  return -1
+endi
+sql insert into st_timestamp_1024 using mt_timestamp tags(-0+1d) values(now,-0 + 1d)
+sql show tags from  st_timestamp_1024
+if $data05 != 86400000 then
+  return -1
+endi
+sql select ts, cast(c as bigint) from st_timestamp_1024
+if $data01 != 86400000 then 
+  return -1
+endi
+sql insert into st_timestamp_1025 using mt_timestamp tags("-0" -1s) values(now,"-0" -1s)
+sql show tags from  st_timestamp_1025
+if $data05 != -1000 then
+  return -1
+endi
+sql select ts, cast(c as bigint) from st_timestamp_1025
+if $data01 != -1000 then 
+  return -1
+endi
+sql insert into st_timestamp_1026 using mt_timestamp tags(+0b01-1a) values(now,+0b01 -1a)
+sql show tags from  st_timestamp_1026
+if $data05 != 0 then
+  return -1
+endi
+sql select ts, cast(c as bigint) from st_timestamp_1026
+if $data01 != 0 then 
+  return -1
+endi
+sql insert into st_timestamp_1027 using mt_timestamp tags(0b01-1s) values(now,0b01 -1s)
+sql show tags from  st_timestamp_1027
+if $data05 != -999 then
+  return -1
+endi
+sql select ts, cast(c as bigint) from st_timestamp_1027
+if $data01 != -999 then 
+  return -1
+endi
+sql insert into st_timestamp_1028 using mt_timestamp tags("0x01" + 1u) values(now,"0x01" +1u)
+sql show tags from  st_timestamp_1028
+if $data05 != 1 then
+  return -1
+endi
+sql select ts, cast(c as bigint) from st_timestamp_1028
+if $data01 != 1 then 
+  return -1
+endi
+sql insert into st_timestamp_1029 using mt_timestamp tags(+0x01 +1b) values(now,+0x01 +1b)
+sql show tags from  st_timestamp_1029
+if $data05 != 1 then
+  return -1
+endi
+sql select ts, cast(c as bigint) from st_timestamp_1029
+if $data01 != 1 then 
+  return -1
+endi
+sql insert into st_timestamp_1030 using mt_timestamp tags (-0b00 -0a) values(now,-0b00 -0a)
+sql show tags from  st_timestamp_1030
+if $data05 != 0 then
+  return -1
+endi
+sql select ts, cast(c as bigint) from st_timestamp_1030
+if $data01 != 0 then 
+  return -1
+endi
+sql insert into st_timestamp_1031 using mt_timestamp tags ("-0x00" +1u) values(now,"-0x00" +1u)
+sql show tags from  st_timestamp_1031
+if $data05 != 0 then
+  return -1
+endi
+sql select ts, cast(c as bigint) from st_timestamp_1031
+if $data01 != 0 then 
+  return -1
+endi
+sql insert into st_timestamp_1032 using mt_timestamp tags (-0x00 +1b) values(now,-0x00 +1b)
+sql show tags from  st_timestamp_1032
+if $data05 != 0 then
+  return -1
+endi
+sql select ts, cast(c as bigint) from st_timestamp_1032
+if $data01 != 0 then 
+  return -1
+endi
+sql insert into st_timestamp_1033 using mt_timestamp tags(now+1b) values(now,now +1b)
+sql show tags from  st_timestamp_1033
+if $data05 < 1711883186000 then
+  return -1
+endi
+sql select ts, cast(c as bigint) from st_timestamp_1033
+if $data01 < 1711883186000 then 
+  return -1
+endi
+sql insert into st_timestamp_1034 using mt_timestamp tags("now" +1b) values(now,"now()" +1b)
+sql show tags from  st_timestamp_1034
+if $data05 < 1711883186000 then
+  return -1
+endi
+sql select ts, cast(c as bigint) from st_timestamp_1034
+if $data01 < 1711883186000 then 
+  return -1
+endi
+sql insert into st_timestamp_1035 using mt_timestamp tags(today() + 1d) values(now,today() +1d)
+sql show tags from  st_timestamp_1035
+if $data05 < 1711883186000 then
+  return -1
+endi
+sql select ts, cast(c as bigint) from st_timestamp_1035
+if $data01 < 1711883186000 then 
+  return -1
+endi
+sql insert into st_timestamp_1036 using mt_timestamp tags("today" +1d) values(now,"today()" +1d)
+sql show tags from  st_timestamp_1036
+if $data05 < 1711883186000 then
+  return -1
+endi
+sql select ts, cast(c as bigint) from st_timestamp_1036
+if $data01 < 1711883186000 then 
+  return -1
+endi
+
+
+
+
 
 ### case 03: alter tag values
 sql alter table st_timestamp_0 set tag tagname=NULL
@@ -567,12 +838,85 @@ sql show tags from  st_timestamp_22
 if $data05 != -1 then
   return -1
 endi
+sql alter table st_timestamp_23 set tag tagname=1+ 1d 
+sql show tags from  st_timestamp_23
+if $data05 != 86400001 then
+  return -1
+endi
+sql alter table st_timestamp_24 set tag tagname=-0 + 1d
+sql show tags from  st_timestamp_24
+if $data05 != 86400000 then
+  return -1
+endi
+sql alter table st_timestamp_25 set tag tagname="-0" -1s
+sql show tags from  st_timestamp_25
+if $data05 != -1000 then
+  return -1
+endi
+sql alter table st_timestamp_26 set tag tagname=+0b01 -1a
+sql show tags from  st_timestamp_26
+if $data05 != 0 then
+  return -1
+endi
+sql alter table st_timestamp_27 set tag tagname=0b01 -1s
+sql show tags from  st_timestamp_27
+if $data05 != -999 then
+  return -1
+endi
+sql alter table st_timestamp_28 set tag tagname="0x01" +1u
+sql show tags from  st_timestamp_28
+if $data05 != 1 then
+  return -1
+endi
+sql alter table st_timestamp_29 set tag tagname=0x01 +1b
+sql show tags from  st_timestamp_29
+if $data05 != 1 then
+  return -1
+endi
+sql alter table st_timestamp_30 set tag tagname==-0b00 -0a
+sql show tags from  st_timestamp_30
+if $data05 != 0 then
+  return -1
+endi
+sql alter table st_timestamp_31 set tag tagname="-0x00" +1u
+sql show tags from  st_timestamp_31
+if $data05 != 0 then
+  return -1
+endi
+sql alter table st_timestamp_32 set tag tagname=-0x00 +1b
+sql show tags from  st_timestamp_32
+if $data05 != 0 then
+  return -1
+endi
+sql alter table st_timestamp_33 set tag tagname=now +1b
+sql show tags from  st_timestamp_33
+if $data05 < 1711883186000 then
+  return -1
+endi
+sql alter table st_timestamp_34 set tag tagname="now()" +1b
+sql show tags from  st_timestamp_34
+if $data05 < 1711883186000 then
+  return -1
+endi
+sql alter table st_timestamp_35 set tag tagname=today( ) +1d
+sql show tags from  st_timestamp_35
+if $data05 < 1711883186000 then
+  return -1
+endi
+sql alter table st_timestamp_36 set tag tagname="today()" +1d
+sql show tags from  st_timestamp_36
+if $data05 < 1711883186000 then
+  return -1
+endi
 
 ## case 04: illegal input
 sql_error create table st_timestamp_e0 using mt_timestamp tags (123abc)
 sql_error create table st_timestamp_e0 using mt_timestamp tags ("123abc")
 sql_error create table st_timestamp_e0 using mt_timestamp tags (abc)
 sql_error create table st_timestamp_e0 using mt_timestamp tags ("abc")
+sql_error create table st_timestamp_e0 using mt_timestamp tags (now()+1d+1s)
+sql_error create table st_timestamp_e0 using mt_timestamp tags (1+1y)
+sql_error create table st_timestamp_e0 using mt_timestamp tags (0x01+1b+1a)
 sql_error create table st_timestamp_e0 using mt_timestamp tags (" ")
 sql_error create table st_timestamp_e0 using mt_timestamp tags ('')
 sql_error create table st_timestamp_104 using mt_timestamp tags ("-123.1")
@@ -590,5 +934,7 @@ sql_error create table st_timestamp_115 using mt_timestamp tags (922337203685477
 sql create table st_timestamp_116 using mt_timestamp tags (-9223372036854775808)
 sql_error create table st_timestamp_117 using mt_timestamp tags (-9223372036854775809)
 sql_error insert into st_timestamp_118 using mt_timestamp tags(9223372036854775807) values(9223372036854775807, 9223372036854775807)
+sql_error insert into st_timestamp_119 using mt_timestamp tags(1+1s-1s) values(now, now)
+sql_error insert into st_timestamp_120 using mt_timestamp tags(1-1s) values(now, now-1s+1d)
 
 system sh/exec.sh -n dnode1 -s stop  -x SIGINT

tests/script/tsim/parser/columnValue_varbinary.sim

@@ -299,6 +299,8 @@ sql_error create table st_varbinary_1012  using mt_varbinary tags(tRue)
 sql_error create table st_varbinary_1013  using mt_varbinary tags(FalsE)
 sql_error create table st_varbinary_1014  using mt_varbinary tags(noW)
 sql_error create table st_varbinary_1015  using mt_varbinary tags(toDay)
+sql_error create table st_varbinary_1016  using mt_varbinary tags(now()+1s)
+sql_error create table st_varbinary_1017  using mt_varbinary tags(1+1s)
 sql_error insert into st_varbinary_106  using mt_varbinary tags(+0123) values(now, NULL);
 sql_error insert into st_varbinary_107  using mt_varbinary tags(-01.23) values(now, NULL);
 sql_error insert into st_varbinary_108  using mt_varbinary tags(+0x01) values(now, NULL);
@@ -309,6 +311,8 @@ sql_error insert into st_varbinary_1012  using mt_varbinary tags(tRue) values(no
 sql_error insert into st_varbinary_1013  using mt_varbinary tags(FalsE) values(now, NULL);
 sql_error insert into st_varbinary_1014  using mt_varbinary tags(noW) values(now, NULL);
 sql_error insert into st_varbinary_1015  using mt_varbinary tags(toDay) values(now, NULL);
+sql_error insert into st_varbinary_1016  using mt_varbinary tags(now()+1s) values(now, NULL);
+sql_error insert into st_varbinary_1017  using mt_varbinary tags(1+1s) values(now, NULL);
 sql_error insert into st_varbinary_106  using mt_varbinary tags(NULL) values(now(), +0123)
 sql_error insert into st_varbinary_107  using mt_varbinary tags(NULL) values(now(), -01.23)
 sql_error insert into st_varbinary_108  using mt_varbinary tags(NULL) values(now(), +0x01)
@@ -319,5 +323,7 @@ sql_error insert into st_varbinary_1012  using mt_varbinary tags(NULL) values(no
 sql_error insert into st_varbinary_1013  using mt_varbinary tags(NULL) values(now(), FalsE)
 sql_error insert into st_varbinary_1014  using mt_varbinary tags(NULL) values(now(), noW)
 sql_error insert into st_varbinary_1015  using mt_varbinary tags(NULL) values(now(), toDay)
+sql_error insert into st_varbinary_1016  using mt_varbinary tags(NULL) values(now(), now()+1s)
+sql_error insert into st_varbinary_1017  using mt_varbinary tags(NULL) values(now(), 1+1s)
 
 system sh/exec.sh -n dnode1 -s stop  -x SIGINT

tests/script/tsim/parser/columnValue_varchar.sim

@@ -410,6 +410,17 @@ endi
 
 
 # case 04: illegal input
+sql_error create table st_varchar_100  using mt_varchar tags(now+1d)
+sql_error create table st_varchar_101  using mt_varchar tags(toDay+1d)
+sql_error create table st_varchar_102  using mt_varchar tags(1+1b)
+sql_error create table st_varchar_103  using mt_varchar tags(0x01+1d)
+sql_error create table st_varchar_104  using mt_varchar tags(0b01+1s)
+sql_error insert into st_varchar_1100  using mt_varchar tags('now') values(now(),now+1d)
+sql_error insert into st_varchar_1101  using mt_varchar tags('now') values(now(),toDay+1d)
+sql_error insert into st_varchar_1102  using mt_varchar tags('now') values(now(),1+1b)
+sql_error insert into st_varchar_1103  using mt_varchar tags('now') values(now(),0x01+1d)
+sql_error insert into st_varchar_1104  using mt_varchar tags('now') values(now(),0b01+1s)
+sql_error alter table st_varchar_15  set tag tagname=now()+1d
 
 
 system sh/exec.sh -n dnode1 -s stop  -x SIGINT

tests/system-test/7-tmq/tmq_ts4563.py

@@ -29,9 +29,11 @@ class TDTestCase:
         tdSql.execute(f'use db_stmt')
 
         tdSql.query("select ts,k from st")
-        tdSql.checkRows(2)
+        tdSql.checkRows(self.expected_affected_rows)
 
         tdSql.execute(f'create topic t_unorder_data as select ts,k from st')
+        tdSql.execute(f'create topic t_unorder_data_none as select i,k from st')
+
         consumer_dict = {
             "group.id": "g1",
             "td.connect.user": "root",
@@ -41,7 +43,7 @@ class TDTestCase:
         consumer = Consumer(consumer_dict)
 
         try:
-            consumer.subscribe(["t_unorder_data"])
+            consumer.subscribe(["t_unorder_data", "t_unorder_data_none"])
         except TmqError:
             tdLog.exit(f"subscribe error")
 
@@ -51,18 +53,15 @@ class TDTestCase:
                 res = consumer.poll(1)
                 print(res)
                 if not res:
-                    if cnt == 0:
+                    if cnt == 0 or  cnt != 2*self.expected_affected_rows:
                         tdLog.exit("consume error")
                     break
                 val = res.value()
                 if val is None:
                     continue
                 for block in val:
+                    print(block.fetchall(),len(block.fetchall()))
                     cnt += len(block.fetchall())
-
-                if cnt != 2:
-                    tdLog.exit("consume error")
-
         finally:
             consumer.close()
 
@@ -114,12 +113,24 @@ class TDTestCase:
             # print(type(stmt))
             tdLog.debug("bind_param_batch start")
             stmt.bind_param_batch(params)
+
             tdLog.debug("bind_param_batch end")
             stmt.execute()
             tdLog.debug("execute end")
+            conn.execute("flush  database %s" % dbname)
+
+            params1 = new_multi_binds(2)
+            params1[0].timestamp((1626861392587,1626861392586))
+            params1[1].int([None,3])     
+            stmt.bind_param_batch(params1)
+            stmt.execute()
+
             end = datetime.now()
             print("elapsed time: ", end - start)
-            assert stmt.affected_rows == 2
+            print(stmt.affected_rows)
+            self.expected_affected_rows = 4
+            if stmt.affected_rows != self.expected_affected_rows :
+                tdLog.exit("affected_rows error")
             tdLog.debug("close start")
 
             stmt.close()

tests/system-test/8-stream/stream_basic.py

@@ -78,14 +78,55 @@ class TDTestCase:
         tdLog.info(cmd)
         os.system(cmd)
 
+    def case1(self):
+
+        tdSql.execute(f'create database if not exists d1 vgroups 1')
+        tdSql.execute(f'use d1')
+        tdSql.execute(f'create table st(ts timestamp, i int) tags(t int)')
+        tdSql.execute(f'insert into t1 using st tags(1) values(now, 1) (now+1s, 2)')
+        tdSql.execute(f'insert into t2 using st tags(2) values(now, 1) (now+1s, 2)')
+        tdSql.execute(f'insert into t3 using st tags(3) values(now, 1) (now+1s, 2)')
+
+        tdSql.execute("create stream stream1 fill_history 1 into sta subtable(concat('new-', tname)) AS SELECT "
+                      "_wstart, count(*), avg(i) FROM st PARTITION BY tbname tname INTERVAL(1m)", show=True)
+
+        tdSql.execute("create stream stream2 fill_history 1 into stb subtable(concat('new-', tname)) AS SELECT "
+                      "_wstart, count(*), avg(i) FROM st PARTITION BY tbname tname INTERVAL(1m)", show=True)
+
+        time.sleep(2)
+        tdSql.query("select * from sta")
+        tdSql.checkRows(3)
+        tdSql.query("select tbname from sta order by tbname")
+        if not tdSql.getData(0, 0).startswith('new-t1_1.d1.sta_'):
+            tdLog.exit("error1")
+
+        if not tdSql.getData(1, 0).startswith('new-t2_1.d1.sta_'):
+            tdLog.exit("error2")
+
+        if not tdSql.getData(2, 0).startswith('new-t3_1.d1.sta_'):
+            tdLog.exit("error3")
+
+        tdSql.query("select * from stb")
+        tdSql.checkRows(3)
+        tdSql.query("select tbname from stb order by tbname")
+        if not tdSql.getData(0, 0).startswith('new-t1_1.d1.stb_'):
+            tdLog.exit("error4")
+
+        if not tdSql.getData(1, 0).startswith('new-t2_1.d1.stb_'):
+            tdLog.exit("error5")
+
+        if not tdSql.getData(2, 0).startswith('new-t3_1.d1.stb_'):
+            tdLog.exit("error6")
+
     # run
     def run(self):
+        self.case1()
         # gen data
         random.seed(int(time.time()))
         self.taosBenchmark(" -d db -t 2 -v 2 -n 1000000 -y")
         # create stream
         tdSql.execute("use db")
-        tdSql.execute("create stream stream1 fill_history 1 into sta as select count(*) as cnt from meters interval(10a);",show=True)
+        tdSql.execute("create stream stream3 fill_history 1 into sta as select count(*) as cnt from meters interval(10a);",show=True)
         sql = "select count(*) from sta"
         # loop wait max 60s to check count is ok
         tdLog.info("loop wait result ...")

tools/CMakeLists.txt

@@ -18,7 +18,7 @@ IF (TD_WEBSOCKET)
             COMMAND     git clean -f -d
             BUILD_COMMAND
             COMMAND cargo update
-            COMMAND RUSTFLAGS=-Ctarget-feature=-crt-static cargo build --release -p taos-ws-sys --features native-tls
+            COMMAND RUSTFLAGS=-Ctarget-feature=-crt-static cargo build --release -p taos-ws-sys --features rustls
             INSTALL_COMMAND
             COMMAND cp target/release/${websocket_lib_file} ${CMAKE_BINARY_DIR}/build/lib
             COMMAND cmake -E make_directory ${CMAKE_BINARY_DIR}/build/include
@@ -37,7 +37,7 @@ IF (TD_WEBSOCKET)
             COMMAND     git clean -f -d
             BUILD_COMMAND
             COMMAND cargo update
-            COMMAND cargo build --release -p taos-ws-sys --features native-tls-vendored
+            COMMAND cargo build --release -p taos-ws-sys --features rustls
             INSTALL_COMMAND
             COMMAND cp target/release/taosws.dll ${CMAKE_BINARY_DIR}/build/lib
             COMMAND cp target/release/taosws.dll.lib ${CMAKE_BINARY_DIR}/build/lib/taosws.lib
@@ -57,7 +57,7 @@ IF (TD_WEBSOCKET)
             COMMAND     git clean -f -d
             BUILD_COMMAND
             COMMAND cargo update
-            COMMAND cargo build --release -p taos-ws-sys --features native-tls-vendored
+            COMMAND cargo build --release -p taos-ws-sys --features rustls
             INSTALL_COMMAND
             COMMAND cp target/release/${websocket_lib_file} ${CMAKE_BINARY_DIR}/build/lib
             COMMAND cmake -E make_directory ${CMAKE_BINARY_DIR}/build/include