Merge branch '3.0' into feature/TD-14481-3.0

examples/c/stream_demo.c

@@ -88,9 +88,9 @@ int32_t create_stream() {
   /*const char* sql = "select min(k), max(k), sum(k) as sum_of_k from st1";*/
   /*const char* sql = "select sum(k) from tu1 interval(10m)";*/
   /*pRes = tmq_create_stream(pConn, "stream1", "out1", sql);*/
-  pRes = taos_query(pConn,
-                    "create stream stream1 trigger at_once into abc2.outstb as select _wstartts, sum(k) from st1  "
-                    "partition by tbname interval(10m) ");
+  pRes = taos_query(
+      pConn,
+      "create stream stream1 trigger at_once into abc1.outstb as select _wstartts, sum(k) from st1 interval(10m) ");
   if (taos_errno(pRes) != 0) {
     printf("failed to create stream stream1, reason:%s\n", taos_errstr(pRes));
     return -1;
@@ -107,11 +107,4 @@ int main(int argc, char* argv[]) {
     code = init_env();
   }
   create_stream();
-#if 0
-  tmq_t*      tmq = build_consumer();
-  tmq_list_t* topic_list = build_topic_list();
-  /*perf_loop(tmq, topic_list);*/
-  /*basic_consume_loop(tmq, topic_list);*/
-  sync_consume_loop(tmq, topic_list);
-#endif
 }

include/common/tmsgcb.h

@@ -34,7 +34,6 @@ typedef enum {
   WRITE_QUEUE,
   APPLY_QUEUE,
   SYNC_QUEUE,
-  MERGE_QUEUE,
   QUEUE_MAX,
 } EQueueType;
 

include/dnode/snode/snode.h

@@ -16,8 +16,8 @@
 #ifndef _TD_SNODE_H_
 #define _TD_SNODE_H_
 
-#include "tmsgcb.h"
 #include "tmsg.h"
+#include "tmsgcb.h"
 #include "trpc.h"
 
 #ifdef __cplusplus
@@ -68,8 +68,8 @@ int32_t sndGetLoad(SSnode *pSnode, SSnodeLoad *pLoad);
  * @param pMsg The request message
  * @param pRsp The response message
  */
-void sndProcessUMsg(SSnode *pSnode, SRpcMsg *pMsg);
-void sndProcessSMsg(SSnode *pSnode, SRpcMsg *pMsg);
+int32_t sndProcessUMsg(SSnode *pSnode, SRpcMsg *pMsg);
+int32_t sndProcessSMsg(SSnode *pSnode, SRpcMsg *pMsg);
 
 #ifdef __cplusplus
 }

include/libs/planner/planner.h

@@ -36,7 +36,6 @@ typedef struct SPlanContext {
   int64_t  watermark;
   char*    pMsg;
   int32_t  msgLen;
-  // double   filesFactor;
 } SPlanContext;
 
 // Create the physical plan for the query, according to the AST.

include/libs/stream/tstream.h

@@ -152,7 +152,7 @@ void*   streamDataBlockDecode(const void* buf, SStreamDataBlock* pInput);
 typedef struct {
   char* qmsg;
   // followings are not applicable to encoder and decoder
-  void* inputHandle;
+  // void* inputHandle;
   void* executor;
 } STaskExec;
 
@@ -240,12 +240,13 @@ struct SStreamTask {
   int8_t  inputType;
   int8_t  status;
 
-  int8_t  sourceType;
   int8_t  execType;
   int8_t  sinkType;
   int8_t  dispatchType;
   int16_t dispatchMsgType;
 
+  int8_t dataScan;
+
   // node info
   int32_t childId;
   int32_t nodeId;

source/common/src/tglobal.c

@@ -185,7 +185,7 @@ char     tsCompressor[32] = "ZSTD_COMPRESSOR";  // ZSTD_COMPRESSOR or GZIP_COMPR
 bool tsStartUdfd = true;
 
 // internal
-int32_t tsTransPullupInterval = 6;
+int32_t tsTransPullupInterval = 2;
 int32_t tsMqRebalanceInterval = 2;
 
 void taosAddDataDir(int32_t index, char *v1, int32_t level, int32_t primary) {

source/dnode/mgmt/mgmt_snode/src/smHandle.c

@@ -95,9 +95,12 @@ SArray *smGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_MON_SM_INFO, smPutNodeMsgToMonitorQueue, 0) == NULL) goto _OVER;
 
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DEPLOY, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TASK_DROP, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RUN, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DISPATCH, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DISPATCH_RSP, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RECOVER, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RECOVER_RSP, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER;
 
   code = 0;
 _OVER:

source/dnode/mgmt/mgmt_snode/src/smWorker.c

@@ -55,7 +55,9 @@ static void smProcessUniqueQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t num
     taosGetQitem(qall, (void **)&pMsg);
 
     dTrace("msg:%p, get from snode-unique queue", pMsg);
-    sndProcessUMsg(pMgmt->pSnode, pMsg);
+    if (sndProcessUMsg(pMgmt->pSnode, pMsg) < 0) {
+      ASSERT(0);
+    }
 
     dTrace("msg:%p, is freed", pMsg);
     rpcFreeCont(pMsg->pCont);
@@ -67,7 +69,9 @@ static void smProcessSharedQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
   SSnodeMgmt *pMgmt = pInfo->ahandle;
 
   dTrace("msg:%p, get from snode-shared queue", pMsg);
-  sndProcessSMsg(pMgmt->pSnode, pMsg);
+  if (sndProcessSMsg(pMgmt->pSnode, pMsg) < 0) {
+    ASSERT(0);
+  }
 
   dTrace("msg:%p, is freed", pMsg);
   rpcFreeCont(pMsg->pCont);

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

@@ -170,10 +170,6 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp
       dTrace("vgId:%d, msg:%p put into vnode-sync queue", pVnode->vgId, pMsg);
       taosWriteQitem(pVnode->pSyncQ, pMsg);
       break;
-    case MERGE_QUEUE:
-      dTrace("vgId:%d, msg:%p put into vnode-merge queue", pVnode->vgId, pMsg);
-      taosWriteQitem(pVnode->pMergeQ, pMsg);
-      break;
     case APPLY_QUEUE:
       dTrace("vgId:%d, msg:%p put into vnode-apply queue", pVnode->vgId, pMsg);
       taosWriteQitem(pVnode->pApplyQ, pMsg);
@@ -196,8 +192,6 @@ int32_t vmPutMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsg
 
 int32_t vmPutMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, FETCH_QUEUE); }
 
-int32_t vmPutMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, MERGE_QUEUE); }
-
 int32_t vmPutMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   dTrace("msg:%p, put into vnode-mgmt queue", pMsg);
   taosWriteQitem(pMgmt->mgmtWorker.queue, pMsg);
@@ -243,9 +237,6 @@ int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) {
       case FETCH_QUEUE:
         size = taosQueueItemSize(pVnode->pFetchQ);
         break;
-      case MERGE_QUEUE:
-        size = taosQueueItemSize(pVnode->pMergeQ);
-        break;
       default:
         break;
     }

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

@@ -65,7 +65,6 @@ int32_t mndConvertRsmaTask(char** pDst, int32_t* pDstLen, const char* ast, int64
       .rSmaQuery = true,
       .triggerType = triggerType,
       .watermark = watermark,
-      /*.filesFactor = filesFactor,*/
   };
 
   if (qCreateQueryPlan(&cxt, &pPlan, NULL) < 0) {
@@ -270,7 +269,6 @@ int32_t mndAddShuffleSinkTasksToStream(SMnode* pMnode, STrans* pTrans, SStreamOb
     pTask->epSet = mndGetVgroupEpset(pMnode, pVgroup);
 
     // source
-    pTask->sourceType = TASK_SOURCE__MERGE;
     pTask->inputType = TASK_INPUT_TYPE__DATA_BLOCK;
 
     // exec
@@ -316,7 +314,6 @@ int32_t mndAddFixedSinkTaskToStream(SMnode* pMnode, STrans* pTrans, SStreamObj*
 #endif
   pTask->epSet = mndGetVgroupEpset(pMnode, &pStream->fixedSinkVg);
   // source
-  pTask->sourceType = TASK_SOURCE__MERGE;
   pTask->inputType = TASK_INPUT_TYPE__DATA_BLOCK;
 
   // exec
@@ -427,6 +424,8 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
       SStreamTask* pTask = tNewSStreamTask(pStream->uid);
       mndAddTaskToTaskSet(taskSourceLevel, pTask);
 
+      pTask->dataScan = 1;
+
       // input
       pTask->inputType = TASK_INPUT_TYPE__SUMBIT_BLOCK;
 
@@ -470,6 +469,8 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
       SStreamTask* pTask = tNewSStreamTask(pStream->uid);
       mndAddTaskToTaskSet(taskOneLevel, pTask);
 
+      pTask->dataScan = 1;
+
       // input
       pTask->inputType = TASK_INPUT_TYPE__SUMBIT_BLOCK;
 

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

@@ -68,8 +68,9 @@ void mndSyncCommitMsg(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbM
       mndTransExecute(pMnode, pTrans);
       mndReleaseTrans(pMnode, pTrans);
     }
-
+#if 0
     sdbWriteFile(pMnode->pSdb, SDB_WRITE_DELTA);
+#endif
   }
 }
 

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

@@ -804,7 +804,7 @@ static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
       sendRsp = true;
     }
   } else {
-    if (pTrans->stage == TRN_STAGE_REDO_ACTION && pTrans->failedTimes > 2) {
+    if (pTrans->stage == TRN_STAGE_REDO_ACTION && pTrans->failedTimes > 3) {
       if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
       sendRsp = true;
     }
@@ -1127,6 +1127,7 @@ static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans)
     }
 
     if (code == 0) {
+      pTrans->failedTimes = 0;
       pTrans->lastAction = action;
       pTrans->lastMsgType = 0;
       pTrans->lastErrorNo = 0;
@@ -1430,8 +1431,7 @@ void mndTransPullup(SMnode *pMnode) {
     mndReleaseTrans(pMnode, pTrans);
   }
 
-  // todo, set to SDB_WRITE_DELTA
-  sdbWriteFile(pMnode->pSdb, 0);
+  sdbWriteFile(pMnode->pSdb, SDB_WRITE_DELTA);
   taosArrayDestroy(pArray);
 }
 

source/dnode/mnode/sdb/inc/sdb.h

@@ -37,7 +37,7 @@ extern "C" {
 #define mTrace(...) { if (mDebugFlag & DEBUG_TRACE) { taosPrintLog("MND ", DEBUG_TRACE, mDebugFlag, __VA_ARGS__); }}
 // clang-format on
 
-#define SDB_WRITE_DELTA 100
+#define SDB_WRITE_DELTA 20
 
 #define SDB_GET_VAL(pData, dataPos, val, pos, func, type) \
   {                                                       \

source/dnode/mnode/sdb/src/sdb.c

@@ -161,9 +161,11 @@ static int32_t sdbCreateDir(SSdb *pSdb) {
 }
 
 void sdbSetApplyInfo(SSdb *pSdb, int64_t index, int64_t term, int64_t config) {
-  mTrace("mnode apply info changed, from index:%" PRId64 " term:%" PRId64 " config:%" PRId64 ", to index:%" PRId64
+#if 1
+  mTrace("mnode apply info changed from index:%" PRId64 " term:%" PRId64 " config:%" PRId64 " to index:%" PRId64
          " term:%" PRId64 " config:%" PRId64,
          pSdb->applyIndex, pSdb->applyTerm, pSdb->applyConfig, index, term, config);
+#endif
   pSdb->applyIndex = index;
   pSdb->applyTerm = term;
   pSdb->applyConfig = config;
@@ -173,7 +175,9 @@ void sdbGetCommitInfo(SSdb *pSdb, int64_t *index, int64_t *term, int64_t *config
   *index = pSdb->commitIndex;
   *term = pSdb->commitTerm;
   *config = pSdb->commitConfig;
+#if 0
   mTrace("mnode current info, apply index:%" PRId64 " term:%" PRId64 " config:%" PRId64 ", commit index:%" PRId64
          " term:%" PRId64 " config:%" PRId64,
          pSdb->applyIndex, pSdb->applyTerm, pSdb->applyConfig, *index, *term, *config);
+#endif
 }

source/dnode/snode/inc/sndInt.h

@@ -56,7 +56,6 @@ SStreamTask* sndMetaGetTask(SStreamMeta* pMeta, int32_t taskId);
 int32_t      sndMetaRemoveTask(SStreamMeta* pMeta, int32_t taskId);
 
 int32_t sndDropTaskOfStream(SStreamMeta* pMeta, int64_t streamId);
-
 int32_t sndStopTaskOfStream(SStreamMeta* pMeta, int64_t streamId);
 int32_t sndResumeTaskOfStream(SStreamMeta* pMeta, int64_t streamId);
 

source/dnode/snode/src/snode.c

@@ -76,45 +76,158 @@ int32_t sndMetaRemoveTask(SStreamMeta *pMeta, int32_t taskId) {
   return taosHashRemove(pMeta->pHash, &taskId, sizeof(int32_t));
 }
 
-static int32_t sndProcessTaskExecReq(SSnode *pSnode, SRpcMsg *pMsg) {
-  /*SStreamExecMsgHead *pHead = pMsg->pCont;*/
-  /*int32_t             taskId = pHead->streamTaskId;*/
-  /*SStreamTask *pTask = sndMetaGetTask(pSnode->pMeta, taskId);*/
-  /*if (pTask == NULL) {*/
-  /*return -1;*/
-  /*}*/
-  return 0;
-}
+static int32_t sndProcessTaskDeployReq(SSnode *pNode, SRpcMsg *pMsg) {
+  SStreamMeta *pMeta = pNode->pMeta;
+  char        *msg = pMsg->pCont;
+  int32_t      msgLen = pMsg->contLen;
 
-void sndProcessUMsg(SSnode *pSnode, SRpcMsg *pMsg) {
-  // stream deploy
-  // stream stop/resume
-  // operator exec
-  if (pMsg->msgType == TDMT_STREAM_TASK_DEPLOY) {
-    void        *msg = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
-    SStreamTask *pTask = taosMemoryMalloc(sizeof(SStreamTask));
+  SStreamTask *pTask = taosMemoryCalloc(1, sizeof(SStreamTask));
   if (pTask == NULL) {
-      ASSERT(0);
-      return;
+    return -1;
   }
   SDecoder decoder;
-    tDecoderInit(&decoder, msg, pMsg->contLen - sizeof(SMsgHead));
-    tDecodeSStreamTask(&decoder, pTask);
+  tDecoderInit(&decoder, (uint8_t *)msg, msgLen);
+  if (tDecodeSStreamTask(&decoder, pTask) < 0) {
+    ASSERT(0);
+  }
   tDecoderClear(&decoder);
 
-    sndMetaDeployTask(pSnode->pMeta, pTask);
-    /*} else if (pMsg->msgType == TDMT_SND_TASK_EXEC) {*/
-    /*sndProcessTaskExecReq(pSnode, pMsg);*/
-  } else {
-    ASSERT(0);
+  pTask->status = TASK_STATUS__IDLE;
+
+  pTask->inputQueue = streamQueueOpen();
+  pTask->outputQueue = streamQueueOpen();
+  pTask->inputStatus = TASK_INPUT_STATUS__NORMAL;
+  pTask->outputStatus = TASK_INPUT_STATUS__NORMAL;
+
+  if (pTask->inputQueue == NULL || pTask->outputQueue == NULL) goto FAIL;
+
+  pTask->pMsgCb = &pNode->msgCb;
+
+  ASSERT(pTask->execType != TASK_EXEC__NONE);
+
+  SReadHandle handle = {
+      .pMsgCb = &pNode->msgCb,
+  };
+
+  /*pTask->exec.inputHandle = NULL;*/
+  pTask->exec.executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, &handle);
+  ASSERT(pTask->exec.executor);
+
+  streamSetupTrigger(pTask);
+
+  qInfo("deploy stream: stream id %ld task id %d child id %d on snode", pTask->streamId, pTask->taskId, pTask->childId);
+
+  return 0;
+
+FAIL:
+  if (pTask->inputQueue) streamQueueClose(pTask->inputQueue);
+  if (pTask->outputQueue) streamQueueClose(pTask->outputQueue);
+  return -1;
+}
+
+static int32_t sndProcessTaskRunReq(SSnode *pNode, SRpcMsg *pMsg) {
+  SStreamMeta       *pMeta = pNode->pMeta;
+  SStreamTaskRunReq *pReq = pMsg->pCont;
+  int32_t            taskId = pReq->taskId;
+  SStreamTask       *pTask = *(SStreamTask **)taosHashGet(pMeta->pHash, &taskId, sizeof(int32_t));
+  streamTaskProcessRunReq(pTask, &pNode->msgCb);
+  return 0;
+}
+
+static int32_t sndProcessTaskDispatchReq(SSnode *pNode, SRpcMsg *pMsg) {
+  SStreamMeta *pMeta = pNode->pMeta;
+
+  char   *msgStr = pMsg->pCont;
+  char   *msgBody = POINTER_SHIFT(msgStr, sizeof(SMsgHead));
+  int32_t msgLen = pMsg->contLen - sizeof(SMsgHead);
+
+  SStreamDispatchReq req;
+  SDecoder           decoder;
+  tDecoderInit(&decoder, msgBody, msgLen);
+  tDecodeStreamDispatchReq(&decoder, &req);
+  int32_t      taskId = req.taskId;
+  SStreamTask *pTask = *(SStreamTask **)taosHashGet(pMeta->pHash, &taskId, sizeof(int32_t));
+  SRpcMsg      rsp = {
+           .info = pMsg->info,
+           .code = 0,
+  };
+  streamProcessDispatchReq(pTask, &pNode->msgCb, &req, &rsp);
+  return 0;
+}
+
+static int32_t sndProcessTaskRecoverReq(SSnode *pNode, SRpcMsg *pMsg) {
+  SStreamMeta *pMeta = pNode->pMeta;
+
+  SStreamTaskRecoverReq *pReq = pMsg->pCont;
+  int32_t                taskId = pReq->taskId;
+  SStreamTask           *pTask = *(SStreamTask **)taosHashGet(pMeta->pHash, &taskId, sizeof(int32_t));
+  streamProcessRecoverReq(pTask, &pNode->msgCb, pReq, pMsg);
+  return 0;
+}
+
+static int32_t sndProcessTaskDispatchRsp(SSnode *pNode, SRpcMsg *pMsg) {
+  SStreamMeta *pMeta = pNode->pMeta;
+
+  SStreamDispatchRsp *pRsp = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
+  int32_t             taskId = pRsp->taskId;
+  SStreamTask        *pTask = *(SStreamTask **)taosHashGet(pMeta->pHash, &taskId, sizeof(int32_t));
+  streamProcessDispatchRsp(pTask, &pNode->msgCb, pRsp);
+  return 0;
+}
+
+static int32_t sndProcessTaskRecoverRsp(SSnode *pNode, SRpcMsg *pMsg) {
+  SStreamMeta *pMeta = pNode->pMeta;
+
+  SStreamTaskRecoverRsp *pRsp = pMsg->pCont;
+  int32_t                taskId = pRsp->taskId;
+  SStreamTask           *pTask = *(SStreamTask **)taosHashGet(pMeta->pHash, &taskId, sizeof(int32_t));
+  streamProcessRecoverRsp(pTask, pRsp);
+  return 0;
+}
+
+static int32_t sndProcessTaskDropReq(SSnode *pNode, SRpcMsg *pMsg) {
+  SStreamMeta *pMeta = pNode->pMeta;
+
+  char                *msg = pMsg->pCont;
+  int32_t              msgLen = pMsg->contLen;
+  SVDropStreamTaskReq *pReq = (SVDropStreamTaskReq *)msg;
+  int32_t              code = taosHashRemove(pMeta->pHash, &pReq->taskId, sizeof(int32_t));
+  ASSERT(code == 0);
+  if (code == 0) {
+    // sendrsp
   }
+  return code;
 }
 
-void sndProcessSMsg(SSnode *pSnode, SRpcMsg *pMsg) {
+int32_t sndProcessUMsg(SSnode *pSnode, SRpcMsg *pMsg) {
+  // stream deploy
+  // stream stop/resume
   // operator exec
-  /*if (pMsg->msgType == TDMT_SND_TASK_EXEC) {*/
-  /*sndProcessTaskExecReq(pSnode, pMsg);*/
-  /*} else {*/
+  switch (pMsg->msgType) {
+    case TDMT_STREAM_TASK_DEPLOY:
+      return sndProcessTaskDeployReq(pSnode, pMsg);
+    case TDMT_VND_STREAM_TASK_DROP:
+      return sndProcessTaskDropReq(pSnode, pMsg);
+    default:
       ASSERT(0);
-  /*}*/
+  }
+  return 0;
+}
+
+int32_t sndProcessSMsg(SSnode *pSnode, SRpcMsg *pMsg) {
+  switch (pMsg->msgType) {
+    case TDMT_STREAM_TASK_RUN:
+      return sndProcessTaskRunReq(pSnode, pMsg);
+    case TDMT_STREAM_TASK_DISPATCH:
+      return sndProcessTaskDispatchReq(pSnode, pMsg);
+    case TDMT_STREAM_TASK_RECOVER:
+      return sndProcessTaskRecoverReq(pSnode, pMsg);
+    case TDMT_STREAM_TASK_DISPATCH_RSP:
+      return sndProcessTaskDispatchRsp(pSnode, pMsg);
+    case TDMT_STREAM_TASK_RECOVER_RSP:
+      return sndProcessTaskRecoverRsp(pSnode, pMsg);
+    default:
+      ASSERT(0);
+  }
+  return 0;
 }

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

@@ -125,10 +125,10 @@ int32_t tqProcessOffsetCommitReq(STQ* pTq, char* msg, int32_t msgLen) {
 
   if (offset.type == TMQ_OFFSET__SNAPSHOT) {
     tqDebug("receive offset commit msg to %s on vg %d, offset(type:snapshot) uid: %ld, ts: %ld", offset.subKey,
-            pTq->pVnode->config.vgId, offset.uid, offset.ts);
+            TD_VID(pTq->pVnode), offset.uid, offset.ts);
   } else if (offset.type == TMQ_OFFSET__LOG) {
     tqDebug("receive offset commit msg to %s on vg %d, offset(type:log) version: %ld", offset.subKey,
-            pTq->pVnode->config.vgId, offset.version);
+            TD_VID(pTq->pVnode), offset.version);
   } else {
     ASSERT(0);
   }
@@ -159,7 +159,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
     if (pOffset != NULL) {
       ASSERT(pOffset->type == TMQ_OFFSET__LOG);
       tqDebug("consumer %ld, restore offset of %s on vg %d, offset(type:log) version: %ld", consumerId, pReq->subKey,
-              pTq->pVnode->config.vgId, pOffset->version);
+              TD_VID(pTq->pVnode), pOffset->version);
       fetchOffset = pOffset->version + 1;
     } else {
       if (pReq->currentOffset == TMQ_CONF__RESET_OFFSET__EARLIEAST) {
@@ -167,13 +167,13 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
       } else if (pReq->currentOffset == TMQ_CONF__RESET_OFFSET__LATEST) {
         fetchOffset = walGetCommittedVer(pTq->pWal);
       } else if (pReq->currentOffset == TMQ_CONF__RESET_OFFSET__NONE) {
-        tqError("tmq poll: no offset committed for consumer %ld in vg %d, subkey %s", consumerId,
-                pTq->pVnode->config.vgId, pReq->subKey);
+        tqError("tmq poll: no offset committed for consumer %ld in vg %d, subkey %s", consumerId, TD_VID(pTq->pVnode),
+                pReq->subKey);
         terrno = TSDB_CODE_TQ_NO_COMMITTED_OFFSET;
         return -1;
       }
       tqDebug("consumer %ld, restore offset of %s on vg %d failed, config is %ld, set to %ld", consumerId, pReq->subKey,
-              pTq->pVnode->config.vgId, pReq->currentOffset, fetchOffset);
+              TD_VID(pTq->pVnode), pReq->currentOffset, fetchOffset);
     }
   }
 
@@ -183,14 +183,14 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
   STqHandle* pHandle = taosHashGet(pTq->handles, pReq->subKey, strlen(pReq->subKey));
   /*ASSERT(pHandle);*/
   if (pHandle == NULL) {
-    tqError("tmq poll: no consumer handle for consumer %ld in vg %d, subkey %s", consumerId, pTq->pVnode->config.vgId,
+    tqError("tmq poll: no consumer handle for consumer %ld in vg %d, subkey %s", consumerId, TD_VID(pTq->pVnode),
             pReq->subKey);
     return -1;
   }
 
   if (pHandle->consumerId != consumerId) {
     tqError("tmq poll: consumer handle mismatch for consumer %ld in vg %d, subkey %s, handle consumer id %ld",
-            consumerId, pTq->pVnode->config.vgId, pReq->subKey, pHandle->consumerId);
+            consumerId, TD_VID(pTq->pVnode), pReq->subKey, pHandle->consumerId);
     return -1;
   }
 
@@ -304,7 +304,6 @@ int32_t tqProcessVgDeleteReq(STQ* pTq, char* msg, int32_t msgLen) {
   return 0;
 }
 
-// TODO: persist meta into tdb
 int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
   SMqRebVgReq req = {0};
   tDecodeSMqRebVgReq(msg, &req);
@@ -346,10 +345,10 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
       pHandle->execHandle.execTb.suid = req.suid;
       SArray* tbUidList = taosArrayInit(0, sizeof(int64_t));
       tsdbGetCtbIdList(pTq->pVnode->pMeta, req.suid, tbUidList);
-      tqDebug("vg %d, tq try get suid: %ld", pTq->pVnode->config.vgId, req.suid);
+      tqDebug("vg %d, tq try get suid: %ld", TD_VID(pTq->pVnode), req.suid);
       for (int32_t i = 0; i < taosArrayGetSize(tbUidList); i++) {
         int64_t tbUid = *(int64_t*)taosArrayGet(tbUidList, i);
-        tqDebug("vg %d, idx %d, uid: %ld", pTq->pVnode->config.vgId, i, tbUid);
+        tqDebug("vg %d, idx %d, uid: %ld", TD_VID(pTq->pVnode), i, tbUid);
       }
       for (int32_t i = 0; i < 5; i++) {
         tqReadHandleSetTbUidList(pHandle->execHandle.pExecReader[i], tbUidList);
@@ -400,6 +399,7 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) {
   // exec
   if (pTask->execType != TASK_EXEC__NONE) {
     // expand runners
+    if (pTask->dataScan) {
       STqReadHandle* pStreamReader = tqInitSubmitMsgScanner(pTq->pVnode->pMeta);
       SReadHandle    handle = {
              .reader = pStreamReader,
@@ -407,9 +407,13 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) {
              .pMsgCb = &pTq->pVnode->msgCb,
              .vnode = pTq->pVnode,
       };
-    pTask->exec.inputHandle = pStreamReader;
+      /*pTask->exec.inputHandle = pStreamReader;*/
       pTask->exec.executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, &handle);
       ASSERT(pTask->exec.executor);
+    } else {
+      pTask->exec.executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, NULL);
+      ASSERT(pTask->exec.executor);
+    }
   }
 
   // sink
@@ -431,7 +435,7 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) {
 
   streamSetupTrigger(pTask);
 
-  tqInfo("deploy stream task id %d child id %d on vg %d", pTask->taskId, pTask->childId, pTq->pVnode->config.vgId);
+  tqInfo("deploy stream task id %d child id %d on vg %d", pTask->taskId, pTask->childId, TD_VID(pTq->pVnode));
 
   taosHashPut(pTq->pStreamTasks, &pTask->taskId, sizeof(int32_t), &pTask, sizeof(void*));
 
@@ -464,7 +468,7 @@ int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* pReq) {
         continue;
       }
 
-      if (streamLaunchByWrite(pTask, pTq->pVnode->config.vgId, &pTq->pVnode->msgCb) < 0) {
+      if (streamLaunchByWrite(pTask, TD_VID(pTq->pVnode), &pTq->pVnode->msgCb) < 0) {
         continue;
       }
     } else {
@@ -534,9 +538,9 @@ int32_t tqProcessTaskRecoverRsp(STQ* pTq, SRpcMsg* pMsg) {
 int32_t tqProcessTaskDropReq(STQ* pTq, char* msg, int32_t msgLen) {
   SVDropStreamTaskReq* pReq = (SVDropStreamTaskReq*)msg;
   int32_t              code = taosHashRemove(pTq->pStreamTasks, &pReq->taskId, sizeof(int32_t));
+  ASSERT(code == 0);
   if (code == 0) {
     // sendrsp
   }
-  ASSERT(code == 0);
   return code;
 }

source/libs/executor/inc/executil.h

@@ -114,7 +114,7 @@ SArray* extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNod
 SExprInfo* createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, int32_t* numOfExprs);
 
 SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput, int32_t** rowEntryInfoOffset);
-void    relocateColumnData(SSDataBlock* pBlock, const SArray* pColMatchInfo, SArray* pCols);
+void    relocateColumnData(SSDataBlock* pBlock, const SArray* pColMatchInfo, SArray* pCols, bool outputEveryColumn);
 void    initExecTimeWindowInfo(SColumnInfoData* pColData, STimeWindow* pQueryWindow);
 
 SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode);

source/libs/executor/src/dataDispatcher.c

@@ -75,12 +75,18 @@ static bool needCompress(const SSDataBlock* pData, int32_t numOfCols) {
 // The length of bitmap is decided by number of rows of this data block, and the length of each column data is
 // recorded in the first segment, next to the struct header
 static void toDataCacheEntry(SDataDispatchHandle* pHandle, const SInputData* pInput, SDataDispatchBuf* pBuf) {
-  int32_t numOfCols = LIST_LENGTH(pHandle->pSchema->pSlots);
-
+  int32_t numOfCols = 0;
+  SNode* pNode;
+  FOREACH(pNode, pHandle->pSchema->pSlots) {
+    SSlotDescNode* pSlotDesc = (SSlotDescNode*)pNode;
+    if (pSlotDesc->output) {
+      ++numOfCols;
+    }
+  }
   SDataCacheEntry* pEntry = (SDataCacheEntry*)pBuf->pData;
   pEntry->compressed = (int8_t)needCompress(pInput->pData, numOfCols);
   pEntry->numOfRows = pInput->pData->info.rows;
-  pEntry->numOfCols = pInput->pData->info.numOfCols;
+  pEntry->numOfCols = numOfCols;
   pEntry->dataLen = 0;
 
   pBuf->useSize = sizeof(SDataCacheEntry);

source/libs/executor/src/executil.c

@@ -193,9 +193,9 @@ SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode) {
   for (int32_t i = 0; i < numOfCols; ++i) {
     SColumnInfoData idata = {{0}};
     SSlotDescNode*  pDescNode = (SSlotDescNode*)nodesListGetNode(pNode->pSlots, i);
-    //    if (!pDescNode->output) {  // todo disable it temporarily
-    //      continue;
-    //    }
+//    if (!pDescNode->output) {  // todo disable it temporarily
+//      continue;
+//    }
 
     idata.info.type = pDescNode->dataType.type;
     idata.info.bytes = pDescNode->dataType.bytes;
@@ -235,7 +235,7 @@ int32_t getTableList(void* metaHandle, SScanPhysiNode* pScanNode, STableListInfo
         terrno = code;
         return code;
       } else {
-        qDebug("sucess to  get tableIds, size: %d, suid: %" PRIu64 "", (int)taosArrayGetSize(res), tableUid);
+        qDebug("success to  get tableIds, size: %d, suid: %" PRIu64 "", (int)taosArrayGetSize(res), tableUid);
       }
 
       for (int i = 0; i < taosArrayGetSize(res); i++) {
@@ -319,7 +319,14 @@ SArray* extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNod
       continue;
     }
 
-    SColMatchInfo* info = taosArrayGet(pList, pNode->slotId);
+    SColMatchInfo* info = NULL;
+    for (int32_t j = 0; j < taosArrayGetSize(pList); ++j) {
+      info = taosArrayGet(pList, j);
+      if (info->targetSlotId == pNode->slotId) {
+        break;
+      }
+    }
+
     if (pNode->output) {
       (*numOfOutputCols) += 1;
     } else {
@@ -578,14 +585,15 @@ SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput,
 }
 
 // NOTE: sources columns are more than the destination SSDatablock columns.
-void relocateColumnData(SSDataBlock* pBlock, const SArray* pColMatchInfo, SArray* pCols) {
+// doFilter in table scan needs every column even its output is false
+void relocateColumnData(SSDataBlock* pBlock, const SArray* pColMatchInfo, SArray* pCols, bool outputEveryColumn) {
   size_t numOfSrcCols = taosArrayGetSize(pCols);
 
   int32_t i = 0, j = 0;
   while (i < numOfSrcCols && j < taosArrayGetSize(pColMatchInfo)) {
     SColumnInfoData* p = taosArrayGet(pCols, i);
     SColMatchInfo*   pmInfo = taosArrayGet(pColMatchInfo, j);
-    if (!pmInfo->output) {
+    if (!outputEveryColumn && !pmInfo->output) {
       j++;
       continue;
     }

source/libs/executor/src/executor.c

@@ -105,7 +105,7 @@ int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numO
 }
 
 qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, void* streamReadHandle) {
-  if (msg == NULL || streamReadHandle == NULL) {
+  if (msg == NULL) {
     return NULL;
   }
 

source/libs/executor/src/executorMain.c

@@ -13,10 +13,10 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "os.h"
-#include "tref.h"
 #include "dataSinkMgt.h"
+#include "os.h"
 #include "tmsg.h"
+#include "tref.h"
 #include "tudf.h"
 
 #include "executor.h"
@@ -26,13 +26,11 @@
 static TdThreadOnce initPoolOnce = PTHREAD_ONCE_INIT;
 int32_t             exchangeObjRefPool = -1;
 
-static void initRefPool() {
-  exchangeObjRefPool = taosOpenRef(1024, doDestroyExchangeOperatorInfo);
-}
+static void initRefPool() { exchangeObjRefPool = taosOpenRef(1024, doDestroyExchangeOperatorInfo); }
 
 int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId, SSubplan* pSubplan,
                         qTaskInfo_t* pTaskInfo, DataSinkHandle* handle, const char* sql, EOPTR_EXEC_MODEL model) {
-  assert(readHandle != NULL && pSubplan != NULL);
+  assert(pSubplan != NULL);
   SExecTaskInfo** pTask = (SExecTaskInfo**)pTaskInfo;
 
   taosThreadOnce(&initPoolOnce, initRefPool);
@@ -58,46 +56,46 @@ int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId,
     code = dsCreateDataSinker(pSubplan->pDataSink, handle, pSinkParam);
   }
 
-  _error:
+_error:
   // if failed to add ref for all tables in this query, abort current query
   return code;
 }
 
 #ifdef TEST_IMPL
 // wait moment
-int waitMoment(SQInfo* pQInfo){
-  if(pQInfo->sql) {
+int waitMoment(SQInfo* pQInfo) {
+  if (pQInfo->sql) {
     int   ms = 0;
     char* pcnt = strstr(pQInfo->sql, " count(*)");
-    if(pcnt) return 0;
+    if (pcnt) return 0;
 
     char* pos = strstr(pQInfo->sql, " t_");
-    if(pos){
+    if (pos) {
       pos += 3;
       ms = atoi(pos);
-      while(*pos >= '0' && *pos <= '9'){
-        pos ++;
+      while (*pos >= '0' && *pos <= '9') {
+        pos++;
       }
       char unit_char = *pos;
-      if(unit_char == 'h'){
-        ms *= 3600*1000;
-      } else if(unit_char == 'm'){
-        ms *= 60*1000;
-      } else if(unit_char == 's'){
+      if (unit_char == 'h') {
+        ms *= 3600 * 1000;
+      } else if (unit_char == 'm') {
+        ms *= 60 * 1000;
+      } else if (unit_char == 's') {
         ms *= 1000;
       }
     }
-    if(ms == 0) return 0;
+    if (ms == 0) return 0;
     printf("test wait sleep %dms. sql=%s ...\n", ms, pQInfo->sql);
 
-    if(ms < 1000) {
+    if (ms < 1000) {
       taosMsleep(ms);
     } else {
       int used_ms = 0;
-      while(used_ms < ms) {
+      while (used_ms < ms) {
         taosMsleep(1000);
         used_ms += 1000;
-        if(isTaskKilled(pQInfo)){
+        if (isTaskKilled(pQInfo)) {
           printf("test check query is canceled, sleep break.%s\n", pQInfo->sql);
           break;
         }
@@ -108,15 +106,14 @@ int waitMoment(SQInfo* pQInfo){
 }
 #endif
 
-int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t *useconds) {
+int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t* useconds) {
   SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
   int64_t        threadId = taosGetSelfPthreadId();
 
   *pRes = NULL;
   int64_t curOwner = 0;
   if ((curOwner = atomic_val_compare_exchange_64(&pTaskInfo->owner, 0, threadId)) != 0) {
-    qError("%s-%p execTask is now executed by thread:%p", GET_TASKID(pTaskInfo), pTaskInfo,
-           (void*)curOwner);
+    qError("%s-%p execTask is now executed by thread:%p", GET_TASKID(pTaskInfo), pTaskInfo, (void*)curOwner);
     pTaskInfo->code = TSDB_CODE_QRY_IN_EXEC;
     return pTaskInfo->code;
   }
@@ -152,18 +149,18 @@ int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t *useconds) {
 
   cleanUpUdfs();
 
-  int32_t current = (*pRes != NULL)? (*pRes)->info.rows:0;
+  int32_t  current = (*pRes != NULL) ? (*pRes)->info.rows : 0;
   uint64_t total = pTaskInfo->pRoot->resultInfo.totalRows;
 
   qDebug("%s task suspended, %d rows returned, total:%" PRId64 " rows, in sinkNode:%d, elapsed:%.2f ms",
-         GET_TASKID(pTaskInfo), current, total, 0, el/1000.0);
+         GET_TASKID(pTaskInfo), current, total, 0, el / 1000.0);
 
   atomic_store_64(&pTaskInfo->owner, 0);
   return pTaskInfo->code;
 }
 
 int32_t qKillTask(qTaskInfo_t qinfo) {
-  SExecTaskInfo *pTaskInfo = (SExecTaskInfo *)qinfo;
+  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)qinfo;
 
   if (pTaskInfo == NULL) {
     return TSDB_CODE_QRY_INVALID_QHANDLE;
@@ -182,7 +179,7 @@ int32_t qKillTask(qTaskInfo_t qinfo) {
 }
 
 int32_t qAsyncKillTask(qTaskInfo_t qinfo) {
-  SExecTaskInfo *pTaskInfo = (SExecTaskInfo *)qinfo;
+  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)qinfo;
 
   if (pTaskInfo == NULL) {
     return TSDB_CODE_QRY_INVALID_QHANDLE;
@@ -195,7 +192,7 @@ int32_t qAsyncKillTask(qTaskInfo_t qinfo) {
 }
 
 int32_t qIsTaskCompleted(qTaskInfo_t qinfo) {
-  SExecTaskInfo *pTaskInfo = (SExecTaskInfo *)qinfo;
+  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)qinfo;
 
   if (pTaskInfo == NULL) {
     return TSDB_CODE_QRY_INVALID_QHANDLE;
@@ -205,15 +202,15 @@ int32_t qIsTaskCompleted(qTaskInfo_t qinfo) {
 }
 
 void qDestroyTask(qTaskInfo_t qTaskHandle) {
-  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*) qTaskHandle;
-  qDebug("%s execTask completed, numOfRows:%"PRId64, GET_TASKID(pTaskInfo), pTaskInfo->pRoot->resultInfo.totalRows);
+  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)qTaskHandle;
+  qDebug("%s execTask completed, numOfRows:%" PRId64, GET_TASKID(pTaskInfo), pTaskInfo->pRoot->resultInfo.totalRows);
 
   queryCostStatis(pTaskInfo);  // print the query cost summary
   doDestroyTask(pTaskInfo);
 }
 
-int32_t qGetExplainExecInfo(qTaskInfo_t tinfo, int32_t *resNum, SExplainExecInfo **pRes) {
-  SExecTaskInfo *pTaskInfo = (SExecTaskInfo *)tinfo;
+int32_t qGetExplainExecInfo(qTaskInfo_t tinfo, int32_t* resNum, SExplainExecInfo** pRes) {
+  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
   int32_t        capacity = 0;
 
   return getOperatorExplainExecInfo(pTaskInfo->pRoot, pRes, &capacity, resNum);

source/libs/executor/src/executorimpl.c

@@ -141,8 +141,8 @@ static int32_t doCopyToSDataBlock(SExecTaskInfo* taskInfo, SSDataBlock* pBlock,
                                   SqlFunctionCtx* pCtx, int32_t numOfExprs);
 
 static void initCtxOutputBuffer(SqlFunctionCtx* pCtx, int32_t size);
-static void doSetTableGroupOutputBuf(SOperatorInfo* pOperator, SAggOperatorInfo* pAggInfo, int32_t numOfOutput, uint64_t groupId);
-
+static void doSetTableGroupOutputBuf(SOperatorInfo* pOperator, SAggOperatorInfo* pAggInfo, int32_t numOfOutput,
+                                     uint64_t groupId);
 
 // setup the output buffer for each operator
 static bool hasNull(SColumn* pColumn, SColumnDataAgg* pStatis) {
@@ -1230,7 +1230,8 @@ void initResultRow(SResultRow* pResultRow) {
  *           offset[0]                                  offset[1]                                   offset[2]
  */
 // TODO refactor: some function move away
-void setFunctionResultOutput(SOperatorInfo *pOperator, SOptrBasicInfo *pInfo, SAggSupporter* pSup, int32_t stage, int32_t numOfExprs) {
+void setFunctionResultOutput(SOperatorInfo* pOperator, SOptrBasicInfo* pInfo, SAggSupporter* pSup, int32_t stage,
+                             int32_t numOfExprs) {
   SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;
   SqlFunctionCtx* pCtx = pOperator->exprSupp.pCtx;
   int32_t*        rowEntryInfoOffset = pOperator->exprSupp.rowEntryInfoOffset;
@@ -1380,7 +1381,8 @@ void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const int8_t* rowR
   }
 }
 
-void doSetTableGroupOutputBuf(SOperatorInfo* pOperator, SAggOperatorInfo* pAggInfo, int32_t numOfOutput, uint64_t groupId) {
+void doSetTableGroupOutputBuf(SOperatorInfo* pOperator, SAggOperatorInfo* pAggInfo, int32_t numOfOutput,
+                              uint64_t groupId) {
   // for simple group by query without interval, all the tables belong to one group result.
   SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;
   SResultRowInfo* pResultRowInfo = &pAggInfo->binfo.resultRowInfo;
@@ -2084,7 +2086,7 @@ int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLo
 
     // data from mnode
     pRes->info.rows = numOfRows;
-    relocateColumnData(pRes, pColList, pBlock->pDataBlock);
+    relocateColumnData(pRes, pColList, pBlock->pDataBlock, false);
     taosArrayDestroy(pBlock->pDataBlock);
     taosMemoryFree(pBlock);
     //    blockDataDestroy(pBlock);
@@ -2165,7 +2167,8 @@ static SSDataBlock* concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SEx
       }
 
       SRetrieveTableRsp* pTableRsp = pDataInfo->pRsp;
-      code = extractDataBlockFromFetchRsp(pExchangeInfo->pResult, pLoadInfo, pTableRsp->numOfRows, pTableRsp->data,
+      code =
+          extractDataBlockFromFetchRsp(pExchangeInfo->pResult, pLoadInfo, pTableRsp->numOfRows, pTableRsp->data,
                                        pTableRsp->compLen, pTableRsp->numOfCols, startTs, &pDataInfo->totalRows, NULL);
       if (code != 0) {
         taosMemoryFreeClear(pDataInfo->pRsp);
@@ -2673,8 +2676,8 @@ static SSDataBlock* doSortedMerge(SOperatorInfo* pOperator) {
   }
 
   int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;
-  pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize,
-                                             numOfBufPage, pInfo->binfo.pRes, "GET_TASKID(pTaskInfo)");
+  pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize, numOfBufPage,
+                                             pInfo->binfo.pRes, "GET_TASKID(pTaskInfo)");
 
   tsortSetFetchRawDataFp(pInfo->pSortHandle, loadNextDataBlock, NULL, NULL);
 
@@ -3403,8 +3406,8 @@ void cleanupAggSup(SAggSupporter* pAggSup) {
   destroyDiskbasedBuf(pAggSup->pResultBuf);
 }
 
-int32_t initAggInfo(SExprSupp* pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols,
-                    size_t keyBufSize, const char* pkey) {
+int32_t initAggInfo(SExprSupp* pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, size_t keyBufSize,
+                    const char* pkey) {
   initExprSupp(pSup, pExprInfo, numOfCols);
   doInitAggInfoSup(pAggSup, pSup->pCtx, numOfCols, keyBufSize, pkey);
   for (int32_t i = 0; i < numOfCols; ++i) {
@@ -3578,7 +3581,8 @@ static SArray* setRowTsColumnOutputInfo(SqlFunctionCtx* pCtx, int32_t numOfCols)
   return pList;
 }
 
-SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhysiNode* pProjPhyNode, SExecTaskInfo* pTaskInfo) {
+SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhysiNode* pProjPhyNode,
+                                         SExecTaskInfo* pTaskInfo) {
   SProjectOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SProjectOperatorInfo));
   SOperatorInfo*        pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   if (pInfo == NULL || pOperator == NULL) {
@@ -3686,8 +3690,8 @@ static SSDataBlock* doApplyIndefinitFunction(SOperatorInfo* pOperator) {
     setInputDataBlock(pOperator, pSup->pCtx, pBlock, order, scanFlag, false);
     blockDataEnsureCapacity(pInfo->pRes, pInfo->pRes->info.rows + pBlock->info.rows);
 
-    code = projectApplyFunctions(pOperator->exprSupp.pExprInfo, pInfo->pRes, pBlock, pSup->pCtx, pOperator->exprSupp.numOfExprs,
-                                 pIndefInfo->pPseudoColInfo);
+    code = projectApplyFunctions(pOperator->exprSupp.pExprInfo, pInfo->pRes, pBlock, pSup->pCtx,
+                                 pOperator->exprSupp.numOfExprs, pIndefInfo->pPseudoColInfo);
     if (code != TSDB_CODE_SUCCESS) {
       longjmp(pTaskInfo->env, code);
     }
@@ -4050,11 +4054,15 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
       STimeWindowAggSupp   twSup = {
             .waterMark = pTableScanNode->watermark, .calTrigger = pTableScanNode->triggerType, .maxTs = INT64_MIN};
       tsdbReaderT pDataReader = NULL;
+
+      if (pHandle) {
         if (pHandle->vnode) {
-        pDataReader = doCreateDataReader(pTableScanNode, pHandle, pTableListInfo, (uint64_t)queryId, taskId, pTagCond);
+          pDataReader =
+              doCreateDataReader(pTableScanNode, pHandle, pTableListInfo, (uint64_t)queryId, taskId, pTagCond);
         } else {
           getTableList(pHandle->meta, pScanPhyNode, pTableListInfo, pTagCond);
         }
+      }
 
       if (pDataReader == NULL && terrno != 0) {
         qDebug("%s pDataReader is NULL", GET_TASKID(pTaskInfo));
@@ -4413,7 +4421,7 @@ int32_t decodeOperator(SOperatorInfo* ops, const char* result, int32_t length) {
     ASSERT(length == *(int32_t*)result);
 
     const char* data = result + sizeof(int32_t);
-    code = ops->fpSet.decodeResultRow(ops, (char*) data);
+    code = ops->fpSet.decodeResultRow(ops, (char*)data);
     if (code != TDB_CODE_SUCCESS) {
       return code;
     }

source/libs/executor/src/scanoperator.c

@@ -259,7 +259,7 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableSca
     return terrno;
   }
 
-  relocateColumnData(pBlock, pTableScanInfo->pColMatchInfo, pCols);
+  relocateColumnData(pBlock, pTableScanInfo->pColMatchInfo, pCols, true);
 
   // currently only the tbname pseudo column
   if (pTableScanInfo->pseudoSup.numOfExprs > 0) {
@@ -527,7 +527,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
     goto _error;
   }
 
-  //taosSsleep(20);
+  // taosSsleep(20);
 
   SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;
 
@@ -741,8 +741,7 @@ static bool prepareDataScan(SStreamBlockScanInfo* pInfo) {
       int64_t              gap = pInfo->sessionSup.gap;
       int32_t              winIndex = 0;
       SResultWindowInfo*   pCurWin =
-          getSessionTimeWindow(pAggSup, tsCols[pInfo->updateResIndex], INT64_MIN,
-              pSDB->info.groupId, gap, &winIndex);
+          getSessionTimeWindow(pAggSup, tsCols[pInfo->updateResIndex], INT64_MIN, pSDB->info.groupId, gap, &winIndex);
       win = pCurWin->win;
       pInfo->updateResIndex +=
           updateSessionWindowInfo(pCurWin, tsCols, NULL, pSDB->info.rows, pInfo->updateResIndex, gap, NULL);
@@ -829,7 +828,7 @@ static SSDataBlock* doDataScan(SStreamBlockScanInfo* pInfo) {
     }
   }
 
-/* Todo(liuyao) for partition by column
+  /* Todo(liuyao) for partition by column
     SSDataBlock* pBlock = createOneDataBlock(pResult, true);
     blockDataCleanup(pResult);
     for (int32_t i = 0; i < pBlock->info.rows; i++) {
@@ -839,7 +838,7 @@ static SSDataBlock* doDataScan(SStreamBlockScanInfo* pInfo) {
       }
     }
     return pResult;
-*/
+  */
 }
 
 static void setUpdateData(SStreamBlockScanInfo* pInfo, SSDataBlock* pBlock, SSDataBlock* pUpdateBlock) {
@@ -854,7 +853,7 @@ static void setUpdateData(SStreamBlockScanInfo* pInfo, SSDataBlock* pBlock, SSDa
     int32_t rowId = *(int32_t*)taosArrayGet(pInfo->tsArray, pInfo->tsArrayIndex);
     pInfo->groupId = getGroupId(pInfo->pOperatorDumy, pBlock, rowId);
     int32_t i = 0;
-    for ( ; i < size; i++) {
+    for (; i < size; i++) {
       rowId = *(int32_t*)taosArrayGet(pInfo->tsArray, i + pInfo->tsArrayIndex);
       uint64_t id = getGroupId(pInfo->pOperatorDumy, pBlock, rowId);
       if (pInfo->groupId != id) {
@@ -1084,9 +1083,6 @@ SOperatorInfo* createStreamScanOperatorInfo(void* pDataReader, SReadHandle* pHan
   SScanPhysiNode* pScanPhyNode = &pTableScanNode->scan;
 
   SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc;
-  SOperatorInfo*      pTableScanDummy = createTableScanOperatorInfo(pTableScanNode, pDataReader, pHandle, pTaskInfo);
-
-  STableScanInfo* pSTInfo = (STableScanInfo*)pTableScanDummy->info;
 
   int32_t numOfCols = 0;
   pInfo->pColMatchInfo = extractColMatchInfo(pScanPhyNode->pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID);
@@ -1103,16 +1099,6 @@ SOperatorInfo* createStreamScanOperatorInfo(void* pDataReader, SReadHandle* pHan
     }
   }
 
-  // set the extract column id to streamHandle
-  tqReadHandleSetColIdList((STqReadHandle*)pHandle->reader, pColIds);
-  SArray* tableIdList = extractTableIdList(&pTaskInfo->tableqinfoList);
-  int32_t code = tqReadHandleSetTbUidList(pHandle->reader, tableIdList);
-  if (code != 0) {
-    taosArrayDestroy(tableIdList);
-    goto _error;
-  }
-  taosArrayDestroy(tableIdList);
-
   pInfo->pBlockLists = taosArrayInit(4, POINTER_BYTES);
   if (pInfo->pBlockLists == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -1124,27 +1110,43 @@ SOperatorInfo* createStreamScanOperatorInfo(void* pDataReader, SReadHandle* pHan
     goto _error;
   }
 
-  if (pSTInfo->interval.interval > 0 && pDataReader) {
+  if (pHandle) {
+    SOperatorInfo*  pTableScanDummy = createTableScanOperatorInfo(pTableScanNode, pDataReader, pHandle, pTaskInfo);
+    STableScanInfo* pSTInfo = (STableScanInfo*)pTableScanDummy->info;
+    if (pSTInfo->interval.interval > 0) {
       pInfo->pUpdateInfo = updateInfoInitP(&pSTInfo->interval, pTwSup->waterMark);
     } else {
       pInfo->pUpdateInfo = NULL;
     }
+    pInfo->pOperatorDumy = pTableScanDummy;
+    pInfo->interval = pSTInfo->interval;
+
+    pInfo->readHandle = *pHandle;
+    ASSERT(pHandle->reader);
+    pInfo->streamBlockReader = pHandle->reader;
+    pInfo->tableUid = pScanPhyNode->uid;
+
+    // set the extract column id to streamHandle
+    tqReadHandleSetColIdList((STqReadHandle*)pHandle->reader, pColIds);
+    SArray* tableIdList = extractTableIdList(&pTaskInfo->tableqinfoList);
+    int32_t code = tqReadHandleSetTbUidList(pHandle->reader, tableIdList);
+    if (code != 0) {
+      taosArrayDestroy(tableIdList);
+      goto _error;
+    }
+    taosArrayDestroy(tableIdList);
+  }
 
   // create the pseduo columns info
   if (pTableScanNode->scan.pScanPseudoCols != NULL) {
     pInfo->pPseudoExpr = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pInfo->numOfPseudoExpr);
   }
 
-  pInfo->readHandle = *pHandle;
-  pInfo->tableUid = pScanPhyNode->uid;
-  pInfo->streamBlockReader = pHandle->reader;
   pInfo->pRes = createResDataBlock(pDescNode);
   pInfo->pUpdateRes = createResDataBlock(pDescNode);
   pInfo->pCondition = pScanPhyNode->node.pConditions;
   pInfo->pDataReader = pDataReader;
   pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
-  pInfo->pOperatorDumy = pTableScanDummy;
-  pInfo->interval = pSTInfo->interval;
   pInfo->sessionSup = (SessionWindowSupporter){.pStreamAggSup = NULL, .gap = -1};
   pInfo->groupId = 0;
 
@@ -1426,15 +1428,15 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
 
           // table comment
           pColInfoData = taosArrayGet(p->pDataBlock, 8);
-          if(pInfo->pCur->mr.me.ctbEntry.commentLen > 0) {
+          if (pInfo->pCur->mr.me.ctbEntry.commentLen > 0) {
             char comment[TSDB_TB_COMMENT_LEN + VARSTR_HEADER_SIZE] = {0};
             STR_TO_VARSTR(comment, pInfo->pCur->mr.me.ctbEntry.comment);
             colDataAppend(pColInfoData, numOfRows, comment, false);
-          }else if(pInfo->pCur->mr.me.ctbEntry.commentLen == 0) {
+          } else if (pInfo->pCur->mr.me.ctbEntry.commentLen == 0) {
             char comment[VARSTR_HEADER_SIZE + VARSTR_HEADER_SIZE] = {0};
             STR_TO_VARSTR(comment, "");
             colDataAppend(pColInfoData, numOfRows, comment, false);
-          }else{
+          } else {
             colDataAppendNULL(pColInfoData, numOfRows);
           }
 
@@ -1462,15 +1464,15 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
 
           // table comment
           pColInfoData = taosArrayGet(p->pDataBlock, 8);
-          if(pInfo->pCur->mr.me.ntbEntry.commentLen > 0) {
+          if (pInfo->pCur->mr.me.ntbEntry.commentLen > 0) {
             char comment[TSDB_TB_COMMENT_LEN + VARSTR_HEADER_SIZE] = {0};
             STR_TO_VARSTR(comment, pInfo->pCur->mr.me.ntbEntry.comment);
             colDataAppend(pColInfoData, numOfRows, comment, false);
-          }else if(pInfo->pCur->mr.me.ntbEntry.commentLen == 0) {
+          } else if (pInfo->pCur->mr.me.ntbEntry.commentLen == 0) {
             char comment[VARSTR_HEADER_SIZE + VARSTR_HEADER_SIZE] = {0};
             STR_TO_VARSTR(comment, "");
             colDataAppend(pColInfoData, numOfRows, comment, false);
-          }else{
+          } else {
             colDataAppendNULL(pColInfoData, numOfRows);
           }
 
@@ -1503,7 +1505,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
       p->info.rows = numOfRows;
       pInfo->pRes->info.rows = numOfRows;
 
-      relocateColumnData(pInfo->pRes, pInfo->scanCols, p->pDataBlock);
+      relocateColumnData(pInfo->pRes, pInfo->scanCols, p->pDataBlock, false);
       doFilterResult(pInfo);
 
       blockDataDestroy(p);
@@ -1595,7 +1597,7 @@ int32_t buildSysDbTableInfo(const SSysTableScanInfo* pInfo, int32_t capacity) {
   getPerfDbMeta(&pSysDbTableMeta, &size);
   p->info.rows = buildDbTableInfoBlock(p, pSysDbTableMeta, size, TSDB_PERFORMANCE_SCHEMA_DB);
 
-  relocateColumnData(pInfo->pRes, pInfo->scanCols, p->pDataBlock);
+  relocateColumnData(pInfo->pRes, pInfo->scanCols, p->pDataBlock, false);
   pInfo->pRes->info.rows = p->info.rows;
   blockDataDestroy(p);
 
@@ -1869,6 +1871,7 @@ SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysi
   SExprInfo* pExprInfo = createExprInfo(pPhyNode->pScanPseudoCols, NULL, &numOfExprs);
   SArray*    colList = extractColMatchInfo(pPhyNode->pScanPseudoCols, pDescNode, &num, COL_MATCH_FROM_COL_ID);
 
+
   initExprSupp(&pOperator->exprSupp, pExprInfo, numOfExprs);
 
   pInfo->pTableList       = pTableListInfo;
@@ -1880,6 +1883,7 @@ SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysi
 
   pOperator->name = "TagScanOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN;
+
   pOperator->blocking     = false;
   pOperator->status       = OP_NOT_OPENED;
   pOperator->info         = pInfo;
@@ -1952,8 +1956,7 @@ typedef struct STableMergeScanInfo {
 int32_t createMultipleDataReaders(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle,
                                   STableListInfo* pTableListInfo, SArray* arrayReader, uint64_t queryId,
                                   uint64_t taskId, SNode* pTagCond) {
-  int32_t code =
-      getTableList(pHandle->meta, &pTableScanNode->scan, pTableListInfo, pTagCond);
+  int32_t code = getTableList(pHandle->meta, &pTableScanNode->scan, pTableListInfo, pTagCond);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }
@@ -2076,7 +2079,7 @@ static int32_t loadDataBlockFromOneTable(SOperatorInfo* pOperator, STableMergeSc
     return terrno;
   }
 
-  relocateColumnData(pBlock, pTableScanInfo->pColMatchInfo, pCols);
+  relocateColumnData(pBlock, pTableScanInfo->pColMatchInfo, pCols, true);
 
   // currently only the tbname pseudo column
   if (pTableScanInfo->numOfPseudoExpr > 0) {
@@ -2177,9 +2180,8 @@ int32_t doOpenTableMergeScanOperator(SOperatorInfo* pOperator) {
 
   int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;
 
-  pInfo->pSortHandle =
-      tsortCreateSortHandle(pInfo->pSortInfo, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize,
-                            numOfBufPage, pInfo->pSortInputBlock, pTaskInfo->id.str);
+  pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize, numOfBufPage,
+                                             pInfo->pSortInputBlock, pTaskInfo->id.str);
 
   tsortSetFetchRawDataFp(pInfo->pSortHandle, getTableDataBlock, NULL, NULL);
 
@@ -2243,8 +2245,7 @@ SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {
     longjmp(pTaskInfo->env, code);
   }
 
-  SSDataBlock* pBlock =
-      getSortedTableMergeScanBlockData(pInfo->pSortHandle, pOperator->resultInfo.capacity, pOperator);
+  SSDataBlock* pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pOperator->resultInfo.capacity, pOperator);
 
   if (pBlock != NULL) {
     pOperator->resultInfo.totalRows += pBlock->info.rows;
@@ -2305,8 +2306,7 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN
   SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;
 
   int32_t numOfCols = 0;
-  SArray* pColList =
-      extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID);
+  SArray* pColList = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID);
 
   int32_t code = initQueryTableDataCond(&pInfo->cond, pTableScanNode);
   if (code != TSDB_CODE_SUCCESS) {

source/libs/function/src/builtinsimpl.c

@@ -2562,9 +2562,6 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
 }
 
 static void firstLastTransferInfo(SFirstLastRes* pInput, SFirstLastRes* pOutput, bool isFirst) {
-  if (!pInput->hasResult) {
-    return;
-  }
   pOutput->bytes = pInput->bytes;
   TSKEY* tsIn = (TSKEY*)(pInput->buf + pInput->bytes);
   TSKEY* tsOut = (TSKEY*)(pOutput->buf + pInput->bytes);
@@ -2598,7 +2595,9 @@ static int32_t firstLastFunctionMergeImpl(SqlFunctionCtx* pCtx, bool isFirstQuer
 
   firstLastTransferInfo(pInputInfo, pInfo, isFirstQuery);
 
-  SET_VAL(GET_RES_INFO(pCtx), 1, 1);
+  int32_t numOfElems = pInputInfo->hasResult ? 1 : 0;
+
+  SET_VAL(GET_RES_INFO(pCtx), numOfElems, 1);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -2623,6 +2622,7 @@ int32_t firstLastFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
 int32_t firstLastPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(pCtx);
   SFirstLastRes*       pRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
+
   int32_t              resultBytes = getFirstLastInfoSize(pRes->bytes);
   char*                res = taosMemoryCalloc(resultBytes + VARSTR_HEADER_SIZE, sizeof(char));
 

source/libs/function/src/tudf.c

@@ -972,6 +972,11 @@ void releaseUdfFuncHandle(char* udfName) {
 }
 
 int32_t cleanUpUdfs() {
+  int8_t initialized = atomic_load_8(&gUdfdProxy.initialized);
+  if (!initialized) {
+    return TSDB_CODE_SUCCESS;
+  }
+
   uv_mutex_lock(&gUdfdProxy.udfStubsMutex);
   int32_t i = 0;
   SArray* udfStubs = taosArrayInit(16, sizeof(SUdfcFuncStub));

source/libs/planner/src/planOptimizer.c

@@ -1082,13 +1082,89 @@ static int32_t partTagsOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLogicSub
   return code;
 }
 
+static bool eliminateProjOptMayBeOptimized(SLogicNode* pNode) {
+  //TODO: enable this optimization after new mechanising that map projection and targets of project node
+  if (NULL != pNode->pParent) {
+    return false;
+  }
+  
+  if (QUERY_NODE_LOGIC_PLAN_PROJECT != nodeType(pNode) || 1 != LIST_LENGTH(pNode->pChildren)) {
+    return false;
+  }
+
+  SProjectLogicNode* pProjectNode = (SProjectLogicNode*)pNode;
+  if (-1 != pProjectNode->limit || -1 != pProjectNode->slimit || -1 != pProjectNode->offset  || -1 != pProjectNode->soffset) {
+    return false;
+  }
+
+  SHashObj* pProjColNameHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
+  SNode* pProjection;
+  FOREACH(pProjection, pProjectNode->pProjections) {
+    SExprNode* pExprNode = (SExprNode*)pProjection;
+    if (QUERY_NODE_COLUMN != nodeType(pExprNode)) {
+      taosHashCleanup(pProjColNameHash);
+      return false;
+    }
+
+    char* projColumnName = ((SColumnNode*)pProjection)->colName;
+    int32_t* pExist = taosHashGet(pProjColNameHash, projColumnName, strlen(projColumnName));
+    if (NULL != pExist) {
+      taosHashCleanup(pProjColNameHash);
+      return false;
+    } else {
+      int32_t exist = 1;
+      taosHashPut(pProjColNameHash, projColumnName, strlen(projColumnName), &exist, sizeof(exist));
+    }
+  }
+  taosHashCleanup(pProjColNameHash);
+
+  return true;
+}
+
+static int32_t eliminateProjOptimizeImpl(SOptimizeContext* pCxt, SLogicSubplan* pLogicSubplan, SProjectLogicNode* pProjectNode) {
+  SLogicNode* pChild = (SLogicNode*)nodesListGetNode(pProjectNode->node.pChildren, 0);
+  SNodeList* pNewChildTargets = nodesMakeList();
+
+  SNode* pProjection = NULL;
+  FOREACH(pProjection, pProjectNode->pProjections) {
+    SNode* pChildTarget = NULL;
+    FOREACH(pChildTarget, pChild->pTargets) {
+      if (strcmp(((SColumnNode*)pProjection)->colName, ((SColumnNode*)pChildTarget)->colName) == 0) {
+        nodesListAppend(pNewChildTargets, nodesCloneNode(pChildTarget));
+        break;
+      }
+    }
+  }
+  nodesDestroyList(pChild->pTargets);
+  pChild->pTargets = pNewChildTargets;
+  
+  int32_t code = replaceLogicNode(pLogicSubplan, (SLogicNode*)pProjectNode, pChild);
+  if (TSDB_CODE_SUCCESS == code) {
+    NODES_CLEAR_LIST(pProjectNode->node.pChildren);
+    nodesDestroyNode((SNode*)pProjectNode);
+  }
+  return code;
+}
+
+static int32_t eliminateProjOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLogicSubplan) {
+  SProjectLogicNode* pProjectNode =
+    (SProjectLogicNode*)optFindPossibleNode(pLogicSubplan->pNode, eliminateProjOptMayBeOptimized);
+
+  if (NULL == pProjectNode) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  return eliminateProjOptimizeImpl(pCxt, pLogicSubplan, pProjectNode);
+}
+
 // clang-format off
 static const SOptimizeRule optimizeRuleSet[] = {
   {.pName = "OptimizeScanData",  .optimizeFunc = osdOptimize},
   {.pName = "ConditionPushDown", .optimizeFunc = cpdOptimize},
   {.pName = "OrderByPrimaryKey", .optimizeFunc = opkOptimize},
   {.pName = "SmaIndex",          .optimizeFunc = smaOptimize},
-  {.pName = "PartitionByTags",   .optimizeFunc = partTagsOptimize}
+  {.pName = "PartitionByTags",   .optimizeFunc = partTagsOptimize},
+  {.pName = "EliminateProject",  .optimizeFunc = eliminateProjOptimize}
 };
 // clang-format on
 

source/libs/planner/src/planUtil.c

@@ -107,6 +107,7 @@ int32_t createColumnByRewriteExpr(SNode* pExpr, SNodeList** pList) {
 int32_t replaceLogicNode(SLogicSubplan* pSubplan, SLogicNode* pOld, SLogicNode* pNew) {
   if (NULL == pOld->pParent) {
     pSubplan->pNode = (SLogicNode*)pNew;
+    pNew->pParent = NULL;
     return TSDB_CODE_SUCCESS;
   }
 

source/libs/planner/test/planOptimizeTest.cpp

@@ -52,3 +52,13 @@ TEST_F(PlanOptimizeTest, orderByPrimaryKey) {
 
   run("SELECT COUNT(*) FROM t1 INTERVAL(10S) ORDER BY _WSTARTTS DESC");
 }
+
+TEST_F(PlanOptimizeTest, eliminateProjection) {
+  useDb("root", "test");
+
+  run("SELECT c1, sum(c3) FROM t1 GROUP BY c1");
+  run("SELECT c1 FROM t1");
+  run("SELECT * FROM st1");
+  run("SELECT c1 FROM st1s3");
+  //run("select 1-abs(c1) from (select unique(c1) c1 from st1s3) order by 1 nulls first");
+}

source/libs/qworker/src/qworker.c

@@ -533,7 +533,7 @@ int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, int8_t taskType, int8_t ex
   atomic_store_8(&ctx->taskType, taskType);
   atomic_store_8(&ctx->explain, explain);
 
-  /*QW_TASK_DLOGL("subplan json string, len:%d, %s", qwMsg->msgLen, qwMsg->msg);*/
+  QW_TASK_DLOGL("subplan json string, len:%d, %s", qwMsg->msgLen, qwMsg->msg);
 
   code = qStringToSubplan(qwMsg->msg, &plan);
   if (TSDB_CODE_SUCCESS != code) {

source/libs/stream/src/streamTask.c

@@ -36,11 +36,11 @@ int32_t tEncodeSStreamTask(SEncoder* pEncoder, const SStreamTask* pTask) {
   if (tEncodeI32(pEncoder, pTask->taskId) < 0) return -1;
   if (tEncodeI8(pEncoder, pTask->inputType) < 0) return -1;
   if (tEncodeI8(pEncoder, pTask->status) < 0) return -1;
-  if (tEncodeI8(pEncoder, pTask->sourceType) < 0) return -1;
   if (tEncodeI8(pEncoder, pTask->execType) < 0) return -1;
   if (tEncodeI8(pEncoder, pTask->sinkType) < 0) return -1;
   if (tEncodeI8(pEncoder, pTask->dispatchType) < 0) return -1;
   if (tEncodeI16(pEncoder, pTask->dispatchMsgType) < 0) return -1;
+  if (tEncodeI8(pEncoder, pTask->dataScan) < 0) return -1;
 
   if (tEncodeI32(pEncoder, pTask->childId) < 0) return -1;
   if (tEncodeI32(pEncoder, pTask->nodeId) < 0) return -1;
@@ -84,11 +84,11 @@ int32_t tDecodeSStreamTask(SDecoder* pDecoder, SStreamTask* pTask) {
   if (tDecodeI32(pDecoder, &pTask->taskId) < 0) return -1;
   if (tDecodeI8(pDecoder, &pTask->inputType) < 0) return -1;
   if (tDecodeI8(pDecoder, &pTask->status) < 0) return -1;
-  if (tDecodeI8(pDecoder, &pTask->sourceType) < 0) return -1;
   if (tDecodeI8(pDecoder, &pTask->execType) < 0) return -1;
   if (tDecodeI8(pDecoder, &pTask->sinkType) < 0) return -1;
   if (tDecodeI8(pDecoder, &pTask->dispatchType) < 0) return -1;
   if (tDecodeI16(pDecoder, &pTask->dispatchMsgType) < 0) return -1;
+  if (tDecodeI8(pDecoder, &pTask->dataScan) < 0) return -1;
 
   if (tDecodeI32(pDecoder, &pTask->childId) < 0) return -1;
   if (tDecodeI32(pDecoder, &pTask->nodeId) < 0) return -1;

source/libs/tdb/src/db/tdbPage.c

@@ -82,7 +82,8 @@ int tdbPageDestroy(SPage *pPage, void (*xFree)(void *arg, void *ptr), void *arg)
   return 0;
 }
 
-void tdbPageZero(SPage *pPage, u8 szAmHdr, int (*xCellSize)(const SPage *, SCell *)) {
+void tdbPageZero(SPage *pPage, u8 szAmHdr, int (*xCellSize)(const SPage *, SCell *, int,
+							    TXN *, SBTree *pBt)) {
   pPage->pPageHdr = pPage->pData + szAmHdr;
   TDB_PAGE_NCELLS_SET(pPage, 0);
   TDB_PAGE_CCELLS_SET(pPage, pPage->pageSize - sizeof(SPageFtr));
@@ -98,7 +99,8 @@ void tdbPageZero(SPage *pPage, u8 szAmHdr, int (*xCellSize)(const SPage *, SCell
   ASSERT((u8 *)pPage->pPageFtr == pPage->pFreeEnd);
 }
 
-void tdbPageInit(SPage *pPage, u8 szAmHdr, int (*xCellSize)(const SPage *, SCell *)) {
+void tdbPageInit(SPage *pPage, u8 szAmHdr, int (*xCellSize)(const SPage *, SCell *, int,
+							    TXN *, SBTree *pBt)) {
   pPage->pPageHdr = pPage->pData + szAmHdr;
   pPage->pCellIdx = pPage->pPageHdr + TDB_PAGE_HDR_SIZE(pPage);
   pPage->pFreeStart = pPage->pCellIdx + TDB_PAGE_OFFSET_SIZE(pPage) * TDB_PAGE_NCELLS(pPage);
@@ -171,12 +173,12 @@ int tdbPageInsertCell(SPage *pPage, int idx, SCell *pCell, int szCell, u8 asOvfl
   return 0;
 }
 
-int tdbPageUpdateCell(SPage *pPage, int idx, SCell *pCell, int szCell) {
-  tdbPageDropCell(pPage, idx);
+int tdbPageUpdateCell(SPage *pPage, int idx, SCell *pCell, int szCell, TXN *pTxn, SBTree *pBt) {
+  tdbPageDropCell(pPage, idx, pTxn, pBt);
   return tdbPageInsertCell(pPage, idx, pCell, szCell, 0);
 }
 
-int tdbPageDropCell(SPage *pPage, int idx) {
+int tdbPageDropCell(SPage *pPage, int idx, TXN *pTxn, SBTree *pBt) {
   int    lidx;
   SCell *pCell;
   int    szCell;
@@ -205,7 +207,7 @@ int tdbPageDropCell(SPage *pPage, int idx) {
 
   lidx = idx - iOvfl;
   pCell = TDB_PAGE_CELL_AT(pPage, lidx);
-  szCell = (*pPage->xCellSize)(pPage, pCell);
+  szCell = (*pPage->xCellSize)(pPage, pCell, 1, pTxn, pBt);
   tdbPageFree(pPage, lidx, pCell, szCell);
   TDB_PAGE_NCELLS_SET(pPage, nCells - 1);
 
@@ -420,7 +422,7 @@ static int tdbPageDefragment(SPage *pPage) {
 
     ASSERT(pCell != NULL);
 
-    szCell = (*pPage->xCellSize)(pPage, pCell);
+    szCell = (*pPage->xCellSize)(pPage, pCell, 0, NULL, NULL);
 
     ASSERT(pCell + szCell <= pNextCell);
     if (pCell + szCell < pNextCell) {

source/libs/tdb/src/inc/tdbInt.h

@@ -116,13 +116,25 @@ typedef struct SBtInfo {
   int   nData;
 } SBtInfo;
 
+#define TDB_CELLD_F_NIL 0x0
+#define TDB_CELLD_F_KEY 0x1
+#define TDB_CELLD_F_VAL 0x2
+
+#define TDB_CELLDECODER_SET_FREE_NIL(pCellDecoder) ((pCellDecoder)->freeKV = TDB_CELLD_F_NIL)
+#define TDB_CELLDECODER_SET_FREE_KEY(pCellDecoder) ((pCellDecoder)->freeKV |= TDB_CELLD_F_KEY)
+#define TDB_CELLDECODER_SET_FREE_VAL(pCellDecoder) ((pCellDecoder)->freeKV |= TDB_CELLD_F_VAL)
+
+#define TDB_CELLDECODER_FREE_KEY(pCellDecoder) ((pCellDecoder)->freeKV & TDB_CELLD_F_KEY)
+#define TDB_CELLDECODER_FREE_VAL(pCellDecoder) ((pCellDecoder)->freeKV & TDB_CELLD_F_VAL)
+
 typedef struct {
   int       kLen;
-  const u8 *pKey;
+  u8       *pKey;
   int       vLen;
-  const u8 *pVal;
+  u8       *pVal;
   SPgno     pgno;
   u8       *pBuf;
+  u8        freeKV;
 } SCellDecoder;
 
 struct SBTC {
@@ -251,7 +263,7 @@ struct SPage {
   int       vLen;  // value length of the page, -1 for unknown
   int       maxLocal;
   int       minLocal;
-  int (*xCellSize)(const SPage *, SCell *);
+  int (*xCellSize)(const SPage *, SCell *, int, TXN *pTxn, SBTree *pBt);
   // Fields used by SPCache
   TDB_PCACHE_PAGE
 };
@@ -298,16 +310,18 @@ static inline int tdbTryLockPage(tdb_spinlock_t *pLock) {
 #define TDB_PAGE_USABLE_SIZE(pPage)        ((u8 *)(pPage)->pPageFtr - (pPage)->pCellIdx)
 #define TDB_PAGE_FREE_SIZE(pPage)          (*(pPage)->pPageMethods->getFreeBytes)(pPage)
 #define TDB_PAGE_PGNO(pPage)               ((pPage)->pgid.pgno)
-#define TDB_BYTES_CELL_TAKEN(pPage, pCell) ((*(pPage)->xCellSize)(pPage, pCell) + (pPage)->pPageMethods->szOffset)
+#define TDB_BYTES_CELL_TAKEN(pPage, pCell) ((*(pPage)->xCellSize)(pPage, pCell, 0, NULL, NULL) + (pPage)->pPageMethods->szOffset)
 #define TDB_PAGE_OFFSET_SIZE(pPage)        ((pPage)->pPageMethods->szOffset)
 
 int  tdbPageCreate(int pageSize, SPage **ppPage, void *(*xMalloc)(void *, size_t), void *arg);
 int  tdbPageDestroy(SPage *pPage, void (*xFree)(void *arg, void *ptr), void *arg);
-void tdbPageZero(SPage *pPage, u8 szAmHdr, int (*xCellSize)(const SPage *, SCell *));
-void tdbPageInit(SPage *pPage, u8 szAmHdr, int (*xCellSize)(const SPage *, SCell *));
+void tdbPageZero(SPage *pPage, u8 szAmHdr, int (*xCellSize)(const SPage *, SCell *, int,
+							    TXN *, SBTree *pBt));
+void tdbPageInit(SPage *pPage, u8 szAmHdr, int (*xCellSize)(const SPage *, SCell *, int,
+							    TXN *, SBTree *pBt));
 int  tdbPageInsertCell(SPage *pPage, int idx, SCell *pCell, int szCell, u8 asOvfl);
-int  tdbPageDropCell(SPage *pPage, int idx);
-int  tdbPageUpdateCell(SPage *pPage, int idx, SCell *pCell, int szCell);
+int  tdbPageDropCell(SPage *pPage, int idx, TXN *pTxn, SBTree *pBt);
+int  tdbPageUpdateCell(SPage *pPage, int idx, SCell *pCell, int szCell, TXN *pTxn, SBTree *pBt);
 void tdbPageCopy(SPage *pFromPage, SPage *pToPage);
 int  tdbPageCapacity(int pageSize, int amHdrSize);
 

source/libs/tdb/test/CMakeLists.txt

@@ -5,3 +5,8 @@ target_link_libraries(tdbTest tdb gtest gtest_main)
 # tdbUtilTest
 add_executable(tdbUtilTest "tdbUtilTest.cpp")
 target_link_libraries(tdbUtilTest tdb gtest gtest_main)
+
+# tdbUtilTest
+add_executable(tdbExOVFLTest "tdbExOVFLTest.cpp")
+target_link_libraries(tdbExOVFLTest tdb gtest gtest_main)
+

source/libs/tdb/test/tdbExOVFLTest.cpp

+#include <gtest/gtest.h>
+
+#define ALLOW_FORBID_FUNC
+#include "os.h"
+#include "tdb.h"
+
+#include <shared_mutex>
+#include <string>
+#include <thread>
+#include <vector>
+
+typedef struct SPoolMem {
+  int64_t          size;
+  struct SPoolMem *prev;
+  struct SPoolMem *next;
+} SPoolMem;
+
+static SPoolMem *openPool() {
+  SPoolMem *pPool = (SPoolMem *)taosMemoryMalloc(sizeof(*pPool));
+
+  pPool->prev = pPool->next = pPool;
+  pPool->size = 0;
+
+  return pPool;
+}
+
+static void clearPool(SPoolMem *pPool) {
+  SPoolMem *pMem;
+
+  do {
+    pMem = pPool->next;
+
+    if (pMem == pPool) break;
+
+    pMem->next->prev = pMem->prev;
+    pMem->prev->next = pMem->next;
+    pPool->size -= pMem->size;
+
+    taosMemoryFree(pMem);
+  } while (1);
+
+  assert(pPool->size == 0);
+}
+
+static void closePool(SPoolMem *pPool) {
+  clearPool(pPool);
+  taosMemoryFree(pPool);
+}
+
+static void *poolMalloc(void *arg, size_t size) {
+  void     *ptr = NULL;
+  SPoolMem *pPool = (SPoolMem *)arg;
+  SPoolMem *pMem;
+
+  pMem = (SPoolMem *)taosMemoryMalloc(sizeof(*pMem) + size);
+  if (pMem == NULL) {
+    assert(0);
+  }
+
+  pMem->size = sizeof(*pMem) + size;
+  pMem->next = pPool->next;
+  pMem->prev = pPool;
+
+  pPool->next->prev = pMem;
+  pPool->next = pMem;
+  pPool->size += pMem->size;
+
+  ptr = (void *)(&pMem[1]);
+  return ptr;
+}
+
+static void poolFree(void *arg, void *ptr) {
+  SPoolMem *pPool = (SPoolMem *)arg;
+  SPoolMem *pMem;
+
+  pMem = &(((SPoolMem *)ptr)[-1]);
+
+  pMem->next->prev = pMem->prev;
+  pMem->prev->next = pMem->next;
+  pPool->size -= pMem->size;
+
+  taosMemoryFree(pMem);
+}
+
+static int tKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2) {
+  int k1, k2;
+
+  std::string s1((char *)pKey1 + 3, kLen1 - 3);
+  std::string s2((char *)pKey2 + 3, kLen2 - 3);
+  k1 = stoi(s1);
+  k2 = stoi(s2);
+
+  if (k1 < k2) {
+    return -1;
+  } else if (k1 > k2) {
+    return 1;
+  } else {
+    return 0;
+  }
+}
+
+static int tDefaultKeyCmpr(const void *pKey1, int keyLen1, const void *pKey2, int keyLen2) {
+  int mlen;
+  int cret;
+
+  ASSERT(keyLen1 > 0 && keyLen2 > 0 && pKey1 != NULL && pKey2 != NULL);
+
+  mlen = keyLen1 < keyLen2 ? keyLen1 : keyLen2;
+  cret = memcmp(pKey1, pKey2, mlen);
+  if (cret == 0) {
+    if (keyLen1 < keyLen2) {
+      cret = -1;
+    } else if (keyLen1 > keyLen2) {
+      cret = 1;
+    } else {
+      cret = 0;
+    }
+  }
+  return cret;
+}
+
+TEST(TdbOVFLPagesTest, TbUpsertTest) {
+
+}
+
+TEST(TdbOVFLPagesTest, TbPGetTest) {
+
+}
+
+static void generateBigVal(char *val, int valLen) {
+  for (int i = 0; i < valLen; ++i) {
+    char c = char(i & 0xff);
+    if (c == 0) {
+      c = 1;
+    }
+    val[i] = c;
+  }
+}
+
+static TDB *openEnv(char const *envName, int const pageSize, int const pageNum) {
+  TDB *pEnv = NULL;
+
+  int ret = tdbOpen(envName, pageSize, pageNum, &pEnv);
+  if (ret) {
+    pEnv = NULL;
+  }
+
+  return pEnv;
+}
+
+static void insertOfp(void) {
+  int ret = 0;
+
+  taosRemoveDir("tdb");
+
+  // open Env
+  int const pageSize = 4096;
+  int const pageNum = 64;
+  TDB *pEnv = openEnv("tdb", pageSize, pageNum);
+  GTEST_ASSERT_NE(pEnv, nullptr);
+
+  // open db
+  TTB *pDb = NULL;
+  tdb_cmpr_fn_t compFunc = tKeyCmpr;
+  ret = tdbTbOpen("ofp_insert.db", -1, -1, compFunc, pEnv, &pDb);
+  GTEST_ASSERT_EQ(ret, 0);
+
+  // open the pool
+  SPoolMem *pPool = openPool();
+
+  // start a transaction
+  TXN txn;
+  int64_t txnid = 0;
+  ++txnid;
+  tdbTxnOpen(&txn, txnid, poolMalloc, poolFree, pPool, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED);
+  tdbBegin(pEnv, &txn);
+
+  // generate value payload
+  char val[((4083 - 4 - 3 - 2)+1)*100]; // pSize(4096) - amSize(1) - pageHdr(8) - footerSize(4)
+  int valLen = sizeof(val) / sizeof(val[0]);
+  generateBigVal(val, valLen);
+
+  // insert the generated big data
+  ret = tdbTbInsert(pDb, "key1", strlen("key1"), val, valLen, &txn);
+  GTEST_ASSERT_EQ(ret, 0);
+
+  // commit current transaction
+  tdbCommit(pEnv, &txn);
+  tdbTxnClose(&txn);
+}
+
+//TEST(TdbOVFLPagesTest, DISABLED_TbInsertTest) {
+TEST(TdbOVFLPagesTest, TbInsertTest) {
+  insertOfp();
+}
+
+//TEST(TdbOVFLPagesTest, DISABLED_TbGetTest) {
+TEST(TdbOVFLPagesTest, TbGetTest) {
+  insertOfp();
+
+  // open Env
+  int const pageSize = 4096;
+  int const pageNum = 64;
+  TDB *pEnv = openEnv("tdb", pageSize, pageNum);
+  GTEST_ASSERT_NE(pEnv, nullptr);
+
+  // open db
+  TTB *pDb = NULL;
+  tdb_cmpr_fn_t compFunc = tKeyCmpr;
+  int ret = tdbTbOpen("ofp_insert.db", -1, -1, compFunc, pEnv, &pDb);
+  GTEST_ASSERT_EQ(ret, 0);
+
+  // generate value payload
+  char val[((4083 - 4 - 3 - 2)+1)*100]; // pSize(4096) - amSize(1) - pageHdr(8) - footerSize(4)
+  int valLen = sizeof(val) / sizeof(val[0]);
+  generateBigVal(val, valLen);
+
+  {  // Query the data
+    void *pVal = NULL;
+    int   vLen;
+
+    ret = tdbTbGet(pDb, "key1", strlen("key1"), &pVal, &vLen);
+    ASSERT(ret == 0);
+    GTEST_ASSERT_EQ(ret, 0);
+
+    GTEST_ASSERT_EQ(vLen, valLen);
+    GTEST_ASSERT_EQ(memcmp(val, pVal, vLen), 0);
+
+    tdbFree(pVal);
+  }
+}
+
+TEST(TdbOVFLPagesTest, TbDeleteTest) {
+  int ret = 0;
+
+  taosRemoveDir("tdb");
+
+  // open Env
+  int const pageSize = 4096;
+  int const pageNum = 64;
+  TDB *pEnv = openEnv("tdb", pageSize, pageNum);
+  GTEST_ASSERT_NE(pEnv, nullptr);
+
+  // open db
+  TTB *pDb = NULL;
+  tdb_cmpr_fn_t compFunc = tKeyCmpr;
+  ret = tdbTbOpen("ofp_insert.db", -1, -1, compFunc, pEnv, &pDb);
+  GTEST_ASSERT_EQ(ret, 0);
+
+  // open the pool
+  SPoolMem *pPool = openPool();
+
+  // start a transaction
+  TXN txn;
+  int64_t txnid = 0;
+  ++txnid;
+  tdbTxnOpen(&txn, txnid, poolMalloc, poolFree, pPool, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED);
+  tdbBegin(pEnv, &txn);
+
+  // generate value payload
+  char val[((4083 - 4 - 3 - 2)+1)*100]; // pSize(4096) - amSize(1) - pageHdr(8) - footerSize(4)
+  int valLen = sizeof(val) / sizeof(val[0]);
+  generateBigVal(val, valLen);
+
+  { // insert the generated big data
+    ret = tdbTbInsert(pDb, "key1", strlen("key1"), val, valLen, &txn);
+    GTEST_ASSERT_EQ(ret, 0);
+  }
+
+  {  // query the data
+    void *pVal = NULL;
+    int   vLen;
+
+    ret = tdbTbGet(pDb, "key1", strlen("key1"), &pVal, &vLen);
+    ASSERT(ret == 0);
+    GTEST_ASSERT_EQ(ret, 0);
+
+    GTEST_ASSERT_EQ(vLen, valLen);
+    GTEST_ASSERT_EQ(memcmp(val, pVal, vLen), 0);
+
+    tdbFree(pVal);
+  }
+    /* open to debug committed file
+  tdbCommit(pEnv, &txn);
+  tdbTxnClose(&txn);
+
+  ++txnid;
+  tdbTxnOpen(&txn, txnid, poolMalloc, poolFree, pPool, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED);
+  tdbBegin(pEnv, &txn);
+    */
+  { // upsert the data
+    ret = tdbTbUpsert(pDb, "key1", strlen("key1"), "value1", strlen("value1"), &txn);
+    GTEST_ASSERT_EQ(ret, 0);
+  }
+
+  {  // query the upserted data
+    void *pVal = NULL;
+    int   vLen;
+
+    ret = tdbTbGet(pDb, "key1", strlen("key1"), &pVal, &vLen);
+    ASSERT(ret == 0);
+    GTEST_ASSERT_EQ(ret, 0);
+
+    GTEST_ASSERT_EQ(vLen, strlen("value1"));
+    GTEST_ASSERT_EQ(memcmp("value1", pVal, vLen), 0);
+
+    tdbFree(pVal);
+  }
+
+  { // delete the data
+    ret = tdbTbDelete(pDb, "key1", strlen("key1"), &txn);
+    GTEST_ASSERT_EQ(ret, 0);
+  }
+
+  {  // query the deleted data
+    void *pVal = NULL;
+    int   vLen = -1;
+
+    ret = tdbTbGet(pDb, "key1", strlen("key1"), &pVal, &vLen);
+    ASSERT(ret == -1);
+    GTEST_ASSERT_EQ(ret, -1);
+
+    GTEST_ASSERT_EQ(vLen, -1);
+    GTEST_ASSERT_EQ(pVal, nullptr);
+
+    tdbFree(pVal);
+  }
+
+  // commit current transaction
+  tdbCommit(pEnv, &txn);
+  tdbTxnClose(&txn);
+}
+
+TEST(tdb_test, DISABLED_simple_insert1) {
+//TEST(tdb_test, simple_insert1) {
+  int           ret;
+  TDB          *pEnv;
+  TTB          *pDb;
+  tdb_cmpr_fn_t compFunc;
+  int           nData = 1;
+  TXN           txn;
+  int const     pageSize = 4096;
+
+  taosRemoveDir("tdb");
+
+  // Open Env
+  ret = tdbOpen("tdb", pageSize, 64, &pEnv);
+  GTEST_ASSERT_EQ(ret, 0);
+
+  // Create a database
+  compFunc = tKeyCmpr;
+  ret = tdbTbOpen("db.db", -1, -1, compFunc, pEnv, &pDb);
+  GTEST_ASSERT_EQ(ret, 0);
+
+  {
+    char      key[64];
+    //char      val[(4083 - 4 - 3 - 2)]; // pSize(4096) - amSize(1) - pageHdr(8) - footerSize(4)
+    char      val[(4083 - 4 - 3 - 2)+1]; // pSize(4096) - amSize(1) - pageHdr(8) - footerSize(4)
+    int64_t   poolLimit = 4096;  // 1M pool limit
+    int64_t   txnid = 0;
+    SPoolMem *pPool;
+
+    // open the pool
+    pPool = openPool();
+
+    // start a transaction
+    txnid++;
+    tdbTxnOpen(&txn, txnid, poolMalloc, poolFree, pPool, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED);
+    tdbBegin(pEnv, &txn);
+
+    for (int iData = 1; iData <= nData; iData++) {
+      sprintf(key, "key0");
+      sprintf(val, "value%d", iData);
+
+      //ret = tdbTbInsert(pDb, key, strlen(key), val, strlen(val), &txn);
+      //GTEST_ASSERT_EQ(ret, 0);
+
+      // generate value payload
+      int valLen = sizeof(val) / sizeof(val[0]);
+      for (int i = 6; i < valLen; ++i) {
+	char c = char(i & 0xff);
+	if (c == 0) {
+	  c = 1;
+	}
+	val[i] = c;
+      }
+
+      ret = tdbTbInsert(pDb, "key1", strlen("key1"), val, valLen, &txn);
+      GTEST_ASSERT_EQ(ret, 0);
+
+      // if pool is full, commit the transaction and start a new one
+      if (pPool->size >= poolLimit) {
+        // commit current transaction
+        tdbCommit(pEnv, &txn);
+        tdbTxnClose(&txn);
+
+        // start a new transaction
+        clearPool(pPool);
+        txnid++;
+        tdbTxnOpen(&txn, txnid, poolMalloc, poolFree, pPool, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED);
+        tdbBegin(pEnv, &txn);
+      }
+    }
+
+    // commit the transaction
+    tdbCommit(pEnv, &txn);
+    tdbTxnClose(&txn);
+
+    {  // Query the data
+      void *pVal = NULL;
+      int   vLen;
+
+      for (int i = 1; i <= nData; i++) {
+        sprintf(key, "key%d", i);
+        sprintf(val, "value%d", i);
+
+        ret = tdbTbGet(pDb, key, strlen(key), &pVal, &vLen);
+        ASSERT(ret == 0);
+        GTEST_ASSERT_EQ(ret, 0);
+
+        GTEST_ASSERT_EQ(vLen, strlen(val));
+        GTEST_ASSERT_EQ(memcmp(val, pVal, vLen), 0);
+      }
+
+      tdbFree(pVal);
+    }
+
+    {  // Iterate to query the DB data
+      TBC  *pDBC;
+      void *pKey = NULL;
+      void *pVal = NULL;
+      int   vLen, kLen;
+      int   count = 0;
+
+      ret = tdbTbcOpen(pDb, &pDBC, NULL);
+      GTEST_ASSERT_EQ(ret, 0);
+
+      tdbTbcMoveToFirst(pDBC);
+
+      for (;;) {
+        ret = tdbTbcNext(pDBC, &pKey, &kLen, &pVal, &vLen);
+        if (ret < 0) break;
+
+        // std::cout.write((char *)pKey, kLen) /* << " " << kLen */ << " ";
+        // std::cout.write((char *)pVal, vLen) /* << " " << vLen */;
+        // std::cout << std::endl;
+
+        count++;
+      }
+
+      GTEST_ASSERT_EQ(count, nData);
+
+      tdbTbcClose(pDBC);
+
+      tdbFree(pKey);
+      tdbFree(pVal);
+    }
+  }
+
+  ret = tdbTbDrop(pDb);
+  GTEST_ASSERT_EQ(ret, 0);
+
+  // Close a database
+  tdbTbClose(pDb);
+
+  // Close Env
+  ret = tdbClose(pEnv);
+  GTEST_ASSERT_EQ(ret, 0);
+}

tests/script/jenkins/basic.txt

@@ -72,14 +72,14 @@
 ./test.sh -f tsim/stream/basic0.sim
 ./test.sh -f tsim/stream/basic1.sim
 ./test.sh -f tsim/stream/basic2.sim
-# ./test.sh -f tsim/stream/distributeInterval0.sim
+./test.sh -f tsim/stream/distributeInterval0.sim
 # ./test.sh -f tsim/stream/distributesession0.sim
 # ./test.sh -f tsim/stream/session0.sim
 # ./test.sh -f tsim/stream/session1.sim
 # ./test.sh -f tsim/stream/state0.sim
-# ./test.sh -f tsim/stream/triggerInterval0.sim
+./test.sh -f tsim/stream/triggerInterval0.sim
 # ./test.sh -f tsim/stream/triggerSession0.sim
-# ./test.sh -f tsim/stream/partitionby.sim
+./test.sh -f tsim/stream/partitionby.sim
 
 
 # ---- transaction

tests/script/tsim/dnode/redistribute_vgroup_replica3_v1_leader.sim

@@ -53,31 +53,10 @@ endi
 if $data(4)[4] != ready then
   goto step1
 endi
-#if $data(5)[4] != ready then
-#  goto step1
-#endi
 
 print =============== step2: create db
 sql create database d1 vgroups 1 replica 3
 
-# dnode not exist
-sql_error redistribute vgroup 3 dnode 6 dnode 3 dnode 4 
-# vgroup not exist
-sql_error redistribute vgroup 3 dnode 5 dnode 3 dnode 4 
-# un changed
-sql_error redistribute vgroup 2 dnode 2 dnode 3 dnode 4
-# no enought vnodes
-sql_error redistribute vgroup 2 dnode 1 dnode 3 dnode 4 
-# offline vnodes
-sql_error redistribute vgroup 2 dnode 5 dnode 3 dnode 4 
-# Invalid replica
-sql_error redistribute vgroup 2 dnode 5 
-sql_error redistribute vgroup 2 dnode 5 dnode 3 
-sql_error redistribute vgroup 2 dnode 2 dnode 3 
-sql_error redistribute vgroup 2 dnode 2 dnode 2 
-sql_error redistribute vgroup 3 dnode 2 dnode 2 
-sql_error redistribute vgroup 2 dnode 2 dnode 2 dnode 3 
-
 system sh/exec.sh -n dnode5 -s start
 $x = 0
 step2:

tests/system-test/1-insert/create_table_comment.py

-###################################################################
-#           Copyright (c) 2016 by TAOS Technologies, Inc.
-#                     All rights reserved.
-#
-#  This file is proprietary and confidential to TAOS Technologies.
-#  No part of this file may be reproduced, stored, transmitted,
-#  disclosed or used in any form or by any means other than as
-#  expressly provided by the written permission from Jianhui Tao
-#
-###################################################################
-
-# -*- coding: utf-8 -*-
-
-import random
-import string
-from util.log import *
-from util.cases import *
-from util.sql import *
-
-class TDTestCase:
-    def init(self, conn, logSql):
-        tdLog.debug("start to execute %s" % __file__)
-        tdSql.init(conn.cursor())
-
-    def get_long_name(self, length, mode="mixed"):
-        """
-        generate long name
-        mode could be numbers/letters/letters_mixed/mixed
-        """
-        if mode == "numbers":
-            population = string.digits
-        elif mode == "letters":
-            population = string.ascii_letters.lower()
-        elif mode == "letters_mixed":
-            population = string.ascii_letters.upper() + string.ascii_letters.lower()
-        else:
-            population = string.ascii_letters.lower() + string.digits
-        return "".join(random.choices(population, k=length))
-
-    def __create_tb(self,dbname,stbname,tbname,comment):
-        tdSql.execute(f'create database if not exists {dbname}')
-        tdSql.execute(f'use {dbname}')
-        tdSql.execute(
-            f'create table {stbname} (ts timestamp,c0 int) tags(t0 int) ')
-        tdSql.execute(
-            f'create table {tbname} using {stbname} tags(1) comment "{comment}"')
-    def __create_normaltb(self,dbname,tbname,comment):
-        tdSql.execute(f'create database if not exists {dbname}')
-        tdSql.execute(f'use {dbname}')
-        tdSql.execute(
-            f'create table {tbname} (ts timestamp,c0 int) comment "{comment}"')
-
-    def check_comment(self):
-        dbname = self.get_long_name(length=10, mode="letters")
-        ntbname = self.get_long_name(length=5, mode="letters")
-
-        # create normal table with comment
-        comment = self.get_long_name(length=10, mode="letters")
-        self.__create_normaltb(dbname,ntbname,comment)
-        ntb_kv_list = tdSql.getResult("show tables")
-        print(ntb_kv_list)
-        tdSql.checkEqual(ntb_kv_list[0][8], comment)
-        tdSql.error('alter table {ntbname} comment "test1"')
-        tdSql.execute(f'drop database {dbname}')
-
-        # max length(1024)
-        comment = self.get_long_name(length=1024, mode="letters")
-        self.__create_normaltb(dbname,ntbname,comment)
-        ntb_kv_list = tdSql.getResult("show tables")
-        tdSql.checkEqual(ntb_kv_list[0][8], comment)
-        tdSql.execute(f'drop database {dbname}')
-
-        # error overlength
-        comment = self.get_long_name(length=1025, mode="letters")
-        tdSql.execute(f'create database if not exists {dbname}')
-        tdSql.execute(f'use {dbname}')
-        tdSql.error(f"create table ntb (ts timestamp,c0 int) comment '{comment}'")
-        tdSql.execute(f'drop database {dbname}')
-
-        # create child table with comment
-        comment = self.get_long_name(length=10, mode="letters")
-        stbname = self.get_long_name(length=5, mode="letters")
-        tbname = self.get_long_name(length=3, mode="letters")
-        self.__create_tb(dbname,stbname,tbname,comment)
-        ntb_kv_list = tdSql.getResult("show tables")
-        tdSql.checkEqual(ntb_kv_list[0][8], comment)
-        tdSql.error(f'alter table {tbname} comment "test1"')
-        tdSql.execute(f'drop database {dbname}')
-
-        # max length 1024
-        comment = self.get_long_name(length=1024, mode="letters")
-        self.__create_tb(dbname,ntbname,comment)
-        ntb_kv_list = tdSql.getResult("show tables")
-        tdSql.checkEqual(ntb_kv_list[0][8], comment)
-        tdSql.execute(f'drop database {dbname}')
-
-        # error overlength
-        comment = self.get_long_name(length=1025, mode="letters")
-        tdSql.execute(f'create database if not exists {dbname}')
-        tdSql.execute(f'use {dbname}')
-        tdSql.execute(f"create table stb (ts timestamp,c0 int) tags(t0 int)")
-        tdSql.error(f'create table stb_1 us stb tags(1) comment "{comment}"')
-        tdSql.execute(f'drop database {dbname}')
-
-    def run(self):
-        self.check_comment()
-
-    def stop(self):
-        tdSql.close()
-        tdLog.success("%s successfully executed" % __file__)
-
-tdCases.addWindows(__file__, TDTestCase())
-tdCases.addLinux(__file__, TDTestCase())
\ No newline at end of file

tests/system-test/1-insert/table_comment.py

+###################################################################
+#           Copyright (c) 2016 by TAOS Technologies, Inc.
+#                     All rights reserved.
+#
+#  This file is proprietary and confidential to TAOS Technologies.
+#  No part of this file may be reproduced, stored, transmitted,
+#  disclosed or used in any form or by any means other than as
+#  expressly provided by the written permission from Jianhui Tao
+#
+###################################################################
+
+# -*- coding: utf-8 -*-
+
+import random
+import string
+
+from util.log import *
+from util.cases import *
+from util.sql import *
+from util.common import *
+
+class TDTestCase:
+    def init(self, conn, logSql):
+        tdLog.debug("start to execute %s" % __file__)
+        tdSql.init(conn.cursor())
+        # prepare data
+        self.ntbname = 'ntb'
+        self.stbname = 'stb'
+        self.column_dict = {
+            'ts':'timestamp',
+            'c1':'int',
+            'c2':'float',
+            'c3':'double',
+            'c4':'timestamp'
+        }
+        self.tag_dict = {
+            't0':'int'
+        }
+        self.comment_length = [0,1024]
+        self.error_comment_length = [1025]
+        self.table_type_list = ['normal_table','stable','child_table']
+        self.comment_flag_list = [True,False]
+
+    def __set_and_alter_comment(self,tb_type='',comment_flag= False):
+        
+        column_sql = ''
+        tag_sql = ''
+        for k,v in self.column_dict.items():
+            column_sql += f"{k} {v},"
+        for k,v in self.tag_dict.items():
+            tag_sql += f"{k} {v},"
+        if tb_type == 'normal_table' or tb_type == '':
+            if comment_flag == False:
+                tdSql.execute(f'create table {self.ntbname} ({column_sql[:-1]})')
+                self.check_comment_info()
+                self.alter_comment(self.ntbname)
+                tdSql.execute(f'drop table {self.ntbname}')
+            elif comment_flag == True:
+                for i in self.comment_length:
+                    comment_info = tdCom.getLongName(i)
+                    tdSql.execute(f'create table {self.ntbname} ({column_sql[:-1]}) comment "{comment_info}"')
+                    self.check_comment_info(comment_info)
+                    self.alter_comment(self.ntbname)
+                    tdSql.execute(f'drop table {self.ntbname}')
+                for i in self.error_comment_length:
+                    comment_info = tdCom.getLongName(i)
+                    tdSql.error(f'create table {self.ntbname} ({column_sql[:-1]}) comment "{comment_info}"')
+        elif tb_type == 'stable':
+            for operation in ['table','stable']:
+                if comment_flag == False:
+                    tdSql.execute(f'create {operation} {self.stbname} ({column_sql[:-1]}) tags({tag_sql[:-1]})')
+                    self.check_comment_info(None,'stable')
+                    self.alter_comment(self.stbname,'stable')
+                    tdSql.execute(f'drop table {self.stbname}')
+                elif comment_flag == True:
+                    for i in self.comment_length:
+                        comment_info = tdCom.getLongName(i)
+                        tdSql.execute(f'create {operation} {self.stbname} ({column_sql[:-1]}) tags({tag_sql[:-1]}) comment "{comment_info}"')
+                        self.check_comment_info(comment_info,'stable')
+                        self.alter_comment(self.stbname,'stable')
+                        tdSql.execute(f'drop table {self.stbname}')   
+        elif tb_type == 'child_table':
+            tdSql.execute(f'create table if not exists {self.stbname} ({column_sql[:-1]}) tags({tag_sql[:-1]})')
+            if comment_flag == False:
+                tdSql.execute(f'create table if not exists {self.stbname}_ctb using {self.stbname} tags(1)')
+                self.check_comment_info()
+                self.alter_comment(f'{self.stbname}_ctb')
+                tdSql.execute(f'drop table {self.stbname}_ctb')
+            elif comment_flag == True:
+                for j in self.comment_length:
+                    comment_info = tdCom.getLongName(j)
+                    tdSql.execute(f'create table if not exists {self.stbname}_ctb using {self.stbname} tags(1) comment "{comment_info}"')
+                    self.check_comment_info(comment_info)
+                    self.alter_comment(f'{self.stbname}_ctb')
+                    tdSql.execute(f'drop table {self.stbname}_ctb')
+            tdSql.execute(f'drop table {self.stbname}')
+    def alter_comment(self,tbname,tb_type=''):
+        for i in self.comment_length:
+            comment_info = tdCom.getLongName(i)
+            print(comment_info)
+            tdSql.execute(f'alter table {tbname} comment "{comment_info}"')
+            self.check_comment_info(comment_info,tb_type)
+        for i in self.error_comment_length:
+            comment_info = tdCom.getLongName(i)
+            tdSql.error(f'alter table {tbname} comment "{comment_info}"')
+    def check_comment_info(self,comment_info=None,tb_type=''):
+        if tb_type == '' or tb_type == 'normal_table' or tb_type == 'child_table':
+            tdSql.query('show tables')
+            if comment_info == None:
+                tdSql.checkEqual(tdSql.queryResult[0][8],None)
+            else :
+                tdSql.checkEqual(tdSql.queryResult[0][8],comment_info)
+        elif tb_type == 'stable':
+            tdSql.query('show stables')
+            if comment_info == None:
+                tdSql.checkEqual(tdSql.queryResult[0][6],None)
+            else :
+                tdSql.checkEqual(tdSql.queryResult[0][6],comment_info)
+    def comment_check_case(self,table_type,comment_flag):
+        tdSql.prepare()
+        for tb in table_type:
+            for flag in comment_flag:
+                self.__set_and_alter_comment(tb,flag)
+        tdSql.execute('drop database db')
+        
+    def run(self):
+        self.comment_check_case(self.table_type_list,self.comment_flag_list)
+
+    def stop(self):
+        tdSql.close()
+        tdLog.success("%s successfully executed" % __file__)
+
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())
\ No newline at end of file

tests/system-test/6-cluster/5dnode3mnodeDrop.py

@@ -75,7 +75,7 @@ class TDTestCase:
         testCluster = False
         valgrind = 0  
         hostname = socket.gethostname()
-        print(hostname)
+        tdLog.debug(hostname)
         dnodes = []
         start_port = 6030
         start_port_sec = 6130
@@ -102,12 +102,12 @@ class TDTestCase:
 
         # create cluster 
         for dnode in self.TDDnodes.dnodes[1:]:
-            # print(dnode.cfgDict)
+            # tdLog.debug(dnode.cfgDict)
             dnode_id = dnode.cfgDict["fqdn"] +  ":" +dnode.cfgDict["serverPort"]
             dnode_first_host = dnode.cfgDict["firstEp"].split(":")[0]
             dnode_first_port = dnode.cfgDict["firstEp"].split(":")[-1]
             cmd = f" taos -h {dnode_first_host} -P {dnode_first_port} -s ' create dnode \"{dnode_id} \" ' ;"
-            print(cmd)
+            tdLog.debug(cmd)
             os.system(cmd)
         
         time.sleep(2)
@@ -119,26 +119,26 @@ class TDTestCase:
             time.sleep(1)
             tdSql.query("show mnodes;")
             if tdSql.checkRows(3) :     
-                print("mnode is  three nodes")           
+                tdLog.debug("mnode is  three nodes")           
             if  tdSql.queryResult[0][2]=='leader' :
                 if  tdSql.queryResult[1][2]=='follower':
                     if  tdSql.queryResult[2][2]=='follower':
-                        print("three mnodes is ready in 10s")
+                        tdLog.debug("three mnodes is ready in 10s")
                         break
             elif tdSql.queryResult[0][2]=='follower' :
                 if  tdSql.queryResult[1][2]=='leader':
                     if  tdSql.queryResult[2][2]=='follower':
-                        print("three mnodes is ready in 10s")
+                        tdLog.debug("three mnodes is ready in 10s")
                         break      
             elif tdSql.queryResult[0][2]=='follower' :
                 if  tdSql.queryResult[1][2]=='follower':
                     if  tdSql.queryResult[2][2]=='leader':
-                        print("three mnodes is ready in 10s")
+                        tdLog.debug("three mnodes is ready in 10s")
                         break                   
             count+=1
         else:
-            print(tdSql.queryResult)
-            print("three mnodes is not ready in 10s ")
+            tdLog.debug(tdSql.queryResult)
+            tdLog.debug("three mnodes is not ready in 10s ")
             return -1
 
         tdSql.query("show mnodes;")       
@@ -156,19 +156,19 @@ class TDTestCase:
             time.sleep(1)
             tdSql.query("show mnodes;")
             if tdSql.checkRows(3) :
-                print("mnode is  three nodes")
+                tdLog.debug("mnode is  three nodes")
             if  tdSql.queryResult[0][2]=='offline' :
                 if  tdSql.queryResult[1][2]=='leader':
                     if  tdSql.queryResult[2][2]=='follower':
-                        print("stop mnodes  on dnode 2 successfully in 10s")
+                        tdLog.debug("stop mnodes  on dnode 2 successfully in 10s")
                         break
                 elif tdSql.queryResult[1][2]=='follower':
                     if  tdSql.queryResult[2][2]=='leader':
-                        print("stop mnodes  on dnode 2 successfully in 10s")
+                        tdLog.debug("stop mnodes  on dnode 2 successfully in 10s")
                         break
             count+=1
         else:
-            print("stop mnodes  on dnode 2 failed in 10s ")
+            tdLog.debug("stop mnodes  on dnode 2 failed in 10s ")
             return -1 
         tdSql.error("drop mnode on dnode 1;")
 
@@ -188,15 +188,15 @@ class TDTestCase:
             time.sleep(1)
             tdSql.query("show mnodes;")
             if tdSql.checkRows(3) :
-                print("mnode is  three nodes")
+                tdLog.debug("mnode is  three nodes")
             if  tdSql.queryResult[0][2]=='leader' :
                 if  tdSql.queryResult[1][2]=='offline':
                     if  tdSql.queryResult[2][2]=='follower':
-                        print("stop mnodes  on dnode 2 successfully in 10s")
+                        tdLog.debug("stop mnodes  on dnode 2 successfully in 10s")
                         break
             count+=1
         else:
-            print("stop mnodes  on dnode 2 failed in 10s ")
+            tdLog.debug("stop mnodes  on dnode 2 failed in 10s ")
             return -1
         tdSql.error("drop mnode on dnode 2;")
 
@@ -218,15 +218,15 @@ class TDTestCase:
             time.sleep(1)
             tdSql.query("show mnodes;")
             if tdSql.checkRows(3) :
-                print("mnode is  three nodes")
+                tdLog.debug("mnode is  three nodes")
             if  tdSql.queryResult[0][2]=='leader' :
                 if  tdSql.queryResult[2][2]=='offline':
                     if  tdSql.queryResult[1][2]=='follower':
-                        print("stop mnodes  on dnode 3 successfully in 10s")
+                        tdLog.debug("stop mnodes  on dnode 3 successfully in 10s")
                         break
             count+=1
         else:
-            print("stop mnodes  on dnode 3 failed in 10s")
+            tdLog.debug("stop mnodes  on dnode 3 failed in 10s")
             return -1
         tdSql.error("drop mnode on dnode 3;")
         tdSql.query("show mnodes;")       
@@ -268,7 +268,7 @@ class TDTestCase:
         tdSql.error("drop mnode on dnode 1")
 
         tdSql.query("show dnodes;")
-        print(tdSql.queryResult)
+        tdLog.debug(tdSql.queryResult)
 
         #  drop follower of mnode 
         dropcount =0 
@@ -282,7 +282,7 @@ class TDTestCase:
                     time.sleep(1)
                     tdSql.query("show mnodes;")
                     if tdSql.checkRows(2):
-                        print("drop mnode %d successfully"%(i+1))
+                        tdLog.debug("drop mnode %d successfully"%(i+1))
                         break
                     count+=1
                 tdLog.debug("create mnode on dnode %d"%(i+1))
@@ -292,7 +292,7 @@ class TDTestCase:
                     time.sleep(1)
                     tdSql.query("show mnodes;")
                     if tdSql.checkRows(3):
-                        print("drop mnode %d successfully"%(i+1))
+                        tdLog.debug("drop mnode %d successfully"%(i+1))
                         break
                     count+=1
             dropcount+=1
@@ -307,7 +307,7 @@ class TDTestCase:
 
 
     def run(self): 
-        # print(self.master_dnode.cfgDict)
+        # tdLog.debug(self.master_dnode.cfgDict)
         self.buildcluster(5)
         self.five_dnode_three_mnode()
 

tests/system-test/6-cluster/5dnode3mnodeStop.py

@@ -99,12 +99,12 @@ class TDTestCase:
 
         # create cluster 
         for dnode in self.TDDnodes.dnodes[1:]:
-            # print(dnode.cfgDict)
+            # tdLog.debug(dnode.cfgDict)
             dnode_id = dnode.cfgDict["fqdn"] +  ":" +dnode.cfgDict["serverPort"]
             dnode_first_host = dnode.cfgDict["firstEp"].split(":")[0]
             dnode_first_port = dnode.cfgDict["firstEp"].split(":")[-1]
             cmd = f" taos -h {dnode_first_host} -P {dnode_first_port} -s ' create dnode \"{dnode_id} \" ' ;"
-            print(cmd)
+            tdLog.debug(cmd)
             os.system(cmd)
         
         time.sleep(2)
@@ -116,25 +116,25 @@ class TDTestCase:
             time.sleep(1)
             tdSql.query("show mnodes;")
             if tdSql.checkRows(3) :     
-                print("mnode is  three nodes")           
+                tdLog.debug("mnode is  three nodes")           
             if  tdSql.queryResult[0][2]=='leader' :
                 if  tdSql.queryResult[1][2]=='follower':
                     if  tdSql.queryResult[2][2]=='follower':
-                        print("three mnodes is ready in 10s")
+                        tdLog.debug("three mnodes is ready in 10s")
                         break
             elif tdSql.queryResult[0][2]=='follower' :
                 if  tdSql.queryResult[1][2]=='leader':
                     if  tdSql.queryResult[2][2]=='follower':
-                        print("three mnodes is ready in 10s")
+                        tdLog.debug("three mnodes is ready in 10s")
                         break      
             elif tdSql.queryResult[0][2]=='follower' :
                 if  tdSql.queryResult[1][2]=='follower':
                     if  tdSql.queryResult[2][2]=='leader':
-                        print("three mnodes is ready in 10s")
+                        tdLog.debug("three mnodes is ready in 10s")
                         break                   
             count+=1
         else:
-            print("three mnodes is not ready in 10s ")
+            tdLog.debug("three mnodes is not ready in 10s ")
             return -1
 
         tdSql.query("show mnodes;")       
@@ -152,19 +152,19 @@ class TDTestCase:
             time.sleep(1)
             tdSql.query("show mnodes;")
             if tdSql.checkRows(3) :
-                print("mnode is  three nodes")
+                tdLog.debug("mnode is  three nodes")
             if  tdSql.queryResult[0][2]=='offline' :
                 if  tdSql.queryResult[1][2]=='leader':
                     if  tdSql.queryResult[2][2]=='follower':
-                        print("stop mnodes  on dnode 2 successfully in 10s")
+                        tdLog.debug("stop mnodes  on dnode 2 successfully in 10s")
                         break
                 elif tdSql.queryResult[1][2]=='follower':
                     if  tdSql.queryResult[2][2]=='leader':
-                        print("stop mnodes  on dnode 2 successfully in 10s")
+                        tdLog.debug("stop mnodes  on dnode 2 successfully in 10s")
                         break
             count+=1
         else:
-            print("stop mnodes  on dnode 2 failed in 10s ")
+            tdLog.debug("stop mnodes  on dnode 2 failed in 10s ")
             return -1
         tdSql.error("drop mnode on dnode 1;")
 
@@ -184,15 +184,15 @@ class TDTestCase:
             time.sleep(1)
             tdSql.query("show mnodes;")
             if tdSql.checkRows(3) :
-                print("mnode is  three nodes")
+                tdLog.debug("mnode is  three nodes")
             if  tdSql.queryResult[0][2]=='leader' :
                 if  tdSql.queryResult[1][2]=='offline':
                     if  tdSql.queryResult[2][2]=='follower':
-                        print("stop mnodes  on dnode 2 successfully in 10s")
+                        tdLog.debug("stop mnodes  on dnode 2 successfully in 10s")
                         break
             count+=1
         else:
-            print("stop mnodes  on dnode 2 failed in 10s ")
+            tdLog.debug("stop mnodes  on dnode 2 failed in 10s ")
             return -1
         tdSql.error("drop mnode on dnode 2;")
 
@@ -214,15 +214,15 @@ class TDTestCase:
             time.sleep(1)
             tdSql.query("show mnodes;")
             if tdSql.checkRows(3) :
-                print("mnode is  three nodes")
+                tdLog.debug("mnode is  three nodes")
             if  tdSql.queryResult[0][2]=='leader' :
                 if  tdSql.queryResult[2][2]=='offline':
                     if  tdSql.queryResult[1][2]=='follower':
-                        print("stop mnodes  on dnode 3 successfully in 10s")
+                        tdLog.debug("stop mnodes  on dnode 3 successfully in 10s")
                         break
             count+=1
         else:
-            print("stop mnodes  on dnode 3 failed in 10s")
+            tdLog.debug("stop mnodes  on dnode 3 failed in 10s")
             return -1
         tdSql.error("drop mnode on dnode 3;")
         tdSql.query("show mnodes;")       
@@ -262,7 +262,7 @@ class TDTestCase:
         tdSql.error("create mnode on dnode 2")
 
         tdSql.query("show dnodes;")
-        print(tdSql.queryResult)
+        tdLog.debug(tdSql.queryResult)
 
         tdLog.debug("stop and follower of mnode") 
         self.TDDnodes.stoptaosd(2)
@@ -303,7 +303,7 @@ class TDTestCase:
 
 
     def run(self): 
-        # print(self.master_dnode.cfgDict)
+        # tdLog.debug(self.master_dnode.cfgDict)
         self.buildcluster(5)
         self.five_dnode_three_mnode(5)
 

tests/system-test/6-cluster/5dnode3mnodeStopInsert.py

@@ -77,7 +77,7 @@ class TDTestCase:
         for couti in range(countstart,countstop):
             tdLog.debug("drop database if exists db%d" %couti)
             tdSql.execute("drop database if exists db%d" %couti)
-            print("create database if not exists db%d replica 1 duration 300" %couti)
+            tdLog.debug("create database if not exists db%d replica 1 duration 300" %couti)
             tdSql.execute("create database if not exists db%d replica 1 duration 300" %couti)
             tdSql.execute("use db%d" %couti)
             tdSql.execute(
@@ -126,12 +126,12 @@ class TDTestCase:
 
         # create cluster 
         for dnode in self.TDDnodes.dnodes[1:]:
-            # print(dnode.cfgDict)
+            # tdLog.debug(dnode.cfgDict)
             dnode_id = dnode.cfgDict["fqdn"] +  ":" +dnode.cfgDict["serverPort"]
             dnode_first_host = dnode.cfgDict["firstEp"].split(":")[0]
             dnode_first_port = dnode.cfgDict["firstEp"].split(":")[-1]
             cmd = f" taos -h {dnode_first_host} -P {dnode_first_port} -s ' create dnode \"{dnode_id} \" ' ;"
-            print(cmd)
+            tdLog.debug(cmd)
             os.system(cmd)
         
         time.sleep(2)
@@ -144,22 +144,22 @@ class TDTestCase:
             statusReadyBumber=0
             tdSql.query("show dnodes;")
             if tdSql.checkRows(dnodenumber) :     
-                print("dnode is %d nodes"%dnodenumber)   
+                tdLog.debug("dnode is %d nodes"%dnodenumber)   
             for i in range(dnodenumber):
                 if tdSql.queryResult[i][4] !='ready'  :
                     status=tdSql.queryResult[i][4]
-                    print("dnode:%d status is %s "%(i,status))
+                    tdLog.debug("dnode:%d status is %s "%(i,status))
                     break
                 else:
                     statusReadyBumber+=1
-            print(statusReadyBumber)
+            tdLog.debug(statusReadyBumber)
             if statusReadyBumber == dnodenumber :
-                print("all of %d mnodes is ready in 10s "%dnodenumber)
+                tdLog.debug("all of %d mnodes is ready in 10s "%dnodenumber)
                 return True
                 break
             count+=1
         else:
-            print("%d mnodes is not ready in 10s "%dnodenumber)
+            tdLog.debug("%d mnodes is not ready in 10s "%dnodenumber)
             return False
            
 
@@ -169,25 +169,25 @@ class TDTestCase:
             time.sleep(1)
             tdSql.query("show mnodes;")
             if tdSql.checkRows(3) :     
-                print("mnode is  three nodes")           
+                tdLog.debug("mnode is  three nodes")           
             if  tdSql.queryResult[0][2]=='leader' :
                 if  tdSql.queryResult[1][2]=='follower':
                     if  tdSql.queryResult[2][2]=='follower':
-                        print("three mnodes is ready in 10s")
+                        tdLog.debug("three mnodes is ready in 10s")
                         break
             elif tdSql.queryResult[0][2]=='follower' :
                 if  tdSql.queryResult[1][2]=='leader':
                     if  tdSql.queryResult[2][2]=='follower':
-                        print("three mnodes is ready in 10s")
+                        tdLog.debug("three mnodes is ready in 10s")
                         break      
             elif tdSql.queryResult[0][2]=='follower' :
                 if  tdSql.queryResult[1][2]=='follower':
                     if  tdSql.queryResult[2][2]=='leader':
-                        print("three mnodes is ready in 10s")
+                        tdLog.debug("three mnodes is ready in 10s")
                         break                   
             count+=1
         else:
-            print("three mnodes is not ready in 10s ")
+            tdLog.debug("three mnodes is not ready in 10s ")
             return -1
 
         tdSql.query("show mnodes;")       
@@ -205,19 +205,19 @@ class TDTestCase:
             time.sleep(1)
             tdSql.query("show mnodes;")
             if tdSql.checkRows(3) :
-                print("mnode is  three nodes")
+                tdLog.debug("mnode is  three nodes")
             if  tdSql.queryResult[0][2]=='offline' :
                 if  tdSql.queryResult[1][2]=='leader':
                     if  tdSql.queryResult[2][2]=='follower':
-                        print("stop mnodes  on dnode 2 successfully in 10s")
+                        tdLog.debug("stop mnodes  on dnode 2 successfully in 10s")
                         break
                 elif tdSql.queryResult[1][2]=='follower':
                     if  tdSql.queryResult[2][2]=='leader':
-                        print("stop mnodes  on dnode 2 successfully in 10s")
+                        tdLog.debug("stop mnodes  on dnode 2 successfully in 10s")
                         break
             count+=1
         else:
-            print("stop mnodes  on dnode 2 failed in 10s ")
+            tdLog.debug("stop mnodes  on dnode 2 failed in 10s ")
             return -1
         tdSql.error("drop mnode on dnode 1;")
 
@@ -237,15 +237,15 @@ class TDTestCase:
             time.sleep(1)
             tdSql.query("show mnodes;")
             if tdSql.checkRows(3) :
-                print("mnode is  three nodes")
+                tdLog.debug("mnode is  three nodes")
             if  tdSql.queryResult[0][2]=='leader' :
                 if  tdSql.queryResult[1][2]=='offline':
                     if  tdSql.queryResult[2][2]=='follower':
-                        print("stop mnodes  on dnode 2 successfully in 10s")
+                        tdLog.debug("stop mnodes  on dnode 2 successfully in 10s")
                         break
             count+=1
         else:
-            print("stop mnodes  on dnode 2 failed in 10s ")
+            tdLog.debug("stop mnodes  on dnode 2 failed in 10s ")
             return -1
         tdSql.error("drop mnode on dnode 2;")
 
@@ -267,15 +267,15 @@ class TDTestCase:
             time.sleep(1)
             tdSql.query("show mnodes;")
             if tdSql.checkRows(3) :
-                print("mnode is  three nodes")
+                tdLog.debug("mnode is  three nodes")
             if  tdSql.queryResult[0][2]=='leader' :
                 if  tdSql.queryResult[2][2]=='offline':
                     if  tdSql.queryResult[1][2]=='follower':
-                        print("stop mnodes  on dnode 3 successfully in 10s")
+                        tdLog.debug("stop mnodes  on dnode 3 successfully in 10s")
                         break
             count+=1
         else:
-            print("stop mnodes  on dnode 3 failed in 10s")
+            tdLog.debug("stop mnodes  on dnode 3 failed in 10s")
             return -1
         tdSql.error("drop mnode on dnode 3;")
         tdSql.query("show mnodes;")       
@@ -311,7 +311,7 @@ class TDTestCase:
 
         tdSql.error("create mnode on dnode 2")
         tdSql.query("show dnodes;")
-        print(tdSql.queryResult)
+        tdLog.debug(tdSql.queryResult)
         tdLog.debug("stop all of mnode ") 
 
         stopcount =0 
@@ -325,10 +325,10 @@ class TDTestCase:
                 self.TDDnodes.starttaosd(i+1)
 
                 if self.checkdnodes(5):
-                    print("123")
+                    tdLog.debug("123")
                     threads.join()
                 else:
-                    print("456")
+                    tdLog.debug("456")
                     self.stop_thread(threads)
                     assert 1 == 2 ,"some dnode started failed"
                     return False
@@ -345,7 +345,7 @@ class TDTestCase:
 
 
     def run(self): 
-        # print(self.master_dnode.cfgDict)
+        # tdLog.debug(self.master_dnode.cfgDict)
         self.buildcluster(5)
         self.five_dnode_three_mnode(5)
 

tests/system-test/fulltest.sh

@@ -22,7 +22,7 @@ python3 ./test.py -f 1-insert/opentsdb_json_taosc_insert.py
 python3 ./test.py -f 1-insert/alter_stable.py
 python3 ./test.py -f 1-insert/alter_table.py
 python3 ./test.py -f 1-insert/insertWithMoreVgroup.py
-# python3 ./test.py -f 1-inerst/create_table_comment.py
+python3 ./test.py -f 1-insert/table_comment.py
 python3 ./test.py -f 2-query/between.py
 python3 ./test.py -f 2-query/distinct.py
 python3 ./test.py -f 2-query/varchar.py