Merge pull request #21119 from taosdata/fix/liaohj_main

refactor: do some internal refactor.

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

@@ -107,7 +107,7 @@ int32_t createStreamRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle) {
 
     if (streamTaskShouldStop(&pTask->status) || status == TASK_STATUS__RECOVER_PREPARE ||
         status == TASK_STATUS__WAIT_DOWNSTREAM) {
-      tqDebug("s-task:%s skip push data, not ready for processing, status %d", pTask->id.idStr, status);
+      tqDebug("s-task:%s skip push data, not ready for processing, status:%d", pTask->id.idStr, status);
       streamMetaReleaseTask(pStreamMeta, pTask);
       continue;
     }

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

@@ -130,6 +130,7 @@ int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMeta
       tqError("vgId:%d, task exec error since %s", pTq->pVnode->config.vgId, terrstr());
       return -1;
     }
+
     tqDebug("tmqsnap task execute end, get %p", pDataBlock);
 
     if (pDataBlock != NULL && pDataBlock->info.rows > 0) {

source/libs/executor/inc/operator.h

@@ -157,6 +157,7 @@ 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);
+int32_t        getOperatorExplainExecInfo(struct SOperatorInfo* operatorInfo, SArray* pExecInfoList);
 
 #ifdef __cplusplus
 }

source/libs/executor/inc/querytask.h

@@ -22,6 +22,17 @@ extern "C" {
 
 #define GET_TASKID(_t) (((SExecTaskInfo*)(_t))->id.str)
 
+enum {
+  // when this task starts to execute, this status will set
+      TASK_NOT_COMPLETED = 0x1u,
+
+  /* Task is over
+   * 1. this status is used in one row result query process, e.g., count/sum/first/last/ avg...etc.
+   * 2. when all data within queried time window, it is also denoted as query_completed
+   */
+      TASK_COMPLETED = 0x2u,
+};
+
 typedef struct STaskIdInfo {
   uint64_t queryId;  // this is also a request id
   uint64_t subplanId;
@@ -44,6 +55,23 @@ typedef struct STaskStopInfo {
   SArray*  pStopInfo;
 } STaskStopInfo;
 
+typedef struct {
+  STqOffsetVal        currentOffset;  // for tmq
+  SMqMetaRsp          metaRsp;        // for tmq fetching meta
+  int64_t             snapshotVer;
+  SPackedData         submit;  // todo remove it
+  SSchemaWrapper*     schema;
+  char                tbName[TSDB_TABLE_NAME_LEN];   // this is the current scan table: todo refactor
+  int8_t              recoverStep;
+  int8_t              recoverScanFinished;
+  SQueryTableDataCond tableCond;
+  int64_t             fillHistoryVer1;
+  int64_t             fillHistoryVer2;
+  SStreamState*       pState;
+  int64_t             dataVersion;
+  int64_t             checkPointId;
+} SStreamTaskInfo;
+
 struct SExecTaskInfo {
   STaskIdInfo           id;
   uint32_t              status;
@@ -75,6 +103,7 @@ 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);
+SArray*        getTableListInfo(const SExecTaskInfo* pTaskInfo);
 
 #ifdef __cplusplus
 }

source/libs/executor/src/aggregateoperator.c

@@ -20,16 +20,16 @@
 #include "tfill.h"
 #include "tname.h"
 
-#include "tdatablock.h"
-#include "tglobal.h"
-#include "executorimpl.h"
+#include "executorInt.h"
 #include "index.h"
+#include "operator.h"
 #include "query.h"
+#include "querytask.h"
 #include "tcompare.h"
+#include "tdatablock.h"
+#include "tglobal.h"
 #include "thash.h"
 #include "ttypes.h"
-#include "operator.h"
-#include "querytask.h"
 
 typedef struct {
   bool    hasAgg;

source/libs/executor/src/cachescanoperator.c

@@ -20,12 +20,12 @@
 #include "tdatablock.h"
 #include "tmsg.h"
 
-#include "executorimpl.h"
+#include "executorInt.h"
+#include "operator.h"
+#include "querytask.h"
 #include "tcompare.h"
 #include "thash.h"
 #include "ttypes.h"
-#include "operator.h"
-#include "querytask.h"
 
 typedef struct SCacheRowsScanInfo {
   SSDataBlock*    pRes;

source/libs/executor/src/dataDeleter.c

@@ -15,7 +15,7 @@
 
 #include "dataSinkInt.h"
 #include "dataSinkMgt.h"
-#include "executorimpl.h"
+#include "executorInt.h"
 #include "planner.h"
 #include "tcompression.h"
 #include "tdatablock.h"

source/libs/executor/src/dataDispatcher.c

@@ -15,7 +15,7 @@
 
 #include "dataSinkInt.h"
 #include "dataSinkMgt.h"
-#include "executorimpl.h"
+#include "executorInt.h"
 #include "planner.h"
 #include "tcompression.h"
 #include "tdatablock.h"

source/libs/executor/src/dataInserter.c

@@ -15,7 +15,7 @@
 
 #include "dataSinkInt.h"
 #include "dataSinkMgt.h"
-#include "executorimpl.h"
+#include "executorInt.h"
 #include "planner.h"
 #include "tcompression.h"
 #include "tdatablock.h"

source/libs/executor/src/eventwindowoperator.c

@@ -13,16 +13,16 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "executorimpl.h"
+#include "executorInt.h"
 #include "filter.h"
 #include "function.h"
 #include "functionMgt.h"
+#include "operator.h"
+#include "querytask.h"
 #include "tcommon.h"
 #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

@@ -13,19 +13,19 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include "executorInt.h"
 #include "filter.h"
 #include "function.h"
-#include "os.h"
-#include "tname.h"
-#include "tref.h"
-#include "tdatablock.h"
-#include "tmsg.h"
-#include "executorimpl.h"
 #include "index.h"
-#include "query.h"
-#include "thash.h"
 #include "operator.h"
+#include "os.h"
+#include "query.h"
 #include "querytask.h"
+#include "tdatablock.h"
+#include "thash.h"
+#include "tmsg.h"
+#include "tname.h"
+#include "tref.h"
 
 typedef struct SFetchRspHandleWrapper {
   uint32_t exchangeId;

source/libs/executor/src/executil.c

@@ -24,9 +24,9 @@
 #include "ttime.h"
 
 #include "executil.h"
-#include "executorimpl.h"
-#include "tcompression.h"
+#include "executorInt.h"
 #include "querytask.h"
+#include "tcompression.h"
 
 typedef struct STableListIdInfo {
   uint64_t suid;

source/libs/executor/src/executor.c

@@ -14,14 +14,14 @@
  */
 
 #include "executor.h"
-#include "executorimpl.h"
+#include "executorInt.h"
+#include "operator.h"
 #include "planner.h"
+#include "querytask.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;
@@ -718,8 +718,6 @@ void qRemoveTaskStopInfo(SExecTaskInfo* pTaskInfo, SExchangeOpStopInfo* pInfo) {
     taosArrayRemove(pTaskInfo->stopInfo.pStopInfo, idx);
   }
   taosWUnLockLatch(&pTaskInfo->stopInfo.lock);
-
-  return;
 }
 
 void qStopTaskOperators(SExecTaskInfo* pTaskInfo) {
@@ -1304,3 +1302,25 @@ SArray* qGetQueriedTableListInfo(qTaskInfo_t tinfo) {
   taosArrayDestroy(plist);
   return pUidList;
 }
+
+static void extractTableList(SArray* pList, const SOperatorInfo* pOperator) {
+  if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
+    SStreamScanInfo* pScanInfo = pOperator->info;
+    STableScanInfo*  pTableScanInfo = pScanInfo->pTableScanOp->info;
+    taosArrayPush(pList, &pTableScanInfo->base.pTableListInfo);
+  } else if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {
+    STableScanInfo* pScanInfo = pOperator->info;
+    taosArrayPush(pList, &pScanInfo->base.pTableListInfo);
+  } else {
+    if (pOperator->pDownstream != NULL && pOperator->pDownstream[0] != NULL) {
+      extractTableList(pList, pOperator->pDownstream[0]);
+    }
+  }
+}
+
+SArray* getTableListInfo(const SExecTaskInfo* pTaskInfo) {
+  SArray*        pArray = taosArrayInit(0, POINTER_BYTES);
+  SOperatorInfo* pOperator = pTaskInfo->pRoot;
+  extractTableList(pArray, pOperator);
+  return pArray;
+}
\ No newline at end of file

source/libs/executor/src/executorimpl.c → source/libs/executor/src/executorInt.c

@@ -25,15 +25,15 @@
 #include "tmsg.h"
 #include "ttime.h"
 
-#include "executorimpl.h"
+#include "executorInt.h"
 #include "index.h"
+#include "operator.h"
 #include "query.h"
+#include "querytask.h"
 #include "tcompare.h"
 #include "thash.h"
 #include "ttypes.h"
 #include "vnode.h"
-#include "operator.h"
-#include "querytask.h"
 
 #define SET_REVERSE_SCAN_FLAG(runtime)    ((runtime)->scanFlag = REVERSE_SCAN)
 #define GET_FORWARD_DIRECTION_FACTOR(ord) (((ord) == TSDB_ORDER_ASC) ? QUERY_ASC_FORWARD_STEP : QUERY_DESC_FORWARD_STEP)
@@ -1059,37 +1059,6 @@ int32_t createDataSinkParam(SDataSinkNode* pNode, void** pParam, SExecTaskInfo*
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SArray* pExecInfoList) {
-  SExplainExecInfo  execInfo = {0};
-  SExplainExecInfo* pExplainInfo = taosArrayPush(pExecInfoList, &execInfo);
-
-  pExplainInfo->numOfRows = operatorInfo->resultInfo.totalRows;
-  pExplainInfo->startupCost = operatorInfo->cost.openCost;
-  pExplainInfo->totalCost = operatorInfo->cost.totalCost;
-  pExplainInfo->verboseLen = 0;
-  pExplainInfo->verboseInfo = NULL;
-
-  if (operatorInfo->fpSet.getExplainFn) {
-    int32_t code =
-        operatorInfo->fpSet.getExplainFn(operatorInfo, &pExplainInfo->verboseInfo, &pExplainInfo->verboseLen);
-    if (code) {
-      qError("%s operator getExplainFn failed, code:%s", GET_TASKID(operatorInfo->pTaskInfo), tstrerror(code));
-      return code;
-    }
-  }
-
-  int32_t code = 0;
-  for (int32_t i = 0; i < operatorInfo->numOfDownstream; ++i) {
-    code = getOperatorExplainExecInfo(operatorInfo->pDownstream[i], pExecInfoList);
-    if (code != TSDB_CODE_SUCCESS) {
-      //      taosMemoryFreeClear(*pRes);
-      return TSDB_CODE_OUT_OF_MEMORY;
-    }
-  }
-
-  return TSDB_CODE_SUCCESS;
-}
-
 int32_t setOutputBuf(SStreamState* pState, STimeWindow* win, SResultRow** pResult, int64_t tableGroupId,
                      SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup) {
   SWinKey key = {
@@ -1331,25 +1300,3 @@ void qStreamCloseTsdbReader(void* task) {
     }
   }
 }
-
-static void extractTableList(SArray* pList, const SOperatorInfo* pOperator) {
-  if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
-    SStreamScanInfo* pScanInfo = pOperator->info;
-    STableScanInfo*  pTableScanInfo = pScanInfo->pTableScanOp->info;
-    taosArrayPush(pList, &pTableScanInfo->base.pTableListInfo);
-  } else if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {
-    STableScanInfo* pScanInfo = pOperator->info;
-    taosArrayPush(pList, &pScanInfo->base.pTableListInfo);
-  } else {
-    if (pOperator->pDownstream != NULL && pOperator->pDownstream[0] != NULL) {
-      extractTableList(pList, pOperator->pDownstream[0]);
-    }
-  }
-}
-
-SArray* getTableListInfo(const SExecTaskInfo* pTaskInfo) {
-  SArray*        pArray = taosArrayInit(0, POINTER_BYTES);
-  SOperatorInfo* pOperator = pTaskInfo->pRoot;
-  extractTableList(pArray, pOperator);
-  return pArray;
-}
\ No newline at end of file

source/libs/executor/src/filloperator.c

@@ -20,7 +20,7 @@
 #include "tmsg.h"
 #include "ttypes.h"
 
-#include "executorimpl.h"
+#include "executorInt.h"
 #include "tcommon.h"
 #include "thash.h"
 #include "ttime.h"

source/libs/executor/src/groupoperator.c

@@ -22,12 +22,11 @@
 #include "tmsg.h"
 
 #include "executorInt.h"
-#include "executorimpl.h"
+#include "operator.h"
+#include "querytask.h"
 #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

@@ -13,18 +13,18 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include "executorInt.h"
 #include "filter.h"
-#include "executorimpl.h"
 #include "function.h"
+#include "operator.h"
 #include "os.h"
 #include "querynodes.h"
+#include "querytask.h"
 #include "tcompare.h"
 #include "tdatablock.h"
 #include "thash.h"
 #include "tmsg.h"
 #include "ttypes.h"
-#include "operator.h"
-#include "querytask.h"
 
 typedef struct SJoinRowCtx {
   bool    rowRemains;

source/libs/executor/src/operator.c

@@ -20,12 +20,12 @@
 
 #include "tglobal.h"
 
-#include "executorimpl.h"
+#include "executorInt.h"
 #include "index.h"
-#include "query.h"
-#include "vnode.h"
 #include "operator.h"
+#include "query.h"
 #include "querytask.h"
+#include "vnode.h"
 
 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,
@@ -75,28 +75,6 @@ void setOperatorInfo(SOperatorInfo* pOperator, const char* name, int32_t type, b
   pOperator->pTaskInfo = pTaskInfo;
 }
 
-void destroyOperator(SOperatorInfo* pOperator) {
-  if (pOperator == NULL) {
-    return;
-  }
-
-  if (pOperator->fpSet.closeFn != NULL) {
-    pOperator->fpSet.closeFn(pOperator->info);
-  }
-
-  if (pOperator->pDownstream != NULL) {
-    for (int32_t i = 0; i < pOperator->numOfDownstream; ++i) {
-      destroyOperator(pOperator->pDownstream[i]);
-    }
-
-    taosMemoryFreeClear(pOperator->pDownstream);
-    pOperator->numOfDownstream = 0;
-  }
-
-  cleanupExprSupp(&pOperator->exprSupp);
-  taosMemoryFreeClear(pOperator);
-}
-
 // each operator should be set their own function to return total cost buffer
 int32_t optrDefaultBufFn(SOperatorInfo* pOperator) {
   if (pOperator->blocking) {
@@ -106,40 +84,6 @@ int32_t optrDefaultBufFn(SOperatorInfo* pOperator) {
   }
 }
 
-//int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scanFlag, bool inheritUsOrder) {
-//  // todo add more information about exchange operation
-//  int32_t type = pOperator->operatorType;
-//  if (type == QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN || type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN ||
-//      type == QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN || type == QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN ||
-//      type == QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN || type == QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN) {
-//    *order = TSDB_ORDER_ASC;
-//    *scanFlag = MAIN_SCAN;
-//    return TSDB_CODE_SUCCESS;
-//  } else if (type == QUERY_NODE_PHYSICAL_PLAN_EXCHANGE) {
-//    if (!inheritUsOrder) {
-//      *order = TSDB_ORDER_ASC;
-//    }
-//    *scanFlag = MAIN_SCAN;
-//    return TSDB_CODE_SUCCESS;
-//  } else if (type == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {
-//    STableScanInfo* pTableScanInfo = pOperator->info;
-//    *order = pTableScanInfo->base.cond.order;
-//    *scanFlag = pTableScanInfo->base.scanFlag;
-//    return TSDB_CODE_SUCCESS;
-//  } else if (type == QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN) {
-//    STableMergeScanInfo* pTableScanInfo = pOperator->info;
-//    *order = pTableScanInfo->base.cond.order;
-//    *scanFlag = pTableScanInfo->base.scanFlag;
-//    return TSDB_CODE_SUCCESS;
-//  } else {
-//    if (pOperator->pDownstream == NULL || pOperator->pDownstream[0] == NULL) {
-//      return TSDB_CODE_INVALID_PARA;
-//    } else {
-//      return getTableScanInfo(pOperator->pDownstream[0], order, scanFlag, inheritUsOrder);
-//    }
-//  }
-//}
-
 static int64_t getQuerySupportBufSize(size_t numOfTables) {
   size_t s1 = sizeof(STableQueryInfo);
   //  size_t s3 = sizeof(STableCheckInfo);  buffer consumption in tsdb
@@ -319,7 +263,7 @@ int32_t stopTableScanOperator(SOperatorInfo* pOperator, const char* pIdStr) {
 }
 
 SOperatorInfo* createOperator(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHandle* pHandle, SNode* pTagCond,
-                                  SNode* pTagIndexCond, const char* pUser, const char* dbname) {
+                              SNode* pTagIndexCond, const char* pUser, const char* dbname) {
   int32_t     type = nodeType(pPhyNode);
   const char* idstr = GET_TASKID(pTaskInfo);
 
@@ -347,7 +291,7 @@ SOperatorInfo* createOperator(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SR
 
       code = initQueriedTableSchemaInfo(pHandle, &pTableScanNode->scan, dbname, pTaskInfo);
       if (code) {
-        pTaskInfo->code = terrno;
+        pTaskInfo->code = code;
         tableListDestroy(pTableListInfo);
         return NULL;
       }
@@ -355,6 +299,7 @@ SOperatorInfo* createOperator(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SR
       pOperator = createTableScanOperatorInfo(pTableScanNode, pHandle, pTableListInfo, pTaskInfo);
       if (NULL == pOperator) {
         pTaskInfo->code = terrno;
+        tableListDestroy(pTableListInfo);
         return NULL;
       }
 
@@ -578,3 +523,56 @@ SOperatorInfo* createOperator(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SR
 
   return pOptr;
 }
+
+void destroyOperator(SOperatorInfo* pOperator) {
+  if (pOperator == NULL) {
+    return;
+  }
+
+  if (pOperator->fpSet.closeFn != NULL) {
+    pOperator->fpSet.closeFn(pOperator->info);
+  }
+
+  if (pOperator->pDownstream != NULL) {
+    for (int32_t i = 0; i < pOperator->numOfDownstream; ++i) {
+      destroyOperator(pOperator->pDownstream[i]);
+    }
+
+    taosMemoryFreeClear(pOperator->pDownstream);
+    pOperator->numOfDownstream = 0;
+  }
+
+  cleanupExprSupp(&pOperator->exprSupp);
+  taosMemoryFreeClear(pOperator);
+}
+
+int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SArray* pExecInfoList) {
+  SExplainExecInfo  execInfo = {0};
+  SExplainExecInfo* pExplainInfo = taosArrayPush(pExecInfoList, &execInfo);
+
+  pExplainInfo->numOfRows = operatorInfo->resultInfo.totalRows;
+  pExplainInfo->startupCost = operatorInfo->cost.openCost;
+  pExplainInfo->totalCost = operatorInfo->cost.totalCost;
+  pExplainInfo->verboseLen = 0;
+  pExplainInfo->verboseInfo = NULL;
+
+  if (operatorInfo->fpSet.getExplainFn) {
+    int32_t code =
+        operatorInfo->fpSet.getExplainFn(operatorInfo, &pExplainInfo->verboseInfo, &pExplainInfo->verboseLen);
+    if (code) {
+      qError("%s operator getExplainFn failed, code:%s", GET_TASKID(operatorInfo->pTaskInfo), tstrerror(code));
+      return code;
+    }
+  }
+
+  int32_t code = 0;
+  for (int32_t i = 0; i < operatorInfo->numOfDownstream; ++i) {
+    code = getOperatorExplainExecInfo(operatorInfo->pDownstream[i], pExecInfoList);
+    if (code != TSDB_CODE_SUCCESS) {
+      //      taosMemoryFreeClear(*pRes);
+      return TSDB_CODE_OUT_OF_MEMORY;
+    }
+  }
+
+  return TSDB_CODE_SUCCESS;
+}

source/libs/executor/src/projectoperator.c

@@ -13,7 +13,7 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "executorimpl.h"
+#include "executorInt.h"
 #include "filter.h"
 #include "functionMgt.h"
 #include "operator.h"

source/libs/executor/src/querytask.c

@@ -24,14 +24,14 @@
 #include "tdatablock.h"
 #include "tmsg.h"
 
-#include "executorimpl.h"
+#include "executorInt.h"
 #include "index.h"
+#include "operator.h"
 #include "query.h"
+#include "querytask.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)))
 

source/libs/executor/src/scanoperator.c

@@ -13,7 +13,7 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "executorimpl.h"
+#include "executorInt.h"
 #include "filter.h"
 #include "function.h"
 #include "functionMgt.h"

source/libs/executor/src/sortoperator.c

@@ -13,11 +13,11 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include "executorInt.h"
 #include "filter.h"
-#include "executorimpl.h"
-#include "tdatablock.h"
 #include "operator.h"
 #include "querytask.h"
+#include "tdatablock.h"
 
 typedef struct SSortOperatorInfo {
   SOptrBasicInfo binfo;

source/libs/executor/src/sysscanoperator.c

@@ -13,7 +13,7 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "executorimpl.h"
+#include "executorInt.h"
 #include "filter.h"
 #include "function.h"
 #include "functionMgt.h"

source/libs/executor/src/tfill.c

@@ -20,7 +20,7 @@
 #include "tmsg.h"
 #include "ttypes.h"
 
-#include "executorimpl.h"
+#include "executorInt.h"
 #include "tcommon.h"
 #include "thash.h"
 #include "ttime.h"

source/libs/executor/src/timesliceoperator.c

@@ -12,17 +12,17 @@
  * 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 "executorimpl.h"
+#include "executorInt.h"
 #include "filter.h"
 #include "function.h"
 #include "functionMgt.h"
+#include "operator.h"
+#include "querytask.h"
 #include "tcommon.h"
 #include "tcompare.h"
 #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

@@ -12,21 +12,27 @@
  * 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 "executorimpl.h"
+#include "executorInt.h"
 #include "filter.h"
 #include "function.h"
 #include "functionMgt.h"
+#include "operator.h"
+#include "querytask.h"
 #include "tcommon.h"
 #include "tcompare.h"
 #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);
 
+typedef struct SStateWindowInfo {
+  SResultWindowInfo winInfo;
+  SStateKeys*       pStateKey;
+} SStateWindowInfo;
+
+
 typedef struct SSessionAggOperatorInfo {
   SOptrBasicInfo     binfo;
   SAggSupporter      aggSup;

source/libs/executor/test/executorTests.cpp

@@ -24,13 +24,13 @@
 #include "os.h"
 
 #include "executor.h"
-#include "executorimpl.h"
+#include "executorInt.h"
 #include "function.h"
+#include "operator.h"
 #include "taos.h"
 #include "tdatablock.h"
 #include "tdef.h"
 #include "tvariant.h"
-#include "operator.h"
 
 namespace {
 

source/libs/executor/test/lhashTests.cpp

@@ -15,7 +15,7 @@
 
 #include <gtest/gtest.h>
 #include <iostream>
-#include "executorimpl.h"
+#include "executorInt.h"
 #include "tlinearhash.h"
 
 #pragma GCC diagnostic push

source/libs/executor/test/sortTests.cpp

@@ -26,7 +26,7 @@
 #include "os.h"
 
 #include "executor.h"
-#include "executorimpl.h"
+#include "executorInt.h"
 #include "taos.h"
 #include "tcompare.h"
 #include "tdatablock.h"