Merge branch '3.0' of https://github.com/taosdata/TDengine into fix/hzcheng_3.0

cmake/cmake.define

@@ -14,25 +14,6 @@ MESSAGE(STATUS "Project binary files output path: " ${PROJECT_BINARY_DIR})
 MESSAGE(STATUS "Project executable files output path: " ${EXECUTABLE_OUTPUT_PATH})
 MESSAGE(STATUS "Project library files output path: " ${LIBRARY_OUTPUT_PATH})
 
-find_package(Git QUIET)
-if(GIT_FOUND AND EXISTS "${TD_SOURCE_DIR}/.git")
-# Update submodules as needed
-    option(GIT_SUBMODULE "Check submodules during build" ON)
-    if(GIT_SUBMODULE)
-        message(STATUS "Submodule update")
-        execute_process(COMMAND cd ${TD_SOURCE_DIR} && ${GIT_EXECUTABLE} submodule update --init --recursive
-                        WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}
-                        RESULT_VARIABLE GIT_SUBMOD_RESULT)
-        if(NOT GIT_SUBMOD_RESULT EQUAL "0")
-            message(WARNING "git submodule update --init --recursive failed with ${GIT_SUBMOD_RESULT}, please checkout submodules")
-        endif()
-    endif()
-endif()
-
-if(NOT EXISTS "${TD_SOURCE_DIR}/tools/taos-tools/CMakeLists.txt")
-    message(WARNING "The submodules were not downloaded! GIT_SUBMODULE was turned off or failed. Please update submodules manually if you need build them.")
-endif()
-
 if (NOT DEFINED TD_GRANT)
   SET(TD_GRANT FALSE)
 endif()

include/util/taoserror.h

@@ -642,6 +642,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_PAR_INVALID_INTERNAL_PK       TAOS_DEF_ERROR_CODE(0, 0x2646)
 #define TSDB_CODE_PAR_INVALID_TIMELINE_FUNC     TAOS_DEF_ERROR_CODE(0, 0x2647)
 #define TSDB_CODE_PAR_INVALID_PASSWD            TAOS_DEF_ERROR_CODE(0, 0x2648)
+#define TSDB_CODE_PAR_INVALID_ALTER_TABLE       TAOS_DEF_ERROR_CODE(0, 0x2649)
 
 //planner
 #define TSDB_CODE_PLAN_INTERNAL_ERROR           TAOS_DEF_ERROR_CODE(0, 0x2700)

source/client/src/clientSml.c

@@ -1007,8 +1007,6 @@ static int32_t smlParseTelnetTags(const char* data, SArray *cols, SHashObj *dump
     while(*sql != '\0') {
       // parse value
       if (*sql == SPACE) {
-        valueLen = sql - value;
-        sql++;
         break;
       }
       if (*sql == EQUAL) {
@@ -1017,9 +1015,7 @@ static int32_t smlParseTelnetTags(const char* data, SArray *cols, SHashObj *dump
       }
       sql++;
     }
-    if(valueLen == 0){
     valueLen = sql - value;
-    }
 
     if(valueLen == 0){
       smlBuildInvalidDataMsg(msg, "invalid value", value);
@@ -1365,7 +1361,7 @@ static void smlDestroySTableMeta(SSmlSTableMeta *meta){
 static void smlDestroyCols(SArray *cols) {
   if (!cols) return;
   for (int i = 0; i < taosArrayGetSize(cols); ++i) {
-    void *kv = taosArrayGet(cols, i);
+    void *kv = taosArrayGetP(cols, i);
     taosMemoryFree(kv);
   }
 }
@@ -2077,12 +2073,16 @@ static int32_t smlParseTelnetLine(SSmlHandle* info, void *data) {
   if(ret != TSDB_CODE_SUCCESS){
     uError("SML:0x%"PRIx64" smlParseTelnetLine failed", info->id);
     smlDestroyTableInfo(info, tinfo);
+    smlDestroyCols(cols);
     taosArrayDestroy(cols);
     return ret;
   }
 
   if(taosArrayGetSize(tinfo->tags) <= 0 || taosArrayGetSize(tinfo->tags) > TSDB_MAX_TAGS){
     smlBuildInvalidDataMsg(&info->msgBuf, "invalidate tags length:[1,128]", NULL);
+    smlDestroyTableInfo(info, tinfo);
+    smlDestroyCols(cols);
+    taosArrayDestroy(cols);
     return TSDB_CODE_SML_INVALID_DATA;
   }
   taosHashClear(info->dumplicateKey);

source/client/test/smlTest.cpp

@@ -516,8 +516,8 @@ TEST(testCase, smlProcess_influx_Test) {
   int ret = smlProcess(info, (char**)sql, sizeof(sql)/sizeof(sql[0]));
   ASSERT_EQ(ret, 0);
 
-  TAOS_RES *res = taos_query(taos, "select * from t_6885c584b98481584ee13dac399e173d");
-  ASSERT_NE(res, nullptr);
+//  TAOS_RES *res = taos_query(taos, "select * from t_6885c584b98481584ee13dac399e173d");
+//  ASSERT_NE(res, nullptr);
 //  int fieldNum = taos_field_count(res);
 //  ASSERT_EQ(fieldNum, 5);
 //  int rowNum = taos_affected_rows(res);
@@ -525,7 +525,7 @@ TEST(testCase, smlProcess_influx_Test) {
 //  for (int i = 0; i < rowNum; ++i) {
 //    TAOS_ROW rows = taos_fetch_row(res);
 //  }
-  taos_free_result(res);
+//  taos_free_result(res);
   destroyRequest(request);
   smlDestroyInfo(info);
 }
@@ -605,8 +605,8 @@ TEST(testCase, smlProcess_telnet_Test) {
   int ret = smlProcess(info, (char**)sql, sizeof(sql)/sizeof(sql[0]));
   ASSERT_EQ(ret, 0);
 
-  TAOS_RES *res = taos_query(taos, "select * from t_8c30283b3c4131a071d1e16cf6d7094a");
-  ASSERT_NE(res, nullptr);
+//  TAOS_RES *res = taos_query(taos, "select * from t_8c30283b3c4131a071d1e16cf6d7094a");
+//  ASSERT_NE(res, nullptr);
 //  int fieldNum = taos_field_count(res);
 //  ASSERT_EQ(fieldNum, 2);
 //  int rowNum = taos_affected_rows(res);
@@ -614,7 +614,7 @@ TEST(testCase, smlProcess_telnet_Test) {
 //  for (int i = 0; i < rowNum; ++i) {
 //   TAOS_ROW rows = taos_fetch_row(res);
 //  }
-  taos_free_result(res);
+//  taos_free_result(res);
 
 //  res = taos_query(taos, "select * from t_6931529054e5637ca92c78a1ad441961");
 //  ASSERT_NE(res, nullptr);
@@ -670,16 +670,16 @@ TEST(testCase, smlProcess_json1_Test) {
   int ret = smlProcess(info, (char **)(&sql), -1);
   ASSERT_EQ(ret, 0);
 
-  TAOS_RES *res = taos_query(taos, "select * from t_cb27a7198d637b4f1c6464bd73f756a7");
-  ASSERT_NE(res, nullptr);
-  int fieldNum = taos_field_count(res);
-  ASSERT_EQ(fieldNum, 2);
+//  TAOS_RES *res = taos_query(taos, "select * from t_cb27a7198d637b4f1c6464bd73f756a7");
+//  ASSERT_NE(res, nullptr);
+//  int fieldNum = taos_field_count(res);
+//  ASSERT_EQ(fieldNum, 2);
   //   int rowNum = taos_affected_rows(res);
   //   ASSERT_EQ(rowNum, 1);
   //   for (int i = 0; i < rowNum; ++i) {
   //     TAOS_ROW rows = taos_fetch_row(res);
   //   }
-  taos_free_result(res);
+//  taos_free_result(res);
   destroyRequest(request);
   smlDestroyInfo(info);
 }

source/common/src/tmsg.c

@@ -4217,7 +4217,6 @@ int32_t tDecodeSVAlterTbReq(SDecoder *pDecoder, SVAlterTbReq *pReq) {
 
   if (tDecodeCStr(pDecoder, &pReq->tbName) < 0) return -1;
   if (tDecodeI8(pDecoder, &pReq->action) < 0) return -1;
-  if (tDecodeCStr(pDecoder, &pReq->colName) < 0) return -1;
   switch (pReq->action) {
     case TSDB_ALTER_TABLE_ADD_COLUMN:
       if (tDecodeCStr(pDecoder, &pReq->colName) < 0) return -1;

source/libs/catalog/src/catalog.c

@@ -2576,12 +2576,6 @@ int32_t catalogGetHandle(uint64_t clusterId, SCatalog** catalogHandle) {
       CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
     }
 
-    SHashObj *metaCache = taosHashInit(gCtgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
-    if (NULL == metaCache) {
-      qError("taosHashInit failed, num:%d", gCtgMgmt.cfg.maxTblCacheNum);
-      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
-    }
-    
     code = taosHashPut(gCtgMgmt.pCluster, &clusterId, sizeof(clusterId), &clusterCtg, POINTER_BYTES);
     if (code) {
       if (HASH_NODE_EXIST(code)) {

source/libs/executor/inc/executorimpl.h

@@ -658,6 +658,7 @@ void    initExecTimeWindowInfo(SColumnInfoData* pColData, STimeWindow* pQueryWin
 void    cleanupAggSup(SAggSupporter* pAggSup);
 void    destroyBasicOperatorInfo(void* param, int32_t numOfOutput);
 void    appendOneRowToDataBlock(SSDataBlock* pBlock, STupleHandle* pTupleHandle);
+void    setTbNameColData(void* pMeta, const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId);
 SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode);
 SColumn extractColumnFromColumnNode(SColumnNode* pColNode);
 

source/libs/executor/src/executorimpl.c

@@ -2055,6 +2055,11 @@ void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const int8_t* rowR
       SColumnInfoData* pDst = taosArrayGet(px->pDataBlock, i);
       SColumnInfoData* pSrc = taosArrayGet(pBlock->pDataBlock, i);
 
+      // it is a reserved column for scalar function, and no data in this column yet.
+      if (pSrc->pData == NULL) {
+        continue;
+      }
+
       int32_t numOfRows = 0;
       for (int32_t j = 0; j < totalRows; ++j) {
         if (rowRes[j] == 0) {

source/libs/executor/src/scanoperator.c

@@ -291,6 +291,19 @@ void addTagPseudoColumnData(STableScanInfo* pTableScanInfo, SSDataBlock* pBlock)
 
     // this is to handle the tbname
     if (fmIsScanPseudoColumnFunc(functionId)) {
+      setTbNameColData(pTableScanInfo->readHandle.meta, pBlock, pColInfoData, functionId);
+    } else {  // these are tags
+      const char* p = metaGetTableTagVal(&mr.me, pExpr->base.pParam[0].pCol->colId);
+      for (int32_t i = 0; i < pBlock->info.rows; ++i) {
+        colDataAppend(pColInfoData, i, p, (p == NULL));
+      }
+    }
+  }
+
+  metaReaderClear(&mr);
+}
+
+void setTbNameColData(void* pMeta, const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId) {
   struct SScalarFuncExecFuncs fpSet = {0};
   fmGetScalarFuncExecFuncs(functionId, &fpSet);
 
@@ -299,21 +312,12 @@ void addTagPseudoColumnData(STableScanInfo* pTableScanInfo, SSDataBlock* pBlock)
   infoData.info.bytes = sizeof(uint64_t);
   colInfoDataEnsureCapacity(&infoData, 0, 1);
 
-      colDataAppendInt64(&infoData, 0, &pBlock->info.uid);
+  colDataAppendInt64(&infoData, 0, (int64_t*) &pBlock->info.uid);
   SScalarParam srcParam = {
-          .numOfRows = pBlock->info.rows, .param = pTableScanInfo->readHandle.meta, .columnData = &infoData};
+      .numOfRows = pBlock->info.rows, .param = pMeta, .columnData = &infoData};
 
   SScalarParam param = {.columnData = pColInfoData};
   fpSet.process(&srcParam, 1, &param);
-    } else {  // these are tags
-      const char* p = metaGetTableTagVal(&mr.me, pExpr->base.pParam[0].pCol->colId);
-      for (int32_t i = 0; i < pBlock->info.rows; ++i) {
-        colDataAppend(pColInfoData, i, p, (p == NULL));
-      }
-    }
-  }
-
-  metaReaderClear(&mr);
 }
 
 static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {

source/libs/parser/inc/sql.y

@@ -241,7 +241,7 @@ alter_table_clause(A) ::=
 alter_table_clause(A) ::= 
   full_table_name(B) RENAME TAG column_name(C) column_name(D).                    { A = createAlterTableRenameCol(pCxt, B, TSDB_ALTER_TABLE_UPDATE_TAG_NAME, &C, &D); }
 alter_table_clause(A) ::=
-  full_table_name(B) SET TAG column_name(C) NK_EQ literal(D).                     { A = createAlterTableSetTag(pCxt, B, &C, D); }
+  full_table_name(B) SET TAG column_name(C) NK_EQ literal(D).                     { A = createAlterTableSetTag(pCxt, B, &C, releaseRawExprNode(pCxt, D)); }
 
 %type multi_create_clause                                                         { SNodeList* }
 %destructor multi_create_clause                                                   { nodesDestroyList($$); }

source/libs/parser/src/parInsert.c

@@ -1041,18 +1041,6 @@ static void destroyInsertParseContextForTable(SInsertParseContext* pCxt) {
   destroyCreateSubTbReq(&pCxt->createTblReq);
 }
 
-static void destroyDataBlock(STableDataBlocks* pDataBlock) {
-  if (pDataBlock == NULL) {
-    return;
-  }
-
-  taosMemoryFreeClear(pDataBlock->pData);
-  if (!pDataBlock->cloned) {
-    destroyBoundColumnInfo(&pDataBlock->boundColumnInfo);
-  }
-  taosMemoryFreeClear(pDataBlock);
-}
-
 static void destroyInsertParseContext(SInsertParseContext* pCxt) {
   destroyInsertParseContextForTable(pCxt);
   taosHashCleanup(pCxt->pVgroupsHashObj);
@@ -1301,6 +1289,7 @@ int32_t qBuildStmtOutput(SQuery* pQuery, SHashObj* pVgHash, SHashObj* pBlockHash
 
   CHECK_CODE(buildOutput(&insertCtx));
 
+  destroyBlockArrayList(insertCtx.pVgDataBlocks);
   return TSDB_CODE_SUCCESS;
 }
 
@@ -1580,16 +1569,25 @@ int32_t qBuildStmtColFields(void* pBlock, int32_t* fieldNum, TAOS_FIELD** fields
 
 // schemaless logic start
 
-typedef struct SmlExecHandle {
-  SHashObj* pBlockHash;
-
+typedef struct SmlExecTableHandle {
   SParsedDataColInfo tags;          // each table
   SKVRowBuilder      tagsBuilder;   // each table
   SVCreateTbReq      createTblReq;  // each table
+} SmlExecTableHandle;
 
-  SQuery* pQuery;
+typedef struct SmlExecHandle {
+  SHashObj*           pBlockHash;
+  SmlExecTableHandle  tableExecHandle;
+  SQuery              *pQuery;
 } SSmlExecHandle;
 
+static void smlDestroyTableHandle(void* pHandle) {
+  SmlExecTableHandle* handle = (SmlExecTableHandle*)pHandle;
+  tdDestroyKVRowBuilder(&handle->tagsBuilder);
+  destroyBoundColumnInfo(&handle->tags);
+  destroyCreateSubTbReq(&handle->createTblReq);
+}
+
 static int32_t smlBoundColumnData(SArray* cols, SParsedDataColInfo* pColList, SSchema* pSchema) {
   col_id_t nCols = pColList->numOfCols;
 
@@ -1692,25 +1690,26 @@ int32_t smlBindData(void *handle, SArray *tags, SArray *colsSchema, SArray *cols
   SMsgBuf pBuf = {.buf = msgBuf, .len = msgBufLen};
 
   SSmlExecHandle* smlHandle = (SSmlExecHandle*)handle;
+  smlDestroyTableHandle(&smlHandle->tableExecHandle);   // free for each table
   SSchema*        pTagsSchema = getTableTagSchema(pTableMeta);
-  setBoundColumnInfo(&smlHandle->tags, pTagsSchema, getNumOfTags(pTableMeta));
-  int ret = smlBoundColumnData(tags, &smlHandle->tags, pTagsSchema);
+  setBoundColumnInfo(&smlHandle->tableExecHandle.tags, pTagsSchema, getNumOfTags(pTableMeta));
+  int ret = smlBoundColumnData(tags, &smlHandle->tableExecHandle.tags, pTagsSchema);
   if (ret != TSDB_CODE_SUCCESS) {
     buildInvalidOperationMsg(&pBuf, "bound tags error");
     return ret;
   }
   SKVRow row = NULL;
-  ret = smlBuildTagRow(tags, &smlHandle->tagsBuilder, &smlHandle->tags, pTagsSchema, &row, &pBuf);
+  ret = smlBuildTagRow(tags, &smlHandle->tableExecHandle.tagsBuilder, &smlHandle->tableExecHandle.tags, pTagsSchema, &row, &pBuf);
   if (ret != TSDB_CODE_SUCCESS) {
     return ret;
   }
 
-  buildCreateTbReq(&smlHandle->createTblReq, tableName, row, pTableMeta->suid);
+  buildCreateTbReq(&smlHandle->tableExecHandle.createTblReq, tableName, row, pTableMeta->suid);
 
   STableDataBlocks* pDataBlock = NULL;
   ret = getDataBlockFromList(smlHandle->pBlockHash, &pTableMeta->uid, sizeof(pTableMeta->uid),
                              TSDB_DEFAULT_PAYLOAD_SIZE, sizeof(SSubmitBlk), getTableInfo(pTableMeta).rowSize,
-                             pTableMeta, &pDataBlock, NULL, &smlHandle->createTblReq);
+                             pTableMeta, &pDataBlock, NULL, &smlHandle->tableExecHandle.createTblReq);
   if (ret != TSDB_CODE_SUCCESS) {
     buildInvalidOperationMsg(&pBuf, "create data block error");
     return ret;
@@ -1826,9 +1825,7 @@ void smlDestroyHandle(void* pHandle) {
   if (!pHandle) return;
   SSmlExecHandle* handle = (SSmlExecHandle*)pHandle;
   destroyBlockHashmap(handle->pBlockHash);
-  tdDestroyKVRowBuilder(&handle->tagsBuilder);
-  destroyBoundColumnInfo(&handle->tags);
-  destroyCreateSubTbReq(&handle->createTblReq);
+  smlDestroyTableHandle(&handle->tableExecHandle);
   taosMemoryFree(handle);
 }
 

source/libs/parser/src/parInsertData.c

@@ -237,9 +237,7 @@ static void destroyDataBlock(STableDataBlocks* pDataBlock) {
   taosMemoryFreeClear(pDataBlock->pData);
   if (!pDataBlock->cloned) {
     // free the refcount for metermeta
-    if (pDataBlock->pTableMeta != NULL) {
     taosMemoryFreeClear(pDataBlock->pTableMeta);
-    }
 
     destroyBoundColumnInfo(&pDataBlock->boundColumnInfo);
   }

source/libs/parser/src/parTranslater.c

@@ -3800,7 +3800,7 @@ static void destroyCreateTbReqBatch(SVgroupCreateTableBatch* pTbBatch) {
   taosArrayDestroy(pTbBatch->req.pArray);
 }
 
-static int32_t rewriteToVnodeModifOpStmt(SQuery* pQuery, SArray* pBufArray) {
+static int32_t rewriteToVnodeModifyOpStmt(SQuery* pQuery, SArray* pBufArray) {
   SVnodeModifOpStmt* pNewStmt = nodesMakeNode(QUERY_NODE_VNODE_MODIF_STMT);
   if (pNewStmt == NULL) {
     return TSDB_CODE_OUT_OF_MEMORY;
@@ -3855,7 +3855,7 @@ static int32_t rewriteCreateTable(STranslateContext* pCxt, SQuery* pQuery) {
     code = buildCreateTableDataBlock(pCxt->pParseCxt->acctId, pStmt, &info, &pBufArray);
   }
   if (TSDB_CODE_SUCCESS == code) {
-    code = rewriteToVnodeModifOpStmt(pQuery, pBufArray);
+    code = rewriteToVnodeModifyOpStmt(pQuery, pBufArray);
     if (TSDB_CODE_SUCCESS != code) {
       destroyCreateTbReqArray(pBufArray);
     }
@@ -4111,7 +4111,7 @@ static int32_t rewriteCreateMultiTable(STranslateContext* pCxt, SQuery* pQuery)
     return TSDB_CODE_OUT_OF_MEMORY;
   }
 
-  return rewriteToVnodeModifOpStmt(pQuery, pBufArray);
+  return rewriteToVnodeModifyOpStmt(pQuery, pBufArray);
 }
 
 typedef struct SVgroupDropTableBatch {
@@ -4251,12 +4251,160 @@ static int32_t rewriteDropTable(STranslateContext* pCxt, SQuery* pQuery) {
     return TSDB_CODE_OUT_OF_MEMORY;
   }
 
-  return rewriteToVnodeModifOpStmt(pQuery, pBufArray);
+  return rewriteToVnodeModifyOpStmt(pQuery, pBufArray);
+}
+
+static int32_t buildAlterTbReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, SVAlterTbReq* pReq) {
+  pReq->tbName = strdup(pStmt->tableName);
+  if (NULL == pReq->tbName) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  pReq->action = pStmt->alterType;
+
+  switch (pStmt->alterType) {
+    case TSDB_ALTER_TABLE_ADD_TAG:
+    case TSDB_ALTER_TABLE_DROP_TAG:
+    case TSDB_ALTER_TABLE_UPDATE_TAG_NAME:
+    case TSDB_ALTER_TABLE_UPDATE_TAG_BYTES:
+      return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_ALTER_TABLE);
+    case TSDB_ALTER_TABLE_UPDATE_TAG_VAL:
+      pReq->tagName = strdup(pStmt->colName);
+      if (NULL == pReq->tagName) {
+        return TSDB_CODE_OUT_OF_MEMORY;
+      }
+      if (DEAL_RES_ERROR == translateValue(pCxt, pStmt->pVal)) {
+        return pCxt->errCode;
+      }
+      pReq->isNull = (TSDB_DATA_TYPE_NULL == pStmt->pVal->node.resType.type);
+      pReq->nTagVal = pStmt->pVal->node.resType.bytes;
+      char* pVal = nodesGetValueFromNode(pStmt->pVal);
+      pReq->pTagVal = IS_VAR_DATA_TYPE(pStmt->pVal->node.resType.type) ? pVal + VARSTR_HEADER_SIZE : pVal;
+      break;
+    case TSDB_ALTER_TABLE_ADD_COLUMN:
+    case TSDB_ALTER_TABLE_DROP_COLUMN:
+      pReq->colName = strdup(pStmt->colName);
+      if (NULL == pReq->colName) {
+        return TSDB_CODE_OUT_OF_MEMORY;
+      }
+      pReq->type = pStmt->dataType.type;
+      pReq->flags = COL_SMA_ON;
+      pReq->bytes = pStmt->dataType.bytes;
+      break;
+    case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES:
+      pReq->colName = strdup(pStmt->colName);
+      if (NULL == pReq->colName) {
+        return TSDB_CODE_OUT_OF_MEMORY;
+      }
+      pReq->colModBytes = calcTypeBytes(pStmt->dataType);
+      break;
+    case TSDB_ALTER_TABLE_UPDATE_OPTIONS:
+      if (-1 != pStmt->pOptions->ttl) {
+        pReq->updateTTL = true;
+        pReq->newTTL = pStmt->pOptions->ttl;
+      }
+      if ('\0' != pStmt->pOptions->comment[0]) {
+        pReq->updateComment = true;
+        pReq->newComment = strdup(pStmt->pOptions->comment);
+        if (NULL == pReq->newComment) {
+          return TSDB_CODE_OUT_OF_MEMORY;
+        }
+      }
+      break;
+    case TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME:
+      pReq->colName = strdup(pStmt->colName);
+      pReq->colNewName = strdup(pStmt->newColName);
+      if (NULL == pReq->colName || NULL == pReq->colNewName) {
+        return TSDB_CODE_OUT_OF_MEMORY;
+      }
+      break;
+    default:
+      break;
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t serializeAlterTbReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, SVAlterTbReq* pReq,
+                                   SArray* pArray) {
+  SVgroupInfo vg = {0};
+  int32_t     code = getTableHashVgroup(pCxt, pStmt->dbName, pStmt->tableName, &vg);
+  int         tlen = 0;
+  if (TSDB_CODE_SUCCESS == code) {
+    tEncodeSize(tEncodeSVAlterTbReq, pReq, tlen, code);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    tlen += sizeof(SMsgHead);
+    void* pMsg = taosMemoryMalloc(tlen);
+    if (NULL == pMsg) {
+      return TSDB_CODE_OUT_OF_MEMORY;
+    }
+    ((SMsgHead*)pMsg)->vgId = htonl(vg.vgId);
+    ((SMsgHead*)pMsg)->contLen = htonl(tlen);
+    void*    pBuf = POINTER_SHIFT(pMsg, sizeof(SMsgHead));
+    SEncoder coder = {0};
+    tEncoderInit(&coder, pBuf, tlen - sizeof(SMsgHead));
+    tEncodeSVAlterTbReq(&coder, pReq);
+    tEncoderClear(&coder);
+
+    SVgDataBlocks* pVgData = taosMemoryCalloc(1, sizeof(SVgDataBlocks));
+    if (NULL == pVgData) {
+      taosMemoryFree(pMsg);
+      return TSDB_CODE_OUT_OF_MEMORY;
+    }
+    pVgData->vg = vg;
+    pVgData->pData = pMsg;
+    pVgData->size = tlen;
+    pVgData->numOfTables = 1;
+    taosArrayPush(pArray, &pVgData);
+  }
+
+  return code;
+}
+
+static int32_t buildModifyVnodeArray(STranslateContext* pCxt, SAlterTableStmt* pStmt, SVAlterTbReq* pReq,
+                                     SArray** pArray) {
+  SArray* pTmpArray = taosArrayInit(1, sizeof(void*));
+  if (NULL == pTmpArray) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  int32_t code = serializeAlterTbReq(pCxt, pStmt, pReq, pTmpArray);
+  if (TSDB_CODE_SUCCESS == code) {
+    *pArray = pTmpArray;
+  } else {
+    taosArrayDestroy(pTmpArray);
+  }
+
+  return code;
 }
 
 static int32_t rewriteAlterTable(STranslateContext* pCxt, SQuery* pQuery) {
-  // todo
+  SAlterTableStmt* pStmt = (SAlterTableStmt*)pQuery->pRoot;
+
+  STableMeta* pTableMeta = NULL;
+  int32_t     code = getTableMeta(pCxt, pStmt->dbName, pStmt->tableName, &pTableMeta);
+  if (TSDB_CODE_SUCCESS != code) {
+    return code;
+  }
+
+  if (TSDB_SUPER_TABLE == pTableMeta->tableType) {
     return TSDB_CODE_SUCCESS;
+  } else if (TSDB_CHILD_TABLE != pTableMeta->tableType && TSDB_NORMAL_TABLE != pTableMeta->tableType) {
+    return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_ALTER_TABLE);
+  }
+
+  SVAlterTbReq req = {0};
+  code = buildAlterTbReq(pCxt, pStmt, &req);
+
+  SArray* pArray = NULL;
+  if (TSDB_CODE_SUCCESS == code) {
+    code = buildModifyVnodeArray(pCxt, pStmt, &req, &pArray);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = rewriteToVnodeModifyOpStmt(pQuery, pArray);
+  }
+
+  return code;
 }
 
 static int32_t rewriteQuery(STranslateContext* pCxt, SQuery* pQuery) {
@@ -4296,9 +4444,7 @@ static int32_t rewriteQuery(STranslateContext* pCxt, SQuery* pQuery) {
       code = rewriteDropTable(pCxt, pQuery);
       break;
     case QUERY_NODE_ALTER_TABLE_STMT:
-      if (TSDB_ALTER_TABLE_UPDATE_TAG_VAL == ((SAlterTableStmt*)pQuery->pRoot)->alterType) {
       code = rewriteAlterTable(pCxt, pQuery);
-      }
       break;
     default:
       break;

source/libs/parser/src/parUtil.c

@@ -152,6 +152,8 @@ static char* getSyntaxErrFormat(int32_t errCode) {
       return "Invalid timeline function";
     case TSDB_CODE_PAR_INVALID_PASSWD:
       return "Invalid password";
+    case TSDB_CODE_PAR_INVALID_ALTER_TABLE:
+      return "Invalid alter table statement";
     case TSDB_CODE_OUT_OF_MEMORY:
       return "Out of memory";
     default:

source/libs/parser/src/sql.c

@@ -3500,7 +3500,7 @@ static YYACTIONTYPE yy_reduce(
   yymsp[-4].minor.yy172 = yylhsminor.yy172;
         break;
       case 121: /* alter_table_clause ::= full_table_name SET TAG column_name NK_EQ literal */
-{ yylhsminor.yy172 = createAlterTableSetTag(pCxt, yymsp[-5].minor.yy172, &yymsp[-2].minor.yy105, yymsp[0].minor.yy172); }
+{ yylhsminor.yy172 = createAlterTableSetTag(pCxt, yymsp[-5].minor.yy172, &yymsp[-2].minor.yy105, releaseRawExprNode(pCxt, yymsp[0].minor.yy172)); }
   yymsp[-5].minor.yy172 = yylhsminor.yy172;
         break;
       case 123: /* multi_create_clause ::= multi_create_clause create_subtable_clause */

source/libs/parser/test/parInitialATest.cpp

@@ -69,7 +69,7 @@ TEST_F(ParserInitialATest, alterDatabase) {
  *   | COMMENT 'string_value'
  * }
  */
-TEST_F(ParserInitialATest, alterTable) {
+TEST_F(ParserInitialATest, alterSTable) {
   useDb("root", "test");
 
   SMAlterStbReq expect = {0};
@@ -119,7 +119,7 @@ TEST_F(ParserInitialATest, alterTable) {
   setCheckDdlFunc([&](const SQuery* pQuery, ParserStage stage) {
     ASSERT_EQ(nodeType(pQuery->pRoot), QUERY_NODE_ALTER_TABLE_STMT);
     SMAlterStbReq req = {0};
-    ASSERT_TRUE(TSDB_CODE_SUCCESS == tDeserializeSMAlterStbReq(pQuery->pCmdMsg->pMsg, pQuery->pCmdMsg->msgLen, &req));
+    ASSERT_EQ(tDeserializeSMAlterStbReq(pQuery->pCmdMsg->pMsg, pQuery->pCmdMsg->msgLen, &req), TSDB_CODE_SUCCESS);
     ASSERT_EQ(std::string(req.name), std::string(expect.name));
     ASSERT_EQ(req.alterType, expect.alterType);
     ASSERT_EQ(req.numOfFields, expect.numOfFields);
@@ -139,24 +139,24 @@ TEST_F(ParserInitialATest, alterTable) {
     }
   });
 
-  setAlterStbReqFunc("t1", TSDB_ALTER_TABLE_UPDATE_OPTIONS, 0, nullptr, 0, 0, nullptr, nullptr, 10);
-  run("ALTER TABLE t1 TTL 10");
+  setAlterStbReqFunc("st1", TSDB_ALTER_TABLE_UPDATE_OPTIONS, 0, nullptr, 0, 0, nullptr, nullptr, 10);
+  run("ALTER TABLE st1 TTL 10");
 
-  setAlterStbReqFunc("t1", TSDB_ALTER_TABLE_UPDATE_OPTIONS, 0, nullptr, 0, 0, nullptr, "test");
-  run("ALTER TABLE t1 COMMENT 'test'");
+  setAlterStbReqFunc("st1", TSDB_ALTER_TABLE_UPDATE_OPTIONS, 0, nullptr, 0, 0, nullptr, "test");
+  run("ALTER TABLE st1 COMMENT 'test'");
 
-  setAlterStbReqFunc("t1", TSDB_ALTER_TABLE_ADD_COLUMN, 1, "cc1", TSDB_DATA_TYPE_BIGINT);
-  run("ALTER TABLE t1 ADD COLUMN cc1 BIGINT");
+  setAlterStbReqFunc("st1", TSDB_ALTER_TABLE_ADD_COLUMN, 1, "cc1", TSDB_DATA_TYPE_BIGINT);
+  run("ALTER TABLE st1 ADD COLUMN cc1 BIGINT");
 
-  setAlterStbReqFunc("t1", TSDB_ALTER_TABLE_DROP_COLUMN, 1, "c1");
-  run("ALTER TABLE t1 DROP COLUMN c1");
+  setAlterStbReqFunc("st1", TSDB_ALTER_TABLE_DROP_COLUMN, 1, "c1");
+  run("ALTER TABLE st1 DROP COLUMN c1");
 
-  setAlterStbReqFunc("t1", TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES, 1, "c1", TSDB_DATA_TYPE_VARCHAR,
+  setAlterStbReqFunc("st1", TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES, 1, "c1", TSDB_DATA_TYPE_VARCHAR,
                      20 + VARSTR_HEADER_SIZE);
-  run("ALTER TABLE t1 MODIFY COLUMN c1 VARCHAR(20)");
+  run("ALTER TABLE st1 MODIFY COLUMN c1 VARCHAR(20)");
 
-  setAlterStbReqFunc("t1", TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME, 2, "c1", 0, 0, "cc1");
-  run("ALTER TABLE t1 RENAME COLUMN c1 cc1");
+  setAlterStbReqFunc("st1", TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME, 2, "c1", 0, 0, "cc1");
+  run("ALTER TABLE st1 RENAME COLUMN c1 cc1");
 
   setAlterStbReqFunc("st1", TSDB_ALTER_TABLE_ADD_TAG, 1, "tag11", TSDB_DATA_TYPE_BIGINT);
   run("ALTER TABLE st1 ADD TAG tag11 BIGINT");
@@ -171,7 +171,127 @@ TEST_F(ParserInitialATest, alterTable) {
   setAlterStbReqFunc("st1", TSDB_ALTER_TABLE_UPDATE_TAG_NAME, 2, "tag1", 0, 0, "tag11");
   run("ALTER TABLE st1 RENAME TAG tag1 tag11");
 
-  // run("ALTER TABLE st1s1 SET TAG tag1=10");
+  // todo
+  // ADD {FULLTEXT | SMA} INDEX index_name (col_name [, col_name] ...) [index_option]
+}
+
+TEST_F(ParserInitialATest, alterTable) {
+  useDb("root", "test");
+
+  SVAlterTbReq expect = {0};
+
+  auto setAlterColFunc = [&](const char* pTbname, int8_t alterType, const char* pColName, int8_t dataType = 0,
+                             int32_t dataBytes = 0, const char* pNewColName = nullptr) {
+    memset(&expect, 0, sizeof(SVAlterTbReq));
+    expect.tbName = strdup(pTbname);
+    expect.action = alterType;
+    expect.colName = strdup(pColName);
+
+    switch (alterType) {
+      case TSDB_ALTER_TABLE_ADD_COLUMN:
+        expect.type = dataType;
+        expect.flags = COL_SMA_ON;
+        expect.bytes = dataBytes > 0 ? dataBytes : (dataType > 0 ? tDataTypes[dataType].bytes : 0);
+        break;
+      case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES:
+        expect.colModBytes = dataBytes;
+        break;
+      case TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME:
+        expect.colNewName = strdup(pNewColName);
+        break;
+      default:
+        break;
+    }
+  };
+
+  auto setAlterTagFunc = [&](const char* pTbname, const char* pTagName, const uint8_t* pNewVal, uint32_t bytes) {
+    memset(&expect, 0, sizeof(SVAlterTbReq));
+    expect.tbName = strdup(pTbname);
+    expect.action = TSDB_ALTER_TABLE_UPDATE_TAG_VAL;
+    expect.tagName = strdup(pTagName);
+
+    expect.isNull = (nullptr == pNewVal);
+    expect.nTagVal = bytes;
+    expect.pTagVal = pNewVal;
+  };
+
+  auto setAlterOptionsFunc = [&](const char* pTbname, int32_t ttl, const char* pComment = nullptr) {
+    memset(&expect, 0, sizeof(SVAlterTbReq));
+    expect.tbName = strdup(pTbname);
+    expect.action = TSDB_ALTER_TABLE_UPDATE_OPTIONS;
+    if (-1 != ttl) {
+      expect.updateTTL = true;
+      expect.newTTL = ttl;
+    }
+    if (nullptr != pComment) {
+      expect.updateComment = true;
+      expect.newComment = pComment;
+    }
+  };
+
+  setCheckDdlFunc([&](const SQuery* pQuery, ParserStage stage) {
+    ASSERT_EQ(nodeType(pQuery->pRoot), QUERY_NODE_VNODE_MODIF_STMT);
+    SVnodeModifOpStmt* pStmt = (SVnodeModifOpStmt*)pQuery->pRoot;
+
+    ASSERT_EQ(pStmt->sqlNodeType, QUERY_NODE_ALTER_TABLE_STMT);
+    ASSERT_NE(pStmt->pDataBlocks, nullptr);
+    ASSERT_EQ(taosArrayGetSize(pStmt->pDataBlocks), 1);
+    SVgDataBlocks* pVgData = (SVgDataBlocks*)taosArrayGetP(pStmt->pDataBlocks, 0);
+    void*          pBuf = POINTER_SHIFT(pVgData->pData, sizeof(SMsgHead));
+    SVAlterTbReq   req = {0};
+    SDecoder       coder = {0};
+    tDecoderInit(&coder, (const uint8_t*)pBuf, pVgData->size);
+    ASSERT_EQ(tDecodeSVAlterTbReq(&coder, &req), TSDB_CODE_SUCCESS);
+
+    ASSERT_EQ(std::string(req.tbName), std::string(expect.tbName));
+    ASSERT_EQ(req.action, expect.action);
+    if (nullptr != expect.colName) {
+      ASSERT_EQ(std::string(req.colName), std::string(expect.colName));
+    }
+    ASSERT_EQ(req.type, expect.type);
+    ASSERT_EQ(req.flags, expect.flags);
+    ASSERT_EQ(req.bytes, expect.bytes);
+    ASSERT_EQ(req.colModBytes, expect.colModBytes);
+    if (nullptr != expect.colNewName) {
+      ASSERT_EQ(std::string(req.colNewName), std::string(expect.colNewName));
+    }
+    if (nullptr != expect.tagName) {
+      ASSERT_EQ(std::string(req.tagName), std::string(expect.tagName));
+    }
+    ASSERT_EQ(req.isNull, expect.isNull);
+    ASSERT_EQ(req.nTagVal, expect.nTagVal);
+    ASSERT_EQ(memcmp(req.pTagVal, expect.pTagVal, expect.nTagVal), 0);
+    ASSERT_EQ(req.updateTTL, expect.updateTTL);
+    ASSERT_EQ(req.newTTL, expect.newTTL);
+    ASSERT_EQ(req.updateComment, expect.updateComment);
+    if (nullptr != expect.newComment) {
+      ASSERT_EQ(std::string(req.newComment), std::string(expect.newComment));
+    }
+
+    tDecoderClear(&coder);
+  });
+
+  setAlterOptionsFunc("t1", 10, nullptr);
+  run("ALTER TABLE t1 TTL 10");
+
+  setAlterOptionsFunc("t1", -1, "test");
+  run("ALTER TABLE t1 COMMENT 'test'");
+
+  setAlterColFunc("t1", TSDB_ALTER_TABLE_ADD_COLUMN, "cc1", TSDB_DATA_TYPE_BIGINT);
+  run("ALTER TABLE t1 ADD COLUMN cc1 BIGINT");
+
+  setAlterColFunc("t1", TSDB_ALTER_TABLE_DROP_COLUMN, "c1");
+  run("ALTER TABLE t1 DROP COLUMN c1");
+
+  setAlterColFunc("t1", TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES, "c1", TSDB_DATA_TYPE_VARCHAR, 20 + VARSTR_HEADER_SIZE);
+  run("ALTER TABLE t1 MODIFY COLUMN c1 VARCHAR(20)");
+
+  setAlterColFunc("t1", TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME, "c1", 0, 0, "cc1");
+  run("ALTER TABLE t1 RENAME COLUMN c1 cc1");
+
+  int64_t val = 10;
+  setAlterTagFunc("st1s1", "tag1", (const uint8_t*)&val, sizeof(val));
+  run("ALTER TABLE st1s1 SET TAG tag1=10");
 
   // todo
   // ADD {FULLTEXT | SMA} INDEX index_name (col_name [, col_name] ...) [index_option]

source/libs/planner/src/planLogicCreater.c

@@ -985,6 +985,8 @@ static int32_t getMsgType(ENodeType sqlType) {
       return TDMT_VND_CREATE_TABLE;
     case QUERY_NODE_DROP_TABLE_STMT:
       return TDMT_VND_DROP_TABLE;
+    case QUERY_NODE_ALTER_TABLE_STMT:
+      return TDMT_VND_ALTER_TABLE;
     default:
       break;
   }

source/libs/scheduler/src/scheduler.c

@@ -256,6 +256,7 @@ int32_t schValidateTaskReceivedMsgType(SSchJob *pJob, SSchTask *pTask, int32_t m
       return TSDB_CODE_SUCCESS;
     case TDMT_VND_CREATE_TABLE_RSP:
     case TDMT_VND_DROP_TABLE_RSP:
+    case TDMT_VND_ALTER_TABLE_RSP:
     case TDMT_VND_SUBMIT_RSP:
       break;
     default:
@@ -1131,6 +1132,24 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
       SCH_ERR_RET(schProcessOnTaskSuccess(pJob, pTask));
       break;
     }
+    case TDMT_VND_ALTER_TABLE_RSP: {
+      SVAlterTbRsp rsp = {0};
+      if (msg) {
+        SDecoder coder = {0};
+        tDecoderInit(&coder, msg, msgSize);
+        code = tDecodeSVAlterTbRsp(&coder, &rsp);
+        tDecoderClear(&coder);
+        SCH_ERR_JRET(code);
+        SCH_ERR_JRET(rsp.code);
+      }
+
+      SCH_ERR_JRET(rspCode);
+
+      if (NULL == msg) {
+        SCH_ERR_JRET(TSDB_CODE_QRY_INVALID_INPUT);
+      }
+      break;
+    }
     case TDMT_VND_SUBMIT_RSP: {
       SCH_ERR_JRET(rspCode);
 
@@ -1391,6 +1410,10 @@ int32_t schHandleDropTableCallback(void *param, const SDataBuf *pMsg, int32_t co
   return schHandleCallback(param, pMsg, TDMT_VND_DROP_TABLE_RSP, code);
 }
 
+int32_t schHandleAlterTableCallback(void *param, const SDataBuf *pMsg, int32_t code) {
+  return schHandleCallback(param, pMsg, TDMT_VND_ALTER_TABLE_RSP, code);
+}
+
 int32_t schHandleQueryCallback(void *param, const SDataBuf *pMsg, int32_t code) {
   return schHandleCallback(param, pMsg, TDMT_VND_QUERY_RSP, code);
 }
@@ -1490,6 +1513,9 @@ int32_t schGetCallbackFp(int32_t msgType, __async_send_cb_fn_t *fp) {
     case TDMT_VND_DROP_TABLE:
       *fp = schHandleDropTableCallback;
       break;
+    case TDMT_VND_ALTER_TABLE:
+      *fp = schHandleAlterTableCallback;
+      break;
     case TDMT_VND_SUBMIT:
       *fp = schHandleSubmitCallback;
       break;
@@ -2010,6 +2036,7 @@ int32_t schBuildAndSendMsg(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *addr,
   switch (msgType) {
     case TDMT_VND_CREATE_TABLE:
     case TDMT_VND_DROP_TABLE:
+    case TDMT_VND_ALTER_TABLE:
     case TDMT_VND_SUBMIT: {
       msgSize = pTask->msgLen;
       msg = taosMemoryCalloc(1, msgSize);

tests/script/tsim/tmq/basic1.sim

@@ -27,6 +27,7 @@ $tstart     = 1640966400000  # 2022-01-01 00:00:00.000
 
 $pullDelay    = 3
 $ifcheckdata  = 1
+$ifmanualcommit = 1
 $showMsg      = 1
 $showRow      = 0
 
@@ -53,8 +54,17 @@ sql create topic topic_ntb_function as select ts, abs(c1), sin(c2) from ntb0
 #  return -1
 #endi
 
+#'group.id:cgrp1,enable.auto.commit:false,auto.commit.interval.ms:6000,auto.offset.reset:earliest'
 $keyList = ' . group.id:cgrp1
+$keyList = $keyList . ,
+$keyList = $keyList . enable.auto.commit:false
+#$keyList = $keyList . ,
+#$keyList = $keyList . auto.commit.interval.ms:6000
+#$keyList = $keyList . ,
+#$keyList = $keyList . auto.offset.reset:earliest
 $keyList = $keyList . '
+print ========== key list:  $keyList
+
 
 $cdb_index = 0
 #=============================== start consume =============================#
@@ -74,7 +84,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -102,7 +112,7 @@ endi
 $consumerId    = 0
 $totalMsgOfStb = $ctbNum * $rowsPerCtb
 $expectmsgcnt  = $totalMsgOfStb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from stb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
@@ -145,7 +155,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -173,7 +183,7 @@ endi
 $consumerId    = 0
 $totalMsgOfCtb = $rowsPerCtb
 $expectmsgcnt  = $totalMsgOfCtb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from ctb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
@@ -216,7 +226,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -244,7 +254,7 @@ endi
 $consumerId    = 0
 $totalMsgOfNtb = $rowsPerCtb
 $expectmsgcnt  = $totalMsgOfNtb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from ntb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start

tests/script/tsim/tmq/basic1Of2Cons.sim

@@ -27,6 +27,7 @@ $tstart     = 1640966400000  # 2022-01-01 00:00:00.000
 
 $pullDelay    = 5
 $ifcheckdata  = 1
+$ifmanualcommit = 1
 $showMsg      = 1
 $showRow      = 0
 
@@ -53,10 +54,19 @@ sql create topic topic_ntb_function as select ts, abs(c1), sin(c2) from ntb0
 #  return -1
 #endi
 
+#'group.id:cgrp1,enable.auto.commit:false,auto.commit.interval.ms:6000,auto.offset.reset:earliest'
 $keyList = ' . group.id:cgrp1
+$keyList = $keyList . ,
+$keyList = $keyList . enable.auto.commit:false
+#$keyList = $keyList . ,
+#$keyList = $keyList . auto.commit.interval.ms:6000
+#$keyList = $keyList . ,
+#$keyList = $keyList . auto.offset.reset:earliest
 $keyList = $keyList . '
+print ========== key list:  $keyList
 
 $cdb_index = 0
+
 #=============================== start consume =============================#
 
 print ================ test consume from stb
@@ -74,7 +84,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table for stb
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -102,10 +112,10 @@ endi
 $consumerId    = 0
 $totalMsgOfStb = $ctbNum * $rowsPerCtb
 $expectmsgcnt  = $totalMsgOfStb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 $consumerId    = 1
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from stb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
@@ -177,7 +187,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table for ctb
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -205,9 +215,9 @@ endi
 $consumerId    = 0
 $totalMsgOfCtb = $rowsPerCtb
 $expectmsgcnt  = $totalMsgOfCtb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 $consumerId    = 1
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from ctb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
@@ -279,7 +289,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table for ntb
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -307,9 +317,9 @@ endi
 $consumerId    = 0
 $totalMsgOfNtb = $rowsPerCtb
 $expectmsgcnt  = $totalMsgOfNtb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 $consumerId    = 1
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from ntb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start

tests/script/tsim/tmq/basic2.sim

@@ -27,6 +27,7 @@ $tstart     = 1640966400000  # 2022-01-01 00:00:00.000
 
 $pullDelay    = 3
 $ifcheckdata  = 1
+$ifmanualcommit = 1
 $showMsg      = 1
 $showRow      = 0
 
@@ -53,8 +54,17 @@ sql create topic topic_ntb_function as select ts, abs(c1), sin(c2) from ntb0
 #  return -1
 #endi
 
+#'group.id:cgrp1,enable.auto.commit:false,auto.commit.interval.ms:6000,auto.offset.reset:earliest'
 $keyList = ' . group.id:cgrp1
+$keyList = $keyList . ,
+$keyList = $keyList . enable.auto.commit:false
+#$keyList = $keyList . ,
+#$keyList = $keyList . auto.commit.interval.ms:6000
+#$keyList = $keyList . ,
+#$keyList = $keyList . auto.offset.reset:earliest
 $keyList = $keyList . '
+print ========== key list:  $keyList
+
 
 $topicNum = 3
 
@@ -74,7 +84,7 @@ $consumerId    = 0
 $totalMsgOfStb = $ctbNum * $rowsPerCtb
 $totalMsgOfStb = $totalMsgOfStb * $topicNum
 $expectmsgcnt  = $totalMsgOfStb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from stb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $dbName -s start
@@ -109,7 +119,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -131,7 +141,7 @@ $topicList = $topicList . '
 $consumerId    = 0
 $totalMsgOfCtb = $rowsPerCtb * $topicNum
 $expectmsgcnt  = $totalMsgOfCtb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from ctb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
@@ -166,7 +176,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -188,7 +198,7 @@ $topicList = $topicList . '
 $consumerId    = 0
 $totalMsgOfNtb = $rowsPerCtb * $topicNum
 $expectmsgcnt  = $totalMsgOfNtb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from ntb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start

tests/script/tsim/tmq/basic2Of2Cons.sim

@@ -27,6 +27,7 @@ $tstart     = 1640966400000  # 2022-01-01 00:00:00.000
 
 $pullDelay    = 5
 $ifcheckdata  = 1
+$ifmanualcommit = 1
 $showMsg      = 1
 $showRow      = 0
 
@@ -53,8 +54,16 @@ sql create topic topic_ntb_function as select ts, abs(c1), sin(c2) from ntb0
 #  return -1
 #endi
 
+#'group.id:cgrp1,enable.auto.commit:false,auto.commit.interval.ms:6000,auto.offset.reset:earliest'
 $keyList = ' . group.id:cgrp1
+$keyList = $keyList . ,
+$keyList = $keyList . enable.auto.commit:false
+#$keyList = $keyList . ,
+#$keyList = $keyList . auto.commit.interval.ms:6000
+#$keyList = $keyList . ,
+#$keyList = $keyList . auto.offset.reset:earliest
 $keyList = $keyList . '
+print ========== key list:  $keyList
 
 $topicNum = 3
 
@@ -74,9 +83,9 @@ $consumerId    = 0
 $totalMsgOfStb = $ctbNum * $rowsPerCtb
 $totalMsgOfStb = $totalMsgOfStb * $topicNum
 $expectmsgcnt  = $totalMsgOfStb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 $consumerId    = 1
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from stb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $dbName -s start
@@ -139,7 +148,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table for ctb
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -161,9 +170,9 @@ $topicList = $topicList . '
 $consumerId    = 0
 $totalMsgOfCtb = $rowsPerCtb * $topicNum
 $expectmsgcnt  = $totalMsgOfCtb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 $consumerId    = 1
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from ctb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
@@ -226,7 +235,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table for ntb
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -248,9 +257,9 @@ $topicList = $topicList . '
 $consumerId    = 0
 $totalMsgOfNtb = $rowsPerCtb * $topicNum
 $expectmsgcnt  = $totalMsgOfNtb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 $consumerId    = 1
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from ntb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start

tests/script/tsim/tmq/basic2Of2ConsOverlap.sim

@@ -27,6 +27,7 @@ $tstart     = 1640966400000  # 2022-01-01 00:00:00.000
 
 $pullDelay    = 5
 $ifcheckdata  = 1
+$ifmanualcommit = 1
 $showMsg      = 1
 $showRow      = 0
 
@@ -53,8 +54,16 @@ sql create topic topic_ntb_function as select ts, abs(c1), sin(c2) from ntb0
 #  return -1
 #endi
 
+#'group.id:cgrp1,enable.auto.commit:false,auto.commit.interval.ms:6000,auto.offset.reset:earliest'
 $keyList = ' . group.id:cgrp1
+$keyList = $keyList . ,
+$keyList = $keyList . enable.auto.commit:false
+#$keyList = $keyList . ,
+#$keyList = $keyList . auto.commit.interval.ms:6000
+#$keyList = $keyList . ,
+#$keyList = $keyList . auto.offset.reset:earliest
 $keyList = $keyList . '
+print ========== key list:  $keyList
 
 $topicNum = 2
 
@@ -72,7 +81,7 @@ $consumerId    = 0
 $totalMsgOfOneTopic = $ctbNum * $rowsPerCtb
 $totalMsgOfStb = $totalMsgOfOneTopic * $topicNum
 $expectmsgcnt  = $totalMsgOfStb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 
 $topicList = ' . topic_stb_all
@@ -80,7 +89,7 @@ $topicList = $topicList . ,
 $topicList = $topicList . topic_stb_function
 $topicList = $topicList . '
 $consumerId    = 1
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from stb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $dbName -s start
@@ -158,7 +167,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table for ctb
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -179,14 +188,14 @@ $consumerId    = 0
 $totalMsgOfOneTopic = $rowsPerCtb
 $totalMsgOfCtb = $totalMsgOfOneTopic * $topicNum
 $expectmsgcnt  = $totalMsgOfCtb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 $topicList = ' . topic_ctb_function
 $topicList = $topicList . ,
 $topicList = $topicList . topic_ctb_all
 $topicList = $topicList . '
 $consumerId    = 1
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from ctb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
@@ -249,7 +258,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table for ntb
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -270,7 +279,7 @@ $consumerId    = 0
 $totalMsgOfOneTopic = $rowsPerCtb
 $totalMsgOfNtb = $totalMsgOfOneTopic * $topicNum
 $expectmsgcnt  = $totalMsgOfNtb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 
 $topicList = ' . topic_ntb_function
@@ -278,7 +287,7 @@ $topicList = $topicList . ,
 $topicList = $topicList . topic_ntb_all
 $topicList = $topicList . '
 $consumerId    = 1
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from ntb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start

tests/script/tsim/tmq/basic3.sim

@@ -27,6 +27,7 @@ $tstart     = 1640966400000  # 2022-01-01 00:00:00.000
 
 $pullDelay    = 3
 $ifcheckdata  = 1
+$ifmanualcommit = 1
 $showMsg      = 1
 $showRow      = 0
 
@@ -53,8 +54,17 @@ sql create topic topic_ntb_function as select ts, abs(c1), sin(c2) from ntb0
 #  return -1
 #endi
 
+#'group.id:cgrp1,enable.auto.commit:false,auto.commit.interval.ms:6000,auto.offset.reset:earliest'
 $keyList = ' . group.id:cgrp1
+$keyList = $keyList . ,
+$keyList = $keyList . enable.auto.commit:false
+#$keyList = $keyList . ,
+#$keyList = $keyList . auto.commit.interval.ms:6000
+#$keyList = $keyList . ,
+#$keyList = $keyList . auto.offset.reset:earliest
 $keyList = $keyList . '
+print ========== key list:  $keyList
+
 
 $cdb_index = 0
 #=============================== start consume =============================#
@@ -74,7 +84,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -102,7 +112,7 @@ endi
 $consumerId    = 0
 $totalMsgOfStb = $ctbNum * $rowsPerCtb
 $expectmsgcnt  = $totalMsgOfStb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from stb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
@@ -145,7 +155,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -173,7 +183,7 @@ endi
 $consumerId    = 0
 $totalMsgOfCtb = $rowsPerCtb
 $expectmsgcnt  = $totalMsgOfCtb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from ctb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
@@ -216,7 +226,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -244,7 +254,7 @@ endi
 $consumerId    = 0
 $totalMsgOfNtb = $rowsPerCtb
 $expectmsgcnt  = $totalMsgOfNtb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from ntb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start

tests/script/tsim/tmq/basic3Of2Cons.sim

@@ -27,6 +27,7 @@ $tstart     = 1640966400000  # 2022-01-01 00:00:00.000
 
 $pullDelay    = 5
 $ifcheckdata  = 1
+$ifmanualcommit = 1
 $showMsg      = 1
 $showRow      = 0
 
@@ -53,8 +54,16 @@ sql create topic topic_ntb_function as select ts, abs(c1), sin(c2) from ntb0
 #  return -1
 #endi
 
+#'group.id:cgrp1,enable.auto.commit:false,auto.commit.interval.ms:6000,auto.offset.reset:earliest'
 $keyList = ' . group.id:cgrp1
+$keyList = $keyList . ,
+$keyList = $keyList . enable.auto.commit:false
+#$keyList = $keyList . ,
+#$keyList = $keyList . auto.commit.interval.ms:6000
+#$keyList = $keyList . ,
+#$keyList = $keyList . auto.offset.reset:earliest
 $keyList = $keyList . '
+print ========== key list:  $keyList
 
 $cdb_index = 0
 #=============================== start consume =============================#
@@ -74,7 +83,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -102,9 +111,9 @@ endi
 $consumerId    = 0
 $totalMsgOfStb = $ctbNum * $rowsPerCtb
 $expectmsgcnt  = $totalMsgOfStb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 $consumerId    = 1
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from stb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
@@ -189,7 +198,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -217,9 +226,9 @@ endi
 $consumerId    = 0
 $totalMsgOfCtb = $rowsPerCtb
 $expectmsgcnt  = $totalMsgOfCtb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 $consumerId    = 1
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from ctb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
@@ -291,7 +300,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -319,9 +328,9 @@ endi
 $consumerId    = 0
 $totalMsgOfNtb = $rowsPerCtb
 $expectmsgcnt  = $totalMsgOfNtb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 $consumerId    = 1
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from ntb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start

tests/script/tsim/tmq/basic4.sim

@@ -27,6 +27,7 @@ $tstart     = 1640966400000  # 2022-01-01 00:00:00.000
 
 $pullDelay    = 3
 $ifcheckdata  = 1
+$ifmanualcommit = 1
 $showMsg      = 1
 $showRow      = 0
 
@@ -53,8 +54,17 @@ sql create topic topic_ntb_function as select ts, abs(c1), sin(c2) from ntb0
 #  return -1
 #endi
 
+#'group.id:cgrp1,enable.auto.commit:false,auto.commit.interval.ms:6000,auto.offset.reset:earliest'
 $keyList = ' . group.id:cgrp1
+$keyList = $keyList . ,
+$keyList = $keyList . enable.auto.commit:false
+#$keyList = $keyList . ,
+#$keyList = $keyList . auto.commit.interval.ms:6000
+#$keyList = $keyList . ,
+#$keyList = $keyList . auto.offset.reset:earliest
 $keyList = $keyList . '
+print ========== key list:  $keyList
+
 
 $topicNum = 3
 
@@ -71,7 +81,7 @@ $consumerId    = 0
 $totalMsgOfStb = $ctbNum * $rowsPerCtb
 $totalMsgOfStb = $totalMsgOfStb * $topicNum
 $expectmsgcnt  = $totalMsgOfStb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from stb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $dbName -s start
@@ -106,7 +116,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -128,7 +138,7 @@ $topicList = $topicList . '
 $consumerId    = 0
 $totalMsgOfCtb = $rowsPerCtb * $topicNum
 $expectmsgcnt  = $totalMsgOfCtb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from ctb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
@@ -163,7 +173,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -185,7 +195,7 @@ $topicList = $topicList . '
 $consumerId    = 0
 $totalMsgOfNtb = $rowsPerCtb * $topicNum
 $expectmsgcnt  = $totalMsgOfNtb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from ntb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start

tests/script/tsim/tmq/basic4Of2Cons.sim

@@ -27,6 +27,7 @@ $tstart     = 1640966400000  # 2022-01-01 00:00:00.000
 
 $pullDelay    = 5
 $ifcheckdata  = 1
+$ifmanualcommit = 1
 $showMsg      = 1
 $showRow      = 0
 
@@ -53,8 +54,16 @@ sql create topic topic_ntb_function as select ts, abs(c1), sin(c2) from ntb0
 #  return -1
 #endi
 
+#'group.id:cgrp1,enable.auto.commit:false,auto.commit.interval.ms:6000,auto.offset.reset:earliest'
 $keyList = ' . group.id:cgrp1
+$keyList = $keyList . ,
+$keyList = $keyList . enable.auto.commit:false
+#$keyList = $keyList . ,
+#$keyList = $keyList . auto.commit.interval.ms:6000
+#$keyList = $keyList . ,
+#$keyList = $keyList . auto.offset.reset:earliest
 $keyList = $keyList . '
+print ========== key list:  $keyList
 
 $topicNum = 3
 
@@ -71,9 +80,9 @@ $consumerId    = 0
 $totalMsgOfStb = $ctbNum * $rowsPerCtb
 $totalMsgOfStb = $totalMsgOfStb * $topicNum
 $expectmsgcnt  = $totalMsgOfStb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 $consumerId    = 1
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from stb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $dbName -s start
@@ -148,7 +157,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -170,9 +179,9 @@ $topicList = $topicList . '
 $consumerId    = 0
 $totalMsgOfCtb = $rowsPerCtb * $topicNum
 $expectmsgcnt  = $totalMsgOfCtb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 $consumerId    = 1
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from ctb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
@@ -236,7 +245,7 @@ sleep 500
 sql use $cdbName
 
 print == create consume info table and consume result table
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables
@@ -258,9 +267,9 @@ $topicList = $topicList . '
 $consumerId    = 0
 $totalMsgOfNtb = $rowsPerCtb * $topicNum
 $expectmsgcnt  = $totalMsgOfNtb
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 $consumerId    = 1
-sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata )
+sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
 
 print == start consumer to pull msgs from ntb
 print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start

tests/script/tsim/tmq/prepareBasicEnv-1vgrp.sim

@@ -48,7 +48,7 @@ endi
 sql use $dbName
 
 print == create consume info table and consume result table
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables

tests/script/tsim/tmq/prepareBasicEnv-4vgrp.sim

@@ -48,7 +48,7 @@ endi
 sql use $dbName
 
 print == create consume info table and consume result table
-sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)
+sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
 sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
 
 sql show tables

tests/system-test/2-query/query_cols_tags_and_or.py

@@ -22,18 +22,6 @@ class TDTestCase:
         tdSql.init(conn.cursor(), logSql)
 
     def insertData(self, tb_name):
-        # insert_sql_list = [f'insert into {tb_name} values ("2021-01-01 12:00:00", 1, 1, 1, 3, 1.1, 1.1, "binary", "nchar", true, 1)',
-        #                    f'insert into {tb_name} values ("2021-01-05 12:00:00", 2, 2, 1, 3, 1.1, 1.1, "binary", "nchar", true, 2)',
-        #                    f'insert into {tb_name} values ("2021-01-07 12:00:00", 1, 3, 1, 2, 1.1, 1.1, "binary", "nchar", true, 3)',
-        #                    f'insert into {tb_name} values ("2021-01-09 12:00:00", 1, 2, 4, 3, 1.1, 1.1, "binary", "nchar", true, 4)',
-        #                    f'insert into {tb_name} values ("2021-01-11 12:00:00", 1, 2, 5, 5, 1.1, 1.1, "binary", "nchar", true, 5)',
-        #                    f'insert into {tb_name} values ("2021-01-13 12:00:00", 1, 2, 1, 3, 6.6, 1.1, "binary", "nchar", true, 6)',
-        #                    f'insert into {tb_name} values ("2021-01-15 12:00:00", 1, 2, 1, 3, 1.1, 7.7, "binary", "nchar", true, 7)',
-        #                    f'insert into {tb_name} values ("2021-01-17 12:00:00", 1, 2, 1, 3, 1.1, 1.1, "binary8", "nchar", true, 8)',
-        #                    f'insert into {tb_name} values ("2021-01-19 12:00:00", 1, 2, 1, 3, 1.1, 1.1, "binary", "nchar9", true, 9)',
-        #                    f'insert into {tb_name} values ("2021-01-21 12:00:00", 1, 2, 1, 3, 1.1, 1.1, "binary", "nchar", false, 10)',
-        #                    f'insert into {tb_name} values ("2021-01-23 12:00:00", 1, 3, 1, 3, 1.1, 1.1, Null, Null, false, 11)'
-        #                    ]
         insert_sql_list = [f'insert into {tb_name} values ("2021-01-01 12:00:00", 1, 1, 1, 3, 1.1, 1.1, "binary", "nchar", true, 1, 2, 3, 4)',
                            f'insert into {tb_name} values ("2021-01-05 12:00:00", 2, 2, 1, 3, 1.1, 1.1, "binary", "nchar", true, 2, 3, 4, 5)',
                            f'insert into {tb_name} values ("2021-01-07 12:00:00", 1, 3, 1, 2, 1.1, 1.1, "binary", "nchar", true, 3, 4, 5, 6)',
@@ -54,7 +42,6 @@ class TDTestCase:
         tb_name = tdCom.getLongName(8, "letters")
         tdSql.execute(
             f"CREATE TABLE {tb_name} (ts timestamp, c1 tinyint, c2 smallint, c3 int, c4 bigint, c5 float, c6 double, c7 binary(100), c8 nchar(200), c9 bool, c10 tinyint unsigned, c11 smallint unsigned, c12 int unsigned, c13 bigint unsigned)")
-            # f"CREATE TABLE {tb_name} (ts timestamp, c1 tinyint, c2 smallint, c3 int, c4 bigint, c5 float, c6 double, c7 binary(100), c8 nchar(200), c9 bool, c10 int)")
         self.insertData(tb_name)
         return tb_name
 
@@ -95,6 +82,31 @@ class TDTestCase:
 
     def queryTsCol(self, tb_name, check_elm=None):
         select_elm = "*" if check_elm is None else check_elm
+        # ts in
+        query_sql = f'select {select_elm} from {tb_name} where ts in ("2021-01-11 12:00:00", "2021-01-13 12:00:00")'
+        tdSql.query(query_sql)
+        tdSql.checkRows(2)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 6) if select_elm == "*" else False
+        # ts not in
+        query_sql = f'select {select_elm} from {tb_name} where ts not in ("2021-01-11 12:00:00", "2021-01-13 12:00:00")'
+        tdSql.query(query_sql)
+        tdSql.checkRows(9)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 11) if select_elm == "*" else False
+        # ts not null
+        query_sql = f'select {select_elm} from {tb_name} where ts is not Null'
+        tdSql.query(query_sql)
+        tdSql.checkRows(11)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 11) if select_elm == "*" else False
+        # ts null
+        query_sql = f'select {select_elm} from {tb_name} where ts is Null'
+        tdSql.query(query_sql)
+        tdSql.checkRows(0)
+        # not support like not like match nmatch
+        tdSql.error(f'select {select_elm} from {tb_name} where ts like ("2021-01-11 12:00:00%")')
+        tdSql.error(f'select {select_elm} from {tb_name} where ts not like ("2021-01-11 12:00:0_")')
+        tdSql.error(f'select {select_elm} from {tb_name} where ts match "2021-01-11 12:00:00%"')
+        tdSql.error(f'select {select_elm} from {tb_name} where ts nmatch "2021-01-11 12:00:00%"')
+
         # ts and ts
         query_sql = f'select {select_elm} from {tb_name} where ts > "2021-01-11 12:00:00" or ts < "2021-01-13 12:00:00"'
         tdSql.query(query_sql)
@@ -1422,9 +1434,9 @@ class TDTestCase:
         tdSql.query(query_sql)
         tdSql.checkRows(11)
         tdSql.checkEqual(self.queryLastC10(query_sql), 11) if select_elm == "*" else False
-        query_sql = f'select c1,c2,c3,c4,c5,c6,c7,c8,c9,c10,c11,c12,c13 from {tb_name} where c9 > "binary" and c9 >= "binary8" or c9 < "binary9" and c9 <= "binary" and c9 != 2 and c9 <> 2 and c9 = 4 or c9 is not null and c9 between 2 and 4 and c9 not between 1 and 2 and c9 in (2,4) and c9 not in (1,2) or c9 match "binary[28]" or c9 nmatch "binary"'
+        query_sql = f'select c1,c2,c3,c4,c5,c6,c7,c8,c9,c10,c11,c12,c13 from {tb_name} where c9 > "binary" and c9 >= "binary8" or c9 < "binary9" and c9 <= "binary" and c9 != 2 and c9 <> 2 and c9 = 4 or c9 is not null and c9 between 2 and 4 and c9 not between 1 and 2 and c9 in (2,4) and c9 not in (1,2)'
         tdSql.query(query_sql)
-        tdSql.checkRows(11)
+        tdSql.checkRows(9)
 
     def queryFullColType(self, tb_name, check_elm=None):
         select_elm = "*" if check_elm is None else check_elm

tests/system-test/2-query/union.py

+import datetime
+
+from util.log import *
+from util.sql import *
+from util.cases import *
+from util.dnodes import *
+
+PRIMARY_COL = "ts"
+
+INT_COL     = "c1"
+BINT_COL    = "c2"
+SINT_COL    = "c3"
+TINT_COL    = "c4"
+FLOAT_COL   = "c5"
+DOUBLE_COL  = "c6"
+BOOL_COL    = "c7"
+
+BINARY_COL  = "c8"
+NCHAR_COL   = "c9"
+TS_COL      = "c10"
+
+NUM_COL     = [ INT_COL, BINT_COL, SINT_COL, TINT_COL, FLOAT_COL, DOUBLE_COL, ]
+CHAR_COL    = [ BINARY_COL, NCHAR_COL, ]
+BOOLEAN_COL = [ BOOL_COL, ]
+TS_TYPE_COL = [ TS_COL, ]
+
+class TDTestCase:
+
+    def init(self, conn, logSql):
+        tdLog.debug(f"start to excute {__file__}")
+        tdSql.init(conn.cursor())
+
+    def __query_condition(self,tbname):
+        query_condition = []
+        for char_col in CHAR_COL:
+            query_condition.extend(
+                (
+                    f"{tbname}.{char_col}",
+                    f"upper( {tbname}.{char_col} )",
+                    f"char_length( {tbname}.{char_col} )",
+                    f"concat( {tbname}.{char_col}, {tbname}.{char_col} )",
+                    f"concat_ws( '_', {tbname}.{char_col}, {tbname}.{char_col} )",
+                    f"length( {tbname}.{char_col} )",
+                    f"lower( {tbname}.{char_col} )",
+                    f"ltrim( {tbname}.{char_col} )",
+                    f"rtrim( {tbname}.{char_col} )",
+                    f"substr( {tbname}.{char_col}, 1 )",
+                    f"count( {tbname}.{char_col} )",
+                    f"cast( {tbname}.{char_col} as nchar(3) )",
+                    f"cast( {tbname}.{char_col} as nchar(8) )",
+                )
+            )
+            query_condition.extend( f"cast( {tbname}.{un_char_col} as binary(16) ) " for un_char_col in NUM_COL)
+            query_condition.extend( f"cast( {tbname}.{char_col} + {tbname}.{char_col_2} as binary(32) ) " for char_col_2 in CHAR_COL )
+            query_condition.extend( f"cast( {tbname}.{char_col} + {tbname}.{un_char_col} as binary(32) ) " for un_char_col in NUM_COL )
+
+        for num_col in NUM_COL:
+            query_condition.extend(
+                (
+                    f"{tbname}.{num_col}",
+                    f"ceil( {tbname}.{num_col} )",
+                    f"abs( {tbname}.{num_col} )",
+                    f"acos( {tbname}.{num_col} )",
+                    f"asin( {tbname}.{num_col} )",
+                    f"atan( {tbname}.{num_col} )",
+                    f"cos( {tbname}.{num_col} )",
+                    f"floor( {tbname}.{num_col} )",
+                    f"log( {tbname}.{num_col},  {tbname}.{num_col})",
+                    f"sin( {tbname}.{num_col} )",
+                    f"sqrt( {tbname}.{num_col} )",
+                    f"tan( {tbname}.{num_col} )",
+                    f"round( {tbname}.{num_col} )",
+                    f"max( {tbname}.{num_col} )",
+                    f"sum( {tbname}.{num_col} )",
+                    f"count( {tbname}.{num_col} )",
+                    f"min( {tbname}.{num_col} )",
+                )
+            )
+            query_condition.extend( f"{tbname}.{num_col} + {tbname}.{num_col_2}" for num_col_2 in NUM_COL )
+            query_condition.extend( f"{tbname}.{num_col} + {tbname}.{char_col} " for char_col in CHAR_COL )
+
+        query_condition.extend(
+            (
+                ''' "test1234!@#$%^&*():'><?/.,][}{" ''',
+                ''' "test12" ''',
+                # 1010,
+            )
+        )
+
+        return query_condition
+
+    def __join_condition(self, tb_list, filter=PRIMARY_COL, INNER=False):
+        table_reference = tb_list[0]
+        join_condition = table_reference
+        join = "inner join" if INNER else "join"
+        for i in range(len(tb_list[1:])):
+            join_condition += f" {join} {tb_list[i+1]} on {table_reference}.{filter}={tb_list[i+1]}.{filter}"
+
+        return join_condition
+
+    def __where_condition(self, col=None, tbname=None, query_conditon=None):
+        if query_conditon and isinstance(query_conditon, str):
+            if query_conditon.startswith("count"):
+                query_conditon = query_conditon[6:-1]
+            elif query_conditon.startswith("max"):
+                query_conditon = query_conditon[4:-1]
+            elif query_conditon.startswith("sum"):
+                query_conditon = query_conditon[4:-1]
+            elif query_conditon.startswith("min"):
+                query_conditon = query_conditon[4:-1]
+
+
+        if query_conditon:
+            return f" where {query_conditon} is not null"
+        if col in NUM_COL:
+            return f" where abs( {tbname}.{col} ) >= 0"
+        if col in CHAR_COL:
+            return f" where lower( {tbname}.{col} ) like 'bina%' or lower( {tbname}.{col} ) like '_cha%' "
+        if col in BOOLEAN_COL:
+            return f" where {tbname}.{col} in (false, true)  "
+        if col in TS_TYPE_COL or col in PRIMARY_COL:
+            return f" where cast( {tbname}.{col} as binary(16) ) is not null "
+
+        return ""
+
+
+    def __group_condition(self, col, having = None):
+        if isinstance(col, str):
+            if col.startswith("count"):
+                col = col[6:-1]
+            elif col.startswith("max"):
+                col = col[4:-1]
+            elif col.startswith("sum"):
+                col = col[4:-1]
+            elif col.startswith("min"):
+                col = col[4:-1]
+        return f" group by {col} having {having}" if having else f" group by {col} "
+
+    def __single_sql(self, select_clause, from_clause, where_condition="", group_condition=""):
+        if isinstance(select_clause, str) and "on" not in from_clause and select_clause.split(".")[0] != from_clause.split(".")[0]:
+            return
+        return f"select {select_clause} from {from_clause} {where_condition} {group_condition}"
+
+
+    @property
+    def __join_tblist(self):
+        return [
+            ["ct1", "ct2"],
+            ["ct1", "ct4"],
+            ["ct1", "t1"],
+            ["ct2", "ct4"],
+            ["ct2", "t1"],
+            ["ct4", "t1"],
+            # ["ct1", "ct2", "ct4"],
+            # ["ct1", "ct2", "t1"],
+            # ["ct1", "ct4", "t1"],
+            # ["ct2", "ct4", "t1"],
+            # ["ct1", "ct2", "ct4", "t1"],
+        ]
+
+    @property
+    def __tb_liast(self):
+        return [
+            "ct1",
+            "ct2",
+            "ct4",
+            "t1",
+        ]
+
+    def sql_list(self):
+        sqls = []
+        __join_tblist = self.__join_tblist
+        for join_tblist in __join_tblist:
+            for join_tb in join_tblist:
+                select_claus_list = self.__query_condition(join_tb)
+                for select_claus in select_claus_list:
+                    group_claus = self.__group_condition( col=select_claus)
+                    where_claus = self.__where_condition(query_conditon=select_claus)
+                    having_claus = self.__group_condition( col=select_claus, having=f"{select_claus} is not null")
+                    sqls.extend(
+                        (
+                            self.__single_sql(select_claus, join_tb, where_claus, group_claus),
+                            self.__single_sql(select_claus, join_tb, where_claus, having_claus),
+                            self.__single_sql(select_claus, self.__join_condition(join_tblist), where_claus, having_claus),
+                            self.__single_sql(select_claus, self.__join_condition(join_tblist, INNER=True), where_claus, having_claus),
+                            self.__single_sql(select_claus, join_tb, where_claus),
+                            self.__single_sql(select_claus, join_tb, having_claus),
+                            self.__single_sql(select_claus, join_tb, group_claus),
+                            self.__single_sql(select_claus, join_tb),
+
+                        )
+                    )
+        __no_join_tblist = self.__tb_liast
+        for tb in __no_join_tblist:
+                select_claus_list = self.__query_condition(tb)
+                for select_claus in select_claus_list:
+                    group_claus = self.__group_condition(col=select_claus)
+                    where_claus = self.__where_condition(query_conditon=select_claus)
+                    having_claus = self.__group_condition(col=select_claus, having=f"{select_claus} is not null")
+                    sqls.extend(
+                        (
+                            self.__single_sql(select_claus, join_tb, where_claus, group_claus),
+                            self.__single_sql(select_claus, join_tb, where_claus, having_claus),
+                            self.__single_sql(select_claus, join_tb, where_claus),
+                            self.__single_sql(select_claus, join_tb, group_claus),
+                            self.__single_sql(select_claus, join_tb, having_claus),
+                            self.__single_sql(select_claus, join_tb),
+                        )
+                    )
+
+        return filter(None, sqls)
+        # return list(filter(None, sqls))
+
+    def __get_type(self, col):
+        if tdSql.cursor.istype(col, "BOOL"):
+            return "BOOL"
+        if tdSql.cursor.istype(col, "INT"):
+            return "INT"
+        if tdSql.cursor.istype(col, "BIGINT"):
+            return "BIGINT"
+        if tdSql.cursor.istype(col, "TINYINT"):
+            return "TINYINT"
+        if tdSql.cursor.istype(col, "SMALLINT"):
+            return "SMALLINT"
+        if tdSql.cursor.istype(col, "FLOAT"):
+            return "FLOAT"
+        if tdSql.cursor.istype(col, "DOUBLE"):
+            return "DOUBLE"
+        if tdSql.cursor.istype(col, "BINARY"):
+            return "BINARY"
+        if tdSql.cursor.istype(col, "NCHAR"):
+            return "NCHAR"
+        if tdSql.cursor.istype(col, "TIMESTAMP"):
+            return "TIMESTAMP"
+        if tdSql.cursor.istype(col, "JSON"):
+            return "JSON"
+        if tdSql.cursor.istype(col, "TINYINT UNSIGNED"):
+            return "TINYINT UNSIGNED"
+        if tdSql.cursor.istype(col, "SMALLINT UNSIGNED"):
+            return "SMALLINT UNSIGNED"
+        if tdSql.cursor.istype(col, "INT UNSIGNED"):
+            return "INT UNSIGNED"
+        if tdSql.cursor.istype(col, "BIGINT UNSIGNED"):
+            return "BIGINT UNSIGNED"
+
+    def union_check(self):
+        sqls = self.sql_list()
+        for sql1 in sqls:
+            tdSql.query(sql1)
+            res1_type = self.__get_type(0)
+            for sql2 in sqls:
+                tdSql.query(sql2)
+                union_type = False
+                res2_type =  self.__get_type(0)
+
+                if res1_type in  ( "BIGINT" , "NCHAR" ):
+                    union_type = True
+                elif res2_type == res1_type:
+                    union_type = True
+                elif res1_type == "TIMESAMP" and res2_type not in ("BINARY", "NCHAR"):
+                    union_type = True
+                elif res1_type == "BINARY" and res2_type != "NCHAR":
+                    union_type = True
+
+                if union_type:
+                    tdSql.query(f"{sql1} union {sql2}")
+                    tdSql.checkCols(1)
+                    tdSql.query(f"{sql1} union all {sql2}")
+                    tdSql.checkCols(1)
+                else:
+                    tdSql.error(f"{sql1} union {sql2}")
+
+    def __test_error(self):
+
+
+        tdSql.error( "show tables union show tables" )
+        tdSql.error( "create table errtb1 union all create table errtb2" )
+        tdSql.error( "drop table ct1 union all drop table ct3" )
+        tdSql.error( "select c1 from ct1 union all drop table ct3" )
+        tdSql.error( "select c1 from ct1 union all '' " )
+        tdSql.error( " '' union all select c1 from ct1 " )
+        tdSql.error( "select c1 from ct1 union select c1 from ct2 union select c1 from ct4 ")
+
+    def all_test(self):
+        self.__test_error()
+        self.union_check()
+
+
+    def __create_tb(self):
+
+        tdLog.printNoPrefix("==========step1:create table")
+        create_stb_sql  =  f'''create table stb1(
+                ts timestamp, {INT_COL} int, {BINT_COL} bigint, {SINT_COL} smallint, {TINT_COL} tinyint,
+                 {FLOAT_COL} float, {DOUBLE_COL} double, {BOOL_COL} bool,
+                 {BINARY_COL} binary(16), {NCHAR_COL} nchar(32), {TS_COL} timestamp
+            ) tags (t1 int)
+            '''
+        create_ntb_sql = f'''create table t1(
+                ts timestamp, {INT_COL} int, {BINT_COL} bigint, {SINT_COL} smallint, {TINT_COL} tinyint,
+                 {FLOAT_COL} float, {DOUBLE_COL} double, {BOOL_COL} bool,
+                 {BINARY_COL} binary(16), {NCHAR_COL} nchar(32), {TS_COL} timestamp
+            )
+            '''
+        tdSql.execute(create_stb_sql)
+        tdSql.execute(create_ntb_sql)
+
+        for i in range(4):
+            tdSql.execute(f'create table ct{i+1} using stb1 tags ( {i+1} )')
+            { i % 32767 }, { i % 127}, { i * 1.11111 }, { i * 1000.1111 }, { i % 2}
+
+    def __insert_data(self, rows):
+        now_time = int(datetime.datetime.timestamp(datetime.datetime.now()) * 1000)
+        for i in range(rows):
+            tdSql.execute(
+                f"insert into ct1 values ( { now_time - i * 1000 }, {i}, {11111 * i}, {111 * i % 32767 }, {11 * i % 127}, {1.11*i}, {1100.0011*i}, {i%2}, 'binary{i}', 'nchar_测试_{i}', { now_time + 1 * i } )"
+            )
+            tdSql.execute(
+                f"insert into ct4 values ( { now_time - i * 7776000000 }, {i}, {11111 * i}, {111 * i % 32767 }, {11 * i % 127}, {1.11*i}, {1100.0011*i}, {i%2}, 'binary{i}', 'nchar_测试_{i}', { now_time + 1 * i } )"
+            )
+            tdSql.execute(
+                f"insert into ct2 values ( { now_time - i * 7776000000 }, {-i},  {-11111 * i}, {-111 * i % 32767 }, {-11 * i % 127}, {-1.11*i}, {-1100.0011*i}, {i%2}, 'binary{i}', 'nchar_测试_{i}', { now_time + 1 * i } )"
+            )
+        tdSql.execute(
+            f'''insert into ct1 values
+            ( { now_time - rows * 5 }, 0, 0, 0, 0, 0, 0, 0, 'binary0', 'nchar_测试_0', { now_time + 8 } )
+            ( { now_time + 10000 }, { rows }, -99999, -999, -99, -9.99, -99.99, 1, 'binary9', 'nchar_测试_9', { now_time + 9 } )
+            '''
+        )
+
+        tdSql.execute(
+            f'''insert into ct4 values
+            ( { now_time - rows * 7776000000 }, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL )
+            ( { now_time - rows * 3888000000 + 10800000 }, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL )
+            ( { now_time +  7776000000 }, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL )
+            (
+                { now_time + 5184000000}, {pow(2,31)-pow(2,15)}, {pow(2,63)-pow(2,30)}, 32767, 127,
+                { 3.3 * pow(10,38) }, { 1.3 * pow(10,308) }, { rows % 2 }, "binary_limit-1", "nchar_测试_limit-1", { now_time - 86400000}
+                )
+            (
+                { now_time + 2592000000 }, {pow(2,31)-pow(2,16)}, {pow(2,63)-pow(2,31)}, 32766, 126,
+                { 3.2 * pow(10,38) }, { 1.2 * pow(10,308) }, { (rows-1) % 2 }, "binary_limit-2", "nchar_测试_limit-2", { now_time - 172800000}
+                )
+            '''
+        )
+
+        tdSql.execute(
+            f'''insert into ct2 values
+            ( { now_time - rows * 7776000000 }, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL )
+            ( { now_time - rows * 3888000000 + 10800000 }, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL )
+            ( { now_time + 7776000000 }, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL )
+            (
+                { now_time + 5184000000 }, { -1 * pow(2,31) + pow(2,15) }, { -1 * pow(2,63) + pow(2,30) }, -32766, -126,
+                { -1 * 3.2 * pow(10,38) }, { -1.2 * pow(10,308) }, { rows % 2 }, "binary_limit-1", "nchar_测试_limit-1", { now_time - 86400000 }
+                )
+            (
+                { now_time + 2592000000 }, { -1 * pow(2,31) + pow(2,16) }, { -1 * pow(2,63) + pow(2,31) }, -32767, -127,
+                { - 3.3 * pow(10,38) }, { -1.3 * pow(10,308) }, { (rows-1) % 2 }, "binary_limit-2", "nchar_测试_limit-2", { now_time - 172800000 }
+                )
+            '''
+        )
+
+        for i in range(rows):
+            insert_data = f'''insert into t1 values
+                ( { now_time - i * 3600000 }, {i}, {i * 11111}, { i % 32767 }, { i % 127}, { i * 1.11111 }, { i * 1000.1111 }, { i % 2},
+                "binary_{i}", "nchar_测试_{i}", { now_time - 1000 * i } )
+                '''
+            tdSql.execute(insert_data)
+        tdSql.execute(
+            f'''insert into t1 values
+            ( { now_time + 10800000 }, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL )
+            ( { now_time - (( rows // 2 ) * 60 + 30) * 60000 }, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL )
+            ( { now_time - rows * 3600000 }, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL )
+            ( { now_time + 7200000 }, { pow(2,31) - pow(2,15) }, { pow(2,63) - pow(2,30) }, 32767, 127,
+                { 3.3 * pow(10,38) }, { 1.3 * pow(10,308) }, { rows % 2 },
+                "binary_limit-1", "nchar_测试_limit-1", { now_time - 86400000 }
+                )
+            (
+                { now_time + 3600000 } , { pow(2,31) - pow(2,16) }, { pow(2,63) - pow(2,31) }, 32766, 126,
+                { 3.2 * pow(10,38) }, { 1.2 * pow(10,308) }, { (rows-1) % 2 },
+                "binary_limit-2", "nchar_测试_limit-2", { now_time - 172800000 }
+                )
+            '''
+        )
+
+
+    def run(self):
+        tdSql.prepare()
+
+        tdLog.printNoPrefix("==========step1:create table")
+        self.__create_tb()
+
+        tdLog.printNoPrefix("==========step2:insert data")
+        self.rows = 10
+        self.__insert_data(self.rows)
+
+        tdLog.printNoPrefix("==========step3:all check")
+        self.all_test()
+
+        tdDnodes.stop(1)
+        tdDnodes.start(1)
+
+        tdSql.execute("use db")
+
+        tdLog.printNoPrefix("==========step4:after wal, all check again ")
+        self.all_test()
+
+    def stop(self):
+        tdSql.close()
+        tdLog.success(f"{__file__} successfully executed")
+
+tdCases.addLinux(__file__, TDTestCase())
+tdCases.addWindows(__file__, TDTestCase())

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

@@ -52,7 +52,7 @@ class TDTestCase:
     def create_tables(self,tsql, dbName,vgroups,stbName,ctbNum,rowsPerTbl):
         tsql.execute("create database if not exists %s vgroups %d"%(dbName, vgroups))        
         tsql.execute("use %s" %dbName)
-        tsql.execute("create table %s (ts timestamp, c1 bigint, c2 binary(16)) tags(t1 int)"%stbName)
+        tsql.execute("create table  if not exists %s (ts timestamp, c1 bigint, c2 binary(16)) tags(t1 int)"%stbName)
         pre_create = "create table"
         sql = pre_create
         #tdLog.debug("doing create one  stable %s and %d  child table in %s  ..." %(stbname, count ,dbname))
@@ -345,11 +345,11 @@ class TDTestCase:
                              after starting consumer, create ctables ")
         # create and start thread
         parameterDict = {'cfg':        '',       \
-                         'dbName':     'db2',    \
+                         'dbName':     'db3',    \
                          'vgroups':    1,        \
                          'stbName':    'stb',    \
                          'ctbNum':     10,       \
-                         'rowsPerTbl': 10000,   \
+                         'rowsPerTbl': 30000,    \
                          'batchNum':   100,       \
                          'startTs':    1640966400000}  # 2022-01-01 00:00:00.000
         parameterDict['cfg'] = cfgPath
@@ -375,21 +375,32 @@ class TDTestCase:
             else:
                 time.sleep(1)
         
+        tdLog.info("create stable2 for the seconde topic")
+        parameterDict2 = {'cfg':        '',       \
+                         'dbName':     'db3',    \
+                         'vgroups':    1,        \
+                         'stbName':    'stb2',    \
+                         'ctbNum':     10,       \
+                         'rowsPerTbl': 30000,    \
+                         'batchNum':   100,       \
+                         'startTs':    1640966400000}  # 2022-01-01 00:00:00.000
+        parameterDict2['cfg'] = cfgPath
+        tdSql.execute("create stable  if not exists %s.%s (ts timestamp, c1 bigint, c2 binary(16)) tags(t1 int)"%(parameterDict2['dbName'], parameterDict2['stbName']))
+       
         tdLog.info("create topics from super table")
-        topicFromStb = 'topic_stb_column2'
-        topicFromCtb = 'topic_ctb_column2'
+        topicFromStb  = 'topic_stb_column3'
+        topicFromStb2 = 'topic_stb_column32'
         
         tdSql.execute("create topic %s as select ts, c1, c2 from %s.%s" %(topicFromStb, parameterDict['dbName'], parameterDict['stbName']))
-        tdSql.execute("create topic %s as select ts, c1, c2 from %s.%s_0" %(topicFromCtb, parameterDict['dbName'], parameterDict['stbName']))
+        tdSql.execute("create topic %s as select ts, c1, c2 from %s.%s" %(topicFromStb2, parameterDict2['dbName'], parameterDict2['stbName']))
         
-        time.sleep(1)
         tdSql.query("show topics")
         topic1 = tdSql.getData(0 , 0)
         topic2 = tdSql.getData(1 , 0)
         tdLog.info("show topics: %s, %s"%(topic1, topic2))
-        if topic1 != topicFromStb and topic1 != topicFromCtb:
+        if topic1 != topicFromStb and topic1 != topicFromStb2:
             tdLog.exit("topic error1") 
-        if topic2 != topicFromStb and topic2 != topicFromCtb:
+        if topic2 != topicFromStb and topic2 != topicFromStb2:
             tdLog.exit("topic error2") 
          
         tdLog.info("create consume info table and consume result table")
@@ -397,10 +408,9 @@ class TDTestCase:
         tdSql.query("create table %s.consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)"%cdbName)
         tdSql.query("create table %s.consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)"%cdbName)
 
-        rowsOfNewCtb = 1000
         consumerId   = 0
-        expectrowcnt = parameterDict["rowsPerTbl"]  * parameterDict["ctbNum"] + rowsOfNewCtb
-        topicList    = topicFromStb
+        expectrowcnt = parameterDict["rowsPerTbl"]  * parameterDict["ctbNum"] +  parameterDict2["rowsPerTbl"]  * parameterDict2["ctbNum"]
+        topicList    = topicFromStb + ',' + topicFromStb2
         ifcheckdata  = 0
         keyList      = 'group.id:cgrp1,\
                         enable.auto.commit:false,\
@@ -432,17 +442,13 @@ class TDTestCase:
         tdLog.info(shellCmd)
         os.system(shellCmd)        
 
-        # create new child table and insert data
-        newCtbName = 'newctb'
-        tdSql.query("create table %s.%s using %s.%s tags(9999)"%(parameterDict["dbName"], newCtbName, parameterDict["dbName"], parameterDict["stbName"]))
-        startTs = parameterDict["startTs"]
-        for j in range(rowsOfNewCtb):
-            sql = "insert into %s.%s values (%d, %d, 'tmqrow_%d') "%(parameterDict["dbName"], newCtbName, startTs + j, j, j)
-            tdSql.execute(sql)
-        tdLog.debug("insert data into new child table ............ [OK]")
+        # start the second thread to create new child table and insert data
+        prepareEnvThread2 = threading.Thread(target=self.prepareEnv, kwargs=parameterDict2)
+        prepareEnvThread2.start()
 
         # wait for data ready
         prepareEnvThread.join()
+        prepareEnvThread2.join()
         
         tdLog.info("insert process end, and start to check consume result")
         while 1:
@@ -457,7 +463,7 @@ class TDTestCase:
         tdSql.checkData(0 , 3, expectrowcnt)
         
         tdSql.query("drop topic %s"%topicFromStb)
-        tdSql.query("drop topic %s"%topicFromCtb)
+        tdSql.query("drop topic %s"%topicFromStb2)
 
         tdLog.printNoPrefix("======== test case 3 end ...... ")
 
@@ -474,7 +480,7 @@ class TDTestCase:
 
         self.tmqCase1(cfgPath, buildPath)
         self.tmqCase2(cfgPath, buildPath)        
-        #self.tmqCase3(cfgPath, buildPath)
+        self.tmqCase3(cfgPath, buildPath)
 
     def stop(self):
         tdSql.close()