tsdb/cache: relayout cache row reader

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

@@ -833,9 +833,25 @@ void              *destroyLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo);
 void              *destroySttBlockReader(SArray *pLDataIterArray, int64_t *blocks, double *el);
 
 // tsdbCache ==============================================================================================
+typedef enum {
+  READ_MODE_COUNT_ONLY = 0x1,
+  READ_MODE_ALL,
+} EReadMode;
+
+typedef struct STsdbReaderInfo {
+  uint64_t      suid;
+  STSchema     *pSchema;
+  EReadMode     readMode;
+  uint64_t      rowsNum;
+  STimeWindow   window;
+  SVersionRange verRange;
+  int16_t       order;
+} STsdbReaderInfo;
+
 typedef struct SCacheRowsReader {
   STsdb          *pTsdb;
-  SVersionRange  verRange;
+  STsdbReaderInfo info;
+  // SVersionRange  verRange;
   TdThreadMutex  readerMutex;
   SVnode        *pVnode;
   STSchema      *pSchema;

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

@@ -1720,25 +1720,10 @@ typedef struct {
   SMergeTree *pMergeTree;
 } SFSLastIter;
 
-static int32_t lastIterClose(SFSLastIter **iter) {
-  int32_t code = 0;
-
-  if ((*iter)->pMergeTree) {
-    tMergeTreeClose((*iter)->pMergeTree);
-    (*iter)->pMergeTree = NULL;
-  }
-
-  *iter = NULL;
-
-  return code;
-}
-
-static int32_t lastIterNext(SFSLastIter *iter, TSDBROW **ppRow, STsdb *pTsdb, STSchema *pTSchema, tb_uid_t suid,
-                            tb_uid_t uid, SCacheRowsReader *pr, int64_t lastTs, STFileSet *pFileSet, int16_t *aCols,
-                            int nCols) {
+static int32_t lastIterOpen(SFSLastIter *iter, STFileSet *pFileSet, STsdb *pTsdb, STSchema *pTSchema, tb_uid_t suid,
+                            tb_uid_t uid, SCacheRowsReader *pr, int64_t lastTs, int16_t *aCols, int nCols) {
   int32_t code = 0;
 
-  if (!iter->pMergeTree) {
   int64_t loadBlocks = 0;
   double  elapse = 0;
   pr->pLDataIterArray = destroySttBlockReader(pr->pLDataIterArray, &loadBlocks, &elapse);
@@ -1767,9 +1752,26 @@ static int32_t lastIterNext(SFSLastIter *iter, TSDBROW **ppRow, STsdb *pTsdb, ST
   }
 
   iter->pMergeTree = &iter->mergeTree;
+
+  return code;
+}
+
+static int32_t lastIterClose(SFSLastIter **iter) {
+  int32_t code = 0;
+
+  if ((*iter)->pMergeTree) {
+    tMergeTreeClose((*iter)->pMergeTree);
+    (*iter)->pMergeTree = NULL;
   }
 
-  // retrieve next row
+  *iter = NULL;
+
+  return code;
+}
+
+static int32_t lastIterNext(SFSLastIter *iter, TSDBROW **ppRow) {
+  int32_t code = 0;
+
   bool hasVal = tMergeTreeNext(iter->pMergeTree);
   if (!hasVal) {
     *ppRow = NULL;
@@ -1823,6 +1825,17 @@ typedef struct SFSNextRowIter {
   struct CacheNextRowIter *pRowIter;
 } SFSNextRowIter;
 
+static void clearLastFileSet(SFSNextRowIter *state) {
+  if (state->pLastIter) {
+    lastIterClose(&state->pLastIter);
+  }
+
+  if (state->pBlockData) {
+    tBlockDataDestroy(state->pBlockData);
+    state->pBlockData = NULL;
+  }
+}
+
 static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlierTs, bool isLast, int16_t *aCols,
                                 int nCols) {
   SFSNextRowIter *state = (SFSNextRowIter *)iter;
@@ -1837,8 +1850,11 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie
 
   if (SFSNEXTROW_FILESET == state->state) {
   _next_fileset:
-    if (--state->iFileSet < 0) {  // no fileset left, cleanup and return NULL row
+    if (--state->iFileSet < 0) {
+      clearLastFileSet(state);
 
+      *ppRow = NULL;
+      return code;
     } else {
       state->pFileSet = TARRAY2_GET(state->aDFileSet, state->iFileSet);
     }
@@ -1872,9 +1888,16 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie
         goto _err;
       }
 
-      // TODO: load tomb data from data and sttt
+      code = lastIterOpen(&state->lastIter, state->pFileSet, state->pTsdb, state->pTSchema, state->suid, state->uid,
+                          state->pr, state->lastTs, aCols, nCols);
+      if (code != TSDB_CODE_SUCCESS) {
+        goto _err;
+      }
+
       state->pLastIter = &state->lastIter;
 
+      // TODO: load tomb data from data and stt
+
       if (!state->pIndexList) {
         state->pIndexList = taosArrayInit(1, sizeof(SBrinBlk));
       } else {
@@ -1990,8 +2013,7 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie
 
     if (!state->pLastRow) {
       // get next row from fslast and process with fs row, --state->Row if select fs row
-      code = lastIterNext(&state->lastIter, &state->pLastRow, state->pTsdb, state->pTSchema, state->suid, state->uid,
-                          state->pr, state->lastTs, state->pFileSet, aCols, nCols);
+      code = lastIterNext(&state->lastIter, &state->pLastRow);
       if (code != TSDB_CODE_SUCCESS) {
         goto _err;
       }

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

@@ -143,7 +143,7 @@ int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList,
   p->type = type;
   p->pVnode = pVnode;
   p->pTsdb = p->pVnode->pTsdb;
-  p->verRange = (SVersionRange){.minVer = 0, .maxVer = UINT64_MAX};
+  p->info.verRange = (SVersionRange){.minVer = 0, .maxVer = UINT64_MAX};
   p->numOfCols = numOfCols;
   p->pCidList = pCidList;
   p->pSlotIds = pSlotIds;

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

@@ -30,12 +30,12 @@ typedef enum {
   EXTERNAL_ROWS_MAIN = 0x2,
   EXTERNAL_ROWS_NEXT = 0x3,
 } EContentData;
-
+/*
 typedef enum {
   READ_MODE_COUNT_ONLY = 0x1,
   READ_MODE_ALL,
 } EReadMode;
-
+*/
 typedef struct {
   STbDataIter* iter;
   int32_t      index;
@@ -461,7 +461,7 @@ static SSHashObj* createDataBlockScanInfo(STsdbReader* pTsdbReader, SBlockInfoBu
 }
 
 static void resetAllDataBlockScanInfo(SSHashObj* pTableMap, int64_t ts, int32_t step) {
-  void   *p = NULL;
+  void*   p = NULL;
   int32_t iter = 0;
 
   while ((p = tSimpleHashIterate(pTableMap, p, &iter)) != NULL) {
@@ -743,7 +743,8 @@ void tsdbReleaseDataBlock(STsdbReader* pReader) {
   }
 }
 
-static int32_t initResBlockInfo(SResultBlockInfo* pResBlockInfo, int64_t capacity, SSDataBlock* pResBlock, SQueryTableDataCond* pCond) {
+static int32_t initResBlockInfo(SResultBlockInfo* pResBlockInfo, int64_t capacity, SSDataBlock* pResBlock,
+                                SQueryTableDataCond* pCond) {
   pResBlockInfo->capacity = capacity;
   pResBlockInfo->pResBlock = pResBlock;
   terrno = 0;
@@ -937,7 +938,8 @@ static void cleanupTableScanInfo(SReaderStatus* pStatus) {
   pStatus->mapDataCleaned = true;
 }
 
-static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, SBlockNumber* pBlockNum, SArray* pTableScanInfoList) {
+static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, SBlockNumber* pBlockNum,
+                               SArray* pTableScanInfoList) {
   size_t sizeInDisk = 0;
   size_t numOfTables = taosArrayGetSize(pIndexList);
 
@@ -1125,18 +1127,18 @@ static int32_t getEndPosInDataBlock(STsdbReader* pReader, SBlockData* pBlockData
     endPos = doBinarySearchKey(pBlockData->aTSKEY, pBlock->nRow, pos, key, pReader->order);
   }
 
-  if ((pReader->verRange.maxVer >= pBlock->minVer && pReader->verRange.maxVer < pBlock->maxVer)||
+  if ((pReader->verRange.maxVer >= pBlock->minVer && pReader->verRange.maxVer < pBlock->maxVer) ||
       (pReader->verRange.minVer <= pBlock->maxVer && pReader->verRange.minVer > pBlock->minVer)) {
     int32_t i = endPos;
 
     if (asc) {
-      for(; i >= 0; --i) {
+      for (; i >= 0; --i) {
         if (pBlockData->aVersion[i] <= pReader->verRange.maxVer) {
           break;
         }
       }
     } else {
-      for(; i < pBlock->nRow; ++i) {
+      for (; i < pBlock->nRow; ++i) {
         if (pBlockData->aVersion[i] >= pReader->verRange.minVer) {
           break;
         }
@@ -1309,17 +1311,17 @@ static int32_t copyBlockDataToSDataBlock(STsdbReader* pReader) {
       ASSERT(pReader->verRange.minVer <= pBlock->maxVer && pReader->verRange.maxVer >= pBlock->minVer);
 
       // find the appropriate start position that satisfies the version requirement.
-      if ((pReader->verRange.maxVer >= pBlock->minVer && pReader->verRange.maxVer < pBlock->maxVer)||
+      if ((pReader->verRange.maxVer >= pBlock->minVer && pReader->verRange.maxVer < pBlock->maxVer) ||
           (pReader->verRange.minVer <= pBlock->maxVer && pReader->verRange.minVer > pBlock->minVer)) {
         int32_t i = pDumpInfo->rowIndex;
         if (asc) {
-          for(; i < pBlock->nRow; ++i) {
+          for (; i < pBlock->nRow; ++i) {
             if (pBlockData->aVersion[i] >= pReader->verRange.minVer) {
               break;
             }
           }
         } else {
-          for(; i >= 0; --i) {
+          for (; i >= 0; --i) {
             if (pBlockData->aVersion[i] <= pReader->verRange.maxVer) {
               break;
             }
@@ -1562,7 +1564,8 @@ static int32_t doSetCurrentBlock(SDataBlockIter* pBlockIter, const char* idStr)
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t initBlockIterator(STsdbReader* pReader, SDataBlockIter* pBlockIter, int32_t numOfBlocks, SArray* pTableList) {
+static int32_t initBlockIterator(STsdbReader* pReader, SDataBlockIter* pBlockIter, int32_t numOfBlocks,
+                                 SArray* pTableList) {
   bool asc = ASCENDING_TRAVERSE(pReader->order);
 
   SBlockOrderSupporter sup = {0};
@@ -1973,7 +1976,8 @@ static bool nextRowFromLastBlocks(SLastBlockReader* pLastBlockReader, STableBloc
     pLastBlockReader->currentKey = key;
     pScanInfo->lastKeyInStt = key;
 
-    if (!hasBeenDropped(pScanInfo->delSkyline, &pScanInfo->lastBlockDelIndex, key, ver, pLastBlockReader->order, pVerRange)) {
+    if (!hasBeenDropped(pScanInfo->delSkyline, &pScanInfo->lastBlockDelIndex, key, ver, pLastBlockReader->order,
+                        pVerRange)) {
       return true;
     }
   }
@@ -2220,7 +2224,8 @@ static int32_t doMergeFileBlockAndLastBlock(SLastBlockReader* pLastBlockReader,
   // create local variable to hold the row value
   TSDBROW fRow = {.iRow = pRow->iRow, .type = TSDBROW_COL_FMT, .pBlockData = pRow->pBlockData};
 
-  tsdbTrace("fRow ptr:%p, %d, uid:%" PRIu64 ", %s", pRow->pBlockData, pRow->iRow, pLastBlockReader->uid, pReader->idStr);
+  tsdbTrace("fRow ptr:%p, %d, uid:%" PRIu64 ", %s", pRow->pBlockData, pRow->iRow, pLastBlockReader->uid,
+            pReader->idStr);
 
   // only last block exists
   if ((!mergeBlockData) || (tsLastBlock != pBlockData->aTSKEY[pDumpInfo->rowIndex])) {
@@ -2240,7 +2245,8 @@ static int32_t doMergeFileBlockAndLastBlock(SLastBlockReader* pLastBlockReader,
 
       TSDBROW* pRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
       tsdbRowMergerAdd(pMerger, pRow1, NULL);
-      doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLastBlock, pMerger, &pReader->verRange, pReader->idStr);
+      doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLastBlock, pMerger, &pReader->verRange,
+                             pReader->idStr);
 
       code = tsdbRowMergerGetRow(pMerger, &pTSRow);
       if (code != TSDB_CODE_SUCCESS) {
@@ -2874,7 +2880,8 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
     }
   } else {  // file blocks not exist
     pBlockScanInfo = *pReader->status.pTableIter;
-    if (pReader->pIgnoreTables && taosHashGet(*pReader->pIgnoreTables, &pBlockScanInfo->uid, sizeof(pBlockScanInfo->uid))) {
+    if (pReader->pIgnoreTables &&
+        taosHashGet(*pReader->pIgnoreTables, &pBlockScanInfo->uid, sizeof(pBlockScanInfo->uid))) {
       setBlockAllDumped(pDumpInfo, pBlock->maxKey.ts, pReader->order);
       return code;
     }
@@ -3261,7 +3268,6 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) {
     return terrno;
   }
 
-
   initLastBlockReader(pLastBlockReader, pScanInfo, pReader);
   TSDBKEY keyInBuf = getCurrentKeyInBuf(pScanInfo, pReader);
 
@@ -3338,7 +3344,7 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) {
     }
   }
 
-  return (pReader->code != TSDB_CODE_SUCCESS)? pReader->code:code;
+  return (pReader->code != TSDB_CODE_SUCCESS) ? pReader->code : code;
 }
 
 static int32_t doSumFileBlockRows(STsdbReader* pReader, SDataFReader* pFileReader) {
@@ -3493,7 +3499,8 @@ static int32_t buildBlockFromBufferSequentially(STsdbReader* pReader) {
     }
 
     STableBlockScanInfo** pBlockScanInfo = pStatus->pTableIter;
-    if (pReader->pIgnoreTables && taosHashGet(*pReader->pIgnoreTables, &(*pBlockScanInfo)->uid, sizeof((*pBlockScanInfo)->uid))) {
+    if (pReader->pIgnoreTables &&
+        taosHashGet(*pReader->pIgnoreTables, &(*pBlockScanInfo)->uid, sizeof((*pBlockScanInfo)->uid))) {
       bool hasNexTable = moveToNextTable(pUidList, pStatus);
       if (!hasNexTable) {
         return TSDB_CODE_SUCCESS;
@@ -3589,7 +3596,7 @@ static ERetrieveType doReadDataFromLastFiles(STsdbReader* pReader) {
   SSDataBlock*    pResBlock = pReader->resBlockInfo.pResBlock;
   SDataBlockIter* pBlockIter = &pReader->status.blockIter;
 
-  while(1) {
+  while (1) {
     terrno = 0;
 
     code = doLoadLastBlockSequentially(pReader);
@@ -3737,13 +3744,14 @@ SVersionRange getQueryVerRange(SVnode* pVnode, SQueryTableDataCond* pCond, const
     endVer = (pCond->endVersion > pVnode->state.applied) ? pVnode->state.applied : pCond->endVersion;
   }
 
-  tsdbDebug("queried verRange:%"PRId64"-%"PRId64", revised query verRange:%"PRId64"-%"PRId64", %s", pCond->startVersion,
-            pCond->endVersion, startVer, endVer, id);
+  tsdbDebug("queried verRange:%" PRId64 "-%" PRId64 ", revised query verRange:%" PRId64 "-%" PRId64 ", %s",
+            pCond->startVersion, pCond->endVersion, startVer, endVer, id);
 
   return (SVersionRange){.minVer = startVer, .maxVer = endVer};
 }
 
-bool hasBeenDropped(const SArray* pDelList, int32_t* index, int64_t key, int64_t ver, int32_t order, SVersionRange* pVerRange) {
+bool hasBeenDropped(const SArray* pDelList, int32_t* index, int64_t key, int64_t ver, int32_t order,
+                    SVersionRange* pVerRange) {
   if (pDelList == NULL) {
     return false;
   }
@@ -3761,8 +3769,7 @@ bool hasBeenDropped(const SArray* pDelList, int32_t* index, int64_t key, int64_t
         return false;
       } else if (key == last->ts) {
         TSDBKEY* prev = taosArrayGet(pDelList, num - 2);
-        return (prev->version >= ver && prev->version <= pVerRange->maxVer &&
-                prev->version >= pVerRange->minVer);
+        return (prev->version >= ver && prev->version <= pVerRange->maxVer && prev->version >= pVerRange->minVer);
       }
     } else {
       TSDBKEY* pCurrent = taosArrayGet(pDelList, *index);
@@ -4070,14 +4077,14 @@ int32_t doMergeMemTableMultiRows(TSDBROW* pRow, uint64_t uid, SIterInfo* pIter,
       return terrno;
     }
 
-    tsdbRowMergerAdd(&pReader->status.merger,pNextRow, pTSchema1);
+    tsdbRowMergerAdd(&pReader->status.merger, pNextRow, pTSchema1);
   } else {  // let's merge rows in file block
     code = tsdbRowMergerAdd(&pReader->status.merger, &current, pReader->pSchema);
     if (code != TSDB_CODE_SUCCESS) {
       return code;
     }
 
-    tsdbRowMergerAdd(&pReader->status.merger,pNextRow, NULL);
+    tsdbRowMergerAdd(&pReader->status.merger, pNextRow, NULL);
   }
 
   code = doMergeRowsInBuf(pIter, uid, TSDBROW_TS(&current), pDelList, pReader);
@@ -4124,9 +4131,8 @@ int32_t doMergeMemIMemRows(TSDBROW* pRow, TSDBROW* piRow, STableBlockScanInfo* p
       return code;
     }
 
-    tsdbRowMergerAdd(&pReader->status.merger,pRow, pSchema);
-    code =
-        doMergeRowsInBuf(&pBlockScanInfo->iter, pBlockScanInfo->uid, k.ts, pBlockScanInfo->delSkyline, pReader);
+    tsdbRowMergerAdd(&pReader->status.merger, pRow, pSchema);
+    code = doMergeRowsInBuf(&pBlockScanInfo->iter, pBlockScanInfo->uid, k.ts, pBlockScanInfo->delSkyline, pReader);
     if (code != TSDB_CODE_SUCCESS) {
       return code;
     }
@@ -4452,13 +4458,14 @@ static int32_t doOpenReaderImpl(STsdbReader* pReader) {
 }
 
 static void freeSchemaFunc(void* param) {
-  void **p = (void **)param;
+  void** p = (void**)param;
   taosMemoryFreeClear(*p);
 }
 
 // ====================================== EXPOSED APIs ======================================
 int32_t tsdbReaderOpen(void* pVnode, SQueryTableDataCond* pCond, void* pTableList, int32_t numOfTables,
-                       SSDataBlock* pResBlock, void** ppReader, const char* idstr, bool countOnly, SHashObj** pIgnoreTables) {
+                       SSDataBlock* pResBlock, void** ppReader, const char* idstr, bool countOnly,
+                       SHashObj** pIgnoreTables) {
   STimeWindow window = pCond->twindows;
   SVnodeCfg*  pConf = &(((SVnode*)pVnode)->config);
 
@@ -4950,8 +4957,9 @@ int32_t tsdbNextDataBlock(STsdbReader* pReader, bool* hasNext) {
 
   *hasNext = false;
 
-  if (isEmptyQueryTimeWindow(&pReader->window) || pReader->step == EXTERNAL_ROWS_NEXT || pReader->code != TSDB_CODE_SUCCESS) {
-    return (pReader->code != TSDB_CODE_SUCCESS)? pReader->code:code;
+  if (isEmptyQueryTimeWindow(&pReader->window) || pReader->step == EXTERNAL_ROWS_NEXT ||
+      pReader->code != TSDB_CODE_SUCCESS) {
+    return (pReader->code != TSDB_CODE_SUCCESS) ? pReader->code : code;
   }
 
   SReaderStatus* pStatus = &pReader->status;
@@ -5087,7 +5095,7 @@ static bool doFillNullColSMA(SBlockLoadSuppInfo* pSup, int32_t numOfRows, int32_
   return hasNullSMA;
 }
 
-int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SSDataBlock* pDataBlock, bool* allHave, bool *hasNullSMA) {
+int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SSDataBlock* pDataBlock, bool* allHave, bool* hasNullSMA) {
   SColumnDataAgg*** pBlockSMA = &pDataBlock->pBlockAgg;
 
   int32_t code = 0;

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

@@ -30,11 +30,6 @@ typedef enum {
   READER_STATUS_NORMAL = 0x2,
 } EReaderStatus;
 
-typedef enum {
-  READ_MODE_COUNT_ONLY = 0x1,
-  READ_MODE_ALL,
-} EReadMode;
-
 typedef enum {
   EXTERNAL_ROWS_PREV = 0x1,
   EXTERNAL_ROWS_MAIN = 0x2,
@@ -69,16 +64,6 @@ typedef struct STableBlockScanInfo {
   bool      iterInit;           // whether to initialize the in-memory skip list iterator or not
 } STableBlockScanInfo;
 
-typedef struct STsdbReaderInfo {
-  uint64_t      suid;
-  STSchema*     pSchema;
-  EReadMode     readMode;
-  uint64_t      rowsNum;
-  STimeWindow   window;
-  SVersionRange verRange;
-  int16_t       order;
-} STsdbReaderInfo;
-
 typedef struct SResultBlockInfo {
   SSDataBlock* pResBlock;
   bool         freeBlock;