/*
 * 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/>.
 */

#include "os.h"
#include "tsdbint.h"
#include <regex.h>

typedef enum { TSDB_TXN_TEMP_FILE = 0, TSDB_TXN_CURR_FILE } TSDB_TXN_FILE_T;
static const char *tsdbTxnFname[] = {"current.t", "current"};
#define TSDB_MAX_FSETS(keep, days) ((keep) / (days) + 3)

static int  tsdbComparFidFSet(const void *arg1, const void *arg2);
static void tsdbResetFSStatus(SFSStatus *pStatus);
static int  tsdbSaveFSStatus(SFSStatus *pStatus, int vid);
static void tsdbApplyFSTxnOnDisk(SFSStatus *pFrom, SFSStatus *pTo);
static void tsdbGetTxnFname(int repoid, TSDB_TXN_FILE_T ftype, char fname[]);
static int  tsdbOpenFSFromCurrent(STsdb *pRepo);
static int  tsdbScanAndTryFixFS(STsdb *pRepo);
static int  tsdbScanRootDir(STsdb *pRepo);
static int  tsdbScanDataDir(STsdb *pRepo);
static bool tsdbIsTFileInFS(STsdbFS *pfs, const TFILE *pf);
static int  tsdbRestoreCurrent(STsdb *pRepo);
static int  tsdbComparTFILE(const void *arg1, const void *arg2);
static void tsdbScanAndTryFixDFilesHeader(STsdb *pRepo, int32_t *nExpired);
static int  tsdbProcessExpiredFS(STsdb *pRepo);
static int  tsdbCreateMeta(STsdb *pRepo);
static int  tsdbFetchTFileSet(STsdb *pRepo, SArray **fArray);

// For backward compatibility
// ================== CURRENT file header info
static int tsdbEncodeFSHeader(void **buf, SFSHeader *pHeader) {
  int tlen = 0;

  tlen += taosEncodeFixedU32(buf, pHeader->version);
  tlen += taosEncodeFixedU32(buf, pHeader->len);

  return tlen;
}

static void *tsdbDecodeFSHeader(void *buf, SFSHeader *pHeader) {
  buf = taosDecodeFixedU32(buf, &(pHeader->version));
  buf = taosDecodeFixedU32(buf, &(pHeader->len));

  return buf;
}

// ================== STsdbFSMeta
static int tsdbEncodeFSMeta(void **buf, STsdbFSMeta *pMeta) {
  int tlen = 0;

  tlen += taosEncodeFixedU32(buf, pMeta->version);
  tlen += taosEncodeFixedI64(buf, pMeta->totalPoints);
  tlen += taosEncodeFixedI64(buf, pMeta->totalStorage);

  return tlen;
}

static void *tsdbDecodeFSMeta(void *buf, STsdbFSMeta *pMeta) {
  buf = taosDecodeFixedU32(buf, &(pMeta->version));
  buf = taosDecodeFixedI64(buf, &(pMeta->totalPoints));
  buf = taosDecodeFixedI64(buf, &(pMeta->totalStorage));

  return buf;
}

// ================== SFSStatus
static int tsdbEncodeDFileSetArray(void **buf, SArray *pArray) {
  int      tlen = 0;
  uint64_t nset = taosArrayGetSize(pArray);

  tlen += taosEncodeFixedU64(buf, nset);
  for (size_t i = 0; i < nset; i++) {
    SDFileSet *pSet = taosArrayGet(pArray, i);

    tlen += tsdbEncodeDFileSet(buf, pSet);
  }

  return tlen;
}

static int tsdbDecodeDFileSetArray(void **originBuf, void *buf, SArray *pArray, SFSHeader *pSFSHeader) {
  uint64_t  nset;
  SDFileSet dset;
  dset.ver = TSDB_FSET_VER_0;  // default value

  taosArrayClear(pArray);

  buf = taosDecodeFixedU64(buf, &nset);

  if (pSFSHeader->version == TSDB_FS_VER_0) {
    // record fver in new version of 'current' file
    uint64_t extendedSize = pSFSHeader->len + nset * TSDB_FILE_MAX * sizeof(TSDB_FVER_TYPE); 
    if (taosTSizeof(*originBuf) < extendedSize) {
      size_t ptrDistance = POINTER_DISTANCE(buf, *originBuf);
      if (tsdbMakeRoom(originBuf, (size_t)extendedSize) < 0) {
        terrno = TSDB_CODE_FS_OUT_OF_MEMORY;
        return -1;
      }
      buf = POINTER_SHIFT(*originBuf, ptrDistance);
    }
  }

  for (size_t i = 0; i < nset; i++) {
    buf = tsdbDecodeDFileSet(buf, &dset, pSFSHeader->version);
    taosArrayPush(pArray, (void *)(&dset));
  }
  return TSDB_CODE_SUCCESS;
}

static int tsdbEncodeFSStatus(void **buf, SFSStatus *pStatus) {
  ASSERT(pStatus->pmf);

  int tlen = 0;

  tlen += tsdbEncodeSMFile(buf, pStatus->pmf);
  tlen += tsdbEncodeDFileSetArray(buf, pStatus->df);

  return tlen;
}

static int tsdbDecodeFSStatus(void **originBuf, void *buf, SFSStatus *pStatus, SFSHeader *pSFSHeader) {
  tsdbResetFSStatus(pStatus);
  pStatus->pmf = &(pStatus->mf);

  buf = tsdbDecodeSMFile(buf, pStatus->pmf);
  return tsdbDecodeDFileSetArray(originBuf, buf, pStatus->df, pSFSHeader);
}

static SFSStatus *tsdbNewFSStatus(int maxFSet) {
  SFSStatus *pStatus = (SFSStatus *)calloc(1, sizeof(*pStatus));
  if (pStatus == NULL) {
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    return NULL;
  }

  TSDB_FILE_SET_CLOSED(&(pStatus->mf));

  pStatus->df = taosArrayInit(maxFSet, sizeof(SDFileSet));
  if (pStatus->df == NULL) {
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    free(pStatus);
    return NULL;
  }

  return pStatus;
}

static SFSStatus *tsdbFreeFSStatus(SFSStatus *pStatus) {
  if (pStatus) {
    pStatus->df = taosArrayDestroy(pStatus->df);
    free(pStatus);
  }

  return NULL;
}

static void tsdbResetFSStatus(SFSStatus *pStatus) {
  if (pStatus == NULL) {
    return;
  }

  TSDB_FILE_SET_CLOSED(&(pStatus->mf));

  pStatus->pmf = NULL;
  taosArrayClear(pStatus->df);
}

static void tsdbSetStatusMFile(SFSStatus *pStatus, const SMFile *pMFile) {
  ASSERT(pStatus->pmf == NULL);

  pStatus->pmf = &(pStatus->mf);
  tsdbInitMFileEx(pStatus->pmf, (SMFile *)pMFile);
}

static int tsdbAddDFileSetToStatus(SFSStatus *pStatus, const SDFileSet *pSet) {
  if (taosArrayPush(pStatus->df, (void *)pSet) == NULL) {
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    return -1;
  }

  TSDB_FSET_SET_CLOSED(((SDFileSet *)taosArrayGetLast(pStatus->df)));

  return 0;
}

// ================== STsdbFS
STsdbFS *tsdbNewFS(STsdbCfg *pCfg) {
  int      keep = pCfg->keep;
  int      days = pCfg->daysPerFile;
  int      maxFSet = TSDB_MAX_FSETS(keep, days);
  STsdbFS *pfs;

  pfs = (STsdbFS *)calloc(1, sizeof(*pfs));
  if (pfs == NULL) {
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    return NULL;
  }

  int code = pthread_rwlock_init(&(pfs->lock), NULL);
  if (code) {
    terrno = TAOS_SYSTEM_ERROR(code);
    free(pfs);
    return NULL;
  }

  pfs->cstatus = tsdbNewFSStatus(maxFSet);
  if (pfs->cstatus == NULL) {
    tsdbFreeFS(pfs);
    return NULL;
  }

  pfs->metaCache = taosHashInit(4096, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);
  if (pfs->metaCache == NULL) {
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    tsdbFreeFS(pfs);
    return NULL;
  }

  pfs->intxn = false;
  pfs->metaCacheComp = NULL;

  pfs->nstatus = tsdbNewFSStatus(maxFSet);
  if (pfs->nstatus == NULL) {
    tsdbFreeFS(pfs);
    return NULL;
  }

  return pfs;
}

void *tsdbFreeFS(STsdbFS *pfs) {
  if (pfs) {
    pfs->nstatus = tsdbFreeFSStatus(pfs->nstatus);
    taosHashCleanup(pfs->metaCache);
    pfs->metaCache = NULL;
    pfs->cstatus = tsdbFreeFSStatus(pfs->cstatus);
    pthread_rwlock_destroy(&(pfs->lock));
    free(pfs);
  }

  return NULL;
}

static int tsdbProcessExpiredFS(STsdb *pRepo) {
  tsdbStartFSTxn(pRepo, 0, 0);
  if (tsdbCreateMeta(pRepo) < 0) {
    tsdbError("vgId:%d failed to create meta since %s", REPO_ID(pRepo), tstrerror(terrno));
    return -1;
  }

  if (tsdbApplyRtn(pRepo) < 0) {
    tsdbEndFSTxnWithError(REPO_FS(pRepo));
    tsdbError("vgId:%d failed to apply rtn since %s", REPO_ID(pRepo), tstrerror(terrno));
    return -1;
  }
  if (tsdbEndFSTxn(pRepo) < 0) {
    tsdbError("vgId:%d failed to end fs txn since %s", REPO_ID(pRepo), tstrerror(terrno));
    return -1;
  }
  return 0;
}

static int tsdbCreateMeta(STsdb *pRepo) {
  STsdbFS *pfs = REPO_FS(pRepo);
  SMFile * pOMFile = pfs->cstatus->pmf;
  SMFile   mf;
  SDiskID  did;

  if (pOMFile != NULL) {
    // keep the old meta file
    tsdbUpdateMFile(pfs, pOMFile);
    return 0;
  }

  // Create a new meta file
  did.level = TFS_PRIMARY_LEVEL;
  did.id = TFS_PRIMARY_ID;
  tsdbInitMFile(&mf, did, REPO_ID(pRepo), FS_TXN_VERSION(REPO_FS(pRepo)));

  if (tsdbCreateMFile(&mf, true) < 0) {
    tsdbError("vgId:%d failed to create META file since %s", REPO_ID(pRepo), tstrerror(terrno));
    return -1;
  }

  tsdbInfo("vgId:%d meta file %s is created", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(&mf));

  if (tsdbUpdateMFileHeader(&mf) < 0) {
    tsdbError("vgId:%d failed to update META file header since %s, revert it", REPO_ID(pRepo), tstrerror(terrno));
    tsdbApplyMFileChange(&mf, pOMFile);
    return -1;
  }

  TSDB_FILE_FSYNC(&mf);
  tsdbCloseMFile(&mf);
  tsdbUpdateMFile(pfs, &mf);

  return 0;
}

int tsdbOpenFS(STsdb *pRepo) {
  STsdbFS *pfs = REPO_FS(pRepo);
  char     current[TSDB_FILENAME_LEN] = "\0";
  int      nExpired = 0;

  ASSERT(pfs != NULL);

  tsdbGetTxnFname(REPO_ID(pRepo), TSDB_TXN_CURR_FILE, current);

  tsdbGetRtnSnap(pRepo, &pRepo->rtn);
  if (access(current, F_OK) == 0) {
    if (tsdbOpenFSFromCurrent(pRepo) < 0) {
      tsdbError("vgId:%d failed to open FS since %s", REPO_ID(pRepo), tstrerror(terrno));
      return -1;
    }

    tsdbScanAndTryFixDFilesHeader(pRepo, &nExpired);
    if (nExpired > 0) {
      tsdbProcessExpiredFS(pRepo);
    }
  } else {
    // should skip expired fileset inside of the function
    if (tsdbRestoreCurrent(pRepo) < 0) {
      tsdbError("vgId:%d failed to restore current file since %s", REPO_ID(pRepo), tstrerror(terrno));
      return -1;
    }
  }

  if (tsdbScanAndTryFixFS(pRepo) < 0) {
    tsdbError("vgId:%d failed to scan and fix FS since %s", REPO_ID(pRepo), tstrerror(terrno));
    return -1;
  }

  // Load meta cache if has meta file
  if ((!(pRepo->state & TSDB_STATE_BAD_META)) && tsdbLoadMetaCache(pRepo, true) < 0) {
    tsdbError("vgId:%d failed to open FS while loading meta cache since %s", REPO_ID(pRepo), tstrerror(terrno));
    return -1;
  }

  return 0;
}

void tsdbCloseFS(STsdb *pRepo) {
  // Do nothing
}

// Start a new transaction to modify the file system
void tsdbStartFSTxn(STsdb *pRepo, int64_t pointsAdd, int64_t storageAdd) {
  STsdbFS *pfs = REPO_FS(pRepo);
  ASSERT(pfs->intxn == false);

  pfs->intxn = true;
  tsdbResetFSStatus(pfs->nstatus);
  pfs->nstatus->meta = pfs->cstatus->meta;
  if (pfs->cstatus->pmf == NULL) {
    pfs->nstatus->meta.version = 0;
  } else {
    pfs->nstatus->meta.version = pfs->cstatus->meta.version + 1;
  }
  pfs->nstatus->meta.totalPoints = pfs->cstatus->meta.totalPoints + pointsAdd;
  pfs->nstatus->meta.totalStorage = pfs->cstatus->meta.totalStorage += storageAdd;
}

void tsdbUpdateFSTxnMeta(STsdbFS *pfs, STsdbFSMeta *pMeta) { pfs->nstatus->meta = *pMeta; }

int tsdbEndFSTxn(STsdb *pRepo) {
  STsdbFS *pfs = REPO_FS(pRepo);
  ASSERT(FS_IN_TXN(pfs));
  SFSStatus *pStatus;

  // Write current file system snapshot
  if (tsdbSaveFSStatus(pfs->nstatus, REPO_ID(pRepo)) < 0) {
    tsdbEndFSTxnWithError(pfs);
    return -1;
  }

  // Make new 
  tsdbWLockFS(pfs);
  pStatus = pfs->cstatus;
  pfs->cstatus = pfs->nstatus;
  pfs->nstatus = pStatus;
  tsdbUnLockFS(pfs);

  // Apply actual change to each file and SDFileSet
  tsdbApplyFSTxnOnDisk(pfs->nstatus, pfs->cstatus);

  pfs->intxn = false;
  return 0;
}

int tsdbEndFSTxnWithError(STsdbFS *pfs) {
  tsdbApplyFSTxnOnDisk(pfs->nstatus, pfs->cstatus);
  // TODO: if mf change, reload pfs->metaCache
  pfs->intxn = false;
  return 0;
}

void tsdbUpdateMFile(STsdbFS *pfs, const SMFile *pMFile) { tsdbSetStatusMFile(pfs->nstatus, pMFile); }

int tsdbUpdateDFileSet(STsdbFS *pfs, const SDFileSet *pSet) { return tsdbAddDFileSetToStatus(pfs->nstatus, pSet); }

static int tsdbSaveFSStatus(SFSStatus *pStatus, int vid) {
  SFSHeader fsheader;
  void *    pBuf = NULL;
  void *    ptr;
  char      hbuf[TSDB_FILE_HEAD_SIZE] = "\0";
  char      tfname[TSDB_FILENAME_LEN] = "\0";
  char      cfname[TSDB_FILENAME_LEN] = "\0";

  tsdbGetTxnFname(vid, TSDB_TXN_TEMP_FILE, tfname);
  tsdbGetTxnFname(vid, TSDB_TXN_CURR_FILE, cfname);

  int fd = open(tfname, O_WRONLY | O_CREAT | O_TRUNC | O_BINARY, 0755);
  if (fd < 0) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
  }

  fsheader.version = TSDB_LATEST_SFS_VER;
  if (pStatus->pmf == NULL) {
    ASSERT(taosArrayGetSize(pStatus->df) == 0);
    fsheader.len = 0;
  } else {
    fsheader.len = tsdbEncodeFSStatus(NULL, pStatus) + sizeof(TSCKSUM);
  }

  // Encode header part and write
  ptr = hbuf;
  tsdbEncodeFSHeader(&ptr, &fsheader);
  tsdbEncodeFSMeta(&ptr, &(pStatus->meta));

  taosCalcChecksumAppend(0, (uint8_t *)hbuf, TSDB_FILE_HEAD_SIZE);

  if (taosWriteFile(fd, hbuf, TSDB_FILE_HEAD_SIZE) < TSDB_FILE_HEAD_SIZE) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    close(fd);
    remove(tfname);
    return -1;
  }

  // Encode file status and write to file
  if (fsheader.len > 0) {
    if (tsdbMakeRoom(&(pBuf), fsheader.len) < 0) {
      close(fd);
      remove(tfname);
      return -1;
    }

    ptr = pBuf;
    tsdbEncodeFSStatus(&ptr, pStatus);
    taosCalcChecksumAppend(0, (uint8_t *)pBuf, fsheader.len);

    if (taosWriteFile(fd, pBuf, fsheader.len) < fsheader.len) {
      terrno = TAOS_SYSTEM_ERROR(errno);
      close(fd);
      (void)remove(tfname);
      taosTZfree(pBuf);
      return -1;
    }
  }

  // fsync, close and rename
  if (taosFsyncFile(fd) < 0) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    close(fd);
    remove(tfname);
    taosTZfree(pBuf);
    return -1;
  }

  (void)close(fd);
  (void)taosRenameFile(tfname, cfname);
  taosTZfree(pBuf);

  return 0;
}

static void tsdbApplyFSTxnOnDisk(SFSStatus *pFrom, SFSStatus *pTo) {
  int        ifrom = 0;
  int        ito = 0;
  size_t     sizeFrom, sizeTo;
  SDFileSet *pSetFrom;
  SDFileSet *pSetTo;

  sizeFrom = taosArrayGetSize(pFrom->df);
  sizeTo = taosArrayGetSize(pTo->df);

  // Apply meta file change
  (void)tsdbApplyMFileChange(pFrom->pmf, pTo->pmf);

  // Apply SDFileSet change
  if (ifrom >= sizeFrom) {
    pSetFrom = NULL;
  } else {
    pSetFrom = taosArrayGet(pFrom->df, ifrom);
  }

  if (ito >= sizeTo) {
    pSetTo = NULL;
  } else {
    pSetTo = taosArrayGet(pTo->df, ito);
  }

  while (true) {
    if ((pSetTo == NULL) && (pSetFrom == NULL)) break;

    if (pSetTo == NULL || (pSetFrom && pSetFrom->fid < pSetTo->fid)) {
      tsdbApplyDFileSetChange(pSetFrom, NULL);

      ifrom++;
      if (ifrom >= sizeFrom) {
        pSetFrom = NULL;
      } else {
        pSetFrom = taosArrayGet(pFrom->df, ifrom);
      }
    } else if (pSetFrom == NULL || pSetFrom->fid > pSetTo->fid) {
      // Do nothing
      ito++;
      if (ito >= sizeTo) {
        pSetTo = NULL;
      } else {
        pSetTo = taosArrayGet(pTo->df, ito);
      }
    } else {
      tsdbApplyDFileSetChange(pSetFrom, pSetTo);

      ifrom++;
      if (ifrom >= sizeFrom) {
        pSetFrom = NULL;
      } else {
        pSetFrom = taosArrayGet(pFrom->df, ifrom);
      }

      ito++;
      if (ito >= sizeTo) {
        pSetTo = NULL;
      } else {
        pSetTo = taosArrayGet(pTo->df, ito);
      }
    }
  }
}

// ================== SFSIter
// ASSUMPTIONS: the FS Should be read locked when calling these functions
void tsdbFSIterInit(SFSIter *pIter, STsdbFS *pfs, int direction) {
  pIter->pfs = pfs;
  pIter->direction = direction;

  size_t size = taosArrayGetSize(pfs->cstatus->df);

  pIter->version = pfs->cstatus->meta.version;

  if (size == 0) {
    pIter->index = -1;
    pIter->fid = TSDB_IVLD_FID;
  } else {
    if (direction == TSDB_FS_ITER_FORWARD) {
      pIter->index = 0;
    } else {
      pIter->index = (int)(size - 1);
    }

    pIter->fid = ((SDFileSet *)taosArrayGet(pfs->cstatus->df, pIter->index))->fid;
  }
}

void tsdbFSIterSeek(SFSIter *pIter, int fid) {
  STsdbFS *pfs = pIter->pfs;
  size_t   size = taosArrayGetSize(pfs->cstatus->df);

  int flags;
  if (pIter->direction == TSDB_FS_ITER_FORWARD) {
    flags = TD_GE;
  } else {
    flags = TD_LE;
  }

  void *ptr = taosbsearch(&fid, pfs->cstatus->df->pData, size, sizeof(SDFileSet), tsdbComparFidFSet, flags);
  if (ptr == NULL) {
    pIter->index = -1;
    pIter->fid = TSDB_IVLD_FID;
  } else {
    pIter->index = (int)(TARRAY_ELEM_IDX(pfs->cstatus->df, ptr));
    pIter->fid = ((SDFileSet *)ptr)->fid;
  }
}

SDFileSet *tsdbFSIterNext(SFSIter *pIter) {
  STsdbFS *  pfs = pIter->pfs;
  SDFileSet *pSet;

  if (pIter->index < 0) {
    ASSERT(pIter->fid == TSDB_IVLD_FID);
    return NULL;
  }

  ASSERT(pIter->fid != TSDB_IVLD_FID);

  if (pIter->version != pfs->cstatus->meta.version) {
    pIter->version = pfs->cstatus->meta.version;
    tsdbFSIterSeek(pIter, pIter->fid);
  }

  if (pIter->index < 0) {
    return NULL;
  }

  pSet = (SDFileSet *)taosArrayGet(pfs->cstatus->df, pIter->index);
  ASSERT(pSet->fid == pIter->fid);

  if (pIter->direction == TSDB_FS_ITER_FORWARD) {
    pIter->index++;
    if (pIter->index >= taosArrayGetSize(pfs->cstatus->df)) {
      pIter->index = -1;
    }
  } else {
    pIter->index--;
  }

  if (pIter->index >= 0) {
    pIter->fid = ((SDFileSet *)taosArrayGet(pfs->cstatus->df, pIter->index))->fid;
  } else {
    pIter->fid = TSDB_IVLD_FID;
  }

  return pSet;
}

static int tsdbComparFidFSet(const void *arg1, const void *arg2) {
  int        fid = *(int *)arg1;
  SDFileSet *pSet = (SDFileSet *)arg2;

  if (fid < pSet->fid) {
    return -1;
  } else if (fid == pSet->fid) {
    return 0;
  } else {
    return 1;
  }
}

static void tsdbGetTxnFname(int repoid, TSDB_TXN_FILE_T ftype, char fname[]) {
  snprintf(fname, TSDB_FILENAME_LEN, "%s/vnode/vnode%d/tsdb/%s", TFS_PRIMARY_PATH(), repoid, tsdbTxnFname[ftype]);
}

static int tsdbOpenFSFromCurrent(STsdb *pRepo) {
  STsdbFS * pfs = REPO_FS(pRepo);
  int       fd = -1;
  void *    buffer = NULL;
  SFSHeader fsheader;
  char      current[TSDB_FILENAME_LEN] = "\0";
  void *    ptr;

  tsdbGetTxnFname(REPO_ID(pRepo), TSDB_TXN_CURR_FILE, current);

  // current file exists, try to recover
  fd = open(current, O_RDONLY | O_BINARY);
  if (fd < 0) {
    tsdbError("vgId:%d failed to open file %s since %s", REPO_ID(pRepo), current, strerror(errno));
    terrno = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  if (tsdbMakeRoom(&buffer, TSDB_FILE_HEAD_SIZE) < 0) {
    goto _err;
  }

  int nread = (int)taosReadFile(fd, buffer, TSDB_FILE_HEAD_SIZE);
  if (nread < 0) {
    tsdbError("vgId:%d failed to read %d bytes from file %s since %s", REPO_ID(pRepo), TSDB_FILENAME_LEN, current,
              strerror(errno));
    terrno = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  if (nread < TSDB_FILE_HEAD_SIZE) {
    tsdbError("vgId:%d failed to read header of file %s, read bytes:%d", REPO_ID(pRepo), current, nread);
    terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
    goto _err;
  }

  if (!taosCheckChecksumWhole((uint8_t *)buffer, TSDB_FILE_HEAD_SIZE)) {
    tsdbError("vgId:%d header of file %s failed checksum check", REPO_ID(pRepo), current);
    terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
    goto _err;
  }

  SFSStatus *pStatus = pfs->cstatus;
  ptr = buffer;
  ptr = tsdbDecodeFSHeader(ptr, &fsheader);
  ptr = tsdbDecodeFSMeta(ptr, &(pStatus->meta));

  if (fsheader.version != TSDB_FS_VER_0) {
    // TODO: handle file version change
  }

  if (fsheader.len > 0) {
    if (tsdbMakeRoom(&buffer, fsheader.len) < 0) {
      goto _err;
    }

    nread = (int)taosReadFile(fd, buffer, fsheader.len);
    if (nread < 0) {
      tsdbError("vgId:%d failed to read file %s since %s", REPO_ID(pRepo), current, strerror(errno));
      terrno = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

    if (nread < fsheader.len) {
      tsdbError("vgId:%d failed to read %d bytes from file %s", REPO_ID(pRepo), fsheader.len, current);
      terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
      goto _err;
    }

    if (!taosCheckChecksumWhole((uint8_t *)buffer, fsheader.len)) {
      tsdbError("vgId:%d file %s is corrupted since wrong checksum", REPO_ID(pRepo), current);
      terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
      goto _err;
    }

    ptr = buffer;
    if (tsdbDecodeFSStatus(&buffer, ptr, pStatus, &fsheader) < 0) {
      goto _err;
    }
  } else {
    tsdbResetFSStatus(pStatus);
  }

  taosTZfree(buffer);
  close(fd);

  return 0;

_err:
  if (fd >= 0) {
    close(fd);
  }
  taosTZfree(buffer);
  return -1;
}

// Scan and try to fix incorrect files
static int tsdbScanAndTryFixFS(STsdb *pRepo) {
  STsdbFS *  pfs = REPO_FS(pRepo);
  SFSStatus *pStatus = pfs->cstatus;

  if (tsdbScanAndTryFixMFile(pRepo) < 0) {
    tsdbError("vgId:%d failed to fix MFile since %s", REPO_ID(pRepo), tstrerror(terrno));
    return -1;
  }

  size_t size = taosArrayGetSize(pStatus->df);

  for (size_t i = 0; i < size; i++) {
    SDFileSet *pSet = (SDFileSet *)taosArrayGet(pStatus->df, i);

    if (tsdbScanAndTryFixDFileSet(pRepo, pSet) < 0) {
      tsdbError("vgId:%d failed to fix DFileSet since %s", REPO_ID(pRepo), tstrerror(terrno));
      return -1;
    }
  }

  // remove those unused files
  tsdbScanRootDir(pRepo);
  tsdbScanDataDir(pRepo);
  return 0;
}

int tsdbLoadMetaCache(STsdb *pRepo, bool recoverMeta) {
  char      tbuf[128];
  STsdbFS * pfs = REPO_FS(pRepo);
  SMFile    mf;
  SMFile *  pMFile = &mf;
  void *    pBuf = NULL;
  SKVRecord rInfo;
  int64_t   maxBufSize = 0;
  SMFInfo   minfo;

  taosHashClear(pfs->metaCache);

  // No meta file, just return
  if (pfs->cstatus->pmf == NULL) return 0;

  mf = pfs->cstatus->mf;
  // Load cache first
  if (tsdbOpenMFile(pMFile, O_RDONLY) < 0) {
    return -1;
  }

  if (tsdbLoadMFileHeader(pMFile, &minfo) < 0) {
    tsdbCloseMFile(pMFile);
    return -1;
  }

  while (true) {
    int64_t tsize = tsdbReadMFile(pMFile, tbuf, sizeof(SKVRecord));
    if (tsize == 0) break;

    if (tsize < 0) {
      tsdbError("vgId:%d failed to read META file since %s", REPO_ID(pRepo), tstrerror(terrno));
      return -1;
    }

    if (tsize < sizeof(SKVRecord)) {
      tsdbError("vgId:%d failed to read %" PRIzu " bytes from file %s", REPO_ID(pRepo), sizeof(SKVRecord),
                TSDB_FILE_FULL_NAME(pMFile));
      terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
      tsdbCloseMFile(pMFile);
      return -1;
    }

    void *ptr = tsdbDecodeKVRecord(tbuf, &rInfo);
    ASSERT(POINTER_DISTANCE(ptr, tbuf) == sizeof(SKVRecord));
    // ASSERT((rInfo.offset > 0) ? (pStore->info.size == rInfo.offset) : true);

    if (rInfo.offset < 0) {
      taosHashRemove(pfs->metaCache, (void *)(&rInfo.uid), sizeof(rInfo.uid));
#if 0
      pStore->info.size += sizeof(SKVRecord);
      pStore->info.nRecords--;
      pStore->info.nDels++;
      pStore->info.tombSize += (rInfo.size + sizeof(SKVRecord) * 2);
#endif
    } else {
      ASSERT(rInfo.offset > 0 && rInfo.size > 0);
      if (taosHashPut(pfs->metaCache, (void *)(&rInfo.uid), sizeof(rInfo.uid), &rInfo, sizeof(rInfo)) < 0) {
        tsdbError("vgId:%d failed to load meta cache from file %s since OOM", REPO_ID(pRepo),
                  TSDB_FILE_FULL_NAME(pMFile));
        terrno = TSDB_CODE_OUT_OF_MEMORY;
        tsdbCloseMFile(pMFile);
        return -1;
      }

      maxBufSize = MAX(maxBufSize, rInfo.size);

      if (tsdbSeekMFile(pMFile, rInfo.size, SEEK_CUR) < 0) {
        tsdbError("vgId:%d failed to lseek file %s since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pMFile),
                  tstrerror(terrno));
        tsdbCloseMFile(pMFile);
        return -1;
      }

#if 0
      pStore->info.size += (sizeof(SKVRecord) + rInfo.size);
      pStore->info.nRecords++;
#endif
    }
  }

  if (recoverMeta) {
    pBuf = malloc((size_t)maxBufSize);
    if (pBuf == NULL) {
      terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
      tsdbCloseMFile(pMFile);
      return -1;
    }

    SKVRecord *pRecord = taosHashIterate(pfs->metaCache, NULL);
    while (pRecord) {
      if (tsdbSeekMFile(pMFile, pRecord->offset + sizeof(SKVRecord), SEEK_SET) < 0) {
        tsdbError("vgId:%d failed to seek file %s since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pMFile),
                  tstrerror(terrno));
        tfree(pBuf);
        tsdbCloseMFile(pMFile);
        return -1;
      }

      int nread = (int)tsdbReadMFile(pMFile, pBuf, pRecord->size);
      if (nread < 0) {
        tsdbError("vgId:%d failed to read file %s since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pMFile),
                  tstrerror(terrno));
        tfree(pBuf);
        tsdbCloseMFile(pMFile);
        return -1;
      }

      if (nread < pRecord->size) {
        tsdbError("vgId:%d failed to read file %s since file corrupted, expected read:%" PRId64 " actual read:%d",
                  REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pMFile), pRecord->size, nread);
        terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
        tfree(pBuf);
        tsdbCloseMFile(pMFile);
        return -1;
      }

      if (tsdbRestoreTable(pRepo, pBuf, (int)pRecord->size) < 0) {
        tsdbError("vgId:%d failed to restore table, uid %" PRId64 ", since %s" PRIu64, REPO_ID(pRepo), pRecord->uid,
                  tstrerror(terrno));
        tfree(pBuf);
        tsdbCloseMFile(pMFile);
        return -1;
      }

      pRecord = taosHashIterate(pfs->metaCache, pRecord);
    }

    tsdbOrgMeta(pRepo);
  }

  tsdbCloseMFile(pMFile);
  tfree(pBuf);
  return 0;
}

static int tsdbScanRootDir(STsdb *pRepo) {
  char         rootDir[TSDB_FILENAME_LEN];
  char         bname[TSDB_FILENAME_LEN];
  STsdbFS *    pfs = REPO_FS(pRepo);
  const TFILE *pf;

  tsdbGetRootDir(REPO_ID(pRepo), rootDir);
  TDIR *tdir = tfsOpendir(rootDir);
  if (tdir == NULL) {
    tsdbError("vgId:%d failed to open directory %s since %s", REPO_ID(pRepo), rootDir, tstrerror(terrno));
    return -1;
  }

  while ((pf = tfsReaddir(tdir))) {
    tfsbasename(pf, bname);

    if (strcmp(bname, tsdbTxnFname[TSDB_TXN_CURR_FILE]) == 0 || strcmp(bname, "data") == 0) {
      // Skip current file and data directory
      continue;
    }

    if (pfs->cstatus->pmf && tfsIsSameFile(pf, &(pfs->cstatus->pmf->f))) {
      continue;
    }

    (void)tfsremove(pf);
    tsdbDebug("vgId:%d invalid file %s is removed", REPO_ID(pRepo), TFILE_NAME(pf));
  }

  tfsClosedir(tdir);

  return 0;
}

static int tsdbScanDataDir(STsdb *pRepo) {
  char         dataDir[TSDB_FILENAME_LEN];
  char         bname[TSDB_FILENAME_LEN];
  STsdbFS *    pfs = REPO_FS(pRepo);
  const TFILE *pf;

  tsdbGetDataDir(REPO_ID(pRepo), dataDir);
  TDIR *tdir = tfsOpendir(dataDir);
  if (tdir == NULL) {
    tsdbError("vgId:%d failed to open directory %s since %s", REPO_ID(pRepo), dataDir, tstrerror(terrno));
    return -1;
  }

  while ((pf = tfsReaddir(tdir))) {
    tfsbasename(pf, bname);

    if (!tsdbIsTFileInFS(pfs, pf)) {
      (void)tfsremove(pf);
      tsdbDebug("vgId:%d invalid file %s is removed", REPO_ID(pRepo), TFILE_NAME(pf));
    }
  }

  tfsClosedir(tdir);

  return 0;
}

static bool tsdbIsTFileInFS(STsdbFS *pfs, const TFILE *pf) {
  SFSIter fsiter;
  tsdbFSIterInit(&fsiter, pfs, TSDB_FS_ITER_FORWARD);
  SDFileSet *pSet;

  while ((pSet = tsdbFSIterNext(&fsiter))) {
    for (TSDB_FILE_T ftype = 0; ftype < tsdbGetNFiles(pSet); ftype++) {
      SDFile *pDFile = TSDB_DFILE_IN_SET(pSet, ftype);
      if (tfsIsSameFile(pf, TSDB_FILE_F(pDFile))) {
        return true;
      }
    }
  }

  return false;
}

static int tsdbRestoreMeta(STsdb *pRepo) {
  char         rootDir[TSDB_FILENAME_LEN];
  char         bname[TSDB_FILENAME_LEN];
  TDIR *       tdir = NULL;
  const TFILE *pf = NULL;
  const char * pattern = "^meta(-ver[0-9]+)?$";
  regex_t      regex;
  STsdbFS *    pfs = REPO_FS(pRepo);

  regcomp(&regex, pattern, REG_EXTENDED);

  tsdbInfo("vgId:%d try to restore meta", REPO_ID(pRepo));

  tsdbGetRootDir(REPO_ID(pRepo), rootDir);

  tdir = tfsOpendir(rootDir);
  if (tdir == NULL) {
    tsdbError("vgId:%d failed to open dir %s since %s", REPO_ID(pRepo), rootDir, tstrerror(terrno));
    regfree(&regex);
    return -1;
  }

  while ((pf = tfsReaddir(tdir))) {
    tfsbasename(pf, bname);

    if (strcmp(bname, "data") == 0) {
      // Skip the data/ directory
      continue;
    }

    if (strcmp(bname, tsdbTxnFname[TSDB_TXN_TEMP_FILE]) == 0) {
      // Skip current.t file
      tsdbInfo("vgId:%d file %s exists, remove it", REPO_ID(pRepo), TFILE_NAME(pf));
      (void)tfsremove(pf);
      continue;
    }

    int code = regexec(&regex, bname, 0, NULL, 0);
    if (code == 0) {
      // Match
      if (pfs->cstatus->pmf != NULL) {
        tsdbError("vgId:%d failed to restore meta since two file exists, file1 %s and file2 %s", REPO_ID(pRepo),
                  TSDB_FILE_FULL_NAME(pfs->cstatus->pmf), TFILE_NAME(pf));
        terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
        tfsClosedir(tdir);
        regfree(&regex);
        return -1;
      } else {
        uint32_t _version = 0;
        if (strcmp(bname, "meta") != 0) {
          sscanf(bname, "meta-ver%" PRIu32, &_version);
          pfs->cstatus->meta.version = _version;
        }

        pfs->cstatus->pmf = &(pfs->cstatus->mf);
        pfs->cstatus->pmf->f = *pf;
        TSDB_FILE_SET_CLOSED(pfs->cstatus->pmf);

        if (tsdbOpenMFile(pfs->cstatus->pmf, O_RDONLY) < 0) {
          tsdbError("vgId:%d failed to restore meta since %s", REPO_ID(pRepo), tstrerror(terrno));
          tfsClosedir(tdir);
          regfree(&regex);
          return -1;
        }

        if (tsdbLoadMFileHeader(pfs->cstatus->pmf, &(pfs->cstatus->pmf->info)) < 0) {
          tsdbError("vgId:%d failed to restore meta since %s", REPO_ID(pRepo), tstrerror(terrno));
          tsdbCloseMFile(pfs->cstatus->pmf);
          tfsClosedir(tdir);
          regfree(&regex);
          return -1;
        }

        if (tsdbForceKeepFile) {
          struct stat tfstat;

          // Get real file size
          if (fstat(pfs->cstatus->pmf->fd, &tfstat) < 0) {
            terrno = TAOS_SYSTEM_ERROR(errno);
            tsdbCloseMFile(pfs->cstatus->pmf);
            tfsClosedir(tdir);
            regfree(&regex);
            return -1;
          }

          if (pfs->cstatus->pmf->info.size != tfstat.st_size) {
            int64_t tfsize = pfs->cstatus->pmf->info.size;
            pfs->cstatus->pmf->info.size = tfstat.st_size;
            tsdbInfo("vgId:%d file %s header size is changed from %" PRId64 " to %" PRId64, REPO_ID(pRepo),
                     TSDB_FILE_FULL_NAME(pfs->cstatus->pmf), tfsize, pfs->cstatus->pmf->info.size);
          }
        }

        tsdbCloseMFile(pfs->cstatus->pmf);
      }
    } else if (code == REG_NOMATCH) {
      // Not match
      tsdbInfo("vgId:%d invalid file %s exists, remove it", REPO_ID(pRepo), TFILE_NAME(pf));
      tfsremove(pf);
      continue;
    } else {
      // Has other error
      tsdbError("vgId:%d failed to restore meta file while run regexec since %s", REPO_ID(pRepo), strerror(code));
      terrno = TAOS_SYSTEM_ERROR(code);
      tfsClosedir(tdir);
      regfree(&regex);
      return -1;
    }
  }

  if (pfs->cstatus->pmf) {
    tsdbInfo("vgId:%d meta file %s is restored", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pfs->cstatus->pmf));
  } else {
    tsdbInfo("vgId:%d no meta file is restored", REPO_ID(pRepo));
  }

  tfsClosedir(tdir);
  regfree(&regex);
  return 0;
}

static int tsdbFetchTFileSet(STsdb *pRepo, SArray **fArray) {
  char         dataDir[TSDB_FILENAME_LEN];
  char         bname[TSDB_FILENAME_LEN];
  TDIR *       tdir = NULL;
  const TFILE *pf = NULL;
  const char * pattern = "^v[0-9]+f[0-9]+\\.(head|data|last|smad|smal)(-ver[0-9]+)?$";
  regex_t      regex;

  tsdbGetDataDir(REPO_ID(pRepo), dataDir);

  // Resource allocation and init
  regcomp(&regex, pattern, REG_EXTENDED);

  *fArray = taosArrayInit(1024, sizeof(TFILE));
  if (*fArray == NULL) {
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    tsdbError("vgId:%d failed to fetch TFileSet while open directory %s since %s", REPO_ID(pRepo), dataDir,
              tstrerror(terrno));
    regfree(&regex);
    return -1;
  }

  tdir = tfsOpendir(dataDir);
  if (tdir == NULL) {
    tsdbError("vgId:%d failed to fetch TFileSet while open directory %s since %s", REPO_ID(pRepo), dataDir,
              tstrerror(terrno));
    taosArrayDestroy(*fArray);
    regfree(&regex);
    return -1;
  }

  while ((pf = tfsReaddir(tdir))) {
    tfsbasename(pf, bname);

    int code = regexec(&regex, bname, 0, NULL, 0);
    if (code == 0) {
      if (taosArrayPush(*fArray, (void *)pf) == NULL) {
        terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
        tfsClosedir(tdir);
        taosArrayDestroy(*fArray);
        regfree(&regex);
        return -1;
      }
    } else if (code == REG_NOMATCH) {
      // Not match
      tsdbInfo("vgId:%d invalid file %s exists, remove it", REPO_ID(pRepo), TFILE_NAME(pf));
      (void)tfsremove(pf);
      continue;
    } else {
      // Has other error
      tsdbError("vgId:%d failed to fetch TFileSet Array while run regexec since %s", REPO_ID(pRepo), strerror(code));
      terrno = TAOS_SYSTEM_ERROR(code);
      tfsClosedir(tdir);
      taosArrayDestroy(*fArray);
      regfree(&regex);
      return -1;
    }
  }

  tfsClosedir(tdir);
  regfree(&regex);

  // Sort the array according to file name
  taosArraySort(*fArray, tsdbComparTFILE);
  return 0;
}

// update the function if the DFileSet definition updates
static bool tsdbIsDFileSetValid(int nFiles) {
  switch (nFiles) {
    case TSDB_FILE_MIN:
    case TSDB_FILE_MAX:
      return true;
    default:
      return false;
  }
}

static int tsdbRestoreDFileSet(STsdb *pRepo) {
  const TFILE *pf = NULL;
  SArray *     fArray = NULL;
  STsdbFS *    pfs = REPO_FS(pRepo);
  char         dataDir[TSDB_FILENAME_LEN] = "\0";
  size_t       fArraySize = 0;

  tsdbGetDataDir(REPO_ID(pRepo), dataDir);

  if (tsdbFetchTFileSet(pRepo, &fArray) < 0) {
    tsdbError("vgId:%d failed to fetch TFileSet from %s to restore since %s", REPO_ID(pRepo), dataDir,
              tstrerror(terrno));
    return -1;
  }

  if ((fArraySize = taosArrayGetSize(fArray)) <= 0) {
    taosArrayDestroy(fArray);
    tsdbInfo("vgId:%d size of DFileSet from %s is %" PRIu32, REPO_ID(pRepo), dataDir, (uint32_t)fArraySize);
    return 0;
  }

  // Loop to recover each file set
  SDFileSet fset = {0};
  uint8_t   nDFiles = 0;
  bool      isOneFSetFinish = true;
  int       lastFType = -1;
  // one fileset ends when (1) the array ends or (2) encounter different fid
  for (size_t index = 0; index < fArraySize; ++index) {
    int         tvid = -1, tfid = -1;
    TSDB_FILE_T ttype = TSDB_FILE_MAX;
    uint32_t    tversion = -1;
    char        bname[TSDB_FILENAME_LEN] = "\0";

    pf = taosArrayGet(fArray, index);
    tfsbasename(pf, bname);
    tsdbParseDFilename(bname, &tvid, &tfid, &ttype, &tversion); 
    ASSERT(tvid == REPO_ID(pRepo));
    SDFile *pDFile = TSDB_DFILE_IN_SET(&fset, ttype);
    if (tfid < pRepo->rtn.minFid) {  // skip the file expired
      continue;
    }
    if ((isOneFSetFinish == false) && (lastFType == ttype)) {  // only fetch the 1st file with same fid and type.
      continue;
    }

    lastFType = ttype;

    if (index == 0) {
      memset(&fset, 0, sizeof(SDFileSet));
      TSDB_FSET_SET_CLOSED(&fset);
      nDFiles = 1;
      fset.fid = tfid;
      pDFile->f = *pf;
      isOneFSetFinish = false;
    } else {
      if (fset.fid == tfid) {
        ++nDFiles;
        pDFile->f = *pf;
        // (1) the array ends
        if (index == fArraySize - 1) {
          if (tsdbIsDFileSetValid(nDFiles)) {
            tsdbInfo("vgId:%d DFileSet %d is fetched, nDFiles=%" PRIu8, REPO_ID(pRepo), fset.fid, nDFiles);
            isOneFSetFinish = true;
          } else {
            // return error in case of removing uncomplete DFileSets
            // terrno = TSDB_CODE_TDB_INCOMPLETE_DFILESET;
            tsdbError("vgId:%d incomplete DFileSet, fid:%d, nDFiles=%" PRIu8, REPO_ID(pRepo), fset.fid, nDFiles);
            taosArrayDestroy(fArray);
            return -1;
          }
        }
      } else {
        // (2) encounter different fid
        if (tsdbIsDFileSetValid(nDFiles)) {
          tsdbInfo("vgId:%d DFileSet %d is fetched, nDFiles=%" PRIu8, REPO_ID(pRepo), fset.fid, nDFiles);
          isOneFSetFinish = true;
        } else {
          // return error in case of removing uncomplete DFileSets
          // terrno = TSDB_CODE_TDB_INCOMPLETE_DFILESET;
          tsdbError("vgId:%d incomplete DFileSet, fid:%d, nDFiles=%" PRIu8, REPO_ID(pRepo), fset.fid, nDFiles);
          taosArrayDestroy(fArray);
          return -1;
#if 0
          // next FSet
          memset(&fset, 0, sizeof(SDFileSet));
          TSDB_FSET_SET_CLOSED(&fset);
          nDFiles = 1;
          fset.fid = tfid;
          pDFile->f = *pf;
          isOneFSetFinish = false;
          continue;
#endif
        }
      }
    }

    if (isOneFSetFinish) {
      for (TSDB_FILE_T ftype = 0; ftype < nDFiles; ++ftype) {
        SDFile * pDFile1 = TSDB_DFILE_IN_SET(&fset, ftype);
        if (tsdbOpenDFile(pDFile1, O_RDONLY) < 0) {
          tsdbError("vgId:%d failed to open DFile %s since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pDFile1),
                    tstrerror(terrno));
          taosArrayDestroy(fArray);
          return -1;
        }

        if (tsdbLoadDFileHeader(pDFile1, &(pDFile1->info)) < 0) {
          tsdbError("vgId:%d failed to load DFile %s header since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pDFile1),
                    tstrerror(terrno));
          taosArrayDestroy(fArray);
          return -1;
        }

        if (tsdbForceKeepFile) {
          struct stat tfstat;

          // Get real file size
          if (fstat(pDFile1->fd, &tfstat) < 0) {
            terrno = TAOS_SYSTEM_ERROR(errno);
            taosArrayDestroy(fArray);
            return -1;
          }

          if (pDFile1->info.size != tfstat.st_size) {
            int64_t tfsize = pDFile1->info.size;
            pDFile1->info.size = tfstat.st_size;
            tsdbInfo("vgId:%d file %s header size is changed from %" PRId64 " to %" PRId64, REPO_ID(pRepo),
                     TSDB_FILE_FULL_NAME(pDFile1), tfsize, pDFile1->info.size);
          }
        }

        tsdbCloseDFile(pDFile1);
      }
      tsdbInfo("vgId:%d FSET %d is restored", REPO_ID(pRepo), fset.fid);

      // TODO: update the logic when TSDB_FSET_VER definition update.
      if (nDFiles == TSDB_FILE_MIN) {
        fset.ver = TSDB_FSET_VER_0;
      } else {
        fset.ver = TSDB_LATEST_FSET_VER;
      }

      taosArrayPush(pfs->cstatus->df, &fset);

      // next FSet
      memset(&fset, 0, sizeof(SDFileSet));
      TSDB_FSET_SET_CLOSED(&fset);
      nDFiles = 1;
      fset.fid = tfid;
      pDFile->f = *pf;
      isOneFSetFinish = false;
    }
  }

  // Resource release
  taosArrayDestroy(fArray);

  return 0;
}

static int tsdbRestoreCurrent(STsdb *pRepo) {
  // Loop to recover mfile
  if (tsdbRestoreMeta(pRepo) < 0) {
    tsdbError("vgId:%d failed to restore current since %s", REPO_ID(pRepo), tstrerror(terrno));
    return -1;
  }

  // Loop to recover dfile set
  if (tsdbRestoreDFileSet(pRepo) < 0) {
    tsdbError("vgId:%d failed to restore DFileSet since %s", REPO_ID(pRepo), tstrerror(terrno));
    return -1;
  }

  if (tsdbSaveFSStatus(pRepo->fs->cstatus, REPO_ID(pRepo)) < 0) {
    tsdbError("vgId:%d failed to restore corrent since %s", REPO_ID(pRepo), tstrerror(terrno));
    return -1;
  }

  return 0;
}

static int tsdbComparTFILE(const void *arg1, const void *arg2) {
  TFILE *pf1 = (TFILE *)arg1;
  TFILE *pf2 = (TFILE *)arg2;

  int         vid1, fid1, vid2, fid2;
  TSDB_FILE_T ftype1, ftype2;
  uint32_t    version1, version2;
  char        bname1[TSDB_FILENAME_LEN];
  char        bname2[TSDB_FILENAME_LEN];

  tfsbasename(pf1, bname1);
  tfsbasename(pf2, bname2);
  tsdbParseDFilename(bname1, &vid1, &fid1, &ftype1, &version1);
  tsdbParseDFilename(bname2, &vid2, &fid2, &ftype2, &version2);

  if (fid1 < fid2) {
    return -1;
  } else if (fid1 > fid2) {
    return 1;
  } else {
    if (ftype1 < ftype2) {
      return -1;
    } else if (ftype1 > ftype2) {
      return 1;
    } else {
      if (version1 < version2) {
        return -1;
      } else if (version1 > version2) {
        return 1;
      } else {
        return 0;
      }
    }
  }
}

static void tsdbScanAndTryFixDFilesHeader(STsdb *pRepo, int32_t *nExpired) {
  STsdbFS *  pfs = REPO_FS(pRepo);
  SFSStatus *pStatus = pfs->cstatus;
  SDFInfo    info;

  for (size_t i = 0; i < taosArrayGetSize(pStatus->df); i++) {
    SDFileSet fset;
    tsdbInitDFileSetEx(&fset, (SDFileSet *)taosArrayGet(pStatus->df, i));
    if (fset.fid < pRepo->rtn.minFid) {
      ++*nExpired;
    }
    tsdbDebug("vgId:%d scan DFileSet %d header", REPO_ID(pRepo), fset.fid);

    if (tsdbOpenDFileSet(&fset, O_RDWR) < 0) {
      tsdbError("vgId:%d failed to open DFileSet %d since %s, continue", REPO_ID(pRepo), fset.fid, tstrerror(terrno));
      continue;
    }

    for (TSDB_FILE_T ftype = 0; ftype < tsdbGetNFiles(&fset); ftype++) {
      SDFile *pDFile = TSDB_DFILE_IN_SET(&fset, ftype);

      if ((tsdbLoadDFileHeader(pDFile, &info) < 0) || pDFile->info.size != info.size ||
          pDFile->info.magic != info.magic) {
        if (tsdbUpdateDFileHeader(pDFile) < 0) {
          tsdbError("vgId:%d failed to update DFile header of %s since %s, continue", REPO_ID(pRepo),
                    TSDB_FILE_FULL_NAME(pDFile), tstrerror(terrno));
        } else {
          tsdbInfo("vgId:%d DFile header of %s is updated", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pDFile));
          TSDB_FILE_FSYNC(pDFile);
        }
      } else {
        tsdbDebug("vgId:%d DFile header of %s is correct", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pDFile));
      }
    }

    tsdbCloseDFileSet(&fset);
  }
}