Merge pull request #8480 from taosdata/enhance/TS-493-M

Enhance/ts 493 m - lazy load cache last

src/tsdb/inc/tsdbMeta.h

@@ -41,6 +41,7 @@ typedef struct STable {
   int16_t        restoreColumnNum;
   bool           hasRestoreLastColumn;
   int            lastColSVersion;
+  int16_t        cacheLastConfigVersion;
   T_REF_DECLARE()
 } STable;
 

src/tsdb/inc/tsdbint.h

@@ -79,8 +79,8 @@ struct STsdbRepo {
   STsdbCfg        save_config;    // save apply config
   bool            config_changed; // config changed flag
   pthread_mutex_t save_mutex;     // protect save config
-  
-  uint8_t         hasCachedLastColumn;
+
+  int16_t         cacheLastConfigVersion;
 
   STsdbAppH       appH;
   STsdbStat       stat;
@@ -110,7 +110,8 @@ int        tsdbUnlockRepo(STsdbRepo* pRepo);
 STsdbMeta* tsdbGetMeta(STsdbRepo* pRepo);
 int        tsdbCheckCommit(STsdbRepo* pRepo);
 int        tsdbRestoreInfo(STsdbRepo* pRepo);
-int        tsdbCacheLastData(STsdbRepo *pRepo, STsdbCfg* oldCfg);
+UNUSED_FUNC int tsdbCacheLastData(STsdbRepo *pRepo, STsdbCfg* oldCfg);
+int32_t    tsdbLoadLastCache(STsdbRepo *pRepo, STable* pTable);
 void       tsdbGetRootDir(int repoid, char dirName[]);
 void       tsdbGetDataDir(int repoid, char dirName[]);
 

src/tsdb/src/tsdbCommitQueue.c

@@ -146,7 +146,9 @@ static void tsdbApplyRepoConfig(STsdbRepo *pRepo) {
 
   if (oldCfg.cacheLastRow != pRepo->config.cacheLastRow) {
     if (tsdbLockRepo(pRepo) < 0) return;
-    tsdbCacheLastData(pRepo, &oldCfg);
+    // tsdbCacheLastData(pRepo, &oldCfg);
+    // lazy load last cache when query or update
+    ++pRepo->cacheLastConfigVersion;
     tsdbUnlockRepo(pRepo);
   }
 

src/tsdb/src/tsdbMain.c

@@ -562,7 +562,7 @@ static STsdbRepo *tsdbNewRepo(STsdbCfg *pCfg, STsdbAppH *pAppH) {
     return NULL;
   }
   pRepo->config_changed = false;
-  atomic_store_8(&pRepo->hasCachedLastColumn, 0);
+  pRepo->cacheLastConfigVersion = 0;
 
   code = tsem_init(&(pRepo->readyToCommit), 0, 1);
   if (code != 0) {
@@ -711,7 +711,7 @@ static int tsdbRestoreLastColumns(STsdbRepo *pRepo, STable *pTable, SReadH* pRea
       tsdbGetBlockStatis(pReadh, pBlockStatis, (int)numColumns);
       loadStatisData = true;
     }
-
+    TSDB_WLOCK_TABLE(pTable);  // lock when update pTable->lastCols[]
     for (int16_t i = 0; i < numColumns && numColumns > pTable->restoreColumnNum; ++i) {
       STColumn *pCol = schemaColAt(pSchema, i);
       // ignore loaded columns
@@ -760,6 +760,7 @@ static int tsdbRestoreLastColumns(STsdbRepo *pRepo, STable *pTable, SReadH* pRea
         break;
       }
     }
+    TSDB_WUNLOCK_TABLE(pTable);
   }
 
 out:
@@ -774,7 +775,6 @@ out:
 }
 
 static int tsdbRestoreLastRow(STsdbRepo *pRepo, STable *pTable, SReadH* pReadh, SBlockIdx *pIdx) {
-  ASSERT(pTable->lastRow == NULL);
   if (tsdbLoadBlockInfo(pReadh, NULL) < 0) {
     return -1;
   }
@@ -788,21 +788,32 @@ static int tsdbRestoreLastRow(STsdbRepo *pRepo, STable *pTable, SReadH* pReadh,
   // Get the data in row
   
   STSchema *pSchema = tsdbGetTableSchema(pTable);
-  pTable->lastRow = taosTMalloc(memRowMaxBytesFromSchema(pSchema));
-  if (pTable->lastRow == NULL) {
+  SMemRow   lastRow = taosTMalloc(memRowMaxBytesFromSchema(pSchema));
+  if (lastRow == NULL) {
     terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
     return -1;
   }
-  memRowSetType(pTable->lastRow, SMEM_ROW_DATA);
-  tdInitDataRow(memRowDataBody(pTable->lastRow), pSchema);
+  memRowSetType(lastRow, SMEM_ROW_DATA);
+  tdInitDataRow(memRowDataBody(lastRow), pSchema);
   for (int icol = 0; icol < schemaNCols(pSchema); icol++) {
     STColumn *pCol = schemaColAt(pSchema, icol);
     SDataCol *pDataCol = pReadh->pDCols[0]->cols + icol;
-    tdAppendColVal(memRowDataBody(pTable->lastRow), tdGetColDataOfRow(pDataCol, pBlock->numOfRows - 1), pCol->type,
+    tdAppendColVal(memRowDataBody(lastRow), tdGetColDataOfRow(pDataCol, pBlock->numOfRows - 1), pCol->type,
                    pCol->offset);
   }
 
-  pTable->lastKey = memRowKey(pTable->lastRow);
+  TSKEY lastKey = memRowKey(lastRow);
+  
+  // during the load data in file, new data would be inserted and last row has been updated
+  TSDB_WLOCK_TABLE(pTable);
+  if (pTable->lastRow == NULL) {
+    pTable->lastKey = lastKey;
+    pTable->lastRow = lastRow;
+    TSDB_WUNLOCK_TABLE(pTable);
+  } else {
+    TSDB_WUNLOCK_TABLE(pTable);
+    taosTZfree(lastRow);
+  }
 
   return 0;
 }
@@ -874,14 +885,105 @@ int tsdbRestoreInfo(STsdbRepo *pRepo) {
 
   tsdbDestroyReadH(&readh);
 
-  if (CACHE_LAST_NULL_COLUMN(pCfg)) {
-    atomic_store_8(&pRepo->hasCachedLastColumn, 1);
+  // if (CACHE_LAST_NULL_COLUMN(pCfg)) {
+  //   atomic_store_8(&pRepo->hasCachedLastColumn, 1);
+  // }
+
+  return 0;
+}
+
+int32_t tsdbLoadLastCache(STsdbRepo *pRepo, STable *pTable) {
+  SFSIter    fsiter;
+  SReadH     readh;
+  SDFileSet *pSet;
+  int cacheLastRowTableNum = 0;
+  int cacheLastColTableNum = 0;
+
+  bool cacheLastRow = CACHE_LAST_ROW(&(pRepo->config));
+  bool cacheLastCol = CACHE_LAST_NULL_COLUMN(&(pRepo->config));
+
+  tsdbDebug("tsdbLoadLastCache for %s, cacheLastRow:%d, cacheLastCol:%d", pTable->name->data, cacheLastRow, cacheLastCol);
+
+  pTable->cacheLastConfigVersion = pRepo->cacheLastConfigVersion;
+
+  if (!cacheLastRow && pTable->lastRow != NULL) {
+    taosTZfree(pTable->lastRow);
+    pTable->lastRow = NULL;
   }
+  if (!cacheLastCol && pTable->lastCols != NULL) {
+    tsdbFreeLastColumns(pTable);
+  }
+
+  if (!cacheLastRow && !cacheLastCol) {
+    return 0;
+  }
+
+  cacheLastRowTableNum = (cacheLastRow && pTable->lastRow  == NULL) ? 1 : 0;
+  cacheLastColTableNum = (cacheLastCol && pTable->lastCols == NULL) ? 1 : 0;
+
+  if (cacheLastRowTableNum == 0 && cacheLastColTableNum == 0) {
+    return 0;
+  }
+
+  if (tsdbInitReadH(&readh, pRepo) < 0) {
+    return -1;
+  }
+
+  tsdbRLockFS(REPO_FS(pRepo));
+  tsdbFSIterInit(&fsiter, REPO_FS(pRepo), TSDB_FS_ITER_BACKWARD);
+
+  while ((cacheLastRowTableNum > 0 || cacheLastColTableNum > 0) && (pSet = tsdbFSIterNext(&fsiter)) != NULL) {
+    if (tsdbSetAndOpenReadFSet(&readh, pSet) < 0) {
+      tsdbUnLockFS(REPO_FS(pRepo));
+      tsdbDestroyReadH(&readh);
+      return -1;
+    }
+
+    if (tsdbLoadBlockIdx(&readh) < 0) {
+      tsdbUnLockFS(REPO_FS(pRepo));
+      tsdbDestroyReadH(&readh);
+      return -1;
+    }
+
+    // tsdbDebug("tsdbRestoreInfo restore vgId:%d,table:%s", REPO_ID(pRepo), pTable->name->data);
+
+    if (tsdbSetReadTable(&readh, pTable) < 0) {
+      tsdbUnLockFS(REPO_FS(pRepo));
+      tsdbDestroyReadH(&readh);
+      return -1;
+    }
+
+    SBlockIdx *pIdx = readh.pBlkIdx;
+
+    if (pIdx && (cacheLastRowTableNum > 0) && (pTable->lastRow == NULL)) {
+      if (tsdbRestoreLastRow(pRepo, pTable, &readh, pIdx) != 0) {
+        tsdbUnLockFS(REPO_FS(pRepo));
+        tsdbDestroyReadH(&readh);
+        return -1;
+      }
+      cacheLastRowTableNum -= 1;
+    }
+
+    // restore NULL columns
+    if (pIdx && (cacheLastColTableNum > 0) && !pTable->hasRestoreLastColumn) {
+      if (tsdbRestoreLastColumns(pRepo, pTable, &readh) != 0) {
+        tsdbUnLockFS(REPO_FS(pRepo));
+        tsdbDestroyReadH(&readh);
+        return -1;
+      }
+      if (pTable->hasRestoreLastColumn) {
+        cacheLastColTableNum -= 1;
+      }
+    }
+  }
+
+  tsdbUnLockFS(REPO_FS(pRepo));
+  tsdbDestroyReadH(&readh);
 
   return 0;
 }
 
-int tsdbCacheLastData(STsdbRepo *pRepo, STsdbCfg* oldCfg) {
+UNUSED_FUNC int tsdbCacheLastData(STsdbRepo *pRepo, STsdbCfg* oldCfg) {
   bool cacheLastRow = false, cacheLastCol = false;
   SFSIter    fsiter;
   SReadH     readh;
@@ -915,9 +1017,9 @@ int tsdbCacheLastData(STsdbRepo *pRepo, STsdbCfg* oldCfg) {
 
   // if close last option,need to free data
   if (need_free_last_row || need_free_last_col) {
-    if (need_free_last_col) {
-      atomic_store_8(&pRepo->hasCachedLastColumn, 0);
-    }
+    // if (need_free_last_col) {
+    //   atomic_store_8(&pRepo->hasCachedLastColumn, 0);
+    // }
     tsdbInfo("free cache last data since cacheLast option changed");    
     for (int i = 1; i <= maxTableIdx; i++) {
       STable *pTable = pMeta->tables[i];
@@ -995,9 +1097,9 @@ int tsdbCacheLastData(STsdbRepo *pRepo, STsdbCfg* oldCfg) {
 
   tsdbDestroyReadH(&readh);
 
-  if (cacheLastCol) {
-    atomic_store_8(&pRepo->hasCachedLastColumn, 1);
-  }
+  // if (cacheLastCol) {
+  //   atomic_store_8(&pRepo->hasCachedLastColumn, 1);
+  // }
   
   return 0;
 }

src/tsdb/src/tsdbMemTable.c

@@ -996,7 +996,8 @@ static void updateTableLatestColumn(STsdbRepo *pRepo, STable *pTable, SMemRow ro
     if ((value == NULL) || isNull(value, pTCol->type)) {
       continue;
     }
-
+    // lock
+    TSDB_WLOCK_TABLE(pTable); 
     SDataCol *pDataCol = &(pLatestCols[idx]);
     if (pDataCol->pData == NULL) {
       pDataCol->pData = malloc(pTCol->bytes);
@@ -1012,6 +1013,8 @@ static void updateTableLatestColumn(STsdbRepo *pRepo, STable *pTable, SMemRow ro
     memcpy(pDataCol->pData, value, bytes);
     //tsdbInfo("updateTableLatestColumn vgId:%d cache column %d for %d,%s", REPO_ID(pRepo), j, pDataCol->bytes, (char*)pDataCol->pData);
     pDataCol->ts = memRowKey(row);
+    // unlock
+    TSDB_WUNLOCK_TABLE(pTable); 
   }
 }
 
@@ -1058,5 +1061,8 @@ static int tsdbUpdateTableLatestInfo(STsdbRepo *pRepo, STable *pTable, SMemRow r
       updateTableLatestColumn(pRepo, pTable, row);
     }
   }
+
+  pTable->cacheLastConfigVersion = pRepo->cacheLastConfigVersion;
+
   return 0;
 }

src/tsdb/src/tsdbMeta.c

@@ -627,27 +627,30 @@ int16_t tsdbGetLastColumnsIndexByColId(STable* pTable, int16_t colId) {
 }
 
 int tsdbInitColIdCacheWithSchema(STable* pTable, STSchema* pSchema) {
-  ASSERT(pTable->lastCols == NULL);
+  TSDB_WLOCK_TABLE(pTable);
+  if (pTable->lastCols == NULL) {
+    int16_t numOfColumn = pSchema->numOfCols;
 
-  int16_t numOfColumn = pSchema->numOfCols;
+    pTable->lastCols = (SDataCol *)malloc(numOfColumn * sizeof(SDataCol));
+    if (pTable->lastCols == NULL) {
+      TSDB_WUNLOCK_TABLE(pTable);
+      return -1;
+    }
 
-  pTable->lastCols = (SDataCol*)malloc(numOfColumn * sizeof(SDataCol));
-  if (pTable->lastCols == NULL) {
-    return -1;
-  }
+    for (int16_t i = 0; i < numOfColumn; ++i) {
+      STColumn *pCol = schemaColAt(pSchema, i);
+      SDataCol *pDataCol = &(pTable->lastCols[i]);
+      pDataCol->bytes = 0;
+      pDataCol->pData = NULL;
+      pDataCol->colId = pCol->colId;
+    }
 
-  for (int16_t i = 0; i < numOfColumn; ++i) {
-    STColumn *pCol = schemaColAt(pSchema, i);
-    SDataCol* pDataCol = &(pTable->lastCols[i]);
-    pDataCol->bytes = 0;
-    pDataCol->pData = NULL;
-    pDataCol->colId = pCol->colId;
+    pTable->lastColSVersion = schemaVersion(pSchema);
+    pTable->maxColNum = numOfColumn;
+    pTable->restoreColumnNum = 0;
+    pTable->hasRestoreLastColumn = false;
   }
-
-  pTable->lastColSVersion = schemaVersion(pSchema);
-  pTable->maxColNum = numOfColumn;
-  pTable->restoreColumnNum = 0;
-  pTable->hasRestoreLastColumn = false;
+  TSDB_WUNLOCK_TABLE(pTable);
   return 0;
 }
 
@@ -797,6 +800,7 @@ static STable *tsdbNewTable() {
 
   pTable->lastCols = NULL;
   pTable->restoreColumnNum = 0;
+  pTable->cacheLastConfigVersion = 0;
   pTable->maxColNum = 0;
   pTable->hasRestoreLastColumn = false;
   pTable->lastColSVersion = -1;

src/tsdb/src/tsdbRead.c

@@ -156,6 +156,7 @@ typedef struct STableGroupSupporter {
 static STimeWindow updateLastrowForEachGroup(STableGroupInfo *groupList);
 static int32_t checkForCachedLastRow(STsdbQueryHandle* pQueryHandle, STableGroupInfo *groupList);
 static int32_t checkForCachedLast(STsdbQueryHandle* pQueryHandle);
+static int32_t lazyLoadCacheLast(STsdbQueryHandle* pQueryHandle);
 static int32_t tsdbGetCachedLastRow(STable* pTable, SMemRow* pRes, TSKEY* lastKey);
 
 static void    changeQueryHandleForInterpQuery(TsdbQueryHandleT pHandle);
@@ -589,6 +590,28 @@ void tsdbResetQueryHandleForNewTable(TsdbQueryHandleT queryHandle, STsdbQueryCon
   pQueryHandle->next = doFreeColumnInfoData(pQueryHandle->next);
 }
 
+static int32_t lazyLoadCacheLast(STsdbQueryHandle* pQueryHandle) {
+  STsdbRepo* pRepo = pQueryHandle->pTsdb;
+
+  size_t  numOfTables = taosArrayGetSize(pQueryHandle->pTableCheckInfo);
+  int32_t code = 0;
+  for (size_t i = 0; i < numOfTables; ++i) {
+    STableCheckInfo* pCheckInfo = taosArrayGet(pQueryHandle->pTableCheckInfo, i);
+    STable*          pTable = pCheckInfo->pTableObj;
+    if (pTable->cacheLastConfigVersion == pRepo->cacheLastConfigVersion) {
+      continue;
+    }
+    code = tsdbLoadLastCache(pRepo, pTable);
+    if (code != 0) {
+      tsdbError("%p uid:%" PRId64 ", tid:%d, failed to load last cache since %s", pQueryHandle, pTable->tableId.uid,
+                pTable->tableId.tid, tstrerror(terrno));
+      break;
+    }
+  }
+
+  return code;
+}
+
 TsdbQueryHandleT tsdbQueryLastRow(STsdbRepo *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList, uint64_t qId, SMemRef* pMemRef) {
   pCond->twindow = updateLastrowForEachGroup(groupList);
 
@@ -602,6 +625,8 @@ TsdbQueryHandleT tsdbQueryLastRow(STsdbRepo *tsdb, STsdbQueryCond *pCond, STable
     return NULL;
   }
 
+  lazyLoadCacheLast(pQueryHandle);
+
   int32_t code = checkForCachedLastRow(pQueryHandle, groupList);
   if (code != TSDB_CODE_SUCCESS) { // set the numOfTables to be 0
     terrno = code;
@@ -616,13 +641,14 @@ TsdbQueryHandleT tsdbQueryLastRow(STsdbRepo *tsdb, STsdbQueryCond *pCond, STable
   return pQueryHandle;
 }
 
-
 TsdbQueryHandleT tsdbQueryCacheLast(STsdbRepo *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList, uint64_t qId, SMemRef* pMemRef) {
   STsdbQueryHandle *pQueryHandle = (STsdbQueryHandle*) tsdbQueryTables(tsdb, pCond, groupList, qId, pMemRef);
   if (pQueryHandle == NULL) {
     return NULL;
   }
 
+  lazyLoadCacheLast(pQueryHandle);
+
   int32_t code = checkForCachedLast(pQueryHandle);
   if (code != TSDB_CODE_SUCCESS) { // set the numOfTables to be 0
     terrno = code;
@@ -2781,6 +2807,9 @@ static bool loadCachedLast(STsdbQueryHandle* pQueryHandle) {
     }
     
     int32_t i = 0, j = 0;
+
+    // lock pTable->lastCols[i] as it would be released when schema update(tsdbUpdateLastColSchema)
+    TSDB_RLOCK_TABLE(pTable);
     while(i < tgNumOfCols && j < numOfCols) {
       pColInfo = taosArrayGet(pQueryHandle->pColumns, i);
       if (pTable->lastCols[j].colId < pColInfo->info.colId) {
@@ -2867,6 +2896,7 @@ static bool loadCachedLast(STsdbQueryHandle* pQueryHandle) {
       i++;
       j++;
     }
+    TSDB_RUNLOCK_TABLE(pTable);
 
     // leave the real ts column as the last row, because last function only (not stable) use the last row as res
     if (priKey != TSKEY_INITIAL_VAL) {
@@ -3198,7 +3228,9 @@ int32_t checkForCachedLast(STsdbQueryHandle* pQueryHandle) {
 
   int32_t code = 0;
   
-  if (pQueryHandle->pTsdb && atomic_load_8(&pQueryHandle->pTsdb->hasCachedLastColumn)){
+  STsdbRepo* pRepo = pQueryHandle->pTsdb;
+
+  if (pRepo && CACHE_LAST_NULL_COLUMN(&(pRepo->config))) {
     pQueryHandle->cachelastrow = TSDB_CACHED_TYPE_LAST;
   }
 

tests/pytest/tools/taosdemoTestTblAlt.py

@@ -56,7 +56,7 @@ class TDTestCase:
         if(threadID == 0):
             print("%staosdemo -y -t %d -n %d -b INT,INT,INT,INT -m t" %
                       (binPath, self.numberOfTables, self.numberOfRecords))
-            os.system("%staosdemo -y -t %d -n %d -b INT,INT,INT,INT" %
+            os.system("%staosdemo -y -t %d -n %d -b INT,INT,INT,INT -m t" %
                       (binPath, self.numberOfTables, self.numberOfRecords))
         if(threadID == 1):
             time.sleep(2)