[td-11818] Refactor API.

include/libs/executor/executor.h

@@ -27,6 +27,10 @@ typedef void* DataSinkHandle;
 struct SRpcMsg;
 struct SSubplan;
 
+typedef struct SReadHandle {
+  void* reader;
+  void* meta;
+} SReadHandle;
  /**
   * Create the exec task for streaming mode
   * @param pMsg
@@ -35,7 +39,13 @@ struct SSubplan;
   */
 qTaskInfo_t qCreateStreamExecTaskInfo(void *msg, void* streamReadHandle);
 
-int32_t     qSetStreamInput(qTaskInfo_t tinfo, void* input);
+/**
+ *
+ * @param tinfo
+ * @param input
+ * @return
+ */
+int32_t     qSetStreamInput(qTaskInfo_t tinfo, const void* input);
 
  /**
   * Create the exec task object according to task json
@@ -46,7 +56,7 @@ int32_t     qSetStreamInput(qTaskInfo_t tinfo, void* input);
   * @param qId
   * @return
   */
-int32_t qCreateExecTask(void* readHandle, int32_t vgId, uint64_t taskId, struct SSubplan* pPlan, qTaskInfo_t* pTaskInfo, DataSinkHandle* handle);
+int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId, struct SSubplan* pPlan, qTaskInfo_t* pTaskInfo, DataSinkHandle* handle);
 
 /**
  * The main task execution function, including query on both table and multiple tables,

include/libs/function/function.h

@@ -107,14 +107,14 @@ typedef struct SPoint1 {
   union{double  val; char* ptr;};
 } SPoint1;
 
-struct SQLFunctionCtx;
+struct SqlFunctionCtx;
 struct SResultRowEntryInfo;
 
 //for selectivity query, the corresponding tag value is assigned if the data is qualified
 typedef struct SExtTagsInfo {
   int16_t                 tagsLen;      // keep the tags data for top/bottom query result
   int16_t                 numOfTagCols;
-  struct SQLFunctionCtx **pTagCtxList;
+  struct SqlFunctionCtx **pTagCtxList;
 } SExtTagsInfo;
 
 typedef struct SResultDataInfo {
@@ -126,18 +126,18 @@ typedef struct SResultDataInfo {
 #define GET_RES_INFO(ctx) ((ctx)->resultInfo)
 
 typedef struct SFunctionFpSet {
-  bool (*init)(struct SQLFunctionCtx *pCtx, struct SResultRowEntryInfo* pResultCellInfo);  // setup the execute environment
-  void (*addInput)(struct SQLFunctionCtx *pCtx);
+  bool (*init)(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResultCellInfo);  // setup the execute environment
+  void (*addInput)(struct SqlFunctionCtx *pCtx);
 
   // finalizer must be called after all exec has been executed to generated final result.
-  void (*finalize)(struct SQLFunctionCtx *pCtx);
-  void (*combine)(struct SQLFunctionCtx *pCtx);
+  void (*finalize)(struct SqlFunctionCtx *pCtx);
+  void (*combine)(struct SqlFunctionCtx *pCtx);
 } SFunctionFpSet;
 
 extern SFunctionFpSet fpSet[1];
 
 // sql function runtime context
-typedef struct SQLFunctionCtx {
+typedef struct SqlFunctionCtx {
   int32_t      size;      // number of rows
   void *       pInput;    // input data buffer
   uint32_t     order;     // asc|desc
@@ -167,7 +167,7 @@ typedef struct SQLFunctionCtx {
 
   int32_t      columnIndex;
   SFunctionFpSet* fpSet;
-} SQLFunctionCtx;
+} SqlFunctionCtx;
 
 enum {
   TEXPR_NODE_DUMMY     = 0x0,
@@ -216,14 +216,14 @@ typedef struct SAggFunctionInfo {
   int8_t    sFunctionId;  // Transfer function for super table query
   uint16_t  status;
 
-  bool (*init)(SQLFunctionCtx *pCtx, struct SResultRowEntryInfo* pResultCellInfo);  // setup the execute environment
-  void (*addInput)(SQLFunctionCtx *pCtx);
+  bool (*init)(SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResultCellInfo);  // setup the execute environment
+  void (*addInput)(SqlFunctionCtx *pCtx);
 
   // finalizer must be called after all exec has been executed to generated final result.
-  void (*finalize)(SQLFunctionCtx *pCtx);
-  void (*combine)(SQLFunctionCtx *pCtx);
+  void (*finalize)(SqlFunctionCtx *pCtx);
+  void (*combine)(SqlFunctionCtx *pCtx);
 
-  int32_t (*dataReqFunc)(SQLFunctionCtx *pCtx, STimeWindow* w, int32_t colId);
+  int32_t (*dataReqFunc)(SqlFunctionCtx *pCtx, STimeWindow* w, int32_t colId);
 } SAggFunctionInfo;
 
 struct SScalarFuncParam;
@@ -279,9 +279,9 @@ void extractFunctionDesc(SArray* pFunctionIdList, SMultiFunctionsDesc* pDesc);
 
 tExprNode* exprdup(tExprNode* pTree);
 
-void resetResultRowEntryResult(SQLFunctionCtx* pCtx, int32_t num);
+void resetResultRowEntryResult(SqlFunctionCtx* pCtx, int32_t num);
 void cleanupResultRowEntry(struct SResultRowEntryInfo* pCell);
-int32_t getNumOfResult(SQLFunctionCtx* pCtx, int32_t num);
+int32_t getNumOfResult(SqlFunctionCtx* pCtx, int32_t num);
 bool isRowEntryCompleted(struct SResultRowEntryInfo* pEntry);
 bool isRowEntryInitialized(struct SResultRowEntryInfo* pEntry);
 

include/libs/function/functionMgt.h

@@ -22,7 +22,7 @@ extern "C" {
 
 #include "nodes.h"
 
-struct SQLFunctionCtx;
+struct SqlFunctionCtx;
 struct SResultRowEntryInfo;
 struct STimeWindow;
 
@@ -32,9 +32,9 @@ typedef struct SFuncExecEnv {
 
 typedef void* FuncMgtHandle;
 typedef bool (*FExecGetEnv)(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
-typedef bool (*FExecInit)(struct SQLFunctionCtx *pCtx, struct SResultRowEntryInfo* pResultCellInfo);
-typedef void (*FExecProcess)(struct SQLFunctionCtx *pCtx);
-typedef void (*FExecFinalize)(struct SQLFunctionCtx *pCtx);
+typedef bool (*FExecInit)(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResultCellInfo);
+typedef void (*FExecProcess)(struct SqlFunctionCtx *pCtx);
+typedef void (*FExecFinalize)(struct SqlFunctionCtx *pCtx);
 
 typedef struct SFuncExecFuncs {
   FExecGetEnv getEnv;

include/libs/planner/plannerOp.h

@@ -24,7 +24,6 @@
 #endif
 
 OP_ENUM_MACRO(StreamScan)
-OP_ENUM_MACRO(TableScan)
 OP_ENUM_MACRO(DataBlocksOptScan)
 OP_ENUM_MACRO(TableSeqScan)
 OP_ENUM_MACRO(TagScan)
@@ -48,3 +47,5 @@ OP_ENUM_MACRO(AllTimeWindow)
 OP_ENUM_MACRO(AllMultiTableTimeInterval)
 OP_ENUM_MACRO(Order)
 OP_ENUM_MACRO(Exchange)
+
+//OP_ENUM_MACRO(TableScan)

source/dnode/vnode/inc/tsdb.h

@@ -91,7 +91,7 @@ int    tsdbCommit(STsdb *pTsdb);
 int  tsdbOptionsInit(STsdbCfg *);
 void tsdbOptionsClear(STsdbCfg *);
 
-typedef void* tsdbReadHandleT;
+typedef void* tsdbReaderT;
 
 /**
  * Get the data block iterator, starting from position according to the query condition
@@ -103,7 +103,7 @@ typedef void* tsdbReadHandleT;
  * @param qinfo      query info handle from query processor
  * @return
  */
-tsdbReadHandleT *tsdbQueryTables(STsdb *tsdb, STsdbQueryCond *pCond, STableGroupInfo *tableInfoGroup, uint64_t qId, uint64_t taskId);
+tsdbReaderT *tsdbQueryTables(STsdb *tsdb, STsdbQueryCond *pCond, STableGroupInfo *tableInfoGroup, uint64_t qId, uint64_t taskId);
 
 /**
  * Get the last row of the given query time window for all the tables in STableGroupInfo object.
@@ -115,13 +115,13 @@ tsdbReadHandleT *tsdbQueryTables(STsdb *tsdb, STsdbQueryCond *pCond, STableGroup
  * @param tableInfo  table list.
  * @return
  */
-//tsdbReadHandleT tsdbQueryLastRow(STsdbRepo *tsdb, STsdbQueryCond *pCond, STableGroupInfo *tableInfo, uint64_t qId,
+//tsdbReaderT tsdbQueryLastRow(STsdbRepo *tsdb, STsdbQueryCond *pCond, STableGroupInfo *tableInfo, uint64_t qId,
 //                                  SMemRef *pRef);
 
 
-tsdbReadHandleT tsdbQueryCacheLast(STsdb *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList, uint64_t qId, void* pMemRef);
+tsdbReaderT tsdbQueryCacheLast(STsdb *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList, uint64_t qId, void* pMemRef);
 
-bool isTsdbCacheLastRow(tsdbReadHandleT* pTsdbReadHandle);
+bool isTsdbCacheLastRow(tsdbReaderT* pTsdbReadHandle);
 
 /**
  *
@@ -148,7 +148,7 @@ int32_t tsdbQuerySTableByTagCond(void* pMeta, uint64_t uid, TSKEY skey, const ch
  * @return row size
  */
 
-int64_t tsdbGetNumOfRowsInMemTable(tsdbReadHandleT* pHandle);
+int64_t tsdbGetNumOfRowsInMemTable(tsdbReaderT* pHandle);
 
 /**
  * move to next block if exists
@@ -156,7 +156,7 @@ int64_t tsdbGetNumOfRowsInMemTable(tsdbReadHandleT* pHandle);
  * @param pTsdbReadHandle
  * @return
  */
-bool tsdbNextDataBlock(tsdbReadHandleT pTsdbReadHandle);
+bool tsdbNextDataBlock(tsdbReaderT pTsdbReadHandle);
 
 /**
  * Get current data block information
@@ -165,7 +165,7 @@ bool tsdbNextDataBlock(tsdbReadHandleT pTsdbReadHandle);
  * @param pBlockInfo
  * @return
  */
-void tsdbRetrieveDataBlockInfo(tsdbReadHandleT *pTsdbReadHandle, SDataBlockInfo *pBlockInfo);
+void tsdbRetrieveDataBlockInfo(tsdbReaderT *pTsdbReadHandle, SDataBlockInfo *pBlockInfo);
 
 /**
  *
@@ -177,7 +177,7 @@ void tsdbRetrieveDataBlockInfo(tsdbReadHandleT *pTsdbReadHandle, SDataBlockInfo
  * @pBlockStatis the pre-calculated value for current data blocks. if the block is a cache block, always return 0
  * @return
  */
-int32_t tsdbRetrieveDataBlockStatisInfo(tsdbReadHandleT *pTsdbReadHandle, SDataStatis **pBlockStatis);
+int32_t tsdbRetrieveDataBlockStatisInfo(tsdbReaderT *pTsdbReadHandle, SDataStatis **pBlockStatis);
 
 /**
  *
@@ -189,7 +189,7 @@ int32_t tsdbRetrieveDataBlockStatisInfo(tsdbReadHandleT *pTsdbReadHandle, SDataS
  * @param pColumnIdList     required data columns id list
  * @return
  */
-SArray *tsdbRetrieveDataBlock(tsdbReadHandleT *pTsdbReadHandle, SArray *pColumnIdList);
+SArray *tsdbRetrieveDataBlock(tsdbReaderT *pTsdbReadHandle, SArray *pColumnIdList);
 
 /**
  * destroy the created table group list, which is generated by tag query
@@ -220,7 +220,7 @@ int32_t tsdbGetTableGroupFromIdList(STsdb *tsdb, SArray *pTableIdList, STableGro
  * clean up the query handle
  * @param queryHandle
  */
-void tsdbCleanupReadHandle(tsdbReadHandleT queryHandle);
+void tsdbCleanupReadHandle(tsdbReaderT queryHandle);
 
 #ifdef __cplusplus
 }

source/dnode/vnode/inc/vnode.h

@@ -71,7 +71,7 @@ typedef struct {
 typedef struct STqReadHandle {
   int64_t        ver;
   uint64_t       tbUid;
-  SSubmitMsg*    pMsg;
+  const SSubmitMsg*    pMsg;
   SSubmitBlk*    pBlock;
   SSubmitMsgIter msgIter;
   SSubmitBlkIter blkIter;

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

@@ -159,7 +159,7 @@ static int32_t checkForCachedLastRow(STsdbReadHandle* pTsdbReadHandle, STableGro
 static int32_t checkForCachedLast(STsdbReadHandle* pTsdbReadHandle);
 //static int32_t tsdbGetCachedLastRow(STable* pTable, SMemRow* pRes, TSKEY* lastKey);
 
-static void    changeQueryHandleForInterpQuery(tsdbReadHandleT pHandle);
+static void    changeQueryHandleForInterpQuery(tsdbReaderT pHandle);
 static void    doMergeTwoLevelData(STsdbReadHandle* pTsdbReadHandle, STableCheckInfo* pCheckInfo, SBlock* pBlock);
 static int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order);
 static int32_t tsdbReadRowsFromCache(STableCheckInfo* pCheckInfo, TSKEY maxKey, int maxRowsToRead, STimeWindow* win, STsdbReadHandle* pTsdbReadHandle);
@@ -167,7 +167,7 @@ static int32_t tsdbCheckInfoCompar(const void* key1, const void* key2);
 //static int32_t doGetExternalRow(STsdbReadHandle* pTsdbReadHandle, int16_t type, void* pMemRef);
 //static void*   doFreeColumnInfoData(SArray* pColumnInfoData);
 //static void*   destroyTableCheckInfo(SArray* pTableCheckInfo);
-static bool    tsdbGetExternalRow(tsdbReadHandleT pHandle);
+static bool    tsdbGetExternalRow(tsdbReaderT pHandle);
 
 static void tsdbInitDataBlockLoadInfo(SDataBlockLoadInfo* pBlockLoadInfo) {
   pBlockLoadInfo->slot = -1;
@@ -208,7 +208,7 @@ static SArray* getDefaultLoadColumns(STsdbReadHandle* pTsdbReadHandle, bool load
   return pLocalIdList;
 }
 
-//int64_t tsdbGetNumOfRowsInMemTable(tsdbReadHandleT* pHandle) {
+//int64_t tsdbGetNumOfRowsInMemTable(tsdbReaderT* pHandle) {
 //  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*) pHandle;
 //
 //  int64_t rows = 0;
@@ -424,7 +424,7 @@ static STsdbReadHandle* tsdbQueryTablesImpl(STsdb* tsdb, STsdbQueryCond* pCond,
   tsdbInitDataBlockLoadInfo(&pReadHandle->dataBlockLoadInfo);
   tsdbInitCompBlockLoadInfo(&pReadHandle->compBlockLoadInfo);
 
-  return (tsdbReadHandleT)pReadHandle;
+  return (tsdbReaderT)pReadHandle;
 
   _end:
   tsdbCleanupReadHandle(pReadHandle);
@@ -432,14 +432,14 @@ static STsdbReadHandle* tsdbQueryTablesImpl(STsdb* tsdb, STsdbQueryCond* pCond,
   return NULL;
 }
 
-tsdbReadHandleT* tsdbQueryTables(STsdb* tsdb, STsdbQueryCond* pCond, STableGroupInfo* groupList, uint64_t qId, uint64_t taskId) {
+tsdbReaderT* tsdbQueryTables(STsdb* tsdb, STsdbQueryCond* pCond, STableGroupInfo* groupList, uint64_t qId, uint64_t taskId) {
   STsdbReadHandle* pTsdbReadHandle = tsdbQueryTablesImpl(tsdb, pCond, qId, taskId);
   if (pTsdbReadHandle == NULL) {
     return NULL;
   }
 
   if (emptyQueryTimewindow(pTsdbReadHandle)) {
-    return (tsdbReadHandleT*) pTsdbReadHandle;
+    return (tsdbReaderT*) pTsdbReadHandle;
   }
 
   // todo apply the lastkey of table check to avoid to load header file
@@ -453,10 +453,10 @@ tsdbReadHandleT* tsdbQueryTables(STsdb* tsdb, STsdbQueryCond* pCond, STableGroup
   tsdbDebug("%p total numOfTable:%" PRIzu " in this query, group %"PRIzu" %s", pTsdbReadHandle, taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo),
       taosArrayGetSize(groupList->pGroupList), pTsdbReadHandle->idStr);
 
-  return (tsdbReadHandleT) pTsdbReadHandle;
+  return (tsdbReaderT) pTsdbReadHandle;
 }
 
-void tsdbResetQueryHandle(tsdbReadHandleT queryHandle, STsdbQueryCond *pCond) {
+void tsdbResetQueryHandle(tsdbReaderT queryHandle, STsdbQueryCond *pCond) {
   STsdbReadHandle* pTsdbReadHandle = queryHandle;
 
   if (emptyQueryTimewindow(pTsdbReadHandle)) {
@@ -493,7 +493,7 @@ void tsdbResetQueryHandle(tsdbReadHandleT queryHandle, STsdbQueryCond *pCond) {
   resetCheckInfo(pTsdbReadHandle);
 }
 
-void tsdbResetQueryHandleForNewTable(tsdbReadHandleT queryHandle, STsdbQueryCond *pCond, STableGroupInfo* groupList) {
+void tsdbResetQueryHandleForNewTable(tsdbReaderT queryHandle, STsdbQueryCond *pCond, STableGroupInfo* groupList) {
   STsdbReadHandle* pTsdbReadHandle = queryHandle;
 
   pTsdbReadHandle->order       = pCond->order;
@@ -533,7 +533,7 @@ void tsdbResetQueryHandleForNewTable(tsdbReadHandleT queryHandle, STsdbQueryCond
 //  pTsdbReadHandle->next = doFreeColumnInfoData(pTsdbReadHandle->next);
 }
 
-tsdbReadHandleT tsdbQueryLastRow(STsdb *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList, uint64_t qId, uint64_t taskId) {
+tsdbReaderT tsdbQueryLastRow(STsdb *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList, uint64_t qId, uint64_t taskId) {
   pCond->twindow = updateLastrowForEachGroup(groupList);
 
   // no qualified table
@@ -561,7 +561,7 @@ tsdbReadHandleT tsdbQueryLastRow(STsdb *tsdb, STsdbQueryCond *pCond, STableGroup
 }
 
 #if 0
-tsdbReadHandleT tsdbQueryCacheLast(STsdb *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList, uint64_t qId, STsdbMemTable* pMemRef) {
+tsdbReaderT tsdbQueryCacheLast(STsdb *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList, uint64_t qId, STsdbMemTable* pMemRef) {
   STsdbReadHandle *pTsdbReadHandle = (STsdbReadHandle*) tsdbQueryTables(tsdb, pCond, groupList, qId, pMemRef);
   if (pTsdbReadHandle == NULL) {
     return NULL;
@@ -581,7 +581,7 @@ tsdbReadHandleT tsdbQueryCacheLast(STsdb *tsdb, STsdbQueryCond *pCond, STableGro
 }
 
 #endif
-SArray* tsdbGetQueriedTableList(tsdbReadHandleT *pHandle) {
+SArray* tsdbGetQueriedTableList(tsdbReaderT *pHandle) {
   assert(pHandle != NULL);
 
   STsdbReadHandle *pTsdbReadHandle = (STsdbReadHandle*) pHandle;
@@ -624,7 +624,7 @@ static STableGroupInfo* trimTableGroup(STimeWindow* window, STableGroupInfo* pGr
   return pNew;
 }
 
-tsdbReadHandleT tsdbQueryRowsInExternalWindow(STsdb *tsdb, STsdbQueryCond* pCond, STableGroupInfo *groupList, uint64_t qId, uint64_t taskId) {
+tsdbReaderT tsdbQueryRowsInExternalWindow(STsdb *tsdb, STsdbQueryCond* pCond, STableGroupInfo *groupList, uint64_t qId, uint64_t taskId) {
   STableGroupInfo* pNew = trimTableGroup(&pCond->twindow, groupList);
 
   if (pNew->numOfTables == 0) {
@@ -2338,7 +2338,7 @@ static void moveToNextDataBlockInCurrentFile(STsdbReadHandle* pTsdbReadHandle) {
   cur->blockCompleted = false;
 }
 #if 0
-int32_t tsdbGetFileBlocksDistInfo(tsdbReadHandleT* queryHandle, STableBlockDist* pTableBlockInfo) {
+int32_t tsdbGetFileBlocksDistInfo(tsdbReaderT* queryHandle, STableBlockDist* pTableBlockInfo) {
   STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*) queryHandle;
 
   pTableBlockInfo->totalSize = 0;
@@ -2494,7 +2494,7 @@ static bool doHasDataInBuffer(STsdbReadHandle* pTsdbReadHandle) {
 }
 
 //todo not unref yet, since it is not support multi-group interpolation query
-static UNUSED_FUNC void changeQueryHandleForInterpQuery(tsdbReadHandleT pHandle) {
+static UNUSED_FUNC void changeQueryHandleForInterpQuery(tsdbReaderT pHandle) {
   // filter the queried time stamp in the first place
   STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*) pHandle;
 
@@ -2635,7 +2635,7 @@ static bool  loadBlockOfActiveTable(STsdbReadHandle* pTsdbReadHandle) {
     }
 
     if (exists) {
-      tsdbRetrieveDataBlock((tsdbReadHandleT*) pTsdbReadHandle, NULL);
+      tsdbRetrieveDataBlock((tsdbReaderT*) pTsdbReadHandle, NULL);
       if (pTsdbReadHandle->currentLoadExternalRows && pTsdbReadHandle->window.skey == pTsdbReadHandle->window.ekey) {
         SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, 0);
         assert(*(int64_t*)pColInfo->pData == pTsdbReadHandle->window.skey);
@@ -2891,7 +2891,7 @@ static bool loadDataBlockFromTableSeq(STsdbReadHandle* pTsdbReadHandle) {
 }
 
 // handle data in cache situation
-bool tsdbNextDataBlock(tsdbReadHandleT pHandle) {
+bool tsdbNextDataBlock(tsdbReaderT pHandle) {
   STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*) pHandle;
 
   if (emptyQueryTimewindow(pTsdbReadHandle)) {
@@ -3055,7 +3055,7 @@ bool tsdbNextDataBlock(tsdbReadHandleT pHandle) {
 //  return terrno;
 //}
 
-bool tsdbGetExternalRow(tsdbReadHandleT pHandle) {
+bool tsdbGetExternalRow(tsdbReaderT pHandle) {
   STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*) pHandle;
   SQueryFilePos* cur = &pTsdbReadHandle->cur;
 
@@ -3112,7 +3112,7 @@ bool tsdbGetExternalRow(tsdbReadHandleT pHandle) {
 //  return code;
 //}
 
-bool isTsdbCacheLastRow(tsdbReadHandleT* pTsdbReadHandle) {
+bool isTsdbCacheLastRow(tsdbReaderT* pTsdbReadHandle) {
   return ((STsdbReadHandle *)pTsdbReadHandle)->cachelastrow > TSDB_CACHED_TYPE_NONE;
 }
 
@@ -3230,7 +3230,7 @@ STimeWindow updateLastrowForEachGroup(STableGroupInfo *groupList) {
   return window;
 }
 
-void tsdbRetrieveDataBlockInfo(tsdbReadHandleT* pTsdbReadHandle, SDataBlockInfo* pDataBlockInfo) {
+void tsdbRetrieveDataBlockInfo(tsdbReaderT* pTsdbReadHandle, SDataBlockInfo* pDataBlockInfo) {
   STsdbReadHandle* pHandle = (STsdbReadHandle*)pTsdbReadHandle;
   SQueryFilePos* cur = &pHandle->cur;
 
@@ -3254,7 +3254,7 @@ void tsdbRetrieveDataBlockInfo(tsdbReadHandleT* pTsdbReadHandle, SDataBlockInfo*
 /*
  * return null for mixed data block, if not a complete file data block, the statistics value will always return NULL
  */
-int32_t tsdbRetrieveDataBlockStatisInfo(tsdbReadHandleT* pTsdbReadHandle, SDataStatis** pBlockStatis) {
+int32_t tsdbRetrieveDataBlockStatisInfo(tsdbReaderT* pTsdbReadHandle, SDataStatis** pBlockStatis) {
   STsdbReadHandle* pHandle = (STsdbReadHandle*) pTsdbReadHandle;
 
   SQueryFilePos* c = &pHandle->cur;
@@ -3309,7 +3309,7 @@ int32_t tsdbRetrieveDataBlockStatisInfo(tsdbReadHandleT* pTsdbReadHandle, SDataS
   return TSDB_CODE_SUCCESS;
 }
 
-SArray* tsdbRetrieveDataBlock(tsdbReadHandleT* pTsdbReadHandle, SArray* pIdList) {
+SArray* tsdbRetrieveDataBlock(tsdbReaderT* pTsdbReadHandle, SArray* pIdList) {
   /**
    * In the following two cases, the data has been loaded to SColumnInfoData.
    * 1. data is from cache, 2. data block is not completed qualified to query time range
@@ -3820,7 +3820,7 @@ static void* destroyTableCheckInfo(SArray* pTableCheckInfo) {
 }
 
 
-void tsdbCleanupReadHandle(tsdbReadHandleT queryHandle) {
+void tsdbCleanupReadHandle(tsdbReaderT queryHandle) {
   STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)queryHandle;
   if (pTsdbReadHandle == NULL) {
     return;

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

@@ -26,12 +26,14 @@ int vnodeQueryOpen(SVnode *pVnode) {
 
 int vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {
   vTrace("message in query queue is processing");
+  SReadHandle handle = {.reader = pVnode->pTsdb, .meta = pVnode->pMeta};
 
   switch (pMsg->msgType) {
-    case TDMT_VND_QUERY:
-      return qWorkerProcessQueryMsg(pVnode->pTsdb, pVnode->pQuery, pMsg);
+    case TDMT_VND_QUERY:{
+      return qWorkerProcessQueryMsg(&handle, pVnode->pQuery, pMsg);
+    }
     case TDMT_VND_QUERY_CONTINUE:
-      return qWorkerProcessCQueryMsg(pVnode->pTsdb, pVnode->pQuery, pMsg);
+      return qWorkerProcessCQueryMsg(&handle, pVnode->pQuery, pMsg);
     default:
       vError("unknown msg type:%d in query queue", pMsg->msgType);
       return TSDB_CODE_VND_APP_ERROR;

source/libs/executor/inc/executorimpl.h

@@ -224,12 +224,12 @@ typedef struct STaskAttr {
 //  SFilterInfo     *pFilters;
   
   void*            tsdb;
-//  SMemRef          memRef;
   STableGroupInfo  tableGroupInfo;       // table <tid, last_key> list  SArray<STableKeyInfo>
   int32_t          vgId;
   SArray          *pUdfInfo;             // no need to free
 } STaskAttr;
 
+typedef int32_t (*__optr_prepare_fn_t)(void* param);
 typedef SSDataBlock* (*__operator_fn_t)(void* param, bool* newgroup);
 typedef void (*__optr_cleanup_fn_t)(void* param, int32_t num);
 
@@ -313,8 +313,9 @@ typedef struct SOperatorInfo {
 
   struct SOperatorInfo **pDownstream;  // downstram pointer list
   int32_t               numOfDownstream; // number of downstream. The value is always ONE expect for join operator
+  __optr_prepare_fn_t   prepareFn;
   __operator_fn_t       exec;
-  __optr_cleanup_fn_t   cleanup;
+  __optr_cleanup_fn_t   cleanupFn;
 } SOperatorInfo;
 
 enum {
@@ -395,7 +396,7 @@ typedef struct STableScanInfo {
   int32_t         current;
   int32_t         reverseTimes; // 0 by default
 
-  SQLFunctionCtx *pCtx;         // next operator query context
+  SqlFunctionCtx *pCtx;         // next operator query context
   SResultRowInfo *pResultRowInfo;
   int32_t        *rowCellInfoOffset;
   SExprInfo      *pExpr;
@@ -425,7 +426,7 @@ typedef struct SStreamBlockScanInfo {
 typedef struct SOptrBasicInfo {
   SResultRowInfo    resultRowInfo;
   int32_t          *rowCellInfoOffset;  // offset value for each row result cell info
-  SQLFunctionCtx   *pCtx;
+  SqlFunctionCtx   *pCtx;
   SSDataBlock      *pRes;
 } SOptrBasicInfo;
 
@@ -564,7 +565,6 @@ typedef struct SOrderOperatorInfo {
 SOperatorInfo* createExchangeOperatorInfo(const SArray* pSources, const SArray* pSchema, SExecTaskInfo* pTaskInfo);
 
 SOperatorInfo* createDataBlocksOptScanInfo(void* pTsdbReadHandle, int32_t order, int32_t numOfOutput, int32_t repeatTime, int32_t reverseTime, SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createTableScanOperatorInfo(void* pTsdbReadHandle, int32_t order, int32_t numOfOutput, int32_t repeatTime, SExecTaskInfo* pTaskInfo);
 SOperatorInfo* createTableSeqScanOperator(void* pTsdbReadHandle, STaskRuntimeEnv* pRuntimeEnv);
 SOperatorInfo* createSubmitBlockScanOperatorInfo(void *pSubmitBlockReadHandle, int32_t numOfOutput, SExecTaskInfo* pTaskInfo);
 
@@ -607,11 +607,11 @@ SSDataBlock* createOutputBuf(SExprInfo* pExpr, int32_t numOfOutput, int32_t numO
 void* destroyOutputBuf(SSDataBlock* pBlock);
 void* doDestroyFilterInfo(SSingleColumnFilterInfo* pFilterInfo, int32_t numOfFilterCols);
 
-void setInputDataBlock(SOperatorInfo* pOperator, SQLFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order);
-void finalizeQueryResult(SOperatorInfo* pOperator, SQLFunctionCtx* pCtx, SResultRowInfo* pResultRowInfo, int32_t* rowCellInfoOffset);
+void setInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order);
+void finalizeQueryResult(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SResultRowInfo* pResultRowInfo, int32_t* rowCellInfoOffset);
 void updateOutputBuf(SOptrBasicInfo* pBInfo, int32_t *bufCapacity, int32_t numOfInputRows);
 void clearOutputBuf(SOptrBasicInfo* pBInfo, int32_t *bufCapacity);
-void copyTsColoum(SSDataBlock* pRes, SQLFunctionCtx* pCtx, int32_t numOfOutput);
+void copyTsColoum(SSDataBlock* pRes, SqlFunctionCtx* pCtx, int32_t numOfOutput);
 
 void freeParam(STaskParam *param);
 int32_t createQueryFunc(SQueriedTableInfo* pTableInfo, int32_t numOfOutput, SExprInfo** pExprInfo,
@@ -659,8 +659,8 @@ void freeQueryAttr(STaskAttr *pQuery);
 
 int32_t getMaximumIdleDurationSec();
 
-void doInvokeUdf(struct SUdfInfo* pUdfInfo, SQLFunctionCtx *pCtx, int32_t idx, int32_t type);
+void doInvokeUdf(struct SUdfInfo* pUdfInfo, SqlFunctionCtx *pCtx, int32_t idx, int32_t type);
 void setTaskStatus(SExecTaskInfo *pTaskInfo, int8_t status);
-int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, void* readerHandle, uint64_t taskId);
+int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId);
 
 #endif  // TDENGINE_EXECUTORIMPL_H

source/libs/executor/src/executor.c

@@ -39,7 +39,7 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, char* id)
   }
 }
 
-int32_t qSetStreamInput(qTaskInfo_t tinfo, void* input) {
+int32_t qSetStreamInput(qTaskInfo_t tinfo, const void* input) {
   if (tinfo == NULL) {
     return TSDB_CODE_QRY_APP_ERROR;
   }
@@ -50,7 +50,7 @@ int32_t qSetStreamInput(qTaskInfo_t tinfo, void* input) {
 
   SExecTaskInfo* pTaskInfo = (SExecTaskInfo*) tinfo;
 
-  int32_t code = doSetStreamBlock(pTaskInfo->pRoot, input, GET_TASKID(pTaskInfo));
+  int32_t code = doSetStreamBlock(pTaskInfo->pRoot, (void*) input, GET_TASKID(pTaskInfo));
   if (code != TSDB_CODE_SUCCESS) {
     qError("%s failed to set the stream block data", GET_TASKID(pTaskInfo));
   } else {

source/libs/executor/src/executorMain.c

@@ -51,25 +51,7 @@ static void freeqinfoFn(void *qhandle) {
   qDestroyTask(*handle);
 }
 
-void freeParam(STaskParam *param) {
-  tfree(param->sql);
-  tfree(param->tagCond);
-  tfree(param->tbnameCond);
-  tfree(param->pTableIdList);
-  taosArrayDestroy(param->pOperator);
-  tfree(param->pExprs);
-  tfree(param->pSecExprs);
-
-  tfree(param->pExpr);
-  tfree(param->pSecExpr);
-
-  tfree(param->pGroupColIndex);
-  tfree(param->pTagColumnInfo);
-  tfree(param->pGroupbyExpr);
-  tfree(param->prevResult);
-}
-
-int32_t qCreateExecTask(void* readHandle, int32_t vgId, uint64_t taskId, SSubplan* pSubplan, qTaskInfo_t* pTaskInfo, DataSinkHandle* handle) {
+int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId, SSubplan* pSubplan, qTaskInfo_t* pTaskInfo, DataSinkHandle* handle) {
   assert(readHandle != NULL && pSubplan != NULL);
   SExecTaskInfo** pTask = (SExecTaskInfo**)pTaskInfo;
 

source/libs/function/inc/taggfunction.h

@@ -73,7 +73,7 @@ typedef struct STwaInfo {
 
 extern int32_t functionCompatList[]; // compatible check array list
 
-bool topbot_datablock_filter(SQLFunctionCtx *pCtx, const char *minval, const char *maxval);
+bool topbot_datablock_filter(SqlFunctionCtx *pCtx, const char *minval, const char *maxval);
 
 /**
  * the numOfRes should be kept, since it may be used later

source/libs/function/src/tudf.c

@@ -121,7 +121,7 @@ void destroyUdfInfo(SUdfInfo* pUdfInfo) {
   tfree(pUdfInfo);
 }
 
-void doInvokeUdf(struct SUdfInfo* pUdfInfo, SQLFunctionCtx *pCtx, int32_t idx, int32_t type) {
+void doInvokeUdf(struct SUdfInfo* pUdfInfo, SqlFunctionCtx *pCtx, int32_t idx, int32_t type) {
   int32_t output = 0;
 
   if (pUdfInfo == NULL || pUdfInfo->funcs[type] == NULL) {

source/libs/planner/src/physicalPlan.c

@@ -287,7 +287,7 @@ static bool needMultiNodeScan(SQueryTableInfo* pTable) {
 static SPhyNode* createSingleTableScanNode(SQueryPlanNode* pPlanNode, SQueryTableInfo* pTableInfo, SSubplan* subplan) {
   SVgroupsInfo* pVgroupsInfo = pTableInfo->pMeta->vgroupList;
   vgroupInfoToNodeAddr(&(pVgroupsInfo->vgroups[0]), &subplan->execNode);
-  int32_t type = (pPlanNode->info.type == QNODE_TABLESCAN)? OP_TableScan:OP_StreamScan;
+  int32_t type = (pPlanNode->info.type == QNODE_TABLESCAN)? OP_DataBlocksOptScan:OP_StreamScan;
   return createUserTableScanNode(pPlanNode, pTableInfo, type);
 }
 

source/libs/planner/src/physicalPlanJson.c

@@ -88,7 +88,6 @@ static const char* jkPnodeType = "Type";
 static int32_t getPnodeTypeSize(cJSON* json) {
   switch (getNumber(json, jkPnodeType)) {
     case OP_StreamScan:
-    case OP_TableScan:
     case OP_DataBlocksOptScan:
     case OP_TableSeqScan:
       return sizeof(STableScanPhyNode);
@@ -831,7 +830,6 @@ static bool specificPhyNodeToJson(const void* obj, cJSON* json) {
   const SPhyNode* phyNode = (const SPhyNode*)obj;
   switch (phyNode->info.type) {
     case OP_StreamScan:
-    case OP_TableScan:
     case OP_DataBlocksOptScan:
     case OP_TableSeqScan:
       return tableScanNodeToJson(obj, json);
@@ -869,7 +867,6 @@ static bool specificPhyNodeToJson(const void* obj, cJSON* json) {
 static bool specificPhyNodeFromJson(const cJSON* json, void* obj) {
   SPhyNode* phyNode = (SPhyNode*)obj;
   switch (phyNode->info.type) {
-    case OP_TableScan:
     case OP_StreamScan:
     case OP_DataBlocksOptScan:
     case OP_TableSeqScan: