Merge pull request #20608 from taosdata/fix/TS-2960

fix(tdb/restore): rollback journal files backward

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

@@ -687,6 +687,8 @@ typedef struct SSttBlockLoadInfo {
   STSchema  *pSchema;
   int16_t   *colIds;
   int32_t    numOfCols;
+  bool       checkRemainingRow;
+  bool       isLast;
   bool       sttBlockLoaded;
   int32_t    numOfStt;
 

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

@@ -590,6 +590,7 @@ typedef struct {
   SDataFReader       **pDataFReader;
   TSDBROW              row;
 
+  bool               checkRemainingRow;
   SMergeTree         mergeTree;
   SMergeTree        *pMergeTree;
   SSttBlockLoadInfo *pLoadInfo;
@@ -600,7 +601,6 @@ static int32_t getNextRowFromFSLast(void *iter, TSDBROW **ppRow, bool *pIgnoreEa
                                     int nCols) {
   SFSLastNextRowIter *state = (SFSLastNextRowIter *)iter;
   int32_t             code = 0;
-  bool                checkRemainingRow = true;
 
   switch (state->state) {
     case SFSLASTNEXTROW_FS:
@@ -633,12 +633,25 @@ static int32_t getNextRowFromFSLast(void *iter, TSDBROW **ppRow, bool *pIgnoreEa
         if (code) goto _err;
       }
 
-      state->pLoadInfo->colIds = aCols;
-      state->pLoadInfo->numOfCols = nCols;
+      for (int i = 0; i < state->pLoadInfo->numOfStt; ++i) {
+        state->pLoadInfo[i].colIds = aCols;
+        state->pLoadInfo[i].numOfCols = nCols;
+        state->pLoadInfo[i].isLast = isLast;
+      }
       tMergeTreeOpen(&state->mergeTree, 1, *state->pDataFReader, state->suid, state->uid,
                      &(STimeWindow){.skey = state->lastTs, .ekey = TSKEY_MAX},
                      &(SVersionRange){.minVer = 0, .maxVer = UINT64_MAX}, state->pLoadInfo, false, NULL, true);
       state->pMergeTree = &state->mergeTree;
+      state->state = SFSLASTNEXTROW_BLOCKROW;
+    }
+    case SFSLASTNEXTROW_BLOCKROW: {
+      if (nCols != state->pLoadInfo->numOfCols) {
+        for (int i = 0; i < state->pLoadInfo->numOfStt; ++i) {
+          state->pLoadInfo[i].numOfCols = nCols;
+
+          state->pLoadInfo[i].checkRemainingRow = state->checkRemainingRow;
+        }
+      }
       bool hasVal = tMergeTreeNext(&state->mergeTree);
       if (!hasVal) {
         if (tMergeTreeIgnoreEarlierTs(&state->mergeTree)) {
@@ -649,76 +662,23 @@ static int32_t getNextRowFromFSLast(void *iter, TSDBROW **ppRow, bool *pIgnoreEa
         state->state = SFSLASTNEXTROW_FILESET;
         goto _next_fileset;
       }
-      state->state = SFSLASTNEXTROW_BLOCKROW;
-      checkRemainingRow = false;
-    }
-    case SFSLASTNEXTROW_BLOCKROW: {
-      bool skipRow = false;
-      do {
-        bool hasVal = false;
-        state->row = tMergeTreeGetRow(&state->mergeTree);
-        *ppRow = &state->row;
-        if (nCols != state->pLoadInfo->numOfCols) {
-          state->pLoadInfo->numOfCols = nCols;
-        }
-        hasVal = tMergeTreeNext(&state->mergeTree);
-        if (TSDBROW_TS(&state->row) <= state->lastTs) {
-          *pIgnoreEarlierTs = true;
-          *ppRow = NULL;
-          return code;
-        }
-
-        *pIgnoreEarlierTs = false;
-        if (!hasVal) {
-          state->state = SFSLASTNEXTROW_FILESET;
-          break;
-        }
-
-        if (checkRemainingRow) {
-          bool skipBlock = true;
-
-          SBlockData *pBlockData = state->row.pBlockData;
+      state->row = tMergeTreeGetRow(&state->mergeTree);
+      *ppRow = &state->row;
 
-          for (int inputColIndex = 0; inputColIndex < nCols; ++inputColIndex) {
-            for (int colIndex = 0; colIndex < pBlockData->nColData; ++colIndex) {
-              SColData *pColData = &pBlockData->aColData[colIndex];
-              int16_t   cid = pColData->cid;
-
-              if (cid == aCols[inputColIndex]) {
-                if (isLast && (pColData->flag & HAS_VALUE)) {
-                  skipBlock = false;
-                  break;
-                } else if (pColData->flag & (HAS_VALUE | HAS_NULL)) {
-                  skipBlock = false;
-                  break;
-                }
-              }
-            }
-          }
-          /*
-          for (int colIndex = 0; colIndex < pBlockData->nColData; ++colIndex) {
-            SColData *pColData = &pBlockData->aColData[colIndex];
-            int16_t   cid = pColData->cid;
-
-            if (inputColIndex < nCols && cid == aCols[inputColIndex]) {
-              if (isLast && (pColData->flag & HAS_VALUE)) {
-                skipBlock = false;
-                break;
-              } else if (pColData->flag & (HAS_VALUE | HAS_NULL)) {
-                skipBlock = false;
-                break;
-              }
+      if (TSDBROW_TS(&state->row) <= state->lastTs) {
+        *pIgnoreEarlierTs = true;
+        *ppRow = NULL;
+        return code;
+      }
 
-              ++inputColIndex;
-            }
-          }
-          */
-          if (skipBlock) {
-            skipRow = true;
-          }
-        }
-      } while (skipRow);
+      *pIgnoreEarlierTs = false;
+      if (!hasVal) {
+        state->state = SFSLASTNEXTROW_FILESET;
+      }
 
+      if (!state->checkRemainingRow) {
+        state->checkRemainingRow = true;
+      }
       return code;
     }
     default:

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

@@ -504,9 +504,34 @@ bool tLDataIterNextRow(SLDataIter *pIter, const char *idStr) {
   pIter->iRow += step;
 
   while (1) {
+    bool skipBlock = false;
+
     findNextValidRow(pIter, idStr);
 
-    if (pIter->iRow >= pBlockData->nRow || pIter->iRow < 0) {
+    if (pIter->pBlockLoadInfo->checkRemainingRow) {
+      skipBlock = true;
+      int16_t *aCols = pIter->pBlockLoadInfo->colIds;
+      int      nCols = pIter->pBlockLoadInfo->numOfCols;
+      bool     isLast = pIter->pBlockLoadInfo->isLast;
+      for (int inputColIndex = 0; inputColIndex < nCols; ++inputColIndex) {
+        for (int colIndex = 0; colIndex < pBlockData->nColData; ++colIndex) {
+          SColData *pColData = &pBlockData->aColData[colIndex];
+          int16_t   cid = pColData->cid;
+
+          if (cid == aCols[inputColIndex]) {
+            if (isLast && (pColData->flag & HAS_VALUE)) {
+              skipBlock = false;
+              break;
+            } else if (pColData->flag & (HAS_VALUE | HAS_NULL)) {
+              skipBlock = false;
+              break;
+            }
+          }
+        }
+      }
+    }
+
+    if (skipBlock || pIter->iRow >= pBlockData->nRow || pIter->iRow < 0) {
       tLDataIterNextBlock(pIter, idStr);
       if (pIter->pSttBlk == NULL) {  // no more data
         goto _exit;

source/libs/tdb/src/db/tdbPager.c

@@ -947,6 +947,12 @@ static int tdbPagerRestore(SPager *pPager, const char *jFileName) {
   return 0;
 }
 
+static int32_t txnIdCompareDesc(const void *pLeft, const void *pRight) {
+  int64_t lhs = *(int64_t *)pLeft;
+  int64_t rhs = *(int64_t *)pRight;
+  return lhs > rhs ? -1 : 1;
+}
+
 int tdbPagerRestoreJournals(SPager *pPager) {
   tdbDirEntryPtr pDirEntry;
   tdbDirPtr      pDir = taosOpenDir(pPager->pEnv->dbName);
@@ -955,23 +961,33 @@ int tdbPagerRestoreJournals(SPager *pPager) {
     return -1;
   }
 
+  SArray *pTxnList = taosArrayInit(16, sizeof(int64_t));
+
   while ((pDirEntry = tdbReadDir(pDir)) != NULL) {
     char *name = tdbDirEntryBaseName(tdbGetDirEntryName(pDirEntry));
     if (strncmp(TDB_MAINDB_NAME "-journal", name, 16) == 0) {
-      char jname[TD_PATH_MAX] = {0};
-      int  dirLen = strlen(pPager->pEnv->dbName);
-      memcpy(jname, pPager->pEnv->dbName, dirLen);
-      jname[dirLen] = '/';
-      memcpy(jname + dirLen + 1, name, strlen(name));
-      if (tdbPagerRestore(pPager, jname) < 0) {
-        tdbCloseDir(&pDir);
-
-        tdbError("failed to restore file due to %s. jFileName:%s", strerror(errno), name);
-        return -1;
-      }
+      int64_t txnId = -1;
+      sscanf(name, TDB_MAINDB_NAME "-journal.%" PRId64, &txnId);
+      taosArrayPush(pTxnList, &txnId);
+    }
+  }
+  taosArraySort(pTxnList, txnIdCompareDesc);
+  for (int i = 0; i < TARRAY_SIZE(pTxnList); ++i) {
+    int64_t *pTxnId = taosArrayGet(pTxnList, i);
+    char     jname[TD_PATH_MAX] = {0};
+    int      dirLen = strlen(pPager->pEnv->dbName);
+    memcpy(jname, pPager->pEnv->dbName, dirLen);
+    jname[dirLen] = '/';
+    sprintf(jname + dirLen + 1, TDB_MAINDB_NAME "-journal.%" PRId64, *pTxnId);
+    if (tdbPagerRestore(pPager, jname) < 0) {
+      tdbCloseDir(&pDir);
+
+      tdbError("failed to restore file due to %s. jFileName:%s", strerror(errno), jname);
+      return -1;
     }
   }
 
+  taosArrayDestroy(pTxnList);
   tdbCloseDir(&pDir);
 
   return 0;