Merge branch '3.0' into fix/TS-2149

docs/zh/12-taos-sql/10-function.md

@@ -1249,4 +1249,4 @@ SELECT SERVER_VERSION();
 SELECT SERVER_STATUS();
 ```
 
-**说明**：返回服务端当前的状态。
+**说明**：检测服务端是否所有 dnode 都在线，如果是则返回成功，否则返回无法建立连接的错误。

include/common/tcommon.h

@@ -174,15 +174,28 @@ typedef struct SColumnDataAgg {
 } SColumnDataAgg;
 #pragma pack(pop)
 
+typedef struct SBlockID {
+  // The uid of table, from which current data block comes. And it is always 0, if current block is the
+  // result of calculation.
+  uint64_t uid;
+
+  // Block id, acquired and assigned from executor, which created according to the hysical planner. Block id is used
+  // to mark the stage of exec task.
+  uint64_t blockId;
+
+  // Generated by group/partition by [value|tags]. Created and assigned by table-scan operator, group-by operator,
+  // and partition by operator.
+  uint64_t groupId;
+} SBlockID;
+
 typedef struct SDataBlockInfo {
   STimeWindow window;
-  int32_t     rows;  // todo hide this attribute
   int32_t     rowSize;
-  uint64_t    uid;      // the uid of table, from which current data block comes
-  uint16_t    blockId;  // block id, generated by physical planner
-  uint64_t    groupId;
-  int16_t     hasVarCol;
+  int32_t     rows;  // todo hide this attribute
   uint32_t    capacity;
+  SBlockID    id;
+  int16_t     hasVarCol;
+
   // TODO: optimize and remove following
   int64_t     version;    // used for stream, and need serialization
   int32_t     childId;    // used for stream, do not serialize

packaging/MPtestJenkinsfile

@@ -120,20 +120,20 @@ pipeline {
                             for verModeSin in ${verModeList}
                             do
                                 cd ${TDENGINE_ROOT_DIR}/packaging
-                                bash testpackage.sh -f server -m ${verModeSin} -f server  -l false -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar 
+                                bash testpackage.sh  -m ${verModeSin} -f server  -l false -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar 
                                 python3 checkPackageRuning.py
                             done 
                             '''  
 
                             sh '''                              
                             cd ${TDENGINE_ROOT_DIR}/packaging
-                            bash testpackage.sh -f server -m community -f server  -l true -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar 
+                            bash testpackage.sh  -m community -f server  -l true -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar 
                             python3 checkPackageRuning.py
                             '''   
 
                             sh ''' 
                             cd ${TDENGINE_ROOT_DIR}/packaging
-                            bash testpackage.sh -f server -m community -f server  -l false -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t deb 
+                            bash testpackage.sh  -m community -f server  -l false -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t deb 
                             python3 checkPackageRuning.py
                             '''
                         }
@@ -152,20 +152,20 @@ pipeline {
                             for verModeSin in ${verModeList}
                             do
                                 cd ${TDENGINE_ROOT_DIR}/packaging
-                                bash testpackage.sh -f server -m ${verModeSin} -f server  -l false -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar 
+                                bash testpackage.sh -m ${verModeSin} -f server  -l false -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar 
                                 python3 checkPackageRuning.py
                             done 
                             '''
 
                             sh '''
                             cd ${TDENGINE_ROOT_DIR}/packaging
-                            bash testpackage.sh -f server -m community -f server  -l true -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar 
+                            bash testpackage.sh -m community -f server  -l true -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar 
                             python3 checkPackageRuning.py
                             '''
 
                             sh '''
                             cd ${TDENGINE_ROOT_DIR}/packaging
-                            bash testpackage.sh -f server -m community -f server  -l false -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t deb 
+                            bash testpackage.sh -m community -f server  -l false -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t deb 
                             python3 checkPackageRuning.py
                             dpkg -r tdengine
                             '''                                
@@ -185,20 +185,20 @@ pipeline {
                             for verModeSin in ${verModeList}
                             do
                                 cd ${TDENGINE_ROOT_DIR}/packaging
-                                bash testpackage.sh -f server -m community -f server  -l false -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar 
+                                bash testpackage.sh -m ${verModeSin} -f server  -l false -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar 
                                 python3 checkPackageRuning.py
                             done 
                             '''
 
                             sh '''                            
                             cd ${TDENGINE_ROOT_DIR}/packaging
-                            bash testpackage.sh -f server -m community -f server  -l true -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar 
+                            bash testpackage.sh -m community -f server  -l true -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar 
                             python3 checkPackageRuning.py
                             '''
 
                             sh '''
                             cd ${TDENGINE_ROOT_DIR}/packaging
-                            bash testpackage.sh -f server -m community -f server  -l false -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t rpm 
+                            bash testpackage.sh -m community -f server  -l false -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t rpm 
                             python3 checkPackageRuning.py
                             '''
                         }
@@ -217,20 +217,20 @@ pipeline {
                             for verModeSin in ${verModeList}
                             do
                                 cd ${TDENGINE_ROOT_DIR}/packaging
-                                bash testpackage.sh -f server -m ${verModeSin} -f server  -l false -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar 
+                                bash testpackage.sh -m ${verModeSin} -f server  -l false -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar 
                                 python3 checkPackageRuning.py
                             done
                             '''
 
                             sh '''
                             cd ${TDENGINE_ROOT_DIR}/packaging
-                            bash testpackage.sh -f server -m community -f server  -l true -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar 
+                            bash testpackage.sh -m community -f server  -l true -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar 
                             python3 checkPackageRuning.py
                             '''
 
                             sh '''
                             cd ${TDENGINE_ROOT_DIR}/packaging
-                            bash testpackage.sh -f server -m community -f server  -l false -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t rpm 
+                            bash testpackage.sh -m community -f server  -l false -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t rpm 
                             python3 checkPackageRuning.py
                             sudo rpm -e tdengine 
                             '''
@@ -250,7 +250,7 @@ pipeline {
                             for verModeSin in ${verModeList}
                             do
                                 cd ${TDENGINE_ROOT_DIR}/packaging
-                                bash testpackage.sh -f server -m ${verModeSin} -f server  -l false -c arm64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar 
+                                bash testpackage.sh -m ${verModeSin} -f server  -l false -c arm64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar 
                                 python3 checkPackageRuning.py
                             done 
                             '''
@@ -273,10 +273,9 @@ pipeline {
                             for verModeSin in ${verModeList}
                             do
                                 cd ${TDENGINE_ROOT_DIR}/packaging
-                                bash testpackage.sh -f server -m ${verModeSin} -f client  -l false -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar 
-                                python3 checkPackageRuning.py
-                            done 
+                                bash testpackage.sh  -m ${verModeSin} -f client  -l false -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar 
                                 python3 checkPackageRuning.py 192.168.0.21
+                            done 
                             '''
                         }
                     }
@@ -286,10 +285,8 @@ pipeline {
                     steps {
                         timeout(time: 30, unit: 'MINUTES'){
                             sh '''
-                            verModeList=community
                             cd ${TDENGINE_ROOT_DIR}/packaging
-                            bash testpackage.sh -f server -m ${verModeSin} -f client  -l true -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar
-                            python3 checkPackageRuning.py
+                            bash testpackage.sh -m community -f client  -l true -c x64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar
                             python3 checkPackageRuning.py  192.168.0.24
                             '''         
                         }
@@ -309,10 +306,9 @@ pipeline {
                     for verModeSin in ${verModeList}
                     do
                         cd ${TDENGINE_ROOT_DIR}/packaging
-                        bash testpackage.sh -f server -m ${verModeSin} -f client  -l false -c arm64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar
-                        python3 checkPackageRuning.py
-                    done 
+                        bash testpackage.sh  -m ${verModeSin} -f client  -l false -c arm64  -v ${version} -o ${baseVersion} -s ${sourcePath}  -t tar
                         python3 checkPackageRuning.py 192.168.0.21
+                    done 
                     '''  
                 }
             }

packaging/testpackage.sh

@@ -206,6 +206,10 @@ else
 fi
 
 
+if [[ ${packgeName} =~ "server" ]] ;then
+    echoColor BD " pkill -9 taosd "
+    pkill -9 taosd
+fi
 
 
 echoColor G "===== new workroom path ====="
@@ -334,7 +338,7 @@ elif  ([[ ${packgeName} =~ "arm64" ]] && [[ ${packgeName} =~ "client" ]]);then
     echoColor G "===== install taos-tools arm when package is arm64-client ====="
     cd ${installPath}
     if [ ! -f "taosTools-2.1.3-Linux-x64.tar.gz " ];then
-        wgetFile taosTools-2.1.3-Linux-x64.tar.gz v2.1.3 web
+        wgetFile taosTools-2.1.3-Linux-arm64.tar.gz v2.1.3 web
         tar xf taosTools-2.1.3-Linux-arm64.tar.gz
     fi    
     

source/common/src/tdatablock.c

@@ -621,7 +621,7 @@ int32_t blockDataFromBuf(SSDataBlock* pBlock, const char* buf) {
 // todo remove this
 int32_t blockDataFromBuf1(SSDataBlock* pBlock, const char* buf, size_t capacity) {
   pBlock->info.rows = *(int32_t*)buf;
-  pBlock->info.groupId = *(uint64_t*)(buf + sizeof(int32_t));
+  pBlock->info.id.groupId = *(uint64_t*)(buf + sizeof(int32_t));
 
   size_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
 
@@ -1140,7 +1140,8 @@ void blockDataCleanup(SSDataBlock* pDataBlock) {
   SDataBlockInfo* pInfo = &pDataBlock->info;
 
   pInfo->rows = 0;
-  pInfo->groupId = 0;
+  pInfo->id.uid = 0;
+  pInfo->id.groupId = 0;
   pInfo->window.ekey = 0;
   pInfo->window.skey = 0;
 
@@ -1334,7 +1335,7 @@ int32_t copyDataBlock(SSDataBlock* dst, const SSDataBlock* src) {
 SSDataBlock* createSpecialDataBlock(EStreamType type) {
   SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
   pBlock->info.hasVarCol = false;
-  pBlock->info.groupId = 0;
+  pBlock->info.id.groupId = 0;
   pBlock->info.rows = 0;
   pBlock->info.type = type;
   pBlock->info.rowSize = sizeof(TSKEY) + sizeof(TSKEY) + sizeof(uint64_t) + sizeof(uint64_t) + sizeof(TSKEY) +
@@ -1675,7 +1676,7 @@ int32_t blockDataKeepFirstNRows(SSDataBlock* pBlock, size_t n) {
 }
 
 int32_t tEncodeDataBlock(void** buf, const SSDataBlock* pBlock) {
-  int64_t tbUid = pBlock->info.uid;
+  int64_t tbUid = pBlock->info.id.uid;
   int16_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
   int16_t hasVarCol = pBlock->info.hasVarCol;
   int32_t rows = pBlock->info.rows;
@@ -1713,7 +1714,7 @@ void* tDecodeDataBlock(const void* buf, SSDataBlock* pBlock) {
 
   int16_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
 
-  buf = taosDecodeFixedU64(buf, &pBlock->info.uid);
+  buf = taosDecodeFixedU64(buf, &pBlock->info.id.uid);
   buf = taosDecodeFixedI16(buf, &numOfCols);
   buf = taosDecodeFixedI16(buf, &pBlock->info.hasVarCol);
   buf = taosDecodeFixedI32(buf, &pBlock->info.rows);
@@ -1834,7 +1835,7 @@ void blockDebugShowDataBlocks(const SArray* dataBlocks, const char* flag) {
     int32_t rows = pDataBlock->info.rows;
     printf("%s |block ver %" PRIi64 " |block type %d |child id %d|group id %" PRIu64 "\n", flag,
            pDataBlock->info.version, (int32_t)pDataBlock->info.type, pDataBlock->info.childId,
-           pDataBlock->info.groupId);
+           pDataBlock->info.id.groupId);
     for (int32_t j = 0; j < rows; j++) {
       printf("%s |", flag);
       for (int32_t k = 0; k < numOfCols; k++) {
@@ -1905,8 +1906,8 @@ char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf)
   len += snprintf(dumpBuf + len, size - len,
                   "===stream===%s|block type %d|child id %d|group id:%" PRIu64 "|uid:%" PRId64
                   "|rows:%d|version:%" PRIu64 "\n",
-                  flag, (int32_t)pDataBlock->info.type, pDataBlock->info.childId, pDataBlock->info.groupId,
-                  pDataBlock->info.uid, pDataBlock->info.rows, pDataBlock->info.version);
+                  flag, (int32_t)pDataBlock->info.type, pDataBlock->info.childId, pDataBlock->info.id.groupId,
+                  pDataBlock->info.id.uid, pDataBlock->info.rows, pDataBlock->info.version);
   if (len >= size - 1) return dumpBuf;
 
   for (int32_t j = 0; j < rows; j++) {
@@ -2035,8 +2036,6 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SSDataBlock* pDataB
   for (int32_t i = 0; i < sz; ++i) {
     int32_t colNum = taosArrayGetSize(pDataBlock->pDataBlock);
     int32_t rows = pDataBlock->info.rows;
-    //    int32_t      rowSize = pDataBlock->info.rowSize;
-    //    int64_t      groupId = pDataBlock->info.groupId;
 
     if (colNum <= 1) {
       // invalid if only with TS col
@@ -2049,7 +2048,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SSDataBlock* pDataB
 
     SSubmitBlk* pSubmitBlk = POINTER_SHIFT(pDataBuf, msgLen);
     pSubmitBlk->suid = suid;
-    pSubmitBlk->uid = pDataBlock->info.groupId;
+    pSubmitBlk->uid = pDataBlock->info.id.groupId;
     pSubmitBlk->numOfRows = rows;
     pSubmitBlk->sversion = pTSchema->version;
 
@@ -2292,7 +2291,7 @@ int32_t blockEncode(const SSDataBlock* pBlock, char* data, int32_t numOfCols) {
   }
 
   *actualLen = dataLen;
-  *groupId = pBlock->info.groupId;
+  *groupId = pBlock->info.id.groupId;
   ASSERT(dataLen > 0);
 
   uDebug("build data block, actualLen:%d, rows:%d, cols:%d", dataLen, *rows, *cols);
@@ -2325,7 +2324,7 @@ const char* blockDecode(SSDataBlock* pBlock, const char* pData) {
   pStart += sizeof(int32_t);
 
   // group id sizeof(uint64_t)
-  pBlock->info.groupId = *(uint64_t*)pStart;
+  pBlock->info.id.groupId = *(uint64_t*)pStart;
   pStart += sizeof(uint64_t);
 
   if (pBlock->pDataBlock == NULL) {

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

@@ -525,6 +525,7 @@ static int32_t mndCreateStbForStream(SMnode *pMnode, STrans *pTrans, const SStre
 
   tFreeSMCreateStbReq(&createReq);
   mndFreeStb(&stbObj);
+  mndReleaseStb(pMnode, pStb);
   mndReleaseDb(pMnode, pDb);
 
   return 0;

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

@@ -434,6 +434,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
       return -1;
     }
     topicObj.stbUid = pStb->uid;
+    mndReleaseStb(pMnode, pStb);
   }
   /*} else if (pCreate->subType == TOPIC_SUB_TYPE__DB) {*/
   /*topicObj.ast = NULL;*/

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

@@ -708,7 +708,7 @@ static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSma
 #endif
     for (int32_t i = 0; i < taosArrayGetSize(pResList); ++i) {
       SSDataBlock *output = taosArrayGetP(pResList, i);
-      smaDebug("result block, uid:%" PRIu64 ", groupid:%" PRIu64 ", rows:%d", output->info.uid, output->info.groupId,
+      smaDebug("result block, uid:%" PRIu64 ", groupid:%" PRIu64 ", rows:%d", output->info.id.uid, output->info.id.groupId,
                output->info.rows);
 
       STsdb      *sinkTsdb = (pItem->level == TSDB_RETENTION_L1 ? pSma->pRSmaTsdb[0] : pSma->pRSmaTsdb[1]);
@@ -718,7 +718,7 @@ static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSma
       if (buildSubmitReqFromDataBlock(&pReq, output, pTSchema, SMA_VID(pSma), suid) < 0) {
         smaError("vgId:%d, build submit req for rsma table suid:%" PRIu64 ", uid:%" PRIu64 ", level %" PRIi8
                  " failed since %s",
-                 SMA_VID(pSma), suid, output->info.groupId, pItem->level, terrstr());
+                 SMA_VID(pSma), suid, output->info.id.groupId, pItem->level, terrstr());
         goto _err;
       }
 
@@ -726,13 +726,13 @@ static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSma
         taosMemoryFreeClear(pReq);
         smaError("vgId:%d, process submit req for rsma suid:%" PRIu64 ", uid:%" PRIu64 " level %" PRIi8
                  " failed since %s",
-                 SMA_VID(pSma), suid, output->info.groupId, pItem->level, terrstr());
+                 SMA_VID(pSma), suid, output->info.id.groupId, pItem->level, terrstr());
         goto _err;
       }
 
       smaDebug("vgId:%d, process submit req for rsma suid:%" PRIu64 ",uid:%" PRIu64 ", level %" PRIi8 " ver %" PRIi64
                " len %" PRIu32,
-               SMA_VID(pSma), suid, output->info.groupId, pItem->level, output->info.version,
+               SMA_VID(pSma), suid, output->info.id.groupId, pItem->level, output->info.version,
                htonl(pReq->header.contLen));
 
       taosMemoryFreeClear(pReq);

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

@@ -299,7 +299,7 @@ int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, STaosxRsp
           taosArrayDestroyP(pSchemas, (FDelete)tDeleteSSchemaWrapper);
           pBlocks = taosArrayInit(0, sizeof(SSDataBlock));
           pSchemas = taosArrayInit(0, sizeof(void*));
-          return -1;
+          continue;
         }
       }
       if (pHandle->fetchMeta) {

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

@@ -61,6 +61,7 @@ int32_t tqOffsetRestoreFromFile(STqOffsetStore* pStore, const char* fname) {
         ASSERT(0);
         // TODO
       }
+      taosMemoryFree(memBuf);
     }
 
     taosCloseFile(&pFile);

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

@@ -530,7 +530,7 @@ int32_t tqRetrieveDataBlock(SSDataBlock* pBlock, STqReader* pReader) {
 
   tInitSubmitBlkIter(&pReader->msgIter, pReader->pBlock, &pReader->blkIter);
 
-  pBlock->info.uid = pReader->msgIter.uid;
+  pBlock->info.id.uid = pReader->msgIter.uid;
   pBlock->info.rows = pReader->msgIter.numOfRows;
   pBlock->info.version = pReader->pMsg->version;
 
@@ -649,7 +649,7 @@ int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas
     }
 
     SSDataBlock* pBlock = taosArrayGetLast(blocks);
-    pBlock->info.uid = pReader->msgIter.uid;
+    pBlock->info.id.uid = pReader->msgIter.uid;
     pBlock->info.rows = 0;
     pBlock->info.version = pReader->pMsg->version;
 

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

@@ -103,7 +103,7 @@ SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchem
       //        STagVal tagVal = {
       //            .cid = pTagSchemaWrapper->pSchema[j].colId,
       //            .type = pTagSchemaWrapper->pSchema[j].type,
-      //            .i64 = (int64_t)pDataBlock->info.groupId,
+      //            .i64 = (int64_t)pDataBlock->info.id.groupId,
       //        };
       //        taosArrayPush(tagArray, &tagVal);
       //        taosArrayPush(tagName, pTagSchemaWrapper->pSchema[j].name);
@@ -134,7 +134,7 @@ SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchem
       STagVal tagVal = {
           .cid = taosArrayGetSize(pDataBlock->pDataBlock) + 1,
           .type = TSDB_DATA_TYPE_UBIGINT,
-          .i64 = (int64_t)pDataBlock->info.groupId,
+          .i64 = (int64_t)pDataBlock->info.id.groupId,
       };
       taosArrayPush(tagArray, &tagVal);
       createTbReq.ctb.tagNum = taosArrayGetSize(tagArray);
@@ -161,7 +161,7 @@ SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchem
       if (pDataBlock->info.parTbName[0]) {
         createTbReq.name = strdup(pDataBlock->info.parTbName);
       } else {
-        createTbReq.name = buildCtbNameByGroupId(stbFullName, pDataBlock->info.groupId);
+        createTbReq.name = buildCtbNameByGroupId(stbFullName, pDataBlock->info.id.groupId);
       }
 
       // save schema len
@@ -358,7 +358,7 @@ void tqSinkToTablePipeline(SStreamTask* pTask, void* vnode, int64_t ver, void* d
       if (pDataBlock->info.parTbName[0]) {
         ctbName = strdup(pDataBlock->info.parTbName);
       } else {
-        ctbName = buildCtbNameByGroupId(stbFullName, pDataBlock->info.groupId);
+        ctbName = buildCtbNameByGroupId(stbFullName, pDataBlock->info.id.groupId);
       }
 
       int32_t schemaLen = 0;
@@ -390,7 +390,7 @@ void tqSinkToTablePipeline(SStreamTask* pTask, void* vnode, int64_t ver, void* d
         STagVal tagVal = {
             .cid = taosArrayGetSize(pDataBlock->pDataBlock) + 1,
             .type = TSDB_DATA_TYPE_UBIGINT,
-            .i64 = (int64_t)pDataBlock->info.groupId,
+            .i64 = (int64_t)pDataBlock->info.id.groupId,
         };
         taosArrayPush(tagArray, &tagVal);
         createTbReq.ctb.tagNum = taosArrayGetSize(tagArray);

source/dnode/vnode/src/tsdb/tsdbRead.c

@@ -1624,7 +1624,7 @@ static int32_t buildDataBlockFromBuf(STsdbReader* pReader, STableBlockScanInfo*
   int32_t code = buildDataBlockFromBufImpl(pBlockScanInfo, endKey, pReader->capacity, pReader);
 
   blockDataUpdateTsWindow(pBlock, 0);
-  pBlock->info.uid = pBlockScanInfo->uid;
+  pBlock->info.id.uid = pBlockScanInfo->uid;
 
   setComposedBlockFlag(pReader, true);
 
@@ -2494,7 +2494,7 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
   }
 
 _end:
-  pResBlock->info.uid = (pBlockScanInfo != NULL) ? pBlockScanInfo->uid : 0;
+  pResBlock->info.id.uid = (pBlockScanInfo != NULL) ? pBlockScanInfo->uid : 0;
   blockDataUpdateTsWindow(pResBlock, 0);
 
   setComposedBlockFlag(pReader, true);
@@ -2506,7 +2506,7 @@ _end:
   if (pResBlock->info.rows > 0) {
     tsdbDebug("%p uid:%" PRIu64 ", composed data block created, brange:%" PRIu64 "-%" PRIu64
               " rows:%d, elapsed time:%.2f ms %s",
-              pReader, pResBlock->info.uid, pResBlock->info.window.skey, pResBlock->info.window.ekey,
+              pReader, pResBlock->info.id.uid, pResBlock->info.window.skey, pResBlock->info.window.ekey,
               pResBlock->info.rows, el, pReader->idStr);
   }
 
@@ -2830,7 +2830,7 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) {
     } else {  // whole block is required, return it directly
       SDataBlockInfo* pInfo = &pReader->pResBlock->info;
       pInfo->rows = pBlock->nRow;
-      pInfo->uid = pScanInfo->uid;
+      pInfo->id.uid = pScanInfo->uid;
       pInfo->window = (STimeWindow){.skey = pBlock->minKey.ts, .ekey = pBlock->maxKey.ts};
       setComposedBlockFlag(pReader, false);
       setBlockAllDumped(&pStatus->fBlockDumpInfo, pBlock->maxKey.ts, pReader->order);
@@ -4020,7 +4020,7 @@ bool tsdbTableNextDataBlock(STsdbReader* pReader, uint64_t uid) {
 static void setBlockInfo(const STsdbReader* pReader, int32_t* rows, uint64_t* uid, STimeWindow* pWindow) {
   ASSERT(pReader != NULL);
   *rows = pReader->pResBlock->info.rows;
-  *uid = pReader->pResBlock->info.uid;
+  *uid = pReader->pResBlock->info.id.uid;
   *pWindow = pReader->pResBlock->info.window;
 }
 

source/dnode/vnode/test/tsdbSmaTest.cpp

@@ -436,7 +436,7 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
     pDataBlock->pBlockAgg = NULL;
     taosArrayGetSize(pDataBlock->pDataBlock) = tSmaNumOfCols;
     pDataBlock->info.rows = tSmaNumOfRows;
-    pDataBlock->info.groupId = tSmaGroupId + g;
+    pDataBlock->info.id.groupId = tSmaGroupId + g;
 
     pDataBlock->pDataBlock = taosArrayInit(tSmaNumOfCols, sizeof(SColumnInfoData *));
     EXPECT_NE(pDataBlock->pDataBlock, nullptr);

source/libs/command/src/command.c

@@ -270,10 +270,10 @@ static void setCreateDBResultIntoDataBlock(SSDataBlock* pBlock, char* dbFName, S
       "CREATE DATABASE `%s` BUFFER %d CACHESIZE %d CACHEMODEL '%s' COMP %d DURATION %dm "
       "WAL_FSYNC_PERIOD %d MAXROWS %d MINROWS %d KEEP %dm,%dm,%dm PAGES %d PAGESIZE %d PRECISION '%s' REPLICA %d "
       "STRICT '%s' WAL_LEVEL %d VGROUPS %d SINGLE_STABLE %d",
-      dbFName, pCfg->buffer, pCfg->cacheSize, cacheModelStr(pCfg->cacheLast), pCfg->compression, pCfg->daysPerFile, pCfg->walFsyncPeriod,
-      pCfg->maxRows, pCfg->minRows, pCfg->daysToKeep0, pCfg->daysToKeep1, pCfg->daysToKeep2, pCfg->pages,
-      pCfg->pageSize, prec, pCfg->replications, strictStr(pCfg->strict), pCfg->walLevel, pCfg->numOfVgroups,
-      1 == pCfg->numOfStables);
+      dbFName, pCfg->buffer, pCfg->cacheSize, cacheModelStr(pCfg->cacheLast), pCfg->compression, pCfg->daysPerFile,
+      pCfg->walFsyncPeriod, pCfg->maxRows, pCfg->minRows, pCfg->daysToKeep0, pCfg->daysToKeep1, pCfg->daysToKeep2,
+      pCfg->pages, pCfg->pageSize, prec, pCfg->replications, strictStr(pCfg->strict), pCfg->walLevel,
+      pCfg->numOfVgroups, 1 == pCfg->numOfStables);
 
   if (retentions) {
     len += sprintf(buf2 + VARSTR_HEADER_SIZE + len, " RETENTIONS %s", retentions);
@@ -691,9 +691,15 @@ static int32_t createSelectResultDataBlock(SNodeList* pProjects, SSDataBlock** p
 
   SNode* pProj = NULL;
   FOREACH(pProj, pProjects) {
+    SExprNode*      pExpr = (SExprNode*)pProj;
     SColumnInfoData infoData = {0};
-    infoData.info.type = ((SExprNode*)pProj)->resType.type;
-    infoData.info.bytes = ((SExprNode*)pProj)->resType.bytes;
+    if (TSDB_DATA_TYPE_NULL == pExpr->resType.type) {
+      infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
+      infoData.info.bytes = 0;
+    } else {
+      infoData.info.type = pExpr->resType.type;
+      infoData.info.bytes = pExpr->resType.bytes;
+    }
     blockDataAppendColInfo(pBlock, &infoData);
   }
   *pOutput = pBlock;

source/libs/executor/inc/executil.h

@@ -38,16 +38,7 @@
     memcpy((_k) + sizeof(uint64_t), (_ori), (_len)); \
   } while (0)
 
-#define SET_RES_EXT_WINDOW_KEY(_k, _ori, _len, _uid, _buf)           \
-  do {                                                               \
-    assert(sizeof(_uid) == sizeof(uint64_t));                        \
-    *(void**)(_k) = (_buf);                                          \
-    *(uint64_t*)((_k) + POINTER_BYTES) = (_uid);                     \
-    memcpy((_k) + POINTER_BYTES + sizeof(uint64_t), (_ori), (_len)); \
-  } while (0)
-
 #define GET_RES_WINDOW_KEY_LEN(_l)     ((_l) + sizeof(uint64_t))
-#define GET_RES_EXT_WINDOW_KEY_LEN(_l) ((_l) + sizeof(uint64_t) + POINTER_BYTES)
 
 #define GET_TASKID(_t) (((SExecTaskInfo*)(_t))->id.str)
 
@@ -110,7 +101,8 @@ int32_t        tableListGetOutputGroups(const STableListInfo* pTableList);
 bool            oneTableForEachGroup(const STableListInfo* pTableList);
 uint64_t        getTableGroupId(const STableListInfo* pTableList, uint64_t tableUid);
 int32_t         tableListAddTableInfo(STableListInfo* pTableList, uint64_t uid, uint64_t gid);
-int32_t        tableListGetGroupList(const STableListInfo* pTableList, int32_t ordinalIndex, STableKeyInfo** pKeyInfo, int32_t* num);
+int32_t         tableListGetGroupList(const STableListInfo* pTableList, int32_t ordinalIndex, STableKeyInfo** pKeyInfo,
+                                      int32_t* num);
 uint64_t        tableListGetSize(const STableListInfo* pTableList);
 uint64_t        tableListGetSuid(const STableListInfo* pTableList);
 STableKeyInfo*  tableListGetInfo(const STableListInfo* pTableList, int32_t index);
@@ -140,7 +132,7 @@ bool hasRemainResults(SGroupResInfo* pGroupResInfo);
 
 int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo);
 
-SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode);
+SSDataBlock* createDataBlockFromDescNode(SDataBlockDescNode* pNode);
 
 EDealRes doTranslateTagExpr(SNode** pNode, void* pContext);
 int32_t  getGroupIdFromTagsVal(void* pMeta, uint64_t uid, SNodeList* pGroupNode, char* keyBuf, uint64_t* pGroupId);

source/libs/executor/inc/executorimpl.h

@@ -537,15 +537,6 @@ typedef struct SStreamIntervalOperatorInfo {
   SWinKey            delKey;
 } SStreamIntervalOperatorInfo;
 
-typedef struct SAggOperatorInfo {
-  SOptrBasicInfo   binfo;
-  SAggSupporter    aggSup;
-  STableQueryInfo* current;
-  uint64_t         groupId;
-  SGroupResInfo    groupResInfo;
-  SExprSupp        scalarExprSup;
-} SAggOperatorInfo;
-
 typedef struct SFillOperatorInfo {
   struct SFillInfo* pFillInfo;
   SSDataBlock*      pRes;
@@ -577,18 +568,6 @@ typedef struct SWindowRowsSup {
   uint64_t    groupId;
 } SWindowRowsSup;
 
-typedef struct SSessionAggOperatorInfo {
-  SOptrBasicInfo binfo;
-  SAggSupporter  aggSup;
-
-  SGroupResInfo      groupResInfo;
-  SWindowRowsSup     winSup;
-  bool               reptScan;  // next round scan
-  int64_t            gap;       // session window gap
-  int32_t            tsSlotId;  // primary timestamp slot id
-  STimeWindowAggSupp twAggSup;
-} SSessionAggOperatorInfo;
-
 typedef struct SResultWindowInfo {
   void*       pOutputBuf;
   SSessionKey sessionWin;
@@ -681,53 +660,30 @@ typedef struct SStreamFillOperatorInfo {
   SStreamFillInfo*      pFillInfo;
 } SStreamFillOperatorInfo;
 
-typedef struct STimeSliceOperatorInfo {
-  SSDataBlock*         pRes;
-  STimeWindow          win;
-  SInterval            interval;
-  int64_t              current;
-  SArray*              pPrevRow;     // SArray<SGroupValue>
-  SArray*              pNextRow;     // SArray<SGroupValue>
-  SArray*              pLinearInfo;  // SArray<SFillLinearInfo>
-  bool                 isPrevRowSet;
-  bool                 isNextRowSet;
-  int32_t              fillType;      // fill type
-  SColumn              tsCol;         // primary timestamp column
-  SExprSupp            scalarSup;     // scalar calculation
-  struct SFillColInfo* pFillColInfo;  // fill column info
-} STimeSliceOperatorInfo;
-
-typedef struct SStateWindowOperatorInfo {
-  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
-  SOptrBasicInfo binfo;
-  SAggSupporter  aggSup;
-  SExprSupp      scalarSup;
-
-  SGroupResInfo      groupResInfo;
-  SWindowRowsSup     winSup;
-  SColumn            stateCol;  // start row index
-  bool               hasKey;
-  SStateKeys         stateKey;
-  int32_t            tsSlotId;  // primary timestamp column slot id
-  STimeWindowAggSupp twAggSup;
-} SStateWindowOperatorInfo;
-
 #define OPTR_IS_OPENED(_optr)  (((_optr)->status & OP_OPENED) == OP_OPENED)
 #define OPTR_SET_OPENED(_optr) ((_optr)->status |= OP_OPENED)
 
 SOperatorFpSet createOperatorFpSet(__optr_open_fn_t openFn, __optr_fn_t nextFn, __optr_fn_t cleanup,
                                    __optr_close_fn_t closeFn, __optr_explain_fn_t explain);
-
 int32_t        operatorDummyOpenFn(SOperatorInfo* pOperator);
 int32_t        appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t num);
+void           setOperatorCompleted(SOperatorInfo* pOperator);
+void           setOperatorInfo(SOperatorInfo* pOperator, const char* name, int32_t type, bool blocking, int32_t status,
+                               void* pInfo, SExecTaskInfo* pTaskInfo);
+void           destroyOperatorInfo(SOperatorInfo* pOperator);
 
 void    initBasicInfo(SOptrBasicInfo* pInfo, SSDataBlock* pBlock);
 void    cleanupBasicInfo(SOptrBasicInfo* pInfo);
+
 int32_t initExprSupp(SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfExpr);
 void    cleanupExprSupp(SExprSupp* pSup);
+
 void    destroyExprInfo(SExprInfo* pExpr, int32_t numOfExprs);
-int32_t initAggInfo(SExprSupp* pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, size_t keyBufSize,
+
+int32_t initAggSup(SExprSupp* pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, size_t keyBufSize,
                    const char* pkey);
+void    cleanupAggSup(SAggSupporter* pAggSup);
+
 void    initResultSizeInfo(SResultInfo* pResultInfo, int32_t numOfRows);
 
 void doBuildStreamResBlock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
@@ -739,8 +695,8 @@ bool    hasLimitOffsetInfo(SLimitInfo* pLimitInfo);
 void initLimitInfo(const SNode* pLimit, const SNode* pSLimit, SLimitInfo* pLimitInfo);
 void applyLimitOffset(SLimitInfo* pLimitInfo, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo, SOperatorInfo* pOperator);
 
-void doApplyFunctions(SExecTaskInfo* taskInfo, SqlFunctionCtx* pCtx, SColumnInfoData* pTimeWindowData, int32_t offset,
-                      int32_t forwardStep, int32_t numOfTotal, int32_t numOfOutput);
+void applyAggFunctionOnPartialTuples(SExecTaskInfo* taskInfo, SqlFunctionCtx* pCtx, SColumnInfoData* pTimeWindowData,
+                                     int32_t offset, int32_t forwardStep, int32_t numOfTotal, int32_t numOfOutput);
 
 int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, char* pData, SArray* pColList, char** pNextStart);
 void    updateLoadRemoteInfo(SLoadRemoteDataInfo* pInfo, int32_t numOfRows, int32_t dataLen, int64_t startTs,
@@ -751,130 +707,108 @@ STimeWindow getFirstQualifiedTimeWindow(int64_t ts, STimeWindow* pWindow, SInter
 int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scanFlag);
 int32_t getBufferPgSize(int32_t rowSize, uint32_t* defaultPgsz, uint32_t* defaultBufsz);
 
-void doDestroyExchangeOperatorInfo(void* param);
+extern void doDestroyExchangeOperatorInfo(void* param);
 
-void    setOperatorCompleted(SOperatorInfo* pOperator);
-void    setOperatorInfo(SOperatorInfo* pOperator, const char* name, int32_t type, bool blocking, int32_t status,
-                        void* pInfo, SExecTaskInfo* pTaskInfo);
 void    doFilter(SSDataBlock* pBlock, SFilterInfo* pFilterInfo, SColMatchInfo* pColMatchInfo);
 int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int32_t numOfExpr, SSDataBlock* pBlock,
                                int32_t rows, const char* idStr, STableMetaCacheInfo* pCache);
 
-void cleanupAggSup(SAggSupporter* pAggSup);
 void appendOneRowToDataBlock(SSDataBlock* pBlock, STupleHandle* pTupleHandle);
 void setTbNameColData(const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId, const char* name);
 
-SSDataBlock* loadNextDataBlock(void* param);
-
 void setResultRowInitCtx(SResultRow* pResult, SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset);
 
 SResultRow* doSetResultOutBufByKey(SDiskbasedBuf* pResultBuf, SResultRowInfo* pResultRowInfo, char* pData,
                                    int16_t bytes, bool masterscan, uint64_t groupId, SExecTaskInfo* pTaskInfo,
                                    bool isIntervalQuery, SAggSupporter* pSup);
-
+// operator creater functions
+// clang-format off
 SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode* pExNode, SExecTaskInfo* pTaskInfo);
 
-SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle,
-                                           SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pPhyNode,
-                                         STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScanPhysiNode* pScanPhyNode,
-                                              const char* pUser, SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScanPhysiNode* pScanPhyNode, const char* pUser, SExecTaskInfo* pTaskInfo);
 
 SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode* pNode, SExecTaskInfo* pTaskInfo);
 
-SOperatorInfo* createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pNode,
-                                                 SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhysiNode* pProjPhyNode,
-                                         SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pNode, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhysiNode* pProjPhyNode, SExecTaskInfo* pTaskInfo);
+
 SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pSortNode, SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** dowStreams, size_t numStreams,
-                                               SMergePhysiNode* pMergePhysiNode, SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createCacherowsScanOperator(SLastRowScanPhysiNode* pTableScanNode, SReadHandle* readHandle,
-                                           SExecTaskInfo* pTaskInfo);
-
-SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SIntervalPhysiNode* pPhyNode,
-                                          SExecTaskInfo* pTaskInfo, bool isStream);
-SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SMergeIntervalPhysiNode* pIntervalPhyNode,
-                                               SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream, SMergeAlignedIntervalPhysiNode* pNode,
-                                                      SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
-                                                     SExecTaskInfo* pTaskInfo, int32_t numOfChild);
-SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SSessionWinodwPhysiNode* pSessionNode,
-                                            SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** dowStreams, size_t numStreams, SMergePhysiNode* pMergePhysiNode, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createCacherowsScanOperator(SLastRowScanPhysiNode* pTableScanNode, SReadHandle* readHandle, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SIntervalPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, bool isStream);
+
+SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SMergeIntervalPhysiNode* pIntervalPhyNode, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream, SMergeAlignedIntervalPhysiNode* pNode, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, int32_t numOfChild);
+
+SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SSessionWinodwPhysiNode* pSessionNode, SExecTaskInfo* pTaskInfo);
+
 SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode* pAggNode, SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDistScanPhysiNode* pBlockScanNode,
-                                               SExecTaskInfo* pTaskInfo);
 
-SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode, SNode* pTagCond,
-                                            SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDistScanPhysiNode* pBlockScanNode, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode, SNode* pTagCond, SExecTaskInfo* pTaskInfo);
 
 SOperatorInfo* createRawScanOperatorInfo(SReadHandle* pHandle, SExecTaskInfo* pTaskInfo);
 
-SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode* pPhyFillNode,
-                                      SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SStateWinodwPhysiNode* pStateNode,
-                                             SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartitionPhysiNode* pPartNode,
-                                           SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode* pPhyFillNode, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SStateWinodwPhysiNode* pStateNode, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartitionPhysiNode* pPartNode, SExecTaskInfo* pTaskInfo);
 
-SOperatorInfo* createStreamPartitionOperatorInfo(SOperatorInfo* downstream, SStreamPartitionPhysiNode* pPartNode,
-                                                 SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createStreamPartitionOperatorInfo(SOperatorInfo* downstream, SStreamPartitionPhysiNode* pPartNode, SExecTaskInfo* pTaskInfo);
 
 SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pNode, SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOfDownstream,
-                                           SSortMergeJoinPhysiNode* pJoinNode, SExecTaskInfo* pTaskInfo);
 
-SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
-                                                  SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createStreamFinalSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
-                                                       SExecTaskInfo* pTaskInfo, int32_t numOfChild);
-SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
-                                                SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOfDownstream, SSortMergeJoinPhysiNode* pJoinNode, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createStreamFinalSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, int32_t numOfChild);
+
+SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo);
 
-SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
-                                                SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFillPhysiNode* pPhyFillNode,
-                                            SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFillPhysiNode* pPhyFillNode, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createGroupSortOperatorInfo(SOperatorInfo* downstream, SGroupSortPhysiNode* pSortPhyNode, SExecTaskInfo* pTaskInfo);
+// clang-format on
 
 int32_t projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBlock* pSrcBlock, SqlFunctionCtx* pCtx,
                               int32_t numOfOutput, SArray* pPseudoList);
 
 void setInputDataBlock(SExprSupp* pExprSupp, SSDataBlock* pBlock, int32_t order, int32_t scanFlag, bool createDummyCol);
 
-bool    isTaskKilled(SExecTaskInfo* pTaskInfo);
 int32_t checkForQueryBuf(size_t numOfTables);
 
+bool    isTaskKilled(SExecTaskInfo* pTaskInfo);
 void    setTaskKilled(SExecTaskInfo* pTaskInfo);
-void    queryCostStatis(SExecTaskInfo* pTaskInfo);
 void    doDestroyTask(SExecTaskInfo* pTaskInfo);
-void    destroyOperatorInfo(SOperatorInfo* pOperator);
-int32_t getMaximumIdleDurationSec();
-
-/*
- * ops:     root operator
- * data:    *data save the result of encode, need to be freed by caller
- * length:  *length save the length of *data
- * nOptrWithVal: *nOptrWithVal save the number of optr with value
- * return:  result code, 0 means success
- */
-int32_t encodeOperator(SOperatorInfo* ops, char** data, int32_t* length, int32_t* nOptrWithVal);
-
-/*
- * ops:    root operator, created by caller
- * data:   save the result of decode
- * length: the length of data
- * return: result code, 0 means success
- */
-int32_t decodeOperator(SOperatorInfo* ops, const char* data, int32_t length);
-
 void    setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status);
+
 int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId,
                                char* sql, EOPTR_EXEC_MODEL model);
 int32_t createDataSinkParam(SDataSinkNode* pNode, void** pParam, qTaskInfo_t* pTaskInfo, SReadHandle* readHandle);
 int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SArray* pExecInfoList);
 
+void    printTaskExecCostInLog(SExecTaskInfo* pTaskInfo);
+
+int32_t getMaximumIdleDurationSec();
+
 STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts, SInterval* pInterval,
                                 int32_t order);
 int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimaryColumn, int32_t startPos, TSKEY ekey,
@@ -897,15 +831,7 @@ void     calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock);
 int32_t finalizeResultRows(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition, SExprSupp* pSup,
                            SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
 
-SOperatorInfo* createGroupSortOperatorInfo(SOperatorInfo* downstream, SGroupSortPhysiNode* pSortPhyNode,
-                                           SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,
-                                                SExecTaskInfo* pTaskInfo);
-
-void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsColIndex);
-
 bool    groupbyTbname(SNodeList* pGroupList);
-void*   destroySqlFunctionCtx(SqlFunctionCtx* pCtx, int32_t numOfOutput);
 int32_t buildDataBlockFromGroupRes(SOperatorInfo* pOperator, SStreamState* pState, SSDataBlock* pBlock, SExprSupp* pSup,
                                    SGroupResInfo* pGroupResInfo);
 int32_t saveSessionDiscBuf(SStreamState* pState, SSessionKey* key, void* buf, int32_t size);

source/libs/executor/src/cachescanoperator.c

@@ -59,7 +59,7 @@ SOperatorInfo* createCacherowsScanOperator(SLastRowScanPhysiNode* pScanNode, SRe
   pInfo->readHandle = *readHandle;
 
   SDataBlockDescNode* pDescNode = pScanNode->scan.node.pOutputDataBlockDesc;
-  pInfo->pRes = createResDataBlock(pDescNode);
+  pInfo->pRes = createDataBlockFromDescNode(pDescNode);
 
   int32_t numOfCols = 0;
   code =
@@ -185,7 +185,7 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
         }
       }
 
-      pRes->info.uid = *(tb_uid_t*)taosArrayGet(pInfo->pUidList, pInfo->indexOfBufferedRes);
+      pRes->info.id.uid = *(tb_uid_t*)taosArrayGet(pInfo->pUidList, pInfo->indexOfBufferedRes);
       pRes->info.rows = 1;
 
       SExprSupp* pSup = &pInfo->pseudoExprSup;
@@ -196,7 +196,7 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
         return NULL;
       }
 
-      pRes->info.groupId = getTableGroupId(pTableList, pRes->info.uid);
+      pRes->info.id.groupId = getTableGroupId(pTableList, pRes->info.id.uid);
       pInfo->indexOfBufferedRes += 1;
       return pRes;
     } else {
@@ -232,12 +232,12 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
           SExprSupp* pSup = &pInfo->pseudoExprSup;
 
           STableKeyInfo* pKeyInfo = &((STableKeyInfo*)pList)[0];
-          pInfo->pRes->info.groupId = pKeyInfo->groupId;
+          pInfo->pRes->info.id.groupId = pKeyInfo->groupId;
 
           if (taosArrayGetSize(pInfo->pUidList) > 0) {
             ASSERT((pInfo->retrieveType & CACHESCAN_RETRIEVE_LAST_ROW) == CACHESCAN_RETRIEVE_LAST_ROW);
 
-            pInfo->pRes->info.uid = *(tb_uid_t*)taosArrayGet(pInfo->pUidList, 0);
+            pInfo->pRes->info.id.uid = *(tb_uid_t*)taosArrayGet(pInfo->pUidList, 0);
             code = addTagPseudoColumnData(&pInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pInfo->pRes, pInfo->pRes->info.rows,
                                           GET_TASKID(pTaskInfo), NULL);
             if (code != TSDB_CODE_SUCCESS) {

source/libs/executor/src/exchangeoperator.c

@@ -303,7 +303,7 @@ SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode
   }
 
   tsem_init(&pInfo->ready, 0, 0);
-  pInfo->pDummyBlock = createResDataBlock(pExNode->node.pOutputDataBlockDesc);
+  pInfo->pDummyBlock = createDataBlockFromDescNode(pExNode->node.pOutputDataBlockDesc);
   pInfo->pResultBlockList = taosArrayInit(64, POINTER_BYTES);
   pInfo->pRecycledBlocks = taosArrayInit(64, POINTER_BYTES);
 
@@ -717,10 +717,10 @@ int32_t prepareLoadRemoteData(SOperatorInfo* pOperator) {
 int32_t handleLimitOffset(SOperatorInfo* pOperator, SLimitInfo* pLimitInfo, SSDataBlock* pBlock, bool holdDataInBuf) {
   if (pLimitInfo->remainGroupOffset > 0) {
     if (pLimitInfo->currentGroupId == 0) {  // it is the first group
-      pLimitInfo->currentGroupId = pBlock->info.groupId;
+      pLimitInfo->currentGroupId = pBlock->info.id.groupId;
       blockDataCleanup(pBlock);
       return PROJECT_RETRIEVE_CONTINUE;
-    } else if (pLimitInfo->currentGroupId != pBlock->info.groupId) {
+    } else if (pLimitInfo->currentGroupId != pBlock->info.id.groupId) {
       // now it is the data from a new group
       pLimitInfo->remainGroupOffset -= 1;
 
@@ -732,11 +732,11 @@ int32_t handleLimitOffset(SOperatorInfo* pOperator, SLimitInfo* pLimitInfo, SSDa
     }
 
     // set current group id of the project operator
-    pLimitInfo->currentGroupId = pBlock->info.groupId;
+    pLimitInfo->currentGroupId = pBlock->info.id.groupId;
   }
 
   // here check for a new group data, we need to handle the data of the previous group.
-  if (pLimitInfo->currentGroupId != 0 && pLimitInfo->currentGroupId != pBlock->info.groupId) {
+  if (pLimitInfo->currentGroupId != 0 && pLimitInfo->currentGroupId != pBlock->info.id.groupId) {
     pLimitInfo->numOfOutputGroups += 1;
     if ((pLimitInfo->slimit.limit > 0) && (pLimitInfo->slimit.limit <= pLimitInfo->numOfOutputGroups)) {
       pOperator->status = OP_EXEC_DONE;
@@ -758,7 +758,7 @@ int32_t handleLimitOffset(SOperatorInfo* pOperator, SLimitInfo* pLimitInfo, SSDa
   // here we reach the start position, according to the limit/offset requirements.
 
   // set current group id
-  pLimitInfo->currentGroupId = pBlock->info.groupId;
+  pLimitInfo->currentGroupId = pBlock->info.id.groupId;
 
   if (pLimitInfo->remainOffset >= pBlock->info.rows) {
     pLimitInfo->remainOffset -= pBlock->info.rows;

source/libs/executor/src/executil.c

@@ -208,12 +208,12 @@ SArray* createSortInfo(SNodeList* pNodeList) {
   return pList;
 }
 
-SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode) {
+SSDataBlock* createDataBlockFromDescNode(SDataBlockDescNode* pNode) {
   int32_t numOfCols = LIST_LENGTH(pNode->pSlots);
 
   SSDataBlock* pBlock = createDataBlock();
 
-  pBlock->info.blockId = pNode->dataBlockId;
+  pBlock->info.id.blockId = pNode->dataBlockId;
   pBlock->info.type = STREAM_INVALID;
   pBlock->info.calWin = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
   pBlock->info.watermark = INT64_MIN;

source/libs/executor/src/executor.c

@@ -712,7 +712,7 @@ void qDestroyTask(qTaskInfo_t qTaskHandle) {
 
   qDebug("%s execTask completed, numOfRows:%" PRId64, GET_TASKID(pTaskInfo), pTaskInfo->pRoot->resultInfo.totalRows);
 
-  queryCostStatis(pTaskInfo);  // print the query cost summary
+  printTaskExecCostInLog(pTaskInfo);  // print the query cost summary
   doDestroyTask(pTaskInfo);
 }
 
@@ -728,12 +728,12 @@ int32_t qSerializeTaskStatus(qTaskInfo_t tinfo, char** pOutput, int32_t* len) {
   }
 
   int32_t nOptrWithVal = 0;
-  int32_t code = encodeOperator(pTaskInfo->pRoot, pOutput, len, &nOptrWithVal);
-  if ((code == TSDB_CODE_SUCCESS) && (nOptrWithVal == 0)) {
-    taosMemoryFreeClear(*pOutput);
-    *len = 0;
-  }
-  return code;
+//  int32_t code = encodeOperator(pTaskInfo->pRoot, pOutput, len, &nOptrWithVal);
+//  if ((code == TSDB_CODE_SUCCESS) && (nOptrWithVal == 0)) {
+//    taosMemoryFreeClear(*pOutput);
+//    *len = 0;
+//  }
+  return 0;
 }
 
 int32_t qDeserializeTaskStatus(qTaskInfo_t tinfo, const char* pInput, int32_t len) {
@@ -743,7 +743,8 @@ int32_t qDeserializeTaskStatus(qTaskInfo_t tinfo, const char* pInput, int32_t le
     return TSDB_CODE_INVALID_PARA;
   }
 
-  return decodeOperator(pTaskInfo->pRoot, pInput, len);
+  return 0;
+//  return decodeOperator(pTaskInfo->pRoot, pInput, len);
 }
 
 int32_t qExtractStreamScanner(qTaskInfo_t tinfo, void** scanner) {

source/libs/executor/src/groupoperator.c

@@ -308,13 +308,13 @@ static void doHashGroupbyAgg(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
 
     len = buildGroupKeys(pInfo->keyBuf, pInfo->pGroupColVals);
     int32_t ret = setGroupResultOutputBuf(pOperator, &(pInfo->binfo), pOperator->exprSupp.numOfExprs, pInfo->keyBuf,
-                                          len, pBlock->info.groupId, pInfo->aggSup.pResultBuf, &pInfo->aggSup);
+                                          len, pBlock->info.id.groupId, pInfo->aggSup.pResultBuf, &pInfo->aggSup);
     if (ret != TSDB_CODE_SUCCESS) {  // null data, too many state code
       T_LONG_JMP(pTaskInfo->env, TSDB_CODE_QRY_APP_ERROR);
     }
 
     int32_t rowIndex = j - num;
-    doApplyFunctions(pTaskInfo, pCtx, NULL, rowIndex, num, pBlock->info.rows, pOperator->exprSupp.numOfExprs);
+    applyAggFunctionOnPartialTuples(pTaskInfo, pCtx, NULL, rowIndex, num, pBlock->info.rows, pOperator->exprSupp.numOfExprs);
 
     // assign the group keys or user input constant values if required
     doAssignGroupKeys(pCtx, pOperator->exprSupp.numOfExprs, pBlock->info.rows, rowIndex);
@@ -325,13 +325,13 @@ static void doHashGroupbyAgg(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
   if (num > 0) {
     len = buildGroupKeys(pInfo->keyBuf, pInfo->pGroupColVals);
     int32_t ret = setGroupResultOutputBuf(pOperator, &(pInfo->binfo), pOperator->exprSupp.numOfExprs, pInfo->keyBuf,
-                                          len, pBlock->info.groupId, pInfo->aggSup.pResultBuf, &pInfo->aggSup);
+                                          len, pBlock->info.id.groupId, pInfo->aggSup.pResultBuf, &pInfo->aggSup);
     if (ret != TSDB_CODE_SUCCESS) {
       T_LONG_JMP(pTaskInfo->env, TSDB_CODE_QRY_APP_ERROR);
     }
 
     int32_t rowIndex = pBlock->info.rows - num;
-    doApplyFunctions(pTaskInfo, pCtx, NULL, rowIndex, num, pBlock->info.rows, pOperator->exprSupp.numOfExprs);
+    applyAggFunctionOnPartialTuples(pTaskInfo, pCtx, NULL, rowIndex, num, pBlock->info.rows, pOperator->exprSupp.numOfExprs);
     doAssignGroupKeys(pCtx, pOperator->exprSupp.numOfExprs, pBlock->info.rows, rowIndex);
   }
 }
@@ -431,7 +431,7 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode*
     goto _error;
   }
 
-  SSDataBlock* pResBlock = createResDataBlock(pAggNode->node.pOutputDataBlockDesc);
+  SSDataBlock* pResBlock = createDataBlockFromDescNode(pAggNode->node.pOutputDataBlockDesc);
   initBasicInfo(&pInfo->binfo, pResBlock);
 
   int32_t    numOfScalarExpr = 0;
@@ -456,7 +456,7 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode*
 
   int32_t    num = 0;
   SExprInfo* pExprInfo = createExprInfo(pAggNode->pAggFuncs, pAggNode->pGroupKeys, &num);
-  code = initAggInfo(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, num, pInfo->groupKeyLen, pTaskInfo->id.str);
+  code = initAggSup(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, num, pInfo->groupKeyLen, pTaskInfo->id.str);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }
@@ -697,7 +697,7 @@ static SSDataBlock* buildPartitionResult(SOperatorInfo* pOperator) {
   releaseBufPage(pInfo->pBuf, page);
 
   blockDataUpdateTsWindow(pInfo->binfo.pRes, 0);
-  pInfo->binfo.pRes->info.groupId = pGroupInfo->groupId;
+  pInfo->binfo.pRes->info.id.groupId = pGroupInfo->groupId;
 
   pOperator->resultInfo.totalRows += pInfo->binfo.pRes->info.rows;
   return pInfo->binfo.pRes;
@@ -823,7 +823,7 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition
   uint32_t defaultPgsz = 0;
   uint32_t defaultBufsz = 0;
 
-  pInfo->binfo.pRes = createResDataBlock(pPartNode->node.pOutputDataBlockDesc);
+  pInfo->binfo.pRes = createDataBlockFromDescNode(pPartNode->node.pOutputDataBlockDesc);
   getBufferPgSize(pInfo->binfo.pRes->info.rowSize, &defaultPgsz, &defaultBufsz);
 
   if (!osTempSpaceAvailable()) {
@@ -952,7 +952,7 @@ static SSDataBlock* buildStreamPartitionResult(SOperatorInfo* pOperator) {
   taosArrayDestroy(pParInfo->rowIds);
   pParInfo->rowIds = NULL;
   blockDataUpdateTsWindow(pDest, pInfo->tsColIndex);
-  pDest->info.groupId = pParInfo->groupId;
+  pDest->info.id.groupId = pParInfo->groupId;
   pOperator->resultInfo.totalRows += pDest->info.rows;
   pInfo->parIte = taosHashIterate(pInfo->pPartitions, pInfo->parIte);
   ASSERT(pDest->info.rows > 0);
@@ -1119,7 +1119,7 @@ SOperatorInfo* createStreamPartitionOperatorInfo(SOperatorInfo* downstream, SStr
   }
   pInfo->partitionSup.needCalc = true;
 
-  pInfo->binfo.pRes = createResDataBlock(pPartNode->part.node.pOutputDataBlockDesc);
+  pInfo->binfo.pRes = createDataBlockFromDescNode(pPartNode->part.node.pOutputDataBlockDesc);
   if (pInfo->binfo.pRes == NULL) {
     goto _error;
   }

source/libs/executor/src/joinoperator.c

@@ -87,7 +87,7 @@ SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t
   }
 
   int32_t      numOfCols = 0;
-  SSDataBlock* pResBlock = createResDataBlock(pJoinNode->node.pOutputDataBlockDesc);
+  SSDataBlock* pResBlock = createDataBlockFromDescNode(pJoinNode->node.pOutputDataBlockDesc);
   SExprInfo*   pExprInfo = createExprInfo(pJoinNode->pTargets, NULL, &numOfCols);
   initResultSizeInfo(&pOperator->resultInfo, 4096);
 
@@ -185,7 +185,7 @@ static void mergeJoinJoinLeftRight(struct SOperatorInfo* pOperator, SSDataBlock*
     int32_t rowIndex = -1;
 
     SColumnInfoData* pSrc = NULL;
-    if (pLeftBlock->info.blockId == blockId) {
+    if (pLeftBlock->info.id.blockId == blockId) {
       pSrc = taosArrayGet(pLeftBlock->pDataBlock, slotId);
       rowIndex = leftPos;
     } else {

source/libs/executor/src/projectoperator.c

@@ -85,7 +85,7 @@ SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhys
   int32_t    numOfCols = 0;
   SExprInfo* pExprInfo = createExprInfo(pProjPhyNode->pProjections, NULL, &numOfCols);
 
-  SSDataBlock* pResBlock = createResDataBlock(pProjPhyNode->node.pOutputDataBlockDesc);
+  SSDataBlock* pResBlock = createDataBlockFromDescNode(pProjPhyNode->node.pOutputDataBlockDesc);
   initLimitInfo(pProjPhyNode->node.pLimit, pProjPhyNode->node.pSlimit, &pInfo->limitInfo);
 
   pInfo->binfo.pRes = pResBlock;
@@ -102,7 +102,7 @@ SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhys
   }
 
   initResultSizeInfo(&pOperator->resultInfo, numOfRows);
-  code = initAggInfo(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str);
+  code = initAggSup(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }
@@ -138,13 +138,13 @@ _error:
 static int32_t discardGroupDataBlock(SSDataBlock* pBlock, SLimitInfo* pLimitInfo) {
   if (pLimitInfo->remainGroupOffset > 0) {
     // it is the first group
-    if (pLimitInfo->currentGroupId == 0 || pLimitInfo->currentGroupId == pBlock->info.groupId) {
-      pLimitInfo->currentGroupId = pBlock->info.groupId;
+    if (pLimitInfo->currentGroupId == 0 || pLimitInfo->currentGroupId == pBlock->info.id.groupId) {
+      pLimitInfo->currentGroupId = pBlock->info.id.groupId;
       return PROJECT_RETRIEVE_CONTINUE;
-    } else if (pLimitInfo->currentGroupId != pBlock->info.groupId) {
+    } else if (pLimitInfo->currentGroupId != pBlock->info.id.groupId) {
       // now it is the data from a new group
       pLimitInfo->remainGroupOffset -= 1;
-      pLimitInfo->currentGroupId = pBlock->info.groupId;
+      pLimitInfo->currentGroupId = pBlock->info.id.groupId;
 
       // ignore data block in current group
       if (pLimitInfo->remainGroupOffset > 0) {
@@ -153,7 +153,7 @@ static int32_t discardGroupDataBlock(SSDataBlock* pBlock, SLimitInfo* pLimitInfo
     }
 
     // set current group id of the project operator
-    pLimitInfo->currentGroupId = pBlock->info.groupId;
+    pLimitInfo->currentGroupId = pBlock->info.id.groupId;
   }
 
   return PROJECT_RETRIEVE_DONE;
@@ -164,7 +164,7 @@ static int32_t setInfoForNewGroup(SSDataBlock* pBlock, SLimitInfo* pLimitInfo, S
   // here check for a new group data, we need to handle the data of the previous group.
   ASSERT(pLimitInfo->remainGroupOffset == 0 || pLimitInfo->remainGroupOffset == -1);
 
-  if (pLimitInfo->currentGroupId != 0 && pLimitInfo->currentGroupId != pBlock->info.groupId) {
+  if (pLimitInfo->currentGroupId != 0 && pLimitInfo->currentGroupId != pBlock->info.id.groupId) {
     pLimitInfo->numOfOutputGroups += 1;
     if ((pLimitInfo->slimit.limit > 0) && (pLimitInfo->slimit.limit <= pLimitInfo->numOfOutputGroups)) {
       setOperatorCompleted(pOperator);
@@ -306,7 +306,7 @@ SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
         T_LONG_JMP(pTaskInfo->env, code);
       }
 
-      status = doIngroupLimitOffset(pLimitInfo, pBlock->info.groupId, pInfo->pRes, pOperator);
+      status = doIngroupLimitOffset(pLimitInfo, pBlock->info.id.groupId, pInfo->pRes, pOperator);
       if (status == PROJECT_RETRIEVE_CONTINUE) {
         continue;
       }
@@ -316,7 +316,7 @@ SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
 
     if (pProjectInfo->mergeDataBlocks) {
       if (pRes->info.rows > 0) {
-        pFinalRes->info.groupId = pRes->info.groupId;
+        pFinalRes->info.id.groupId = pRes->info.id.groupId;
         pFinalRes->info.version = pRes->info.version;
 
         // continue merge data, ignore the group id
@@ -385,7 +385,7 @@ SOperatorInfo* createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhy
     }
   }
 
-  SSDataBlock* pResBlock = createResDataBlock(pPhyNode->node.pOutputDataBlockDesc);
+  SSDataBlock* pResBlock = createDataBlockFromDescNode(pPhyNode->node.pOutputDataBlockDesc);
 
   int32_t numOfRows = 4096;
   size_t  keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
@@ -400,7 +400,7 @@ SOperatorInfo* createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhy
   initResultSizeInfo(&pOperator->resultInfo, numOfRows);
   blockDataEnsureCapacity(pResBlock, numOfRows);
 
-  int32_t code = initAggInfo(pSup, &pInfo->aggSup, pExprInfo, numOfExpr, keyBufSize, pTaskInfo->id.str);
+  int32_t code = initAggSup(pSup, &pInfo->aggSup, pExprInfo, numOfExpr, keyBufSize, pTaskInfo->id.str);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }
@@ -511,11 +511,11 @@ SSDataBlock* doApplyIndefinitFunction(SOperatorInfo* pOperator) {
           break;
         }
 
-        if (pIndefInfo->groupId == 0 && pBlock->info.groupId != 0) {
-          pIndefInfo->groupId = pBlock->info.groupId;  // this is the initial group result
+        if (pIndefInfo->groupId == 0 && pBlock->info.id.groupId != 0) {
+          pIndefInfo->groupId = pBlock->info.id.groupId;  // this is the initial group result
         } else {
-          if (pIndefInfo->groupId != pBlock->info.groupId) {  // reset output buffer and computing status
-            pIndefInfo->groupId = pBlock->info.groupId;
+          if (pIndefInfo->groupId != pBlock->info.id.groupId) {  // reset output buffer and computing status
+            pIndefInfo->groupId = pBlock->info.id.groupId;
             pIndefInfo->pNextGroupRes = pBlock;
             break;
           }
@@ -643,3 +643,182 @@ SSDataBlock* doGenerateSourceData(SOperatorInfo* pOperator) {
 
   return (pRes->info.rows > 0) ? pRes : NULL;
 }
+
+static void setPseudoOutputColInfo(SSDataBlock* pResult, SqlFunctionCtx* pCtx, SArray* pPseudoList) {
+  size_t num = (pPseudoList != NULL) ? taosArrayGetSize(pPseudoList) : 0;
+  for (int32_t i = 0; i < num; ++i) {
+    pCtx[i].pOutput = taosArrayGet(pResult->pDataBlock, i);
+  }
+}
+
+int32_t projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBlock* pSrcBlock, SqlFunctionCtx* pCtx,
+                              int32_t numOfOutput, SArray* pPseudoList) {
+  setPseudoOutputColInfo(pResult, pCtx, pPseudoList);
+
+  if (pSrcBlock == NULL) {
+    for (int32_t k = 0; k < numOfOutput; ++k) {
+      int32_t outputSlotId = pExpr[k].base.resSchema.slotId;
+
+      ASSERT(pExpr[k].pExpr->nodeType == QUERY_NODE_VALUE);
+      SColumnInfoData* pColInfoData = taosArrayGet(pResult->pDataBlock, outputSlotId);
+
+      int32_t type = pExpr[k].base.pParam[0].param.nType;
+      if (TSDB_DATA_TYPE_NULL == type) {
+        colDataAppendNNULL(pColInfoData, 0, 1);
+      } else {
+        colDataAppend(pColInfoData, 0, taosVariantGet(&pExpr[k].base.pParam[0].param, type), false);
+      }
+    }
+
+    pResult->info.rows = 1;
+    return TSDB_CODE_SUCCESS;
+  }
+
+  if (pResult != pSrcBlock) {
+    pResult->info.id.groupId = pSrcBlock->info.id.groupId;
+    memcpy(pResult->info.parTbName, pSrcBlock->info.parTbName, TSDB_TABLE_NAME_LEN);
+  }
+
+  // if the source equals to the destination, it is to create a new column as the result of scalar
+  // function or some operators.
+  bool createNewColModel = (pResult == pSrcBlock);
+  if (createNewColModel) {
+    blockDataEnsureCapacity(pResult, pResult->info.rows);
+  }
+
+  int32_t numOfRows = 0;
+
+  for (int32_t k = 0; k < numOfOutput; ++k) {
+    int32_t               outputSlotId = pExpr[k].base.resSchema.slotId;
+    SqlFunctionCtx*       pfCtx = &pCtx[k];
+    SInputColumnInfoData* pInputData = &pfCtx->input;
+
+    if (pExpr[k].pExpr->nodeType == QUERY_NODE_COLUMN) {  // it is a project query
+      SColumnInfoData* pColInfoData = taosArrayGet(pResult->pDataBlock, outputSlotId);
+      if (pResult->info.rows > 0 && !createNewColModel) {
+        colDataMergeCol(pColInfoData, pResult->info.rows, (int32_t*)&pResult->info.capacity, pInputData->pData[0],
+                        pInputData->numOfRows);
+      } else {
+        colDataAssign(pColInfoData, pInputData->pData[0], pInputData->numOfRows, &pResult->info);
+      }
+
+      numOfRows = pInputData->numOfRows;
+    } else if (pExpr[k].pExpr->nodeType == QUERY_NODE_VALUE) {
+      SColumnInfoData* pColInfoData = taosArrayGet(pResult->pDataBlock, outputSlotId);
+
+      int32_t offset = createNewColModel ? 0 : pResult->info.rows;
+
+      int32_t type = pExpr[k].base.pParam[0].param.nType;
+      if (TSDB_DATA_TYPE_NULL == type) {
+        colDataAppendNNULL(pColInfoData, offset, pSrcBlock->info.rows);
+      } else {
+        char* p = taosVariantGet(&pExpr[k].base.pParam[0].param, type);
+        for (int32_t i = 0; i < pSrcBlock->info.rows; ++i) {
+          colDataAppend(pColInfoData, i + offset, p, false);
+        }
+      }
+
+      numOfRows = pSrcBlock->info.rows;
+    } else if (pExpr[k].pExpr->nodeType == QUERY_NODE_OPERATOR) {
+      SArray* pBlockList = taosArrayInit(4, POINTER_BYTES);
+      taosArrayPush(pBlockList, &pSrcBlock);
+
+      SColumnInfoData* pResColData = taosArrayGet(pResult->pDataBlock, outputSlotId);
+      SColumnInfoData  idata = {.info = pResColData->info, .hasNull = true};
+
+      SScalarParam dest = {.columnData = &idata};
+      int32_t      code = scalarCalculate(pExpr[k].pExpr->_optrRoot.pRootNode, pBlockList, &dest);
+      if (code != TSDB_CODE_SUCCESS) {
+        taosArrayDestroy(pBlockList);
+        return code;
+      }
+
+      int32_t startOffset = createNewColModel ? 0 : pResult->info.rows;
+      ASSERT(pResult->info.capacity > 0);
+
+      colDataMergeCol(pResColData, startOffset, (int32_t*)&pResult->info.capacity, &idata, dest.numOfRows);
+      colDataDestroy(&idata);
+
+      numOfRows = dest.numOfRows;
+      taosArrayDestroy(pBlockList);
+    } else if (pExpr[k].pExpr->nodeType == QUERY_NODE_FUNCTION) {
+      // _rowts/_c0, not tbname column
+      if (fmIsPseudoColumnFunc(pfCtx->functionId) && (!fmIsScanPseudoColumnFunc(pfCtx->functionId))) {
+        // do nothing
+      } else if (fmIsIndefiniteRowsFunc(pfCtx->functionId)) {
+        SResultRowEntryInfo* pResInfo = GET_RES_INFO(pfCtx);
+        pfCtx->fpSet.init(pfCtx, pResInfo);
+
+        pfCtx->pOutput = taosArrayGet(pResult->pDataBlock, outputSlotId);
+        pfCtx->offset = createNewColModel ? 0 : pResult->info.rows;  // set the start offset
+
+        // set the timestamp(_rowts) output buffer
+        if (taosArrayGetSize(pPseudoList) > 0) {
+          int32_t* outputColIndex = taosArrayGet(pPseudoList, 0);
+          pfCtx->pTsOutput = (SColumnInfoData*)pCtx[*outputColIndex].pOutput;
+        }
+
+        // link pDstBlock to set selectivity value
+        if (pfCtx->subsidiaries.num > 0) {
+          pfCtx->pDstBlock = pResult;
+        }
+
+        int32_t code = pfCtx->fpSet.process(pfCtx);
+        if (code != TSDB_CODE_SUCCESS) {
+          return code;
+        }
+        numOfRows = pResInfo->numOfRes;
+      } else if (fmIsAggFunc(pfCtx->functionId)) {
+        // selective value output should be set during corresponding function execution
+        if (fmIsSelectValueFunc(pfCtx->functionId)) {
+          continue;
+        }
+        // _group_key function for "partition by tbname" + csum(col_name) query
+        SColumnInfoData* pOutput = taosArrayGet(pResult->pDataBlock, outputSlotId);
+        int32_t          slotId = pfCtx->param[0].pCol->slotId;
+
+        // todo handle the json tag
+        SColumnInfoData* pInput = taosArrayGet(pSrcBlock->pDataBlock, slotId);
+        for (int32_t f = 0; f < pSrcBlock->info.rows; ++f) {
+          bool isNull = colDataIsNull_s(pInput, f);
+          if (isNull) {
+            colDataAppendNULL(pOutput, pResult->info.rows + f);
+          } else {
+            char* data = colDataGetData(pInput, f);
+            colDataAppend(pOutput, pResult->info.rows + f, data, isNull);
+          }
+        }
+
+      } else {
+        SArray* pBlockList = taosArrayInit(4, POINTER_BYTES);
+        taosArrayPush(pBlockList, &pSrcBlock);
+
+        SColumnInfoData* pResColData = taosArrayGet(pResult->pDataBlock, outputSlotId);
+        SColumnInfoData  idata = {.info = pResColData->info, .hasNull = true};
+
+        SScalarParam dest = {.columnData = &idata};
+        int32_t      code = scalarCalculate((SNode*)pExpr[k].pExpr->_function.pFunctNode, pBlockList, &dest);
+        if (code != TSDB_CODE_SUCCESS) {
+          taosArrayDestroy(pBlockList);
+          return code;
+        }
+
+        int32_t startOffset = createNewColModel ? 0 : pResult->info.rows;
+        ASSERT(pResult->info.capacity > 0);
+        colDataMergeCol(pResColData, startOffset, (int32_t*)&pResult->info.capacity, &idata, dest.numOfRows);
+        colDataDestroy(&idata);
+
+        numOfRows = dest.numOfRows;
+        taosArrayDestroy(pBlockList);
+      }
+    } else {
+      return TSDB_CODE_OPS_NOT_SUPPORT;
+    }
+  }
+
+  if (!createNewColModel) {
+    pResult->info.rows += numOfRows;
+  }
+
+  return TSDB_CODE_SUCCESS;
+}

source/libs/executor/src/scanoperator.c

@@ -184,7 +184,7 @@ static int32_t doDynamicPruneDataBlock(SOperatorInfo* pOperator, SDataBlockInfo*
   SExprSupp* pSup1 = pTableScanInfo->base.pdInfo.pExprSup;
 
   SFilePage*  pPage = NULL;
-  SResultRow* pRow = getTableGroupOutputBuf(pOperator, pBlockInfo->groupId, &pPage);
+  SResultRow* pRow = getTableGroupOutputBuf(pOperator, pBlockInfo->id.groupId, &pPage);
 
   if (pRow == NULL) {
     return TSDB_CODE_SUCCESS;
@@ -484,13 +484,13 @@ int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int
   // 1. check if it is existed in meta cache
   if (pCache == NULL) {
     metaReaderInit(&mr, pHandle->meta, 0);
-    code = metaGetTableEntryByUidCache(&mr, pBlock->info.uid);
+    code = metaGetTableEntryByUidCache(&mr, pBlock->info.id.uid);
     if (code != TSDB_CODE_SUCCESS) {
       if (terrno == TSDB_CODE_PAR_TABLE_NOT_EXIST) {
-        qWarn("failed to get table meta, table may have been dropped, uid:0x%" PRIx64 ", code:%s, %s", pBlock->info.uid,
+        qWarn("failed to get table meta, table may have been dropped, uid:0x%" PRIx64 ", code:%s, %s", pBlock->info.id.uid,
               tstrerror(terrno), idStr);
       } else {
-        qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", pBlock->info.uid, tstrerror(terrno), idStr);
+        qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", pBlock->info.id.uid, tstrerror(terrno), idStr);
       }
       metaReaderClear(&mr);
       return terrno;
@@ -505,16 +505,16 @@ int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int
   } else {
     pCache->metaFetch += 1;
 
-    h = taosLRUCacheLookup(pCache->pTableMetaEntryCache, &pBlock->info.uid, sizeof(pBlock->info.uid));
+    h = taosLRUCacheLookup(pCache->pTableMetaEntryCache, &pBlock->info.id.uid, sizeof(pBlock->info.id.uid));
     if (h == NULL) {
       metaReaderInit(&mr, pHandle->meta, 0);
-      code = metaGetTableEntryByUidCache(&mr, pBlock->info.uid);
+      code = metaGetTableEntryByUidCache(&mr, pBlock->info.id.uid);
       if (code != TSDB_CODE_SUCCESS) {
         if (terrno == TSDB_CODE_PAR_TABLE_NOT_EXIST) {
           qWarn("failed to get table meta, table may have been dropped, uid:0x%" PRIx64 ", code:%s, %s",
-                pBlock->info.uid, tstrerror(terrno), idStr);
+                pBlock->info.id.uid, tstrerror(terrno), idStr);
         } else {
-          qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", pBlock->info.uid, tstrerror(terrno),
+          qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", pBlock->info.id.uid, tstrerror(terrno),
                  idStr);
         }
         metaReaderClear(&mr);
@@ -528,7 +528,7 @@ int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int
       val = *pVal;
       freeReader = true;
 
-      int32_t ret = taosLRUCacheInsert(pCache->pTableMetaEntryCache, &pBlock->info.uid, sizeof(uint64_t), pVal,
+      int32_t ret = taosLRUCacheInsert(pCache->pTableMetaEntryCache, &pBlock->info.id.uid, sizeof(uint64_t), pVal,
                                        sizeof(STableCachedVal), freeCachedMetaItem, NULL, TAOS_LRU_PRIORITY_LOW);
       if (ret != TAOS_LRU_STATUS_OK) {
         qError("failed to put meta into lru cache, code:%d, %s", ret, idStr);
@@ -642,13 +642,13 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
     SDataBlockInfo* pBInfo = &pBlock->info;
 
     int32_t rows = 0;
-    tsdbRetrieveDataBlockInfo(pTableScanInfo->base.dataReader, &rows, &pBInfo->uid, &pBInfo->window);
+    tsdbRetrieveDataBlockInfo(pTableScanInfo->base.dataReader, &rows, &pBInfo->id.uid, &pBInfo->window);
 
     blockDataEnsureCapacity(pBlock, rows);  // todo remove it latter
     pBInfo->rows = rows;
 
-    ASSERT(pBInfo->uid != 0);
-    pBlock->info.groupId = getTableGroupId(pTaskInfo->pTableInfoList, pBlock->info.uid);
+    ASSERT(pBInfo->id.uid != 0);
+    pBlock->info.id.groupId = getTableGroupId(pTaskInfo->pTableInfoList, pBlock->info.id.uid);
 
     uint32_t status = 0;
     int32_t  code = loadDataBlock(pOperator, &pTableScanInfo->base, pBlock, &status);
@@ -668,13 +668,13 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
     pOperator->cost.totalCost = pTableScanInfo->base.readRecorder.elapsedTime;
 
     // todo refactor
-    /*pTableScanInfo->lastStatus.uid = pBlock->info.uid;*/
+    /*pTableScanInfo->lastStatus.uid = pBlock->info.id.uid;*/
     /*pTableScanInfo->lastStatus.ts = pBlock->info.window.ekey;*/
     pTaskInfo->streamInfo.lastStatus.type = TMQ_OFFSET__SNAPSHOT_DATA;
-    pTaskInfo->streamInfo.lastStatus.uid = pBlock->info.uid;
+    pTaskInfo->streamInfo.lastStatus.uid = pBlock->info.id.uid;
     pTaskInfo->streamInfo.lastStatus.ts = pBlock->info.window.ekey;
 
-    ASSERT(pBlock->info.uid != 0);
+    ASSERT(pBlock->info.id.uid != 0);
     return pBlock;
   }
   return NULL;
@@ -786,7 +786,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
 
     SSDataBlock* result = doGroupedTableScan(pOperator);
     if (result != NULL) {
-      ASSERT(result->info.uid != 0);
+      ASSERT(result->info.id.uid != 0);
       return result;
     }
 
@@ -885,7 +885,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
   pInfo->base.dataBlockLoadFlag = pTableScanNode->dataRequired;
 
   initResultSizeInfo(&pOperator->resultInfo, 4096);
-  pInfo->pResBlock = createResDataBlock(pDescNode);
+  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);
   blockDataEnsureCapacity(pInfo->pResBlock, pOperator->resultInfo.capacity);
 
   code = filterInitFromNode((SNode*)pTableScanNode->scan.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);
@@ -1009,7 +1009,7 @@ static SSDataBlock* readPreVersionData(SOperatorInfo* pTableScanOp, uint64_t tbU
     SDataBlockInfo* pBInfo = &pBlock->info;
 
     int32_t rows = 0;
-    tsdbRetrieveDataBlockInfo(pReader, &rows, &pBInfo->uid, &pBInfo->window);
+    tsdbRetrieveDataBlockInfo(pReader, &rows, &pBInfo->id.uid, &pBInfo->window);
 
     SArray* pCols = tsdbRetrieveDataBlock(pReader, NULL);
     blockDataEnsureCapacity(pBlock, rows);
@@ -1018,7 +1018,7 @@ static SSDataBlock* readPreVersionData(SOperatorInfo* pTableScanOp, uint64_t tbU
     relocateColumnData(pBlock, pTableScanInfo->base.matchInfo.pList, pCols, true);
     doSetTagColumnData(&pTableScanInfo->base, pBlock, pTaskInfo, rows);
 
-    pBlock->info.groupId = getTableGroupId(pTaskInfo->pTableInfoList, pBInfo->uid);
+    pBlock->info.id.groupId = getTableGroupId(pTaskInfo->pTableInfoList, pBInfo->id.uid);
   }
 
   tsdbReaderClose(pReader);
@@ -1175,7 +1175,7 @@ static SSDataBlock* doRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32
         pResult->info.calWin = pInfo->updateWin;
         return pResult;
       }
-    } else if (pResult->info.groupId == pInfo->groupId) {
+    } else if (pResult->info.id.groupId == pInfo->groupId) {
       pResult->info.calWin = pInfo->updateWin;
       return pResult;
     }
@@ -1366,7 +1366,7 @@ void calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock) {
   if (pBlock == NULL || pBlock->info.rows == 0) return;
 
   void* tbname = NULL;
-  if (streamStateGetParName(pInfo->pStreamScanOp->pTaskInfo->streamInfo.pState, pBlock->info.groupId, &tbname) < 0) {
+  if (streamStateGetParName(pInfo->pStreamScanOp->pTaskInfo->streamInfo.pState, pBlock->info.id.groupId, &tbname) < 0) {
     pBlock->info.parTbName[0] = 0;
   } else {
     memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);
@@ -1399,8 +1399,8 @@ void calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock) {
     pBlock->info.parTbName[0] = 0;
   }
 
-  if (pBlock->info.groupId && pBlock->info.parTbName[0]) {
-    streamStatePutParName(pState, pBlock->info.groupId, pBlock->info.parTbName);
+  if (pBlock->info.id.groupId && pBlock->info.parTbName[0]) {
+    streamStatePutParName(pState, pBlock->info.id.groupId, pBlock->info.parTbName);
   }
 
   blockDataDestroy(pSrcBlock);
@@ -1434,7 +1434,7 @@ static void checkUpdateData(SStreamScanInfo* pInfo, bool invertible, SSDataBlock
   SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex);
   ASSERT(pColDataInfo->info.type == TSDB_DATA_TYPE_TIMESTAMP);
   TSKEY* tsCol = (TSKEY*)pColDataInfo->pData;
-  bool   tableInserted = updateInfoIsTableInserted(pInfo->pUpdateInfo, pBlock->info.uid);
+  bool   tableInserted = updateInfoIsTableInserted(pInfo->pUpdateInfo, pBlock->info.id.uid);
   for (int32_t rowId = 0; rowId < pBlock->info.rows; rowId++) {
     SResultRowInfo dumyInfo;
     dumyInfo.cur.pageId = -1;
@@ -1445,18 +1445,18 @@ static void checkUpdateData(SStreamScanInfo* pInfo, bool invertible, SSDataBlock
       isClosed = isCloseWindow(&win, &pInfo->twAggSup);
     }
     // must check update info first.
-    bool update = updateInfoIsUpdated(pInfo->pUpdateInfo, pBlock->info.uid, tsCol[rowId]);
+    bool update = updateInfoIsUpdated(pInfo->pUpdateInfo, pBlock->info.id.uid, tsCol[rowId]);
     bool closedWin = isClosed && isSignleIntervalWindow(pInfo) &&
-                     isDeletedStreamWindow(&win, pBlock->info.groupId,
+                     isDeletedStreamWindow(&win, pBlock->info.id.groupId,
                                            pInfo->pTableScanOp->pTaskInfo->streamInfo.pState, &pInfo->twAggSup);
     if ((update || closedWin) && out) {
       qDebug("stream update check not pass, update %d, closedWin %d", update, closedWin);
       uint64_t gpId = 0;
-      appendOneRowToStreamSpecialBlock(pInfo->pUpdateDataRes, tsCol + rowId, tsCol + rowId, &pBlock->info.uid, &gpId,
+      appendOneRowToStreamSpecialBlock(pInfo->pUpdateDataRes, tsCol + rowId, tsCol + rowId, &pBlock->info.id.uid, &gpId,
                                        NULL);
       if (closedWin && pInfo->partitionSup.needCalc) {
         gpId = calGroupIdByData(&pInfo->partitionSup, pInfo->pPartScalarSup, pBlock, rowId);
-        appendOneRowToStreamSpecialBlock(pInfo->pUpdateDataRes, tsCol + rowId, tsCol + rowId, &pBlock->info.uid, &gpId,
+        appendOneRowToStreamSpecialBlock(pInfo->pUpdateDataRes, tsCol + rowId, tsCol + rowId, &pBlock->info.id.uid, &gpId,
                                          NULL);
       }
     }
@@ -1476,11 +1476,11 @@ static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock
   blockDataEnsureCapacity(pInfo->pRes, pBlock->info.rows);
 
   pInfo->pRes->info.rows = pBlock->info.rows;
-  pInfo->pRes->info.uid = pBlock->info.uid;
+  pInfo->pRes->info.id.uid = pBlock->info.id.uid;
   pInfo->pRes->info.type = STREAM_NORMAL;
   pInfo->pRes->info.version = pBlock->info.version;
 
-  pInfo->pRes->info.groupId = getTableGroupId(pTaskInfo->pTableInfoList, pBlock->info.uid);
+  pInfo->pRes->info.id.groupId = getTableGroupId(pTaskInfo->pTableInfoList, pBlock->info.id.uid);
 
   // todo extract method
   for (int32_t i = 0; i < taosArrayGetSize(pInfo->matchInfo.pList); ++i) {
@@ -1807,8 +1807,8 @@ FETCH_NEXT_BLOCK:
 
     int32_t      current = pInfo->validBlockIndex++;
     SSDataBlock* pBlock = taosArrayGetP(pInfo->pBlockLists, current);
-    if (pBlock->info.groupId && pBlock->info.parTbName[0]) {
-      streamStatePutParName(pTaskInfo->streamInfo.pState, pBlock->info.groupId, pBlock->info.parTbName);
+    if (pBlock->info.id.groupId && pBlock->info.parTbName[0]) {
+      streamStatePutParName(pTaskInfo->streamInfo.pState, pBlock->info.id.groupId, pBlock->info.parTbName);
     }
     // TODO move into scan
     pBlock->info.calWin.skey = INT64_MIN;
@@ -1960,7 +1960,7 @@ FETCH_NEXT_BLOCK:
 
         setBlockIntoRes(pInfo, &block, false);
 
-        if (updateInfoIgnore(pInfo->pUpdateInfo, &pInfo->pRes->info.window, pInfo->pRes->info.groupId,
+        if (updateInfoIgnore(pInfo->pUpdateInfo, &pInfo->pRes->info.window, pInfo->pRes->info.id.groupId,
                              pInfo->pRes->info.version)) {
           printDataBlock(pInfo->pRes, "stream scan ignore");
           blockDataCleanup(pInfo->pRes);
@@ -2050,7 +2050,7 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
       }
 
       int32_t rows = 0;
-      tsdbRetrieveDataBlockInfo(pInfo->dataReader, &rows, &pBlock->info.uid, &pBlock->info.window);
+      tsdbRetrieveDataBlockInfo(pInfo->dataReader, &rows, &pBlock->info.id.uid, &pBlock->info.window);
       pBlock->info.rows = rows;
 
       SArray* pCols = tsdbRetrieveDataBlock(pInfo->dataReader, NULL);
@@ -2059,9 +2059,9 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
         longjmp(pTaskInfo->env, terrno);
       }
 
-      qDebug("tmqsnap doRawScan get data uid:%" PRId64 "", pBlock->info.uid);
+      qDebug("tmqsnap doRawScan get data uid:%" PRId64 "", pBlock->info.id.uid);
       pTaskInfo->streamInfo.lastStatus.type = TMQ_OFFSET__SNAPSHOT_DATA;
-      pTaskInfo->streamInfo.lastStatus.uid = pBlock->info.uid;
+      pTaskInfo->streamInfo.lastStatus.uid = pBlock->info.id.uid;
       pTaskInfo->streamInfo.lastStatus.ts = pBlock->info.window.ekey;
       return pBlock;
     }
@@ -2352,7 +2352,7 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
     goto _error;
   }
 
-  pInfo->pRes = createResDataBlock(pDescNode);
+  pInfo->pRes = createDataBlockFromDescNode(pDescNode);
   pInfo->pUpdateRes = createSpecialDataBlock(STREAM_CLEAR);
   pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
   pInfo->windowSup = (SWindowSupporter){.pStreamAggSup = NULL, .gap = -1, .parentType = QUERY_NODE_PHYSICAL_PLAN};
@@ -2476,8 +2476,7 @@ static void destroyTagScanOperatorInfo(void* param) {
   taosMemoryFreeClear(param);
 }
 
-SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pPhyNode,
-                                         STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo) {
+SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo) {
   STagScanInfo*  pInfo = taosMemoryCalloc(1, sizeof(STagScanInfo));
   SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   if (pInfo == NULL || pOperator == NULL) {
@@ -2499,7 +2498,7 @@ SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysi
     goto _error;
   }
 
-  pInfo->pRes = createResDataBlock(pDescNode);
+  pInfo->pRes = createDataBlockFromDescNode(pDescNode);
   pInfo->readHandle = *pReadHandle;
   pInfo->curPos = 0;
 
@@ -2556,7 +2555,7 @@ static SSDataBlock* getTableDataBlockImpl(void* param) {
     blockDataCleanup(pBlock);
 
     int32_t rows = 0;
-    tsdbRetrieveDataBlockInfo(reader, &rows, &pBlock->info.uid, &pBlock->info.window);
+    tsdbRetrieveDataBlockInfo(reader, &rows, &pBlock->info.id.uid, &pBlock->info.window);
     blockDataEnsureCapacity(pBlock, rows);
     pBlock->info.rows = rows;
 
@@ -2578,7 +2577,7 @@ static SSDataBlock* getTableDataBlockImpl(void* param) {
       continue;
     }
 
-    pBlock->info.groupId = getTableGroupId(pTaskInfo->pTableInfoList, pBlock->info.uid);
+    pBlock->info.id.groupId = getTableGroupId(pTaskInfo->pTableInfoList, pBlock->info.id.uid);
 
     pOperator->resultInfo.totalRows += pBlock->info.rows;
     pTableScanInfo->base.readRecorder.elapsedTime += (taosGetTimestampUs() - st) / 1000.0;
@@ -2613,7 +2612,7 @@ SArray* generateSortByTsInfo(SArray* colMatchInfo, int32_t order) {
   return pList;
 }
 
-int32_t dumpSQueryTableCond(const SQueryTableDataCond* src, SQueryTableDataCond* dst) {
+int32_t dumpQueryTableCond(const SQueryTableDataCond* src, SQueryTableDataCond* dst) {
   memcpy((void*)dst, (void*)src, sizeof(SQueryTableDataCond));
   dst->colList = taosMemoryCalloc(src->numOfCols, sizeof(SColumnInfo));
   for (int i = 0; i < src->numOfCols; i++) {
@@ -2664,7 +2663,7 @@ int32_t startGroupTableMergeScan(SOperatorInfo* pOperator) {
     taosArrayPush(pInfo->sortSourceParams, &param);
 
     SQueryTableDataCond cond;
-    dumpSQueryTableCond(&pInfo->base.cond, &cond);
+    dumpQueryTableCond(&pInfo->base.cond, &cond);
     taosArrayPush(pInfo->queryConds, &cond);
   }
 
@@ -2774,7 +2773,7 @@ SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {
     pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pInfo->pResBlock, pOperator->resultInfo.capacity,
                                               pOperator);
     if (pBlock != NULL) {
-      pBlock->info.groupId = pInfo->groupId;
+      pBlock->info.id.groupId = pInfo->groupId;
       pOperator->resultInfo.totalRows += pBlock->info.rows;
       return pBlock;
     } else {
@@ -2900,7 +2899,7 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN
   }
 
   initResultSizeInfo(&pOperator->resultInfo, 1024);
-  pInfo->pResBlock = createResDataBlock(pDescNode);
+  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);
   blockDataEnsureCapacity(pInfo->pResBlock, pOperator->resultInfo.capacity);
 
   pInfo->sortSourceParams = taosArrayInit(64, sizeof(STableMergeScanSortSourceParam));

source/libs/executor/src/sortoperator.c

@@ -47,7 +47,7 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode*
   SDataBlockDescNode* pDescNode = pSortNode->node.pOutputDataBlockDesc;
 
   int32_t      numOfCols = 0;
-  SSDataBlock* pResBlock = createResDataBlock(pDescNode);
+  SSDataBlock* pResBlock = createDataBlockFromDescNode(pDescNode);
   SExprInfo*   pExprInfo = createExprInfo(pSortNode->pExprs, NULL, &numOfCols);
 
   int32_t numOfOutputCols = 0;
@@ -359,7 +359,7 @@ SSDataBlock* fetchNextGroupSortDataBlock(void* param) {
     SOperatorInfo* childOp = source->childOpInfo;
     SSDataBlock*   block = childOp->fpSet.getNextFn(childOp);
     if (block != NULL) {
-      if (block->info.groupId == grpSortOpInfo->currGroupId) {
+      if (block->info.id.groupId == grpSortOpInfo->currGroupId) {
         grpSortOpInfo->childOpStatus = CHILD_OP_SAME_GROUP;
         return block;
       } else {
@@ -439,7 +439,7 @@ SSDataBlock* doGroupSort(SOperatorInfo* pOperator) {
       setOperatorCompleted(pOperator);
       return NULL;
     }
-    pInfo->currGroupId = pInfo->prefetchedSortInput->info.groupId;
+    pInfo->currGroupId = pInfo->prefetchedSortInput->info.id.groupId;
     pInfo->childOpStatus = CHILD_OP_NEW_GROUP;
     beginSortGroup(pOperator);
   }
@@ -451,13 +451,13 @@ SSDataBlock* doGroupSort(SOperatorInfo* pOperator) {
     pBlock = getGroupSortedBlockData(pInfo->pCurrSortHandle, pInfo->binfo.pRes, pOperator->resultInfo.capacity,
                                      pInfo->matchInfo.pList, pInfo);
     if (pBlock != NULL) {
-      pBlock->info.groupId = pInfo->currGroupId;
+      pBlock->info.id.groupId = pInfo->currGroupId;
       pOperator->resultInfo.totalRows += pBlock->info.rows;
       return pBlock;
     } else {
       if (pInfo->childOpStatus == CHILD_OP_NEW_GROUP) {
         finishSortGroup(pOperator);
-        pInfo->currGroupId = pInfo->prefetchedSortInput->info.groupId;
+        pInfo->currGroupId = pInfo->prefetchedSortInput->info.id.groupId;
         beginSortGroup(pOperator);
       } else if (pInfo->childOpStatus == CHILD_OP_FINISHED) {
         finishSortGroup(pOperator);
@@ -509,7 +509,7 @@ SOperatorInfo* createGroupSortOperatorInfo(SOperatorInfo* downstream, SGroupSort
   initResultSizeInfo(&pOperator->resultInfo, 1024);
   pOperator->exprSupp.pCtx = createSqlFunctionCtx(pExprInfo, numOfCols, &pOperator->exprSupp.rowEntryInfoOffset);
 
-  pInfo->binfo.pRes = createResDataBlock(pDescNode);
+  pInfo->binfo.pRes = createDataBlockFromDescNode(pDescNode);
   blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
 
   int32_t numOfOutputCols = 0;
@@ -691,10 +691,10 @@ SSDataBlock* getMultiwaySortedBlockData(SSortHandle* pHandle, SSDataBlock* pData
 
     pInfo->limitInfo.numOfOutputRows += p->info.rows;
     pDataBlock->info.rows = p->info.rows;
-    pDataBlock->info.groupId = pInfo->groupId;
+    pDataBlock->info.id.groupId = pInfo->groupId;
   }
 
-  qDebug("%s get sorted block, groupId:0x%" PRIx64 " rows:%d", GET_TASKID(pTaskInfo), pDataBlock->info.groupId,
+  qDebug("%s get sorted block, groupId:0x%" PRIx64 " rows:%d", GET_TASKID(pTaskInfo), pDataBlock->info.id.groupId,
          pDataBlock->info.rows);
 
   return (pDataBlock->info.rows > 0) ? pDataBlock : NULL;
@@ -766,7 +766,7 @@ SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** downStreams, size
   }
 
   initLimitInfo(pMergePhyNode->node.pLimit, pMergePhyNode->node.pSlimit, &pInfo->limitInfo);
-  pInfo->binfo.pRes = createResDataBlock(pDescNode);
+  pInfo->binfo.pRes = createDataBlockFromDescNode(pDescNode);
 
   int32_t rowSize = pInfo->binfo.pRes->info.rowSize;
   ASSERT(rowSize < 100 * 1024 * 1024);
@@ -779,7 +779,7 @@ SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** downStreams, size
   }
 
   SPhysiNode*  pChildNode = (SPhysiNode*)nodesListGetNode(pPhyNode->pChildren, 0);
-  SSDataBlock* pInputBlock = createResDataBlock(pChildNode->pOutputDataBlockDesc);
+  SSDataBlock* pInputBlock = createDataBlockFromDescNode(pChildNode->pOutputDataBlockDesc);
 
   initResultSizeInfo(&pOperator->resultInfo, 4096);
   blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);

source/libs/executor/src/sysscanoperator.c

@@ -1411,7 +1411,7 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScan
   pInfo->pUser = taosMemoryStrDup((void*)pUser);
   pInfo->sysInfo = pScanPhyNode->sysInfo;
   pInfo->showRewrite = pScanPhyNode->showRewrite;
-  pInfo->pRes = createResDataBlock(pDescNode);
+  pInfo->pRes = createDataBlockFromDescNode(pDescNode);
 
   pInfo->pCondition = pScanNode->node.pConditions;
   code = filterInitFromNode(pScanNode->node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);
@@ -1928,7 +1928,7 @@ SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDi
     pInfo->readHandle = *readHandle;
     pInfo->uid = pBlockScanNode->suid;
 
-    pInfo->pResBlock = createResDataBlock(pBlockScanNode->node.pOutputDataBlockDesc);
+    pInfo->pResBlock = createDataBlockFromDescNode(pBlockScanNode->node.pOutputDataBlockDesc);
     blockDataEnsureCapacity(pInfo->pResBlock, 1);
 
     int32_t    numOfCols = 0;

source/libs/executor/src/tfill.c

@@ -1092,7 +1092,7 @@ static bool checkResult(SStreamFillSupporter* pFillSup, TSKEY ts, uint64_t group
 }
 
 static void buildFillResult(SResultRowData* pResRow, SStreamFillSupporter* pFillSup, TSKEY ts, SSDataBlock* pBlock) {
-  uint64_t groupId = pBlock->info.groupId;
+  uint64_t groupId = pBlock->info.id.groupId;
   if (pFillSup->hasDelete && !checkResult(pFillSup, ts, groupId)) {
     return;
   }
@@ -1131,7 +1131,7 @@ static void doStreamFillNormal(SStreamFillSupporter* pFillSup, SStreamFillInfo*
 
 static void doStreamFillLinear(SStreamFillSupporter* pFillSup, SStreamFillInfo* pFillInfo, SSDataBlock* pBlock) {
   while (hasRemainCalc(pFillInfo) && pBlock->info.rows < pBlock->info.capacity) {
-    uint64_t groupId = pBlock->info.groupId;
+    uint64_t groupId = pBlock->info.id.groupId;
     SWinKey  key = {.groupId = groupId, .ts = pFillInfo->current};
     if (pFillSup->hasDelete && !checkResult(pFillSup, pFillInfo->current, groupId)) {
       pFillInfo->current = taosTimeAdd(pFillInfo->current, pFillSup->interval.sliding, pFillSup->interval.slidingUnit,
@@ -1230,7 +1230,7 @@ void keepBlockRowInDiscBuf(SOperatorInfo* pOperator, SStreamFillInfo* pFillInfo,
 
 static void doFillResults(SOperatorInfo* pOperator, SStreamFillSupporter* pFillSup, SStreamFillInfo* pFillInfo,
                           SSDataBlock* pBlock, TSKEY* tsCol, int32_t rowId, SSDataBlock* pRes) {
-  uint64_t groupId = pBlock->info.groupId;
+  uint64_t groupId = pBlock->info.id.groupId;
   getWindowFromDiscBuf(pOperator, tsCol[rowId], groupId, pFillSup);
   if (pFillSup->prev.key == pFillInfo->preRowKey) {
     resetFillWindow(&pFillSup->prev);
@@ -1245,9 +1245,9 @@ static void doStreamFillImpl(SOperatorInfo* pOperator) {
   SStreamFillSupporter*    pFillSup = pInfo->pFillSup;
   SStreamFillInfo*         pFillInfo = pInfo->pFillInfo;
   SSDataBlock*             pBlock = pInfo->pSrcBlock;
-  uint64_t                 groupId = pBlock->info.groupId;
+  uint64_t                 groupId = pBlock->info.id.groupId;
   SSDataBlock*             pRes = pInfo->pRes;
-  pRes->info.groupId = groupId;
+  pRes->info.id.groupId = groupId;
   if (hasRemainCalc(pFillInfo)) {
     doStreamFillRange(pFillInfo, pFillSup, pRes);
   }
@@ -1282,7 +1282,9 @@ static void doStreamFillImpl(SOperatorInfo* pOperator) {
   blockDataCleanup(pInfo->pSrcBlock);
 }
 
-static void buildDeleteRange(TSKEY start, TSKEY end, uint64_t groupId, SSDataBlock* delRes) {
+static void buildDeleteRange(SOperatorInfo* pOp, TSKEY start, TSKEY end, uint64_t groupId, SSDataBlock* delRes) {
+  SStreamState* pState = pOp->pTaskInfo->streamInfo.pState;
+
   SSDataBlock*     pBlock = delRes;
   SColumnInfoData* pStartCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
   SColumnInfoData* pEndCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);
@@ -1290,25 +1292,42 @@ static void buildDeleteRange(TSKEY start, TSKEY end, uint64_t groupId, SSDataBlo
   SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
   SColumnInfoData* pCalStartCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);
   SColumnInfoData* pCalEndCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);
+  SColumnInfoData* pTbNameCol = taosArrayGet(pBlock->pDataBlock, TABLE_NAME_COLUMN_INDEX);
   colDataAppend(pStartCol, pBlock->info.rows, (const char*)&start, false);
   colDataAppend(pEndCol, pBlock->info.rows, (const char*)&end, false);
   colDataAppendNULL(pUidCol, pBlock->info.rows);
   colDataAppend(pGroupCol, pBlock->info.rows, (const char*)&groupId, false);
   colDataAppendNULL(pCalStartCol, pBlock->info.rows);
   colDataAppendNULL(pCalEndCol, pBlock->info.rows);
+
+  SColumnInfoData* pTableCol = taosArrayGet(pBlock->pDataBlock, TABLE_NAME_COLUMN_INDEX);
+
+  void* tbname = NULL;
+  streamStateGetParName(pOp->pTaskInfo->streamInfo.pState, groupId, &tbname);
+  if (tbname == NULL) {
+    colDataAppendNULL(pTableCol, pBlock->info.rows);
+  } else {
+    char parTbName[VARSTR_HEADER_SIZE + TSDB_TABLE_NAME_LEN];
+    STR_WITH_MAXSIZE_TO_VARSTR(parTbName, tbname, sizeof(parTbName));
+    colDataAppend(pTableCol, pBlock->info.rows, (const char*)parTbName, false);
+  }
+  tdbFree(tbname);
+
   pBlock->info.rows++;
 }
 
-static void buildDeleteResult(SStreamFillSupporter* pFillSup, TSKEY startTs, TSKEY endTs, uint64_t groupId,
+static void buildDeleteResult(SOperatorInfo* pOperator, TSKEY startTs, TSKEY endTs, uint64_t groupId,
                               SSDataBlock* delRes) {
+  SStreamFillOperatorInfo* pInfo = pOperator->info;
+  SStreamFillSupporter*    pFillSup = pInfo->pFillSup;
   if (hasPrevWindow(pFillSup)) {
     TSKEY start = getNextWindowTs(pFillSup->prev.key, &pFillSup->interval);
-    buildDeleteRange(start, endTs, groupId, delRes);
+    buildDeleteRange(pOperator, start, endTs, groupId, delRes);
   } else if (hasNextWindow(pFillSup)) {
     TSKEY end = getPrevWindowTs(pFillSup->next.key, &pFillSup->interval);
-    buildDeleteRange(startTs, end, groupId, delRes);
+    buildDeleteRange(pOperator, startTs, end, groupId, delRes);
   } else {
-    buildDeleteRange(startTs, endTs, groupId, delRes);
+    buildDeleteRange(pOperator, startTs, endTs, groupId, delRes);
   }
 }
 
@@ -1319,7 +1338,7 @@ static void doDeleteFillResultImpl(SOperatorInfo* pOperator, TSKEY startTs, TSKE
   SWinKey key = {.ts = startTs, .groupId = groupId};
   if (!pInfo->pFillInfo->needFill) {
     streamStateFillDel(pOperator->pTaskInfo->streamInfo.pState, &key);
-    buildDeleteResult(pInfo->pFillSup, startTs, endTs, groupId, pInfo->pDelRes);
+    buildDeleteResult(pOperator, startTs, endTs, groupId, pInfo->pDelRes);
   } else {
     STimeRange tw = {
         .skey = startTs,
@@ -1342,14 +1361,14 @@ static void doDeleteFillFinalize(SOperatorInfo* pOperator) {
   tSimpleHashClear(pInfo->pFillSup->pResMap);
   for (; pFillInfo->delIndex < size; pFillInfo->delIndex++) {
     STimeRange* range = taosArrayGet(pFillInfo->delRanges, pFillInfo->delIndex);
-    if (pInfo->pRes->info.groupId != 0 && pInfo->pRes->info.groupId != range->groupId) {
+    if (pInfo->pRes->info.id.groupId != 0 && pInfo->pRes->info.id.groupId != range->groupId) {
       return;
     }
     getWindowFromDiscBuf(pOperator, range->skey, range->groupId, pInfo->pFillSup);
     setDeleteFillValueInfo(range->skey, range->ekey, pInfo->pFillSup, pInfo->pFillInfo);
     if (pInfo->pFillInfo->needFill) {
       doStreamFillRange(pInfo->pFillInfo, pInfo->pFillSup, pInfo->pRes);
-      pInfo->pRes->info.groupId = range->groupId;
+      pInfo->pRes->info.id.groupId = range->groupId;
     }
     SWinKey key = {.ts = range->skey, .groupId = range->groupId};
     streamStateFillDel(pOperator->pTaskInfo->streamInfo.pState, &key);
@@ -1435,7 +1454,7 @@ static void doApplyStreamScalarCalculation(SOperatorInfo* pOperator, SSDataBlock
   pSup = &pInfo->pFillSup->notFillExprSup;
   setInputDataBlock(pSup, pSrcBlock, TSDB_ORDER_ASC, MAIN_SCAN, false);
   projectApplyFunctions(pSup->pExprInfo, pDstBlock, pSrcBlock, pSup->pCtx, pSup->numOfExprs, NULL);
-  pDstBlock->info.groupId = pSrcBlock->info.groupId;
+  pDstBlock->info.id.groupId = pSrcBlock->info.id.groupId;
 
   blockDataUpdateTsWindow(pDstBlock, pInfo->primaryTsCol);
 }
@@ -1651,9 +1670,9 @@ SOperatorInfo* createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFi
   }
 
   initResultSizeInfo(&pOperator->resultInfo, 4096);
-  pInfo->pRes = createResDataBlock(pPhyFillNode->node.pOutputDataBlockDesc);
-  pInfo->pSrcBlock = createResDataBlock(pPhyFillNode->node.pOutputDataBlockDesc);
-  pInfo->pPrevSrcBlock = createResDataBlock(pPhyFillNode->node.pOutputDataBlockDesc);
+  pInfo->pRes = createDataBlockFromDescNode(pPhyFillNode->node.pOutputDataBlockDesc);
+  pInfo->pSrcBlock = createDataBlockFromDescNode(pPhyFillNode->node.pOutputDataBlockDesc);
+  pInfo->pPrevSrcBlock = createDataBlockFromDescNode(pPhyFillNode->node.pOutputDataBlockDesc);
   blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);
   blockDataEnsureCapacity(pInfo->pSrcBlock, pOperator->resultInfo.capacity);
   blockDataEnsureCapacity(pInfo->pPrevSrcBlock, pOperator->resultInfo.capacity);
@@ -1715,9 +1734,9 @@ SOperatorInfo* createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFi
   }
 
   pInfo->srcRowIndex = 0;
-  setOperatorInfo(pOperator, "StreamFillOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL, false, OP_NOT_OPENED, pInfo, pTaskInfo);
-  pOperator->fpSet =
-      createOperatorFpSet(operatorDummyOpenFn, doStreamFill, NULL, destroyStreamFillOperatorInfo, NULL);
+  setOperatorInfo(pOperator, "StreamFillOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL, false, OP_NOT_OPENED, pInfo,
+                  pTaskInfo);
+  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamFill, NULL, destroyStreamFillOperatorInfo, NULL);
 
   code = appendDownstream(pOperator, &downstream, 1);
   if (code != TSDB_CODE_SUCCESS) {

source/libs/executor/src/tsort.c

@@ -417,8 +417,8 @@ int32_t msortComparFn(const void* pLeft, const void* pRight, void* param) {
   SSDataBlock* pRightBlock = pRightSource->src.pBlock;
 
   if (pParam->cmpGroupId) {
-    if (pLeftBlock->info.groupId != pRightBlock->info.groupId) {
-      return pLeftBlock->info.groupId < pRightBlock->info.groupId ? -1 : 1;
+    if (pLeftBlock->info.id.groupId != pRightBlock->info.id.groupId) {
+      return pLeftBlock->info.id.groupId < pRightBlock->info.id.groupId ? -1 : 1;
     }
   }
 
@@ -826,7 +826,7 @@ void* tsortGetValue(STupleHandle* pVHandle, int32_t colIndex) {
   }
 }
 
-uint64_t tsortGetGroupId(STupleHandle* pVHandle) { return pVHandle->pBlock->info.groupId; }
+uint64_t tsortGetGroupId(STupleHandle* pVHandle) { return pVHandle->pBlock->info.id.groupId; }
 
 SSortExecInfo tsortGetSortExecInfo(SSortHandle* pHandle) {
   SSortExecInfo info = {0};

source/libs/function/src/tudf.c

@@ -1096,7 +1096,7 @@ int32_t udfAggProcess(struct SqlFunctionCtx *pCtx) {
 
   SSDataBlock *pTempBlock = createDataBlock();
   pTempBlock->info.rows = pInput->totalRows;
-  pTempBlock->info.uid = pInput->uid;
+  pTempBlock->info.id.uid = pInput->uid;
   for (int32_t i = 0; i < numOfCols; ++i) {
     blockDataAppendColInfo(pTempBlock, pInput->pData[i]);
   }

source/libs/index/test/index_executor_tests.cpp

@@ -78,7 +78,7 @@ void sifAppendReservedSlot(SArray *pBlockList, int16_t *dataBlockId, int16_t *sl
     blockDataEnsureCapacity(res, rows);
 
     *dataBlockId = taosArrayGetSize(pBlockList) - 1;
-    res->info.blockId = *dataBlockId;
+    res->info.id.blockId = *dataBlockId;
     *slotId = 0;
   } else {
     SSDataBlock *res = *(SSDataBlock **)taosArrayGetLast(pBlockList);

source/libs/parser/inc/sql.y

@@ -654,7 +654,7 @@ cgroup_name(A) ::= NK_ID(B).
 
 /************************************************ expression **********************************************************/
 expr_or_subquery(A) ::= expression(B).                                            { A = B; }
-expr_or_subquery(A) ::= subquery(B).                                              { A = B; }
+//expr_or_subquery(A) ::= subquery(B).                                              { A = createTempTableNode(pCxt, releaseRawExprNode(pCxt, B), NULL); }
 
 expression(A) ::= literal(B).                                                     { A = B; }
 expression(A) ::= pseudo_column(B).                                               { A = B; }

source/libs/parser/src/parCalcConst.c

@@ -372,14 +372,42 @@ static bool isEmptyResultQuery(SNode* pStmt) {
   return isEmptyResult;
 }
 
+static void resetProjectNullTypeImpl(SNodeList* pProjects) {
+  SNode* pProj = NULL;
+  FOREACH(pProj, pProjects) {
+    SExprNode* pExpr = (SExprNode*)pProj;
+    if (TSDB_DATA_TYPE_NULL == pExpr->resType.type) {
+      pExpr->resType.type = TSDB_DATA_TYPE_VARCHAR;
+      pExpr->resType.bytes = 0;
+    }
+  }
+}
+
+static void resetProjectNullType(SNode* pStmt) {
+  switch (nodeType(pStmt)) {
+    case QUERY_NODE_SELECT_STMT:
+      resetProjectNullTypeImpl(((SSelectStmt*)pStmt)->pProjectionList);
+      break;
+    case QUERY_NODE_SET_OPERATOR: {
+      resetProjectNullTypeImpl(((SSetOperator*)pStmt)->pProjectionList);
+      break;
+    }
+    default:
+      break;
+  }
+}
+
 int32_t calculateConstant(SParseContext* pParseCxt, SQuery* pQuery) {
   SCalcConstContext cxt = {.pParseCxt = pParseCxt,
                            .msgBuf.buf = pParseCxt->pMsg,
                            .msgBuf.len = pParseCxt->msgLen,
                            .code = TSDB_CODE_SUCCESS};
   int32_t           code = calcConstQuery(&cxt, pQuery->pRoot, false);
-  if (TSDB_CODE_SUCCESS == code && isEmptyResultQuery(pQuery->pRoot)) {
+  if (TSDB_CODE_SUCCESS == code) {
+    resetProjectNullType(pQuery->pRoot);
+    if (isEmptyResultQuery(pQuery->pRoot)) {
       pQuery->execMode = QUERY_EXEC_MODE_EMPTY_RESULT;
     }
+  }
   return code;
 }