Merge pull request #14128 from taosdata/feature/wal

feat(wal): provide is empty api

include/libs/wal/wal.h

@@ -195,7 +195,6 @@ void            walCloseReadHandle(SWalReadHandle *);
 int32_t         walReadWithHandle(SWalReadHandle *pRead, int64_t ver);
 
 // only for tq usage
-// int32_t walReadWithHandle_s(SWalReadHandle *pRead, int64_t ver, SWalReadHead **ppHead);
 void    walSetReaderCapacity(SWalReadHandle *pRead, int32_t capacity);
 int32_t walFetchHead(SWalReadHandle *pRead, int64_t ver, SWalHead *pHead);
 int32_t walFetchBody(SWalReadHandle *pRead, SWalHead **ppHead);
@@ -211,13 +210,8 @@ void     walCloseRef(SWalRef *);
 int32_t  walRefVer(SWalRef *, int64_t ver);
 int32_t  walUnrefVer(SWal *);
 
-// deprecated
-#if 0
-int32_t walRead(SWal *, SWalHead **, int64_t ver);
-int32_t walReadWithFp(SWal *, FWalWrite writeFp, int64_t verStart, int32_t readNum);
-#endif
-
 // lifecycle check
+bool    walIsEmpty(SWal *);
 int64_t walGetFirstVer(SWal *);
 int64_t walGetSnapshotVer(SWal *);
 int64_t walGetLastVer(SWal *);

source/libs/wal/src/walMeta.c

@@ -19,6 +19,8 @@
 #include "tref.h"
 #include "walInt.h"
 
+bool FORCE_INLINE walIsEmpty(SWal* pWal) { return pWal->vers.firstVer == -1; }
+
 int64_t FORCE_INLINE walGetFirstVer(SWal* pWal) { return pWal->vers.firstVer; }
 
 int64_t FORCE_INLINE walGetSnaphostVer(SWal* pWal) { return pWal->vers.snapshotVer; }

source/libs/wal/src/walWrite.c

@@ -141,7 +141,7 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
   // validate offset
   SWalHead head;
   ASSERT(taosValidFile(pLogTFile));
-  int size = taosReadFile(pLogTFile, &head, sizeof(SWalHead));
+  int64_t size = taosReadFile(pLogTFile, &head, sizeof(SWalHead));
   if (size != sizeof(SWalHead)) {
     return -1;
   }
@@ -149,22 +149,33 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
 
   ASSERT(code == 0);
   if (code != 0) {
+    terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
     return -1;
   }
   if (head.head.version != ver) {
-    // TODO
+    ASSERT(0);
+    terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
     return -1;
   }
+
   // truncate old files
   code = taosFtruncateFile(pLogTFile, entry.offset);
   if (code < 0) {
+    ASSERT(0);
+    terrno = TAOS_SYSTEM_ERROR(errno);
     return -1;
   }
   code = taosFtruncateFile(pIdxTFile, idxOff);
   if (code < 0) {
+    ASSERT(0);
+    terrno = TAOS_SYSTEM_ERROR(errno);
     return -1;
   }
   pWal->vers.lastVer = ver - 1;
+  if (pWal->vers.lastVer < pWal->vers.firstVer) {
+    ASSERT(pWal->vers.lastVer == pWal->vers.firstVer - 1);
+    pWal->vers.firstVer = -1;
+  }
   ((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->lastVer = ver - 1;
   ((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->fileSize = entry.offset;
   taosCloseFile(&pIdxTFile);