Merge from origin/develop

src/dnode/src/dnodeMain.c

@@ -19,6 +19,7 @@
 #include "tutil.h"
 #include "tconfig.h"
 #include "tglobal.h"
+#include "twal.h"
 #include "dnode.h"
 #include "dnodeInt.h"
 #include "dnodeMgmt.h"
@@ -58,6 +59,7 @@ static const SDnodeComponent tsDnodeComponents[] = {
   {"dnodeeps",  dnodeInitEps,        dnodeCleanupEps},
   {"mnodeinfos",dnodeInitMInfos,     dnodeCleanupMInfos},
   {"globalcfg" ,taosCheckGlobalCfg,  NULL},
+  {"wal",       walInit,             walCleanUp},
   {"check",     dnodeInitCheck,      dnodeCleanupCheck},     // NOTES: dnodeInitCheck must be behind the dnodeinitStorage component !!!
   {"vread",     dnodeInitVnodeRead,  dnodeCleanupVnodeRead},
   {"vwrite",    dnodeInitVnodeWrite, dnodeCleanupVnodeWrite},

src/inc/tsync.h

@@ -68,7 +68,7 @@ typedef uint32_t (*FGetFileInfo)(void *ahandle, char *name, uint32_t *index, uin
 
 // get the wal file from index or after
 // return value, -1: error, 1:more wal files, 0:last WAL. if name[0]==0, no WAL file
-typedef int      (*FGetWalInfo)(void *ahandle, char *name, uint32_t *index); 
+typedef int32_t  (*FGetWalInfo)(void *ahandle, char *fileName, int64_t *fileId); 
  
 // when a forward pkt is received, call this to handle data 
 typedef int      (*FWriteToCache)(void *ahandle, void *pHead, int type);

src/inc/twal.h

@@ -19,9 +19,11 @@
 extern "C" {
 #endif
 
-#define TAOS_WAL_NOLOG   0
-#define TAOS_WAL_WRITE   1
-#define TAOS_WAL_FSYNC   2
+typedef enum {
+  TAOS_WAL_NOLOG = 0,
+  TAOS_WAL_WRITE = 1,
+  TAOS_WAL_FSYNC = 2
+} EWalType;
 
 typedef struct {
   int8_t   msgType;
@@ -34,8 +36,9 @@ typedef struct {
 } SWalHead;
 
 typedef struct {
-  int8_t  walLevel;     // wal level
+  int32_t vgId;
   int32_t fsyncPeriod;  // millisecond
+  int8_t  walLevel;     // wal level
   int8_t  wals;         // number of WAL files;
   int8_t  keep;         // keep the wal file when closed
 } SWalCfg;
@@ -43,14 +46,17 @@ typedef struct {
 typedef void* twalh;  // WAL HANDLE
 typedef int (*FWalWrite)(void *ahandle, void *pHead, int type);
 
-twalh   walOpen(const char *path, const SWalCfg *pCfg);
-int     walAlter(twalh pWal, const SWalCfg *pCfg);
+int32_t walInit();
+void    walCleanUp();
+
+twalh   walOpen(char *path, SWalCfg *pCfg);
+int32_t walAlter(twalh pWal, SWalCfg *pCfg);
 void    walClose(twalh);
-int     walRenew(twalh);
-int     walWrite(twalh, SWalHead *);
+int32_t walRenew(twalh);
+int32_t walWrite(twalh, SWalHead *);
 void    walFsync(twalh);
-int     walRestore(twalh, void *pVnode, FWalWrite writeFp);
-int     walGetWalFile(twalh, char *name, uint32_t *index);
+int32_t walRestore(twalh, void *pVnode, FWalWrite writeFp);
+int32_t walGetWalFile(twalh, char *fileName, int64_t *fileId);
 int64_t walGetVersion(twalh);
 
 #ifdef __cplusplus

src/mnode/src/mnodeSdb.c

@@ -175,7 +175,7 @@ static void *sdbGetTableFromId(int32_t tableId) {
 }
 
 static int32_t sdbInitWal() {
-  SWalCfg walCfg = {.walLevel = 2, .wals = 2, .keep = 1, .fsyncPeriod = 0};
+  SWalCfg walCfg = {.vgId = 1, .walLevel = 2, .wals = 2, .keep = 1, .fsyncPeriod = 0};
   char temp[TSDB_FILENAME_LEN];
   sprintf(temp, "%s/wal", tsMnodeDir);
   tsSdbObj.wal = walOpen(temp, &walCfg);
@@ -237,8 +237,8 @@ static uint32_t sdbGetFileInfo(void *ahandle, char *name, uint32_t *index, uint3
   return 0;
 }
 
-static int sdbGetWalInfo(void *ahandle, char *name, uint32_t *index) {
-  return walGetWalFile(tsSdbObj.wal, name, index);
+static int32_t sdbGetWalInfo(void *ahandle, char *fileName, int64_t *fileId) {
+  return walGetWalFile(tsSdbObj.wal, fileName, fileId);
 }
 
 static void sdbNotifyRole(void *ahandle, int8_t role) {

src/sync/src/syncRetrieve.c

@@ -287,7 +287,7 @@ static int syncRetrieveLastWal(SSyncPeer *pPeer, char *name, uint64_t fversion,
   return -1;
 }
 
-static int syncProcessLastWal(SSyncPeer *pPeer, char *wname, uint32_t index) {
+static int syncProcessLastWal(SSyncPeer *pPeer, char *wname, int64_t index) {
   SSyncNode *pNode = pPeer->pSyncNode;
   int        code = -1;
   char       fname[TSDB_FILENAME_LEN * 2];  // full path to wal file
@@ -377,7 +377,7 @@ static int syncRetrieveWal(SSyncPeer *pPeer) {
   int32_t     size;
   struct stat fstat;
   int         code = -1;
-  uint32_t    index = 0;
+  int64_t     index = 0;
 
   while (1) {
     // retrieve wal info

src/sync/test/syncServer.c

@@ -254,7 +254,7 @@ uint32_t getFileInfo(void *ahandle, char *name, uint32_t *index, uint32_t eindex
   return magic;
 }
 
-int getWalInfo(void *ahandle, char *name, uint32_t *index) {
+int getWalInfo(void *ahandle, char *name, int64_t *index) {
   struct stat fstat;
   char        aname[280];
 

src/wal/inc/walMgmt.h → src/util/inc/talloc.h

@@ -13,13 +13,31 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#ifndef TDENGINE_WAL_MGMT_H
-#define TDENGINE_WAL_MGMT_H
+#ifndef TDENGINE_UTIL_ALLOC_H
+#define TDENGINE_UTIL_ALLOC_H
 
 #ifdef __cplusplus
 extern "C" {
 #endif
 
+#define TSDB_USE_SYS_MEM
+
+#ifdef TSDB_USE_SYS_MEM
+  #define tmalloc(size) malloc(size)
+  #define tcalloc(size) calloc(1, size)
+  #define trealloc(p, size) realloc(p, size)
+  #define tmemalign(alignment, size) malloc(size)
+  #define tfree(p) free(p)
+  #define tmemzero(p, size) memset(p, 0, size)
+#else
+  void *tmalloc(int32_t size);
+  void *tcalloc(int32_t size);
+  void *trealloc(void *p, int32_t size);
+  void *tmemalign(int32_t alignment, int32_t size);
+  void  tfree(void *p);
+  void  tmemzero(void *p, int32_t size);
+#endif
+
 #ifdef __cplusplus
 }
 #endif

src/util/src/talloc.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#define _DEFAULT_SOURCE
+#include "os.h"
+#include "taoserror.h"
+#include "tulog.h"
+#include "talloc.h"
+
+#define TSDB_HAVE_MEMALIGN
+#ifndef TSDB_USE_SYS_MEM
+
+void *tmalloc(int32_t size) {
+  void *p = malloc(size);
+  if (p == NULL) {
+    terrno = TAOS_SYSTEM_ERROR(errno);
+    uError("failed to malloc memory, size:%d reason:%s", size, strerror(errno));
+  }
+
+  return p;
+}
+
+void *tcalloc(int32_t size) {
+  void *p = calloc(1, size);
+  if (p == NULL) {
+    terrno = TAOS_SYSTEM_ERROR(errno);
+    uError("failed to calloc memory, size:%d reason:%s", size, strerror(errno));
+  }
+
+  return p;
+}
+
+void *trealloc(void *p, int32_t size) {
+  p = realloc(p, size);
+  if (p == NULL) {
+    terrno = TAOS_SYSTEM_ERROR(errno);
+    uError("failed to realloc memory, size:%d reason:%s", size, strerror(errno));
+  }
+
+  return p;
+}
+
+void tfree(void *p) { free(p); }
+
+void tmemzero(void *p, int32_t size) { memset(p, 0, size); }
+
+#ifdef TSDB_HAVE_MEMALIGN
+
+void *tmemalign(int32_t alignment, int32_t size) {
+  void *p;
+
+  int err = posix_memalign(&p, alignment, size);
+  if (err) {
+    terrno = TAOS_SYSTEM_ERROR(errno);
+    uError("failed to memalign memory, alignment:%d size:%d reason:%s", alignment, size, strerror(err));
+    p = NULL;
+  }
+
+  return p;
+}
+
+#else
+
+void *tmemalign(int32_t alignment, int32_t size) { return tmalloc(size); }
+
+#endif
+#endif
\ No newline at end of file

src/vnode/src/vnodeMain.c

@@ -38,7 +38,7 @@ static SHashObj*tsDnodeVnodesHash;
 static void     vnodeCleanUp(SVnodeObj *pVnode);
 static int      vnodeProcessTsdbStatus(void *arg, int status);
 static uint32_t vnodeGetFileInfo(void *ahandle, char *name, uint32_t *index, uint32_t eindex, int64_t *size, uint64_t *fversion);
-static int      vnodeGetWalInfo(void *ahandle, char *name, uint32_t *index);
+static int      vnodeGetWalInfo(void *ahandle, char *fileName, int64_t *fileId);
 static void     vnodeNotifyRole(void *ahandle, int8_t role);
 static void     vnodeCtrlFlow(void *handle, int32_t mseconds); 
 static int      vnodeNotifyFileSynced(void *ahandle, uint64_t fversion);
@@ -298,6 +298,7 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
   }
 
   sprintf(temp, "%s/wal", rootDir);
+  pVnode->walCfg.vgId = pVnode->vgId;
   pVnode->wal = walOpen(temp, &pVnode->walCfg);
   if (pVnode->wal == NULL) { 
     vnodeCleanUp(pVnode);
@@ -618,9 +619,9 @@ static uint32_t vnodeGetFileInfo(void *ahandle, char *name, uint32_t *index, uin
   return tsdbGetFileInfo(pVnode->tsdb, name, index, eindex, size);
 }
 
-static int vnodeGetWalInfo(void *ahandle, char *name, uint32_t *index) {
+static int vnodeGetWalInfo(void *ahandle, char *fileName, int64_t *fileId) {
   SVnodeObj *pVnode = ahandle;
-  return walGetWalFile(pVnode->wal, name, index);
+  return walGetWalFile(pVnode->wal, fileName, fileId);
 }
 
 static void vnodeNotifyRole(void *ahandle, int8_t role) {

src/wal/inc/walInt.h

@@ -31,8 +31,32 @@ extern int32_t wDebugFlag;
 #define wDebug(...) { if (wDebugFlag & DEBUG_DEBUG) { taosPrintLog("WAL ", wDebugFlag, __VA_ARGS__); }}
 #define wTrace(...) { if (wDebugFlag & DEBUG_TRACE) { taosPrintLog("WAL ", wDebugFlag, __VA_ARGS__); }}
 
-#define walPrefix "wal"
-#define walSignature (uint32_t)(0xFAFBFDFE)
+#define WAL_PREFIX     "wal"
+#define WAL_PREFIX_LEN 3
+#define WAL_REFRESH_MS 1000
+#define WAL_MAX_SIZE   (1024 * 1024)
+#define WAL_SIGNATURE  ((uint32_t)(0xFAFBFDFE))
+#define WAL_PATH_LEN   (TSDB_FILENAME_LEN + 12)
+#define WAL_FILE_LEN   (TSDB_FILENAME_LEN + 32)
+#define WAL_FILE_NUM   3
+
+typedef struct {
+  uint64_t version;
+  int64_t  fileId;
+  int32_t  vgId;
+  int32_t  fd;
+  int32_t  keep;
+  int32_t  level;
+  int32_t  fsyncPeriod;
+  int32_t  fsyncSeq;
+  char     path[WAL_PATH_LEN];
+  char     name[WAL_FILE_LEN];
+  pthread_mutex_t mutex;
+} SWal;
+
+int32_t walGetNextFile(SWal *pWal, int64_t *nextFileId);
+int32_t walGetOldFile(SWal *pWal, int64_t curFileId, int32_t minDiff, int64_t *oldFileId);
+int32_t walGetNewFile(SWal *pWal, int64_t *newFileId);
 
 #ifdef __cplusplus
 }

src/wal/src/walMgmt.c

@@ -16,6 +16,223 @@
 #define _DEFAULT_SOURCE
 #include "os.h"
 #include "taoserror.h"
+#include "talloc.h"
+#include "tref.h"
 #include "twal.h"
 #include "walInt.h"
-#include "walMgmt.h"
\ No newline at end of file
+
+typedef struct {
+  int32_t   refId;
+  int32_t   seq;
+  int8_t    stop;
+  pthread_t thread;
+  pthread_mutex_t mutex;
+} SWalMgmt;
+
+static SWalMgmt tsWal;
+static int32_t  walCreateThread();
+static void     walStopThread();
+static int32_t  walInitObj(SWal *pWal);
+static void     walFreeObj(void *pWal);
+
+int32_t walInit() {
+  tmemzero(&tsWal, sizeof(SWalMgmt));
+  tsWal.refId = taosOpenRef(TSDB_MIN_VNODES, walFreeObj);
+
+  int32_t code = walCreateThread();
+  if (code != TSDB_CODE_SUCCESS) {
+    wError("failed to init wal module since %s", tstrerror(code));
+    return code;
+  }
+
+  wInfo("wal module is initialized, refId:%d", tsWal.refId);
+  return code;
+}
+
+void walCleanUp() {
+  walStopThread();
+  taosCloseRef(tsWal.refId);
+  wInfo("wal module is cleaned up");
+}
+
+void *walOpen(char *path, SWalCfg *pCfg) {
+  SWal *pWal = tcalloc(sizeof(SWal));
+  if (pWal == NULL) {
+    terrno = TAOS_SYSTEM_ERROR(errno);
+    return NULL;
+  }
+
+  pWal->vgId = pCfg->vgId;
+  pWal->fd = -1;
+  pWal->fileId = -1;
+  pWal->level = pCfg->walLevel;
+  pWal->keep = pCfg->keep;
+  pWal->fsyncPeriod = pCfg->fsyncPeriod;
+  tstrncpy(pWal->path, path, sizeof(pWal->path));
+  pthread_mutex_init(&pWal->mutex, NULL);
+
+  pWal->fsyncSeq = pCfg->fsyncPeriod % 1000;
+  if (pWal->fsyncSeq <= 0) pWal->fsyncSeq = 1;
+
+  if (walInitObj(pWal) != TSDB_CODE_SUCCESS) {
+    walFreeObj(pWal);
+    return NULL;
+  }
+
+  if (taosAddRef(tsWal.refId, pWal) != TSDB_CODE_SUCCESS) {
+    walFreeObj(pWal);
+    return NULL;
+  }
+
+  wDebug("vgId:%d, wal:%p is opened, level:%d fsyncPeriod:%d", pWal->vgId, pWal, pWal->level, pWal->fsyncPeriod);
+
+  return pWal;
+}
+
+int32_t walAlter(void *handle, SWalCfg *pCfg) {
+  if (handle == NULL) return TSDB_CODE_WAL_APP_ERROR;
+  SWal *pWal = handle;
+
+  if (pWal->level == pCfg->walLevel && pWal->fsyncPeriod == pCfg->fsyncPeriod) {
+    wDebug("vgId:%d, old walLevel:%d fsync:%d, new walLevel:%d fsync:%d not change", pWal->vgId, pWal->level,
+           pWal->fsyncPeriod, pCfg->walLevel, pCfg->fsyncPeriod);
+    return TSDB_CODE_SUCCESS;
+  }
+
+  wInfo("vgId:%d, change old walLevel:%d fsync:%d, new walLevel:%d fsync:%d", pWal->vgId, pWal->level,
+        pWal->fsyncPeriod, pCfg->walLevel, pCfg->fsyncPeriod);
+
+  pWal->level = pCfg->walLevel;
+  pWal->fsyncPeriod = pCfg->fsyncPeriod;
+  pWal->fsyncSeq = pCfg->fsyncPeriod % 1000;
+  if (pWal->fsyncSeq <= 0) pWal->fsyncSeq = 1;
+
+  return TSDB_CODE_SUCCESS;
+}
+
+void walClose(void *handle) {
+  if (handle == NULL) return;
+
+  SWal *pWal = handle;
+  pthread_mutex_lock(&pWal->mutex);
+
+  taosClose(pWal->fd);
+
+  if (!pWal->keep) {
+    int64_t fileId = -1;
+    while (walGetNextFile(pWal, &fileId) >= 0) {
+      snprintf(pWal->name, sizeof(pWal->name), "%s/%s%" PRId64, pWal->path, WAL_PREFIX, fileId);
+
+      if (fileId == pWal->fileId) {
+        wDebug("vgId:%d, wal:%p file:%s, it is closed and kept", pWal->vgId, pWal, pWal->name);
+      } else if (remove(pWal->name) < 0) {
+        wError("vgId:%d, wal:%p file:%s, failed to remove", pWal->vgId, pWal, pWal->name);
+      } else {
+        wDebug("vgId:%d, wal:%p file:%s, it is removed", pWal->vgId, pWal, pWal->name);
+      }
+    }
+  } else {
+    wDebug("vgId:%d, wal:%p file:%s, it is closed and kept", pWal->vgId, pWal, pWal->name);
+  }
+
+  pthread_mutex_unlock(&pWal->mutex);
+  taosRemoveRef(tsWal.refId, pWal);
+}
+
+static int32_t walInitObj(SWal *pWal) {
+  if (taosMkDir(pWal->path, 0755) != 0) {
+    wError("vgId:%d, file:%s, failed to create directory since %s", pWal->vgId, pWal->path, strerror(errno));
+    return TAOS_SYSTEM_ERROR(errno);
+  }
+
+  if (pWal->keep) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  walRenew(pWal);
+
+  if (pWal && pWal->fd < 0) {
+    wError("vgId:%d, file:%s, failed to open file since %s", pWal->vgId, pWal->path, strerror(errno));
+    return TAOS_SYSTEM_ERROR(errno);
+  }
+
+  wDebug("vgId:%d, file is initialized", pWal->vgId);
+  return TSDB_CODE_SUCCESS;
+}
+
+static void walFreeObj(void *wal) {
+  SWal *pWal = wal;
+  wDebug("vgId:%d, wal:%p is freed", pWal->vgId, pWal);
+
+  taosClose(pWal->fd);
+  pthread_mutex_destroy(&pWal->mutex);
+  tfree(pWal);
+}
+
+static bool walNeedFsync(SWal *pWal) {
+  if (pWal->fsyncPeriod <= 0 || pWal->level != TAOS_WAL_FSYNC) {
+    return false;
+  }
+
+  if (tsWal.seq % pWal->fsyncSeq == 0) {
+    return true;
+  }
+
+  return false;
+}
+
+static void walUpdateSeq() {
+  taosMsleep(WAL_REFRESH_MS);
+  if (++tsWal.seq <= 0) {
+    tsWal.seq = 1;
+  }
+}
+
+static void walFsyncAll() {
+  SWal *pWal = taosIterateRef(tsWal.refId, NULL);
+  while (pWal) {
+    if (walNeedFsync(pWal)) {
+      wTrace("vgId:%d, do fsync, level:%d seq:%d rseq:%d", pWal->vgId, pWal->level, pWal->fsyncSeq, tsWal.seq);
+      int32_t code = fsync(pWal->fd);
+      if (code != 0) {
+        wError("vgId:%d, file:%s, failed to fsync since %s", pWal->vgId, pWal->name, strerror(code));
+      }
+    }
+    pWal = taosIterateRef(tsWal.refId, pWal);
+  }
+}
+
+static void *walThreadFunc(void *param) {
+  while (1) {
+    walUpdateSeq();
+    walFsyncAll();
+    if (tsWal.stop) break;
+  }
+
+  return NULL;
+}
+
+static int32_t walCreateThread() {
+  pthread_attr_t thAttr;
+  pthread_attr_init(&thAttr);
+  pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE);
+
+  if (pthread_create(&tsWal.thread, &thAttr, walThreadFunc, NULL) != 0) {
+    wError("failed to create wal thread since %s", strerror(errno));
+    return TAOS_SYSTEM_ERROR(errno);
+  }
+
+  pthread_attr_destroy(&thAttr);
+  wDebug("wal thread is launched");
+
+  return TSDB_CODE_SUCCESS;
+}
+
+static void walStopThread() {
+  tsWal.stop = 1;
+  if (tsWal.thread) {
+    pthread_join(tsWal.thread, NULL);
+  }
+
+  wDebug("wal thread is stopped");
+}

src/wal/src/walUtil.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#define _DEFAULT_SOURCE
+#include "os.h"
+#include "walInt.h"
+
+int32_t walGetNextFile(SWal *pWal, int64_t *nextFileId) {
+  int64_t curFileId = *nextFileId;
+  int64_t minFileId = INT64_MAX;
+
+  DIR *dir = opendir(pWal->path);
+  if (dir == NULL) {
+    wError("vgId:%d, path:%s, failed to open since %s", pWal->vgId, pWal->path, strerror(errno));
+    return -1;
+  }
+
+  struct dirent *ent;
+  while ((ent = readdir(dir)) != NULL) {
+    char *name = ent->d_name;
+
+    if (strncmp(name, WAL_PREFIX, WAL_PREFIX_LEN) == 0) {
+      int64_t id = atoll(name + WAL_PREFIX_LEN);
+      if (id <= curFileId) continue;
+
+      if (id < minFileId) {
+        minFileId = id;
+      }
+    }
+  }
+  closedir(dir);
+
+  if (minFileId == INT64_MAX) return -1;
+
+  *nextFileId = minFileId;
+  wTrace("vgId:%d, path:%s, curFileId:%" PRId64 " nextFileId:%" PRId64, pWal->vgId, pWal->path, curFileId, *nextFileId);
+
+  return 0;
+}
+
+int32_t walGetOldFile(SWal *pWal, int64_t curFileId, int32_t minDiff, int64_t *oldFileId) {
+  int64_t minFileId = INT64_MAX;
+
+  DIR *dir = opendir(pWal->path);
+  if (dir == NULL) {
+    wError("vgId:%d, path:%s, failed to open since %s", pWal->vgId, pWal->path, strerror(errno));
+    return -1;
+  }
+
+  struct dirent *ent;
+  while ((ent = readdir(dir)) != NULL) {
+    char *name = ent->d_name;
+
+    if (strncmp(name, WAL_PREFIX, WAL_PREFIX_LEN) == 0) {
+      int64_t id = atoll(name + WAL_PREFIX_LEN);
+      if (id >= curFileId) continue;
+
+      minDiff--;
+      if (id < minFileId) {
+        minFileId = id;
+      }
+    }
+  }
+  closedir(dir);
+
+  if (minFileId == INT64_MAX) return -1;
+  if (minDiff > 0) return -1;
+
+  *oldFileId = minFileId;
+  wTrace("vgId:%d, path:%s, curFileId:%" PRId64 " oldFildId:%" PRId64, pWal->vgId, pWal->path, curFileId, *oldFileId);
+
+  return 0;
+}
+
+int32_t walGetNewFile(SWal *pWal, int64_t *newFileId) {
+  int64_t maxFileId = INT64_MIN;
+
+  DIR *dir = opendir(pWal->path);
+  if (dir == NULL) {
+    wError("vgId:%d, path:%s, failed to open since %s", pWal->vgId, pWal->path, strerror(errno));
+    return -1;
+  }
+
+  struct dirent *ent;
+  while ((ent = readdir(dir)) != NULL) {
+    char *name = ent->d_name;
+
+    if (strncmp(name, WAL_PREFIX, WAL_PREFIX_LEN) == 0) {
+      int64_t id = atoll(name + WAL_PREFIX_LEN);
+      if (id > maxFileId) {
+        maxFileId = id;
+      }
+    }
+  }
+  closedir(dir);
+
+  if (maxFileId == INT64_MIN) {
+    *newFileId = 0;
+  } else {
+    *newFileId = maxFileId;
+  }
+
+  wTrace("vgId:%d, path:%s, newFileId:%" PRId64, pWal->vgId, pWal->path, *newFileId);
+
+  return 0;
+}
\ No newline at end of file

src/wal/test/waltest.c

@@ -115,17 +115,17 @@ int main(int argc, char *argv[]) {
 
   printf("%d wal files are written\n", total);
 
-  uint32_t index = 0;
-  char     name[256];
+  int64_t index = 0;
+  char    name[256];
 
   while (1) {
     int code = walGetWalFile(pWal, name, &index);
     if (code == -1) {
-      printf("failed to get wal file, index:%d\n", index);
+      printf("failed to get wal file, index:%" PRId64 "\n", index);
       break;
     }
 
-    printf("index:%d wal:%s\n", index, name);
+    printf("index:%" PRId64 " wal:%s\n", index, name);
     if (code == 0) break;
 
     index++;

tests/script/unique/dnode/offline1.sim

@@ -49,7 +49,7 @@ print dnode1 $data4_2
 if $data4_1 != ready then
   return -1
 endi
-if $data4_2 != offline then
+if $data4_2 == ready then
   return -1
 endi