Merge branch 'main' into feature/3_liaohj

cmake/taostools_CMakeLists.txt.in

@@ -2,7 +2,7 @@
 # taos-tools
 ExternalProject_Add(taos-tools
         GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
-        GIT_TAG 2864326
+        GIT_TAG e82b9fc
         SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
         BINARY_DIR ""
         #BUILD_IN_SOURCE TRUE

include/common/tdatablock.h

@@ -178,7 +178,7 @@ int32_t getJsonValueLen(const char* data);
 
 int32_t colDataSetVal(SColumnInfoData* pColumnInfoData, uint32_t rowIndex, const char* pData, bool isNull);
 int32_t colDataAppend(SColumnInfoData* pColumnInfoData, uint32_t rowIndex, const char* pData, bool isNull);
-int32_t colDataSetNItems(SColumnInfoData* pColumnInfoData, uint32_t rowIndex, const char* pData, uint32_t numOfRows);
+int32_t colDataSetNItems(SColumnInfoData* pColumnInfoData, uint32_t rowIndex, const char* pData, uint32_t numOfRows, bool trimValue);
 int32_t colDataMergeCol(SColumnInfoData* pColumnInfoData, int32_t numOfRow1, int32_t* capacity,
                         const SColumnInfoData* pSource, int32_t numOfRow2);
 int32_t colDataAssign(SColumnInfoData* pColumnInfoData, const SColumnInfoData* pSource, int32_t numOfRows,

include/common/tmsgcb.h

@@ -34,7 +34,7 @@ typedef enum {
   WRITE_QUEUE,
   APPLY_QUEUE,
   SYNC_QUEUE,
-  SYNC_CTRL_QUEUE,
+  SYNC_RD_QUEUE,
   STREAM_QUEUE,
   QUEUE_MAX,
 } EQueueType;

include/common/tmsgdef.h

@@ -259,7 +259,7 @@ enum {
 
   TD_NEW_MSG_SEG(TDMT_SYNC_MSG)
   TD_DEF_MSG_TYPE(TDMT_SYNC_TIMEOUT, "sync-timer", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_SYNC_PING, "sync-ping", NULL, NULL)             // no longer used
+  TD_DEF_MSG_TYPE(TDMT_SYNC_TIMEOUT_ELECTION, "sync-elect", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_SYNC_PING_REPLY, "sync-ping-reply", NULL, NULL) // no longer used
   TD_DEF_MSG_TYPE(TDMT_SYNC_CLIENT_REQUEST, "sync-client-request", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_SYNC_CLIENT_REQUEST_BATCH, "sync-client-request-batch", NULL, NULL)

include/libs/stream/streamState.h

@@ -23,20 +23,19 @@ extern "C" {
 #ifndef _STREAM_STATE_H_
 #define _STREAM_STATE_H_
 
-typedef struct SStreamTask SStreamTask;
-
 typedef bool (*state_key_cmpr_fn)(void* pKey1, void* pKey2);
 
 typedef struct STdbState {
-  SStreamTask* pOwner;
-  TDB*         db;
-  TTB*         pStateDb;
-  TTB*         pFuncStateDb;
-  TTB*         pFillStateDb;  // todo refactor
-  TTB*         pSessionStateDb;
-  TTB*         pParNameDb;
-  TTB*         pParTagDb;
-  TXN*         txn;
+  struct SStreamTask* pOwner;
+
+  TDB* db;
+  TTB* pStateDb;
+  TTB* pFuncStateDb;
+  TTB* pFillStateDb;  // todo refactor
+  TTB* pSessionStateDb;
+  TTB* pParNameDb;
+  TTB* pParTagDb;
+  TXN* txn;
 } STdbState;
 
 // incremental state storage
@@ -45,7 +44,7 @@ typedef struct {
   int32_t    number;
 } SStreamState;
 
-SStreamState* streamStateOpen(char* path, SStreamTask* pTask, bool specPath, int32_t szPage, int32_t pages);
+SStreamState* streamStateOpen(char* path, struct SStreamTask* pTask, bool specPath, int32_t szPage, int32_t pages);
 void          streamStateClose(SStreamState* pState);
 int32_t       streamStateBegin(SStreamState* pState);
 int32_t       streamStateCommit(SStreamState* pState);

include/libs/stream/tstream.h

@@ -295,7 +295,7 @@ typedef struct {
   SEpSet  epSet;
 } SStreamChildEpInfo;
 
-typedef struct SStreamTask {
+struct SStreamTask {
   int64_t streamId;
   int32_t taskId;
   int32_t totalLevel;
@@ -362,8 +362,7 @@ typedef struct SStreamTask {
 
   int64_t checkpointingId;
   int32_t checkpointAlignCnt;
-
-} SStreamTask;
+};
 
 int32_t tEncodeStreamEpInfo(SEncoder* pEncoder, const SStreamChildEpInfo* pInfo);
 int32_t tDecodeStreamEpInfo(SDecoder* pDecoder, SStreamChildEpInfo* pInfo);
@@ -587,7 +586,7 @@ int32_t streamProcessDispatchReq(SStreamTask* pTask, SStreamDispatchReq* pReq, S
 int32_t streamProcessDispatchRsp(SStreamTask* pTask, SStreamDispatchRsp* pRsp, int32_t code);
 
 int32_t streamProcessRetrieveReq(SStreamTask* pTask, SStreamRetrieveReq* pReq, SRpcMsg* pMsg);
-int32_t streamProcessRetrieveRsp(SStreamTask* pTask, SStreamRetrieveRsp* pRsp);
+// int32_t streamProcessRetrieveRsp(SStreamTask* pTask, SStreamRetrieveRsp* pRsp);
 
 int32_t streamTryExec(SStreamTask* pTask);
 int32_t streamSchedExec(SStreamTask* pTask);

include/util/tsimplehash.h

@@ -48,6 +48,13 @@ SSHashObj *tSimpleHashInit(size_t capacity, _hash_fn_t fn);
  */
 int32_t tSimpleHashGetSize(const SSHashObj *pHashObj);
 
+/**
+ * set the free function pointer
+ * @param pHashObj
+ * @param freeFp
+ */
+void tSimpleHashSetFreeFp(SSHashObj* pHashObj, _hash_free_fn_t freeFp);
+
 int32_t tSimpleHashPrint(const SSHashObj *pHashObj);
 
 /**

packaging/rpm/tdengine.spec

@@ -90,45 +90,33 @@ cp %{_compiledir}/../include/libs/function/taosudf.h       %{buildroot}%{homepat
 cp -r %{_compiledir}/../examples/*                  %{buildroot}%{homepath}/examples
 
 if [ -f %{_compiledir}/build/bin/jemalloc-config ]; then
-    mkdir -p %{buildroot}%{userlocalpath}/bin
-    mkdir -p %{buildroot}%{userlocalpath}/lib
-    mkdir -p %{buildroot}%{userlocalpath}/lib/pkgconfig
-    mkdir -p %{buildroot}%{userlocalpath}/include
-    mkdir -p %{buildroot}%{userlocalpath}/include/jemalloc
-    mkdir -p %{buildroot}%{userlocalpath}/share
-    mkdir -p %{buildroot}%{userlocalpath}/share/doc
-    mkdir -p %{buildroot}%{userlocalpath}/share/doc/jemalloc
-    mkdir -p %{buildroot}%{userlocalpath}/share/man
-    mkdir -p %{buildroot}%{userlocalpath}/share/man/man3
-
-    cp %{_compiledir}/build/bin/jemalloc-config %{buildroot}%{userlocalpath}/bin/
+    mkdir -p %{buildroot}%{homepath}/jemalloc/ ||:
+    mkdir -p %{buildroot}%{homepath}/jemalloc/include/jemalloc/ ||:
+    mkdir -p %{buildroot}%{homepath}/jemalloc/lib/ ||:
+    mkdir -p %{buildroot}%{homepath}/jemalloc/lib/pkgconfig ||:
+
+    cp %{_compiledir}/build/bin/jemalloc-config %{buildroot}%{homepath}/jemalloc/bin
     if [ -f %{_compiledir}/build/bin/jemalloc.sh ]; then
-        cp %{_compiledir}/build/bin/jemalloc.sh %{buildroot}%{userlocalpath}/bin/
+        cp %{_compiledir}/build/bin/jemalloc.sh %{buildroot}%{homepath}/jemalloc/bin
     fi
     if [ -f %{_compiledir}/build/bin/jeprof ]; then
-        cp %{_compiledir}/build/bin/jeprof %{buildroot}%{userlocalpath}/bin/
+        cp %{_compiledir}/build/bin/jeprof %{buildroot}%{homepath}/jemalloc/bin
     fi
     if [ -f %{_compiledir}/build/include/jemalloc/jemalloc.h ]; then
-        cp %{_compiledir}/build/include/jemalloc/jemalloc.h %{buildroot}%{userlocalpath}/include/jemalloc/
+        cp %{_compiledir}/build/include/jemalloc/jemalloc.h %{buildroot}%{homepath}/jemalloc/include/jemalloc/
     fi
     if [ -f %{_compiledir}/build/lib/libjemalloc.so.2 ]; then
-        cp %{_compiledir}/build/lib/libjemalloc.so.2 %{buildroot}%{userlocalpath}/lib/
-        ln -sf libjemalloc.so.2 %{buildroot}%{userlocalpath}/lib/libjemalloc.so
+        cp %{_compiledir}/build/lib/libjemalloc.so.2 %{buildroot}%{homepath}/jemalloc/lib
+        ln -sf libjemalloc.so.2 %{buildroot}%{homepath}/jemalloc/lib/libjemalloc.so
     fi
     if [ -f %{_compiledir}/build/lib/libjemalloc.a ]; then
-        cp %{_compiledir}/build/lib/libjemalloc.a %{buildroot}%{userlocalpath}/lib/
+        cp %{_compiledir}/build/lib/libjemalloc.a %{buildroot}%{homepath}/jemalloc/lib
     fi
     if [ -f %{_compiledir}/build/lib/libjemalloc_pic.a ]; then
-        cp %{_compiledir}/build/lib/libjemalloc_pic.a %{buildroot}%{userlocalpath}/lib/
+        cp %{_compiledir}/build/lib/libjemalloc_pic.a %{buildroot}%{homepath}/jemalloc/lib
     fi
     if [ -f %{_compiledir}/build/lib/pkgconfig/jemalloc.pc ]; then
-        cp %{_compiledir}/build/lib/pkgconfig/jemalloc.pc %{buildroot}%{userlocalpath}/lib/pkgconfig/
-    fi
-    if [ -f %{_compiledir}/build/share/doc/jemalloc/jemalloc.html ]; then
-        cp %{_compiledir}/build/share/doc/jemalloc/jemalloc.html %{buildroot}%{userlocalpath}/share/doc/jemalloc/
-    fi
-    if [ -f %{_compiledir}/build/share/man/man3/jemalloc.3 ]; then
-        cp %{_compiledir}/build/share/man/man3/jemalloc.3 %{buildroot}%{userlocalpath}/share/man/man3/
+        cp %{_compiledir}/build/lib/pkgconfig/jemalloc.pc %{buildroot}%{homepath}/jemalloc/lib/pkgconfig
     fi
 fi
 

packaging/tools/makepkg.sh

@@ -51,9 +51,9 @@ fi
 
 if [ -d ${top_dir}/tools/taos-tools/packaging/deb ]; then
   cd ${top_dir}/tools/taos-tools/packaging/deb
-  [ -z "$taos_tools_ver" ] && taos_tools_ver="0.1.0"
-
+  
   taostools_ver=$(git for-each-ref --sort=taggerdate --format '%(tag)' refs/tags|grep -v taos | tail -1)
+  [ -z "$taos_tools_ver" ] && taos_tools_ver="0.1.0"
   taostools_install_dir="${release_dir}/${clientName2}Tools-${taostools_ver}"
 
   cd ${curr_dir}

packaging/tools/post.sh

@@ -145,6 +145,59 @@ function install_include() {
     log_print "install include success"
 }
 
+function install_jemalloc() {
+  jemalloc_dir=${script_dir}/../jemalloc
+
+  if [ -d ${jemalloc_dir} ]; then
+    ${csudo}/usr/bin/install -c -d /usr/local/bin
+
+    if [ -f ${jemalloc_dir}/bin/jemalloc-config ]; then
+      ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/bin/jemalloc-config /usr/local/bin
+    fi
+    if [ -f ${jemalloc_dir}/bin/jemalloc.sh ]; then
+      ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/bin/jemalloc.sh /usr/local/bin
+    fi
+    if [ -f ${jemalloc_dir}/bin/jeprof ]; then
+      ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/bin/jeprof /usr/local/bin
+    fi
+    if [ -f ${jemalloc_dir}/include/jemalloc/jemalloc.h ]; then
+      ${csudo}/usr/bin/install -c -d /usr/local/include/jemalloc
+      ${csudo}/usr/bin/install -c -m 644 ${jemalloc_dir}/include/jemalloc/jemalloc.h /usr/local/include/jemalloc
+    fi
+    if [ -f ${jemalloc_dir}/lib/libjemalloc.so.2 ]; then
+      ${csudo}/usr/bin/install -c -d /usr/local/lib
+      ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc.so.2 /usr/local/lib
+      ${csudo}ln -sf libjemalloc.so.2 /usr/local/lib/libjemalloc.so
+      ${csudo}/usr/bin/install -c -d /usr/local/lib
+      if [ -f ${jemalloc_dir}/lib/libjemalloc.a ]; then
+        ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc.a /usr/local/lib
+      fi
+      if [ -f ${jemalloc_dir}/lib/libjemalloc_pic.a ]; then
+        ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc_pic.a /usr/local/lib
+      fi
+      if [ -f ${jemalloc_dir}/lib/libjemalloc_pic.a ]; then
+        ${csudo}/usr/bin/install -c -d /usr/local/lib/pkgconfig
+        ${csudo}/usr/bin/install -c -m 644 ${jemalloc_dir}/lib/pkgconfig/jemalloc.pc /usr/local/lib/pkgconfig
+      fi
+    fi
+    if [ -f ${jemalloc_dir}/share/doc/jemalloc/jemalloc.html ]; then
+      ${csudo}/usr/bin/install -c -d /usr/local/share/doc/jemalloc
+      ${csudo}/usr/bin/install -c -m 644 ${jemalloc_dir}/share/doc/jemalloc/jemalloc.html /usr/local/share/doc/jemalloc
+    fi
+    if [ -f ${jemalloc_dir}/share/man/man3/jemalloc.3 ]; then
+      ${csudo}/usr/bin/install -c -d /usr/local/share/man/man3
+      ${csudo}/usr/bin/install -c -m 644 ${jemalloc_dir}/share/man/man3/jemalloc.3 /usr/local/share/man/man3
+    fi
+
+    if [ -d /etc/ld.so.conf.d ]; then
+      echo "/usr/local/lib" | ${csudo}tee /etc/ld.so.conf.d/jemalloc.conf >/dev/null || echo -e "failed to write /etc/ld.so.conf.d/jemalloc.conf"
+      ${csudo}ldconfig
+    else
+      echo "/etc/ld.so.conf.d not found!"
+    fi
+  fi
+}
+
 function install_lib() {
     log_print "start install lib from ${lib_dir} to ${lib_link_dir}"
     ${csudo}rm -f ${lib_link_dir}/libtaos* || :
@@ -663,6 +716,7 @@ function install_TDengine() {
     # Install include, lib, binary and service
     install_include &&
     install_lib &&
+    install_jemalloc
     install_bin
 
     if [[ "$?" != 0 ]];then

source/client/src/clientMsgHandler.c

@@ -76,7 +76,7 @@ int32_t processConnectRsp(void* param, SDataBuf* pMsg, int32_t code) {
     goto End;
   }
 
-  if ((code = taosCheckVersionCompatibleFromStr(version, connectRsp.sVer, 2)) != 0) {
+  if ((code = taosCheckVersionCompatibleFromStr(version, connectRsp.sVer, 3)) != 0) {
     setErrno(pRequest, code);
     tsem_post(&pRequest->body.rspSem);
     goto End;
@@ -506,7 +506,7 @@ int32_t processShowVariablesRsp(void* param, SDataBuf* pMsg, int32_t code) {
       code = setQueryResultFromRsp(&pRequest->body.resInfo, pRes, false, true);
     }
 
-    if(code != 0){
+    if (code != 0) {
       taosMemoryFree(pRes);
     }
     tFreeSShowVariablesRsp(&rsp);

source/common/src/tdatablock.c

@@ -147,9 +147,17 @@ int32_t colDataReserve(SColumnInfoData* pColumnInfoData, size_t newSize) {
   return TSDB_CODE_SUCCESS;
 }
 
-static void doCopyNItems(struct SColumnInfoData* pColumnInfoData, int32_t currentRow, const char* pData,
-                         int32_t itemLen, int32_t numOfRows) {
-  ASSERT(pColumnInfoData->info.bytes >= itemLen);
+static int32_t doCopyNItems(struct SColumnInfoData* pColumnInfoData, int32_t currentRow, const char* pData,
+                         int32_t itemLen, int32_t numOfRows, bool trimValue) {
+  if (pColumnInfoData->info.bytes < itemLen) {
+    uWarn("column/tag actual data len %d is bigger than schema len %d, trim it:%d", itemLen, pColumnInfoData->info.bytes, trimValue);
+    if (trimValue) {
+      itemLen = pColumnInfoData->info.bytes;
+    } else {
+      return TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER;
+    }
+  }
+  
   size_t start = 1;
 
   // the first item
@@ -178,10 +186,12 @@ static void doCopyNItems(struct SColumnInfoData* pColumnInfoData, int32_t curren
 
     pColumnInfoData->varmeta.length += numOfRows * itemLen;
   }
+
+  return TSDB_CODE_SUCCESS;
 }
 
 int32_t colDataSetNItems(SColumnInfoData* pColumnInfoData, uint32_t currentRow, const char* pData,
-                            uint32_t numOfRows) {
+                            uint32_t numOfRows, bool trimValue) {
   int32_t len = pColumnInfoData->info.bytes;
   if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
     len = varDataTLen(pData);
@@ -193,8 +203,7 @@ int32_t colDataSetNItems(SColumnInfoData* pColumnInfoData, uint32_t currentRow,
     }
   }
 
-  doCopyNItems(pColumnInfoData, currentRow, pData, len, numOfRows);
-  return TSDB_CODE_SUCCESS;
+  return doCopyNItems(pColumnInfoData, currentRow, pData, len, numOfRows, trimValue);
 }
 
 static void doBitmapMerge(SColumnInfoData* pColumnInfoData, int32_t numOfRow1, const SColumnInfoData* pSource,

source/dnode/mgmt/mgmt_mnode/inc/mmInt.h

@@ -34,7 +34,7 @@ typedef struct SMnodeMgmt {
   SSingleWorker  readWorker;
   SSingleWorker  writeWorker;
   SSingleWorker  syncWorker;
-  SSingleWorker  syncCtrlWorker;
+  SSingleWorker  syncRdWorker;
   bool           stopped;
   int32_t        refCount;
   TdThreadRwlock lock;
@@ -54,7 +54,7 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt);
 void    mmStopWorker(SMnodeMgmt *pMgmt);
 int32_t mmPutMsgToWriteQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t mmPutMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t mmPutMsgToSyncCtrlQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t mmPutMsgToSyncRdQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t mmPutMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t mmPutMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t mmPutMsgToFetchQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);

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

@@ -188,7 +188,8 @@ SArray *mmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_INDEX_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_DISABLE_WRITE_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
 
-  if (dmSetMgmtHandle(pArray, TDMT_SYNC_TIMEOUT, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_TIMEOUT_ELECTION, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_HEARTBEAT, mmPutMsgToSyncRdQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_SYNC_CLIENT_REQUEST, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_SYNC_CLIENT_REQUEST_BATCH, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_SYNC_CLIENT_REQUEST_REPLY, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
@@ -198,11 +199,12 @@ SArray *mmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_SYNC_APPEND_ENTRIES_BATCH, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_SYNC_APPEND_ENTRIES_REPLY, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_SYNC_SNAPSHOT_SEND, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_SYNC_SNAPSHOT_RSP, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_SYNC_PRE_SNAPSHOT, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_SYNC_PRE_SNAPSHOT_REPLY, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_SYNC_HEARTBEAT, mmPutMsgToSyncCtrlQueue, 1) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_SYNC_HEARTBEAT_REPLY, mmPutMsgToSyncCtrlQueue, 1) == NULL) goto _OVER;
+
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_TIMEOUT, mmPutMsgToSyncRdQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_HEARTBEAT_REPLY, mmPutMsgToSyncRdQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_SNAPSHOT_RSP, mmPutMsgToSyncRdQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_PRE_SNAPSHOT_REPLY, mmPutMsgToSyncRdQueue, 1) == NULL) goto _OVER;
 
   code = 0;
 

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

@@ -111,8 +111,8 @@ int32_t mmPutMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   return mmPutMsgToWorker(pMgmt, &pMgmt->syncWorker, pMsg);
 }
 
-int32_t mmPutMsgToSyncCtrlQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
-  return mmPutMsgToWorker(pMgmt, &pMgmt->syncCtrlWorker, pMsg);
+int32_t mmPutMsgToSyncRdQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
+  return mmPutMsgToWorker(pMgmt, &pMgmt->syncRdWorker, pMsg);
 }
 
 int32_t mmPutMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
@@ -151,8 +151,8 @@ int32_t mmPutMsgToQueue(SMnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) {
     case SYNC_QUEUE:
       pWorker = &pMgmt->syncWorker;
       break;
-    case SYNC_CTRL_QUEUE:
-      pWorker = &pMgmt->syncCtrlWorker;
+    case SYNC_RD_QUEUE:
+      pWorker = &pMgmt->syncRdWorker;
       break;
     default:
       terrno = TSDB_CODE_INVALID_PARA;
@@ -238,12 +238,12 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
   SSingleWorkerCfg scCfg = {
       .min = 1,
       .max = 1,
-      .name = "mnode-sync-ctrl",
+      .name = "mnode-sync-rd",
       .fp = (FItem)mmProcessSyncMsg,
       .param = pMgmt,
   };
-  if (tSingleWorkerInit(&pMgmt->syncCtrlWorker, &scCfg) != 0) {
-    dError("failed to start mnode mnode-sync-ctrl worker since %s", terrstr());
+  if (tSingleWorkerInit(&pMgmt->syncRdWorker, &scCfg) != 0) {
+    dError("failed to start mnode mnode-sync-rd worker since %s", terrstr());
     return -1;
   }
 
@@ -259,6 +259,6 @@ void mmStopWorker(SMnodeMgmt *pMgmt) {
   tSingleWorkerCleanup(&pMgmt->readWorker);
   tSingleWorkerCleanup(&pMgmt->writeWorker);
   tSingleWorkerCleanup(&pMgmt->syncWorker);
-  tSingleWorkerCleanup(&pMgmt->syncCtrlWorker);
+  tSingleWorkerCleanup(&pMgmt->syncRdWorker);
   dDebug("mnode workers are closed");
 }

source/dnode/mgmt/mgmt_vnode/inc/vmInt.h

@@ -59,7 +59,7 @@ typedef struct {
   SVnode       *pImpl;
   SMultiWorker  pWriteW;
   SMultiWorker  pSyncW;
-  SMultiWorker  pSyncCtrlW;
+  SMultiWorker  pSyncRdW;
   SMultiWorker  pApplyW;
   STaosQueue   *pQueryQ;
   STaosQueue   *pStreamQ;
@@ -107,7 +107,7 @@ int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc);
 
 int32_t vmPutMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t vmPutMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t vmPutMsgToSyncCtrlQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t vmPutMsgToSyncRdQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t vmPutMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t vmPutMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t vmPutMsgToStreamQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);

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

@@ -549,7 +549,8 @@ SArray *vmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_VNODE, vmPutMsgToMgmtQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_VNODE, vmPutMsgToMgmtQueue, 0) == NULL) goto _OVER;
 
-  if (dmSetMgmtHandle(pArray, TDMT_SYNC_TIMEOUT, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_TIMEOUT_ELECTION, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_HEARTBEAT, vmPutMsgToSyncRdQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_SYNC_CLIENT_REQUEST, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_SYNC_CLIENT_REQUEST_BATCH, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_SYNC_CLIENT_REQUEST_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
@@ -559,12 +560,12 @@ SArray *vmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_SYNC_APPEND_ENTRIES_BATCH, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_SYNC_APPEND_ENTRIES_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_SYNC_SNAPSHOT_SEND, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_SYNC_SNAPSHOT_RSP, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_SYNC_PRE_SNAPSHOT, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_SYNC_PRE_SNAPSHOT_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
 
-  if (dmSetMgmtHandle(pArray, TDMT_SYNC_HEARTBEAT, vmPutMsgToSyncCtrlQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_SYNC_HEARTBEAT_REPLY, vmPutMsgToSyncCtrlQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_TIMEOUT, vmPutMsgToSyncRdQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_HEARTBEAT_REPLY, vmPutMsgToSyncRdQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_SNAPSHOT_RSP, vmPutMsgToSyncRdQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_PRE_SNAPSHOT_REPLY, vmPutMsgToSyncRdQueue, 0) == NULL) goto _OVER;
 
   code = 0;
 

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

@@ -98,9 +98,9 @@ void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode, bool commitAndRemoveWal)
         pVnode->pSyncW.queue->threadId);
   tMultiWorkerCleanup(&pVnode->pSyncW);
 
-  dInfo("vgId:%d, wait for vnode sync ctrl queue:%p is empty, thread:%08" PRId64, pVnode->vgId,
-        pVnode->pSyncCtrlW.queue, pVnode->pSyncCtrlW.queue->threadId);
-  tMultiWorkerCleanup(&pVnode->pSyncCtrlW);
+  dInfo("vgId:%d, wait for vnode sync rd queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncRdW.queue,
+        pVnode->pSyncRdW.queue->threadId);
+  tMultiWorkerCleanup(&pVnode->pSyncRdW);
 
   dInfo("vgId:%d, wait for vnode apply queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pApplyW.queue,
         pVnode->pApplyW.queue->threadId);

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

@@ -216,9 +216,9 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp
       dGTrace("vgId:%d, msg:%p put into vnode-sync queue", pVnode->vgId, pMsg);
       taosWriteQitem(pVnode->pSyncW.queue, pMsg);
       break;
-    case SYNC_CTRL_QUEUE:
-      dGTrace("vgId:%d, msg:%p put into vnode-sync-ctrl queue", pVnode->vgId, pMsg);
-      taosWriteQitem(pVnode->pSyncCtrlW.queue, pMsg);
+    case SYNC_RD_QUEUE:
+      dGTrace("vgId:%d, msg:%p put into vnode-sync-rd queue", pVnode->vgId, pMsg);
+      taosWriteQitem(pVnode->pSyncRdW.queue, pMsg);
       break;
     case APPLY_QUEUE:
       dGTrace("vgId:%d, msg:%p put into vnode-apply queue", pVnode->vgId, pMsg);
@@ -234,9 +234,7 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp
   return code;
 }
 
-int32_t vmPutMsgToSyncCtrlQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
-  return vmPutMsgToQueue(pMgmt, pMsg, SYNC_CTRL_QUEUE);
-}
+int32_t vmPutMsgToSyncRdQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, SYNC_RD_QUEUE); }
 
 int32_t vmPutMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, SYNC_QUEUE); }
 
@@ -327,18 +325,18 @@ int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) {
 int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
   SMultiWorkerCfg wcfg = {.max = 1, .name = "vnode-write", .fp = (FItems)vnodeProposeWriteMsg, .param = pVnode->pImpl};
   SMultiWorkerCfg scfg = {.max = 1, .name = "vnode-sync", .fp = (FItems)vmProcessSyncQueue, .param = pVnode};
-  SMultiWorkerCfg sccfg = {.max = 1, .name = "vnode-sync-ctrl", .fp = (FItems)vmProcessSyncQueue, .param = pVnode};
+  SMultiWorkerCfg sccfg = {.max = 1, .name = "vnode-sync-rd", .fp = (FItems)vmProcessSyncQueue, .param = pVnode};
   SMultiWorkerCfg acfg = {.max = 1, .name = "vnode-apply", .fp = (FItems)vnodeApplyWriteMsg, .param = pVnode->pImpl};
   (void)tMultiWorkerInit(&pVnode->pWriteW, &wcfg);
   (void)tMultiWorkerInit(&pVnode->pSyncW, &scfg);
-  (void)tMultiWorkerInit(&pVnode->pSyncCtrlW, &sccfg);
+  (void)tMultiWorkerInit(&pVnode->pSyncRdW, &sccfg);
   (void)tMultiWorkerInit(&pVnode->pApplyW, &acfg);
 
   pVnode->pQueryQ = tQWorkerAllocQueue(&pMgmt->queryPool, pVnode, (FItem)vmProcessQueryQueue);
   pVnode->pStreamQ = tAutoQWorkerAllocQueue(&pMgmt->streamPool, pVnode, (FItem)vmProcessStreamQueue);
   pVnode->pFetchQ = tWWorkerAllocQueue(&pMgmt->fetchPool, pVnode, (FItems)vmProcessFetchQueue);
 
-  if (pVnode->pWriteW.queue == NULL || pVnode->pSyncW.queue == NULL || pVnode->pSyncCtrlW.queue == NULL ||
+  if (pVnode->pWriteW.queue == NULL || pVnode->pSyncW.queue == NULL || pVnode->pSyncRdW.queue == NULL ||
       pVnode->pApplyW.queue == NULL || pVnode->pQueryQ == NULL || pVnode->pStreamQ == NULL || pVnode->pFetchQ == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
@@ -348,8 +346,8 @@ int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
         pVnode->pWriteW.queue->threadId);
   dInfo("vgId:%d, sync-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncW.queue,
         pVnode->pSyncW.queue->threadId);
-  dInfo("vgId:%d, sync-ctrl-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncCtrlW.queue,
-        pVnode->pSyncCtrlW.queue->threadId);
+  dInfo("vgId:%d, sync-rd-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncRdW.queue,
+        pVnode->pSyncRdW.queue->threadId);
   dInfo("vgId:%d, apply-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pApplyW.queue,
         pVnode->pApplyW.queue->threadId);
   dInfo("vgId:%d, query-queue:%p is alloced", pVnode->vgId, pVnode->pQueryQ);

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

@@ -226,7 +226,7 @@ static int32_t mndProcessConnectReq(SRpcMsg *pReq) {
     goto _OVER;
   }
 
-  if ((code = taosCheckVersionCompatibleFromStr(connReq.sVer, version, 2)) != 0) {
+  if ((code = taosCheckVersionCompatibleFromStr(connReq.sVer, version, 3)) != 0) {
     terrno = code;
     goto _OVER;
   }

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

@@ -33,7 +33,7 @@ static int32_t mndSyncEqCtrlMsg(const SMsgCb *msgcb, SRpcMsg *pMsg) {
     return -1;
   }
 
-  int32_t code = tmsgPutToQueue(msgcb, SYNC_CTRL_QUEUE, pMsg);
+  int32_t code = tmsgPutToQueue(msgcb, SYNC_RD_QUEUE, pMsg);
   if (code != 0) {
     rpcFreeCont(pMsg->pCont);
     pMsg->pCont = NULL;

source/dnode/vnode/CMakeLists.txt

@@ -57,7 +57,7 @@ target_sources(
 
     # tq
     "src/tq/tq.c"
-    "src/tq/tqExec.c"
+    "src/tq/tqScan.c"
     "src/tq/tqMeta.c"
     "src/tq/tqRead.c"
     "src/tq/tqOffset.c"

source/dnode/vnode/inc/vnode.h

@@ -182,7 +182,7 @@ int32_t tsdbReaderOpen(SVnode *pVnode, SQueryTableDataCond *pCond, void *pTableL
 
 void         tsdbReaderSetId(STsdbReader* pReader, const char* idstr);
 void         tsdbReaderClose(STsdbReader *pReader);
-bool         tsdbNextDataBlock(STsdbReader *pReader);
+int32_t      tsdbNextDataBlock(STsdbReader *pReader, bool *hasNext);
 int32_t      tsdbRetrieveDatablockSMA(STsdbReader *pReader, SSDataBlock *pDataBlock, bool *allHave);
 void         tsdbReleaseDataBlock(STsdbReader *pReader);
 SSDataBlock *tsdbRetrieveDataBlock(STsdbReader *pTsdbReadHandle, SArray *pColumnIdList);

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

@@ -193,7 +193,7 @@ STQ*    tqOpen(const char* path, SVnode* pVnode);
 void    tqClose(STQ*);
 int     tqPushMsg(STQ*, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver);
 int     tqRegisterPushEntry(STQ* pTq, void* pHandle, const SMqPollReq* pRequest, SRpcMsg* pRpcMsg, SMqDataRsp* pDataRsp, int32_t type);
-int     tqRemovePushEntry(STQ* pTq, const char* pKey, int32_t keyLen, uint64_t consumerId, bool rspConsumer);
+int     tqUnregisterPushEntry(STQ* pTq, const char* pKey, int32_t keyLen, uint64_t consumerId, bool rspConsumer);
 
 int     tqCommit(STQ*);
 int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, bool isAdd);

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

@@ -223,19 +223,6 @@ static int32_t doSendDataRsp(const SRpcHandleInfo* pRpcHandleInfo, const SMqData
 
 int32_t tqPushDataRsp(STQ* pTq, STqPushEntry* pPushEntry) {
   SMqDataRsp* pRsp = pPushEntry->pDataRsp;
-
-#if 0
-  A(taosArrayGetSize(pRsp->blockData) == pRsp->blockNum);
-  A(taosArrayGetSize(pRsp->blockDataLen) == pRsp->blockNum);
-
-  A(!pRsp->withSchema);
-  A(taosArrayGetSize(pRsp->blockSchema) == 0);
-
-  if (pRsp->reqOffset.type == TMQ_OFFSET__LOG) {
-    A(pRsp->rspOffset.version > pRsp->reqOffset.version);
-  }
-#endif
-
   SMqRspHead* pHeader = &pPushEntry->pDataRsp->head;
   doSendDataRsp(&pPushEntry->info, pRsp, pHeader->epoch, pHeader->consumerId, pHeader->mqMsgType);
 
@@ -486,6 +473,10 @@ static int32_t extractDataAndRspForNormalSubscribe(STQ* pTq, STqHandle* pHandle,
 
   qSetTaskId(pHandle->execHandle.task, consumerId, pRequest->reqId);
   code = tqScanData(pTq, pHandle, &dataRsp, pOffset);
+  if (code != TSDB_CODE_SUCCESS) {
+    taosWUnLockLatch(&pTq->lock);
+    return code;
+  }
 
   // till now, all data has been transferred to consumer, new data needs to push client once arrived.
   if (dataRsp.blockNum == 0 && dataRsp.reqOffset.type == TMQ_OFFSET__LOG &&
@@ -905,7 +896,7 @@ int32_t tqProcessSubscribeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg
       atomic_store_32(&pHandle->epoch, -1);
 
       // remove if it has been register in the push manager, and return one empty block to consumer
-      tqRemovePushEntry(pTq, req.subKey, (int32_t)strlen(req.subKey), pHandle->consumerId, true);
+      tqUnregisterPushEntry(pTq, req.subKey, (int32_t)strlen(req.subKey), pHandle->consumerId, true);
 
       atomic_store_64(&pHandle->consumerId, req.newConsumerId);
       atomic_add_fetch_32(&pHandle->epoch, 1);

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

@@ -206,7 +206,84 @@ int32_t tqPushMsgNew(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_
 }
 #endif
 
-int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) {
+typedef struct {
+  void* pKey;
+  int64_t keyLen;
+} SItem;
+
+static void recordPushedEntry(SArray* cachedKey, void* pIter);
+
+static void freeItem(void* param) {
+  SItem* p = (SItem*) param;
+  taosMemoryFree(p->pKey);
+}
+
+static void doRemovePushedEntry(SArray* pCachedKeys, STQ* pTq) {
+  int32_t vgId = TD_VID(pTq->pVnode);
+  int32_t  numOfKeys = (int32_t) taosArrayGetSize(pCachedKeys);
+
+  for (int32_t i = 0; i < numOfKeys; i++) {
+    SItem* pItem = taosArrayGet(pCachedKeys, i);
+    if (taosHashRemove(pTq->pPushMgr, pItem->pKey, pItem->keyLen) != 0) {
+      tqError("vgId:%d, tq push hash remove key error, key: %s", vgId, (char*) pItem->pKey);
+    }
+  }
+
+  if (numOfKeys > 0) {
+    tqDebug("vgId:%d, pushed %d items and remain:%d", vgId, numOfKeys, (int32_t)taosHashGetSize(pTq->pPushMgr));
+  }
+}
+
+static void doPushDataForEntry(void* pIter, STqExecHandle* pExec, STQ* pTq, int64_t ver, int32_t vgId, char* pData,
+                               int32_t dataLen, SArray* pCachedKey) {
+  STqPushEntry* pPushEntry = *(STqPushEntry**)pIter;
+
+  SMqDataRsp* pRsp = pPushEntry->pDataRsp;
+  if (pRsp->reqOffset.version >= ver) {
+    tqDebug("vgId:%d, push entry req version %" PRId64 ", while push version %" PRId64 ", skip", vgId,
+            pRsp->reqOffset.version, ver);
+    return;
+  }
+
+  qTaskInfo_t pTaskInfo = pExec->task;
+
+  // prepare scan mem data
+  SPackedData submit = {.msgStr = pData, .msgLen = dataLen, .ver = ver};
+
+  if (qStreamSetScanMemData(pTaskInfo, submit) != 0) {
+    return;
+  }
+
+  // here start to scan submit block to extract the subscribed data
+  int32_t totalRows = 0;
+
+  while (1) {
+    SSDataBlock* pDataBlock = NULL;
+    uint64_t     ts = 0;
+    if (qExecTask(pTaskInfo, &pDataBlock, &ts) < 0) {
+      tqDebug("vgId:%d, tq exec error since %s", vgId, terrstr());
+    }
+
+    if (pDataBlock == NULL) {
+      break;
+    }
+
+    tqAddBlockDataToRsp(pDataBlock, pRsp, pExec->numOfCols, pTq->pVnode->config.tsdbCfg.precision);
+    pRsp->blockNum++;
+    totalRows += pDataBlock->info.rows;
+  }
+
+  tqDebug("vgId:%d, tq handle push, subkey:%s, block num:%d, rows:%d", vgId, pPushEntry->subKey, pRsp->blockNum,
+      totalRows);
+
+  if (pRsp->blockNum > 0) {
+    tqOffsetResetToLog(&pRsp->rspOffset, ver);
+    tqPushDataRsp(pTq, pPushEntry);
+    recordPushedEntry(pCachedKey, pIter);
+  }
+}
+
+int32_t tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) {
   void*   pReq = POINTER_SHIFT(msg, sizeof(SSubmitReq2Msg));
   int32_t len = msgLen - sizeof(SSubmitReq2Msg);
   int32_t vgId = TD_VID(pTq->pVnode);
@@ -220,24 +297,19 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
       tqDebug("vgId:%d tq push msg version:%" PRId64 " type:%s, head:%p, body:%p len:%d, numOfPushed consumers:%d",
           vgId, ver, TMSG_INFO(msgType), msg, pReq, len, numOfRegisteredPush);
 
-      SArray* cachedKeys = taosArrayInit(0, sizeof(void*));
-      SArray* cachedKeyLens = taosArrayInit(0, sizeof(size_t));
-
-      void*   data = taosMemoryMalloc(len);
+      void* data = taosMemoryMalloc(len);
       if (data == NULL) {
         terrno = TSDB_CODE_OUT_OF_MEMORY;
-        tqError("failed to copy data for stream since out of memory");
-        taosArrayDestroyP(cachedKeys, (FDelete)taosMemoryFree);
-        taosArrayDestroy(cachedKeyLens);
-
-        // unlock
+        tqError("failed to copy data for stream since out of memory, vgId:%d", vgId);
         taosWUnLockLatch(&pTq->lock);
         return -1;
       }
 
       memcpy(data, pReq, len);
 
-      void* pIter = NULL;
+      SArray* cachedKey = taosArrayInit(0, sizeof(SItem));
+      void*   pIter = NULL;
+
       while (1) {
         pIter = taosHashIterate(pTq->pPushMgr, pIter);
         if (pIter == NULL) {
@@ -248,83 +320,29 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
 
         STqHandle* pHandle = taosHashGet(pTq->pHandle, pPushEntry->subKey, strlen(pPushEntry->subKey));
         if (pHandle == NULL) {
-          tqDebug("vgId:%d, cannot find handle %s", pTq->pVnode->config.vgId, pPushEntry->subKey);
-          continue;
-        }
-
-        SMqDataRsp* pRsp = pPushEntry->pDataRsp;
-        if (pRsp->reqOffset.version >= ver) {
-          tqDebug("vgId:%d, push entry req version %" PRId64 ", while push version %" PRId64 ", skip", vgId,
-                  pRsp->reqOffset.version, ver);
+          tqDebug("vgId:%d, failed to find handle %s in pushing data to consumer, ignore", pTq->pVnode->config.vgId, pPushEntry->subKey);
           continue;
         }
 
         STqExecHandle* pExec = &pHandle->execHandle;
-        qTaskInfo_t    task = pExec->task;
-
-        // prepare scan mem data
-        SPackedData submit = {
-            .msgStr = data,
-            .msgLen = len,
-            .ver = ver,
-        };
-        if(qStreamSetScanMemData(task, submit) != 0){
-          continue;
-        }
-
-        // here start to scan submit block to extract the subscribed data
-        while (1) {
-          SSDataBlock* pDataBlock = NULL;
-          uint64_t     ts = 0;
-          if (qExecTask(task, &pDataBlock, &ts) < 0) {
-            tqDebug("vgId:%d, tq exec error since %s", vgId, terrstr());
-          }
-
-          if (pDataBlock == NULL) {
-            break;
-          }
-
-          tqAddBlockDataToRsp(pDataBlock, pRsp, pExec->numOfCols, pTq->pVnode->config.tsdbCfg.precision);
-          pRsp->blockNum++;
-        }
-
-        tqDebug("vgId:%d, tq handle push, subkey:%s, block num:%d", vgId, pPushEntry->subKey, pRsp->blockNum);
-        if (pRsp->blockNum > 0) {
-          // set offset
-          tqOffsetResetToLog(&pRsp->rspOffset, ver);
-
-          // remove from hash
-          size_t kLen;
-          void*  key = taosHashGetKey(pIter, &kLen);
-          void*  keyCopy = taosMemoryCalloc(1, kLen + 1);
-          memcpy(keyCopy, key, kLen);
-
-          taosArrayPush(cachedKeys, &keyCopy);
-          taosArrayPush(cachedKeyLens, &kLen);
-
-          tqPushDataRsp(pTq, pPushEntry);
-        }
-      }
-
-      // delete entry
-      for (int32_t i = 0; i < taosArrayGetSize(cachedKeys); i++) {
-        void*  key = taosArrayGetP(cachedKeys, i);
-        size_t kLen = *(size_t*)taosArrayGet(cachedKeyLens, i);
-        if (taosHashRemove(pTq->pPushMgr, key, kLen) != 0) {
-          tqError("vgId:%d, tq push hash remove key error, key: %s", pTq->pVnode->config.vgId, (char*)key);
-        }
+        doPushDataForEntry(pIter, pExec, pTq, ver, vgId, data, len, cachedKey);
       }
 
-      taosArrayDestroyP(cachedKeys, (FDelete)taosMemoryFree);
-      taosArrayDestroy(cachedKeyLens);
+      doRemovePushedEntry(cachedKey, pTq);
+      taosArrayDestroyEx(cachedKey, freeItem);
       taosMemoryFree(data);
     }
+
     // unlock
     taosWUnLockLatch(&pTq->lock);
   }
 
+  // push data for stream processing
   if (!tsDisableStream && vnodeIsRoleLeader(pTq->pVnode)) {
-    if (taosHashGetSize(pTq->pStreamMeta->pTasks) == 0) return 0;
+    if (taosHashGetSize(pTq->pStreamMeta->pTasks) == 0) {
+      return 0;
+    }
+
     if (msgType == TDMT_VND_SUBMIT) {
       void* data = taosMemoryMalloc(len);
       if (data == NULL) {
@@ -332,12 +350,9 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
         tqError("failed to copy data for stream since out of memory");
         return -1;
       }
+
       memcpy(data, pReq, len);
-      SPackedData submit = {
-          .msgStr = data,
-          .msgLen = len,
-          .ver = ver,
-      };
+      SPackedData submit = {.msgStr = data, .msgLen = len, .ver = ver};
 
       tqDebug("tq copy write msg %p %d %" PRId64 " from %p", data, len, ver, pReq);
       tqProcessSubmitReq(pTq, submit);
@@ -351,6 +366,13 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
   return 0;
 }
 
+void recordPushedEntry(SArray* cachedKey, void* pIter) {
+  size_t kLen = 0;
+  void*  key = taosHashGetKey(pIter, &kLen);
+  SItem item = {.pKey = strndup(key, kLen), .keyLen = kLen};
+  taosArrayPush(cachedKey, &item);
+}
+
 int32_t tqRegisterPushEntry(STQ* pTq, void* pHandle, const SMqPollReq* pRequest, SRpcMsg* pRpcMsg,
                             SMqDataRsp* pDataRsp, int32_t type) {
   uint64_t   consumerId = pRequest->consumerId;
@@ -388,8 +410,8 @@ int32_t tqRegisterPushEntry(STQ* pTq, void* pHandle, const SMqPollReq* pRequest,
   return 0;
 }
 
-int32_t tqRemovePushEntry(STQ* pTq, const char* pKey, int32_t keyLen, uint64_t consumerId, bool rspConsumer) {
-  int32_t       vgId = TD_VID(pTq->pVnode);
+int32_t tqUnregisterPushEntry(STQ* pTq, const char* pKey, int32_t keyLen, uint64_t consumerId, bool rspConsumer) {
+  int32_t        vgId = TD_VID(pTq->pVnode);
   STqPushEntry** pEntry = taosHashGet(pTq->pPushMgr, pKey, keyLen);
 
   if (pEntry != NULL) {

source/dnode/vnode/src/tq/tqExec.c → source/dnode/vnode/src/tq/tqScan.c

@@ -18,7 +18,9 @@
 int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t numOfCols, int8_t precision) {
   int32_t dataStrLen = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock);
   void*   buf = taosMemoryCalloc(1, dataStrLen);
-  if (buf == NULL) return -1;
+  if (buf == NULL) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
 
   SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)buf;
   pRetrieve->useconds = 0;
@@ -31,7 +33,8 @@ int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t
   actualLen += sizeof(SRetrieveTableRsp);
   taosArrayPush(pRsp->blockDataLen, &actualLen);
   taosArrayPush(pRsp->blockData, &buf);
-  return 0;
+
+  return TSDB_CODE_SUCCESS;
 }
 
 static int32_t tqAddBlockSchemaToRsp(const STqExecHandle* pExec, STaosxRsp* pRsp) {
@@ -62,72 +65,85 @@ static int32_t tqAddTbNameToRsp(const STQ* pTq, int64_t uid, STaosxRsp* pRsp, in
 }
 
 int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVal* pOffset) {
-  const STqExecHandle* pExec = &pHandle->execHandle;
+  const int32_t MAX_ROWS_TO_RETURN = 4096;
+  int32_t       vgId = TD_VID(pTq->pVnode);
+  int32_t       code = 0;
+  int32_t       totalRows = 0;
 
-  qTaskInfo_t task = pExec->task;
-  int32_t vgId = TD_VID(pTq->pVnode);
+  const STqExecHandle* pExec = &pHandle->execHandle;
+  qTaskInfo_t          task = pExec->task;
 
   if (qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType) < 0) {
     tqDebug("prepare scan failed, vgId:%d, consumer:0x%"PRIx64, vgId, pHandle->consumerId);
     if (pOffset->type == TMQ_OFFSET__LOG) {
       pRsp->rspOffset = *pOffset;
-      return 0;
+      return code;
     } else {
       tqOffsetResetToLog(pOffset, pHandle->snapshotVer);
       if (qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType) < 0) {
         tqDebug("prepare scan failed, vgId:%d, consumer:0x%"PRIx64, vgId, pHandle->consumerId);
         pRsp->rspOffset = *pOffset;
-        return 0;
+        return code;
       }
     }
   }
 
-  int32_t rowCnt = 0;
   while (1) {
     SSDataBlock* pDataBlock = NULL;
     uint64_t     ts = 0;
 
-    tqDebug("vgId:%d, tmq task start to execute, consumer:0x%"PRIx64, vgId, pHandle->consumerId);
-    if (qExecTask(task, &pDataBlock, &ts) < 0) {
+    tqDebug("vgId:%d, tmq task start to execute, consumer:0x%" PRIx64, vgId, pHandle->consumerId);
+
+    code = qExecTask(task, &pDataBlock, &ts);
+    if (code != TSDB_CODE_SUCCESS) {
       tqError("vgId:%d, task exec error since %s, consumer:0x%" PRIx64, vgId, terrstr(),
               pHandle->consumerId);
-      return -1;
+      return code;
     }
 
-    tqDebug("consumer:0x%"PRIx64" vgId:%d, tmq task executed, get %p", pHandle->consumerId, vgId, pDataBlock);
-
-    // current scan should be stopped asap, since the rebalance occurs.
+    // current scan should be stopped ASAP, since the re-balance occurs.
     if (pDataBlock == NULL) {
       break;
     }
 
-    tqAddBlockDataToRsp(pDataBlock, pRsp, pExec->numOfCols, pTq->pVnode->config.tsdbCfg.precision);
+    code = tqAddBlockDataToRsp(pDataBlock, pRsp, pExec->numOfCols, pTq->pVnode->config.tsdbCfg.precision);
+    if (code != TSDB_CODE_SUCCESS) {
+      tqError("vgId:%d, failed to add block to rsp msg", vgId);
+      return code;
+    }
+
     pRsp->blockNum++;
 
+    tqDebug("vgId:%d, consumer:0x%" PRIx64 " tmq task executed, rows:%d, total blocks:%d", vgId, pHandle->consumerId,
+            pDataBlock->info.rows, pRsp->blockNum);
+
     if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {
-      rowCnt += pDataBlock->info.rows;
-      if (rowCnt >= 4096) {
+      totalRows += pDataBlock->info.rows;
+      if (totalRows >= MAX_ROWS_TO_RETURN) {
         break;
       }
     }
   }
 
-  if (qStreamExtractOffset(task, &pRsp->rspOffset) < 0) {
-    return -1;
-  }
+  qStreamExtractOffset(task, &pRsp->rspOffset);
 
   if (pRsp->rspOffset.type == 0) {
+    code = TSDB_CODE_INVALID_PARA;
     tqError("vgId:%d, expected rsp offset: type %d %" PRId64 " %" PRId64 " %" PRId64, vgId, pRsp->rspOffset.type,
             pRsp->rspOffset.ts, pRsp->rspOffset.uid, pRsp->rspOffset.version);
-    return -1;
+    return code;
   }
 
   if (pRsp->withTbName || pRsp->withSchema) {
+    code = TSDB_CODE_INVALID_PARA;
     tqError("vgId:%d, get column should not with meta:%d,%d", vgId, pRsp->withTbName, pRsp->withSchema);
-    return -1;
+    return code;
   }
 
-  return 0;
+  tqDebug("vgId:%d, consumer:0x%" PRIx64 " tmq task executed, total blocks:%d, rows:%d", vgId, pHandle->consumerId,
+          pRsp->blockNum, totalRows);
+
+  return code;
 }
 
 int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMetaRsp* pMetaRsp, STqOffsetVal* pOffset) {

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

@@ -378,7 +378,7 @@ static int32_t vnodeSyncEqCtrlMsg(const SMsgCb *msgcb, SRpcMsg *pMsg) {
     return -1;
   }
 
-  int32_t code = tmsgPutToQueue(msgcb, SYNC_CTRL_QUEUE, pMsg);
+  int32_t code = tmsgPutToQueue(msgcb, SYNC_RD_QUEUE, pMsg);
   if (code != 0) {
     rpcFreeCont(pMsg->pCont);
     pMsg->pCont = NULL;

source/libs/executor/src/scanoperator.c

@@ -584,10 +584,16 @@ int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int
       if (isNullVal) {
         colDataSetNNULL(pColInfoData, 0, pBlock->info.rows);
       } else if (pColInfoData->info.type != TSDB_DATA_TYPE_JSON) {
-        colDataSetNItems(pColInfoData, 0, data, pBlock->info.rows);
+        code = colDataSetNItems(pColInfoData, 0, data, pBlock->info.rows, false);
         if (IS_VAR_DATA_TYPE(((const STagVal*)p)->type)) {
           taosMemoryFree(data);
         }
+        if (code) {
+          if (freeReader) {
+            metaReaderClear(&mr);
+          }
+          return code;
+        }
       } else {  // todo opt for json tag
         for (int32_t i = 0; i < pBlock->info.rows; ++i) {
           colDataSetVal(pColInfoData, i, data, false);
@@ -634,10 +640,22 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
   STableScanInfo* pTableScanInfo = pOperator->info;
   SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;
   SSDataBlock*    pBlock = pTableScanInfo->pResBlock;
+  bool            hasNext = false;
+  int32_t         code = TSDB_CODE_SUCCESS;
 
   int64_t st = taosGetTimestampUs();
 
-  while (tsdbNextDataBlock(pTableScanInfo->base.dataReader)) {
+  while (true) {
+    code = tsdbNextDataBlock(pTableScanInfo->base.dataReader, &hasNext);
+    if (code) {
+      tsdbReleaseDataBlock(pTableScanInfo->base.dataReader);
+      T_LONG_JMP(pTaskInfo->env, code);
+    }
+
+    if (!hasNext) {
+      break;
+    }
+    
     if (isTaskKilled(pTaskInfo)) {
       tsdbReleaseDataBlock(pTableScanInfo->base.dataReader);
       T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
@@ -1015,7 +1033,15 @@ static SSDataBlock* readPreVersionData(SOperatorInfo* pTableScanOp, uint64_t tbU
     return NULL;
   }
 
-  if (tsdbNextDataBlock(pReader)) {
+  bool hasNext = false;
+  code = tsdbNextDataBlock(pReader, &hasNext);
+  if (code != TSDB_CODE_SUCCESS) {
+    terrno = code;
+    T_LONG_JMP(pTaskInfo->env, code);
+    return NULL;
+  }
+
+  if (hasNext) {
     /*SSDataBlock* p = */ tsdbRetrieveDataBlock(pReader, NULL);
     doSetTagColumnData(&pTableScanInfo->base, pBlock, pTaskInfo, pBlock->info.rows);
     pBlock->info.id.groupId = getTableGroupId(pTaskInfo->pTableInfoList, pBlock->info.id.uid);
@@ -2100,12 +2126,22 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
   // NOTE: this operator does never check if current status is done or not
   SExecTaskInfo*      pTaskInfo = pOperator->pTaskInfo;
   SStreamRawScanInfo* pInfo = pOperator->info;
+  int32_t             code = TSDB_CODE_SUCCESS;
   pTaskInfo->streamInfo.metaRsp.metaRspLen = 0;  // use metaRspLen !=0 to judge if data is meta
   pTaskInfo->streamInfo.metaRsp.metaRsp = NULL;
 
   qDebug("tmqsnap doRawScan called");
   if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_DATA) {
-    if (pInfo->dataReader && tsdbNextDataBlock(pInfo->dataReader)) {
+    bool hasNext = false;
+    if (pInfo->dataReader) {
+      code = tsdbNextDataBlock(pInfo->dataReader, &hasNext);
+      if (code) {
+        tsdbReleaseDataBlock(pInfo->dataReader);
+        longjmp(pTaskInfo->env, code);
+      }
+    }
+    
+    if (pInfo->dataReader && hasNext) {
       if (isTaskKilled(pTaskInfo)) {
         tsdbReleaseDataBlock(pInfo->dataReader);
         longjmp(pTaskInfo->env, pTaskInfo->code);
@@ -2606,8 +2642,21 @@ static SSDataBlock* getTableDataBlockImpl(void* param) {
   
   pInfo->base.dataReader = source->dataReader;
   STsdbReader* reader = pInfo->base.dataReader;
+  bool hasNext = false;
   qTrace("tsdb/read-table-data: %p, enter next reader", reader);
-  while (tsdbNextDataBlock(reader)) {
+
+  while (true) {
+    code = tsdbNextDataBlock(reader, &hasNext);
+    if (code != 0) {
+      tsdbReleaseDataBlock(reader);
+      pInfo->base.dataReader = NULL;
+      T_LONG_JMP(pTaskInfo->env, code);
+    }
+
+    if (!hasNext) {
+      break;
+    }
+  
     if (isTaskKilled(pTaskInfo)) {
       tsdbReleaseDataBlock(reader);
       pInfo->base.dataReader = NULL;

source/libs/executor/src/sysscanoperator.c

@@ -1627,7 +1627,7 @@ static void sysTableScanFillTbName(SOperatorInfo* pOperator, const SSysTableScan
     char             varTbName[TSDB_TABLE_FNAME_LEN - 1 + VARSTR_HEADER_SIZE] = {0};
     STR_TO_VARSTR(varTbName, name);
 
-    colDataSetNItems(pColumnInfoData, 0, varTbName, pBlock->info.rows);
+    colDataSetNItems(pColumnInfoData, 0, varTbName, pBlock->info.rows, true);
   }
 
   doFilter(pBlock, pOperator->exprSupp.pFilterInfo, NULL);

source/libs/function/src/builtins.c

@@ -2375,7 +2375,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "spread",
     .type = FUNCTION_TYPE_SPREAD,
-    .classification = FUNC_MGT_AGG_FUNC,
+    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SPECIAL_DATA_REQUIRED,
     .translateFunc = translateSpread,
     .dataRequiredFunc = statisDataRequired,
     .getEnvFunc   = getSpreadFuncEnv,
@@ -2417,7 +2417,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "elapsed",
     .type = FUNCTION_TYPE_ELAPSED,
-    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_INTERVAL_INTERPO_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
+    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_INTERVAL_INTERPO_FUNC | FUNC_MGT_FORBID_STREAM_FUNC | FUNC_MGT_SPECIAL_DATA_REQUIRED,
     .dataRequiredFunc = statisDataRequired,
     .translateFunc = translateElapsed,
     .getEnvFunc   = getElapsedFuncEnv,
@@ -2500,7 +2500,8 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .sprocessFunc = firstLastScalarFunction,
     .pPartialFunc = "_last_row_partial",
     .pMergeFunc   = "_last_row_merge",
-    .finalizeFunc = firstLastFinalize
+    .finalizeFunc = firstLastFinalize,
+    .combineFunc  = lastCombine
   },
   {
     .name = "_cache_last_row",
@@ -2809,7 +2810,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "mode",
     .type = FUNCTION_TYPE_MODE,
-    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC,
+    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
     .translateFunc = translateMode,
     .getEnvFunc   = getModeFuncEnv,
     .initFunc     = modeFunctionSetup,
@@ -3212,7 +3213,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "_block_dist",
     .type = FUNCTION_TYPE_BLOCK_DIST,
-    .classification = FUNC_MGT_AGG_FUNC,
+    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
     .translateFunc = translateBlockDistFunc,
     .getEnvFunc   = getBlockDistFuncEnv,
     .initFunc     = blockDistSetup,

source/libs/parser/src/parInsertUtil.c

@@ -631,10 +631,10 @@ int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreate
     ret = TSDB_CODE_INVALID_PARA;
     goto end;
   }
-  for (int c = 0; c < boundInfo->numOfBound; ++c) {
-    SSchema*  pColSchema = &pSchema[c];
-    SColData* pCol = taosArrayGet(pTableCxt->pData->aCol, c);
-    if(tFields == NULL || findFileds(pColSchema, tFields, numFields)){
+  if(tFields == NULL){
+    for (int j = 0; j < boundInfo->numOfBound; j++){
+      SSchema*  pColSchema = &pSchema[j];
+      SColData* pCol = taosArrayGet(pTableCxt->pData->aCol, j);
       if (*fields != pColSchema->type && *(int32_t*)(fields + sizeof(int8_t)) != pColSchema->bytes) {
         uError("type or bytes not equal");
         ret = TSDB_CODE_INVALID_PARA;
@@ -652,12 +652,52 @@ int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreate
       tColDataAddValueByDataBlock(pCol, pColSchema->type, pColSchema->bytes, numOfRows, offset, pData);
       fields += sizeof(int8_t) + sizeof(int32_t);
       if (needChangeLength) {
-        pStart += htonl(colLength[c]);
+        pStart += htonl(colLength[j]);
       } else {
-        pStart += colLength[c];
+        pStart += colLength[j];
+      }
+    }
+  }else{
+    for (int i = 0; i < numFields; i++) {
+      for (int j = 0; j < boundInfo->numOfBound; j++){
+        SSchema*  pColSchema = &pSchema[j];
+        if(strcmp(pColSchema->name, tFields[i].name) == 0){
+          if (*fields != pColSchema->type && *(int32_t*)(fields + sizeof(int8_t)) != pColSchema->bytes) {
+            uError("type or bytes not equal");
+            ret = TSDB_CODE_INVALID_PARA;
+            goto end;
+          }
+
+          int8_t* offset = pStart;
+          if (IS_VAR_DATA_TYPE(pColSchema->type)) {
+            pStart += numOfRows * sizeof(int32_t);
+          } else {
+            pStart += BitmapLen(numOfRows);
+          }
+          char* pData = pStart;
+
+          SColData* pCol = taosArrayGet(pTableCxt->pData->aCol, j);
+          tColDataAddValueByDataBlock(pCol, pColSchema->type, pColSchema->bytes, numOfRows, offset, pData);
+          fields += sizeof(int8_t) + sizeof(int32_t);
+          if (needChangeLength) {
+            pStart += htonl(colLength[i]);
+          } else {
+            pStart += colLength[i];
+          }
+          boundInfo->pColIndex[j] = -1;
+          break;
+        }
+      }
+
+    }
+
+    for (int c = 0; c < boundInfo->numOfBound; ++c) {
+      if( boundInfo->pColIndex[c] != -1){
+        SColData* pCol = taosArrayGet(pTableCxt->pData->aCol, c);
+        tColDataAddValueByDataBlock(pCol, 0, 0, numOfRows, NULL, NULL);
+      }else{
+        boundInfo->pColIndex[c] = c;  // restore for next block
       }
-    }else{
-      tColDataAddValueByDataBlock(pCol, pColSchema->type, pColSchema->bytes, numOfRows, NULL, NULL);
     }
   }
 

source/libs/scalar/src/sclfunc.c

@@ -1755,7 +1755,11 @@ int32_t winEndTsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
 int32_t qTbnameFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
   char* p = colDataGetVarData(pInput->columnData, 0);
 
-  colDataSetNItems(pOutput->columnData, pOutput->numOfRows, p, pInput->numOfRows);
+  int32_t code = colDataSetNItems(pOutput->columnData, pOutput->numOfRows, p, pInput->numOfRows, true);
+  if (code) {
+    return code;
+  }
+  
   pOutput->numOfRows += pInput->numOfRows;
   return TSDB_CODE_SUCCESS;
 }

source/libs/stream/src/stream.c

@@ -275,7 +275,7 @@ int32_t streamProcessRetrieveReq(SStreamTask* pTask, SStreamRetrieveReq* pReq, S
   return 0;
 }
 
-int32_t streamProcessRetrieveRsp(SStreamTask* pTask, SStreamRetrieveRsp* pRsp) {
-  //
-  return 0;
-}
+// int32_t streamProcessRetrieveRsp(SStreamTask* pTask, SStreamRetrieveRsp* pRsp) {
+//   //
+//   return 0;
+// }

source/libs/sync/src/syncAppendEntries.c

@@ -105,6 +105,7 @@ int32_t syncNodeOnAppendEntries(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
   SRpcMsg            rpcRsp = {0};
   bool               accepted = false;
   SSyncRaftEntry*    pEntry = NULL;
+  bool               resetElect = false;
 
   // if already drop replica, do not process
   if (!syncNodeInRaftGroup(ths, &(pMsg->srcId))) {
@@ -137,7 +138,7 @@ int32_t syncNodeOnAppendEntries(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
   }
 
   syncNodeStepDown(ths, pMsg->term);
-  syncNodeResetElectTimer(ths);
+  resetElect = true;
 
   if (pMsg->dataLen < sizeof(SSyncRaftEntry)) {
     sError("vgId:%d, incomplete append entries received. prev index:%" PRId64 ", term:%" PRId64 ", datalen:%d",
@@ -184,10 +185,9 @@ _SEND_RESPONSE:
   // commit index, i.e. leader notice me
   if (syncLogBufferCommit(ths->pLogBuf, ths, ths->commitIndex) < 0) {
     sError("vgId:%d, failed to commit raft fsm log since %s.", ths->vgId, terrstr());
-    goto _out;
   }
 
-_out:
+  if (resetElect) syncNodeResetElectTimer(ths);
   return 0;
 
 _IGNORE:

source/libs/sync/src/syncElection.c

@@ -115,6 +115,5 @@ int32_t syncNodeElect(SSyncNode* pSyncNode) {
   ASSERT(ret == 0);
 
   syncNodeResetElectTimer(pSyncNode);
-
   return ret;
 }

source/libs/sync/src/syncMain.c

@@ -182,6 +182,9 @@ int32_t syncProcessMsg(int64_t rid, SRpcMsg* pMsg) {
     case TDMT_SYNC_TIMEOUT:
       code = syncNodeOnTimeout(pSyncNode, pMsg);
       break;
+    case TDMT_SYNC_TIMEOUT_ELECTION:
+      code = syncNodeOnTimeout(pSyncNode, pMsg);
+      break;
     case TDMT_SYNC_CLIENT_REQUEST:
       code = syncNodeOnClientRequest(pSyncNode, pMsg, NULL);
       break;
@@ -1593,8 +1596,8 @@ void syncNodeBecomeFollower(SSyncNode* pSyncNode, const char* debugStr) {
   pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER;
   syncNodeStopHeartbeatTimer(pSyncNode);
 
-  // reset elect timer
-  syncNodeResetElectTimer(pSyncNode);
+  // trace log
+  sNTrace(pSyncNode, "become follower %s", debugStr);
 
   // send rsp to client
   syncNodeLeaderChangeRsp(pSyncNode);
@@ -1610,8 +1613,8 @@ void syncNodeBecomeFollower(SSyncNode* pSyncNode, const char* debugStr) {
   // reset log buffer
   syncLogBufferReset(pSyncNode->pLogBuf, pSyncNode);
 
-  // trace log
-  sNTrace(pSyncNode, "become follower %s", debugStr);
+  // reset elect timer
+  syncNodeResetElectTimer(pSyncNode);
 }
 
 // TLA+ Spec
@@ -2277,6 +2280,7 @@ static int32_t syncNodeAppendNoopOld(SSyncNode* ths) {
 
 int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
   SyncHeartbeat* pMsg = pRpcMsg->pCont;
+  bool           resetElect = false;
 
   const STraceId* trace = &pRpcMsg->info.traceId;
   char            tbuf[40] = {0};
@@ -2300,12 +2304,11 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
 
   if (pMsg->term == currentTerm && ths->state != TAOS_SYNC_STATE_LEADER) {
     syncIndexMgrSetRecvTime(ths->pNextIndex, &(pMsg->srcId), tsMs);
+    resetElect = true;
 
-    syncNodeResetElectTimer(ths);
     ths->minMatchIndex = pMsg->minMatchIndex;
 
     if (ths->state == TAOS_SYNC_STATE_FOLLOWER) {
-      // syncNodeFollowerCommit(ths, pMsg->commitIndex);
       SRpcMsg rpcMsgLocalCmd = {0};
       (void)syncBuildLocalCmd(&rpcMsgLocalCmd, ths->vgId);
 
@@ -2328,7 +2331,6 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
   }
 
   if (pMsg->term >= currentTerm && ths->state != TAOS_SYNC_STATE_FOLLOWER) {
-    // syncNodeStepDown(ths, pMsg->term);
     SRpcMsg rpcMsgLocalCmd = {0};
     (void)syncBuildLocalCmd(&rpcMsgLocalCmd, ths->vgId);
 
@@ -2348,15 +2350,10 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
     }
   }
 
-  /*
-    // htonl
-    SMsgHead* pHead = rpcMsg.pCont;
-    pHead->contLen = htonl(pHead->contLen);
-    pHead->vgId = htonl(pHead->vgId);
-  */
-
   // reply
   syncNodeSendMsgById(&pMsgReply->destId, ths, &rpcMsg);
+
+  if (resetElect) syncNodeResetElectTimer(ths);
   return 0;
 }
 

source/libs/sync/src/syncMessage.c

@@ -22,7 +22,7 @@ int32_t syncBuildTimeout(SRpcMsg* pMsg, ESyncTimeoutType timeoutType, uint64_t l
                          SSyncNode* pNode) {
   int32_t bytes = sizeof(SyncTimeout);
   pMsg->pCont = rpcMallocCont(bytes);
-  pMsg->msgType = TDMT_SYNC_TIMEOUT;
+  pMsg->msgType = (timeoutType == SYNC_TIMEOUT_ELECTION) ? TDMT_SYNC_TIMEOUT_ELECTION : TDMT_SYNC_TIMEOUT;
   pMsg->contLen = bytes;
   if (pMsg->pCont == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -31,7 +31,7 @@ int32_t syncBuildTimeout(SRpcMsg* pMsg, ESyncTimeoutType timeoutType, uint64_t l
 
   SyncTimeout* pTimeout = pMsg->pCont;
   pTimeout->bytes = bytes;
-  pTimeout->msgType = TDMT_SYNC_TIMEOUT;
+  pTimeout->msgType = pMsg->msgType;
   pTimeout->vgId = pNode->vgId;
   pTimeout->timeoutType = timeoutType;
   pTimeout->logicClock = logicClock;

source/libs/sync/src/syncRequestVote.c

@@ -89,6 +89,7 @@ static bool syncNodeOnRequestVoteLogOK(SSyncNode* ths, SyncRequestVote* pMsg) {
 int32_t syncNodeOnRequestVote(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
   int32_t          ret = 0;
   SyncRequestVote* pMsg = pRpcMsg->pCont;
+  bool             resetElect = false;
 
   // if already drop replica, do not process
   if (!syncNodeInRaftGroup(ths, &pMsg->srcId)) {
@@ -115,7 +116,7 @@ int32_t syncNodeOnRequestVote(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
     syncNodeStepDown(ths, currentTerm);
 
     // forbid elect for this round
-    syncNodeResetElectTimer(ths);
+    resetElect = true;
   }
 
   // send msg
@@ -134,5 +135,7 @@ int32_t syncNodeOnRequestVote(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
   syncLogRecvRequestVote(ths, pMsg, pReply->voteGranted, "");
   syncLogSendRequestVoteReply(ths, pReply, "");
   syncNodeSendMsgById(&pReply->destId, ths, &rpcMsg);
+
+  if (resetElect) syncNodeResetElectTimer(ths);
   return 0;
 }

source/libs/sync/src/syncSnapshot.c

@@ -798,7 +798,6 @@ int32_t syncNodeOnSnapshot(SSyncNode *pSyncNode, const SRpcMsg *pRpcMsg) {
   if (pMsg->term > raftStoreGetTerm(pSyncNode)) {
     syncNodeStepDown(pSyncNode, pMsg->term);
   }
-  syncNodeResetElectTimer(pSyncNode);
 
   // state, term, seq/ack
   int32_t code = 0;
@@ -840,6 +839,7 @@ int32_t syncNodeOnSnapshot(SSyncNode *pSyncNode, const SRpcMsg *pRpcMsg) {
     code = -1;
   }
 
+  syncNodeResetElectTimer(pSyncNode);
   return code;
 }
 

source/libs/sync/src/syncTimeout.c

@@ -120,9 +120,6 @@ int32_t syncNodeOnTimeout(SSyncNode* ths, const SRpcMsg* pRpc) {
     if (atomic_load_64(&ths->pingTimerLogicClockUser) <= pMsg->logicClock) {
       ++(ths->pingTimerCounter);
 
-      // syncNodePingAll(ths);
-      // syncNodePingPeers(ths);
-
       syncNodeTimerRoutine(ths);
     }
 
@@ -138,8 +135,6 @@ int32_t syncNodeOnTimeout(SSyncNode* ths, const SRpcMsg* pRpc) {
       ++(ths->heartbeatTimerCounter);
       sTrace("vgId:%d, sync timer, type:replicate count:%" PRIu64 ", lc-user:%" PRIu64, ths->vgId,
              ths->heartbeatTimerCounter, ths->heartbeatTimerLogicClockUser);
-
-      // syncNodeReplicate(ths, true);
     }
 
   } else {

source/util/src/tsimplehash.c

@@ -28,19 +28,23 @@
 
 #define HASH_INDEX(v, c) ((v) & ((c)-1))
 
-#define FREE_HASH_NODE(_n)   \
-  do {                       \
-    taosMemoryFreeClear(_n); \
+#define FREE_HASH_NODE(_n, fp) \
+  do {                         \
+    if (fp) {                  \
+      fp((_n)->data);          \
+    }                          \
+    taosMemoryFreeClear(_n);   \
   } while (0);
 
 struct SSHashObj {
-  SHNode    **hashList;
-  size_t      capacity;  // number of slots
-  int64_t     size;      // number of elements in hash table
-  _hash_fn_t  hashFp;    // hash function
-  _equal_fn_t equalFp;   // equal function
-  SArray*     pHashNodeBuf;// hash node allocation buffer, 1k size of each page by default
-  int32_t     offset;      // allocation offset in current page
+  SHNode        **hashList;
+  size_t          capacity;  // number of slots
+  int64_t         size;      // number of elements in hash table
+  _hash_fn_t      hashFp;    // hash function
+  _equal_fn_t     equalFp;   // equal function
+  _hash_free_fn_t freeFp;    // free function
+  SArray         *pHashNodeBuf;  // hash node allocation buffer, 1k size of each page by default
+  int32_t         offset;        // allocation offset in current page
 };
 
 static FORCE_INLINE int32_t taosHashCapacity(int32_t length) {
@@ -71,7 +75,7 @@ SSHashObj *tSimpleHashInit(size_t capacity, _hash_fn_t fn) {
   pHashObj->capacity = taosHashCapacity((int32_t)capacity);
   pHashObj->equalFp = memcmp;
 
-  pHashObj->pHashNodeBuf = taosArrayInit(10, sizeof(void*));
+  pHashObj->freeFp = NULL;
   pHashObj->offset = 0;
   pHashObj->size = 0;
   
@@ -92,6 +96,10 @@ int32_t tSimpleHashGetSize(const SSHashObj *pHashObj) {
   return (int32_t) pHashObj->size;
 }
 
+void tSimpleHashSetFreeFp(SSHashObj* pHashObj, _hash_free_fn_t freeFp) {
+  pHashObj->freeFp = freeFp;
+}
+
 static void* doInternalAlloc(SSHashObj* pHashObj, int32_t size) {
 #if 0
   void** p = taosArrayGetLast(pHashObj->pHashNodeBuf);
@@ -306,7 +314,8 @@ int32_t tSimpleHashRemove(SSHashObj *pHashObj, const void *key, size_t keyLen) {
       } else {
         pPrev->next = pNode->next;
       }
-      FREE_HASH_NODE(pNode);
+
+      FREE_HASH_NODE(pNode, pHashObj->freeFp);
       pHashObj->size -= 1;
       code = TSDB_CODE_SUCCESS;
       break;
@@ -341,7 +350,7 @@ int32_t tSimpleHashIterateRemove(SSHashObj *pHashObj, const void *key, size_t ke
         *pIter = pPrev ? GET_SHASH_NODE_DATA(pPrev) : NULL;
       }
 
-      FREE_HASH_NODE(pNode);
+      FREE_HASH_NODE(pNode, pHashObj->freeFp);
       pHashObj->size -= 1;
       break;
     }
@@ -370,14 +379,13 @@ void tSimpleHashClear(SSHashObj *pHashObj) {
 
     while (pNode) {
       pNext = pNode->next;
-      FREE_HASH_NODE(pNode);
+      FREE_HASH_NODE(pNode, pHashObj->freeFp);
       pNode = pNext;
     }
 
     pHashObj->hashList[i] = NULL;
   }
 
-  taosArrayClearEx(pHashObj->pHashNodeBuf, destroyItems);
   pHashObj->offset = 0;
   pHashObj->size = 0;
 }
@@ -388,7 +396,6 @@ void tSimpleHashCleanup(SSHashObj *pHashObj) {
   }
 
   tSimpleHashClear(pHashObj);
-  taosArrayDestroy(pHashObj->pHashNodeBuf);
   taosMemoryFreeClear(pHashObj->hashList);
   taosMemoryFree(pHashObj);
 }

source/util/src/tversion.c

@@ -26,8 +26,9 @@ int32_t taosVersionStrToInt(const char *vstr, int32_t *vint) {
   int32_t vnum[4] = {0};
   int32_t len = strlen(vstr);
   char    tmp[16] = {0};
+  int32_t vpos = 0;
 
-  for (int32_t spos = 0, tpos = 0, vpos = 0; spos < len && vpos < 4; ++spos) {
+  for (int32_t spos = 0, tpos = 0; spos < len && vpos < 4; ++spos) {
     if (vstr[spos] != '.') {
       tmp[spos - tpos] = vstr[spos];
     } else {
@@ -38,6 +39,10 @@ int32_t taosVersionStrToInt(const char *vstr, int32_t *vint) {
     }
   }
 
+  if ('\0' != tmp[0] && vpos < 4) {
+    vnum[vpos] = atoi(tmp);
+  }
+
   if (vnum[0] <= 0) {
     terrno = TSDB_CODE_INVALID_VERSION_STRING;
     return -1;
@@ -66,16 +71,16 @@ int32_t taosCheckVersionCompatible(int32_t clientVer, int32_t serverVer, int32_t
     case 4:
       break;
     case 3:
-      clientVer %= 100;
-      serverVer %= 100;
+      clientVer /= 100;
+      serverVer /= 100;
       break;
     case 2:
-      clientVer %= 10000;
-      serverVer %= 10000;
+      clientVer /= 10000;
+      serverVer /= 10000;
       break;
     case 1:
-      clientVer %= 1000000;
-      serverVer %= 1000000;
+      clientVer /= 1000000;
+      serverVer /= 1000000;
       break;
     default:
       terrno = TSDB_CODE_INVALID_VERSION_NUMBER;

tests/develop-test/5-taos-tools/taosbenchmark/json/sml_telnet_tcp.json

@@ -16,7 +16,7 @@
   "num_of_records_per_req": 10,
   "databases": [{
     "dbinfo": {
-      "name": "db",
+      "name": "opentsdb_telnet",
       "drop": "yes"
     },
     "super_tables": [{

tests/parallel_test/cases.task

@@ -14,6 +14,8 @@
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_math.py -Q 2
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_time.py -Q 2
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery.py -Q 2
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/columnLenUpdated.py -Q 2
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/columnLenUpdated.py -Q 3
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery.py -Q 3
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_str.py -Q 3
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_math.py -Q 3
@@ -22,6 +24,7 @@
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_str.py -Q 4
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_math.py -Q 4
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_time.py -Q 4
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/columnLenUpdated.py -Q 4
 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqShow.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDropStb.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/subscribeStb0.py
@@ -1103,9 +1106,9 @@
 ,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/json_tag.py
 ,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/query_json.py
 ,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/sample_csv_json.py
-#,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/sml_json_alltypes.py
+,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/sml_json_alltypes.py
 ,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/taosdemoTestQueryWithJson.py -R
-#,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/telnet_tcp.py -R
+,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/telnet_tcp.py -R
 
 #docs-examples test
 ,,n,docs-examples-test,bash python.sh

tests/script/tsim/stream/distributeInterval0.sim

@@ -284,7 +284,7 @@ sql create table ccc using st tags(3,2,2);
 sql create table ddd using st tags(4,2,2);
 
 
-sql create stream streams1 ignore expired 0 fill_history 0 watermark 3s into streamst subtable(c) as select  _wstart, c , count(*) c1  from st partition by c interval(1s) ;
+sql create stream streams1 ignore expired 0 fill_history 0 watermark 3s into streamst subtable(c) as select  _wstart, c , count(*) c1, last_row(b) c2  from st partition by c interval(1s) ;
 
 sql insert into aaa values(1648791221001,2,2,"/a1/aa/aa");
 sql insert into bbb values(1648791221001,2,2,"/a1/aa/aa");

tests/script/tsim/stream/distributeIntervalRetrive0.sim

@@ -3,7 +3,6 @@ system sh/deploy.sh -n dnode1 -i 1
 system sh/deploy.sh -n dnode2 -i 2
 
 system sh/exec.sh -n dnode1 -s start 
-#==system sh/exec.sh -n dnode1 -s start -v
 
 sleep 50
 sql connect
@@ -11,7 +10,6 @@ sql connect
 sql create dnode $hostname2 port 7200
 
 system sh/exec.sh -n dnode2 -s start
-#==system sh/exec.sh -n dnode2 -s start -v
 
 print ===== step1
 $x = 0
@@ -37,14 +35,14 @@ endi
 
 print ===== step2
 
-sql create database test  vgroups 4;
+sql create database test  vgroups 10;
 sql use test;
 sql create stable st(ts timestamp, a int, b int , c int, d double) tags(ta int,tb int,tc int);
 sql create table ts1 using st tags(1,1,1);
 sql create table ts2 using st tags(2,2,2);
 sql create table ts3 using st tags(3,2,2);
 sql create table ts4 using st tags(4,2,2);
-sql create stream stream_t1 trigger at_once IGNORE EXPIRED 0 into streamtST1 as select  _wstart, count(*) c1, sum(a) c3 , max(b)  c4, min(c) c5 from st interval(10s);
+sql create stream stream_t1 trigger at_once IGNORE EXPIRED 0 delete_mark 10s into streamtST1 as select  _wstart, count(*) c1, sum(a) c3 , max(b)  c4, min(c) c5 from st interval(10s);
 
 sleep 1000
 
@@ -235,6 +233,29 @@ endi
 
 print loop3 over
 
+sql insert into ts1 values(1648791200001,1,12,3,1.0);
+sql insert into ts2 values(1648791200001,1,12,3,1.0);
+sql insert into ts3 values(1648791200001,1,12,3,1.0);
+sql insert into ts4 values(1648791200001,1,12,3,1.0);
+
+$loop_count = 0
+loop31:
+sleep 1000
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+sql select * from streamtST1;
+
+if $rows <= 4 then
+  print =====rows=$rows
+  goto loop31
+endi
+
+print loop31 over
+
 
 sql drop stream if exists streams1;
 sql drop database if exists test1;
@@ -243,7 +264,7 @@ sql use test1;
 sql create stable st(ts timestamp,a int,b int,c int) tags(ta int,tb int,tc int);
 sql create table t1 using st tags(1,1,1);
 sql create table t2 using st tags(2,2,2);
-sql create stream streams1 trigger at_once IGNORE EXPIRED 0  into streamt1 as select  _wstart as c0, count(*) c1, count(a) c2  from st interval(10s) ;
+sql create stream streams1 trigger at_once IGNORE EXPIRED 0 delete_mark 20s into streamt1 as select  _wstart as c0, count(*) c1, count(a) c2  from st interval(10s) ;
 
 sql insert into t1 values(1648791211000,1,2,3);
 

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

+
+import taos
+import sys
+import time
+import socket
+import os
+import platform
+if platform.system().lower() == 'windows':
+    import wexpect as taosExpect
+else:
+    import pexpect as taosExpect
+
+from util.log import *
+from util.sql import *
+from util.cases import *
+from util.dnodes import *
+
+def taos_command (buildPath, key, value, expectString, sqlString=''):
+    if len(key) == 0:
+        tdLog.exit("taos test key is null!")
+
+    if platform.system().lower() == 'windows':
+        taosCmd = buildPath + '\\build\\bin\\taos.exe '
+        taosCmd = taosCmd.replace('\\','\\\\')
+    else:
+        taosCmd = buildPath + '/build/bin/taos '
+
+    cfgPath = buildPath + "/../sim/psim/cfg"
+    taosCmd = taosCmd + ' -c' + cfgPath + ' -' + key
+    if len(value) != 0:
+        taosCmd = taosCmd + ' ' + value
+
+    tdLog.info ("taos cmd: %s" % taosCmd)
+
+    child = taosExpect.spawn(taosCmd, timeout=20)
+    #output = child.readline()
+    #print (output.decode())
+    if len(expectString) != 0:
+        i = child.expect([expectString, taosExpect.TIMEOUT, taosExpect.EOF], timeout=20)
+    else:
+        i = child.expect([taosExpect.TIMEOUT, taosExpect.EOF], timeout=20)
+
+    if platform.system().lower() == 'windows':
+        retResult = child.before
+    else:
+        retResult = child.before.decode()
+    print(retResult)
+    #print(child.after.decode())
+    if i == 0:
+        print ('taos login success! Here can run sql, taos> ')
+        return  "TAOS_OK"
+    else:
+        return "TAOS_FAIL"
+
+class TDTestCase:
+    #updatecfgDict = {'clientCfg': {'serverPort': 7080, 'firstEp': 'trd02:7080', 'secondEp':'trd02:7080'},\
+    #                 'serverPort': 7080, 'firstEp': 'trd02:7080'}
+    hostname = socket.gethostname()
+    if (platform.system().lower() == 'windows' and not tdDnodes.dnodes[0].remoteIP == ""):
+        try:
+            config = eval(tdDnodes.dnodes[0].remoteIP)
+            hostname = config["host"]
+        except Exception:
+            hostname = tdDnodes.dnodes[0].remoteIP
+    serverPort = '7080'
+    rpcDebugFlagVal = '143'
+    clientCfgDict = {'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}
+    clientCfgDict["serverPort"]    = serverPort
+    clientCfgDict["firstEp"]       = hostname + ':' + serverPort
+    clientCfgDict["secondEp"]      = hostname + ':' + serverPort
+    clientCfgDict["rpcDebugFlag"]  = rpcDebugFlagVal
+    clientCfgDict["fqdn"] = hostname
+
+    updatecfgDict = {'clientCfg': {}, 'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}
+    updatecfgDict["clientCfg"]  = clientCfgDict
+    updatecfgDict["serverPort"] = serverPort
+    updatecfgDict["firstEp"]    = hostname + ':' + serverPort
+    updatecfgDict["secondEp"]   = hostname + ':' + serverPort
+    updatecfgDict["fqdn"] = hostname
+
+    print ("===================: ", updatecfgDict)
+
+    def init(self, conn, logSql, replicaVar=1):
+        self.replicaVar = int(replicaVar)
+        tdLog.debug(f"start to excute {__file__}")
+        tdSql.init(conn.cursor())
+
+    def getBuildPath(self):
+        selfPath = os.path.dirname(os.path.realpath(__file__))
+
+        if ("community" in selfPath):
+            projPath = selfPath[:selfPath.find("community")]
+        else:
+            projPath = selfPath[:selfPath.find("tests")]
+
+        for root, dirs, files in os.walk(projPath):
+            if ("taosd" in files or "taosd.exe" in files):
+                rootRealPath = os.path.dirname(os.path.realpath(root))
+                if ("packaging" not in rootRealPath):
+                    buildPath = root[:len(root) - len("/build/bin")]
+                    break
+        return buildPath
+
+    def run(self):  # sourcery skip: extract-duplicate-method, remove-redundant-fstring
+        tdSql.prepare()
+        # time.sleep(2)
+        tdSql.query("create user testpy pass 'testpy'")
+
+        buildPath = self.getBuildPath()
+        if (buildPath == ""):
+            tdLog.exit("taosd not found!")
+        else:
+            tdLog.info("taosd found in %s" % buildPath)
+        cfgPath = buildPath + "/../sim/psim/cfg"
+        tdLog.info("cfgPath: %s" % cfgPath)
+
+        checkNetworkStatus = ['0: unavailable', '1: network ok', '2: service ok', '3: service degraded', '4: exiting']
+        netrole            = ['client', 'server']
+
+        keyDict = {'h':'', 'P':'6030', 'p':'testpy', 'u':'testpy', 'a':'', 'A':'', 'c':'', 'C':'', 's':'', 'r':'', 'f':'', \
+                   'k':'', 't':'', 'n':'', 'l':'1024', 'N':'100', 'V':'', 'd':'db', 'w':'30', '-help':'', '-usage':'', '?':''}
+
+        keyDict['h'] = self.hostname
+        keyDict['c'] = cfgPath
+        keyDict['P'] = self.serverPort
+
+        tdSql.query("drop database if exists db1")
+        tdSql.query("create database if not exists db1 vgroups 1")
+        tdSql.query("use db1")
+        tdSql.query("create table tba (ts timestamp, f1 binary(2))")
+        tdSql.query("insert into tba values (now, '22')")
+        tdSql.query("select * from tba")
+        tdSql.checkData(0, 1, '22')
+
+        keyDict['s'] = "\"alter table db1.tba modify column f1 binary(5) \""
+        retCode = taos_command(buildPath, "s", keyDict['s'], "Query OK", '')
+        if retCode != "TAOS_OK":
+            tdLog.exit("taos -s fail")
+
+        keyDict['s'] = "\"insert into db1.tba values (now, '55555')\""
+        retCode = taos_command(buildPath, "s", keyDict['s'], "Insert OK", '')
+        if retCode != "TAOS_OK":
+            tdLog.exit("taos -s fail")
+
+        tdSql.query("select * from tba order by ts")
+        tdSql.checkData(0, 1, '22')
+        tdSql.checkData(1, 1, '55555')
+
+
+        tdSql.query("create table stb (ts timestamp, f1 int) tags (tg1 binary(2))")
+        tdSql.query("create table tb1 using stb tags('bb')")
+        tdSql.query("insert into tb1 values (now, 2)")
+        tdSql.query("select count(*) from stb group by tg1")
+        tdSql.checkData(0, 0, 1)
+
+        keyDict['s'] = "\"alter table db1.stb modify tag tg1 binary(5) \""
+        retCode = taos_command(buildPath, "s", keyDict['s'], "Query OK", '')
+        if retCode != "TAOS_OK":
+            tdLog.exit("taos -s fail")
+
+        keyDict['s'] = "\"create table db1.tb2 using db1.stb tags('bbbbb')\""
+        retCode = taos_command(buildPath, "s", keyDict['s'], "Create OK", '')
+        if retCode != "TAOS_OK":
+            tdLog.exit("taos -s fail")
+
+        keyDict['s'] = "\"insert into db1.tb2 values (now, 2)\""
+        retCode = taos_command(buildPath, "s", keyDict['s'], "Insert OK", '')
+        if retCode != "TAOS_OK":
+            tdLog.exit("taos -s fail")
+
+        tdSql.query("select count(*) from stb group by tg1")
+        tdSql.checkData(0, 0, 1)
+        tdSql.checkData(1, 0, 1)
+
+
+    def stop(self):
+        tdSql.close()
+        tdLog.success(f"{__file__} successfully executed")
+
+tdCases.addLinux(__file__, TDTestCase())
+tdCases.addWindows(__file__, TDTestCase())