#4875

Committed 09 Dec 2025 01:22AM UTC coverage: 64.472% (-0.2%) from 64.623%

Build # #4875

Build Type

push

travis-ci

Committed by

guanshengliang

Commit Message

fix: temporarily disable memory leak detection for UDF tests (#33856)

Run Details

162014 of 251293 relevant lines covered (64.47%)

104318075.66 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

1.34

/source/dnode/vnode/src/tsdb/tsdbUpgrade.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 "tsdbUpgrade.h"

// old
#include "tsdb.h"
// extern void    tsdbGetCurrentFName(STsdb *pTsdb, char *current, char *current_t);

// new
#include "tsdbDataFileRW.h"
#include "tsdbFS2.h"
#include "tsdbSttFileRW.h"

static int32_t tsdbUpgradeHead(STsdb *tsdb, SDFileSet *pDFileSet, SDataFReader *reader, STFileSet *fset) {
  int32_t code = 0;
  int32_t lino = 0;

  // init
  struct {
    // config
    int32_t maxRow;
    int8_t  cmprAlg;
    int32_t szPage;
    SBuffer buffers[10];
    // reader
    SArray    *aBlockIdx;
    SMapData   mDataBlk[1];
    SBlockData blockData[1];
    // writer
    STsdbFD      *fd;
    SBrinBlock    brinBlock[1];
    TBrinBlkArray brinBlkArray[1];
    SHeadFooter   footer[1];
  } ctx[1] = {{
      .maxRow = tsdb->pVnode->config.tsdbCfg.maxRows,
      .cmprAlg = tsdb->pVnode->config.tsdbCfg.compression,
      .szPage = tsdb->pVnode->config.tsdbPageSize,
  }};

  // read SBlockIdx array
  if ((ctx->aBlockIdx = taosArrayInit(0, sizeof(SBlockIdx))) == NULL) {
    TAOS_CHECK_GOTO(terrno, &lino, _exit);
  }

  TAOS_CHECK_GOTO(tsdbReadBlockIdx(reader, ctx->aBlockIdx), &lino, _exit);

  if (taosArrayGetSize(ctx->aBlockIdx) > 0) {
    // init/open file fd
    STFile file = {
        .type = TSDB_FTYPE_HEAD,
        .did = pDFileSet->diskId,
        .fid = fset->fid,
        .cid = pDFileSet->pHeadF->commitID,
        .size = pDFileSet->pHeadF->size,
        .minVer = VERSION_MAX,
        .maxVer = VERSION_MIN,
    };

    TAOS_CHECK_GOTO(tsdbTFileObjInit(tsdb, &file, &fset->farr[TSDB_FTYPE_HEAD]), &lino, _exit);

    // open fd
    char fname[TSDB_FILENAME_LEN];
    tsdbTFileName(tsdb, &file, fname);

    TAOS_CHECK_GOTO(tsdbOpenFile(fname, tsdb, TD_FILE_READ | TD_FILE_WRITE, &ctx->fd, 0), &lino, _exit);

    // convert
    for (int32_t iBlockIdx = 0; iBlockIdx < taosArrayGetSize(ctx->aBlockIdx); ++iBlockIdx) {
      SBlockIdx *pBlockIdx = taosArrayGet(ctx->aBlockIdx, iBlockIdx);

      TAOS_CHECK_GOTO(tsdbReadDataBlk(reader, pBlockIdx, ctx->mDataBlk), &lino, _exit);

      for (int32_t iDataBlk = 0; iDataBlk < ctx->mDataBlk->nItem; ++iDataBlk) {
        SDataBlk dataBlk[1];
        tMapDataGetItemByIdx(ctx->mDataBlk, iDataBlk, dataBlk, tGetDataBlk);

        SBrinRecord record = {
            .suid = pBlockIdx->suid,
            .uid = pBlockIdx->uid,
            .firstKey =
                (STsdbRowKey){
                    .key =
                        (SRowKey){
                            .ts = dataBlk->minKey.ts,
                            .numOfPKs = 0,
                        },
                    .version = dataBlk->minKey.version,
                },
            .lastKey =
                (STsdbRowKey){
                    .key =
                        (SRowKey){
                            .ts = dataBlk->maxKey.ts,
                            .numOfPKs = 0,
                        },
                    .version = dataBlk->maxKey.version,
                },
            .minVer = dataBlk->minVer,
            .maxVer = dataBlk->maxVer,
            .blockOffset = dataBlk->aSubBlock->offset,
            .smaOffset = dataBlk->smaInfo.offset,
            .blockSize = dataBlk->aSubBlock->szBlock,
            .blockKeySize = dataBlk->aSubBlock->szKey,
            .smaSize = dataBlk->smaInfo.size,
            .numRow = dataBlk->nRow,
            .count = dataBlk->nRow,
        };

        if (dataBlk->hasDup) {
          record.count = 0;
        }

        TAOS_CHECK_GOTO(tBrinBlockPut(ctx->brinBlock, &record), &lino, _exit);

        if (ctx->brinBlock->numOfRecords >= ctx->maxRow) {
          SVersionRange range = {.minVer = VERSION_MAX, .maxVer = VERSION_MIN};
          SEncryptData *pEncryptData = &(tsdb->pVnode->config.tsdbCfg.encryptData);
          code = tsdbFileWriteBrinBlock(ctx->fd, ctx->brinBlock, ctx->cmprAlg, &fset->farr[TSDB_FTYPE_HEAD]->f->size,
                                        ctx->brinBlkArray, ctx->buffers, &range, pEncryptData);
          TSDB_CHECK_CODE(code, lino, _exit);
        }
      }
    }

    SEncryptData *pEncryptData = &(tsdb->pVnode->config.tsdbCfg.encryptData);

    if (ctx->brinBlock->numOfRecords > 0) {
      SVersionRange range = {.minVer = VERSION_MAX, .maxVer = VERSION_MIN};
      TAOS_CHECK_GOTO(
          tsdbFileWriteBrinBlock(ctx->fd, ctx->brinBlock, ctx->cmprAlg, &fset->farr[TSDB_FTYPE_HEAD]->f->size,
                                 ctx->brinBlkArray, ctx->buffers, &range, pEncryptData),
          &lino, _exit);
    }

    TAOS_CHECK_GOTO(tsdbFileWriteBrinBlk(ctx->fd, ctx->brinBlkArray, ctx->footer->brinBlkPtr,
                                         &fset->farr[TSDB_FTYPE_HEAD]->f->size, pEncryptData),
                    &lino, _exit);

    code = tsdbFileWriteHeadFooter(ctx->fd, &fset->farr[TSDB_FTYPE_HEAD]->f->size, ctx->footer, pEncryptData);
    TSDB_CHECK_CODE(code, lino, _exit);

    TAOS_CHECK_GOTO(tsdbFsyncFile(ctx->fd, pEncryptData), &lino, _exit);

    tsdbCloseFile(&ctx->fd);
  }

_exit:
  if (code) {
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, lino, tstrerror(code));
  }
  TARRAY2_DESTROY(ctx->brinBlkArray, NULL);
  tBrinBlockDestroy(ctx->brinBlock);
  tBlockDataDestroy(ctx->blockData);
  tMapDataClear(ctx->mDataBlk);
  taosArrayDestroy(ctx->aBlockIdx);
  for (int32_t i = 0; i < ARRAY_SIZE(ctx->buffers); ++i) {
    tBufferDestroy(ctx->buffers + i);
  }
  return code;
}

static int32_t tsdbUpgradeData(STsdb *tsdb, SDFileSet *pDFileSet, SDataFReader *reader, STFileSet *fset) {
  int32_t code = 0;
  int32_t lino = 0;

  if (fset->farr[TSDB_FTYPE_HEAD] == NULL) {
    return 0;
  }

  STFile file = {
      .type = TSDB_FTYPE_DATA,
      .did = pDFileSet->diskId,
      .fid = fset->fid,
      .cid = pDFileSet->pDataF->commitID,
      .size = pDFileSet->pDataF->size,
      .minVer = VERSION_MAX,
      .maxVer = VERSION_MIN,
  };

  TAOS_CHECK_GOTO(tsdbTFileObjInit(tsdb, &file, &fset->farr[TSDB_FTYPE_DATA]), &lino, _exit);

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

static int32_t tsdbUpgradeSma(STsdb *tsdb, SDFileSet *pDFileSet, SDataFReader *reader, STFileSet *fset) {
  int32_t code = 0;
  int32_t lino = 0;

  if (fset->farr[TSDB_FTYPE_HEAD] == NULL) {
    return 0;
  }

  STFile file = {
      .type = TSDB_FTYPE_SMA,
      .did = pDFileSet->diskId,
      .fid = fset->fid,
      .cid = pDFileSet->pSmaF->commitID,
      .size = pDFileSet->pSmaF->size,
      .minVer = VERSION_MAX,
      .maxVer = VERSION_MIN,
  };

  TAOS_CHECK_GOTO(tsdbTFileObjInit(tsdb, &file, &fset->farr[TSDB_FTYPE_SMA]), &lino, _exit);

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

static int32_t tsdbUpgradeSttFile(STsdb *tsdb, SDFileSet *pDFileSet, SDataFReader *reader, STFileSet *fset,
                                  int32_t iStt, SSttLvl *lvl) {
  int32_t code = 0;
  int32_t lino = 0;

  SArray *aSttBlk = taosArrayInit(0, sizeof(SSttBlk));
  if (aSttBlk == NULL) {
    TAOS_CHECK_GOTO(terrno, &lino, _exit);
  }

  TAOS_CHECK_GOTO(tsdbReadSttBlk(reader, iStt, aSttBlk), &lino, _exit);

  if (taosArrayGetSize(aSttBlk) > 0) {
    SSttFile  *pSttF = pDFileSet->aSttF[iStt];
    STFileObj *fobj;
    struct {
      int32_t szPage;

      // writer
      STsdbFD     *fd;
      TSttBlkArray sttBlkArray[1];
      SSttFooter   footer[1];
    } ctx[1] = {{
        .szPage = tsdb->pVnode->config.tsdbPageSize,
    }};

    STFile file = {
        .type = TSDB_FTYPE_STT,
        .did = pDFileSet->diskId,
        .fid = fset->fid,
        .cid = pSttF->commitID,
        .size = pSttF->size,
        .minVer = VERSION_MAX,
        .maxVer = VERSION_MIN,
    };
    TAOS_CHECK_GOTO(tsdbTFileObjInit(tsdb, &file, &fobj), &lino, _exit);

    TAOS_CHECK_GOTO(tsdbOpenFile(fobj->fname, tsdb, TD_FILE_READ | TD_FILE_WRITE, &ctx->fd, 0), &lino, _exit);

    for (int32_t iSttBlk = 0; iSttBlk < taosArrayGetSize(aSttBlk); iSttBlk++) {
      TAOS_CHECK_GOTO(TARRAY2_APPEND_PTR(ctx->sttBlkArray, (SSttBlk *)taosArrayGet(aSttBlk, iSttBlk)), &lino, _exit);
    }

    SEncryptData *pEncryptData = &(tsdb->pVnode->config.tsdbCfg.encryptData);

    code = tsdbFileWriteSttBlk(ctx->fd, ctx->sttBlkArray, ctx->footer->sttBlkPtr, &fobj->f->size, pEncryptData);
    TSDB_CHECK_CODE(code, lino, _exit1);

    TAOS_CHECK_GOTO(tsdbFileWriteSttFooter(ctx->fd, ctx->footer, &fobj->f->size, pEncryptData), &lino, _exit);

    TAOS_CHECK_GOTO(tsdbFsyncFile(ctx->fd, pEncryptData), &lino, _exit);

    tsdbCloseFile(&ctx->fd);

    TAOS_CHECK_GOTO(TARRAY2_APPEND(lvl->fobjArr, fobj), &lino, _exit);

  _exit1:
    TARRAY2_DESTROY(ctx->sttBlkArray, NULL);
  }

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

static int32_t tsdbUpgradeStt(STsdb *tsdb, SDFileSet *pDFileSet, SDataFReader *reader, STFileSet *fset) {
  int32_t code = 0;
  int32_t lino = 0;

  if (pDFileSet->nSttF == 0) {
    return 0;
  }

  SSttLvl *lvl;
  TAOS_CHECK_GOTO(tsdbSttLvlInit(0, &lvl), &lino, _exit);

  for (int32_t iStt = 0; iStt < pDFileSet->nSttF; ++iStt) {
    TAOS_CHECK_GOTO(tsdbUpgradeSttFile(tsdb, pDFileSet, reader, fset, iStt, lvl), &lino, _exit);
  }

  if (TARRAY2_SIZE(lvl->fobjArr) > 0) {
    TAOS_CHECK_GOTO(TARRAY2_APPEND(fset->lvlArr, lvl), &lino, _exit);
  } else {
    tsdbSttLvlClear(&lvl);
  }

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

static int32_t tsdbUpgradeFileSet(STsdb *tsdb, SDFileSet *pDFileSet, TFileSetArray *fileSetArray) {
  int32_t code = 0;
  int32_t lino = 0;

  tsdbInfo("vgId:%d upgrade file set start, fid:%d", TD_VID(tsdb->pVnode), pDFileSet->fid);

  SDataFReader *reader;
  STFileSet    *fset;

  TAOS_CHECK_GOTO(tsdbTFileSetInit(pDFileSet->fid, &fset), &lino, _exit);

  TAOS_CHECK_GOTO(tsdbDataFReaderOpen(&reader, tsdb, pDFileSet), &lino, _exit);

  // .head
  TAOS_CHECK_GOTO(tsdbUpgradeHead(tsdb, pDFileSet, reader, fset), &lino, _exit);

  // .data
  TAOS_CHECK_GOTO(tsdbUpgradeData(tsdb, pDFileSet, reader, fset), &lino, _exit);

  // .sma
  TAOS_CHECK_GOTO(tsdbUpgradeSma(tsdb, pDFileSet, reader, fset), &lino, _exit);

  // .stt
  if (pDFileSet->nSttF > 0) {
    TAOS_CHECK_GOTO(tsdbUpgradeStt(tsdb, pDFileSet, reader, fset), &lino, _exit);
  }

  tsdbDataFReaderClose(&reader);

  TAOS_CHECK_GOTO(TARRAY2_APPEND(fileSetArray, fset), &lino, _exit);

  tsdbInfo("vgId:%d upgrade file set end, fid:%d", TD_VID(tsdb->pVnode), pDFileSet->fid);

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

static int32_t tsdbUpgradeOpenTombFile(STsdb *tsdb, STFileSet *fset, STsdbFD **fd, STFileObj **fobj, bool *toStt) {
  int32_t code = 0;
  int32_t lino = 0;

  if (TARRAY2_SIZE(fset->lvlArr) == 0) {  // to .tomb file
    *toStt = false;

    STFile file = {
        .type = TSDB_FTYPE_TOMB,
        .did = fset->farr[TSDB_FTYPE_HEAD]->f->did,
        .fid = fset->fid,
        .cid = 0,
        .size = 0,
        .minVer = VERSION_MAX,
        .maxVer = VERSION_MIN,
    };

    TAOS_CHECK_GOTO(tsdbTFileObjInit(tsdb, &file, fobj), &lino, _exit);

    fset->farr[TSDB_FTYPE_TOMB] = *fobj;
  } else {  // to .stt file
    *toStt = true;
    SSttLvl *lvl = TARRAY2_GET(fset->lvlArr, 0);

    STFile file = {
        .type = TSDB_FTYPE_STT,
        .did = TARRAY2_GET(lvl->fobjArr, 0)->f->did,
        .fid = fset->fid,
        .cid = 0,
        .size = 0,
        .minVer = VERSION_MAX,
        .maxVer = VERSION_MIN,
    };

    TAOS_CHECK_GOTO(tsdbTFileObjInit(tsdb, &file, fobj), &lino, _exit);

    TAOS_CHECK_GOTO(TARRAY2_APPEND(lvl->fobjArr, fobj[0]), &lino, _exit);
  }

  char fname[TSDB_FILENAME_LEN] = {0};
  TAOS_CHECK_GOTO(
      tsdbOpenFile(fobj[0]->fname, tsdb, TD_FILE_READ | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_CREATE, fd, 0), &lino,
      _exit);

  uint8_t hdr[TSDB_FHDR_SIZE] = {0};
  SEncryptData *pEncryptData = &(tsdb->pVnode->config.tsdbCfg.encryptData);

  TAOS_CHECK_GOTO(tsdbWriteFile(fd[0], 0, hdr, TSDB_FHDR_SIZE, pEncryptData), &lino, _exit);
  fobj[0]->f->size += TSDB_FHDR_SIZE;

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

static int32_t tsdbDumpTombDataToFSet(STsdb *tsdb, SDelFReader *reader, SArray *aDelIdx, STFileSet *fset) {
  int32_t code = 0;
  int32_t lino = 0;

  struct {
    // context
    bool    toStt;
    int8_t  cmprAlg;
    int32_t maxRow;
    int64_t minKey;
    int64_t maxKey;
    SBuffer buffers[10];
    // reader
    SArray *aDelData;
    // writer
    STsdbFD      *fd;
    STFileObj    *fobj;
    STombBlock    tombBlock[1];
    TTombBlkArray tombBlkArray[1];
    STombFooter   tombFooter[1];
    SSttFooter    sttFooter[1];
  } ctx[1] = {{
      .maxRow = tsdb->pVnode->config.tsdbCfg.maxRows,
      .cmprAlg = tsdb->pVnode->config.tsdbCfg.compression,
  }};

  SEncryptData *pEncryptData = &(tsdb->pVnode->config.tsdbCfg.encryptData);

  tsdbFidKeyRange(fset->fid, tsdb->keepCfg.days, tsdb->keepCfg.precision, &ctx->minKey, &ctx->maxKey);

  if ((ctx->aDelData = taosArrayInit(0, sizeof(SDelData))) == NULL) {
    TAOS_CHECK_GOTO(terrno, &lino, _exit);
  }

  for (int32_t iDelIdx = 0; iDelIdx < taosArrayGetSize(aDelIdx); iDelIdx++) {
    SDelIdx *pDelIdx = (SDelIdx *)taosArrayGet(aDelIdx, iDelIdx);

    TAOS_CHECK_GOTO(tsdbReadDelData(reader, pDelIdx, ctx->aDelData), &lino, _exit);

    for (int32_t iDelData = 0; iDelData < taosArrayGetSize(ctx->aDelData); iDelData++) {
      SDelData *pDelData = (SDelData *)taosArrayGet(ctx->aDelData, iDelData);

      STombRecord record = {
          .suid = pDelIdx->suid,
          .uid = pDelIdx->uid,
          .version = pDelData->version,
          .skey = pDelData->sKey,
          .ekey = pDelData->eKey,
      };

      TAOS_CHECK_GOTO(tTombBlockPut(ctx->tombBlock, &record), &lino, _exit);

      if (TOMB_BLOCK_SIZE(ctx->tombBlock) > ctx->maxRow) {
        if (ctx->fd == NULL) {
          TAOS_CHECK_GOTO(tsdbUpgradeOpenTombFile(tsdb, fset, &ctx->fd, &ctx->fobj, &ctx->toStt), &lino, _exit);
        }
        SVersionRange tombRange = {.minVer = VERSION_MAX, .maxVer = VERSION_MIN};
        TAOS_CHECK_GOTO(tsdbFileWriteTombBlock(ctx->fd, ctx->tombBlock, ctx->cmprAlg, &ctx->fobj->f->size,
                                               ctx->tombBlkArray, ctx->buffers, &tombRange, pEncryptData),
                        &lino, _exit);
      }
    }
  }

  if (TOMB_BLOCK_SIZE(ctx->tombBlock) > 0) {
    if (ctx->fd == NULL) {
      TAOS_CHECK_GOTO(tsdbUpgradeOpenTombFile(tsdb, fset, &ctx->fd, &ctx->fobj, &ctx->toStt), &lino, _exit);
    }
    SVersionRange tombRange = {.minVer = VERSION_MAX, .maxVer = VERSION_MIN};
    TAOS_CHECK_GOTO(tsdbFileWriteTombBlock(ctx->fd, ctx->tombBlock, ctx->cmprAlg, &ctx->fobj->f->size,
                                           ctx->tombBlkArray, ctx->buffers, &tombRange, pEncryptData),
                    &lino, _exit);
  }

  if (ctx->fd != NULL) {
    if (ctx->toStt) {
      code = tsdbFileWriteTombBlk(ctx->fd, ctx->tombBlkArray, ctx->sttFooter->tombBlkPtr, &ctx->fobj->f->size,
                                  pEncryptData);
      TSDB_CHECK_CODE(code, lino, _exit);

      code = tsdbFileWriteSttFooter(ctx->fd, ctx->sttFooter, &ctx->fobj->f->size, pEncryptData);
      TSDB_CHECK_CODE(code, lino, _exit);
    } else {
      code = tsdbFileWriteTombBlk(ctx->fd, ctx->tombBlkArray, ctx->tombFooter->tombBlkPtr, &ctx->fobj->f->size,
                                  pEncryptData);
      TSDB_CHECK_CODE(code, lino, _exit);

      code = tsdbFileWriteTombFooter(ctx->fd, ctx->tombFooter, &ctx->fobj->f->size, pEncryptData);
      TSDB_CHECK_CODE(code, lino, _exit);
    }

    TAOS_CHECK_GOTO(tsdbFsyncFile(ctx->fd, pEncryptData), &lino, _exit);

    tsdbCloseFile(&ctx->fd);
  }

_exit:
  if (code) {
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, lino, tstrerror(code));
  }
  for (int32_t i = 0; i < ARRAY_SIZE(ctx->buffers); i++) {
    tBufferDestroy(ctx->buffers + i);
  }
  TARRAY2_DESTROY(ctx->tombBlkArray, NULL);
  tTombBlockDestroy(ctx->tombBlock);
  taosArrayDestroy(ctx->aDelData);
  return code;
}

static int32_t tsdbUpgradeTombFile(STsdb *tsdb, SDelFile *pDelFile, TFileSetArray *fileSetArray) {
  int32_t code = 0;
  int32_t lino = 0;

  SDelFReader *reader = NULL;
  SArray      *aDelIdx = NULL;

  if ((aDelIdx = taosArrayInit(0, sizeof(SDelIdx))) == NULL) {
    TAOS_CHECK_GOTO(terrno, &lino, _exit);
  }

  TAOS_CHECK_GOTO(tsdbDelFReaderOpen(&reader, pDelFile, tsdb), &lino, _exit);

  TAOS_CHECK_GOTO(tsdbReadDelIdx(reader, aDelIdx), &lino, _exit);

  if (taosArrayGetSize(aDelIdx) > 0) {
    STFileSet *fset;
    TARRAY2_FOREACH(fileSetArray, fset) {
      TAOS_CHECK_GOTO(tsdbDumpTombDataToFSet(tsdb, reader, aDelIdx, fset), &lino, _exit);
    }
  }

_exit:
  if (code) {
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, lino, tstrerror(code));
  }
  tsdbDelFReaderClose(&reader);
  taosArrayDestroy(aDelIdx);
  return code;
}

static int32_t tsdbDoUpgradeFileSystem(STsdb *tsdb, TFileSetArray *fileSetArray) {
  int32_t code = 0;
  int32_t lino = 0;

  // upgrade each file set
  for (int32_t i = 0; i < taosArrayGetSize(tsdb->fs.aDFileSet); i++) {
    TAOS_CHECK_GOTO(tsdbUpgradeFileSet(tsdb, taosArrayGet(tsdb->fs.aDFileSet, i), fileSetArray), &lino, _exit);
  }

  // upgrade tomb file
  if (tsdb->fs.pDelFile != NULL) {
    TAOS_CHECK_GOTO(tsdbUpgradeTombFile(tsdb, tsdb->fs.pDelFile, fileSetArray), &lino, _exit);
  }

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

static int32_t tsdbUpgradeFileSystem(STsdb *tsdb, int8_t rollback) {
  int32_t code = 0;
  int32_t lino = 0;

  TFileSetArray fileSetArray[1] = {0};

  // open old file system
  TAOS_CHECK_GOTO(tsdbFSOpen(tsdb, rollback), &lino, _exit);

  TAOS_CHECK_GOTO(tsdbDoUpgradeFileSystem(tsdb, fileSetArray), &lino, _exit);

  // close file system
  TAOS_CHECK_GOTO(tsdbFSClose(tsdb), &lino, _exit);

  // save new file system
  char fname[TSDB_FILENAME_LEN];
  current_fname(tsdb, fname, TSDB_FCURRENT);
  TAOS_CHECK_GOTO(save_fs(fileSetArray, fname), &lino, _exit);

_exit:
  if (code) {
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, lino, tstrerror(code));
    TSDB_ERROR_LOG(TD_VID(tsdb->pVnode), lino, code);
  }
  TARRAY2_DESTROY(fileSetArray, tsdbTFileSetClear);
  return code;
}

int32_t tsdbCheckAndUpgradeFileSystem(STsdb *tsdb, int8_t rollback) {
  char fname[TSDB_FILENAME_LEN];

  tsdbGetCurrentFName(tsdb, fname, NULL);
  if (!taosCheckExistFile(fname)) return 0;

  TAOS_CHECK_RETURN(tsdbUpgradeFileSystem(tsdb, rollback));

  tsdbRemoveFile(fname);
  return 0;
}

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#include "tsdbUpgrade.h"
17
18	// old
19	#include "tsdb.h"
20	// extern void tsdbGetCurrentFName(STsdb pTsdb, char current, char *current_t);
21
22	// new
23	#include "tsdbDataFileRW.h"
24	#include "tsdbFS2.h"
25	#include "tsdbSttFileRW.h"
26
27	static int32_t tsdbUpgradeHead(STsdb tsdb, SDFileSet pDFileSet, SDataFReader reader, STFileSet fset) {	×
28	int32_t code = 0;	×
29	int32_t lino = 0;	×
30
31	// init
32	struct {
33	// config
34	int32_t maxRow;
35	int8_t cmprAlg;
36	int32_t szPage;
37	SBuffer buffers[10];
38	// reader
39	SArray *aBlockIdx;
40	SMapData mDataBlk[1];
41	SBlockData blockData[1];
42	// writer
43	STsdbFD *fd;
44	SBrinBlock brinBlock[1];
45	TBrinBlkArray brinBlkArray[1];
46	SHeadFooter footer[1];
47	} ctx[1] = {{	×
48	.maxRow = tsdb->pVnode->config.tsdbCfg.maxRows,	×
49	.cmprAlg = tsdb->pVnode->config.tsdbCfg.compression,	×
50	.szPage = tsdb->pVnode->config.tsdbPageSize,	×
51	}};
52
53	// read SBlockIdx array
54	if ((ctx->aBlockIdx = taosArrayInit(0, sizeof(SBlockIdx))) == NULL) {	×
55	TAOS_CHECK_GOTO(terrno, &lino, _exit);	×
56	}
57
58	TAOS_CHECK_GOTO(tsdbReadBlockIdx(reader, ctx->aBlockIdx), &lino, _exit);	×
59
60	if (taosArrayGetSize(ctx->aBlockIdx) > 0) {	×
61	// init/open file fd
62	STFile file = {	×
63	.type = TSDB_FTYPE_HEAD,
64	.did = pDFileSet->diskId,
65	.fid = fset->fid,	×
66	.cid = pDFileSet->pHeadF->commitID,	×
67	.size = pDFileSet->pHeadF->size,	×
68	.minVer = VERSION_MAX,
69	.maxVer = VERSION_MIN,
70	};
71
72	TAOS_CHECK_GOTO(tsdbTFileObjInit(tsdb, &file, &fset->farr[TSDB_FTYPE_HEAD]), &lino, _exit);	×
73
74	// open fd
75	char fname[TSDB_FILENAME_LEN];	×
76	tsdbTFileName(tsdb, &file, fname);	×
77
78	TAOS_CHECK_GOTO(tsdbOpenFile(fname, tsdb, TD_FILE_READ \| TD_FILE_WRITE, &ctx->fd, 0), &lino, _exit);	×
79
80	// convert
81	for (int32_t iBlockIdx = 0; iBlockIdx < taosArrayGetSize(ctx->aBlockIdx); ++iBlockIdx) {	×
82	SBlockIdx *pBlockIdx = taosArrayGet(ctx->aBlockIdx, iBlockIdx);	×
83
84	TAOS_CHECK_GOTO(tsdbReadDataBlk(reader, pBlockIdx, ctx->mDataBlk), &lino, _exit);	×
85
86	for (int32_t iDataBlk = 0; iDataBlk < ctx->mDataBlk->nItem; ++iDataBlk) {	×
87	SDataBlk dataBlk[1];	×
88	tMapDataGetItemByIdx(ctx->mDataBlk, iDataBlk, dataBlk, tGetDataBlk);	×
89
90	SBrinRecord record = {	×
91	.suid = pBlockIdx->suid,	×
92	.uid = pBlockIdx->uid,	×
93	.firstKey =
94	(STsdbRowKey){
95	.key =
96	(SRowKey){
97	.ts = dataBlk->minKey.ts,	×
98	.numOfPKs = 0,
99	},
100	.version = dataBlk->minKey.version,	×
101	},
102	.lastKey =
103	(STsdbRowKey){
104	.key =
105	(SRowKey){
106	.ts = dataBlk->maxKey.ts,	×
107	.numOfPKs = 0,
108	},
109	.version = dataBlk->maxKey.version,	×
110	},
111	.minVer = dataBlk->minVer,	×
112	.maxVer = dataBlk->maxVer,	×
113	.blockOffset = dataBlk->aSubBlock->offset,	×
114	.smaOffset = dataBlk->smaInfo.offset,	×
115	.blockSize = dataBlk->aSubBlock->szBlock,	×
116	.blockKeySize = dataBlk->aSubBlock->szKey,	×
117	.smaSize = dataBlk->smaInfo.size,	×
118	.numRow = dataBlk->nRow,	×
119	.count = dataBlk->nRow,	×
120	};
121
122	if (dataBlk->hasDup) {	×
123	record.count = 0;	×
124	}
125
126	TAOS_CHECK_GOTO(tBrinBlockPut(ctx->brinBlock, &record), &lino, _exit);	×
127
128	if (ctx->brinBlock->numOfRecords >= ctx->maxRow) {	×
129	SVersionRange range = {.minVer = VERSION_MAX, .maxVer = VERSION_MIN};	×
130	SEncryptData *pEncryptData = &(tsdb->pVnode->config.tsdbCfg.encryptData);	×
131	code = tsdbFileWriteBrinBlock(ctx->fd, ctx->brinBlock, ctx->cmprAlg, &fset->farr[TSDB_FTYPE_HEAD]->f->size,	×
132	ctx->brinBlkArray, ctx->buffers, &range, pEncryptData);
133	TSDB_CHECK_CODE(code, lino, _exit);	×
134	}
135	}
136	}
137
138	SEncryptData *pEncryptData = &(tsdb->pVnode->config.tsdbCfg.encryptData);	×
139
140	if (ctx->brinBlock->numOfRecords > 0) {	×
141	SVersionRange range = {.minVer = VERSION_MAX, .maxVer = VERSION_MIN};	×
142	TAOS_CHECK_GOTO(	×
143	tsdbFileWriteBrinBlock(ctx->fd, ctx->brinBlock, ctx->cmprAlg, &fset->farr[TSDB_FTYPE_HEAD]->f->size,
144	ctx->brinBlkArray, ctx->buffers, &range, pEncryptData),
145	&lino, _exit);
146	}
147
148	TAOS_CHECK_GOTO(tsdbFileWriteBrinBlk(ctx->fd, ctx->brinBlkArray, ctx->footer->brinBlkPtr,	×
149	&fset->farr[TSDB_FTYPE_HEAD]->f->size, pEncryptData),
150	&lino, _exit);
151
152	code = tsdbFileWriteHeadFooter(ctx->fd, &fset->farr[TSDB_FTYPE_HEAD]->f->size, ctx->footer, pEncryptData);	×
153	TSDB_CHECK_CODE(code, lino, _exit);	×
154
155	TAOS_CHECK_GOTO(tsdbFsyncFile(ctx->fd, pEncryptData), &lino, _exit);	×
156
157	tsdbCloseFile(&ctx->fd);	×
158	}
159
160	_exit:	×
161	if (code) {	×
162	tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, lino, tstrerror(code));	×
163	}
164	TARRAY2_DESTROY(ctx->brinBlkArray, NULL);	×
165	tBrinBlockDestroy(ctx->brinBlock);	×
166	tBlockDataDestroy(ctx->blockData);	×
167	tMapDataClear(ctx->mDataBlk);	×
168	taosArrayDestroy(ctx->aBlockIdx);	×
169	for (int32_t i = 0; i < ARRAY_SIZE(ctx->buffers); ++i) {	×
170	tBufferDestroy(ctx->buffers + i);	×
171	}
172	return code;	×
173	}
174
175	static int32_t tsdbUpgradeData(STsdb tsdb, SDFileSet pDFileSet, SDataFReader reader, STFileSet fset) {	×
176	int32_t code = 0;	×
177	int32_t lino = 0;	×
178
179	if (fset->farr[TSDB_FTYPE_HEAD] == NULL) {	×
180	return 0;	×
181	}
182
183	STFile file = {	×
184	.type = TSDB_FTYPE_DATA,
185	.did = pDFileSet->diskId,
186	.fid = fset->fid,	×
187	.cid = pDFileSet->pDataF->commitID,	×
188	.size = pDFileSet->pDataF->size,	×
189	.minVer = VERSION_MAX,
190	.maxVer = VERSION_MIN,
191	};
192
193	TAOS_CHECK_GOTO(tsdbTFileObjInit(tsdb, &file, &fset->farr[TSDB_FTYPE_DATA]), &lino, _exit);	×
194
195	_exit:	×
196	if (code) {	×
197	tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, lino, tstrerror(code));	×
198	}
199	return code;	×
200	}
201
202	static int32_t tsdbUpgradeSma(STsdb tsdb, SDFileSet pDFileSet, SDataFReader reader, STFileSet fset) {	×
203	int32_t code = 0;	×
204	int32_t lino = 0;	×
205
206	if (fset->farr[TSDB_FTYPE_HEAD] == NULL) {	×
207	return 0;	×
208	}
209
210	STFile file = {	×
211	.type = TSDB_FTYPE_SMA,
212	.did = pDFileSet->diskId,
213	.fid = fset->fid,	×
214	.cid = pDFileSet->pSmaF->commitID,	×
215	.size = pDFileSet->pSmaF->size,	×
216	.minVer = VERSION_MAX,
217	.maxVer = VERSION_MIN,
218	};
219
220	TAOS_CHECK_GOTO(tsdbTFileObjInit(tsdb, &file, &fset->farr[TSDB_FTYPE_SMA]), &lino, _exit);	×
221
222	_exit:	×
223	if (code) {	×
224	tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, lino, tstrerror(code));	×
225	}
226	return code;	×
227	}
228
229	static int32_t tsdbUpgradeSttFile(STsdb tsdb, SDFileSet pDFileSet, SDataFReader reader, STFileSet fset,	×
230	int32_t iStt, SSttLvl *lvl) {
231	int32_t code = 0;	×
232	int32_t lino = 0;	×
233
234	SArray *aSttBlk = taosArrayInit(0, sizeof(SSttBlk));	×
235	if (aSttBlk == NULL) {	×
236	TAOS_CHECK_GOTO(terrno, &lino, _exit);	×
237	}
238
239	TAOS_CHECK_GOTO(tsdbReadSttBlk(reader, iStt, aSttBlk), &lino, _exit);	×
240
241	if (taosArrayGetSize(aSttBlk) > 0) {	×
242	SSttFile *pSttF = pDFileSet->aSttF[iStt];	×
243	STFileObj *fobj;	×
244	struct {
245	int32_t szPage;
246
247	// writer
248	STsdbFD *fd;
249	TSttBlkArray sttBlkArray[1];
250	SSttFooter footer[1];
251	} ctx[1] = {{	×
252	.szPage = tsdb->pVnode->config.tsdbPageSize,	×
253	}};
254
255	STFile file = {	×
256	.type = TSDB_FTYPE_STT,
257	.did = pDFileSet->diskId,
258	.fid = fset->fid,	×
259	.cid = pSttF->commitID,	×
260	.size = pSttF->size,	×
261	.minVer = VERSION_MAX,
262	.maxVer = VERSION_MIN,
263	};
264	TAOS_CHECK_GOTO(tsdbTFileObjInit(tsdb, &file, &fobj), &lino, _exit);	×
265
266	TAOS_CHECK_GOTO(tsdbOpenFile(fobj->fname, tsdb, TD_FILE_READ \| TD_FILE_WRITE, &ctx->fd, 0), &lino, _exit);	×
267
268	for (int32_t iSttBlk = 0; iSttBlk < taosArrayGetSize(aSttBlk); iSttBlk++) {	×
269	TAOS_CHECK_GOTO(TARRAY2_APPEND_PTR(ctx->sttBlkArray, (SSttBlk *)taosArrayGet(aSttBlk, iSttBlk)), &lino, _exit);	×
270	}
271
272	SEncryptData *pEncryptData = &(tsdb->pVnode->config.tsdbCfg.encryptData);	×
273
274	code = tsdbFileWriteSttBlk(ctx->fd, ctx->sttBlkArray, ctx->footer->sttBlkPtr, &fobj->f->size, pEncryptData);	×
275	TSDB_CHECK_CODE(code, lino, _exit1);	×
276
277	TAOS_CHECK_GOTO(tsdbFileWriteSttFooter(ctx->fd, ctx->footer, &fobj->f->size, pEncryptData), &lino, _exit);	×
278
279	TAOS_CHECK_GOTO(tsdbFsyncFile(ctx->fd, pEncryptData), &lino, _exit);	×
280
281	tsdbCloseFile(&ctx->fd);	×
282
283	TAOS_CHECK_GOTO(TARRAY2_APPEND(lvl->fobjArr, fobj), &lino, _exit);	×
284
285	_exit1:	×
286	TARRAY2_DESTROY(ctx->sttBlkArray, NULL);	×
287	}
288
289	_exit:	×
290	if (code) {	×
291	tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, lino, tstrerror(code));	×
292	}
293	taosArrayDestroy(aSttBlk);	×
294	return code;	×
295	}
296
297	static int32_t tsdbUpgradeStt(STsdb tsdb, SDFileSet pDFileSet, SDataFReader reader, STFileSet fset) {	×
298	int32_t code = 0;	×
299	int32_t lino = 0;	×
300
301	if (pDFileSet->nSttF == 0) {	×
302	return 0;	×
303	}
304
305	SSttLvl *lvl;	×
306	TAOS_CHECK_GOTO(tsdbSttLvlInit(0, &lvl), &lino, _exit);	×
307
308	for (int32_t iStt = 0; iStt < pDFileSet->nSttF; ++iStt) {	×
309	TAOS_CHECK_GOTO(tsdbUpgradeSttFile(tsdb, pDFileSet, reader, fset, iStt, lvl), &lino, _exit);	×
310	}
311
312	if (TARRAY2_SIZE(lvl->fobjArr) > 0) {	×
313	TAOS_CHECK_GOTO(TARRAY2_APPEND(fset->lvlArr, lvl), &lino, _exit);	×
314	} else {
315	tsdbSttLvlClear(&lvl);	×
316	}
317
318	_exit:	×
319	if (code) {	×
320	tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, lino, tstrerror(code));	×
321	}
322	return code;	×
323	}
324
325	static int32_t tsdbUpgradeFileSet(STsdb tsdb, SDFileSet pDFileSet, TFileSetArray *fileSetArray) {	×
326	int32_t code = 0;	×
327	int32_t lino = 0;	×
328
329	tsdbInfo("vgId:%d upgrade file set start, fid:%d", TD_VID(tsdb->pVnode), pDFileSet->fid);	×
330
331	SDataFReader *reader;	×
332	STFileSet *fset;	×
333
334	TAOS_CHECK_GOTO(tsdbTFileSetInit(pDFileSet->fid, &fset), &lino, _exit);	×
335
336	TAOS_CHECK_GOTO(tsdbDataFReaderOpen(&reader, tsdb, pDFileSet), &lino, _exit);	×
337
338	// .head
339	TAOS_CHECK_GOTO(tsdbUpgradeHead(tsdb, pDFileSet, reader, fset), &lino, _exit);	×
340
341	// .data
342	TAOS_CHECK_GOTO(tsdbUpgradeData(tsdb, pDFileSet, reader, fset), &lino, _exit);	×
343
344	// .sma
345	TAOS_CHECK_GOTO(tsdbUpgradeSma(tsdb, pDFileSet, reader, fset), &lino, _exit);	×
346
347	// .stt
348	if (pDFileSet->nSttF > 0) {	×
349	TAOS_CHECK_GOTO(tsdbUpgradeStt(tsdb, pDFileSet, reader, fset), &lino, _exit);	×
350	}
351
352	tsdbDataFReaderClose(&reader);	×
353
354	TAOS_CHECK_GOTO(TARRAY2_APPEND(fileSetArray, fset), &lino, _exit);	×
355
356	tsdbInfo("vgId:%d upgrade file set end, fid:%d", TD_VID(tsdb->pVnode), pDFileSet->fid);	×
357
358	_exit:	×
359	if (code) {	×
360	tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, lino, tstrerror(code));	×
361	}
362	return code;	×
363	}
364
365	static int32_t tsdbUpgradeOpenTombFile(STsdb tsdb, STFileSet fset, STsdbFD fd, STFileObj fobj, bool *toStt) {	×
366	int32_t code = 0;	×
367	int32_t lino = 0;	×
368
369	if (TARRAY2_SIZE(fset->lvlArr) == 0) { // to .tomb file	×
370	*toStt = false;	×
371
372	STFile file = {	×
373	.type = TSDB_FTYPE_TOMB,
374	.did = fset->farr[TSDB_FTYPE_HEAD]->f->did,	×
375	.fid = fset->fid,	×
376	.cid = 0,
377	.size = 0,
378	.minVer = VERSION_MAX,
379	.maxVer = VERSION_MIN,
380	};
381
382	TAOS_CHECK_GOTO(tsdbTFileObjInit(tsdb, &file, fobj), &lino, _exit);	×
383
384	fset->farr[TSDB_FTYPE_TOMB] = *fobj;	×
385	} else { // to .stt file
386	*toStt = true;	×
387	SSttLvl *lvl = TARRAY2_GET(fset->lvlArr, 0);	×
388
389	STFile file = {	×
390	.type = TSDB_FTYPE_STT,
391	.did = TARRAY2_GET(lvl->fobjArr, 0)->f->did,	×
392	.fid = fset->fid,	×
393	.cid = 0,
394	.size = 0,
395	.minVer = VERSION_MAX,
396	.maxVer = VERSION_MIN,
397	};
398
399	TAOS_CHECK_GOTO(tsdbTFileObjInit(tsdb, &file, fobj), &lino, _exit);	×
400
401	TAOS_CHECK_GOTO(TARRAY2_APPEND(lvl->fobjArr, fobj[0]), &lino, _exit);	×
402	}
403
404	char fname[TSDB_FILENAME_LEN] = {0};	×
405	TAOS_CHECK_GOTO(	×
406	tsdbOpenFile(fobj[0]->fname, tsdb, TD_FILE_READ \| TD_FILE_WRITE \| TD_FILE_TRUNC \| TD_FILE_CREATE, fd, 0), &lino,
407	_exit);
408
409	uint8_t hdr[TSDB_FHDR_SIZE] = {0};	×
410	SEncryptData *pEncryptData = &(tsdb->pVnode->config.tsdbCfg.encryptData);	×
411
412	TAOS_CHECK_GOTO(tsdbWriteFile(fd[0], 0, hdr, TSDB_FHDR_SIZE, pEncryptData), &lino, _exit);	×
413	fobj[0]->f->size += TSDB_FHDR_SIZE;	×
414
415	_exit:	×
416	if (code) {	×
417	tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, lino, tstrerror(code));	×
418	}
419	return code;	×
420	}
421
422	static int32_t tsdbDumpTombDataToFSet(STsdb tsdb, SDelFReader reader, SArray aDelIdx, STFileSet fset) {	×
423	int32_t code = 0;	×
424	int32_t lino = 0;	×
425
426	struct {
427	// context
428	bool toStt;
429	int8_t cmprAlg;
430	int32_t maxRow;
431	int64_t minKey;
432	int64_t maxKey;
433	SBuffer buffers[10];
434	// reader
435	SArray *aDelData;
436	// writer
437	STsdbFD *fd;
438	STFileObj *fobj;
439	STombBlock tombBlock[1];
440	TTombBlkArray tombBlkArray[1];
441	STombFooter tombFooter[1];
442	SSttFooter sttFooter[1];
443	} ctx[1] = {{	×
444	.maxRow = tsdb->pVnode->config.tsdbCfg.maxRows,	×
445	.cmprAlg = tsdb->pVnode->config.tsdbCfg.compression,	×
446	}};
447
448	SEncryptData *pEncryptData = &(tsdb->pVnode->config.tsdbCfg.encryptData);	×
449
450	tsdbFidKeyRange(fset->fid, tsdb->keepCfg.days, tsdb->keepCfg.precision, &ctx->minKey, &ctx->maxKey);	×
451
452	if ((ctx->aDelData = taosArrayInit(0, sizeof(SDelData))) == NULL) {	×
453	TAOS_CHECK_GOTO(terrno, &lino, _exit);	×
454	}
455
456	for (int32_t iDelIdx = 0; iDelIdx < taosArrayGetSize(aDelIdx); iDelIdx++) {	×
457	SDelIdx pDelIdx = (SDelIdx )taosArrayGet(aDelIdx, iDelIdx);	×
458
459	TAOS_CHECK_GOTO(tsdbReadDelData(reader, pDelIdx, ctx->aDelData), &lino, _exit);	×
460
461	for (int32_t iDelData = 0; iDelData < taosArrayGetSize(ctx->aDelData); iDelData++) {	×
462	SDelData pDelData = (SDelData )taosArrayGet(ctx->aDelData, iDelData);	×
463
464	STombRecord record = {	×
465	.suid = pDelIdx->suid,	×
466	.uid = pDelIdx->uid,	×
467	.version = pDelData->version,	×
468	.skey = pDelData->sKey,	×
469	.ekey = pDelData->eKey,	×
470	};
471
472	TAOS_CHECK_GOTO(tTombBlockPut(ctx->tombBlock, &record), &lino, _exit);	×
473
474	if (TOMB_BLOCK_SIZE(ctx->tombBlock) > ctx->maxRow) {	×
475	if (ctx->fd == NULL) {	×
476	TAOS_CHECK_GOTO(tsdbUpgradeOpenTombFile(tsdb, fset, &ctx->fd, &ctx->fobj, &ctx->toStt), &lino, _exit);	×
477	}
478	SVersionRange tombRange = {.minVer = VERSION_MAX, .maxVer = VERSION_MIN};	×
479	TAOS_CHECK_GOTO(tsdbFileWriteTombBlock(ctx->fd, ctx->tombBlock, ctx->cmprAlg, &ctx->fobj->f->size,	×
480	ctx->tombBlkArray, ctx->buffers, &tombRange, pEncryptData),
481	&lino, _exit);
482	}
483	}
484	}
485
486	if (TOMB_BLOCK_SIZE(ctx->tombBlock) > 0) {	×
487	if (ctx->fd == NULL) {	×
488	TAOS_CHECK_GOTO(tsdbUpgradeOpenTombFile(tsdb, fset, &ctx->fd, &ctx->fobj, &ctx->toStt), &lino, _exit);	×
489	}
490	SVersionRange tombRange = {.minVer = VERSION_MAX, .maxVer = VERSION_MIN};	×
491	TAOS_CHECK_GOTO(tsdbFileWriteTombBlock(ctx->fd, ctx->tombBlock, ctx->cmprAlg, &ctx->fobj->f->size,	×
492	ctx->tombBlkArray, ctx->buffers, &tombRange, pEncryptData),
493	&lino, _exit);
494	}
495
496	if (ctx->fd != NULL) {	×
497	if (ctx->toStt) {	×
498	code = tsdbFileWriteTombBlk(ctx->fd, ctx->tombBlkArray, ctx->sttFooter->tombBlkPtr, &ctx->fobj->f->size,	×
499	pEncryptData);
500	TSDB_CHECK_CODE(code, lino, _exit);	×
501
502	code = tsdbFileWriteSttFooter(ctx->fd, ctx->sttFooter, &ctx->fobj->f->size, pEncryptData);	×
503	TSDB_CHECK_CODE(code, lino, _exit);	×
504	} else {
505	code = tsdbFileWriteTombBlk(ctx->fd, ctx->tombBlkArray, ctx->tombFooter->tombBlkPtr, &ctx->fobj->f->size,	×
506	pEncryptData);
507	TSDB_CHECK_CODE(code, lino, _exit);	×
508
509	code = tsdbFileWriteTombFooter(ctx->fd, ctx->tombFooter, &ctx->fobj->f->size, pEncryptData);	×
510	TSDB_CHECK_CODE(code, lino, _exit);	×
511	}
512
513	TAOS_CHECK_GOTO(tsdbFsyncFile(ctx->fd, pEncryptData), &lino, _exit);	×
514
515	tsdbCloseFile(&ctx->fd);	×
516	}
517
518	_exit:	×
519	if (code) {	×
520	tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, lino, tstrerror(code));	×
521	}
522	for (int32_t i = 0; i < ARRAY_SIZE(ctx->buffers); i++) {	×
523	tBufferDestroy(ctx->buffers + i);	×
524	}
525	TARRAY2_DESTROY(ctx->tombBlkArray, NULL);	×
526	tTombBlockDestroy(ctx->tombBlock);	×
527	taosArrayDestroy(ctx->aDelData);	×
528	return code;	×
529	}
530
531	static int32_t tsdbUpgradeTombFile(STsdb tsdb, SDelFile pDelFile, TFileSetArray *fileSetArray) {	×
532	int32_t code = 0;	×
533	int32_t lino = 0;	×
534
535	SDelFReader *reader = NULL;	×
536	SArray *aDelIdx = NULL;	×
537
538	if ((aDelIdx = taosArrayInit(0, sizeof(SDelIdx))) == NULL) {	×
539	TAOS_CHECK_GOTO(terrno, &lino, _exit);	×
540	}
541
542	TAOS_CHECK_GOTO(tsdbDelFReaderOpen(&reader, pDelFile, tsdb), &lino, _exit);	×
543
544	TAOS_CHECK_GOTO(tsdbReadDelIdx(reader, aDelIdx), &lino, _exit);	×
545
546	if (taosArrayGetSize(aDelIdx) > 0) {	×
547	STFileSet *fset;
548	TARRAY2_FOREACH(fileSetArray, fset) {	×
549	TAOS_CHECK_GOTO(tsdbDumpTombDataToFSet(tsdb, reader, aDelIdx, fset), &lino, _exit);	×
550	}
551	}
552
553	_exit:	×
554	if (code) {	×
555	tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, lino, tstrerror(code));	×
556	}
557	tsdbDelFReaderClose(&reader);	×
558	taosArrayDestroy(aDelIdx);	×
559	return code;	×
560	}
561
562	static int32_t tsdbDoUpgradeFileSystem(STsdb tsdb, TFileSetArray fileSetArray) {	×
563	int32_t code = 0;	×
564	int32_t lino = 0;	×
565
566	// upgrade each file set
567	for (int32_t i = 0; i < taosArrayGetSize(tsdb->fs.aDFileSet); i++) {	×
568	TAOS_CHECK_GOTO(tsdbUpgradeFileSet(tsdb, taosArrayGet(tsdb->fs.aDFileSet, i), fileSetArray), &lino, _exit);	×
569	}
570
571	// upgrade tomb file
572	if (tsdb->fs.pDelFile != NULL) {	×
573	TAOS_CHECK_GOTO(tsdbUpgradeTombFile(tsdb, tsdb->fs.pDelFile, fileSetArray), &lino, _exit);	×
574	}
575
576	_exit:	×
577	if (code) {	×
578	tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, lino, tstrerror(code));	×
579	}
580	return code;	×
581	}
582
583	static int32_t tsdbUpgradeFileSystem(STsdb *tsdb, int8_t rollback) {	×
584	int32_t code = 0;	×
585	int32_t lino = 0;	×
586
587	TFileSetArray fileSetArray[1] = {0};	×
588
589	// open old file system
590	TAOS_CHECK_GOTO(tsdbFSOpen(tsdb, rollback), &lino, _exit);	×
591
592	TAOS_CHECK_GOTO(tsdbDoUpgradeFileSystem(tsdb, fileSetArray), &lino, _exit);	×
593
594	// close file system
595	TAOS_CHECK_GOTO(tsdbFSClose(tsdb), &lino, _exit);	×
596
597	// save new file system
598	char fname[TSDB_FILENAME_LEN];	×
599	current_fname(tsdb, fname, TSDB_FCURRENT);	×
600	TAOS_CHECK_GOTO(save_fs(fileSetArray, fname), &lino, _exit);	×
601
602	_exit:	×
603	if (code) {	×
604	tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, lino, tstrerror(code));	×
605	TSDB_ERROR_LOG(TD_VID(tsdb->pVnode), lino, code);	×
606	}
607	TARRAY2_DESTROY(fileSetArray, tsdbTFileSetClear);	×
608	return code;	×
609	}
610
611	int32_t tsdbCheckAndUpgradeFileSystem(STsdb *tsdb, int8_t rollback) {	4,185,351✔
612	char fname[TSDB_FILENAME_LEN];	4,179,134✔
613
614	tsdbGetCurrentFName(tsdb, fname, NULL);	4,193,376✔
615	if (!taosCheckExistFile(fname)) return 0;	4,193,493✔
616
617	TAOS_CHECK_RETURN(tsdbUpgradeFileSystem(tsdb, rollback));	×
618
619	tsdbRemoveFile(fname);	×
620	return 0;	×
621	}

taosdata / TDengine / #4875

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous