Merge branch '3.0' of https://github.com/taosdata/TDengine into feat/TS-4243-3.0

docs/zh/12-taos-sql/12-distinguished.md

@@ -49,6 +49,7 @@ window_clause: {
   | STATE_WINDOW(col)
   | INTERVAL(interval_val [, interval_offset]) [SLIDING (sliding_val)] [FILL(fill_mod_and_val)]
   | EVENT_WINDOW START WITH start_trigger_condition END WITH end_trigger_condition
+  | COUNT_WINDOW(count_val[, sliding_val])
 }
 ```
 
@@ -180,6 +181,19 @@ select _wstart, _wend, count(*) from t event_window start with c1 > 0 end with c
 
 ![TDengine Database 事件窗口示意图](./event_window.webp)
 
+### 计数窗口
+
+计数窗口按固定的数据行数来划分窗口。默认将数据按时间戳排序，再按照count_val的值，将数据划分为多个窗口，然后做聚合计算。count_val表示每个count window包含的最大数据行数，总数据行数不能整除count_val时，最后一个窗口的行数会小于count_val。sliding_val是常量，表示窗口滑动的数量，类似于 interval的SLIDING。
+
+以下面的 SQL 语句为例，计数窗口切分如图所示：
+```sql
+select _wstart, _wend, count(*) from t count_window(4);
+```
+
+![count_window](https://github.com/taosdata/TDengine/assets/38781207/ff86278e-cbff-477f-a866-6e5cc150b712)
+
+
+
 ### 时间戳伪列
 
 窗口聚合查询结果中，如果 SQL 语句中没有指定输出查询结果中的时间戳列，那么最终结果中不会自动包含窗口的时间列信息。如果需要在结果中输出聚合结果所对应的时间窗口信息，需要在 SELECT 子句中使用时间戳相关的伪列: 时间窗口起始时间 (\_WSTART), 时间窗口结束时间 (\_WEND), 时间窗口持续时间 (\_WDURATION), 以及查询整体窗口相关的伪列: 查询窗口起始时间(\_QSTART) 和查询窗口结束时间(\_QEND)。需要注意的是时间窗口起始时间和结束时间均是闭区间，时间窗口持续时间是数据当前时间分辨率下的数值。例如，如果当前数据库的时间分辨率是毫秒，那么结果中 500 就表示当前时间窗口的持续时间是 500毫秒 (500 ms)。

include/libs/executor/storageapi.h

@@ -223,10 +223,10 @@ typedef struct SStoreTqReader {
   bool (*tqReaderCurrentBlockConsumed)();
 
   struct SWalReader* (*tqReaderGetWalReader)();  // todo remove it
-  int32_t (*tqReaderRetrieveTaosXBlock)();       // todo remove it
+//  int32_t (*tqReaderRetrieveTaosXBlock)();       // todo remove it
 
   int32_t (*tqReaderSetSubmitMsg)();  // todo remove it
-  bool (*tqReaderNextBlockFilterOut)();
+//  bool (*tqReaderNextBlockFilterOut)();
 } SStoreTqReader;
 
 typedef struct SStoreSnapshotFn {

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

@@ -106,7 +106,7 @@ typedef struct SQueryNode         SQueryNode;
 #define VND_INFO_FNAME     "vnode.json"
 #define VND_INFO_FNAME_TMP "vnode_tmp.json"
 
-#define VNODE_METRIC_SQL_COUNT "taos_sql_req:count"
+#define VNODE_METRIC_SQL_COUNT "taosd_sql_req:count"
 
 #define VNODE_METRIC_TAG_NAME_SQL_TYPE "sql_type"
 #define VNODE_METRIC_TAG_NAME_CLUSTER_ID "cluster_id"

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

@@ -368,24 +368,11 @@ int32_t extractMsgFromWal(SWalReader* pReader, void** pItem, int64_t maxVer, con
   }
 }
 
-// todo ignore the error in wal?
 bool tqNextBlockInWal(STqReader* pReader, const char* id, int sourceExcluded) {
   SWalReader*  pWalReader = pReader->pWalReader;
-  SSDataBlock* pDataBlock = NULL;
 
   uint64_t st = taosGetTimestampMs();
   while (1) {
-    // try next message in wal file
-    if (walNextValidMsg(pWalReader) < 0) {
-      return false;
-    }
-
-    void*   pBody = POINTER_SHIFT(pWalReader->pHead->head.body, sizeof(SSubmitReq2Msg));
-    int32_t bodyLen = pWalReader->pHead->head.bodyLen - sizeof(SSubmitReq2Msg);
-    int64_t ver = pWalReader->pHead->head.version;
-
-    tqReaderSetSubmitMsg(pReader, pBody, bodyLen, ver);
-    pReader->nextBlk = 0;
     int32_t numOfBlocks = taosArrayGetSize(pReader->submit.aSubmitTbData);
     while (pReader->nextBlk < numOfBlocks) {
       tqTrace("tq reader next data block %d/%d, len:%d %" PRId64, pReader->nextBlk, numOfBlocks, pReader->msg.msgLen,
@@ -400,33 +387,32 @@ bool tqNextBlockInWal(STqReader* pReader, const char* id, int sourceExcluded) {
         tqTrace("tq reader return submit block, uid:%" PRId64, pSubmitTbData->uid);
         SSDataBlock* pRes = NULL;
         int32_t      code = tqRetrieveDataBlock(pReader, &pRes, NULL);
-        if (code == TSDB_CODE_SUCCESS && pRes->info.rows > 0) {
-          if (pDataBlock == NULL) {
-            pDataBlock = createOneDataBlock(pRes, true);
-          } else {
-            blockDataMerge(pDataBlock, pRes);
-          }
+        if (code == TSDB_CODE_SUCCESS) {
+          return true;
         }
       } else {
         pReader->nextBlk += 1;
         tqTrace("tq reader discard submit block, uid:%" PRId64 ", continue", pSubmitTbData->uid);
       }
     }
+
     tDestroySubmitReq(&pReader->submit, TSDB_MSG_FLG_DECODE);
     pReader->msg.msgStr = NULL;
 
-    if (pDataBlock != NULL) {
-      blockDataCleanup(pReader->pResBlock);
-      copyDataBlock(pReader->pResBlock, pDataBlock);
-      blockDataDestroy(pDataBlock);
-      return true;
-    } else {
-      qTrace("stream scan return empty, all %d submit blocks consumed, %s", numOfBlocks, id);
-    }
-
     if (taosGetTimestampMs() - st > 1000) {
       return false;
     }
+
+    // try next message in wal file
+    if (walNextValidMsg(pWalReader) < 0) {
+      return false;
+    }
+
+    void*   pBody = POINTER_SHIFT(pWalReader->pHead->head.body, sizeof(SSubmitReq2Msg));
+    int32_t bodyLen = pWalReader->pHead->head.bodyLen - sizeof(SSubmitReq2Msg);
+    int64_t ver = pWalReader->pHead->head.version;
+    tqReaderSetSubmitMsg(pReader, pBody, bodyLen, ver);
+    pReader->nextBlk = 0;
   }
 }
 

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

@@ -128,12 +128,12 @@ void initTqAPI(SStoreTqReader* pTq) {
   pTq->tqReaderCurrentBlockConsumed = tqCurrentBlockConsumed;
 
   pTq->tqReaderGetWalReader = tqGetWalReader;              // todo remove it
-  pTq->tqReaderRetrieveTaosXBlock = tqRetrieveTaosxBlock;  // todo remove it
+//  pTq->tqReaderRetrieveTaosXBlock = tqRetrieveTaosxBlock;  // todo remove it
 
   pTq->tqReaderSetSubmitMsg = tqReaderSetSubmitMsg;  // todo remove it
   pTq->tqGetResultBlock = tqGetResultBlock;
 
-  pTq->tqReaderNextBlockFilterOut = tqNextDataBlockFilterOut;
+//  pTq->tqReaderNextBlockFilterOut = tqNextDataBlockFilterOut;
   pTq->tqGetResultBlockTime = tqGetResultBlockTime;
 
   pTq->tqGetStreamExecProgress = tqGetStreamExecInfo;

source/libs/parser/src/parTranslater.c

@@ -4547,7 +4547,7 @@ static int32_t translateWhere(STranslateContext* pCxt, SSelectStmt* pSelect) {
   if (TSDB_CODE_SUCCESS == code) {
     code = getQueryTimeRange(pCxt, pSelect->pWhere, &pSelect->timeRange);
   }
-  if (pSelect->pWhere != NULL) {
+  if (pSelect->pWhere != NULL && pCxt->pParseCxt->topicQuery == false) {
     setTableVgroupsFromEqualTbnameCond(pCxt, pSelect);
   }
   return code;
@@ -8147,7 +8147,9 @@ static int32_t adjustTagsForCreateTable(STranslateContext* pCxt, SCreateStreamSt
     SColumnDefNode* pDef = (SColumnDefNode*)pTagDef;
     if (!dataTypeEqual(&pDef->dataType, &((SExprNode*)pTagExpr)->resType)) {
       SNode*  pFunc = NULL;
-      int32_t code = createCastFunc(pCxt, pTagExpr, pDef->dataType, &pFunc);
+      SDataType defType = pDef->dataType;
+      defType.bytes = calcTypeBytes(defType);
+      int32_t code = createCastFunc(pCxt, pTagExpr, defType, &pFunc);
       if (TSDB_CODE_SUCCESS != code) {
         return code;
       }

source/libs/scheduler/src/schRemote.c

@@ -940,6 +940,10 @@ int32_t schAsyncSendMsg(SSchJob *pJob, SSchTask *pTask, SSchTrans *trans, SQuery
   SCH_ERR_JRET(schGenerateCallBackInfo(pJob, pTask, msg, msgSize, msgType, trans, isHb, &pMsgSendInfo));
   SCH_ERR_JRET(schUpdateSendTargetInfo(pMsgSendInfo, addr, pTask));
 
+  if (isHb && persistHandle && trans->pHandle == 0) {
+    trans->pHandle = rpcAllocHandle();
+  } 
+
   if (pJob && pTask) {
     SCH_TASK_DLOG("start to send %s msg to node[%d,%s,%d], pTrans:%p, pHandle:%p", TMSG_INFO(msgType), addr->nodeId,
            epSet->eps[epSet->inUse].fqdn, epSet->eps[epSet->inUse].port, trans->pTrans, trans->pHandle);

source/libs/stream/src/streamBackendRocksdb.c

@@ -906,6 +906,7 @@ int32_t chkpMayDelObsolete(void* arg, int64_t chkpId, char* path) {
   return 0;
 }
 
+#ifdef BUILD_NO_CALL
 static int32_t chkpIdComp(const void* a, const void* b) {
   int64_t x = *(int64_t*)a;
   int64_t y = *(int64_t*)b;
@@ -964,6 +965,7 @@ int32_t streamBackendLoadCheckpointInfo(void* arg) {
   taosMemoryFree(chkpPath);
   return 0;
 }
+#endif
 
 #ifdef BUILD_NO_CALL
 int32_t chkpGetAllDbCfHandle(SStreamMeta* pMeta, rocksdb_column_family_handle_t*** ppHandle, SArray* refs) {

source/libs/stream/src/streamExec.c

@@ -353,7 +353,8 @@ int32_t streamDoTransferStateToStreamTask(SStreamTask* pTask) {
   if (pStreamTask->info.taskLevel == TASK_LEVEL__SOURCE) {
     ASSERT(status == TASK_STATUS__HALT || status == TASK_STATUS__DROPPING || status == TASK_STATUS__STOP);
   } else {
-    ASSERT(status == TASK_STATUS__READY || status == TASK_STATUS__DROPPING || status == TASK_STATUS__STOP);
+    ASSERT(status == TASK_STATUS__READY || status == TASK_STATUS__PAUSE || status == TASK_STATUS__DROPPING ||
+           status == TASK_STATUS__STOP);
     int32_t code = streamTaskHandleEvent(pStreamTask->status.pSM, TASK_EVENT_HALT);
     if (code != TSDB_CODE_SUCCESS) {
       stError("s-task:%s halt stream task:%s failed, code:%s not transfer state to stream task", id,

source/libs/stream/test/backendTest.cpp

@@ -29,7 +29,7 @@ class BackendEnv : public ::testing::Test {
 
 void *backendCreate() {
   const char *streamPath = "/tmp";
-  void       *p = NULL;
+  void *      p = NULL;
 
   // char *absPath = NULL;
   // // SBackendWrapper *p = (SBackendWrapper *)streamBackendInit(streamPath, -1, 2);
@@ -52,7 +52,7 @@ SStreamState *stateCreate(const char *path) {
 }
 void *backendOpen() {
   streamMetaInit();
-  const char   *path = "/tmp/backend";
+  const char *  path = "/tmp/backend";
   SStreamState *p = stateCreate(path);
   ASSERT(p != NULL);
 
@@ -79,7 +79,7 @@ void *backendOpen() {
 
     const char *val = "value data";
     int32_t     len = 0;
-    char       *newVal = NULL;
+    char *      newVal = NULL;
     streamStateGet_rocksdb(p, &key, (void **)&newVal, &len);
     ASSERT(len == strlen(val));
   }
@@ -100,7 +100,7 @@ void *backendOpen() {
 
     const char *val = "value data";
     int32_t     len = 0;
-    char       *newVal = NULL;
+    char *      newVal = NULL;
     int32_t     code = streamStateGet_rocksdb(p, &key, (void **)&newVal, &len);
     ASSERT(code != 0);
   }
@@ -130,7 +130,7 @@ void *backendOpen() {
 
   winkey.groupId = 0;
   winkey.ts = tsArray[0];
-  char   *val = NULL;
+  char *  val = NULL;
   int32_t len = 0;
 
   pCurr = streamStateSeekKeyNext_rocksdb(p, &winkey);
@@ -157,7 +157,7 @@ void *backendOpen() {
     key.ts = tsArray[i];
     key.exprIdx = i;
 
-    char   *val = NULL;
+    char *  val = NULL;
     int32_t len = 0;
     streamStateFuncGet_rocksdb(p, &key, (void **)&val, &len);
     ASSERT(len == strlen("Value"));
@@ -168,7 +168,7 @@ void *backendOpen() {
     key.ts = tsArray[i];
     key.exprIdx = i;
 
-    char   *val = NULL;
+    char *  val = NULL;
     int32_t len = 0;
     streamStateFuncDel_rocksdb(p, &key);
   }
@@ -213,7 +213,7 @@ void *backendOpen() {
   {
     SSessionKey key;
     memset(&key, 0, sizeof(key));
-    char   *val = NULL;
+    char *  val = NULL;
     int32_t vlen = 0;
     code = streamStateSessionGetKVByCur_rocksdb(pCurr, &key, (void **)&val, &vlen);
     ASSERT(code == 0);
@@ -260,7 +260,7 @@ void *backendOpen() {
     SWinKey key = {0};  // {.groupId = (uint64_t)(i), .ts = tsArray[i]};
     key.groupId = (uint64_t)(i);
     key.ts = tsArray[i];
-    char   *val = NULL;
+    char *  val = NULL;
     int32_t vlen = 0;
     ASSERT(streamStateFillGet_rocksdb(p, &key, (void **)&val, &vlen) == 0);
     taosMemoryFreeClear(val);
@@ -272,7 +272,7 @@ void *backendOpen() {
     SStreamStateCur *pCurr = streamStateFillGetCur_rocksdb(p, &key);
     ASSERT(pCurr != NULL);
 
-    char   *val = NULL;
+    char *  val = NULL;
     int32_t vlen = 0;
     ASSERT(0 == streamStateFillGetKVByCur_rocksdb(pCurr, &key, (const void **)&val, &vlen));
     ASSERT(vlen == strlen("Value"));
@@ -296,7 +296,7 @@ void *backendOpen() {
     SWinKey key = {0};  // {.groupId = (uint64_t)(i), .ts = tsArray[i]};
     key.groupId = (uint64_t)(i);
     key.ts = tsArray[i];
-    char   *val = NULL;
+    char *  val = NULL;
     int32_t vlen = 0;
     ASSERT(streamStateFillDel_rocksdb(p, &key) == 0);
     taosMemoryFreeClear(val);
@@ -338,7 +338,7 @@ void *backendOpen() {
     char key[128] = {0};
     sprintf(key, "tbname_%d", i);
 
-    char   *val = NULL;
+    char *  val = NULL;
     int32_t len = 0;
     code = streamDefaultGet_rocksdb(p, key, (void **)&val, &len);
     ASSERT(code == 0);
@@ -354,7 +354,7 @@ TEST_F(BackendEnv, checkOpen) {
   SStreamState *p = (SStreamState *)backendOpen();
   int64_t       tsStart = taosGetTimestampMs();
   {
-    void   *pBatch = streamStateCreateBatch();
+    void *  pBatch = streamStateCreateBatch();
     int32_t size = 0;
     for (int i = 0; i < size; i++) {
       char key[128] = {0};
@@ -368,7 +368,7 @@ TEST_F(BackendEnv, checkOpen) {
     streamStateDestroyBatch(pBatch);
   }
   {
-    void   *pBatch = streamStateCreateBatch();
+    void *  pBatch = streamStateCreateBatch();
     int32_t size = 0;
     char    valBuf[256] = {0};
     for (int i = 0; i < size; i++) {
@@ -385,7 +385,7 @@ TEST_F(BackendEnv, checkOpen) {
   // do checkpoint 2
   taskDbDoCheckpoint(p->pTdbState->pOwner->pBackend, 2);
   {
-    void   *pBatch = streamStateCreateBatch();
+    void *  pBatch = streamStateCreateBatch();
     int32_t size = 0;
     char    valBuf[256] = {0};
     for (int i = 0; i < size; i++) {
@@ -407,12 +407,22 @@ TEST_F(BackendEnv, checkOpen) {
   // taosMkDir(dump);
   taosMulMkDir(dump);
   SBkdMgt *mgt = bkdMgtCreate((char *)path);
-  SArray  *result = taosArrayInit(4, sizeof(void *));
+  SArray * result = taosArrayInit(4, sizeof(void *));
   bkdMgtGetDelta(mgt, p->pTdbState->idstr, 3, result, (char *)dump);
 
+  taskDbDoCheckpoint(p->pTdbState->pOwner->pBackend, 4);
+
+  taosArrayClear(result);
+  bkdMgtGetDelta(mgt, p->pTdbState->idstr, 4, result, (char *)dump);
   bkdMgtDestroy(mgt);
   streamStateClose((SStreamState *)p, true);
+  // {
+  //   taosRemoveDir("/tmp/backend");
+  //   const char *  path = "/tmp/backend";
+  //   SStreamState *p = stateCreate(path);
+  // }
   taosRemoveDir(path);
+  // streamStateClose((SStreamState *)p, true);
 }
 
 TEST_F(BackendEnv, backendChkp) { const char *path = "/tmp"; }
@@ -430,6 +440,20 @@ TEST_F(BackendEnv, backendUtil) {
     ASSERT_EQ(nextPow2((uint32_t)(kvDict[i].k)), kvDict[i].v);
   }
 }
+TEST_F(BackendEnv, oldBackendInit) {
+  const char *path = "/tmp/backend1";
+  taosMulMkDir(path);
+  {
+    SBackendWrapper *p = (SBackendWrapper *)streamBackendInit(path, 10, 10);
+    streamBackendCleanup((void *)p);
+  }
+  {
+    SBackendWrapper *p = (SBackendWrapper *)streamBackendInit(path, 10, 10);
+    streamBackendCleanup((void *)p);
+  }
+
+  taosRemoveDir(path);
+}
 
 int main(int argc, char **argv) {
   testing::InitGoogleTest(&argc, argv);

source/libs/transport/src/transCli.c

@@ -584,8 +584,8 @@ void* destroyConnPool(SCliThrd* pThrd) {
 
 static SCliConn* getConnFromPool(SCliThrd* pThrd, char* key, bool* exceed) {
   void*      pool = pThrd->pool;
-  SConnList* plist = taosHashGet((SHashObj*)pool, key, strlen(key) + 1);
   STrans*    pTranInst = pThrd->pTransInst;
+  SConnList* plist = taosHashGet((SHashObj*)pool, key, strlen(key) + 1);
   if (plist == NULL) {
     SConnList list = {0};
     taosHashPut((SHashObj*)pool, key, strlen(key) + 1, (void*)&list, sizeof(list));
@@ -867,17 +867,18 @@ static void cliDestroyConn(SCliConn* conn, bool clear) {
   QUEUE_INIT(&conn->q);
 
   conn->broken = true;
+  if (conn->list == NULL) {
+    conn->list = taosHashGet((SHashObj*)pThrd->pool, conn->dstAddr, strlen(conn->dstAddr));
+  }
 
-  if (conn->list != NULL) {
-    SConnList* connList = conn->list;
-    connList->list->numOfConn--;
-    connList->size--;
-  } else {
-    if (pThrd->pool) {
-      SConnList* connList = taosHashGet((SHashObj*)pThrd->pool, conn->dstAddr, strlen(conn->dstAddr) + 1);
-      if (connList != NULL) connList->list->numOfConn--;
+  if (conn->list) {
+    SConnList* list = conn->list;
+    list->list->numOfConn--;
+    if (conn->status == ConnInPool) {
+      list->size--;
     }
   }
+
   conn->list = NULL;
   pThrd->newConnCount--;
 

source/libs/transport/src/transSvr.c

@@ -159,7 +159,7 @@ static void uvStartSendResp(SSvrMsg* msg);
 
 static void uvNotifyLinkBrokenToApp(SSvrConn* conn);
 
-static FORCE_INLINE void destroySmsg(SSvrMsg* smsg);
+static FORCE_INLINE void      destroySmsg(SSvrMsg* smsg);
 static FORCE_INLINE SSvrConn* createConn(void* hThrd);
 static FORCE_INLINE void      destroyConn(SSvrConn* conn, bool clear /*clear handle or not*/);
 static FORCE_INLINE void      destroyConnRegArg(SSvrConn* conn);
@@ -527,6 +527,10 @@ void uvOnSendCb(uv_write_t* req, int status) {
       if (!transQueueEmpty(&conn->srvMsgs)) {
         msg = (SSvrMsg*)transQueueGet(&conn->srvMsgs, 0);
         if (msg->type == Register && conn->status == ConnAcquire) {
+          if (conn->regArg.init) {
+            transFreeMsg(conn->regArg.msg.pCont);
+            conn->regArg.init = 0;
+          }
           conn->regArg.notifyCount = 0;
           conn->regArg.init = 1;
           conn->regArg.msg = msg->msg;
@@ -1350,6 +1354,11 @@ void uvHandleRegister(SSvrMsg* msg, SWorkThrd* thrd) {
       return;
     }
     transQueuePop(&conn->srvMsgs);
+
+    if (conn->regArg.init) {
+      transFreeMsg(conn->regArg.msg.pCont);
+      conn->regArg.init = 0;
+    }
     conn->regArg.notifyCount = 0;
     conn->regArg.init = 1;
     conn->regArg.msg = msg->msg;

tests/parallel_test/cases.task

@@ -228,7 +228,7 @@
 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/stbTagFilter-1ctb.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/dataFromTsdbNWal.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/dataFromTsdbNWal-multiCtb.py
-,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmq_taosx.py
+,,n,system-test,python3 ./test.py -f 7-tmq/tmq_taosx.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmq_replay.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqSeekAndCommit.py
 ,,n,system-test,python3 ./test.py -f 7-tmq/tmq_offset.py

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

@@ -359,8 +359,74 @@ class TDTestCase:
         finally:
             consumer.close()
 
+    def consume_TS_4540_Test(self):
+        tdSql.execute(f'create database if not exists test')
+        tdSql.execute(f'use test')
+        tdSql.execute(f'CREATE STABLE `test`.`b` ( `time` TIMESTAMP , `task_id` NCHAR(1000) ) TAGS( `key` NCHAR(1000))')
+        tdSql.execute(f"insert into `test`.b1 using `test`.`b`(key) tags('1') (time, task_id) values ('2024-03-04 12:50:01.000', '32') `test`.b2 using `test`.`b`(key) tags('2') (time, task_id) values ('2024-03-04 12:50:01.000', '43') `test`.b3 using `test`.`b`(key) tags('3') (time, task_id) values ('2024-03-04 12:50:01.000', '123456')")
+
+        tdSql.execute(f'create topic tt as select tbname,task_id,key from b')
+
+        consumer_dict = {
+            "group.id": "g1",
+            "td.connect.user": "root",
+            "td.connect.pass": "taosdata",
+            "auto.offset.reset": "earliest",
+        }
+        consumer = Consumer(consumer_dict)
+
+        try:
+          consumer.subscribe(["tt"])
+        except TmqError:
+            tdLog.exit(f"subscribe error")
+
+        try:
+            while True:
+                res = consumer.poll(1)
+                if not res:
+                    break
+                val = res.value()
+                if val is None:
+                    continue
+                for block in val:
+                    data = block.fetchall()
+                    print(data)
+                    if data != [('b1', '32', '1')] and data != [('b2', '43', '2')] and data != [('b3', '123456', '3')]:
+                        tdLog.exit(f"index = 0 table b1 error")
+
+        finally:
+            consumer.close()
+        
+    def consume_ts_4544(self):
+        tdSql.execute(f'create database if not exists d1')
+        tdSql.execute(f'use d1')
+        tdSql.execute(f'create table stt(ts timestamp, i int) tags(t int)')
+        tdSql.execute(f'insert into tt1 using stt tags(1) values(now, 1) (now+1s, 2)')
+        tdSql.execute(f'insert into tt2 using stt tags(2) values(now, 1) (now+1s, 2)')
+        tdSql.execute(f'insert into tt3 using stt tags(3) values(now, 1) (now+1s, 2)')
+        tdSql.execute(f'insert into tt1 using stt tags(1) values(now+5s, 11) (now+10s, 12)')
+
+        tdSql.execute(f'create topic topic_in as select * from stt where tbname in ("tt2")')
+
+        consumer_dict = {
+            "group.id": "g1",
+            "td.connect.user": "root",
+            "td.connect.pass": "taosdata",
+            "auto.offset.reset": "earliest",
+        }
+        consumer = Consumer(consumer_dict)
+
+        try:
+            consumer.subscribe(["topic_in"])
+        except TmqError:
+            tdLog.exit(f"subscribe error")
+
+        consumer.close()
+
     def run(self):
         self.consumeTest()
+        self.consume_ts_4544()
+        self.consume_TS_4540_Test()
 
         tdSql.prepare()
         self.checkWal1VgroupOnlyMeta()