Merge pull request #21331 from taosdata/fix/TD-24097

enh(cache/batchread): load columns in one trip

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

@@ -305,10 +305,6 @@ void    tsdbUntakeReadSnap(STsdbReader *pReader, STsdbReadSnap *pSnap, bool proa
 // tsdbMerge.c ==============================================================================================
 int32_t tsdbMerge(STsdb *pTsdb);
 
-#define TSDB_CACHE_NO(c)       ((c).cacheLast == 0)
-#define TSDB_CACHE_LAST_ROW(c) (((c).cacheLast & 1) > 0)
-#define TSDB_CACHE_LAST(c)     (((c).cacheLast & 2) > 0)
-
 // tsdbDiskData ==============================================================================================
 int32_t tDiskDataBuilderCreate(SDiskDataBuilder **ppBuilder);
 void   *tDiskDataBuilderDestroy(SDiskDataBuilder *pBuilder);
@@ -346,13 +342,17 @@ struct STsdbFS {
 };
 
 typedef struct {
-  rocksdb_t              *db;
-  rocksdb_options_t      *options;
-  rocksdb_flushoptions_t *flushoptions;
-  rocksdb_writeoptions_t *writeoptions;
-  rocksdb_readoptions_t  *readoptions;
-  rocksdb_writebatch_t   *writebatch;
-  TdThreadMutex           rMutex;
+  rocksdb_t                           *db;
+  rocksdb_comparator_t                *my_comparator;
+  rocksdb_cache_t                     *blockcache;
+  rocksdb_block_based_table_options_t *tableoptions;
+  rocksdb_options_t                   *options;
+  rocksdb_flushoptions_t              *flushoptions;
+  rocksdb_writeoptions_t              *writeoptions;
+  rocksdb_readoptions_t               *readoptions;
+  rocksdb_writebatch_t                *writebatch;
+  TdThreadMutex                        rMutex;
+  STSchema                            *pTSchema;
 } SRocksCache;
 
 struct STsdb {
@@ -782,7 +782,7 @@ typedef struct SLDataIter {
 #define tMergeTreeGetRow(_t) (&((_t)->pIter->rInfo.row))
 int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFReader, uint64_t suid, uint64_t uid,
                        STimeWindow *pTimeWindow, SVersionRange *pVerRange, SSttBlockLoadInfo *pBlockLoadInfo,
-                       bool destroyLoadInfo, const char *idStr, bool strictTimeRange, SLDataIter* pLDataIter);
+                       bool destroyLoadInfo, const char *idStr, bool strictTimeRange, SLDataIter *pLDataIter);
 void    tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter);
 bool    tMergeTreeNext(SMergeTree *pMTree);
 bool    tMergeTreeIgnoreEarlierTs(SMergeTree *pMTree);
@@ -822,13 +822,15 @@ typedef struct SCacheRowsReader {
 
 typedef struct {
   TSKEY   ts;
+  int8_t  dirty;
   SColVal colVal;
 } SLastCol;
 
 int32_t tsdbOpenCache(STsdb *pTsdb);
 void    tsdbCloseCache(STsdb *pTsdb);
 int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSDBROW *row);
-int32_t tsdbCacheGet(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArray, SCacheRowsReader *pr, int32_t ltype);
+int32_t tsdbCacheGetBatch(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArray, SCacheRowsReader *pr, int8_t ltype);
+int32_t tsdbCacheGet(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArray, SCacheRowsReader *pr, int8_t ltype);
 int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKEY eKey);
 
 int32_t tsdbCacheInsertLast(SLRUCache *pCache, tb_uid_t uid, TSDBROW *row, STsdb *pTsdb);

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

@@ -197,7 +197,7 @@ void tqClose(STQ*);
 int  tqPushMsg(STQ*, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver);
 int  tqRegisterPushHandle(STQ* pTq, void* handle, SRpcMsg* pMsg);
 int  tqUnregisterPushHandle(STQ* pTq, void* pHandle);
-int  tqStartStreamTasks(STQ* pTq);    // restore all stream tasks after vnode launching completed.
+int  tqStartStreamTasks(STQ* pTq);  // restore all stream tasks after vnode launching completed.
 
 int     tqCommit(STQ*);
 int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, bool isAdd);
@@ -403,6 +403,10 @@ struct SVnode {
 #define VND_IS_RSMA(v)      ((v)->config.isRsma == 1)
 #define VND_IS_TSMA(v)      ((v)->config.isTsma == 1)
 
+#define TSDB_CACHE_NO(c)       ((c).cacheLast == 0)
+#define TSDB_CACHE_LAST_ROW(c) (((c).cacheLast & 1) > 0)
+#define TSDB_CACHE_LAST(c)     (((c).cacheLast & 2) > 0)
+
 struct STbUidStore {
   tb_uid_t  suid;
   SArray*   tbUids;

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

@@ -180,8 +180,7 @@ int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList,
   }
 
   SVnodeCfg* pCfg = &((SVnode*)pVnode)->config;
-
-  int32_t numOfStt = pCfg->sttTrigger;
+  int32_t    numOfStt = pCfg->sttTrigger;
   p->pLoadInfo = tCreateLastBlockLoadInfo(p->pSchema, NULL, 0, numOfStt);
   if (p->pLoadInfo == NULL) {
     tsdbCacherowsReaderClose(p);
@@ -215,7 +214,7 @@ void* tsdbCacherowsReaderClose(void* pReader) {
     taosMemoryFree(p->pSchema);
   }
 
-  taosMemoryFreeClear(p->pDataIter);
+  taosMemoryFree(p->pDataIter);
   taosMemoryFree(p->pCurrSchema);
 
   destroyLastBlockLoadInfo(p->pLoadInfo);
@@ -306,23 +305,27 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
   pr->pDataFReader = NULL;
   pr->pDataFReaderLast = NULL;
 
-  int32_t ltype = (pr->type & CACHESCAN_RETRIEVE_LAST) >> 3;
+  int8_t ltype = (pr->type & CACHESCAN_RETRIEVE_LAST) >> 3;
 
   // retrieve the only one last row of all tables in the uid list.
   if (HASTYPE(pr->type, CACHESCAN_RETRIEVE_TYPE_SINGLE)) {
     int64_t st = taosGetTimestampUs();
     int64_t totalLastTs = INT64_MAX;
+
     for (int32_t i = 0; i < pr->numOfTables; ++i) {
       STableKeyInfo* pKeyInfo = &pr->pTableList[i];
 
-      tsdbCacheGet(pr->pTsdb, pKeyInfo->uid, pRow, pr, ltype);
+      tsdbCacheGetBatch(pr->pTsdb, pKeyInfo->uid, pRow, pr, ltype);
+      // tsdbCacheGet(pr->pTsdb, pKeyInfo->uid, pRow, pr, ltype);
       if (TARRAY_SIZE(pRow) <= 0) {
-        taosArrayClearEx(pRow, freeItem);
+        // taosArrayClearEx(pRow, freeItem);
+        taosArrayClear(pRow);
         continue;
       }
-      SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, 0);
+      SLastCol* pColVal = taosArrayGet(pRow, 0);
       if (COL_VAL_IS_NONE(&pColVal->colVal)) {
-        taosArrayClearEx(pRow, freeItem);
+        // taosArrayClearEx(pRow, freeItem);
+        taosArrayClear(pRow);
         continue;
       }
 
@@ -373,13 +376,14 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
         }
       }
 
-      if (taosArrayGetSize(pTableUidList) == 0) {
+      if (TARRAY_SIZE(pTableUidList) == 0) {
         taosArrayPush(pTableUidList, &pKeyInfo->uid);
       } else {
         taosArraySet(pTableUidList, 0, &pKeyInfo->uid);
       }
 
-      taosArrayClearEx(pRow, freeItem);
+      // taosArrayClearEx(pRow, freeItem);
+      taosArrayClear(pRow);
     }
 
     if (hasRes) {
@@ -387,25 +391,28 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
     }
   } else if (HASTYPE(pr->type, CACHESCAN_RETRIEVE_TYPE_ALL)) {
     for (int32_t i = pr->tableIndex; i < pr->numOfTables; ++i) {
-      STableKeyInfo* pKeyInfo = &pr->pTableList[i];
+      tb_uid_t uid = pr->pTableList[i].uid;
 
-      tsdbCacheGet(pr->pTsdb, pKeyInfo->uid, pRow, pr, ltype);
+      tsdbCacheGetBatch(pr->pTsdb, uid, pRow, pr, ltype);
       if (TARRAY_SIZE(pRow) <= 0) {
-        taosArrayClearEx(pRow, freeItem);
+        // taosArrayClearEx(pRow, freeItem);
+        taosArrayClear(pRow);
         continue;
       }
       SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, 0);
       if (COL_VAL_IS_NONE(&pColVal->colVal)) {
-        taosArrayClearEx(pRow, freeItem);
+        // taosArrayClearEx(pRow, freeItem);
+        taosArrayClear(pRow);
         continue;
       }
 
       saveOneRow(pRow, pResBlock, pr, slotIds, dstSlotIds, pRes, pr->idstr);
-      taosArrayClearEx(pRow, freeItem);
+      // taosArrayClearEx(pRow, freeItem);
+      taosArrayClear(pRow);
 
-      taosArrayPush(pTableUidList, &pKeyInfo->uid);
+      taosArrayPush(pTableUidList, &uid);
 
-      pr->tableIndex += 1;
+      ++pr->tableIndex;
       if (pResBlock->info.rows >= pResBlock->info.capacity) {
         goto _end;
       }
@@ -429,7 +436,9 @@ _end:
   }
 
   taosMemoryFree(pRes);
-  taosArrayDestroyEx(pRow, freeItem);
+  // taosArrayDestroyEx(pRow, freeItem);
+  taosArrayDestroy(pRow);
   taosArrayDestroyEx(pLastCols, freeItem);
+
   return code;
 }

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

@@ -302,12 +302,12 @@ int64_t tsdbCountTbDataRows(STbData *pTbData) {
   return rowsNum;
 }
 
-void tsdbMemTableCountRows(SMemTable *pMemTable, SSHashObj* pTableMap, int64_t *rowsNum) {
+void tsdbMemTableCountRows(SMemTable *pMemTable, SSHashObj *pTableMap, int64_t *rowsNum) {
   taosRLockLatch(&pMemTable->latch);
   for (int32_t i = 0; i < pMemTable->nBucket; ++i) {
     STbData *pTbData = pMemTable->aBucket[i];
     while (pTbData) {
-      void* p = tSimpleHashGet(pTableMap, &pTbData->uid, sizeof(pTbData->uid));
+      void *p = tSimpleHashGet(pTableMap, &pTbData->uid, sizeof(pTbData->uid));
       if (p == NULL) {
         pTbData = pTbData->next;
         continue;
@@ -673,7 +673,10 @@ static int32_t tsdbInsertColDataToTable(SMemTable *pMemTable, STbData *pTbData,
   if (key.ts >= pTbData->maxKey) {
     pTbData->maxKey = key.ts;
   }
-  tsdbCacheUpdate(pMemTable->pTsdb, pTbData->suid, pTbData->uid, &lRow);
+
+  if (!TSDB_CACHE_NO(pMemTable->pTsdb->pVnode->config)) {
+    tsdbCacheUpdate(pMemTable->pTsdb, pTbData->suid, pTbData->uid, &lRow);
+  }
 
   // SMemTable
   pMemTable->minKey = TMIN(pMemTable->minKey, pTbData->minKey);
@@ -734,7 +737,9 @@ static int32_t tsdbInsertRowDataToTable(SMemTable *pMemTable, STbData *pTbData,
   if (key.ts >= pTbData->maxKey) {
     pTbData->maxKey = key.ts;
   }
-  tsdbCacheUpdate(pMemTable->pTsdb, pTbData->suid, pTbData->uid, &lRow);
+  if (!TSDB_CACHE_NO(pMemTable->pTsdb->pVnode->config)) {
+    tsdbCacheUpdate(pMemTable->pTsdb, pTbData->suid, pTbData->uid, &lRow);
+  }
 
   // SMemTable
   pMemTable->minKey = TMIN(pMemTable->minKey, pTbData->minKey);

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

@@ -439,8 +439,10 @@ static int vnodeCommitImpl(SCommitInfo *pInfo) {
   code = tsdbCommit(pVnode->pTsdb, pInfo);
   TSDB_CHECK_CODE(code, lino, _exit);
 
-  code = tsdbCacheCommit(pVnode->pTsdb);
-  TSDB_CHECK_CODE(code, lino, _exit);
+  if (!TSDB_CACHE_NO(pVnode->config)) {
+    code = tsdbCacheCommit(pVnode->pTsdb);
+    TSDB_CHECK_CODE(code, lino, _exit);
+  }
 
   if (VND_IS_RSMA(pVnode)) {
     code = smaCommit(pVnode->pSma, pInfo);