refactor: do some internal refactor.

include/common/tdatablock.h

@@ -215,7 +215,7 @@ size_t blockDataGetSerialMetaSize(uint32_t numOfCols);
 int32_t blockDataSort(SSDataBlock* pDataBlock, SArray* pOrderInfo);
 int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirst);
 
-int32_t colInfoDataEnsureCapacity(SColumnInfoData* pColumn, uint32_t numOfRows);
+int32_t colInfoDataEnsureCapacity(SColumnInfoData* pColumn, uint32_t numOfRows, bool clearPayload);
 int32_t blockDataEnsureCapacity(SSDataBlock* pDataBlock, uint32_t numOfRows);
 
 void colInfoDataCleanup(SColumnInfoData* pColumn, uint32_t numOfRows);

include/common/tdataformat.h

@@ -241,13 +241,6 @@ struct STag {
     memcpy(varDataVal(x), (str), __len);          \
   } while (0);
 
-#define STR_TO_NET_VARSTR(x, str)                 \
-  do {                                            \
-    VarDataLenT __len = (VarDataLenT)strlen(str); \
-    *(VarDataLenT *)(x) = htons(__len);           \
-    memcpy(varDataVal(x), (str), __len);          \
-  } while (0);
-
 #define STR_WITH_MAXSIZE_TO_VARSTR(x, str, _maxs)                         \
   do {                                                                    \
     char *_e = stpncpy(varDataVal(x), (str), (_maxs)-VARSTR_HEADER_SIZE); \

include/libs/function/function.h

@@ -182,7 +182,7 @@ struct SScalarParam {
 };
 
 void    cleanupResultRowEntry(struct SResultRowEntryInfo *pCell);
-int32_t getNumOfResult(SqlFunctionCtx *pCtx, int32_t num, SSDataBlock *pResBlock);
+//int32_t getNumOfResult(SqlFunctionCtx *pCtx, int32_t num, SSDataBlock *pResBlock);
 bool    isRowEntryCompleted(struct SResultRowEntryInfo *pEntry);
 bool    isRowEntryInitialized(struct SResultRowEntryInfo *pEntry);
 

include/libs/function/functionMgt.h

@@ -224,7 +224,7 @@ int32_t fmGetDistMethod(const SFunctionNode* pFunc, SFunctionNode** pPartialFunc
 
 typedef enum EFuncDataRequired {
   FUNC_DATA_REQUIRED_DATA_LOAD = 1,
-  FUNC_DATA_REQUIRED_STATIS_LOAD,
+  FUNC_DATA_REQUIRED_SMA_LOAD,
   FUNC_DATA_REQUIRED_NOT_LOAD,
   FUNC_DATA_REQUIRED_FILTEROUT,
 } EFuncDataRequired;

source/common/src/tdatablock.c

@@ -1143,7 +1143,7 @@ void blockDataCleanup(SSDataBlock* pDataBlock) {
   }
 }
 
-static int32_t doEnsureCapacity(SColumnInfoData* pColumn, const SDataBlockInfo* pBlockInfo, uint32_t numOfRows) {
+static int32_t doEnsureCapacity(SColumnInfoData* pColumn, const SDataBlockInfo* pBlockInfo, uint32_t numOfRows, bool clearPayload) {
   ASSERT(numOfRows > 0 && pBlockInfo->capacity >= pBlockInfo->rows);
   if (numOfRows < pBlockInfo->capacity) {
     return TSDB_CODE_SUCCESS;
@@ -1182,8 +1182,10 @@ static int32_t doEnsureCapacity(SColumnInfoData* pColumn, const SDataBlockInfo*
       return TSDB_CODE_OUT_OF_MEMORY;
     }
 
-    memset(tmp + pColumn->info.bytes * existedRows, 0, pColumn->info.bytes * (numOfRows - existedRows));
     pColumn->pData = tmp;
+    if (clearPayload) {
+      memset(tmp + pColumn->info.bytes * existedRows, 0, pColumn->info.bytes * (numOfRows - existedRows));
+    }
   }
 
   return TSDB_CODE_SUCCESS;
@@ -1203,9 +1205,9 @@ void colInfoDataCleanup(SColumnInfoData* pColumn, uint32_t numOfRows) {
   }
 }
 
-int32_t colInfoDataEnsureCapacity(SColumnInfoData* pColumn, uint32_t numOfRows) {
+int32_t colInfoDataEnsureCapacity(SColumnInfoData* pColumn, uint32_t numOfRows, bool clearPayload) {
   SDataBlockInfo info = {0};
-  return doEnsureCapacity(pColumn, &info, numOfRows);
+  return doEnsureCapacity(pColumn, &info, numOfRows, clearPayload);
 }
 
 int32_t blockDataEnsureCapacity(SSDataBlock* pDataBlock, uint32_t numOfRows) {
@@ -1221,7 +1223,7 @@ int32_t blockDataEnsureCapacity(SSDataBlock* pDataBlock, uint32_t numOfRows) {
   size_t numOfCols = taosArrayGetSize(pDataBlock->pDataBlock);
   for (int32_t i = 0; i < numOfCols; ++i) {
     SColumnInfoData* p = taosArrayGet(pDataBlock->pDataBlock, i);
-    code = doEnsureCapacity(p, &pDataBlock->info, numOfRows);
+    code = doEnsureCapacity(p, &pDataBlock->info, numOfRows, false);
     if (code) {
       return code;
     }

source/dnode/vnode/src/inc/tsdb.h

@@ -644,6 +644,13 @@ typedef struct SSttBlockLoadInfo {
   int16_t   *colIds;
   int32_t    numOfCols;
   bool       sttBlockLoaded;
+
+  // keep the last access position, this position may be used to reduce the binary times for
+  // starting last block data for a new table
+  struct {
+    int32_t blockIndex;
+    int32_t rowIndex;
+  } prevEndPos;
 } SSttBlockLoadInfo;
 
 typedef struct SMergeTree {

source/libs/executor/inc/executorimpl.h

@@ -536,10 +536,10 @@ typedef struct SSysTableScanInfo {
 } SSysTableScanInfo;
 
 typedef struct SBlockDistInfo {
-  SSDataBlock* pResBlock;
-  STsdbReader* pHandle;
-  SReadHandle  readHandle;
-  uint64_t     uid;  // table uid
+  SSDataBlock*   pResBlock;
+  STsdbReader*   pHandle;
+  SReadHandle    readHandle;
+  uint64_t       uid;  // table uid
 } SBlockDistInfo;
 
 // todo remove this
@@ -550,7 +550,6 @@ typedef struct SOptrBasicInfo {
 } SOptrBasicInfo;
 
 typedef struct SIntervalAggOperatorInfo {
-  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
   SOptrBasicInfo     binfo;              // basic info
   SAggSupporter      aggSup;             // aggregate supporter
   SExprSupp          scalarSupp;         // supporter for perform scalar function
@@ -571,7 +570,6 @@ typedef struct SIntervalAggOperatorInfo {
 typedef struct SMergeAlignedIntervalAggOperatorInfo {
   SIntervalAggOperatorInfo* intervalAggOperatorInfo;
 
-  //  bool         hasGroupId;
   uint64_t     groupId;  // current groupId
   int64_t      curTs;    // current ts
   SSDataBlock* prefetchedBlock;
@@ -839,10 +837,6 @@ typedef struct SSortOperatorInfo {
   SNode*         pCondition;
 } SSortOperatorInfo;
 
-typedef struct STagFilterOperatorInfo {
-  SOptrBasicInfo binfo;
-} STagFilterOperatorInfo;
-
 typedef struct SJoinOperatorInfo {
   SSDataBlock* pRes;
   int32_t      joinType;
@@ -907,7 +901,7 @@ int32_t addTagPseudoColumnData(SReadHandle* pHandle, SExprInfo* pPseudoExpr, int
 
 void cleanupAggSup(SAggSupporter* pAggSup);
 void appendOneRowToDataBlock(SSDataBlock* pBlock, STupleHandle* pTupleHandle);
-void setTbNameColData(void* pMeta, const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId);
+void setTbNameColData(const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId, const char* name);
 
 int32_t doPrepareScan(SOperatorInfo* pOperator, uint64_t uid, int64_t ts);
 int32_t doGetScanStatus(SOperatorInfo* pOperator, uint64_t* uid, int64_t* ts);

source/libs/executor/src/cachescanoperator.c

@@ -170,14 +170,12 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
       pRes->info.uid = *(tb_uid_t*)taosArrayGet(pInfo->pUidList, pInfo->indexOfBufferedRes);
       pRes->info.rows = 1;
 
-      if (pInfo->pseudoExprSup.numOfExprs > 0) {
-        SExprSupp* pSup = &pInfo->pseudoExprSup;
-        int32_t    code = addTagPseudoColumnData(&pInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pRes, pRes->info.rows,
-                                                 GET_TASKID(pTaskInfo));
-        if (code != TSDB_CODE_SUCCESS) {
-          pTaskInfo->code = code;
-          return NULL;
-        }
+      SExprSupp* pSup = &pInfo->pseudoExprSup;
+      int32_t code = addTagPseudoColumnData(&pInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pRes,
+                                            pRes->info.rows, GET_TASKID(pTaskInfo));
+      if (code != TSDB_CODE_SUCCESS) {
+        pTaskInfo->code = code;
+        return NULL;
       }
 
       pRes->info.groupId = getTableGroupId(pTableList, pRes->info.uid);

source/libs/executor/src/executil.c

@@ -379,7 +379,7 @@ static int32_t createResultData(SDataType* pType, int32_t numOfRows, SScalarPara
   pColumnData->info.scale = pType->scale;
   pColumnData->info.precision = pType->precision;
 
-  int32_t code = colInfoDataEnsureCapacity(pColumnData, numOfRows);
+  int32_t code = colInfoDataEnsureCapacity(pColumnData, numOfRows, true);
   if (code != TSDB_CODE_SUCCESS) {
     terrno = code;
     taosMemoryFree(pColumnData);
@@ -1828,8 +1828,6 @@ static int32_t orderbyGroupIdComparFn(const void* p1, const void* p2) {
 }
 
 static int32_t sortTableGroup(STableListInfo* pTableListInfo) {
-  int32_t code = TSDB_CODE_SUCCESS;
-
   taosArraySort(pTableListInfo->pTableList, orderbyGroupIdComparFn);
   int32_t size = taosArrayGetSize(pTableListInfo->pTableList);
 
@@ -1851,6 +1849,11 @@ static int32_t sortTableGroup(STableListInfo* pTableListInfo) {
 
   pTableListInfo->numOfOuputGroups = taosArrayGetSize(pList);
   pTableListInfo->groupOffset = taosMemoryMalloc(sizeof(int32_t) * pTableListInfo->numOfOuputGroups);
+  if (pTableListInfo->groupOffset == NULL) {
+    taosArrayDestroy(pList);
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
   memcpy(pTableListInfo->groupOffset, taosArrayGet(pList, 0), sizeof(int32_t) * pTableListInfo->numOfOuputGroups);
   taosArrayDestroy(pList);
   return TDB_CODE_SUCCESS;

source/libs/executor/src/executorimpl.c

@@ -87,7 +87,7 @@ static void releaseQueryBuf(size_t numOfTables);
 
 static void destroyFillOperatorInfo(void* param);
 static void destroyProjectOperatorInfo(void* param);
-static void destroyOrderOperatorInfo(void* param);
+static void destroySortOperatorInfo(void* param);
 static void destroyAggOperatorInfo(void* param);
 
 static void destroyIntervalOperatorInfo(void* param);
@@ -322,7 +322,7 @@ void initExecTimeWindowInfo(SColumnInfoData* pColData, STimeWindow* pQueryWindow
   pColData->info.type = TSDB_DATA_TYPE_TIMESTAMP;
   pColData->info.bytes = sizeof(int64_t);
 
-  colInfoDataEnsureCapacity(pColData, 5);
+  colInfoDataEnsureCapacity(pColData, 5, false);
   colDataAppendInt64(pColData, 0, &pQueryWindow->skey);
   colDataAppendInt64(pColData, 1, &pQueryWindow->ekey);
 
@@ -439,7 +439,7 @@ static int32_t doCreateConstantValColumnInfo(SInputColumnInfoData* pInput, SFunc
     pColInfo = pInput->pData[paramIndex];
   }
 
-  colInfoDataEnsureCapacity(pColInfo, numOfRows);
+  colInfoDataEnsureCapacity(pColInfo, numOfRows, false);
 
   int8_t type = pFuncParam->param.nType;
   if (type == TSDB_DATA_TYPE_BIGINT || type == TSDB_DATA_TYPE_UBIGINT) {
@@ -1047,8 +1047,6 @@ void doFilter(const SNode* pFilterNode, SSDataBlock* pBlock, SColMatchInfo* pCol
   SFilterInfo* filter = pFilterInfo;
   int64_t      st = taosGetTimestampUs();
 
-  //  pError("start filter");
-
   // todo move to the initialization function
   int32_t code = 0;
   bool    needFree = false;

source/libs/executor/src/sortoperator.c

@@ -20,7 +20,7 @@ static SSDataBlock* doSort(SOperatorInfo* pOperator);
 static int32_t      doOpenSortOperator(SOperatorInfo* pOperator);
 static int32_t      getExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len);
 
-static void destroyOrderOperatorInfo(void* param);
+static void destroySortOperatorInfo(void* param);
 
 // todo add limit/offset impl
 SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pSortNode, SExecTaskInfo* pTaskInfo) {
@@ -64,7 +64,7 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode*
   // TODO dynamic set the available sort buffer
 
   pOperator->fpSet =
-      createOperatorFpSet(doOpenSortOperator, doSort, NULL, NULL, destroyOrderOperatorInfo, getExplainExecInfo);
+      createOperatorFpSet(doOpenSortOperator, doSort, NULL, NULL, destroySortOperatorInfo, getExplainExecInfo);
 
   code = appendDownstream(pOperator, &downstream, 1);
   if (code != TSDB_CODE_SUCCESS) {
@@ -249,7 +249,7 @@ SSDataBlock* doSort(SOperatorInfo* pOperator) {
   return blockDataGetNumOfRows(pBlock) > 0 ? pBlock : NULL;
 }
 
-void destroyOrderOperatorInfo(void* param) {
+void destroySortOperatorInfo(void* param) {
   SSortOperatorInfo* pInfo = (SSortOperatorInfo*)param;
   pInfo->binfo.pRes = blockDataDestroy(pInfo->binfo.pRes);
 
@@ -729,12 +729,11 @@ SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** downStreams, size
 
   initLimitInfo(pMergePhyNode->node.pLimit, pMergePhyNode->node.pSlimit, &pInfo->limitInfo);
   pInfo->binfo.pRes = createResDataBlock(pDescNode);
+
   int32_t rowSize = pInfo->binfo.pRes->info.rowSize;
   ASSERT(rowSize < 100 * 1024 * 1024);
 
-  SArray* pSortInfo = createSortInfo(pMergePhyNode->pMergeKeys);
   int32_t numOfOutputCols = 0;
-
   code = extractColMatchInfo(pMergePhyNode->pTargets, pDescNode, &numOfOutputCols, COL_MATCH_FROM_SLOT_ID,
                              &pInfo->matchInfo);
   if (code != TSDB_CODE_SUCCESS) {
@@ -746,7 +745,7 @@ SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** downStreams, size
   initResultSizeInfo(&pOperator->resultInfo, 1024);
 
   pInfo->groupSort = pMergePhyNode->groupSort;
-  pInfo->pSortInfo = pSortInfo;
+  pInfo->pSortInfo = createSortInfo(pMergePhyNode->pMergeKeys);
   pInfo->pInputBlock = pInputBlock;
   pInfo->bufPageSize = getProperSortPageSize(rowSize);
   pInfo->sortBufSize = pInfo->bufPageSize * (numStreams + 1);  // one additional is reserved for merged result.
@@ -768,11 +767,11 @@ SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** downStreams, size
   return pOperator;
 
 _error:
-  pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;
   if (pInfo != NULL) {
     destroyMultiwayMergeOperatorInfo(pInfo);
   }
 
+  pTaskInfo->code = code;
   taosMemoryFree(pOperator);
   return NULL;
 }

source/libs/function/src/builtinsimpl.c

@@ -489,7 +489,7 @@ EFuncDataRequired countDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWind
   if (QUERY_NODE_COLUMN == nodeType(pParam) && PRIMARYKEY_TIMESTAMP_COL_ID == ((SColumnNode*)pParam)->colId) {
     return FUNC_DATA_REQUIRED_NOT_LOAD;
   }
-  return FUNC_DATA_REQUIRED_STATIS_LOAD;
+  return FUNC_DATA_REQUIRED_SMA_LOAD;
 }
 
 bool getCountFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
@@ -1103,7 +1103,7 @@ int32_t avgPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
 }
 
 EFuncDataRequired statisDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWindow) {
-  return FUNC_DATA_REQUIRED_STATIS_LOAD;
+  return FUNC_DATA_REQUIRED_SMA_LOAD;
 }
 
 bool minmaxFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo) {

source/libs/planner/src/planOptimizer.c

@@ -279,7 +279,7 @@ static EFuncDataRequired scanPathOptPromoteDataRequired(EFuncDataRequired l, EFu
   switch (l) {
     case FUNC_DATA_REQUIRED_DATA_LOAD:
       return l;
-    case FUNC_DATA_REQUIRED_STATIS_LOAD:
+    case FUNC_DATA_REQUIRED_SMA_LOAD:
       return FUNC_DATA_REQUIRED_DATA_LOAD == r ? r : l;
     case FUNC_DATA_REQUIRED_NOT_LOAD:
       return FUNC_DATA_REQUIRED_FILTEROUT == r ? l : r;

source/libs/scalar/CMakeLists.txt

@@ -14,7 +14,6 @@ target_link_libraries(scalar
     PRIVATE nodes
     PRIVATE function
     PRIVATE qcom
-    PRIVATE vnode
 )
 
 if(${BUILD_TEST})

source/libs/scalar/src/scalar.c

@@ -49,7 +49,7 @@ int32_t sclCreateColumnInfoData(SDataType *pType, int32_t numOfRows, SScalarPara
   pColumnData->info.scale = pType->scale;
   pColumnData->info.precision = pType->precision;
 
-  int32_t code = colInfoDataEnsureCapacity(pColumnData, numOfRows);
+  int32_t code = colInfoDataEnsureCapacity(pColumnData, numOfRows, true);
   if (code != TSDB_CODE_SUCCESS) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     taosMemoryFree(pColumnData);
@@ -70,7 +70,7 @@ int32_t sclConvertValueToSclParam(SValueNode* pValueNode, SScalarParam* out, int
 
   colDataAppend(in.columnData, 0, nodesGetValueFromNode(pValueNode), false);
 
-  colInfoDataEnsureCapacity(out->columnData, 1);
+  colInfoDataEnsureCapacity(out->columnData, 1, true);
   code = vectorConvertSingleColImpl(&in, out, overflow, -1, -1);
   sclFreeParam(&in);
 
@@ -88,7 +88,7 @@ int32_t sclExtendResRows(SScalarParam *pDst, SScalarParam *pSrc, SArray *pBlockL
   pLeft->numOfRows = pb->info.rows;
 
   if (pDst->numOfRows < pb->info.rows) {
-    colInfoDataEnsureCapacity(pDst->columnData, pb->info.rows);
+    colInfoDataEnsureCapacity(pDst->columnData, pb->info.rows, true);
   }
   
   _bin_scalar_fn_t OperatorFn = getBinScalarOperatorFn(OP_TYPE_ASSIGN);
@@ -1604,7 +1604,7 @@ int32_t scalarCalculate(SNode *pNode, SArray *pBlockList, SScalarParam *pDst) {
     if (1 == res->numOfRows) {
       SCL_ERR_JRET(sclExtendResRows(pDst, res, pBlockList));
     } else {
-      colInfoDataEnsureCapacity(pDst->columnData, res->numOfRows);
+      colInfoDataEnsureCapacity(pDst->columnData, res->numOfRows, true);
       colDataAssign(pDst->columnData, res->columnData, res->numOfRows, NULL);
       pDst->numOfRows = res->numOfRows;
       pDst->numOfQualified = res->numOfQualified;

source/libs/scalar/src/sclfunc.c

@@ -6,7 +6,6 @@
 #include "tdatablock.h"
 #include "tjson.h"
 #include "ttime.h"
-#include "vnode.h"
 
 typedef float (*_float_fn)(float);
 typedef double (*_double_fn)(double);
@@ -1718,12 +1717,9 @@ int32_t winEndTsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
 
 int32_t qTbnameFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
   ASSERT(inputNum == 1);
+  char* p = colDataGetVarData(pInput->columnData, 0);
 
-  uint64_t uid = *(uint64_t *)colDataGetData(pInput->columnData, 0);
-
-  char str[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
-  metaGetTableNameByUid(pInput->param, uid, str);
-  colDataAppendNItems(pOutput->columnData, pOutput->numOfRows, str, pInput->numOfRows);
+  colDataAppendNItems(pOutput->columnData, pOutput->numOfRows, p, pInput->numOfRows);
   pOutput->numOfRows += pInput->numOfRows;
   return TSDB_CODE_SUCCESS;
 }