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

#include "tsdb.h"

// =================================================================================================
static int32_t tsdbFSToBinary(uint8_t *p, STsdbFS *pFS) {
  int32_t  n = 0;
  int8_t   hasDel = pFS->pDelFile ? 1 : 0;
  uint32_t nSet = taosArrayGetSize(pFS->aDFileSet);

  // version
  n += tPutI8(p ? p + n : p, 0);

  // SDelFile
  n += tPutI8(p ? p + n : p, hasDel);
  if (hasDel) {
    n += tPutDelFile(p ? p + n : p, pFS->pDelFile);
  }

  // SArray<SDFileSet>
  n += tPutU32v(p ? p + n : p, nSet);
  for (uint32_t iSet = 0; iSet < nSet; iSet++) {
    n += tPutDFileSet(p ? p + n : p, (SDFileSet *)taosArrayGet(pFS->aDFileSet, iSet));
  }

  return n;
}

static int32_t tsdbBinaryToFS(uint8_t *pData, int64_t nData, STsdbFS *pFS) {
  int32_t code = 0;
  int32_t n = 0;

  // version
  n += tGetI8(pData + n, NULL);

  // SDelFile
  int8_t hasDel = 0;
  n += tGetI8(pData + n, &hasDel);
  if (hasDel) {
    pFS->pDelFile = (SDelFile *)taosMemoryCalloc(1, sizeof(SDelFile));
    if (pFS->pDelFile == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _exit;
    }

    n += tGetDelFile(pData + n, pFS->pDelFile);
    pFS->pDelFile->nRef = 1;
  } else {
    pFS->pDelFile = NULL;
  }

  // aDFileSet
  taosArrayClear(pFS->aDFileSet);
  uint32_t nSet = 0;
  n += tGetU32v(pData + n, &nSet);
  for (uint32_t iSet = 0; iSet < nSet; iSet++) {
    SDFileSet fSet = {0};

    int32_t nt = tGetDFileSet(pData + n, &fSet);
    if (nt < 0) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _exit;
    }

    n += nt;
    if (taosArrayPush(pFS->aDFileSet, &fSet) == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _exit;
    }
  }

  ASSERT(n + sizeof(TSCKSUM) == nData);

_exit:
  return code;
}

static int32_t tsdbSaveFSToFile(STsdbFS *pFS, const char *fname) {
  int32_t code = 0;
  int32_t lino = 0;

  // encode to binary
  int32_t  size = tsdbFSToBinary(NULL, pFS) + sizeof(TSCKSUM);
  uint8_t *pData = taosMemoryMalloc(size);
  if (pData == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    TSDB_CHECK_CODE(code, lino, _exit);
  }
  tsdbFSToBinary(pData, pFS);
  taosCalcChecksumAppend(0, pData, size);

  // save to file
  TdFilePtr pFD = taosOpenFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
  if (pFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  int64_t n = taosWriteFile(pFD, pData, size);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    taosCloseFile(&pFD);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  if (taosFsyncFile(pFD) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    taosCloseFile(&pFD);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  taosCloseFile(&pFD);

_exit:
  if (pData) taosMemoryFree(pData);
  if (code) {
    tsdbError("%s failed at line %d since %s, fname:%s", __func__, lino, tstrerror(code), fname);
  }
  return code;
}

int32_t tsdbFSCreate(STsdbFS *pFS) {
  int32_t code = 0;

  pFS->pDelFile = NULL;
  pFS->aDFileSet = taosArrayInit(0, sizeof(SDFileSet));
  if (pFS->aDFileSet == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

_exit:
  return code;
}

void tsdbFSDestroy(STsdbFS *pFS) {
  if (pFS->pDelFile) {
    taosMemoryFree(pFS->pDelFile);
  }

  for (int32_t iSet = 0; iSet < taosArrayGetSize(pFS->aDFileSet); iSet++) {
    SDFileSet *pSet = (SDFileSet *)taosArrayGet(pFS->aDFileSet, iSet);
    taosMemoryFree(pSet->pHeadF);
    taosMemoryFree(pSet->pDataF);
    taosMemoryFree(pSet->pSmaF);
    for (int32_t iStt = 0; iStt < pSet->nSttF; iStt++) {
      taosMemoryFree(pSet->aSttF[iStt]);
    }
  }

  taosArrayDestroy(pFS->aDFileSet);
}

static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
  int32_t code = 0;
  int32_t lino = 0;
  int64_t size = 0;
  char    fname[TSDB_FILENAME_LEN] = {0};

  // SDelFile
  if (pTsdb->fs.pDelFile) {
    tsdbDelFileName(pTsdb, pTsdb->fs.pDelFile, fname);
    if (taosStatFile(fname, &size, NULL)) {
      code = TAOS_SYSTEM_ERROR(errno);
      TSDB_CHECK_CODE(code, lino, _exit);
    }

    if (size != tsdbLogicToFileSize(pTsdb->fs.pDelFile->size, pTsdb->pVnode->config.tsdbPageSize)) {
      code = TSDB_CODE_FILE_CORRUPTED;
      TSDB_CHECK_CODE(code, lino, _exit);
    }
  }

  // SArray<SDFileSet>
  int32_t fid = 0;
  for (int32_t iSet = 0; iSet < taosArrayGetSize(pTsdb->fs.aDFileSet); iSet++) {
    SDFileSet *pSet = (SDFileSet *)taosArrayGet(pTsdb->fs.aDFileSet, iSet);
    fid = pSet->fid;

    // head =========
    tsdbHeadFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pHeadF, fname);
    if (taosStatFile(fname, &size, NULL)) {
      code = TAOS_SYSTEM_ERROR(errno);
      TSDB_CHECK_CODE(code, lino, _exit);
    }
    if (size != tsdbLogicToFileSize(pSet->pHeadF->size, pTsdb->pVnode->config.tsdbPageSize)) {
      code = TSDB_CODE_FILE_CORRUPTED;
      TSDB_CHECK_CODE(code, lino, _exit);
    }

    // data =========
    tsdbDataFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pDataF, fname);
    if (taosStatFile(fname, &size, NULL)) {
      code = TAOS_SYSTEM_ERROR(errno);
      TSDB_CHECK_CODE(code, lino, _exit);
    }
    if (size < tsdbLogicToFileSize(pSet->pDataF->size, pTsdb->pVnode->config.tsdbPageSize)) {
      code = TSDB_CODE_FILE_CORRUPTED;
      TSDB_CHECK_CODE(code, lino, _exit);
    }
    // else if (size > tsdbLogicToFileSize(pSet->pDataF->size, pTsdb->pVnode->config.tsdbPageSize)) {
    //   code = tsdbDFileRollback(pTsdb, pSet, TSDB_DATA_FILE);
    //   TSDB_CHECK_CODE(code, lino, _exit);
    // }

    // sma =============
    tsdbSmaFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pSmaF, fname);
    if (taosStatFile(fname, &size, NULL)) {
      code = TAOS_SYSTEM_ERROR(errno);
      TSDB_CHECK_CODE(code, lino, _exit);
    }
    if (size < tsdbLogicToFileSize(pSet->pSmaF->size, pTsdb->pVnode->config.tsdbPageSize)) {
      code = TSDB_CODE_FILE_CORRUPTED;
      TSDB_CHECK_CODE(code, lino, _exit);
    }
    // else if (size > tsdbLogicToFileSize(pSet->pSmaF->size, pTsdb->pVnode->config.tsdbPageSize)) {
    //   code = tsdbDFileRollback(pTsdb, pSet, TSDB_SMA_FILE);
    //   TSDB_CHECK_CODE(code, lino, _exit);
    // }

    // stt ===========
    for (int32_t iStt = 0; iStt < pSet->nSttF; iStt++) {
      tsdbSttFileName(pTsdb, pSet->diskId, pSet->fid, pSet->aSttF[iStt], fname);
      if (taosStatFile(fname, &size, NULL)) {
        code = TAOS_SYSTEM_ERROR(errno);
        TSDB_CHECK_CODE(code, lino, _exit);
      }
      if (size != tsdbLogicToFileSize(pSet->aSttF[iStt]->size, pTsdb->pVnode->config.tsdbPageSize)) {
        code = TSDB_CODE_FILE_CORRUPTED;
        TSDB_CHECK_CODE(code, lino, _exit);
      }
    }
  }

  {
    // remove those invalid files (todo)
  }

_exit:
  if (code) {
    tsdbError("vgId:%d %s failed at line %d since %s, fid:%d", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code),
              fid);
  }
  return code;
}

int32_t tDFileSetCmprFn(const void *p1, const void *p2) {
  if (((SDFileSet *)p1)->fid < ((SDFileSet *)p2)->fid) {
    return -1;
  } else if (((SDFileSet *)p1)->fid > ((SDFileSet *)p2)->fid) {
    return 1;
  }

  return 0;
}

static void tsdbGetCurrentFName(STsdb *pTsdb, char *current, char *current_t) {
  SVnode *pVnode = pTsdb->pVnode;
  if (pVnode->pTfs) {
    if (current) {
      snprintf(current, TSDB_FILENAME_LEN - 1, "%s%s%s%sCURRENT", tfsGetPrimaryPath(pTsdb->pVnode->pTfs), TD_DIRSEP,
               pTsdb->path, TD_DIRSEP);
    }
    if (current_t) {
      snprintf(current_t, TSDB_FILENAME_LEN - 1, "%s%s%s%sCURRENT.t", tfsGetPrimaryPath(pTsdb->pVnode->pTfs), TD_DIRSEP,
               pTsdb->path, TD_DIRSEP);
    }
  } else {
    if (current) {
      snprintf(current, TSDB_FILENAME_LEN - 1, "%s%sCURRENT", pTsdb->path, TD_DIRSEP);
    }
    if (current_t) {
      snprintf(current_t, TSDB_FILENAME_LEN - 1, "%s%sCURRENT.t", pTsdb->path, TD_DIRSEP);
    }
  }
}

static int32_t tsdbLoadFSFromFile(const char *fname, STsdbFS *pFS) {
  int32_t  code = 0;
  int32_t  lino = 0;
  uint8_t *pData = NULL;

  // load binary
  TdFilePtr pFD = taosOpenFile(fname, TD_FILE_READ);
  if (pFD == NULL) {
    code = terrno;
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  int64_t size;
  if (taosFStatFile(pFD, &size, NULL) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    taosCloseFile(&pFD);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  pData = taosMemoryMalloc(size);
  if (pData == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    taosCloseFile(&pFD);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  if (taosReadFile(pFD, pData, size) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    taosCloseFile(&pFD);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  if (!taosCheckChecksumWhole(pData, size)) {
    code = TSDB_CODE_FILE_CORRUPTED;
    taosCloseFile(&pFD);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  taosCloseFile(&pFD);

  // decode binary
  code = tsdbBinaryToFS(pData, size, pFS);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  if (pData) taosMemoryFree(pData);
  if (code) {
    tsdbError("%s failed at line %d since %s, fname:%s", __func__, lino, tstrerror(code), fname);
  }
  return code;
}

static int32_t tsdbRemoveFileSet(STsdb *pTsdb, SDFileSet *pSet) {
  int32_t code = 0;
  char    fname[TSDB_FILENAME_LEN] = {0};

  int32_t nRef = atomic_sub_fetch_32(&pSet->pHeadF->nRef, 1);
  if (nRef == 0) {
    tsdbHeadFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pHeadF, fname);
    (void)taosRemoveFile(fname);
    taosMemoryFree(pSet->pHeadF);
  }

  nRef = atomic_sub_fetch_32(&pSet->pDataF->nRef, 1);
  if (nRef == 0) {
    tsdbDataFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pDataF, fname);
    taosRemoveFile(fname);
    taosMemoryFree(pSet->pDataF);
  }

  nRef = atomic_sub_fetch_32(&pSet->pSmaF->nRef, 1);
  if (nRef == 0) {
    tsdbSmaFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pSmaF, fname);
    taosRemoveFile(fname);
    taosMemoryFree(pSet->pSmaF);
  }

  for (int8_t iStt = 0; iStt < pSet->nSttF; iStt++) {
    nRef = atomic_sub_fetch_32(&pSet->aSttF[iStt]->nRef, 1);
    if (nRef == 0) {
      tsdbSttFileName(pTsdb, pSet->diskId, pSet->fid, pSet->aSttF[iStt], fname);
      taosRemoveFile(fname);
      taosMemoryFree(pSet->aSttF[iStt]);
    }
  }

_exit:
  return code;
}

static int32_t tsdbNewFileSet(STsdb *pTsdb, SDFileSet *pSetTo, SDFileSet *pSetFrom) {
  int32_t code = 0;
  int32_t lino = 0;

  *pSetTo = (SDFileSet){.diskId = pSetFrom->diskId, .fid = pSetFrom->fid, .nSttF = 0};

  // head
  pSetTo->pHeadF = (SHeadFile *)taosMemoryMalloc(sizeof(SHeadFile));
  if (pSetTo->pHeadF == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    TSDB_CHECK_CODE(code, lino, _exit);
  }
  *pSetTo->pHeadF = *pSetFrom->pHeadF;
  pSetTo->pHeadF->nRef = 1;

  // data
  pSetTo->pDataF = (SDataFile *)taosMemoryMalloc(sizeof(SDataFile));
  if (pSetTo->pDataF == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    TSDB_CHECK_CODE(code, lino, _exit);
  }
  *pSetTo->pDataF = *pSetFrom->pDataF;
  pSetTo->pDataF->nRef = 1;

  // sma
  pSetTo->pSmaF = (SSmaFile *)taosMemoryMalloc(sizeof(SSmaFile));
  if (pSetTo->pSmaF == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    TSDB_CHECK_CODE(code, lino, _exit);
  }
  *pSetTo->pSmaF = *pSetFrom->pSmaF;
  pSetTo->pSmaF->nRef = 1;

  // stt
  for (int32_t iStt = 0; iStt < pSetFrom->nSttF; iStt++) {
    pSetTo->aSttF[iStt] = (SSttFile *)taosMemoryMalloc(sizeof(SSttFile));
    if (pSetTo->aSttF[iStt] == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      TSDB_CHECK_CODE(code, lino, _exit);
    }

    pSetTo->nSttF++;
    *pSetTo->aSttF[iStt] = *pSetFrom->aSttF[iStt];
    pSetTo->aSttF[iStt]->nRef = 1;
  }

_exit:
  if (code) {
    tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
  }
  return code;
}

static int32_t tsdbMergeFileSet(STsdb *pTsdb, SDFileSet *pSetOld, SDFileSet *pSetNew) {
  int32_t code = 0;
  int32_t lino = 0;
  int32_t nRef = 0;
  bool    sameDisk = ((pSetOld->diskId.level == pSetNew->diskId.level) && (pSetOld->diskId.id == pSetNew->diskId.id));
  char    fname[TSDB_FILENAME_LEN] = {0};

  // head
  SHeadFile *pHeadF = pSetOld->pHeadF;
  if ((!sameDisk) || (pHeadF->commitID != pSetNew->pHeadF->commitID)) {
    pSetOld->pHeadF = (SHeadFile *)taosMemoryMalloc(sizeof(SHeadFile));
    if (pSetOld->pHeadF == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      TSDB_CHECK_CODE(code, lino, _exit);
    }
    *pSetOld->pHeadF = *pSetNew->pHeadF;
    pSetOld->pHeadF->nRef = 1;

    nRef = atomic_sub_fetch_32(&pHeadF->nRef, 1);
    if (nRef == 0) {
      tsdbHeadFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pHeadF, fname);
      (void)taosRemoveFile(fname);
      taosMemoryFree(pHeadF);
    }
  } else {
    nRef = pHeadF->nRef;
    *pHeadF = *pSetNew->pHeadF;
    pHeadF->nRef = nRef;
  }

  // data
  SDataFile *pDataF = pSetOld->pDataF;
  if ((!sameDisk) || (pDataF->commitID != pSetNew->pDataF->commitID)) {
    pSetOld->pDataF = (SDataFile *)taosMemoryMalloc(sizeof(SDataFile));
    if (pSetOld->pDataF == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      TSDB_CHECK_CODE(code, lino, _exit);
    }
    *pSetOld->pDataF = *pSetNew->pDataF;
    pSetOld->pDataF->nRef = 1;

    nRef = atomic_sub_fetch_32(&pDataF->nRef, 1);
    if (nRef == 0) {
      tsdbDataFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pDataF, fname);
      (void)taosRemoveFile(fname);
      taosMemoryFree(pDataF);
    }
  } else {
    nRef = pDataF->nRef;
    *pDataF = *pSetNew->pDataF;
    pDataF->nRef = nRef;
  }

  // sma
  SSmaFile *pSmaF = pSetOld->pSmaF;
  if ((!sameDisk) || (pSmaF->commitID != pSetNew->pSmaF->commitID)) {
    pSetOld->pSmaF = (SSmaFile *)taosMemoryMalloc(sizeof(SSmaFile));
    if (pSetOld->pSmaF == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      TSDB_CHECK_CODE(code, lino, _exit);
    }
    *pSetOld->pSmaF = *pSetNew->pSmaF;
    pSetOld->pSmaF->nRef = 1;

    nRef = atomic_sub_fetch_32(&pSmaF->nRef, 1);
    if (nRef == 0) {
      tsdbSmaFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pSmaF, fname);
      (void)taosRemoveFile(fname);
      taosMemoryFree(pSmaF);
    }
  } else {
    nRef = pSmaF->nRef;
    *pSmaF = *pSetNew->pSmaF;
    pSmaF->nRef = nRef;
  }

  // stt
  if (sameDisk) {
    if (pSetNew->nSttF > pSetOld->nSttF) {
      ASSERT(pSetNew->nSttF == pSetOld->nSttF + 1);
      pSetOld->aSttF[pSetOld->nSttF] = (SSttFile *)taosMemoryMalloc(sizeof(SSttFile));
      if (pSetOld->aSttF[pSetOld->nSttF] == NULL) {
        code = TSDB_CODE_OUT_OF_MEMORY;
        TSDB_CHECK_CODE(code, lino, _exit);
      }
      *pSetOld->aSttF[pSetOld->nSttF] = *pSetNew->aSttF[pSetOld->nSttF];
      pSetOld->aSttF[pSetOld->nSttF]->nRef = 1;
      pSetOld->nSttF++;
    } else if (pSetNew->nSttF < pSetOld->nSttF) {
      ASSERT(pSetNew->nSttF == 1);
      for (int32_t iStt = 0; iStt < pSetOld->nSttF; iStt++) {
        SSttFile *pSttFile = pSetOld->aSttF[iStt];
        nRef = atomic_sub_fetch_32(&pSttFile->nRef, 1);
        if (nRef == 0) {
          tsdbSttFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pSttFile, fname);
          taosRemoveFile(fname);
          taosMemoryFree(pSttFile);
        }
        pSetOld->aSttF[iStt] = NULL;
      }

      pSetOld->nSttF = 1;
      pSetOld->aSttF[0] = (SSttFile *)taosMemoryMalloc(sizeof(SSttFile));
      if (pSetOld->aSttF[0] == NULL) {
        code = TSDB_CODE_OUT_OF_MEMORY;
        TSDB_CHECK_CODE(code, lino, _exit);
      }
      *pSetOld->aSttF[0] = *pSetNew->aSttF[0];
      pSetOld->aSttF[0]->nRef = 1;
    } else {
      for (int32_t iStt = 0; iStt < pSetOld->nSttF; iStt++) {
        if (pSetOld->aSttF[iStt]->commitID != pSetNew->aSttF[iStt]->commitID) {
          SSttFile *pSttFile = pSetOld->aSttF[iStt];
          nRef = atomic_sub_fetch_32(&pSttFile->nRef, 1);
          if (nRef == 0) {
            tsdbSttFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pSttFile, fname);
            taosRemoveFile(fname);
            taosMemoryFree(pSttFile);
          }

          pSetOld->aSttF[iStt] = (SSttFile *)taosMemoryMalloc(sizeof(SSttFile));
          if (pSetOld->aSttF[iStt] == NULL) {
            code = TSDB_CODE_OUT_OF_MEMORY;
            TSDB_CHECK_CODE(code, lino, _exit);
          }
          *pSetOld->aSttF[iStt] = *pSetNew->aSttF[iStt];
          pSetOld->aSttF[iStt]->nRef = 1;
        } else {
          ASSERT(pSetOld->aSttF[iStt]->size == pSetOld->aSttF[iStt]->size);
          ASSERT(pSetOld->aSttF[iStt]->offset == pSetOld->aSttF[iStt]->offset);
        }
      }
    }
  } else {
    for (int32_t iStt = 0; iStt < pSetOld->nSttF; iStt++) {
      SSttFile *pSttFile = pSetOld->aSttF[iStt];
      nRef = atomic_sub_fetch_32(&pSttFile->nRef, 1);
      if (nRef == 0) {
        tsdbSttFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pSttFile, fname);
        (void)taosRemoveFile(fname);
        taosMemoryFree(pSttFile);
      }
    }

    pSetOld->nSttF = 0;
    for (int32_t iStt = 0; iStt < pSetNew->nSttF; iStt++) {
      pSetOld->aSttF[iStt] = (SSttFile *)taosMemoryMalloc(sizeof(SSttFile));
      if (pSetOld->aSttF[iStt] == NULL) {
        code = TSDB_CODE_OUT_OF_MEMORY;
        TSDB_CHECK_CODE(code, lino, _exit);
      }

      *pSetOld->aSttF[iStt] = *pSetNew->aSttF[iStt];
      pSetOld->aSttF[iStt]->nRef = 1;

      pSetOld->nSttF++;
    }
  }

  if (!sameDisk) {
    pSetOld->diskId = pSetNew->diskId;
  }

_exit:
  if (code) {
    tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
  }
  return code;
}

static int32_t tsdbFSApplyChange(STsdb *pTsdb, STsdbFS *pFS) {
  int32_t code = 0;
  int32_t lino = 0;

  int32_t nRef = 0;
  char    fname[TSDB_FILENAME_LEN] = {0};

  // SDelFile
  if (pFS->pDelFile) {
    SDelFile *pDelFile = pTsdb->fs.pDelFile;

    if (pDelFile == NULL || (pDelFile->commitID != pFS->pDelFile->commitID)) {
      pTsdb->fs.pDelFile = (SDelFile *)taosMemoryMalloc(sizeof(SDelFile));
      if (pTsdb->fs.pDelFile == NULL) {
        code = TSDB_CODE_OUT_OF_MEMORY;
        TSDB_CHECK_CODE(code, lino, _exit);
      }

      *pTsdb->fs.pDelFile = *pFS->pDelFile;
      pTsdb->fs.pDelFile->nRef = 1;

      if (pDelFile) {
        nRef = atomic_sub_fetch_32(&pDelFile->nRef, 1);
        if (nRef == 0) {
          tsdbDelFileName(pTsdb, pDelFile, fname);
          (void)taosRemoveFile(fname);
          taosMemoryFree(pDelFile);
        }
      }
    }
  } else {
    ASSERT(pTsdb->fs.pDelFile == NULL);
  }

  // aDFileSet
  int32_t iOld = 0;
  int32_t iNew = 0;
  while (true) {
    int32_t   nOld = taosArrayGetSize(pTsdb->fs.aDFileSet);
    int32_t   nNew = taosArrayGetSize(pFS->aDFileSet);
    SDFileSet fSet = {0};
    int8_t    sameDisk = 0;

    if (iOld >= nOld && iNew >= nNew) break;

    SDFileSet *pSetOld = (iOld < nOld) ? taosArrayGet(pTsdb->fs.aDFileSet, iOld) : NULL;
    SDFileSet *pSetNew = (iNew < nNew) ? taosArrayGet(pFS->aDFileSet, iNew) : NULL;

    if (pSetOld && pSetNew) {
      if (pSetOld->fid == pSetNew->fid) {
        code = tsdbMergeFileSet(pTsdb, pSetOld, pSetNew);
        TSDB_CHECK_CODE(code, lino, _exit);

        iOld++;
        iNew++;
      } else if (pSetOld->fid < pSetNew->fid) {
        code = tsdbRemoveFileSet(pTsdb, pSetOld);
        TSDB_CHECK_CODE(code, lino, _exit);
        taosArrayRemove(pTsdb->fs.aDFileSet, iOld);
      } else {
        code = tsdbNewFileSet(pTsdb, &fSet, pSetNew);
        TSDB_CHECK_CODE(code, lino, _exit)

        if (taosArrayInsert(pTsdb->fs.aDFileSet, iOld, &fSet) == NULL) {
          code = TSDB_CODE_OUT_OF_MEMORY;
          TSDB_CHECK_CODE(code, lino, _exit);
        }

        iOld++;
        iNew++;
      }
    } else if (pSetOld) {
      code = tsdbRemoveFileSet(pTsdb, pSetOld);
      TSDB_CHECK_CODE(code, lino, _exit);
      taosArrayRemove(pTsdb->fs.aDFileSet, iOld);
    } else {
      code = tsdbNewFileSet(pTsdb, &fSet, pSetNew);
      TSDB_CHECK_CODE(code, lino, _exit)

      if (taosArrayInsert(pTsdb->fs.aDFileSet, iOld, &fSet) == NULL) {
        code = TSDB_CODE_OUT_OF_MEMORY;
        TSDB_CHECK_CODE(code, lino, _exit);
      }

      iOld++;
      iNew++;
    }
  }

_exit:
  if (code) {
    tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
  }
  return code;
}

// EXPOSED APIS ====================================================================================
int32_t tsdbFSCommit(STsdb *pTsdb) {
  int32_t code = 0;
  int32_t lino = 0;

  char current[TSDB_FILENAME_LEN] = {0};
  char current_t[TSDB_FILENAME_LEN] = {0};
  tsdbGetCurrentFName(pTsdb, current, current_t);

  // rename the file
  if (taosRenameFile(current_t, current) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  // Load the new FS
  STsdbFS fs = {0};
  code = tsdbFSCreate(&fs);
  TSDB_CHECK_CODE(code, lino, _exit);

  code = tsdbLoadFSFromFile(current_t, &fs);
  TSDB_CHECK_CODE(code, lino, _exit);

  // apply file change
  code = tsdbFSApplyChange(pTsdb, &fs);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  tsdbFSDestroy(&fs);
  if (code) {
    tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(errno));
  }
  return code;
}

int32_t tsdbFSRollback(STsdb *pTsdb) {
  int32_t code = 0;
  int32_t lino = 0;

  char current_t[TSDB_FILENAME_LEN] = {0};
  tsdbGetCurrentFName(pTsdb, NULL, current_t);
  (void)taosRemoveFile(current_t);

_exit:
  if (code) {
    tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(errno));
  }
  return code;
}

int32_t tsdbFSOpen(STsdb *pTsdb, int8_t rollback) {
  int32_t code = 0;
  int32_t lino = 0;
  SVnode *pVnode = pTsdb->pVnode;

  // open handle
  code = tsdbFSCreate(&pTsdb->fs);
  TSDB_CHECK_CODE(code, lino, _exit);

  // open impl
  char current[TSDB_FILENAME_LEN] = {0};
  char current_t[TSDB_FILENAME_LEN] = {0};
  tsdbGetCurrentFName(pTsdb, current, current_t);

  if (taosCheckExistFile(current)) {
    code = tsdbLoadFSFromFile(current, &pTsdb->fs);
    TSDB_CHECK_CODE(code, lino, _exit);

    if (taosCheckExistFile(current_t)) {
      if (rollback) {
        code = tsdbFSRollback(pTsdb);
        TSDB_CHECK_CODE(code, lino, _exit);
      } else {
        code = tsdbFSCommit(pTsdb);
        TSDB_CHECK_CODE(code, lino, _exit);
      }
    }
  } else {
    // empty one
    code = tsdbSaveFSToFile(&pTsdb->fs, current);
    TSDB_CHECK_CODE(code, lino, _exit);

    ASSERT(!rollback);
  }

  // scan and fix FS
  code = tsdbScanAndTryFixFS(pTsdb);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  if (code) {
    tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
  }
  return code;
}

int32_t tsdbFSClose(STsdb *pTsdb) {
  int32_t code = 0;

  if (pTsdb->fs.pDelFile) {
    ASSERT(pTsdb->fs.pDelFile->nRef == 1);
    taosMemoryFree(pTsdb->fs.pDelFile);
  }

  for (int32_t iSet = 0; iSet < taosArrayGetSize(pTsdb->fs.aDFileSet); iSet++) {
    SDFileSet *pSet = (SDFileSet *)taosArrayGet(pTsdb->fs.aDFileSet, iSet);

    // head
    ASSERT(pSet->pHeadF->nRef == 1);
    taosMemoryFree(pSet->pHeadF);

    // data
    ASSERT(pSet->pDataF->nRef == 1);
    taosMemoryFree(pSet->pDataF);

    // sma
    ASSERT(pSet->pSmaF->nRef == 1);
    taosMemoryFree(pSet->pSmaF);

    // stt
    for (int32_t iStt = 0; iStt < pSet->nSttF; iStt++) {
      ASSERT(pSet->aSttF[iStt]->nRef == 1);
      taosMemoryFree(pSet->aSttF[iStt]);
    }
  }

  taosArrayDestroy(pTsdb->fs.aDFileSet);

  return code;
}

int32_t tsdbFSCopy(STsdb *pTsdb, STsdbFS *pFS) {
  int32_t code = 0;
  int32_t lino = 0;

  pFS->pDelFile = NULL;
  if (pFS->aDFileSet) {
    taosArrayClear(pFS->aDFileSet);
  } else {
    pFS->aDFileSet = taosArrayInit(taosArrayGetSize(pTsdb->fs.aDFileSet), sizeof(SDFileSet));
    if (pFS->aDFileSet == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      TSDB_CHECK_CODE(code, lino, _exit);
    }
  }

  if (pTsdb->fs.pDelFile) {
    pFS->pDelFile = (SDelFile *)taosMemoryMalloc(sizeof(SDelFile));
    if (pFS->pDelFile == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      TSDB_CHECK_CODE(code, lino, _exit);
    }

    *pFS->pDelFile = *pTsdb->fs.pDelFile;
  }

  for (int32_t iSet = 0; iSet < taosArrayGetSize(pTsdb->fs.aDFileSet); iSet++) {
    SDFileSet *pSet = (SDFileSet *)taosArrayGet(pTsdb->fs.aDFileSet, iSet);
    SDFileSet  fSet = {.diskId = pSet->diskId, .fid = pSet->fid};

    // head
    fSet.pHeadF = (SHeadFile *)taosMemoryMalloc(sizeof(SHeadFile));
    if (fSet.pHeadF == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      TSDB_CHECK_CODE(code, lino, _exit);
    }
    *fSet.pHeadF = *pSet->pHeadF;

    // data
    fSet.pDataF = (SDataFile *)taosMemoryMalloc(sizeof(SDataFile));
    if (fSet.pDataF == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      TSDB_CHECK_CODE(code, lino, _exit);
    }
    *fSet.pDataF = *pSet->pDataF;

    // sma
    fSet.pSmaF = (SSmaFile *)taosMemoryMalloc(sizeof(SSmaFile));
    if (fSet.pSmaF == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      TSDB_CHECK_CODE(code, lino, _exit);
    }
    *fSet.pSmaF = *pSet->pSmaF;

    // stt
    for (fSet.nSttF = 0; fSet.nSttF < pSet->nSttF; fSet.nSttF++) {
      fSet.aSttF[fSet.nSttF] = (SSttFile *)taosMemoryMalloc(sizeof(SSttFile));
      if (fSet.aSttF[fSet.nSttF] == NULL) {
        code = TSDB_CODE_OUT_OF_MEMORY;
        TSDB_CHECK_CODE(code, lino, _exit);
      }
      *fSet.aSttF[fSet.nSttF] = *pSet->aSttF[fSet.nSttF];
    }

    if (taosArrayPush(pFS->aDFileSet, &fSet) == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      TSDB_CHECK_CODE(code, lino, _exit);
    }
  }

_exit:
  if (code) {
    tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
  }
  return code;
}

int32_t tsdbFSUpsertDelFile(STsdbFS *pFS, SDelFile *pDelFile) {
  int32_t code = 0;

  if (pFS->pDelFile == NULL) {
    pFS->pDelFile = (SDelFile *)taosMemoryMalloc(sizeof(SDelFile));
    if (pFS->pDelFile == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _exit;
    }
  }
  *pFS->pDelFile = *pDelFile;

_exit:
  return code;
}

int32_t tsdbFSUpsertFSet(STsdbFS *pFS, SDFileSet *pSet) {
  int32_t code = 0;
  int32_t idx = taosArraySearchIdx(pFS->aDFileSet, pSet, tDFileSetCmprFn, TD_GE);

  if (idx < 0) {
    idx = taosArrayGetSize(pFS->aDFileSet);
  } else {
    SDFileSet *pDFileSet = (SDFileSet *)taosArrayGet(pFS->aDFileSet, idx);
    int32_t    c = tDFileSetCmprFn(pSet, pDFileSet);
    if (c == 0) {
      *pDFileSet->pHeadF = *pSet->pHeadF;
      *pDFileSet->pDataF = *pSet->pDataF;
      *pDFileSet->pSmaF = *pSet->pSmaF;
      // stt
      if (pSet->nSttF > pDFileSet->nSttF) {
        ASSERT(pSet->nSttF == pDFileSet->nSttF + 1);

        pDFileSet->aSttF[pDFileSet->nSttF] = (SSttFile *)taosMemoryMalloc(sizeof(SSttFile));
        if (pDFileSet->aSttF[pDFileSet->nSttF] == NULL) {
          code = TSDB_CODE_OUT_OF_MEMORY;
          goto _exit;
        }
        *pDFileSet->aSttF[pDFileSet->nSttF] = *pSet->aSttF[pSet->nSttF - 1];
        pDFileSet->nSttF++;
      } else if (pSet->nSttF < pDFileSet->nSttF) {
        ASSERT(pSet->nSttF == 1);
        for (int32_t iStt = 1; iStt < pDFileSet->nSttF; iStt++) {
          taosMemoryFree(pDFileSet->aSttF[iStt]);
        }

        *pDFileSet->aSttF[0] = *pSet->aSttF[0];
        pDFileSet->nSttF = 1;
      } else {
        for (int32_t iStt = 0; iStt < pSet->nSttF; iStt++) {
          *pDFileSet->aSttF[iStt] = *pSet->aSttF[iStt];
        }
      }

      goto _exit;
    }
  }

  ASSERT(pSet->nSttF == 1);
  SDFileSet fSet = {.diskId = pSet->diskId, .fid = pSet->fid, .nSttF = 1};

  // head
  fSet.pHeadF = (SHeadFile *)taosMemoryMalloc(sizeof(SHeadFile));
  if (fSet.pHeadF == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }
  *fSet.pHeadF = *pSet->pHeadF;

  // data
  fSet.pDataF = (SDataFile *)taosMemoryMalloc(sizeof(SDataFile));
  if (fSet.pDataF == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }
  *fSet.pDataF = *pSet->pDataF;

  // sma
  fSet.pSmaF = (SSmaFile *)taosMemoryMalloc(sizeof(SSmaFile));
  if (fSet.pSmaF == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }
  *fSet.pSmaF = *pSet->pSmaF;

  // stt
  fSet.aSttF[0] = (SSttFile *)taosMemoryMalloc(sizeof(SSttFile));
  if (fSet.aSttF[0] == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }
  *fSet.aSttF[0] = *pSet->aSttF[0];

  if (taosArrayInsert(pFS->aDFileSet, idx, &fSet) == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

_exit:
  return code;
}

int32_t tsdbFSPrepareCommit(STsdb *pTsdb, STsdbFS *pFSNew) {
  int32_t code = 0;
  int32_t lino = 0;
  char    tfname[TSDB_FILENAME_LEN];

  tsdbGetCurrentFName(pTsdb, NULL, tfname);

  // gnrt CURRENT.t
  code = tsdbSaveFSToFile(pFSNew, tfname);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  if (code) {
    tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
  }
  return code;
}

int32_t tsdbFSRef(STsdb *pTsdb, STsdbFS *pFS) {
  int32_t code = 0;
  int32_t nRef;

  pFS->aDFileSet = taosArrayInit(taosArrayGetSize(pTsdb->fs.aDFileSet), sizeof(SDFileSet));
  if (pFS->aDFileSet == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

  pFS->pDelFile = pTsdb->fs.pDelFile;
  if (pFS->pDelFile) {
    nRef = atomic_fetch_add_32(&pFS->pDelFile->nRef, 1);
    ASSERT(nRef > 0);
  }

  SDFileSet fSet;
  for (int32_t iSet = 0; iSet < taosArrayGetSize(pTsdb->fs.aDFileSet); iSet++) {
    SDFileSet *pSet = (SDFileSet *)taosArrayGet(pTsdb->fs.aDFileSet, iSet);
    fSet = *pSet;

    nRef = atomic_fetch_add_32(&pSet->pHeadF->nRef, 1);
    ASSERT(nRef > 0);

    nRef = atomic_fetch_add_32(&pSet->pDataF->nRef, 1);
    ASSERT(nRef > 0);

    nRef = atomic_fetch_add_32(&pSet->pSmaF->nRef, 1);
    ASSERT(nRef > 0);

    for (int32_t iStt = 0; iStt < pSet->nSttF; iStt++) {
      nRef = atomic_fetch_add_32(&pSet->aSttF[iStt]->nRef, 1);
      ASSERT(nRef > 0);
    }

    if (taosArrayPush(pFS->aDFileSet, &fSet) == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _exit;
    }
  }

_exit:
  return code;
}

void tsdbFSUnref(STsdb *pTsdb, STsdbFS *pFS) {
  int32_t nRef;
  char    fname[TSDB_FILENAME_LEN];

  if (pFS->pDelFile) {
    nRef = atomic_sub_fetch_32(&pFS->pDelFile->nRef, 1);
    ASSERT(nRef >= 0);
    if (nRef == 0) {
      tsdbDelFileName(pTsdb, pFS->pDelFile, fname);
      (void)taosRemoveFile(fname);
      taosMemoryFree(pFS->pDelFile);
    }
  }

  for (int32_t iSet = 0; iSet < taosArrayGetSize(pFS->aDFileSet); iSet++) {
    SDFileSet *pSet = (SDFileSet *)taosArrayGet(pFS->aDFileSet, iSet);

    // head
    nRef = atomic_sub_fetch_32(&pSet->pHeadF->nRef, 1);
    ASSERT(nRef >= 0);
    if (nRef == 0) {
      tsdbHeadFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pHeadF, fname);
      (void)taosRemoveFile(fname);
      taosMemoryFree(pSet->pHeadF);
    }

    // data
    nRef = atomic_sub_fetch_32(&pSet->pDataF->nRef, 1);
    ASSERT(nRef >= 0);
    if (nRef == 0) {
      tsdbDataFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pDataF, fname);
      taosRemoveFile(fname);
      taosMemoryFree(pSet->pDataF);
    }

    // sma
    nRef = atomic_sub_fetch_32(&pSet->pSmaF->nRef, 1);
    ASSERT(nRef >= 0);
    if (nRef == 0) {
      tsdbSmaFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pSmaF, fname);
      taosRemoveFile(fname);
      taosMemoryFree(pSet->pSmaF);
    }

    // stt
    for (int32_t iStt = 0; iStt < pSet->nSttF; iStt++) {
      nRef = atomic_sub_fetch_32(&pSet->aSttF[iStt]->nRef, 1);
      ASSERT(nRef >= 0);
      if (nRef == 0) {
        tsdbSttFileName(pTsdb, pSet->diskId, pSet->fid, pSet->aSttF[iStt], fname);
        taosRemoveFile(fname);
        taosMemoryFree(pSet->aSttF[iStt]);
        /* code */
      }
    }
  }

  taosArrayDestroy(pFS->aDFileSet);
}