Merge pull request #24311 from taosdata/fix/TD-28032

fix:[TD-28032]concat subtable and hash val as new subtable name to av…

include/common/tdatablock.h

@@ -262,6 +262,9 @@ char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** dumpBuf, c
 int32_t buildSubmitReqFromDataBlock(SSubmitReq2** pReq, const SSDataBlock* pDataBlocks, const STSchema* pTSchema, int64_t uid, int32_t vgId,
                                     tb_uid_t suid);
 
+bool  alreadyAddGroupId(char* ctbName);
+bool  isAutoTableName(char* ctbName);
+void  buildCtbNameAddGruopId(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/libs/stream/tstream.h

@@ -62,9 +62,10 @@ typedef struct SStreamTask   SStreamTask;
 typedef struct SStreamQueue  SStreamQueue;
 typedef struct SStreamTaskSM SStreamTaskSM;
 
-#define SSTREAM_TASK_VER              2
+#define SSTREAM_TASK_VER              3
 #define SSTREAM_TASK_INCOMPATIBLE_VER 1
 #define SSTREAM_TASK_NEED_CONVERT_VER 2
+#define SSTREAM_TASK_SUBTABLE_CHANGED_VER 3
 
 enum {
   STREAM_STATUS__NORMAL = 0,

source/common/src/tdatablock.c

@@ -2118,6 +2118,30 @@ _end:
   return TSDB_CODE_SUCCESS;
 }
 
+void  buildCtbNameAddGruopId(char* ctbName, uint64_t groupId){
+  char tmp[TSDB_TABLE_NAME_LEN] = {0};
+  snprintf(tmp, TSDB_TABLE_NAME_LEN, "_%"PRIu64, groupId);
+  ctbName[TSDB_TABLE_NAME_LEN - strlen(tmp) - 1] = 0;  // put groupId to the end
+  strcat(ctbName, tmp);
+}
+
+bool  isAutoTableName(char* ctbName){
+  return (strlen(ctbName) == 34 && ctbName[0] == 't' && ctbName[1] == '_');
+}
+
+bool  alreadyAddGroupId(char* ctbName){
+  size_t len = strlen(ctbName);
+  size_t _location = len - 1;
+  while(_location > 0){
+    if(ctbName[_location] < '0' || ctbName[_location] > '9'){
+      break;
+    }
+    _location--;
+  }
+
+  return ctbName[_location] == '_' &&  len - 1 - _location > 15;  //15 means the min length of groupid
+}
+
 char* buildCtbNameByGroupId(const char* stbFullName, uint64_t groupId) {
   char* pBuf = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN + 1);
   if (!pBuf) {

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

@@ -146,7 +146,7 @@ int32_t         tqOffsetCommitFile(STqOffsetStore* pStore);
 
 // tqSink
 int32_t tqBuildDeleteReq(STQ* pTq, const char* stbFullName, const SSDataBlock* pDataBlock, SBatchDeleteReq* deleteReq,
-                         const char* pIdStr);
+                         const char* pIdStr, bool newSubTableRule);
 void    tqSinkDataIntoDstTable(SStreamTask* pTask, void* vnode, void* data);
 
 // tqOffset
@@ -165,7 +165,7 @@ int32_t setDstTableDataPayload(uint64_t suid, const STSchema* pTSchema, int32_t
 int32_t doMergeExistedRows(SSubmitTbData* pExisted, const SSubmitTbData* pNew, const char* id);
 
 SVCreateTbReq* buildAutoCreateTableReq(const char* stbFullName, int64_t suid, int32_t numOfCols,
-                                       SSDataBlock* pDataBlock, SArray* pTagArray);
+                                       SSDataBlock* pDataBlock, SArray* pTagArray, bool newSubTableRule);
 
 #ifdef __cplusplus
 }

source/dnode/vnode/src/sma/smaRollup.c

@@ -703,7 +703,7 @@ static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSma
           code = terrno;
           TSDB_CHECK_CODE(code, lino, _exit);
         }
-        code = tqBuildDeleteReq(pSma->pVnode->pTq, NULL, output, &deleteReq, "");
+        code = tqBuildDeleteReq(pSma->pVnode->pTq, NULL, output, &deleteReq, "", true);
         TSDB_CHECK_CODE(code, lino, _exit);
         code = tdRSmaProcessDelReq(pSma, suid, pItem->level, &deleteReq);
         TSDB_CHECK_CODE(code, lino, _exit);

source/dnode/vnode/src/sma/smaTimeRange.c

@@ -188,7 +188,7 @@ int32_t smaBlockToSubmit(SVnode *pVnode, const SArray *pBlocks, const STSchema *
     if (pDataBlock->info.type == STREAM_DELETE_RESULT) {
       pDeleteReq->suid = suid;
       pDeleteReq->deleteReqs = taosArrayInit(0, sizeof(SSingleDeleteReq));
-      code = tqBuildDeleteReq(pVnode->pTq, stbFullName, pDataBlock, pDeleteReq, "");
+      code = tqBuildDeleteReq(pVnode->pTq, stbFullName, pDataBlock, pDeleteReq, "", true);
       TSDB_CHECK_CODE(code, lino, _exit);
       continue;
     }
@@ -196,7 +196,7 @@ int32_t smaBlockToSubmit(SVnode *pVnode, const SArray *pBlocks, const STSchema *
     SSubmitTbData tbData = {.suid = suid, .uid = 0, .sver = pTSchema->version, .flags = SUBMIT_REQ_AUTO_CREATE_TABLE,};
 
     int32_t cid = taosArrayGetSize(pDataBlock->pDataBlock) + 1;
-    tbData.pCreateTbReq = buildAutoCreateTableReq(stbFullName, suid, cid, pDataBlock, tagArray);
+    tbData.pCreateTbReq = buildAutoCreateTableReq(stbFullName, suid, cid, pDataBlock, tagArray, true);
 
     {
       uint64_t groupId = pDataBlock->info.id.groupId;

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

@@ -41,10 +41,10 @@ static bool    isValidDstChildTable(SMetaReader* pReader, int32_t vgId, const ch
 static int32_t initCreateTableMsg(SVCreateTbReq* pCreateTableReq, uint64_t suid, const char* stbFullName, int32_t numOfTags);
 static SArray* createDefaultTagColName();
 static void    setCreateTableMsgTableName(SVCreateTbReq* pCreateTableReq, SSDataBlock* pDataBlock, const char* stbFullName,
-                                   int64_t gid);
+                                   int64_t gid, bool newSubTableRule);
 
 int32_t tqBuildDeleteReq(STQ* pTq, const char* stbFullName, const SSDataBlock* pDataBlock, SBatchDeleteReq* deleteReq,
-                         const char* pIdStr) {
+                         const char* pIdStr, bool newSubTableRule) {
   int32_t          totalRows = pDataBlock->info.rows;
   SColumnInfoData* pStartTsCol = taosArrayGet(pDataBlock->pDataBlock, START_TS_COLUMN_INDEX);
   SColumnInfoData* pEndTsCol = taosArrayGet(pDataBlock->pDataBlock, END_TS_COLUMN_INDEX);
@@ -68,6 +68,12 @@ 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) {
+        buildCtbNameAddGruopId(name, groupId);
+      }
     } else if (stbFullName) {
       name = buildCtbNameByGroupId(stbFullName, groupId);
     } else {
@@ -173,9 +179,18 @@ SArray* createDefaultTagColName() {
 }
 
 void setCreateTableMsgTableName(SVCreateTbReq* pCreateTableReq, SSDataBlock* pDataBlock, const char* stbFullName,
-                                int64_t gid) {
+                                int64_t gid, bool newSubTableRule) {
   if (pDataBlock->info.parTbName[0]) {
-    pCreateTableReq->name = taosStrdup(pDataBlock->info.parTbName);
+    if(newSubTableRule &&
+        !isAutoTableName(pDataBlock->info.parTbName) &&
+        !alreadyAddGroupId(pDataBlock->info.parTbName) &&
+        gid != 0) {
+      pCreateTableReq->name = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN);
+      strcpy(pCreateTableReq->name, pDataBlock->info.parTbName);
+      buildCtbNameAddGruopId(pCreateTableReq->name, gid);
+    }else{
+      pCreateTableReq->name = taosStrdup(pDataBlock->info.parTbName);
+    }
   } else {
     pCreateTableReq->name = buildCtbNameByGroupId(stbFullName, gid);
   }
@@ -247,7 +262,7 @@ static int32_t doBuildAndSendCreateTableMsg(SVnode* pVnode, char* stbFullName, S
       ASSERT(gid == *(int64_t*)pGpIdData);
     }
 
-    setCreateTableMsgTableName(pCreateTbReq, pDataBlock, stbFullName, gid);
+    setCreateTableMsgTableName(pCreateTbReq, pDataBlock, stbFullName, gid, pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER);
 
     taosArrayPush(reqs.pArray, pCreateTbReq);
     tqDebug("s-task:%s build create table:%s msg complete", pTask->id.idStr, pCreateTbReq->name);
@@ -357,7 +372,8 @@ int32_t doBuildAndSendDeleteMsg(SVnode* pVnode, char* stbFullName, SSDataBlock*
                                 int64_t suid) {
   SBatchDeleteReq deleteReq = {.suid = suid, .deleteReqs = taosArrayInit(0, sizeof(SSingleDeleteReq))};
 
-  int32_t code = tqBuildDeleteReq(pVnode->pTq, stbFullName, pDataBlock, &deleteReq, pTask->id.idStr);
+  int32_t code = tqBuildDeleteReq(pVnode->pTq, stbFullName, pDataBlock, &deleteReq, pTask->id.idStr,
+                                  pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER);
   if (code != TSDB_CODE_SUCCESS) {
     return code;
   }
@@ -411,7 +427,7 @@ bool isValidDstChildTable(SMetaReader* pReader, int32_t vgId, const char* ctbNam
 }
 
 SVCreateTbReq* buildAutoCreateTableReq(const char* stbFullName, int64_t suid, int32_t numOfCols,
-                                       SSDataBlock* pDataBlock, SArray* pTagArray) {
+                                       SSDataBlock* pDataBlock, SArray* pTagArray, bool newSubTableRule) {
   SVCreateTbReq* pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));
   if (pCreateTbReq == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -436,7 +452,7 @@ SVCreateTbReq* buildAutoCreateTableReq(const char* stbFullName, int64_t suid, in
   pCreateTbReq->ctb.tagName = createDefaultTagColName();
 
   // set table name
-  setCreateTableMsgTableName(pCreateTbReq, pDataBlock, stbFullName, pDataBlock->info.id.groupId);
+  setCreateTableMsgTableName(pCreateTbReq, pDataBlock, stbFullName, pDataBlock->info.id.groupId, newSubTableRule);
   return pCreateTbReq;
 }
 
@@ -637,6 +653,7 @@ int32_t setDstTableDataUid(SVnode* pVnode, SStreamTask* pTask, SSDataBlock* pDat
       tqDebug("s-task:%s vgId:%d, gropuId:%" PRIu64 " datablock table name is null, set name:%s", id, vgId, groupId,
               dstTableName);
     } else {
+      tstrncpy(dstTableName, pTableSinkInfo->name.data, pTableSinkInfo->name.len + 1);
       if (pTableSinkInfo->uid != 0) {
         tqDebug("s-task:%s write %d rows into groupId:%" PRIu64 " dstTable:%s(uid:%" PRIu64 ")", id, numOfRows, groupId,
                 dstTableName, pTableSinkInfo->uid);
@@ -649,10 +666,17 @@ int32_t setDstTableDataUid(SVnode* pVnode, SStreamTask* pTask, SSDataBlock* pDat
     if (dstTableName[0] == 0) {
       memset(dstTableName, 0, TSDB_TABLE_NAME_LEN);
       buildCtbNameByGroupIdImpl(stbFullName, groupId, dstTableName);
+    }else{
+      if(pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER &&
+          !isAutoTableName(dstTableName) &&
+          !alreadyAddGroupId(dstTableName) &&
+          groupId != 0) {
+        buildCtbNameAddGruopId(dstTableName, groupId);
+      }
     }
 
     int32_t nameLen = strlen(dstTableName);
-    pTableSinkInfo = taosMemoryCalloc(1, sizeof(STableSinkInfo) + nameLen);
+    pTableSinkInfo = taosMemoryCalloc(1, sizeof(STableSinkInfo) + nameLen + 1);
     if (pTableSinkInfo == NULL) {
       return TSDB_CODE_OUT_OF_MEMORY;
     }
@@ -690,7 +714,7 @@ int32_t setDstTableDataUid(SVnode* pVnode, SStreamTask* pTask, SSDataBlock* pDat
 
       pTableData->flags = SUBMIT_REQ_AUTO_CREATE_TABLE;
       pTableData->pCreateTbReq =
-          buildAutoCreateTableReq(stbFullName, suid, pTSchema->numOfCols + 1, pDataBlock, pTagArray);
+          buildAutoCreateTableReq(stbFullName, suid, pTSchema->numOfCols + 1, pDataBlock, pTagArray, pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER);
       taosArrayDestroy(pTagArray);
 
       if (pTableData->pCreateTbReq == NULL) {
@@ -719,7 +743,7 @@ int32_t setDstTableDataUid(SVnode* pVnode, SStreamTask* pTask, SSDataBlock* pDat
     }
   }
 
-  return TSDB_CODE_SUCCESS;
+  return TDB_CODE_SUCCESS;
 }
 
 int32_t setDstTableDataPayload(uint64_t suid, const STSchema *pTSchema, int32_t blockIndex, SSDataBlock* pDataBlock,

source/libs/stream/src/streamDispatch.c

@@ -547,25 +547,21 @@ int32_t streamSearchAndAddBlock(SStreamTask* pTask, SStreamDispatchReq* pReqs, S
       memcpy(pDataBlock->info.parTbName, pBln->parTbName, strlen(pBln->parTbName));
     }
   } else {
-    char* ctbName = taosMemoryCalloc(1, TSDB_TABLE_FNAME_LEN);
+    char ctbName[TSDB_TABLE_FNAME_LEN] = {0};
-    if (ctbName == NULL) {
-      return -1;
-    }
-
     if (pDataBlock->info.parTbName[0]) {
-      snprintf(ctbName, TSDB_TABLE_NAME_LEN, "%s.%s", pTask->outputInfo.shuffleDispatcher.dbInfo.db,
+      if(pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER &&
-               pDataBlock->info.parTbName);
+          !isAutoTableName(pDataBlock->info.parTbName) &&
+          !alreadyAddGroupId(pDataBlock->info.parTbName) &&
+          groupId != 0){
+        buildCtbNameAddGruopId(pDataBlock->info.parTbName, groupId);
+      }
     } else {
       buildCtbNameByGroupIdImpl(pTask->outputInfo.shuffleDispatcher.stbFullName, groupId, pDataBlock->info.parTbName);
-      snprintf(ctbName, TSDB_TABLE_NAME_LEN, "%s.%s", pTask->outputInfo.shuffleDispatcher.dbInfo.db,
-               pDataBlock->info.parTbName);
     }
-
+    snprintf(ctbName, TSDB_TABLE_NAME_LEN, "%s.%s", pTask->outputInfo.shuffleDispatcher.dbInfo.db, pDataBlock->info.parTbName);
     /*uint32_t hashValue = MurmurHash3_32(ctbName, strlen(ctbName));*/
     SUseDbRsp* pDbInfo = &pTask->outputInfo.shuffleDispatcher.dbInfo;
-    hashValue =
+    hashValue = taosGetTbHashVal(ctbName, strlen(ctbName), pDbInfo->hashMethod, pDbInfo->hashPrefix, pDbInfo->hashSuffix);
-        taosGetTbHashVal(ctbName, strlen(ctbName), pDbInfo->hashMethod, pDbInfo->hashPrefix, pDbInfo->hashSuffix);
-    taosMemoryFree(ctbName);
     SBlockName bln = {0};
     bln.hashValue = hashValue;
     memcpy(bln.parTbName, pDataBlock->info.parTbName, strlen(pDataBlock->info.parTbName));

source/libs/stream/src/streamMeta.c

@@ -180,10 +180,8 @@ int32_t streamMetaCheckBackendCompatible(SStreamMeta* pMeta) {
     }
     if (info.msgVer <= SSTREAM_TASK_INCOMPATIBLE_VER) {
       ret = STREAM_STATA_NO_COMPATIBLE;
-    } else if (info.msgVer == SSTREAM_TASK_NEED_CONVERT_VER) {
+    } else if (info.msgVer >= SSTREAM_TASK_NEED_CONVERT_VER) {
       ret = STREAM_STATA_NEED_CONVERT;
-    } else if (info.msgVer == SSTREAM_TASK_VER) {
-      ret = STREAM_STATA_COMPATIBLE;
     }
     tDecoderClear(&decoder);
     break;

source/libs/stream/src/streamTask.c

@@ -213,7 +213,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_VER) return -1;
+  if (pTask->ver <= SSTREAM_TASK_INCOMPATIBLE_VER) return -1;
 
   if (tDecodeI64(pDecoder, &pTask->id.streamId) < 0) return -1;
   if (tDecodeI32(pDecoder, &pTask->id.taskId) < 0) return -1;
@@ -300,7 +300,7 @@ int32_t tDecodeStreamTaskChkInfo(SDecoder* pDecoder, SCheckpointInfo* pChkpInfo)
 
   if (tStartDecode(pDecoder) < 0) return -1;
   if (tDecodeI64(pDecoder, &pChkpInfo->msgVer) < 0) return -1;
-  // if (ver != SSTREAM_TASK_VER) return -1;
+  // if (ver <= SSTREAM_TASK_INCOMPATIBLE_VER) return -1;
 
   if (tDecodeI64(pDecoder, &skip64) < 0) return -1;
   if (tDecodeI32(pDecoder, &skip32) < 0) return -1;
@@ -328,7 +328,7 @@ int32_t tDecodeStreamTaskId(SDecoder* pDecoder, STaskId* pTaskId) {
   int64_t ver;
   if (tStartDecode(pDecoder) < 0) return -1;
   if (tDecodeI64(pDecoder, &ver) < 0) return -1;
-  if (ver != SSTREAM_TASK_VER) return -1;
+  if (ver <= SSTREAM_TASK_INCOMPATIBLE_VER) return -1;
 
   if (tDecodeI64(pDecoder, &pTaskId->streamId) < 0) return -1;
 

tests/pytest/util/common.py

@@ -1845,6 +1845,23 @@ class TDCom:
                 if i == 1:
                     self.record_history_ts = ts_value
 
+    def get_subtable(self, tbname_pre):
+        tdSql.query(f'show tables')
+        tbname_list = list(map(lambda x:x[0], tdSql.queryResult))
+        for tbname in tbname_list:
+            if tbname_pre in tbname:
+                return tbname
+
+    def get_subtable_wait(self, tbname_pre):
+        tbname = self.get_subtable(tbname_pre)
+        latency = 0
+        while tbname is None:
+            tbname = self.get_subtable(tbname_pre)
+            if latency < self.stream_timeout:
+                latency += 1
+                time.sleep(1)
+        return tbname
+
 def is_json(msg):
     if isinstance(msg, str):
         try:

tests/script/tsim/stream/udTableAndTag0.sim

@@ -48,16 +48,6 @@ if $rows != 2 then
   goto loop0
 endi
 
-if $data00 != aaa-t1 then
-  print =====data00=$data00
-  goto loop0
-endi
-
-if $data10 != aaa-t2 then
-  print =====data10=$data10
-  goto loop0
-endi
-
 $loop_count = 0
 loop1:
 
@@ -264,17 +254,6 @@ if $rows != 2 then
   goto loop6
 endi
 
-if $data00 != tbn-t1 then
-  print =====data00=$data00
-  goto loop6
-endi
-
-if $data10 != tbn-t2 then
-  print =====data10=$data10
-  goto loop6
-endi
-
-
 print ===== step5
 print ===== tag name + table name
 
@@ -312,21 +291,6 @@ if $rows != 3 then
   goto loop7
 endi
 
-if $data00 != tbn-t1 then
-  print =====data00=$data00
-  goto loop7
-endi
-
-if $data10 != tbn-t2 then
-  print =====data10=$data10
-  goto loop7
-endi
-
-if $data20 != tbn-t3 then
-  print =====data20=$data20
-  goto loop7
-endi
-
 $loop_count = 0
 loop8:
 

tests/script/tsim/stream/udTableAndTag1.sim

@@ -47,16 +47,6 @@ if $rows != 2 then
   goto loop0
 endi
 
-if $data00 != aaa-1 then
-  print =====data00=$data00
-  goto loop0
-endi
-
-if $data10 != aaa-2 then
-  print =====data10=$data10
-  goto loop0
-endi
-
 $loop_count = 0
 loop1:
 
@@ -264,16 +254,6 @@ if $rows != 2 then
   goto loop6
 endi
 
-if $data00 != tbn-1 then
-  print =====data00=$data00
-  goto loop6
-endi
-
-if $data10 != tbn-2 then
-  print =====data10=$data10
-  goto loop6
-endi
-
 print ===== step5
 print ===== tag name + table name
 
@@ -311,21 +291,6 @@ if $rows != 3 then
   goto loop7
 endi
 
-if $data00 != tbn-t1 then
-  print =====data00=$data00
-  goto loop7
-endi
-
-if $data10 != tbn-t2 then
-  print =====data10=$data10
-  goto loop7
-endi
-
-if $data20 != tbn-t3 then
-  print =====data20=$data20
-  goto loop7
-endi
-
 $loop_count = 0
 loop8:
 

tests/script/tsim/stream/udTableAndTag2.sim

@@ -415,11 +415,6 @@ if $data00 != aaa then
   goto loop9
 endi
 
-if $data10 != aaa-1 then
-  print =====data00=$data00
-  goto loop9
-endi
-
 $loop_count = 0
 loop10:
 

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

@@ -106,14 +106,18 @@ class TDTestCase:
                 ptn_counter = 0
                 for c1_value in tdSql.queryResult:
                     if partition == "c1":
-                        tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}`;')
+                        tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`')
                     elif partition is None:
-                        tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}no_partition{self.tdCom.subtable_suffix}`;')
+                        tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}no_partition{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`')
                     elif partition == "abs(c1)":
                         abs_c1_value = abs(c1_value[1])
-                        tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;')
+                        tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`')
                     elif partition == "tbname" and ptn_counter == 0:
-                        tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}`;')
+                        tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`')
                         ptn_counter += 1
                     tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True)
 
@@ -121,14 +125,18 @@ class TDTestCase:
             ptn_counter = 0
             for c1_value in tdSql.queryResult:
                 if partition == "c1":
-                    tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}`;')
+                    tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}')
+                    tdSql.query(f'select count(*) from `{tbname}`')
                 elif partition is None:
-                    tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}no_partition{self.tdCom.subtable_suffix}`;')
+                    tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}no_partition{self.tdCom.subtable_suffix}')
+                    tdSql.query(f'select count(*) from `{tbname}`')
                 elif partition == "abs(c1)":
                     abs_c1_value = abs(c1_value[1])
-                    tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;')
+                    tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}')
+                    tdSql.query(f'select count(*) from `{tbname}`')
                 elif partition == "tbname" and ptn_counter == 0:
-                    tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}`;')
+                    tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}')
+                    tdSql.query(f'select count(*) from `{tbname}`')
                     ptn_counter += 1
 
                 tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True)
@@ -208,8 +216,8 @@ class TDTestCase:
         self.at_once_interval(interval=random.randint(10, 15), partition=None, delete=True)
         self.at_once_interval(interval=random.randint(10, 15), partition=self.tdCom.stream_case_when_tbname, case_when=f'case when {self.tdCom.stream_case_when_tbname} = tbname then {self.tdCom.partition_tbname_alias} else tbname end')
         self.at_once_interval(interval=random.randint(10, 15), partition="tbname", fill_history_value=1, fill_value="NULL")
-        # for fill_value in ["NULL", "PREV", "NEXT", "LINEAR", "VALUE,1,2,3,4,5,6,7,8,9,10,11,1,2,3,4,5,6,7,8,9,10,11"]:
+        for fill_value in ["NULL", "PREV", "NEXT", "LINEAR", "VALUE,1,2,3,4,5,6,7,8,9,10,11,1,2,3,4,5,6,7,8,9,10,11"]:
-        for fill_value in ["PREV", "NEXT", "LINEAR", "VALUE,1,2,3,4,5,6,7,8,9,10,11,1,2,3,4,5,6,7,8,9,10,11"]:
+        # for fill_value in ["PREV", "NEXT", "LINEAR", "VALUE,1,2,3,4,5,6,7,8,9,10,11,1,2,3,4,5,6,7,8,9,10,11"]:
             self.at_once_interval(interval=random.randint(10, 15), partition="tbname", fill_value=fill_value)
             self.at_once_interval(interval=random.randint(10, 15), partition="tbname", fill_value=fill_value, delete=True)
 

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

@@ -153,12 +153,15 @@ class TDTestCase:
                 ptn_counter = 0
                 for c1_value in tdSql.queryResult:
                     if partition == "c1":
-                        tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}`;')
+                        tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`')
                     elif partition == "abs(c1)":
                         abs_c1_value = abs(c1_value[1])
-                        tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;')
+                        tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`')
                     elif partition == "tbname" and ptn_counter == 0:
-                        tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}`;')
+                        tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`')
                         ptn_counter += 1
                     else:
                         tdSql.query(f'select cast(cast(cast({c1_value[1]} as int unsigned) as bigint) as varchar(100))')
@@ -166,7 +169,8 @@ class TDTestCase:
                         if subtable == "constant":
                             return
                         else:
-                            tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{subtable_value}{self.tdCom.subtable_suffix}`;')
+                            tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{subtable_value}{self.tdCom.subtable_suffix}')
+                            tdSql.query(f'select count(*) from `{tbname}`')
                     tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True)
 
     def run(self):

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

@@ -171,35 +171,43 @@ class TDTestCase:
             for c1_value in tdSql.queryResult:
                 if c1_value[1] is not None:
                     if partition == "c1":
-                        tdSql.query(f'select count(*) from `{self.ctb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}`;')
+                        tbname = self.tdCom.get_subtable_wait(f'{self.ctb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`')
                     elif partition == "abs(c1)":
                         if subtable:
                             abs_c1_value = abs(c1_value[1])
-                            tdSql.query(f'select count(*) from `{self.ctb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;')
+                            tbname = self.tdCom.get_subtable_wait(f'{self.ctb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}')
+                            tdSql.query(f'select count(*) from `{tbname}`')
                         else:
-                            tdSql.query(f'select count(*) from `{self.ctb_name}_{self.tdCom.subtable_prefix}{partition_elm_alias}{self.tdCom.subtable_suffix}`;')
+                            tbname = self.tdCom.get_subtable_wait(f'{self.ctb_name}_{self.tdCom.subtable_prefix}{partition_elm_alias}{self.tdCom.subtable_suffix}')
+                            tdSql.query(f'select count(*) from `{tbname}`')
                     elif partition == "tbname" and ptn_counter == 0:
-                        tdSql.query(f'select count(*) from `{self.ctb_name}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}`;')
+                        tbname = self.tdCom.get_subtable_wait(f'{self.ctb_name}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`')
                         ptn_counter += 1
-                    tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True)
+                    tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True) if subtable is not None else tdSql.checkEqual(tdSql.queryResult[0][0] >= 0, True)
 
             tdSql.query(f'select * from {self.tb_name}')
             ptn_counter = 0
             for c1_value in tdSql.queryResult:
                 if c1_value[1] is not None:
                     if partition == "c1":
-                        tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}`;')
+                        tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`')
                     elif partition == "abs(c1)":
                         if subtable:
                             abs_c1_value = abs(c1_value[1])
-                            tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;')
+                            tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}')
+                            tdSql.query(f'select count(*) from `{tbname}`')
                         else:
-                            tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{partition_elm_alias}{self.tdCom.subtable_suffix}`;')
+                            tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{partition_elm_alias}{self.tdCom.subtable_suffix}')
+                            tdSql.query(f'select count(*) from `{tbname}`')
                     elif partition == "tbname" and ptn_counter == 0:
-                        tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}`;')
+                        tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`')
                         ptn_counter += 1
 
-                    tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True)
+                    tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True) if subtable is not None else tdSql.checkEqual(tdSql.queryResult[0][0] >= 0, True)
 
 
 
@@ -224,4 +232,4 @@ class TDTestCase:
 event = threading.Event()
 
 tdCases.addLinux(__file__, TDTestCase())
-tdCases.addWindows(__file__, TDTestCase())
+tdCases.addWindows(__file__, TDTestCase())

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

@@ -94,15 +94,19 @@ class TDTestCase:
             for c1_value in tdSql.queryResult:
                 if partition == "c1":
                     if subtable:
-                        tdSql.query(f'select count(*) from `{self.ctb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}`;')
+                        tbname = self.tdCom.get_subtable_wait(f'{self.ctb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`')
                     else:
-                        tdSql.query(f'select count(*) from `{self.ctb_name}_{self.tdCom.subtable_prefix}{partition_elm_alias}{self.tdCom.subtable_suffix}`;')
+                        tbname = self.tdCom.get_subtable_wait(f'{self.ctb_name}_{self.tdCom.subtable_prefix}{partition_elm_alias}{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`')
                         return
                 elif partition == "abs(c1)":
                     abs_c1_value = abs(c1_value[1])
-                    tdSql.query(f'select count(*) from `{self.ctb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;')
+                    tbname = self.tdCom.get_subtable_wait(f'{self.ctb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}')
+                    tdSql.query(f'select count(*) from `{tbname}`')
                 elif partition == "tbname" and ptn_counter == 0:
-                    tdSql.query(f'select count(*) from `{self.ctb_name}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}`;')
+                    tbname = self.tdCom.get_subtable_wait(f'{self.ctb_name}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}')
+                    tdSql.query(f'select count(*) from `{tbname}`')
                     ptn_counter += 1
                 tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True)
 
@@ -111,17 +115,20 @@ class TDTestCase:
             for c1_value in tdSql.queryResult:
                 if partition == "c1":
                     if subtable:
-                        tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}`;')
+                        tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`')
                     else:
-                        tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{partition_elm_alias}{self.tdCom.subtable_suffix}`;')
+                        tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{partition_elm_alias}{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`')
                         return
                 elif partition == "abs(c1)":
                     abs_c1_value = abs(c1_value[1])
-                    tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;')
+                    tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}')
+                    tdSql.query(f'select count(*) from `{tbname}`')
                 elif partition == "tbname" and ptn_counter == 0:
-                    tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}`;')
+                    tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}')
+                    tdSql.query(f'select count(*) from `{tbname}`')
                     ptn_counter += 1
-
                 tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True)
 
 

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

@@ -117,14 +117,18 @@ class TDTestCase:
                 ptn_counter = 0
                 for c1_value in tdSql.queryResult:
                     if partition == "c1":
-                        tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}`;')
+                        tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`')
                     elif partition is None:
-                        tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}no_partition{self.tdCom.subtable_suffix}`;')
+                        tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}no_partition{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`')
                     elif partition == "abs(c1)":
                         abs_c1_value = abs(c1_value[1])
-                        tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;')
+                        tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`')
                     elif partition == "tbname" and ptn_counter == 0:
-                        tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}`;')
+                        tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`')
                         ptn_counter += 1
                     tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True)
 
@@ -132,14 +136,18 @@ class TDTestCase:
             ptn_counter = 0
             for c1_value in tdSql.queryResult:
                 if partition == "c1":
-                    tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}`;')
+                    tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}')
+                    tdSql.query(f'select count(*) from `{tbname}`')
                 elif partition is None:
-                    tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}no_partition{self.tdCom.subtable_suffix}`;')
+                    tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}no_partition{self.tdCom.subtable_suffix}')
+                    tdSql.query(f'select count(*) from `{tbname}`')
                 elif partition == "abs(c1)":
                     abs_c1_value = abs(c1_value[1])
-                    tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;')
+                    tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}')
+                    tdSql.query(f'select count(*) from `{tbname}`')
                 elif partition == "tbname" and ptn_counter == 0:
-                    tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}`;')
+                    tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}')
+                    tdSql.query(f'select count(*) from `{tbname}`')
                     ptn_counter += 1
 
                 tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True)

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

@@ -146,12 +146,15 @@ class TDTestCase:
                 ptn_counter = 0
                 for c1_value in tdSql.queryResult:
                     if partition == "c1":
-                        tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}`;', count_expected_res=self.tdCom.range_count)
+                        tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`', count_expected_res=self.tdCom.range_count)
                     elif partition == "abs(c1)":
                         abs_c1_value = abs(c1_value[1])
-                        tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;', count_expected_res=self.tdCom.range_count)
+                        tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`', count_expected_res=self.tdCom.range_count)
                     elif partition == "tbname" and ptn_counter == 0:
-                        tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}`;', count_expected_res=self.tdCom.range_count)
+                        tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}')
+                        tdSql.query(f'select count(*) from `{tbname}`', count_expected_res=self.tdCom.range_count)
                         ptn_counter += 1
 
                     tdSql.checkEqual(tdSql.queryResult[0][0] , self.tdCom.range_count)
@@ -161,12 +164,15 @@ class TDTestCase:
             ptn_counter = 0
             for c1_value in tdSql.queryResult:
                 if partition == "c1":
-                    tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}`;')
+                    tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}')
+                    tdSql.query(f'select count(*) from `{tbname}`')
                 elif partition == "abs(c1)":
                     abs_c1_value = abs(c1_value[1])
-                    tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;')
+                    tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}')
+                    tdSql.query(f'select count(*) from `{tbname}`')
                 elif partition == "tbname" and ptn_counter == 0:
-                    tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}`;')
+                    tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}')
+                    tdSql.query(f'select count(*) from `{tbname}`')
                     ptn_counter += 1
 
                 tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True)