[td-13039] refactor.

include/common/tcommon.h

@@ -193,20 +193,19 @@ typedef struct SColumn {
   uint8_t scale;
 } SColumn;
 
-typedef struct SLimit {
-  int64_t limit;
-  int64_t offset;
-} SLimit;
-
-typedef struct SOrder {
-  uint32_t order;
-  SColumn  col;
-} SOrder;
-
-typedef struct SGroupbyExpr {
-  SArray* columnInfo;  // SArray<SColIndex>, group by columns information
-  bool    groupbyTag;  // group by tag or column
-} SGroupbyExpr;
+typedef struct STableBlockDistInfo {
+  uint16_t  rowSize;
+  uint16_t  numOfFiles;
+  uint32_t  numOfTables;
+  uint64_t  totalSize;
+  uint64_t  totalRows;
+  int32_t   maxRows;
+  int32_t   minRows;
+  int32_t   firstSeekTimeUs;
+  uint32_t  numOfRowsInMemTable;
+  uint32_t  numOfSmallBlocks;
+  SArray   *dataBlockInfos;
+} STableBlockDistInfo;
 
 enum {
   FUNC_PARAM_TYPE_VALUE = 0x1,
@@ -241,15 +240,6 @@ typedef struct SExprInfo {
   struct tExprNode*     pExpr;
 } SExprInfo;
 
-typedef struct SStateWindow {
-  SColumn col;
-} SStateWindow;
-
-typedef struct SSessionWindow {
-  int64_t gap;  // gap between two session window(in microseconds)
-  SColumn col;
-} SSessionWindow;
-
 #define QUERY_ASC_FORWARD_STEP 1
 #define QUERY_DESC_FORWARD_STEP -1
 

include/libs/function/function.h

@@ -52,6 +52,11 @@ typedef struct SFuncExecFuncs {
   FExecFinalize finalize;
 } SFuncExecFuncs;
 
+typedef struct SFileBlockInfo {
+  int32_t numBlocksOfStep;
+} SFileBlockInfo;
+
+#define TSDB_BLOCK_DIST_STEP_ROWS 8
 #define MAX_INTERVAL_TIME_WINDOW  1000000  // maximum allowed time windows in final results
 
 #define FUNCTION_TYPE_SCALAR       1
@@ -101,10 +106,6 @@ typedef struct SFuncExecFuncs {
 #define FUNCTION_DERIVATIVE   32
 #define FUNCTION_BLKINFO      33
 
-#define FUNCTION_HISTOGRAM    34
-#define FUNCTION_HLL          35
-#define FUNCTION_MODE         36
-#define FUNCTION_SAMPLE       37
 
 #define FUNCTION_COV          38
 

source/dnode/vnode/inc/tsdb.h

@@ -171,6 +171,8 @@ tsdbReaderT *tsdbQueryTables(STsdb *tsdb, STsdbQueryCond *pCond, STableGroupInfo
 
 tsdbReaderT tsdbQueryCacheLast(STsdb *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList, uint64_t qId, void* pMemRef);
 
+int32_t tsdbGetFileBlocksDistInfo(tsdbReaderT* queryHandle, STableBlockDistInfo* pTableBlockInfo);
+
 bool isTsdbCacheLastRow(tsdbReaderT* pTsdbReadHandle);
 
 /**

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

@@ -14,7 +14,7 @@
  */
 
 #include "tsdbDef.h"
-#include <tdatablock.h>
+#include "tdatablock.h"
 #include "os.h"
 #include "talgo.h"
 #include "tcompare.h"
@@ -31,6 +31,7 @@
 #include "tlosertree.h"
 #include "tsdbDef.h"
 #include "tmsg.h"
+#include "tsdbCommit.h"
 
 #define EXTRA_BYTES 2
 #define ASCENDING_TRAVERSE(o)   (o == TSDB_ORDER_ASC)
@@ -209,34 +210,34 @@ static SArray* getDefaultLoadColumns(STsdbReadHandle* pTsdbReadHandle, bool load
   return pLocalIdList;
 }
 
-//int64_t tsdbGetNumOfRowsInMemTable(tsdbReaderT* pHandle) {
-//  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*) pHandle;
-//
-//  int64_t rows = 0;
-//  STsdbMemTable* pMemTable = pTsdbReadHandle->pMemTable;
-//  if (pMemTable == NULL) { return rows; }
-//
-////  STableData* pMem  = NULL;
-////  STableData* pIMem = NULL;
-//
-////  SMemTable* pMemT = pMemRef->snapshot.mem;
-////  SMemTable* pIMemT = pMemRef->snapshot.imem;
-//
-//  size_t size = taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
-//  for (int32_t i = 0; i < size; ++i) {
-//    STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, i);
-//
-////    if (pMemT && pCheckInfo->tableId < pMemT->maxTables) {
-////      pMem = pMemT->tData[pCheckInfo->tableId];
-////      rows += (pMem && pMem->uid == pCheckInfo->tableId) ? pMem->numOfRows : 0;
-////    }
-////    if (pIMemT && pCheckInfo->tableId < pIMemT->maxTables) {
-////      pIMem = pIMemT->tData[pCheckInfo->tableId];
-////      rows += (pIMem && pIMem->uid == pCheckInfo->tableId) ? pIMem->numOfRows : 0;
-////    }
+int64_t tsdbGetNumOfRowsInMemTable(tsdbReaderT* pHandle) {
+  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*) pHandle;
+
+  int64_t rows = 0;
+  STsdbMemTable* pMemTable = NULL;//pTsdbReadHandle->pMemTable;
+  if (pMemTable == NULL) { return rows; }
+
+//  STableData* pMem  = NULL;
+//  STableData* pIMem = NULL;
+
+//  SMemTable* pMemT = pMemRef->snapshot.mem;
+//  SMemTable* pIMemT = pMemRef->snapshot.imem;
+
+  size_t size = taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
+  for (int32_t i = 0; i < size; ++i) {
+    STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, i);
+
+//    if (pMemT && pCheckInfo->tableId < pMemT->maxTables) {
+//      pMem = pMemT->tData[pCheckInfo->tableId];
+//      rows += (pMem && pMem->uid == pCheckInfo->tableId) ? pMem->numOfRows : 0;
 //    }
-//  return rows;
-//}
+//    if (pIMemT && pCheckInfo->tableId < pIMemT->maxTables) {
+//      pIMem = pIMemT->tData[pCheckInfo->tableId];
+//      rows += (pIMem && pIMem->uid == pCheckInfo->tableId) ? pIMem->numOfRows : 0;
+//    }
+  }
+  return rows;
+}
 
 static SArray* createCheckInfoFromTableGroup(STsdbReadHandle* pTsdbReadHandle, STableGroupInfo* pGroupList) {
   size_t numOfGroup = taosArrayGetSize(pGroupList->pGroupList);
@@ -2261,8 +2262,8 @@ static void moveToNextDataBlockInCurrentFile(STsdbReadHandle* pTsdbReadHandle) {
   cur->mixBlock       = false;
   cur->blockCompleted = false;
 }
-#if 0
-int32_t tsdbGetFileBlocksDistInfo(tsdbReaderT* queryHandle, STableBlockDist* pTableBlockInfo) {
+
+int32_t tsdbGetFileBlocksDistInfo(tsdbReaderT* queryHandle, STableBlockDistInfo* pTableBlockInfo) {
   STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*) queryHandle;
 
   pTableBlockInfo->totalSize = 0;
@@ -2284,7 +2285,7 @@ int32_t tsdbGetFileBlocksDistInfo(tsdbReaderT* queryHandle, STableBlockDist* pTa
   int32_t     code = TSDB_CODE_SUCCESS;
   int32_t     numOfBlocks = 0;
   int32_t     numOfTables = (int32_t)taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
-  int         defaultRows = TSDB_DEFAULT_BLOCK_ROWS(pCfg->maxRowsPerFileBlock);
+  int         defaultRows = 4096;//TSDB_DEFAULT_BLOCK_ROWS(pCfg->maxRowsPerFileBlock);
   STimeWindow win = TSWINDOW_INITIALIZER;
 
   while (true) {
@@ -2345,16 +2346,15 @@ int32_t tsdbGetFileBlocksDistInfo(tsdbReaderT* queryHandle, STableBlockDist* pTa
         if (numOfRows > pTableBlockInfo->maxRows) pTableBlockInfo->maxRows = numOfRows;
         if (numOfRows < pTableBlockInfo->minRows) pTableBlockInfo->minRows = numOfRows;
         if (numOfRows < defaultRows) pTableBlockInfo->numOfSmallBlocks+=1;
-        int32_t  stepIndex = (numOfRows-1)/TSDB_BLOCK_DIST_STEP_ROWS;
-        SFileBlockInfo *blockInfo = (SFileBlockInfo*)taosArrayGet(pTableBlockInfo->dataBlockInfos, stepIndex);
-        blockInfo->numBlocksOfStep++;
+//        int32_t  stepIndex = (numOfRows-1)/TSDB_BLOCK_DIST_STEP_ROWS;
+//        SFileBlockInfo *blockInfo = (SFileBlockInfo*)taosArrayGet(pTableBlockInfo->dataBlockInfos, stepIndex);
+//        blockInfo->numBlocksOfStep++;
       }
     }
   }
 
   return code;
 }
-#endif
 
 static int32_t getDataBlocksInFiles(STsdbReadHandle* pTsdbReadHandle, bool* exists) {
   STsdbFS*       pFileHandle = REPO_FS(pTsdbReadHandle->pTsdb);

source/libs/executor/inc/executorimpl.h

@@ -128,6 +128,11 @@ typedef struct {
   int64_t sumRunTimes;
 } SOperatorProfResult;
 
+typedef struct SLimit {
+  int64_t limit;
+  int64_t offset;
+} SLimit;
+
 typedef struct STaskCostInfo {
   int64_t created;
   int64_t start;
@@ -163,6 +168,11 @@ typedef struct SOperatorCostInfo {
   uint64_t execCost;
 } SOperatorCostInfo;
 
+typedef struct SOrder {
+  uint32_t order;
+  SColumn  col;
+} SOrder;
+
 // The basic query information extracted from the SQueryInfo tree to support the
 // execution of query in a data node.
 typedef struct STaskAttr {
@@ -196,7 +206,6 @@ typedef struct STaskAttr {
 
   STimeWindow    window;
   SInterval      interval;
-  SSessionWindow sw;
   int16_t        precision;
   int16_t        numOfOutput;
   int16_t        fillType;
@@ -206,13 +215,8 @@ typedef struct STaskAttr {
   int32_t       intermediateResultRowSize;  // intermediate result row size, in case of top-k query.
   int32_t       maxTableColumnWidth;
   int32_t       tagLen;  // tag value length of current query
-  SGroupbyExpr* pGroupbyExpr;
 
   SExprInfo* pExpr1;
-  SExprInfo* pExpr2;
-  int32_t    numOfExpr2;
-  SExprInfo* pExpr3;
-  int32_t    numOfExpr3;
 
   SColumnInfo*         tableCols;
   SColumnInfo*         tagColList;
@@ -220,8 +224,6 @@ typedef struct STaskAttr {
   int64_t*             fillVal;
 
   SSingleColumnFilterInfo* pFilterInfo;
-  //  SFilterInfo     *pFilters;
-
   void*           tsdb;
   STableGroupInfo tableGroupInfo;  // table <tid, last_key> list  SArray<STableKeyInfo>
   int32_t         vgId;
@@ -384,7 +386,7 @@ typedef struct SExchangeInfo {
 } SExchangeInfo;
 
 typedef struct STableScanInfo {
-  void*           pTsdbReadHandle;
+  void*           dataReader;
   int32_t         numOfBlocks;  // extract basic running information.
   int32_t         numOfSkipped;
   int32_t         numOfBlockStatis;
@@ -644,12 +646,13 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo*
 SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, int64_t gap, SExecTaskInfo* pTaskInfo);
 SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock,
                                        SArray* pGroupColList, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
+SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SExecTaskInfo* pTaskInfo);
+
 SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols, SInterval* pInterval, SSDataBlock* pResBlock,
                                       int32_t fillType, char* fillVal, bool multigroupResult, SExecTaskInfo* pTaskInfo);
 SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols, SSDataBlock* pResBlock, SExecTaskInfo* pTaskInfo);
 SOperatorInfo* createDistinctOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr,
                                           int32_t numOfOutput);
-SOperatorInfo* createTableBlockInfoScanOperator(void* pTsdbReadHandle, STaskRuntimeEnv* pRuntimeEnv);
 
 SOperatorInfo* createTableSeqScanOperatorInfo(void* pTsdbReadHandle, STaskRuntimeEnv* pRuntimeEnv);
 SOperatorInfo* createAllTimeIntervalOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream,

source/libs/function/src/taggfunction.c

@@ -176,26 +176,6 @@ typedef struct SResPair {
   double avg;
 } SResPair;
 
-#define TSDB_BLOCK_DIST_STEP_ROWS 16
-
-typedef struct STableBlockDist {
-  uint16_t  rowSize;
-  uint16_t  numOfFiles;
-  uint32_t  numOfTables;
-  uint64_t  totalSize;
-  uint64_t  totalRows;
-  int32_t   maxRows;
-  int32_t   minRows;
-  int32_t   firstSeekTimeUs;
-  uint32_t  numOfRowsInMemTable;
-  uint32_t  numOfSmallBlocks;
-  SArray   *dataBlockInfos;
-} STableBlockDist;
-
-typedef struct SFileBlockInfo {
-  int32_t numBlocksOfStep;
-} SFileBlockInfo;
-
 void cleanupResultRowEntry(struct SResultRowEntryInfo* pCell) {
   pCell->initialized = false;
 }
@@ -3984,7 +3964,7 @@ static void irate_function(SqlFunctionCtx *pCtx) {
   }
 }
 
-static void blockDistInfoFromBinary(const char* data, int32_t len, STableBlockDist* pDist) {
+static void blockDistInfoFromBinary(const char* data, int32_t len, STableBlockDistInfo* pDist) {
   SBufferReader br = tbufInitReader(data, len, false);
 
   pDist->numOfTables = tbufReadUint32(&br);
@@ -4024,7 +4004,7 @@ static void blockDistInfoFromBinary(const char* data, int32_t len, STableBlockDi
 
 static void blockInfo_func(SqlFunctionCtx* pCtx) {
   SResultRowEntryInfo *pResInfo = GET_RES_INFO(pCtx);
-  STableBlockDist* pDist = (STableBlockDist*) GET_ROWCELL_INTERBUF(pResInfo);
+  STableBlockDistInfo* pDist = (STableBlockDistInfo*) GET_ROWCELL_INTERBUF(pResInfo);
 
   int32_t len = *(int32_t*) pCtx->pInput;
   blockDistInfoFromBinary((char*)pCtx->pInput + sizeof(int32_t), len, pDist);
@@ -4036,8 +4016,8 @@ static void blockInfo_func(SqlFunctionCtx* pCtx) {
   //pResInfo->hasResult = DATA_SET_FLAG;
 }
 
-static void mergeTableBlockDist(SResultRowEntryInfo* pResInfo, const STableBlockDist* pSrc) {
-  STableBlockDist* pDist = (STableBlockDist*) GET_ROWCELL_INTERBUF(pResInfo);
+static void mergeTableBlockDist(SResultRowEntryInfo* pResInfo, const STableBlockDistInfo* pSrc) {
+  STableBlockDistInfo* pDist = (STableBlockDistInfo*) GET_ROWCELL_INTERBUF(pResInfo);
   assert(pDist != NULL && pSrc != NULL);
 
   pDist->numOfTables += pSrc->numOfTables;
@@ -4071,7 +4051,7 @@ static void mergeTableBlockDist(SResultRowEntryInfo* pResInfo, const STableBlock
 }
 
 void block_func_merge(SqlFunctionCtx* pCtx) {
-  STableBlockDist info = {0};
+  STableBlockDistInfo info = {0};
   int32_t len = *(int32_t*) pCtx->pInput;
   blockDistInfoFromBinary(((char*)pCtx->pInput) + sizeof(int32_t), len, &info);
   SResultRowEntryInfo *pResInfo = GET_RES_INFO(pCtx);
@@ -4082,7 +4062,7 @@ void block_func_merge(SqlFunctionCtx* pCtx) {
   //pResInfo->hasResult = DATA_SET_FLAG;
 }
 
-void getPercentiles(STableBlockDist *pTableBlockDist, int64_t totalBlocks, int32_t numOfPercents,
+void getPercentiles(STableBlockDistInfo *pTableBlockDist, int64_t totalBlocks, int32_t numOfPercents,
                     double* percents, int32_t* percentiles) {
   if (totalBlocks == 0) {
     for (int32_t i = 0; i < numOfPercents; ++i) {
@@ -4117,7 +4097,7 @@ void getPercentiles(STableBlockDist *pTableBlockDist, int64_t totalBlocks, int32
   }
 }
 
-void generateBlockDistResult(STableBlockDist *pTableBlockDist, char* result) {
+void generateBlockDistResult(STableBlockDistInfo *pTableBlockDist, char* result) {
   if (pTableBlockDist == NULL) {
     return;
   }
@@ -4178,7 +4158,7 @@ void generateBlockDistResult(STableBlockDist *pTableBlockDist, char* result) {
 
 void blockinfo_func_finalizer(SqlFunctionCtx* pCtx) {
   SResultRowEntryInfo *pResInfo = GET_RES_INFO(pCtx);
-  STableBlockDist* pDist = (STableBlockDist*) GET_ROWCELL_INTERBUF(pResInfo);
+  STableBlockDistInfo* pDist = (STableBlockDistInfo*) GET_ROWCELL_INTERBUF(pResInfo);
 
   pDist->rowSize = (uint16_t)pCtx->param[0].i;
   generateBlockDistResult(pDist, pCtx->pOutput);