refactor commit process

src/tsdb/inc/tsdbMain.h

@@ -96,6 +96,11 @@ typedef struct {
 } STsdbBufPool;
 
 // ------------------ tsdbMemTable.c
+typedef struct {
+  STable *           pTable;
+  SSkipListIterator *pIter;
+} SCommitIter;
+
 typedef struct {
   uint64_t   uid;
   TSKEY      keyFirst;
@@ -377,6 +382,24 @@ int   tsdbUnRefMemTable(STsdbRepo* pRepo, SMemTable* pMemTable);
 int   tsdbTakeMemSnapshot(STsdbRepo* pRepo, SMemTable** pMem, SMemTable** pIMem);
 void* tsdbAllocBytes(STsdbRepo* pRepo, int bytes);
 int   tsdbAsyncCommit(STsdbRepo* pRepo);
+int   tsdbLoadDataFromCache(STable* pTable, SSkipListIterator* pIter, TSKEY maxKey, int maxRowsToRead, SDataCols* pCols,
+                            TSKEY* filterKeys, int nFilterKeys);
+
+static FORCE_INLINE SDataRow tsdbNextIterRow(SSkipListIterator* pIter) {
+  if (pIter == NULL) return NULL;
+
+  SSkipListNode* node = tSkipListIterGet(pIter);
+  if (node == NULL) return NULL;
+
+  return SL_GET_NODE_DATA(node);
+}
+
+static FORCE_INLINE TSKEY tsdbNextIterKey(SSkipListIterator* pIter) {
+  SDataRow row = tsdbNextIterRow(pIter);
+  if (row == NULL) return -1;
+
+  return dataRowKey(row);
+}
 
 // ------------------ tsdbFile.c
 #define TSDB_KEY_FILEID(key, daysPerFile, precision) ((key) / tsMsPerDay[(precision)] / (daysPerFile))
@@ -421,25 +444,36 @@ void        tsdbRemoveFileGroup(STsdbRepo* pRepo, SFileGroup* pFGroup);
 #define helperType(h) (h)->type
 #define helperRepo(h) (h)->pRepo
 #define helperState(h) (h)->state
-
-int   tsdbInitReadHelper(SRWHelper* pHelper, STsdbRepo* pRepo);
-int   tsdbInitWriteHelper(SRWHelper* pHelper, STsdbRepo* pRepo);
-void  tsdbDestroyHelper(SRWHelper* pHelper);
-void  tsdbResetHelper(SRWHelper* pHelper);
-int   tsdbSetAndOpenHelperFile(SRWHelper* pHelper, SFileGroup* pGroup);
-int   tsdbCloseHelperFile(SRWHelper* pHelper, bool hasError);
-void  tsdbSetHelperTable(SRWHelper* pHelper, STable* pTable, STsdbRepo* pRepo);
-int   tsdbWriteDataBlock(SRWHelper* pHelper, SDataCols* pDataCols);
-int   tsdbMoveLastBlockIfNeccessary(SRWHelper* pHelper);
-int   tsdbWriteCompInfo(SRWHelper* pHelper);
-int   tsdbWriteCompIdx(SRWHelper* pHelper);
-int   tsdbLoadCompIdx(SRWHelper* pHelper, void* target);
-int   tsdbLoadCompInfo(SRWHelper* pHelper, void* target);
-int   tsdbLoadCompData(SRWHelper* phelper, SCompBlock* pcompblock, void* target);
-void  tsdbGetDataStatis(SRWHelper* pHelper, SDataStatis* pStatis, int numOfCols);
-int   tsdbLoadBlockDataCols(SRWHelper* pHelper, SCompBlock* pCompBlock, SCompInfo* pCompInfo, int16_t* colIds,
-                            int numOfColIds);
-int   tsdbLoadBlockData(SRWHelper* pHelper, SCompBlock* pCompBlock, SCompInfo* pCompInfo);
+#define TSDB_NLAST_FILE_OPENED(h) ((h)->files.nLastF.fd > 0)
+
+int  tsdbInitReadHelper(SRWHelper* pHelper, STsdbRepo* pRepo);
+int  tsdbInitWriteHelper(SRWHelper* pHelper, STsdbRepo* pRepo);
+void tsdbDestroyHelper(SRWHelper* pHelper);
+void tsdbResetHelper(SRWHelper* pHelper);
+int  tsdbSetAndOpenHelperFile(SRWHelper* pHelper, SFileGroup* pGroup);
+int  tsdbCloseHelperFile(SRWHelper* pHelper, bool hasError);
+void tsdbSetHelperTable(SRWHelper* pHelper, STable* pTable, STsdbRepo* pRepo);
+int  tsdbCommitTableData(SRWHelper* pHelper, SCommitIter* pCommitIter, SDataCols* pDataCols, TSKEY maxKey);
+int  tsdbMoveLastBlockIfNeccessary(SRWHelper* pHelper);
+int  tsdbWriteCompInfo(SRWHelper* pHelper);
+int  tsdbWriteCompIdx(SRWHelper* pHelper);
+int  tsdbLoadCompIdx(SRWHelper* pHelper, void* target);
+int  tsdbLoadCompInfo(SRWHelper* pHelper, void* target);
+int  tsdbLoadCompData(SRWHelper* phelper, SCompBlock* pcompblock, void* target);
+void tsdbGetDataStatis(SRWHelper* pHelper, SDataStatis* pStatis, int numOfCols);
+int  tsdbLoadBlockDataCols(SRWHelper* pHelper, SCompBlock* pCompBlock, SCompInfo* pCompInfo, int16_t* colIds,
+                           int numOfColIds);
+int  tsdbLoadBlockData(SRWHelper* pHelper, SCompBlock* pCompBlock, SCompInfo* pCompInfo);
+
+static FORCE_INLINE int compTSKEY(const void* key1, const void* key2) {
+  if (*(TSKEY*)key1 > *(TSKEY*)key2) {
+    return 1;
+  } else if (*(TSKEY*)key1 == *(TSKEY*)key2) {
+    return 0;
+  } else {
+    return -1;
+  }
+}
 
 // ------------------ tsdbMain.c
 #define REPO_ID(r) (r)->config.tsdbId

src/tsdb/src/tsdbMemTable.c

@@ -18,11 +18,6 @@
 
 #define TSDB_DATA_SKIPLIST_LEVEL 5
 
-typedef struct {
-  STable *           pTable;
-  SSkipListIterator *pIter;
-} SCommitIter;
-
 static FORCE_INLINE STsdbBufBlock *tsdbGetCurrBufBlock(STsdbRepo *pRepo);
 
 static void        tsdbFreeBytes(STsdbRepo *pRepo, void *ptr, int bytes);
@@ -34,14 +29,11 @@ static char *      tsdbGetTsTupleKey(const void *data);
 static void *      tsdbCommitData(void *arg);
 static int         tsdbCommitMeta(STsdbRepo *pRepo);
 static void        tsdbEndCommit(STsdbRepo *pRepo);
-static TSKEY       tsdbNextIterKey(SCommitIter *pIter);
 static int         tsdbHasDataToCommit(SCommitIter *iters, int nIters, TSKEY minKey, TSKEY maxKey);
 static int  tsdbCommitToFile(STsdbRepo *pRepo, int fid, SCommitIter *iters, SRWHelper *pHelper, SDataCols *pDataCols);
 static void tsdbGetFidKeyRange(int daysPerFile, int8_t precision, int fileId, TSKEY *minKey, TSKEY *maxKey);
 static SCommitIter *tsdbCreateTableIters(STsdbRepo *pRepo);
 static void         tsdbDestroyTableIters(SCommitIter *iters, int maxTables);
-static int          tsdbReadRowsFromCache(STsdbMeta *pMeta, STable *pTable, SSkipListIterator *pIter, TSKEY maxKey,
-                                          int maxRowsToRead, SDataCols *pCols);
 
 // ---------------- INTERNAL FUNCTIONS ----------------
 int tsdbInsertRowToMem(STsdbRepo *pRepo, SDataRow row, STable *pTable) {
@@ -248,6 +240,66 @@ int tsdbAsyncCommit(STsdbRepo *pRepo) {
   return 0;
 }
 
+int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead, SDataCols *pCols,
+                          TSKEY *filterKeys, int nFilterKeys) {
+  ASSERT(maxRowsToRead > 0 && nFilterKeys >= 0);
+  if (pIter == NULL) return 0;
+  STSchema *pSchema = NULL;
+  int       numOfRows = 0;
+  TSKEY     keyNext = 0;
+  int       filterIter = 0;
+
+  if (nFilterKeys != 0) { // for filter purpose
+    ASSERT(filterKeys != NULL);
+    keyNext = tsdbNextIterKey(pIter);
+    if (keyNext < 0 || keyNext > maxKey) return numOfRows;
+    void *ptr = taosbsearch((void *)(&keyNext), (void *)filterKeys, nFilterKeys, sizeof(TSKEY), compTSKEY, TD_GE);
+    filterIter = (ptr == NULL) ? nFilterKeys : (POINTER_DISTANCE(ptr, filterKeys) / sizeof(TSKEY));
+  }
+
+  do {
+    if (numOfRows >= maxRowsToRead) break;
+
+    SDataRow row = tsdbNextIterRow(pIter);
+    if (row == NULL) break;
+
+    keyNext = dataRowKey(row);
+    if (keyNext < 0 || keyNext > maxKey) break;
+
+    bool keyFiltered = false;
+    if (nFilterKeys != 0) {
+      while (true) {
+        if (filterIter >= nFilterKeys) break;
+        if (keyNext == filterKeys[filterIter]) {
+          keyFiltered = true;
+          filterIter++;
+          break;
+        } else if (keyNext < filterKeys[filterIter]) {
+          break;
+        } else {
+          filterIter++;
+        }
+      }
+    }
+
+    if (!keyFiltered) {
+      if (pCols) {
+        if (pSchema == NULL || schemaVersion(pSchema) != dataRowVersion(row)) {
+          pSchema = tsdbGetTableSchemaImpl(pTable, false, false, dataRowVersion(row));
+          if (pSchema == NULL) {
+            ASSERT(0);
+          }
+        }
+
+        tdAppendDataRowToDataCol(row, pSchema, pCols);
+      }
+      numOfRows++;
+    }
+  } while (tSkipListIterNext(pIter));
+
+  return numOfRows;
+}
+
 // ---------------- LOCAL FUNCTIONS ----------------
 static FORCE_INLINE STsdbBufBlock *tsdbGetCurrBufBlock(STsdbRepo *pRepo) {
   ASSERT(pRepo != NULL);
@@ -475,19 +527,9 @@ static void tsdbEndCommit(STsdbRepo *pRepo) {
   if (pRepo->appH.notifyStatus) pRepo->appH.notifyStatus(pRepo->appH.appH, TSDB_STATUS_COMMIT_OVER);
 }
 
-static TSKEY tsdbNextIterKey(SCommitIter *pIter) {
-  if (pIter == NULL) return -1;
-
-  SSkipListNode *node = tSkipListIterGet(pIter->pIter);
-  if (node == NULL) return -1;
-
-  SDataRow row = SL_GET_NODE_DATA(node);
-  return dataRowKey(row);
-}
-
 static int tsdbHasDataToCommit(SCommitIter *iters, int nIters, TSKEY minKey, TSKEY maxKey) {
   for (int i = 0; i < nIters; i++) {
-    TSKEY nextKey = tsdbNextIterKey(iters + i);
+    TSKEY nextKey = tsdbNextIterKey((iters + i)->pIter);
     if (nextKey > 0 && (nextKey >= minKey && nextKey <= maxKey)) return 1;
   }
   return 0;
@@ -500,7 +542,6 @@ static void tsdbGetFidKeyRange(int daysPerFile, int8_t precision, int fileId, TS
 
 static int tsdbCommitToFile(STsdbRepo *pRepo, int fid, SCommitIter *iters, SRWHelper *pHelper, SDataCols *pDataCols) {
   char *      dataDir = NULL;
-  STsdbMeta * pMeta = pRepo->tsdbMeta;
   STsdbCfg *  pCfg = &pRepo->config;
   STsdbFileH *pFileH = pRepo->tsdbFileH;
   SFileGroup *pGroup = NULL;
@@ -545,33 +586,13 @@ static int tsdbCommitToFile(STsdbRepo *pRepo, int fid, SCommitIter *iters, SRWHe
     if (pIter->pIter != NULL) {
       tdInitDataCols(pDataCols, tsdbGetTableSchemaImpl(pIter->pTable, false, false, -1));
 
-      int maxRowsToRead = pCfg->maxRowsPerFileBlock * 4 / 5;
-      int nLoop = 0;
-      while (true) {
-        int rowsRead = tsdbReadRowsFromCache(pMeta, pIter->pTable, pIter->pIter, maxKey, maxRowsToRead, pDataCols);
-        ASSERT(rowsRead >= 0);
-        if (pDataCols->numOfRows == 0) break;
-        nLoop++;
-
-        ASSERT(dataColsKeyFirst(pDataCols) >= minKey && dataColsKeyFirst(pDataCols) <= maxKey);
-        ASSERT(dataColsKeyLast(pDataCols) >= minKey && dataColsKeyLast(pDataCols) <= maxKey);
-
-        int rowsWritten = tsdbWriteDataBlock(pHelper, pDataCols);
-        ASSERT(rowsWritten != 0);
-        if (rowsWritten < 0) {
-          taosRUnLockLatch(&(pIter->pTable->latch));
-          tsdbError("vgId:%d failed to write data block to table %s tid %d uid %" PRIu64 " since %s", REPO_ID(pRepo),
-                    TABLE_CHAR_NAME(pIter->pTable), TABLE_TID(pIter->pTable), TABLE_UID(pIter->pTable),
-                    tstrerror(terrno));
-          goto _err;
-        }
-        ASSERT(rowsWritten <= pDataCols->numOfRows);
-
-        tdPopDataColsPoints(pDataCols, rowsWritten);
-        maxRowsToRead = pCfg->maxRowsPerFileBlock * 4 / 5 - pDataCols->numOfRows;
+      if (tsdbCommitTableData(pHelper, pIter, pDataCols, maxKey) < 0) {
+        taosRUnLockLatch(&(pIter->pTable->latch));
+        tsdbError("vgId:%d failed to write data of table %s tid %d uid %" PRIu64 " since %s", REPO_ID(pRepo),
+                  TABLE_CHAR_NAME(pIter->pTable), TABLE_TID(pIter->pTable), TABLE_UID(pIter->pTable),
+                  tstrerror(terrno));
+        goto _err;
       }
-
-      ASSERT(pDataCols->numOfRows == 0);
     }
 
     taosRUnLockLatch(&(pIter->pTable->latch));
@@ -666,35 +687,4 @@ static void tsdbDestroyTableIters(SCommitIter *iters, int maxTables) {
   }
 
   free(iters);
-}
-
-static int tsdbReadRowsFromCache(STsdbMeta *pMeta, STable *pTable, SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead, SDataCols *pCols) {
-  ASSERT(maxRowsToRead > 0);
-  if (pIter == NULL) return 0;
-  STSchema *pSchema = NULL;
-
-  int numOfRows = 0;
-
-  do {
-    if (numOfRows >= maxRowsToRead) break;
-
-    SSkipListNode *node = tSkipListIterGet(pIter);
-    if (node == NULL) break;
-
-    SDataRow row = SL_GET_NODE_DATA(node);
-    if (dataRowKey(row) > maxKey) break;
-
-    if (pSchema == NULL || schemaVersion(pSchema) != dataRowVersion(row)) {
-      pSchema = tsdbGetTableSchemaImpl(pTable, true, false, dataRowVersion(row));
-      if (pSchema == NULL) {
-        // TODO: deal with the error here
-        ASSERT(0);
-      }
-    }
-
-    tdAppendDataRowToDataCol(row, pSchema, pCols);
-    numOfRows++;
-  } while (tSkipListIterNext(pIter));
-
-  return numOfRows;
 }
\ No newline at end of file