Merge pull request #21116 from taosdata/fix/liaohj_main

fix(query): stop tsdbreade ASAP, and do some internal refactor. TD-23249

source/libs/executor/inc/executil.h

@@ -39,8 +39,6 @@
 
 #define GET_RES_WINDOW_KEY_LEN(_l)     ((_l) + sizeof(uint64_t))
 
-#define GET_TASKID(_t) (((SExecTaskInfo*)(_t))->id.str)
-
 typedef struct SGroupResInfo {
   int32_t index;
   SArray* pRows;  // SArray<SResKeyPos>

source/libs/executor/inc/executorimpl.h

@@ -85,39 +85,6 @@ typedef struct SLimit {
 
 typedef struct STableScanAnalyzeInfo SFileBlockLoadRecorder;
 
-typedef struct STaskCostInfo {
-  int64_t                 created;
-  int64_t                 start;
-  uint64_t                elapsedTime;
-  double                  extractListTime;
-  double                  groupIdMapTime;
-  SFileBlockLoadRecorder* pRecoder;
-} STaskCostInfo;
-
-typedef struct SOperatorCostInfo {
-  double openCost;
-  double totalCost;
-} SOperatorCostInfo;
-
-struct SOperatorInfo;
-
-typedef int32_t (*__optr_encode_fn_t)(struct SOperatorInfo* pOperator, char** result, int32_t* length);
-typedef int32_t (*__optr_decode_fn_t)(struct SOperatorInfo* pOperator, char* result);
-
-typedef int32_t (*__optr_open_fn_t)(struct SOperatorInfo* pOptr);
-typedef SSDataBlock* (*__optr_fn_t)(struct SOperatorInfo* pOptr);
-typedef void (*__optr_close_fn_t)(void* param);
-typedef int32_t (*__optr_explain_fn_t)(struct SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len);
-typedef int32_t (*__optr_reqBuf_fn_t)(struct SOperatorInfo* pOptr);
-
-typedef struct STaskIdInfo {
-  uint64_t queryId;  // this is also a request id
-  uint64_t subplanId;
-  uint64_t templateId;
-  char*    str;
-  int32_t  vgId;
-} STaskIdInfo;
-
 enum {
   STREAM_RECOVER_STEP__NONE = 0,
   STREAM_RECOVER_STEP__PREPARE1,
@@ -156,51 +123,6 @@ typedef struct SExchangeOpStopInfo {
   int64_t refId;
 } SExchangeOpStopInfo;
 
-typedef struct STaskStopInfo {
-  SRWLatch lock;
-  SArray*  pStopInfo;
-} STaskStopInfo;
-
-struct SExecTaskInfo {
-  STaskIdInfo           id;
-  uint32_t              status;
-  STimeWindow           window;
-  STaskCostInfo         cost;
-  int64_t               owner;  // if it is in execution
-  int32_t               code;
-  int32_t               qbufQuota;  // total available buffer (in KB) during execution query
-  int64_t               version;    // used for stream to record wal version, why not move to sschemainfo
-  SStreamTaskInfo       streamInfo;
-  SSchemaInfo           schemaInfo;
-  const char*           sql;        // query sql string
-  jmp_buf               env;        // jump to this position when error happens.
-  EOPTR_EXEC_MODEL      execModel;  // operator execution model [batch model|stream model]
-  SSubplan*             pSubplan;
-  struct SOperatorInfo* pRoot;
-  SLocalFetch           localFetch;
-  SArray*               pResultBlockList;  // result block list
-  STaskStopInfo         stopInfo;
-  SRWLatch              lock;  // secure the access of STableListInfo
-};
-
-enum {
-  OP_NOT_OPENED = 0x0,
-  OP_OPENED = 0x1,
-  OP_RES_TO_RETURN = 0x5,
-  OP_EXEC_DONE = 0x9,
-};
-
-typedef struct SOperatorFpSet {
-  __optr_open_fn_t    _openFn;  // DO NOT invoke this function directly
-  __optr_fn_t         getNextFn;
-  __optr_fn_t         cleanupFn;  // call this function to release the allocated resources ASAP
-  __optr_close_fn_t   closeFn;
-  __optr_reqBuf_fn_t  reqBufFn;  // total used buffer for blocking operator
-  __optr_encode_fn_t  encodeResultRow;
-  __optr_decode_fn_t  decodeResultRow;
-  __optr_explain_fn_t getExplainFn;
-} SOperatorFpSet;
-
 typedef struct SExprSupp {
   SExprInfo*      pExprInfo;
   int32_t         numOfExprs;  // the number of scalar expression in group operator
@@ -209,22 +131,6 @@ typedef struct SExprSupp {
   SFilterInfo*    pFilterInfo;
 } SExprSupp;
 
-typedef struct SOperatorInfo {
-  uint16_t               operatorType;
-  int16_t                resultDataBlockId;
-  bool                   blocking;  // block operator or not
-  uint8_t                status;    // denote if current operator is completed
-  char*                  name;      // name, for debug purpose
-  void*                  info;      // extension attribution
-  SExprSupp              exprSupp;
-  SExecTaskInfo*         pTaskInfo;
-  SOperatorCostInfo      cost;
-  SResultInfo            resultInfo;
-  struct SOperatorInfo** pDownstream;      // downstram pointer list
-  int32_t                numOfDownstream;  // number of downstream. The value is always ONE expect for join operator
-  SOperatorFpSet         fpSet;
-} SOperatorInfo;
-
 typedef enum {
   EX_SOURCE_DATA_NOT_READY = 0x1,
   EX_SOURCE_DATA_READY = 0x2,
@@ -449,8 +355,8 @@ typedef struct SStreamScanInfo {
   SUpdateInfo* pUpdateInfo;
 
   EStreamScanMode       scanMode;
-  SOperatorInfo*        pStreamScanOp;
-  SOperatorInfo*        pTableScanOp;
+  struct SOperatorInfo*        pStreamScanOp;
+  struct SOperatorInfo*        pTableScanOp;
   SArray*               childIds;
   SWindowSupporter      windowSup;
   SPartitionBySupporter partitionSup;
@@ -676,18 +582,8 @@ typedef struct SStreamFillOperatorInfo {
 #define OPTR_IS_OPENED(_optr)  (((_optr)->status & OP_OPENED) == OP_OPENED)
 #define OPTR_SET_OPENED(_optr) ((_optr)->status |= OP_OPENED)
 
-SExecTaskInfo* doCreateExecTaskInfo(uint64_t queryId, uint64_t taskId, int32_t vgId, EOPTR_EXEC_MODEL model,
-                                    char* dbFName);
-
-SOperatorFpSet createOperatorFpSet(__optr_open_fn_t openFn, __optr_fn_t nextFn, __optr_fn_t cleanup,
-                                   __optr_close_fn_t closeFn, __optr_reqBuf_fn_t reqBufFn, __optr_explain_fn_t explain);
-int32_t        optrDummyOpenFn(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);
-int32_t        optrDefaultBufFn(SOperatorInfo* pOperator);
+SSchemaWrapper* extractQueriedColumnSchema(SScanPhysiNode* pScanNode);
+int32_t initQueriedTableSchemaInfo(SReadHandle* pHandle, SScanPhysiNode* pScanNode, const char* dbName, SExecTaskInfo* pTaskInfo);
 
 void initBasicInfo(SOptrBasicInfo* pInfo, SSDataBlock* pBlock);
 void cleanupBasicInfo(SOptrBasicInfo* pInfo);
@@ -703,9 +599,9 @@ void    cleanupAggSup(SAggSupporter* pAggSup);
 
 void initResultSizeInfo(SResultInfo* pResultInfo, int32_t numOfRows);
 
-void doBuildStreamResBlock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
+void doBuildStreamResBlock(struct SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
                            SDiskbasedBuf* pBuf);
-void doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
+void doBuildResultDatablock(struct SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
                             SDiskbasedBuf* pBuf);
 
 bool hasLimitOffsetInfo(SLimitInfo* pLimitInfo);
@@ -719,12 +615,10 @@ void applyAggFunctionOnPartialTuples(SExecTaskInfo* taskInfo, SqlFunctionCtx* pC
 
 int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, char* pData, SArray* pColList, char** pNextStart);
 void    updateLoadRemoteInfo(SLoadRemoteDataInfo* pInfo, int64_t numOfRows, int32_t dataLen, int64_t startTs,
-                             SOperatorInfo* pOperator);
+                             struct SOperatorInfo* pOperator);
 
 STimeWindow getFirstQualifiedTimeWindow(int64_t ts, STimeWindow* pWindow, SInterval* pInterval, int32_t order);
 
-SOperatorInfo* extractOperatorInTree(SOperatorInfo* pOperator, int32_t type, const char* id);
-int32_t        getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scanFlag, bool inheritUsOrder);
 int32_t        getBufferPgSize(int32_t rowSize, uint32_t* defaultPgsz, uint32_t* defaultBufsz);
 
 extern void doDestroyExchangeOperatorInfo(void* param);
@@ -742,76 +636,6 @@ void clearResultRowInitFlag(SqlFunctionCtx* pCtx, int32_t numOfOutput);
 SResultRow* doSetResultOutBufByKey(SDiskbasedBuf* pResultBuf, SResultRowInfo* pResultRowInfo, char* pData,
                                    int16_t bytes, bool masterscan, uint64_t groupId, SExecTaskInfo* pTaskInfo,
                                    bool isIntervalQuery, SAggSupporter* pSup, bool keepGroup);
-// operator creater functions
-// clang-format off
-SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode* pExNode, SExecTaskInfo* pTaskInfo);
-
-SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle, STableListInfo* pTableList, SExecTaskInfo* pTaskInfo);
-
-SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle, STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo);
-
-SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pPhyNode, STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo);
-
-SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScanPhysiNode* pScanPhyNode, const char* pUser, SExecTaskInfo* pTaskInfo);
-
-SOperatorInfo* createTableCountScanOperatorInfo(SReadHandle* handle, STableCountScanPhysiNode* pNode, 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* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pSortNode, SExecTaskInfo* pTaskInfo);
-
-SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** dowStreams, size_t numStreams, SMergePhysiNode* pMergePhysiNode, SExecTaskInfo* pTaskInfo);
-
-SOperatorInfo* createCacherowsScanOperator(SLastRowScanPhysiNode* pTableScanNode, SReadHandle* readHandle, STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo);
-
-SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SIntervalPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo);
-
-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, STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo);
-
-SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode, SNode* pTagCond, STableListInfo* pTableListInfo, 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* 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* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo);
-
-SOperatorInfo* createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFillPhysiNode* pPhyFillNode, SExecTaskInfo* pTaskInfo);
-
-SOperatorInfo* createGroupSortOperatorInfo(SOperatorInfo* downstream, SGroupSortPhysiNode* pSortPhyNode, SExecTaskInfo* pTaskInfo);
-
-SOperatorInfo* createEventwindowOperatorInfo(SOperatorInfo* downstream, SPhysiNode* physiNode, SExecTaskInfo* pTaskInfo);
-// clang-format on
 
 int32_t projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBlock* pSrcBlock, SqlFunctionCtx* pCtx,
                               int32_t numOfOutput, SArray* pPseudoList);
@@ -820,19 +644,10 @@ void setInputDataBlock(SExprSupp* pExprSupp, SSDataBlock* pBlock, int32_t order,
 
 int32_t checkForQueryBuf(size_t numOfTables);
 
-bool isTaskKilled(SExecTaskInfo* pTaskInfo);
-void setTaskKilled(SExecTaskInfo* pTaskInfo, int32_t rspCode);
-void doDestroyTask(SExecTaskInfo* pTaskInfo);
-void setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status);
-
-void buildTaskId(uint64_t taskId, uint64_t queryId, char* dst);
-
 SArray* getTableListInfo(const SExecTaskInfo* pTaskInfo);
 
-int32_t createExecTaskInfo(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId,
-                           int32_t vgId, char* sql, EOPTR_EXEC_MODEL model);
 int32_t createDataSinkParam(SDataSinkNode* pNode, void** pParam, SExecTaskInfo* pTask, SReadHandle* readHandle);
-int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SArray* pExecInfoList);
+int32_t getOperatorExplainExecInfo(struct SOperatorInfo* operatorInfo, SArray* pExecInfoList);
 
 STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts, SInterval* pInterval,
                                 int32_t order);
@@ -854,17 +669,16 @@ int32_t finalizeResultRows(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPos
                            SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
 
 bool    groupbyTbname(SNodeList* pGroupList);
-int32_t buildDataBlockFromGroupRes(SOperatorInfo* pOperator, SStreamState* pState, SSDataBlock* pBlock, SExprSupp* pSup,
+int32_t buildDataBlockFromGroupRes(struct SOperatorInfo* pOperator, SStreamState* pState, SSDataBlock* pBlock, SExprSupp* pSup,
                                    SGroupResInfo* pGroupResInfo);
 int32_t saveSessionDiscBuf(SStreamState* pState, SSessionKey* key, void* buf, int32_t size);
-int32_t buildSessionResultDataBlock(SOperatorInfo* pOperator, SStreamState* pState, SSDataBlock* pBlock,
+int32_t buildSessionResultDataBlock(struct SOperatorInfo* pOperator, SStreamState* pState, SSDataBlock* pBlock,
                                     SExprSupp* pSup, SGroupResInfo* pGroupResInfo);
 int32_t setOutputBuf(SStreamState* pState, STimeWindow* win, SResultRow** pResult, int64_t tableGroupId,
                      SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup);
 int32_t releaseOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult);
 int32_t saveOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult, int32_t resSize);
 void    getNextIntervalWindow(SInterval* pInterval, STimeWindow* tw, int32_t order);
-int32_t qAppendTaskStopInfo(SExecTaskInfo* pTaskInfo, SExchangeOpStopInfo* pInfo);
 int32_t getForwardStepsInBlock(int32_t numOfRows, __block_search_fn_t searchFn, TSKEY ekey, int32_t pos, int32_t order,
                                int64_t* pData);
 void    appendCreateTableRow(SStreamState* pState, SExprSupp* pTableSup, SExprSupp* pTagSup, uint64_t groupId,

source/libs/executor/inc/operator.h

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef TDENGINE_OPERATOR_H
+#define TDENGINE_OPERATOR_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef struct SOperatorCostInfo {
+  double openCost;
+  double totalCost;
+} SOperatorCostInfo;
+
+struct SOperatorInfo;
+
+typedef int32_t (*__optr_encode_fn_t)(struct SOperatorInfo* pOperator, char** result, int32_t* length);
+typedef int32_t (*__optr_decode_fn_t)(struct SOperatorInfo* pOperator, char* result);
+
+typedef int32_t (*__optr_open_fn_t)(struct SOperatorInfo* pOptr);
+typedef SSDataBlock* (*__optr_fn_t)(struct SOperatorInfo* pOptr);
+typedef void (*__optr_close_fn_t)(void* param);
+typedef int32_t (*__optr_explain_fn_t)(struct SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len);
+typedef int32_t (*__optr_reqBuf_fn_t)(struct SOperatorInfo* pOptr);
+
+typedef struct SOperatorFpSet {
+  __optr_open_fn_t    _openFn;  // DO NOT invoke this function directly
+  __optr_fn_t         getNextFn;
+  __optr_fn_t         cleanupFn;  // call this function to release the allocated resources ASAP
+  __optr_close_fn_t   closeFn;
+  __optr_reqBuf_fn_t  reqBufFn;  // total used buffer for blocking operator
+  __optr_encode_fn_t  encodeResultRow;
+  __optr_decode_fn_t  decodeResultRow;
+  __optr_explain_fn_t getExplainFn;
+} SOperatorFpSet;
+
+enum {
+  OP_NOT_OPENED = 0x0,
+  OP_OPENED = 0x1,
+  OP_RES_TO_RETURN = 0x5,
+  OP_EXEC_DONE = 0x9,
+};
+
+typedef struct SOperatorInfo {
+  uint16_t               operatorType;
+  int16_t                resultDataBlockId;
+  bool                   blocking;  // block operator or not
+  uint8_t                status;    // denote if current operator is completed
+  char*                  name;      // name, for debug purpose
+  void*                  info;      // extension attribution
+  SExprSupp              exprSupp;
+  SExecTaskInfo*         pTaskInfo;
+  SOperatorCostInfo      cost;
+  SResultInfo            resultInfo;
+  struct SOperatorInfo** pDownstream;      // downstram pointer list
+  int32_t                numOfDownstream;  // number of downstream. The value is always ONE expect for join operator
+  SOperatorFpSet         fpSet;
+} SOperatorInfo;
+
+// operator creater functions
+// clang-format off
+SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode* pExNode, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle, STableListInfo* pTableList, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle, STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pPhyNode, STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScanPhysiNode* pScanPhyNode, const char* pUser, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createTableCountScanOperatorInfo(SReadHandle* handle, STableCountScanPhysiNode* pNode, 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* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pSortNode, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** dowStreams, size_t numStreams, SMergePhysiNode* pMergePhysiNode, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createCacherowsScanOperator(SLastRowScanPhysiNode* pTableScanNode, SReadHandle* readHandle, STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SIntervalPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo);
+
+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, STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode, SNode* pTagCond, STableListInfo* pTableListInfo, 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* 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* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFillPhysiNode* pPhyFillNode, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createGroupSortOperatorInfo(SOperatorInfo* downstream, SGroupSortPhysiNode* pSortPhyNode, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createEventwindowOperatorInfo(SOperatorInfo* downstream, SPhysiNode* physiNode, SExecTaskInfo* pTaskInfo);
+// clang-format on
+
+SOperatorFpSet createOperatorFpSet(__optr_open_fn_t openFn, __optr_fn_t nextFn, __optr_fn_t cleanup,
+                                   __optr_close_fn_t closeFn, __optr_reqBuf_fn_t reqBufFn, __optr_explain_fn_t explain);
+int32_t        optrDummyOpenFn(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);
+int32_t        optrDefaultBufFn(SOperatorInfo* pOperator);
+
+SOperatorInfo* createOperator(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHandle* pHandle, SNode* pTagCond,
+                              SNode* pTagIndexCond, const char* pUser, const char* dbname);
+void           destroyOperator(SOperatorInfo* pOperator);
+
+SOperatorInfo* extractOperatorInTree(SOperatorInfo* pOperator, int32_t type, const char* id);
+int32_t        getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scanFlag, bool inheritUsOrder);
+int32_t        stopTableScanOperator(SOperatorInfo* pOperator, const char* pIdStr);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif  // TDENGINE_OPERATOR_H
\ No newline at end of file

source/libs/executor/inc/querytask.h

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef TDENGINE_QUERYTASK_H
+#define TDENGINE_QUERYTASK_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define GET_TASKID(_t) (((SExecTaskInfo*)(_t))->id.str)
+
+typedef struct STaskIdInfo {
+  uint64_t queryId;  // this is also a request id
+  uint64_t subplanId;
+  uint64_t templateId;
+  char*    str;
+  int32_t  vgId;
+} STaskIdInfo;
+
+typedef struct STaskCostInfo {
+  int64_t                 created;
+  int64_t                 start;
+  uint64_t                elapsedTime;
+  double                  extractListTime;
+  double                  groupIdMapTime;
+  SFileBlockLoadRecorder* pRecoder;
+} STaskCostInfo;
+
+typedef struct STaskStopInfo {
+  SRWLatch lock;
+  SArray*  pStopInfo;
+} STaskStopInfo;
+
+struct SExecTaskInfo {
+  STaskIdInfo           id;
+  uint32_t              status;
+  STimeWindow           window;
+  STaskCostInfo         cost;
+  int64_t               owner;  // if it is in execution
+  int32_t               code;
+  int32_t               qbufQuota;  // total available buffer (in KB) during execution query
+  int64_t               version;    // used for stream to record wal version, why not move to sschemainfo
+  SStreamTaskInfo       streamInfo;
+  SSchemaInfo           schemaInfo;
+  const char*           sql;        // query sql string
+  jmp_buf               env;        // jump to this position when error happens.
+  EOPTR_EXEC_MODEL      execModel;  // operator execution model [batch model|stream model]
+  SSubplan*             pSubplan;
+  struct SOperatorInfo* pRoot;
+  SLocalFetch           localFetch;
+  SArray*               pResultBlockList;  // result block list
+  STaskStopInfo         stopInfo;
+  SRWLatch              lock;  // secure the access of STableListInfo
+};
+
+void           buildTaskId(uint64_t taskId, uint64_t queryId, char* dst);
+SExecTaskInfo* doCreateTask(uint64_t queryId, uint64_t taskId, int32_t vgId, EOPTR_EXEC_MODEL model);
+void           doDestroyTask(SExecTaskInfo* pTaskInfo);
+bool           isTaskKilled(SExecTaskInfo* pTaskInfo);
+void           setTaskKilled(SExecTaskInfo* pTaskInfo, int32_t rspCode);
+void           setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status);
+int32_t        createExecTaskInfo(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId,
+                                  int32_t vgId, char* sql, EOPTR_EXEC_MODEL model);
+int32_t        qAppendTaskStopInfo(SExecTaskInfo* pTaskInfo, SExchangeOpStopInfo* pInfo);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif  // TDENGINE_QUERYTASK_H

source/libs/executor/src/aggregateoperator.c

@@ -15,7 +15,6 @@
 
 #include "filter.h"
 #include "function.h"
-#include "functionMgt.h"
 #include "os.h"
 #include "querynodes.h"
 #include "tfill.h"
@@ -23,16 +22,14 @@
 
 #include "tdatablock.h"
 #include "tglobal.h"
-#include "tmsg.h"
-#include "ttime.h"
-
 #include "executorimpl.h"
 #include "index.h"
 #include "query.h"
 #include "tcompare.h"
 #include "thash.h"
 #include "ttypes.h"
-#include "vnode.h"
+#include "operator.h"
+#include "querytask.h"
 
 typedef struct {
   bool    hasAgg;

source/libs/executor/src/cachescanoperator.c

@@ -24,6 +24,8 @@
 #include "tcompare.h"
 #include "thash.h"
 #include "ttypes.h"
+#include "operator.h"
+#include "querytask.h"
 
 typedef struct SCacheRowsScanInfo {
   SSDataBlock*    pRes;

source/libs/executor/src/eventwindowoperator.c

@@ -21,6 +21,8 @@
 #include "tcompare.h"
 #include "tdatablock.h"
 #include "ttime.h"
+#include "operator.h"
+#include "querytask.h"
 
 typedef struct SEventWindowOperatorInfo {
   SOptrBasicInfo     binfo;

source/libs/executor/src/exchangeoperator.c

@@ -24,6 +24,8 @@
 #include "index.h"
 #include "query.h"
 #include "thash.h"
+#include "operator.h"
+#include "querytask.h"
 
 typedef struct SFetchRspHandleWrapper {
   uint32_t exchangeId;

source/libs/executor/src/executil.c

@@ -26,6 +26,7 @@
 #include "executil.h"
 #include "executorimpl.h"
 #include "tcompression.h"
+#include "querytask.h"
 
 typedef struct STableListIdInfo {
   uint64_t suid;

source/libs/executor/src/executor.c

@@ -14,13 +14,14 @@
  */
 
 #include "executor.h"
-#include <vnode.h>
 #include "executorimpl.h"
 #include "planner.h"
 #include "tdatablock.h"
 #include "tref.h"
 #include "tudf.h"
 #include "vnode.h"
+#include "operator.h"
+#include "querytask.h"
 
 static TdThreadOnce initPoolOnce = PTHREAD_ONCE_INIT;
 int32_t             exchangeObjRefPool = -1;
@@ -249,7 +250,7 @@ int32_t qSetSMAInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numOfBlocks,
 qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* pReaderHandle, int32_t vgId, int32_t* numOfCols,
                                      uint64_t id) {
   if (msg == NULL) {  // create raw scan
-    SExecTaskInfo* pTaskInfo = doCreateExecTaskInfo(0, id, vgId, OPTR_EXEC_MODEL_QUEUE, "");
+    SExecTaskInfo* pTaskInfo = doCreateTask(0, id, vgId, OPTR_EXEC_MODEL_QUEUE);
     if (NULL == pTaskInfo) {
       terrno = TSDB_CODE_OUT_OF_MEMORY;
       return NULL;

source/libs/executor/src/filloperator.c

@@ -25,11 +25,13 @@
 #include "thash.h"
 #include "ttime.h"
 
-#include "executorInt.h"
 #include "function.h"
 #include "querynodes.h"
 #include "tdatablock.h"
 #include "tfill.h"
+#include "operator.h"
+#include "querytask.h"
+
 
 #define FILL_POS_INVALID 0
 #define FILL_POS_START   1

source/libs/executor/src/groupoperator.c

@@ -26,6 +26,8 @@
 #include "tcompare.h"
 #include "thash.h"
 #include "ttypes.h"
+#include "operator.h"
+#include "querytask.h"
 
 typedef struct SGroupbyOperatorInfo {
   SOptrBasicInfo binfo;

source/libs/executor/src/joinoperator.c

@@ -23,6 +23,8 @@
 #include "thash.h"
 #include "tmsg.h"
 #include "ttypes.h"
+#include "operator.h"
+#include "querytask.h"
 
 typedef struct SJoinRowCtx {
   bool    rowRemains;

source/libs/executor/src/projectoperator.c

@@ -16,6 +16,8 @@
 #include "executorimpl.h"
 #include "filter.h"
 #include "functionMgt.h"
+#include "operator.h"
+#include "querytask.h"
 
 typedef struct SProjectOperatorInfo {
   SOptrBasicInfo binfo;

source/libs/executor/src/querytask.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "filter.h"
+#include "function.h"
+#include "functionMgt.h"
+#include "os.h"
+#include "querynodes.h"
+#include "tfill.h"
+#include "tname.h"
+
+#include "tdatablock.h"
+#include "tmsg.h"
+
+#include "executorimpl.h"
+#include "index.h"
+#include "query.h"
+#include "thash.h"
+#include "ttypes.h"
+#include "vnode.h"
+#include "operator.h"
+#include "querytask.h"
+
+#define CLEAR_QUERY_STATUS(q, st) ((q)->status &= (~(st)))
+
+SExecTaskInfo* doCreateTask(uint64_t queryId, uint64_t taskId, int32_t vgId, EOPTR_EXEC_MODEL model) {
+  SExecTaskInfo* pTaskInfo = taosMemoryCalloc(1, sizeof(SExecTaskInfo));
+  if (pTaskInfo == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return NULL;
+  }
+
+  setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);
+  pTaskInfo->cost.created = taosGetTimestampUs();
+
+  pTaskInfo->execModel = model;
+  pTaskInfo->stopInfo.pStopInfo = taosArrayInit(4, sizeof(SExchangeOpStopInfo));
+  pTaskInfo->pResultBlockList = taosArrayInit(128, POINTER_BYTES);
+
+  taosInitRWLatch(&pTaskInfo->lock);
+
+  pTaskInfo->id.vgId = vgId;
+  pTaskInfo->id.queryId = queryId;
+  pTaskInfo->id.str = taosMemoryMalloc(64);
+  buildTaskId(taskId, queryId, pTaskInfo->id.str);
+
+  return pTaskInfo;
+}
+
+bool isTaskKilled(SExecTaskInfo* pTaskInfo) { return (0 != pTaskInfo->code); }
+
+void setTaskKilled(SExecTaskInfo* pTaskInfo, int32_t rspCode) {
+  pTaskInfo->code = rspCode;
+  stopTableScanOperator(pTaskInfo->pRoot, pTaskInfo->id.str);
+}
+
+void setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status) {
+  if (status == TASK_NOT_COMPLETED) {
+    pTaskInfo->status = status;
+  } else {
+    // QUERY_NOT_COMPLETED is not compatible with any other status, so clear its position first
+    CLEAR_QUERY_STATUS(pTaskInfo, TASK_NOT_COMPLETED);
+    pTaskInfo->status |= status;
+  }
+}
+
+int32_t createExecTaskInfo(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId,
+                           int32_t vgId, char* sql, EOPTR_EXEC_MODEL model) {
+  *pTaskInfo = doCreateTask(pPlan->id.queryId, taskId, vgId, model);
+  if (*pTaskInfo == NULL) {
+    taosMemoryFree(sql);
+    return terrno;
+  }
+
+  if (pHandle) {
+    if (pHandle->pStateBackend) {
+      (*pTaskInfo)->streamInfo.pState = pHandle->pStateBackend;
+    }
+  }
+
+  TSWAP((*pTaskInfo)->sql, sql);
+
+  (*pTaskInfo)->pSubplan = pPlan;
+  (*pTaskInfo)->pRoot = createOperator(pPlan->pNode, *pTaskInfo, pHandle, pPlan->pTagCond, pPlan->pTagIndexCond,
+                                       pPlan->user, pPlan->dbFName);
+
+  if (NULL == (*pTaskInfo)->pRoot) {
+    int32_t code = (*pTaskInfo)->code;
+    doDestroyTask(*pTaskInfo);
+    return code;
+  } else {
+    return TSDB_CODE_SUCCESS;
+  }
+}
+
+void cleanupQueriedTableScanInfo(SSchemaInfo* pSchemaInfo) {
+  taosMemoryFreeClear(pSchemaInfo->dbname);
+  taosMemoryFreeClear(pSchemaInfo->tablename);
+  tDeleteSSchemaWrapper(pSchemaInfo->sw);
+  tDeleteSSchemaWrapper(pSchemaInfo->qsw);
+}
+
+int32_t initQueriedTableSchemaInfo(SReadHandle* pHandle, SScanPhysiNode* pScanNode, const char* dbName, SExecTaskInfo* pTaskInfo) {
+  SMetaReader mr = {0};
+  if (pHandle == NULL) {
+    terrno = TSDB_CODE_INVALID_PARA;
+    return terrno;
+  }
+
+  metaReaderInit(&mr, pHandle->meta, 0);
+  int32_t code = metaGetTableEntryByUidCache(&mr, pScanNode->uid);
+  if (code != TSDB_CODE_SUCCESS) {
+    qError("failed to get the table meta, uid:0x%" PRIx64 ", suid:0x%" PRIx64 ", %s", pScanNode->uid, pScanNode->suid,
+           GET_TASKID(pTaskInfo));
+
+    metaReaderClear(&mr);
+    return terrno;
+  }
+
+  SSchemaInfo* pSchemaInfo = &pTaskInfo->schemaInfo;
+
+  pSchemaInfo->tablename = taosStrdup(mr.me.name);
+  pSchemaInfo->dbname = taosStrdup(dbName);
+
+  if (mr.me.type == TSDB_SUPER_TABLE) {
+    pSchemaInfo->sw = tCloneSSchemaWrapper(&mr.me.stbEntry.schemaRow);
+    pSchemaInfo->tversion = mr.me.stbEntry.schemaTag.version;
+  } else if (mr.me.type == TSDB_CHILD_TABLE) {
+    tDecoderClear(&mr.coder);
+
+    tb_uid_t suid = mr.me.ctbEntry.suid;
+    code = metaGetTableEntryByUidCache(&mr, suid);
+    if (code != TSDB_CODE_SUCCESS) {
+      metaReaderClear(&mr);
+      return terrno;
+    }
+
+    pSchemaInfo->sw = tCloneSSchemaWrapper(&mr.me.stbEntry.schemaRow);
+    pSchemaInfo->tversion = mr.me.stbEntry.schemaTag.version;
+  } else {
+    pSchemaInfo->sw = tCloneSSchemaWrapper(&mr.me.ntbEntry.schemaRow);
+  }
+
+  metaReaderClear(&mr);
+
+  pSchemaInfo->qsw = extractQueriedColumnSchema(pScanNode);
+  return TSDB_CODE_SUCCESS;
+}
+
+SSchemaWrapper* extractQueriedColumnSchema(SScanPhysiNode* pScanNode) {
+  int32_t numOfCols = LIST_LENGTH(pScanNode->pScanCols);
+  int32_t numOfTags = LIST_LENGTH(pScanNode->pScanPseudoCols);
+
+  SSchemaWrapper* pqSw = taosMemoryCalloc(1, sizeof(SSchemaWrapper));
+  pqSw->pSchema = taosMemoryCalloc(numOfCols + numOfTags, sizeof(SSchema));
+
+  for (int32_t i = 0; i < numOfCols; ++i) {
+    STargetNode* pNode = (STargetNode*)nodesListGetNode(pScanNode->pScanCols, i);
+    SColumnNode* pColNode = (SColumnNode*)pNode->pExpr;
+
+    SSchema* pSchema = &pqSw->pSchema[pqSw->nCols++];
+    pSchema->colId = pColNode->colId;
+    pSchema->type = pColNode->node.resType.type;
+    pSchema->bytes = pColNode->node.resType.bytes;
+    tstrncpy(pSchema->name, pColNode->colName, tListLen(pSchema->name));
+  }
+
+  // this the tags and pseudo function columns, we only keep the tag columns
+  for (int32_t i = 0; i < numOfTags; ++i) {
+    STargetNode* pNode = (STargetNode*)nodesListGetNode(pScanNode->pScanPseudoCols, i);
+
+    int32_t type = nodeType(pNode->pExpr);
+    if (type == QUERY_NODE_COLUMN) {
+      SColumnNode* pColNode = (SColumnNode*)pNode->pExpr;
+
+      SSchema* pSchema = &pqSw->pSchema[pqSw->nCols++];
+      pSchema->colId = pColNode->colId;
+      pSchema->type = pColNode->node.resType.type;
+      pSchema->bytes = pColNode->node.resType.bytes;
+      tstrncpy(pSchema->name, pColNode->colName, tListLen(pSchema->name));
+    }
+  }
+
+  return pqSw;
+}
+
+static void cleanupStreamInfo(SStreamTaskInfo* pStreamInfo) { tDeleteSSchemaWrapper(pStreamInfo->schema); }
+
+static void freeBlock(void* pParam) {
+  SSDataBlock* pBlock = *(SSDataBlock**)pParam;
+  blockDataDestroy(pBlock);
+}
+
+void doDestroyTask(SExecTaskInfo* pTaskInfo) {
+  qDebug("%s execTask is freed", GET_TASKID(pTaskInfo));
+  destroyOperator(pTaskInfo->pRoot);
+  cleanupQueriedTableScanInfo(&pTaskInfo->schemaInfo);
+  cleanupStreamInfo(&pTaskInfo->streamInfo);
+
+  if (!pTaskInfo->localFetch.localExec) {
+    nodesDestroyNode((SNode*)pTaskInfo->pSubplan);
+  }
+
+  taosArrayDestroyEx(pTaskInfo->pResultBlockList, freeBlock);
+  taosArrayDestroy(pTaskInfo->stopInfo.pStopInfo);
+  taosMemoryFreeClear(pTaskInfo->sql);
+  taosMemoryFreeClear(pTaskInfo->id.str);
+  taosMemoryFreeClear(pTaskInfo);
+}
+
+void buildTaskId(uint64_t taskId, uint64_t queryId, char* dst) {
+  char* p = dst;
+
+  int32_t offset = 6;
+  memcpy(p, "TID:0x", offset);
+  offset += tintToHex(taskId, &p[offset]);
+
+  memcpy(&p[offset], " QID:0x", 7);
+  offset += 7;
+  offset += tintToHex(queryId, &p[offset]);
+
+  p[offset] = 0;
+}

source/libs/executor/src/scanoperator.c

@@ -30,6 +30,8 @@
 #include "tcompare.h"
 #include "thash.h"
 #include "ttypes.h"
+#include "operator.h"
+#include "querytask.h"
 
 int32_t scanDebug = 0;
 
@@ -2304,7 +2306,7 @@ static void destroyStreamScanOperatorInfo(void* param) {
   SStreamScanInfo* pStreamScan = (SStreamScanInfo*)param;
 
   if (pStreamScan->pTableScanOp && pStreamScan->pTableScanOp->info) {
-    destroyOperatorInfo(pStreamScan->pTableScanOp);
+    destroyOperator(pStreamScan->pTableScanOp);
   }
 
   if (pStreamScan->tqReader) {

source/libs/executor/src/sortoperator.c

@@ -16,6 +16,8 @@
 #include "filter.h"
 #include "executorimpl.h"
 #include "tdatablock.h"
+#include "operator.h"
+#include "querytask.h"
 
 typedef struct SSortOperatorInfo {
   SOptrBasicInfo binfo;

source/libs/executor/src/sysscanoperator.c

@@ -31,6 +31,9 @@
 #include "thash.h"
 #include "ttypes.h"
 #include "vnode.h"
+#include "operator.h"
+#include "querytask.h"
+
 
 typedef int (*__optSysFilter)(void* a, void* b, int16_t dtype);
 typedef int32_t (*__sys_filte)(void* pMeta, SNode* cond, SArray* result);

source/libs/executor/src/timesliceoperator.c

@@ -21,6 +21,8 @@
 #include "tdatablock.h"
 #include "tfill.h"
 #include "ttime.h"
+#include "operator.h"
+#include "querytask.h"
 
 typedef struct STimeSliceOperatorInfo {
   SSDataBlock*         pRes;

source/libs/executor/src/timewindowoperator.c

@@ -21,6 +21,8 @@
 #include "tdatablock.h"
 #include "tfill.h"
 #include "ttime.h"
+#include "operator.h"
+#include "querytask.h"
 
 #define IS_FINAL_OP(op)    ((op)->isFinal)
 #define DEAULT_DELETE_MARK (1000LL * 60LL * 60LL * 24LL * 365LL * 10LL);
@@ -1606,7 +1608,7 @@ void destroyStreamFinalIntervalOperatorInfo(void* param) {
     int32_t size = taosArrayGetSize(pInfo->pChildren);
     for (int32_t i = 0; i < size; i++) {
       SOperatorInfo* pChildOp = taosArrayGetP(pInfo->pChildren, i);
-      destroyOperatorInfo(pChildOp);
+      destroyOperator(pChildOp);
     }
     taosArrayDestroy(pInfo->pChildren);
   }
@@ -2835,7 +2837,7 @@ void destroyStreamSessionAggOperatorInfo(void* param) {
     int32_t size = taosArrayGetSize(pInfo->pChildren);
     for (int32_t i = 0; i < size; i++) {
       SOperatorInfo* pChild = taosArrayGetP(pInfo->pChildren, i);
-      destroyOperatorInfo(pChild);
+      destroyOperator(pChild);
     }
     taosArrayDestroy(pInfo->pChildren);
   }
@@ -3812,7 +3814,7 @@ void destroyStreamStateOperatorInfo(void* param) {
     int32_t size = taosArrayGetSize(pInfo->pChildren);
     for (int32_t i = 0; i < size; i++) {
       SOperatorInfo* pChild = taosArrayGetP(pInfo->pChildren, i);
-      destroyOperatorInfo(pChild);
+      destroyOperator(pChild);
     }
     taosArrayDestroy(pInfo->pChildren);
   }

source/libs/executor/test/executorTests.cpp

@@ -29,11 +29,8 @@
 #include "taos.h"
 #include "tdatablock.h"
 #include "tdef.h"
-#include "tglobal.h"
-#include "tmsg.h"
-#include "tname.h"
-#include "trpc.h"
 #include "tvariant.h"
+#include "operator.h"
 
 namespace {