#4791

Committed 13 Oct 2025 06:50AM UTC coverage: 57.628% (-0.8%) from 58.476%

Build # #4791

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #33213 from taosdata/fix/huoh/timemoe_model_directory

fix: fix tdgpt timemoe model directory

Run Details

136628 of 303332 branches covered (45.04%)

Branch coverage included in aggregate %.

208121 of 294900 relevant lines covered (70.57%)

4250784.02 hits per line

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

66.84

/source/dnode/vnode/src/tsdb/tsdbSnapInfo.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"
#include "tsdbFS2.h"

#define TSDB_SNAP_MSG_VER 1

// fset partition
static int32_t tsdbFSetPartCmprFn(STsdbFSetPartition* x, STsdbFSetPartition* y) {
  if (x->fid < y->fid) return -1;
  if (x->fid > y->fid) return 1;
  return 0;
}

static int32_t tVersionRangeCmprFn(SVersionRange* x, SVersionRange* y) {
  if (x->minVer < y->minVer) return -1;
  if (x->minVer > y->minVer) return 1;
  if (x->maxVer < y->maxVer) return -1;
  if (x->maxVer > y->maxVer) return 1;
  return 0;
}

static int32_t tsdbTFileSetRangeCmprFn(STFileSetRange* x, STFileSetRange* y) {
  if (x->fid < y->fid) return -1;
  if (x->fid > y->fid) return 1;
  return 0;
}

STsdbFSetPartition* tsdbFSetPartitionCreate() {
  STsdbFSetPartition* pSP = taosMemoryCalloc(1, sizeof(STsdbFSetPartition));
  if (pSP == NULL) {
    return NULL;
  }
  for (int32_t i = 0; i < TSDB_FSET_RANGE_TYP_MAX; i++) {
    TARRAY2_INIT(&pSP->verRanges[i]);
  }
  return pSP;
}

void tsdbFSetPartitionClear(STsdbFSetPartition** ppSP) {
  if (ppSP == NULL || ppSP[0] == NULL) {
    return;
  }
  for (int32_t i = 0; i < TSDB_FSET_RANGE_TYP_MAX; i++) {
    TARRAY2_DESTROY(&ppSP[0]->verRanges[i], NULL);
  }
  taosMemoryFree(ppSP[0]);
  ppSP[0] = NULL;
}

static int32_t tsdbFTypeToFRangeType(tsdb_ftype_t ftype) {
  switch (ftype) {
    case TSDB_FTYPE_HEAD:
      return TSDB_FSET_RANGE_TYP_HEAD;
    case TSDB_FTYPE_DATA:
      return TSDB_FSET_RANGE_TYP_DATA;
    case TSDB_FTYPE_SMA:
      return TSDB_FSET_RANGE_TYP_SMA;
    case TSDB_FTYPE_TOMB:
      return TSDB_FSET_RANGE_TYP_TOMB;
    case TSDB_FTYPE_STT:
      return TSDB_FSET_RANGE_TYP_STT;
  }
  return TSDB_FSET_RANGE_TYP_MAX;
}

static int32_t tsdbTFileSetToFSetPartition(STFileSet* fset, STsdbFSetPartition** ppSP) {
  STsdbFSetPartition* p = tsdbFSetPartitionCreate();
  if (p == NULL) {
    return terrno;
  }

  p->fid = fset->fid;

  int32_t code = 0;
  int32_t typ = 0;
  int32_t corrupt = false;
  int32_t count = 0;
  for (int32_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) {
    if (fset->farr[ftype] == NULL) continue;
    typ = tsdbFTypeToFRangeType(ftype);
    STFile* f = fset->farr[ftype]->f;
    if (f->maxVer > fset->maxVerValid) {
      corrupt = true;
      tsdbError("skip incomplete data file: fid:%d, maxVerValid:%" PRId64 ", minVer:%" PRId64 ", maxVer:%" PRId64
                ", ftype: %d",
                fset->fid, fset->maxVerValid, f->minVer, f->maxVer, ftype);
      continue;
    }
    count++;
    SVersionRange vr = {.minVer = f->minVer, .maxVer = f->maxVer};
    code = TARRAY2_SORT_INSERT(&p->verRanges[typ], vr, tVersionRangeCmprFn);
    if (code) {
      tsdbFSetPartitionClear(&p);
      return code;
    }
  }

  typ = TSDB_FSET_RANGE_TYP_STT;
  const SSttLvl* lvl;
  TARRAY2_FOREACH(fset->lvlArr, lvl) {
    STFileObj* fobj;
    TARRAY2_FOREACH(lvl->fobjArr, fobj) {
      STFile* f = fobj->f;
      if (f->maxVer > fset->maxVerValid) {
        corrupt = true;
        tsdbError("skip incomplete stt file.fid:%d, maxVerValid:%" PRId64 ", minVer:%" PRId64 ", maxVer:%" PRId64
                  ", ftype: %d",
                  fset->fid, fset->maxVerValid, f->minVer, f->maxVer, typ);
        continue;
      }
      count++;
      SVersionRange vr = {.minVer = f->minVer, .maxVer = f->maxVer};
      code = TARRAY2_SORT_INSERT(&p->verRanges[typ], vr, tVersionRangeCmprFn);
      if (code) {
        tsdbFSetPartitionClear(&p);
        return code;
      }
    }
  }
  if (corrupt && count == 0) {
    SVersionRange vr = {.minVer = VERSION_MIN, .maxVer = fset->maxVerValid};
    code = TARRAY2_SORT_INSERT(&p->verRanges[typ], vr, tVersionRangeCmprFn);
    if (code) {
      tsdbFSetPartitionClear(&p);
      return code;
    }
  }
  ppSP[0] = p;
  return 0;
}

// fset partition list
STsdbFSetPartList* tsdbFSetPartListCreate() {
  STsdbFSetPartList* pList = taosMemoryCalloc(1, sizeof(STsdbFSetPartList));
  if (pList == NULL) {
    return NULL;
  }
  TARRAY2_INIT(pList);
  return pList;
}

void tsdbFSetPartListDestroy(STsdbFSetPartList** ppList) {
  if (ppList == NULL || ppList[0] == NULL) return;

  TARRAY2_DESTROY(ppList[0], tsdbFSetPartitionClear);
  taosMemoryFree(ppList[0]);
  ppList[0] = NULL;
}

int32_t tsdbFSetPartListToRangeDiff(STsdbFSetPartList* pList, TFileSetRangeArray** ppRanges) {
  int32_t code = 0;

  TFileSetRangeArray* pDiff = taosMemoryCalloc(1, sizeof(TFileSetRangeArray));
  if (pDiff == NULL) {
    code = terrno;
    goto _err;
  }
  TARRAY2_INIT(pDiff);

  STsdbFSetPartition* part;
  TARRAY2_FOREACH(pList, part) {
    STFileSetRange* r = taosMemoryCalloc(1, sizeof(STFileSetRange));
    if (r == NULL) {
      code = terrno;
      goto _err;
    }
    int64_t maxVerValid = -1;
    int32_t typMax = TSDB_FSET_RANGE_TYP_MAX;
    for (int32_t i = 0; i < typMax; i++) {
      SVerRangeList* iList = &part->verRanges[i];
      SVersionRange  vr = {0};
      TARRAY2_FOREACH(iList, vr) {
        if (vr.maxVer < vr.minVer) {
          continue;
        }
        maxVerValid = TMAX(maxVerValid, vr.maxVer);
      }
    }
    r->fid = part->fid;
    r->sver = maxVerValid + 1;
    r->ever = VERSION_MAX;
    tsdbDebug("range diff fid:%" PRId64 ", sver:%" PRId64 ", ever:%" PRId64, part->fid, r->sver, r->ever);
    code = TARRAY2_SORT_INSERT(pDiff, r, tsdbTFileSetRangeCmprFn);
    if (code) {
      taosMemoryFree(r);
      goto _err;
    }
  }
  ppRanges[0] = pDiff;

  tsdbInfo("pDiff size:%d", TARRAY2_SIZE(pDiff));
  return 0;

_err:
  if (pDiff) {
    tsdbTFileSetRangeArrayDestroy(&pDiff);
  }
  return code;
}

// serialization
int32_t tTsdbFSetPartListDataLenCalc(STsdbFSetPartList* pList) {
  int32_t hdrLen = sizeof(int32_t);
  int32_t datLen = 0;

  int8_t  msgVer = 1;
  int32_t len = TARRAY2_SIZE(pList);
  hdrLen += sizeof(msgVer);
  hdrLen += sizeof(len);
  datLen += hdrLen;

  for (int32_t u = 0; u < len; u++) {
    STsdbFSetPartition* p = TARRAY2_GET(pList, u);
    int32_t             typMax = TSDB_FSET_RANGE_TYP_MAX;
    int32_t             uItem = 0;
    uItem += sizeof(STsdbFSetPartition);
    uItem += sizeof(typMax);

    for (int32_t i = 0; i < typMax; i++) {
      int32_t iLen = TARRAY2_SIZE(&p->verRanges[i]);
      int32_t jItem = 0;
      jItem += sizeof(SVersionRange);
      jItem += sizeof(int64_t);
      uItem += sizeof(iLen) + jItem * iLen;
    }
    datLen += uItem;
  }
  return datLen;
}

static int32_t tSerializeTsdbFSetPartList(void* buf, int32_t bufLen, STsdbFSetPartList* pList, int32_t* encodeSize) {
  SEncoder encoder = {0};
  int8_t   reserved8 = 0;
  int16_t  reserved16 = 0;
  int64_t  reserved64 = 0;
  int8_t   msgVer = TSDB_SNAP_MSG_VER;
  int32_t  len = TARRAY2_SIZE(pList);
  int32_t  code = 0;

  tEncoderInit(&encoder, buf, bufLen);
  if ((code = tStartEncode(&encoder))) goto _exit;
  if ((code = tEncodeI8(&encoder, msgVer))) goto _exit;
  if ((code = tEncodeI32(&encoder, len))) goto _exit;

  for (int32_t u = 0; u < len; u++) {
    STsdbFSetPartition* p = TARRAY2_GET(pList, u);
    if ((code = tEncodeI64(&encoder, p->fid))) goto _exit;
    if ((code = tEncodeI8(&encoder, p->stat))) goto _exit;
    if ((code = tEncodeI8(&encoder, reserved8))) goto _exit;
    if ((code = tEncodeI16(&encoder, reserved16))) goto _exit;

    int32_t typMax = TSDB_FSET_RANGE_TYP_MAX;
    if ((code = tEncodeI32(&encoder, typMax))) goto _exit;

    for (int32_t i = 0; i < typMax; i++) {
      SVerRangeList* iList = &p->verRanges[i];
      int32_t        iLen = TARRAY2_SIZE(iList);

      if ((code = tEncodeI32(&encoder, iLen))) goto _exit;
      for (int32_t j = 0; j < iLen; j++) {
        SVersionRange r = TARRAY2_GET(iList, j);
        if ((code = tEncodeI64(&encoder, r.minVer))) goto _exit;
        if ((code = tEncodeI64(&encoder, r.maxVer))) goto _exit;
        if ((code = tEncodeI64(&encoder, reserved64))) goto _exit;
      }
    }
  }

  tEndEncode(&encoder);

  if (encodeSize) {
    encodeSize[0] = encoder.pos;
  }

_exit:
  tEncoderClear(&encoder);
  return code;
}

int32_t tDeserializeTsdbFSetPartList(void* buf, int32_t bufLen, STsdbFSetPartList* pList) {
  SDecoder decoder = {0};
  int8_t   reserved8 = 0;
  int16_t  reserved16 = 0;
  int64_t  reserved64 = 0;
  int32_t  code = 0;

  STsdbFSetPartition* p = NULL;

  tDecoderInit(&decoder, buf, bufLen);
  int8_t  msgVer = 0;
  int32_t len = 0;
  if ((code = tStartDecode(&decoder))) goto _err;
  if ((code = tDecodeI8(&decoder, &msgVer))) goto _err;
  if (msgVer != TSDB_SNAP_MSG_VER) {
    code = TSDB_CODE_INVALID_MSG;
    goto _err;
  }
  if ((code = tDecodeI32(&decoder, &len))) goto _err;

  for (int32_t u = 0; u < len; u++) {
    p = tsdbFSetPartitionCreate();
    if (p == NULL) {
      code = terrno;
      goto _err;
    }

    if ((code = tDecodeI64(&decoder, &p->fid))) goto _err;
    if ((code = tDecodeI8(&decoder, &p->stat))) goto _err;
    if ((code = tDecodeI8(&decoder, &reserved8))) goto _err;
    if ((code = tDecodeI16(&decoder, &reserved16))) goto _err;

    int32_t typMax = 0;
    if ((code = tDecodeI32(&decoder, &typMax))) goto _err;

    for (int32_t i = 0; i < typMax; i++) {
      SVerRangeList* iList = &p->verRanges[i];
      int32_t        iLen = 0;
      if ((code = tDecodeI32(&decoder, &iLen))) goto _err;
      for (int32_t j = 0; j < iLen; j++) {
        SVersionRange r = {0};
        if ((code = tDecodeI64(&decoder, &r.minVer))) goto _err;
        if ((code = tDecodeI64(&decoder, &r.maxVer))) goto _err;
        if ((code = tDecodeI64(&decoder, &reserved64))) goto _err;
        if ((code = TARRAY2_APPEND(iList, r))) goto _err;
      }
    }
    if ((code = TARRAY2_APPEND(pList, p))) goto _err;
    p = NULL;
  }

  tEndDecode(&decoder);
  tDecoderClear(&decoder);
  return 0;

_err:
  if (p) {
    tsdbFSetPartitionClear(&p);
  }
  tDecoderClear(&decoder);
  return code;
}

// fs state
static STsdbFSetPartList* tsdbSnapGetFSetPartList(STFileSystem* fs) {
  STsdbFSetPartList* pList = tsdbFSetPartListCreate();
  if (pList == NULL) {
    return NULL;
  }

  int32_t code = 0;
  (void)taosThreadMutexLock(&fs->tsdb->mutex);
  STFileSet* fset;
  TARRAY2_FOREACH(fs->fSetArr, fset) {
    STsdbFSetPartition* pItem = NULL;
    code = tsdbTFileSetToFSetPartition(fset, &pItem);
    if (code) {
      terrno = code;
      break;
    }
    code = TARRAY2_SORT_INSERT(pList, pItem, tsdbFSetPartCmprFn);
    if (code) {
      terrno = code;
      break;
    }
  }
  (void)taosThreadMutexUnlock(&fs->tsdb->mutex);

  if (code) {
    TARRAY2_DESTROY(pList, tsdbFSetPartitionClear);
    taosMemoryFree(pList);
    pList = NULL;
  }
  return pList;
}

ETsdbFsState tsdbSnapGetFsState(SVnode* pVnode) { return pVnode->pTsdb->pFS->fsstate; }

// description
typedef struct STsdbPartitionInfo {
  int32_t            vgId;
  int32_t            tsdbMaxCnt;
  int32_t            subTyps[TSDB_RETENTION_MAX];
  STsdbFSetPartList* pLists[TSDB_RETENTION_MAX];
} STsdbPartitionInfo;

static int32_t tsdbPartitionInfoInit(SVnode* pVnode, STsdbPartitionInfo* pInfo) {
  int32_t subTyps[TSDB_RETENTION_MAX] = {SNAP_DATA_TSDB, SNAP_DATA_RSMA1, SNAP_DATA_RSMA2};
  pInfo->vgId = TD_VID(pVnode);
  pInfo->tsdbMaxCnt = 1;

  if (!(sizeof(pInfo->subTyps) == sizeof(subTyps))) {
    return TSDB_CODE_INVALID_PARA;
  }
  memcpy(pInfo->subTyps, (char*)subTyps, sizeof(subTyps));

  // fset partition list
  memset(pInfo->pLists, 0, sizeof(pInfo->pLists[0]) * TSDB_RETENTION_MAX);
  for (int32_t j = 0; j < pInfo->tsdbMaxCnt; ++j) {
    STsdb* pTsdb = SMA_RSMA_GET_TSDB(pVnode, j);
    pInfo->pLists[j] = tsdbSnapGetFSetPartList(pTsdb->pFS);
    if (pInfo->pLists[j] == NULL) {
      return terrno;
    }
  }
  return 0;
}

static void tsdbPartitionInfoClear(STsdbPartitionInfo* pInfo) {
  for (int32_t j = 0; j < pInfo->tsdbMaxCnt; ++j) {
    if (pInfo->pLists[j] == NULL) continue;
    tsdbFSetPartListDestroy(&pInfo->pLists[j]);
  }
}

static int32_t tsdbPartitionInfoEstSize(STsdbPartitionInfo* pInfo) {
  int32_t dataLen = 0;
  for (int32_t j = 0; j < pInfo->tsdbMaxCnt; ++j) {
    dataLen += sizeof(SSyncTLV);  // subTyps[j]
    dataLen += tTsdbFSetPartListDataLenCalc(pInfo->pLists[j]);
  }
  return dataLen;
}

static int32_t tsdbPartitionInfoSerialize(STsdbPartitionInfo* pInfo, uint8_t* buf, int32_t bufLen) {
  int32_t tlen = 0;
  int32_t offset = 0;
  for (int32_t j = 0; j < pInfo->tsdbMaxCnt; ++j) {
    SSyncTLV* pSubHead = (void*)((char*)buf + offset);
    int32_t   valOffset = offset + sizeof(*pSubHead);
    int32_t   code = tSerializeTsdbFSetPartList(pSubHead->val, bufLen - valOffset, pInfo->pLists[j], &tlen);
    if (code) {
      tsdbError("vgId:%d, failed to serialize fset partition list of tsdb %d since %s", pInfo->vgId, j, terrstr());
      return code;
    }
    pSubHead->typ = pInfo->subTyps[j];
    pSubHead->len = tlen;
    offset += sizeof(*pSubHead) + tlen;
  }
  return offset;
}

// tsdb replication opts
static int32_t tTsdbRepOptsDataLenCalc(STsdbRepOpts* pInfo) {
  int32_t hdrLen = sizeof(int32_t);
  int32_t datLen = 0;

  int8_t  msgVer = 0;
  int64_t reserved64 = 0;
  int16_t format = 0;
  hdrLen += sizeof(msgVer);
  datLen += hdrLen;
  datLen += sizeof(format);
  datLen += sizeof(reserved64);
  datLen += sizeof(*pInfo);
  return datLen;
}

int32_t tSerializeTsdbRepOpts(void* buf, int32_t bufLen, STsdbRepOpts* pOpts) {
  int32_t  code = 0;
  SEncoder encoder = {0};
  int64_t  reserved64 = 0;
  int8_t   msgVer = TSDB_SNAP_MSG_VER;

  tEncoderInit(&encoder, buf, bufLen);

  if ((code = tStartEncode(&encoder))) goto _err;
  if ((code = tEncodeI8(&encoder, msgVer))) goto _err;
  int16_t format = pOpts->format;
  if ((code = tEncodeI16(&encoder, format))) goto _err;
  if ((code = tEncodeI64(&encoder, reserved64))) goto _err;

  tEndEncode(&encoder);
  int32_t tlen = encoder.pos;
  tEncoderClear(&encoder);
  return tlen;

_err:
  tEncoderClear(&encoder);
  return code;
}

int32_t tDeserializeTsdbRepOpts(void* buf, int32_t bufLen, STsdbRepOpts* pOpts) {
  int32_t  code;
  SDecoder decoder = {0};
  int64_t  reserved64 = 0;
  int8_t   msgVer = 0;

  tDecoderInit(&decoder, buf, bufLen);

  if ((code = tStartDecode(&decoder))) goto _err;
  if ((code = tDecodeI8(&decoder, &msgVer))) goto _err;
  if (msgVer != TSDB_SNAP_MSG_VER) goto _err;
  int16_t format = 0;
  if ((code = tDecodeI16(&decoder, &format))) goto _err;
  pOpts->format = format;
  if ((code = tDecodeI64(&decoder, &reserved64))) goto _err;

  tEndDecode(&decoder);
  tDecoderClear(&decoder);
  return 0;

_err:
  tDecoderClear(&decoder);
  return code;
}

static int32_t tsdbRepOptsEstSize(STsdbRepOpts* pOpts) {
  int32_t dataLen = 0;
  dataLen += sizeof(SSyncTLV);
  dataLen += tTsdbRepOptsDataLenCalc(pOpts);
  return dataLen;
}

static int32_t tsdbRepOptsSerialize(STsdbRepOpts* pOpts, void* buf, int32_t bufLen) {
  SSyncTLV* pSubHead = buf;
  int32_t   offset = 0;
  int32_t   tlen = 0;
  if ((tlen = tSerializeTsdbRepOpts(pSubHead->val, bufLen, pOpts)) < 0) {
    return tlen;
  }
  pSubHead->typ = SNAP_DATA_RAW;
  pSubHead->len = tlen;
  offset += sizeof(*pSubHead) + tlen;
  return offset;
}

// snap info
static int32_t tsdbSnapPrepDealWithSnapInfo(SVnode* pVnode, SSnapshot* pSnap, STsdbRepOpts* pInfo) {
  if (!pSnap->data) {
    return 0;
  }
  int32_t code = 0;

  SSyncTLV* pHead = (void*)pSnap->data;
  int32_t   offset = 0;

  while (offset + sizeof(*pHead) < pHead->len) {
    SSyncTLV* pField = (void*)(pHead->val + offset);
    offset += sizeof(*pField) + pField->len;
    void*   buf = pField->val;
    int32_t bufLen = pField->len;

    switch (pField->typ) {
      case SNAP_DATA_TSDB:
      case SNAP_DATA_RSMA1:
      case SNAP_DATA_RSMA2: {
      } break;
      case SNAP_DATA_RAW: {
        code = tDeserializeTsdbRepOpts(buf, bufLen, pInfo);
        if (code < 0) {
          tsdbError("vgId:%d, failed to deserialize tsdb rep opts since %s", TD_VID(pVnode), terrstr());
          return code;
        }
      } break;
      default:
        code = TSDB_CODE_INVALID_MSG;
        tsdbError("vgId:%d, unexpected subfield type of snap info. typ:%d", TD_VID(pVnode), pField->typ);
        return code;
    }
  }

  return code;
}

int32_t tsdbSnapPrepDescription(SVnode* pVnode, SSnapshot* pSnap) {
  STsdbPartitionInfo  partitionInfo = {0};
  int                 code = 0;
  STsdbPartitionInfo* pInfo = &partitionInfo;

  code = tsdbPartitionInfoInit(pVnode, pInfo);
  if (code) {
    goto _out;
  }

  // deal with snap info for reply
  STsdbRepOpts opts = {.format = TSDB_SNAP_REP_FMT_RAW};
  if (pSnap->type == TDMT_SYNC_PREP_SNAPSHOT_REPLY) {
    STsdbRepOpts leaderOpts = {0};
    if ((code = tsdbSnapPrepDealWithSnapInfo(pVnode, pSnap, &leaderOpts)) < 0) {
      tsdbError("vgId:%d, failed to deal with snap info for reply since %s", TD_VID(pVnode), terrstr());
      goto _out;
    }
    opts.format = TMIN(opts.format, leaderOpts.format);
  }

  // info data realloc
  const int32_t headLen = sizeof(SSyncTLV);
  int32_t       bufLen = headLen;
  bufLen += tsdbPartitionInfoEstSize(pInfo);
  bufLen += tsdbRepOptsEstSize(&opts);
  if ((code = syncSnapInfoDataRealloc(pSnap, bufLen)) != 0) {
    tsdbError("vgId:%d, failed to realloc memory for data of snap info. bytes:%d", TD_VID(pVnode), bufLen);
    goto _out;
  }

  // serialization
  char*   buf = (void*)pSnap->data;
  int32_t offset = headLen;
  int32_t tlen = 0;

  if ((tlen = tsdbPartitionInfoSerialize(pInfo, (uint8_t*)(buf + offset), bufLen - offset)) < 0) {
    code = tlen;
    tsdbError("vgId:%d, failed to serialize tsdb partition info since %s", TD_VID(pVnode), terrstr());
    goto _out;
  }
  offset += tlen;

  if ((tlen = tsdbRepOptsSerialize(&opts, buf + offset, bufLen - offset)) < 0) {
    code = tlen;
    tsdbError("vgId:%d, failed to serialize tsdb rep opts since %s", TD_VID(pVnode), terrstr());
    goto _out;
  }
  offset += tlen;

  // set header of info data
  SSyncTLV* pHead = pSnap->data;
  pHead->typ = pSnap->type;
  pHead->len = offset - headLen;

  tsdbInfo("vgId:%d, tsdb snap info prepared. type:%s, val length:%d", TD_VID(pVnode), TMSG_INFO(pHead->typ),
           pHead->len);

_out:
  tsdbPartitionInfoClear(pInfo);
  return code;
}

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 "tsdb.h"
17	#include "tsdbFS2.h"
18
19	#define TSDB_SNAP_MSG_VER 1
20
21	// fset partition
22	static int32_t tsdbFSetPartCmprFn(STsdbFSetPartition* x, STsdbFSetPartition* y) {	183✔
23	if (x->fid < y->fid) return -1;	183✔
24	if (x->fid > y->fid) return 1;	173!
25	return 0;	×
26	}
27
28	static int32_t tVersionRangeCmprFn(SVersionRange* x, SVersionRange* y) {	65✔
29	if (x->minVer < y->minVer) return -1;	65✔
30	if (x->minVer > y->minVer) return 1;	36!
31	if (x->maxVer < y->maxVer) return -1;	×
32	if (x->maxVer > y->maxVer) return 1;	×
33	return 0;	×
34	}
35
36	static int32_t tsdbTFileSetRangeCmprFn(STFileSetRange* x, STFileSetRange* y) {	×
37	if (x->fid < y->fid) return -1;	×
38	if (x->fid > y->fid) return 1;	×
39	return 0;	×
40	}
41
42	STsdbFSetPartition* tsdbFSetPartitionCreate() {	153✔
43	STsdbFSetPartition* pSP = taosMemoryCalloc(1, sizeof(STsdbFSetPartition));	153!
44	if (pSP == NULL) {	153!
45	return NULL;	×
46	}
47	for (int32_t i = 0; i < TSDB_FSET_RANGE_TYP_MAX; i++) {	918✔
48	TARRAY2_INIT(&pSP->verRanges[i]);	765✔
49	}
50	return pSP;	153✔
51	}
52
53	void tsdbFSetPartitionClear(STsdbFSetPartition** ppSP) {	153✔
54	if (ppSP == NULL \|\| ppSP[0] == NULL) {	153!
55	return;	×
56	}
57	for (int32_t i = 0; i < TSDB_FSET_RANGE_TYP_MAX; i++) {	918✔
58	TARRAY2_DESTROY(&ppSP[0]->verRanges[i], NULL);	765!
59	}
60	taosMemoryFree(ppSP[0]);	153!
61	ppSP[0] = NULL;	153✔
62	}
63
64	static int32_t tsdbFTypeToFRangeType(tsdb_ftype_t ftype) {	240✔
65	switch (ftype) {	240!
66	case TSDB_FTYPE_HEAD:	80✔
67	return TSDB_FSET_RANGE_TYP_HEAD;	80✔
68	case TSDB_FTYPE_DATA:	80✔
69	return TSDB_FSET_RANGE_TYP_DATA;	80✔
70	case TSDB_FTYPE_SMA:	80✔
71	return TSDB_FSET_RANGE_TYP_SMA;	80✔
72	case TSDB_FTYPE_TOMB:	×
73	return TSDB_FSET_RANGE_TYP_TOMB;	×
74	case TSDB_FTYPE_STT:	×
75	return TSDB_FSET_RANGE_TYP_STT;	×
76	}
77	return TSDB_FSET_RANGE_TYP_MAX;	×
78	}
79
80	static int32_t tsdbTFileSetToFSetPartition(STFileSet* fset, STsdbFSetPartition** ppSP) {	147✔
81	STsdbFSetPartition* p = tsdbFSetPartitionCreate();	147✔
82	if (p == NULL) {	147!
83	return terrno;	×
84	}
85
86	p->fid = fset->fid;	147✔
87
88	int32_t code = 0;	147✔
89	int32_t typ = 0;	147✔
90	int32_t corrupt = false;	147✔
91	int32_t count = 0;	147✔
92	for (int32_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) {	735✔
93	if (fset->farr[ftype] == NULL) continue;	588✔
94	typ = tsdbFTypeToFRangeType(ftype);	240✔
95	STFile* f = fset->farr[ftype]->f;	240✔
96	if (f->maxVer > fset->maxVerValid) {	240!
97	corrupt = true;	×
98	tsdbError("skip incomplete data file: fid:%d, maxVerValid:%" PRId64 ", minVer:%" PRId64 ", maxVer:%" PRId64	×
99	", ftype: %d",
100	fset->fid, fset->maxVerValid, f->minVer, f->maxVer, ftype);
101	continue;	×
102	}
103	count++;	240✔
104	SVersionRange vr = {.minVer = f->minVer, .maxVer = f->maxVer};	240✔
105	code = TARRAY2_SORT_INSERT(&p->verRanges[typ], vr, tVersionRangeCmprFn);	240✔
106	if (code) {	240!
107	tsdbFSetPartitionClear(&p);	×
108	return code;	×
109	}
110	}
111
112	typ = TSDB_FSET_RANGE_TYP_STT;	147✔
113	const SSttLvl* lvl;
114	TARRAY2_FOREACH(fset->lvlArr, lvl) {	284✔
115	STFileObj* fobj;
116	TARRAY2_FOREACH(lvl->fobjArr, fobj) {	310✔
117	STFile* f = fobj->f;	173✔
118	if (f->maxVer > fset->maxVerValid) {	173!
119	corrupt = true;	×
120	tsdbError("skip incomplete stt file.fid:%d, maxVerValid:%" PRId64 ", minVer:%" PRId64 ", maxVer:%" PRId64	×
121	", ftype: %d",
122	fset->fid, fset->maxVerValid, f->minVer, f->maxVer, typ);
123	continue;	×
124	}
125	count++;	173✔
126	SVersionRange vr = {.minVer = f->minVer, .maxVer = f->maxVer};	173✔
127	code = TARRAY2_SORT_INSERT(&p->verRanges[typ], vr, tVersionRangeCmprFn);	173✔
128	if (code) {	173!
129	tsdbFSetPartitionClear(&p);	×
130	return code;	×
131	}
132	}
133	}
134	if (corrupt && count == 0) {	147!
135	SVersionRange vr = {.minVer = VERSION_MIN, .maxVer = fset->maxVerValid};	×
136	code = TARRAY2_SORT_INSERT(&p->verRanges[typ], vr, tVersionRangeCmprFn);	×
137	if (code) {	×
138	tsdbFSetPartitionClear(&p);	×
139	return code;	×
140	}
141	}
142	ppSP[0] = p;	147✔
143	return 0;	147✔
144	}
145
146	// fset partition list
147	STsdbFSetPartList* tsdbFSetPartListCreate() {	366✔
148	STsdbFSetPartList* pList = taosMemoryCalloc(1, sizeof(STsdbFSetPartList));	366!
149	if (pList == NULL) {	366!
150	return NULL;	×
151	}
152	TARRAY2_INIT(pList);	366✔
153	return pList;	366✔
154	}
155
156	void tsdbFSetPartListDestroy(STsdbFSetPartList** ppList) {	366✔
157	if (ppList == NULL \|\| ppList[0] == NULL) return;	366!
158
159	TARRAY2_DESTROY(ppList[0], tsdbFSetPartitionClear);	519!
160	taosMemoryFree(ppList[0]);	366!
161	ppList[0] = NULL;	366✔
162	}
163
164	int32_t tsdbFSetPartListToRangeDiff(STsdbFSetPartList* pList, TFileSetRangeArray** ppRanges) {	183✔
165	int32_t code = 0;	183✔
166
167	TFileSetRangeArray* pDiff = taosMemoryCalloc(1, sizeof(TFileSetRangeArray));	183!
168	if (pDiff == NULL) {	183!
169	code = terrno;	×
170	goto _err;	×
171	}
172	TARRAY2_INIT(pDiff);	183✔
173
174	STsdbFSetPartition* part;
175	TARRAY2_FOREACH(pList, part) {	189✔
176	STFileSetRange* r = taosMemoryCalloc(1, sizeof(STFileSetRange));	6!
177	if (r == NULL) {	6!
178	code = terrno;	×
179	goto _err;	×
180	}
181	int64_t maxVerValid = -1;	6✔
182	int32_t typMax = TSDB_FSET_RANGE_TYP_MAX;	6✔
183	for (int32_t i = 0; i < typMax; i++) {	36✔
184	SVerRangeList* iList = &part->verRanges[i];	30✔
185	SVersionRange vr = {0};	30✔
186	TARRAY2_FOREACH(iList, vr) {	36✔
187	if (vr.maxVer < vr.minVer) {	6!
188	continue;	×
189	}
190	maxVerValid = TMAX(maxVerValid, vr.maxVer);	6✔
191	}
192	}
193	r->fid = part->fid;	6✔
194	r->sver = maxVerValid + 1;	6✔
195	r->ever = VERSION_MAX;	6✔
196	tsdbDebug("range diff fid:%" PRId64 ", sver:%" PRId64 ", ever:%" PRId64, part->fid, r->sver, r->ever);	6!
197	code = TARRAY2_SORT_INSERT(pDiff, r, tsdbTFileSetRangeCmprFn);	6✔
198	if (code) {	6!
199	taosMemoryFree(r);	×
200	goto _err;	×
201	}
202	}
203	ppRanges[0] = pDiff;	183✔
204
205	tsdbInfo("pDiff size:%d", TARRAY2_SIZE(pDiff));	183!
206	return 0;	183✔
207
208	_err:	×
209	if (pDiff) {	×
210	tsdbTFileSetRangeArrayDestroy(&pDiff);	×
211	}
212	return code;	×
213	}
214
215	// serialization
216	int32_t tTsdbFSetPartListDataLenCalc(STsdbFSetPartList* pList) {	183✔
217	int32_t hdrLen = sizeof(int32_t);	183✔
218	int32_t datLen = 0;	183✔
219
220	int8_t msgVer = 1;	183✔
221	int32_t len = TARRAY2_SIZE(pList);	183✔
222	hdrLen += sizeof(msgVer);	183✔
223	hdrLen += sizeof(len);	183✔
224	datLen += hdrLen;	183✔
225
226	for (int32_t u = 0; u < len; u++) {	330✔
227	STsdbFSetPartition* p = TARRAY2_GET(pList, u);	147✔
228	int32_t typMax = TSDB_FSET_RANGE_TYP_MAX;	147✔
229	int32_t uItem = 0;	147✔
230	uItem += sizeof(STsdbFSetPartition);	147✔
231	uItem += sizeof(typMax);	147✔
232
233	for (int32_t i = 0; i < typMax; i++) {	882✔
234	int32_t iLen = TARRAY2_SIZE(&p->verRanges[i]);	735✔
235	int32_t jItem = 0;	735✔
236	jItem += sizeof(SVersionRange);	735✔
237	jItem += sizeof(int64_t);	735✔
238	uItem += sizeof(iLen) + jItem * iLen;	735✔
239	}
240	datLen += uItem;	147✔
241	}
242	return datLen;	183✔
243	}
244
245	static int32_t tSerializeTsdbFSetPartList(void* buf, int32_t bufLen, STsdbFSetPartList* pList, int32_t* encodeSize) {	183✔
246	SEncoder encoder = {0};	183✔
247	int8_t reserved8 = 0;	183✔
248	int16_t reserved16 = 0;	183✔
249	int64_t reserved64 = 0;	183✔
250	int8_t msgVer = TSDB_SNAP_MSG_VER;	183✔
251	int32_t len = TARRAY2_SIZE(pList);	183✔
252	int32_t code = 0;	183✔
253
254	tEncoderInit(&encoder, buf, bufLen);	183✔
255	if ((code = tStartEncode(&encoder))) goto _exit;	183!
256	if ((code = tEncodeI8(&encoder, msgVer))) goto _exit;	366!
257	if ((code = tEncodeI32(&encoder, len))) goto _exit;	183!
258
259	for (int32_t u = 0; u < len; u++) {	330✔
260	STsdbFSetPartition* p = TARRAY2_GET(pList, u);	147✔
261	if ((code = tEncodeI64(&encoder, p->fid))) goto _exit;	294!
262	if ((code = tEncodeI8(&encoder, p->stat))) goto _exit;	294!
263	if ((code = tEncodeI8(&encoder, reserved8))) goto _exit;	294!
264	if ((code = tEncodeI16(&encoder, reserved16))) goto _exit;	294!
265
266	int32_t typMax = TSDB_FSET_RANGE_TYP_MAX;	147!
267	if ((code = tEncodeI32(&encoder, typMax))) goto _exit;	147!
268
269	for (int32_t i = 0; i < typMax; i++) {	882✔
270	SVerRangeList* iList = &p->verRanges[i];	735✔
271	int32_t iLen = TARRAY2_SIZE(iList);	735!
272
273	if ((code = tEncodeI32(&encoder, iLen))) goto _exit;	735!
274	for (int32_t j = 0; j < iLen; j++) {	1,148✔
275	SVersionRange r = TARRAY2_GET(iList, j);	413✔
276	if ((code = tEncodeI64(&encoder, r.minVer))) goto _exit;	826!
277	if ((code = tEncodeI64(&encoder, r.maxVer))) goto _exit;	826!
278	if ((code = tEncodeI64(&encoder, reserved64))) goto _exit;	413!
279	}
280	}
281	}
282
283	tEndEncode(&encoder);	183✔
284
285	if (encodeSize) {	183!
286	encodeSize[0] = encoder.pos;	183✔
287	}
288
289	_exit:	×
290	tEncoderClear(&encoder);	183✔
291	return code;	183✔
292	}
293
294	int32_t tDeserializeTsdbFSetPartList(void* buf, int32_t bufLen, STsdbFSetPartList* pList) {	183✔
295	SDecoder decoder = {0};	183✔
296	int8_t reserved8 = 0;	183✔
297	int16_t reserved16 = 0;	183✔
298	int64_t reserved64 = 0;	183✔
299	int32_t code = 0;	183✔
300
301	STsdbFSetPartition* p = NULL;	183✔
302
303	tDecoderInit(&decoder, buf, bufLen);	183✔
304	int8_t msgVer = 0;	183✔
305	int32_t len = 0;	183✔
306	if ((code = tStartDecode(&decoder))) goto _err;	183!
307	if ((code = tDecodeI8(&decoder, &msgVer))) goto _err;	183!
308	if (msgVer != TSDB_SNAP_MSG_VER) {	183!
309	code = TSDB_CODE_INVALID_MSG;	×
310	goto _err;	×
311	}
312	if ((code = tDecodeI32(&decoder, &len))) goto _err;	183!
313
314	for (int32_t u = 0; u < len; u++) {	189✔
315	p = tsdbFSetPartitionCreate();	6✔
316	if (p == NULL) {	6!
317	code = terrno;	×
318	goto _err;	×
319	}
320
321	if ((code = tDecodeI64(&decoder, &p->fid))) goto _err;	12!
322	if ((code = tDecodeI8(&decoder, &p->stat))) goto _err;	12!
323	if ((code = tDecodeI8(&decoder, &reserved8))) goto _err;	6!
324	if ((code = tDecodeI16(&decoder, &reserved16))) goto _err;	6!
325
326	int32_t typMax = 0;	6✔
327	if ((code = tDecodeI32(&decoder, &typMax))) goto _err;	6!
328
329	for (int32_t i = 0; i < typMax; i++) {	36✔
330	SVerRangeList* iList = &p->verRanges[i];	30✔
331	int32_t iLen = 0;	30✔
332	if ((code = tDecodeI32(&decoder, &iLen))) goto _err;	30!
333	for (int32_t j = 0; j < iLen; j++) {	36✔
334	SVersionRange r = {0};	6✔
335	if ((code = tDecodeI64(&decoder, &r.minVer))) goto _err;	6!
336	if ((code = tDecodeI64(&decoder, &r.maxVer))) goto _err;	6!
337	if ((code = tDecodeI64(&decoder, &reserved64))) goto _err;	6!
338	if ((code = TARRAY2_APPEND(iList, r))) goto _err;	12!
339	}
340	}
341	if ((code = TARRAY2_APPEND(pList, p))) goto _err;	12!
342	p = NULL;	6✔
343	}
344
345	tEndDecode(&decoder);	183✔
346	tDecoderClear(&decoder);	183✔
347	return 0;	183✔
348
349	_err:	×
350	if (p) {	×
351	tsdbFSetPartitionClear(&p);	×
352	}
353	tDecoderClear(&decoder);	×
354	return code;	×
355	}
356
357	// fs state
358	static STsdbFSetPartList* tsdbSnapGetFSetPartList(STFileSystem* fs) {	183✔
359	STsdbFSetPartList* pList = tsdbFSetPartListCreate();	183✔
360	if (pList == NULL) {	183!
361	return NULL;	×
362	}
363
364	int32_t code = 0;	183✔
365	(void)taosThreadMutexLock(&fs->tsdb->mutex);	183✔
366	STFileSet* fset;
367	TARRAY2_FOREACH(fs->fSetArr, fset) {	330✔
368	STsdbFSetPartition* pItem = NULL;	147✔
369	code = tsdbTFileSetToFSetPartition(fset, &pItem);	147✔
370	if (code) {	147!
371	terrno = code;	×
372	break;	×
373	}
374	code = TARRAY2_SORT_INSERT(pList, pItem, tsdbFSetPartCmprFn);	147✔
375	if (code) {	147!
376	terrno = code;	×
377	break;	×
378	}
379	}
380	(void)taosThreadMutexUnlock(&fs->tsdb->mutex);	183✔
381
382	if (code) {	183!
383	TARRAY2_DESTROY(pList, tsdbFSetPartitionClear);	×
384	taosMemoryFree(pList);	×
385	pList = NULL;	×
386	}
387	return pList;	183✔
388	}
389
390	ETsdbFsState tsdbSnapGetFsState(SVnode* pVnode) { return pVnode->pTsdb->pFS->fsstate; }	245,974✔
391
392	// description
393	typedef struct STsdbPartitionInfo {
394	int32_t vgId;
395	int32_t tsdbMaxCnt;
396	int32_t subTyps[TSDB_RETENTION_MAX];
397	STsdbFSetPartList* pLists[TSDB_RETENTION_MAX];
398	} STsdbPartitionInfo;
399
400	static int32_t tsdbPartitionInfoInit(SVnode* pVnode, STsdbPartitionInfo* pInfo) {	183✔
401	int32_t subTyps[TSDB_RETENTION_MAX] = {SNAP_DATA_TSDB, SNAP_DATA_RSMA1, SNAP_DATA_RSMA2};	183✔
402	pInfo->vgId = TD_VID(pVnode);	183✔
403	pInfo->tsdbMaxCnt = 1;	183✔
404
405	if (!(sizeof(pInfo->subTyps) == sizeof(subTyps))) {
406	return TSDB_CODE_INVALID_PARA;
407	}
408	memcpy(pInfo->subTyps, (char*)subTyps, sizeof(subTyps));	183✔
409
410	// fset partition list
411	memset(pInfo->pLists, 0, sizeof(pInfo->pLists[0]) * TSDB_RETENTION_MAX);	183✔
412	for (int32_t j = 0; j < pInfo->tsdbMaxCnt; ++j) {	366✔
413	STsdb* pTsdb = SMA_RSMA_GET_TSDB(pVnode, j);	183!
414	pInfo->pLists[j] = tsdbSnapGetFSetPartList(pTsdb->pFS);	183✔
415	if (pInfo->pLists[j] == NULL) {	183!
416	return terrno;	×
417	}
418	}
419	return 0;	183✔
420	}
421
422	static void tsdbPartitionInfoClear(STsdbPartitionInfo* pInfo) {	183✔
423	for (int32_t j = 0; j < pInfo->tsdbMaxCnt; ++j) {	366✔
424	if (pInfo->pLists[j] == NULL) continue;	183!
425	tsdbFSetPartListDestroy(&pInfo->pLists[j]);	183✔
426	}
427	}	183✔
428
429	static int32_t tsdbPartitionInfoEstSize(STsdbPartitionInfo* pInfo) {	183✔
430	int32_t dataLen = 0;	183✔
431	for (int32_t j = 0; j < pInfo->tsdbMaxCnt; ++j) {	366✔
432	dataLen += sizeof(SSyncTLV); // subTyps[j]	183✔
433	dataLen += tTsdbFSetPartListDataLenCalc(pInfo->pLists[j]);	183✔
434	}
435	return dataLen;	183✔
436	}
437
438	static int32_t tsdbPartitionInfoSerialize(STsdbPartitionInfo* pInfo, uint8_t* buf, int32_t bufLen) {	183✔
439	int32_t tlen = 0;	183✔
440	int32_t offset = 0;	183✔
441	for (int32_t j = 0; j < pInfo->tsdbMaxCnt; ++j) {	366✔
442	SSyncTLV* pSubHead = (void)((char)buf + offset);	183✔
443	int32_t valOffset = offset + sizeof(*pSubHead);	183✔
444	int32_t code = tSerializeTsdbFSetPartList(pSubHead->val, bufLen - valOffset, pInfo->pLists[j], &tlen);	183✔
445	if (code) {	183!
446	tsdbError("vgId:%d, failed to serialize fset partition list of tsdb %d since %s", pInfo->vgId, j, terrstr());	×
447	return code;	×
448	}
449	pSubHead->typ = pInfo->subTyps[j];	183✔
450	pSubHead->len = tlen;	183✔
451	offset += sizeof(*pSubHead) + tlen;	183✔
452	}
453	return offset;	183✔
454	}
455
456	// tsdb replication opts
457	static int32_t tTsdbRepOptsDataLenCalc(STsdbRepOpts* pInfo) {	183✔
458	int32_t hdrLen = sizeof(int32_t);	183✔
459	int32_t datLen = 0;	183✔
460
461	int8_t msgVer = 0;	183✔
462	int64_t reserved64 = 0;	183✔
463	int16_t format = 0;	183✔
464	hdrLen += sizeof(msgVer);	183✔
465	datLen += hdrLen;	183✔
466	datLen += sizeof(format);	183✔
467	datLen += sizeof(reserved64);	183✔
468	datLen += sizeof(*pInfo);	183✔
469	return datLen;	183✔
470	}
471
472	int32_t tSerializeTsdbRepOpts(void* buf, int32_t bufLen, STsdbRepOpts* pOpts) {	183✔
473	int32_t code = 0;	183✔
474	SEncoder encoder = {0};	183✔
475	int64_t reserved64 = 0;	183✔
476	int8_t msgVer = TSDB_SNAP_MSG_VER;	183✔
477
478	tEncoderInit(&encoder, buf, bufLen);	183✔
479
480	if ((code = tStartEncode(&encoder))) goto _err;	183!
481	if ((code = tEncodeI8(&encoder, msgVer))) goto _err;	366!
482	int16_t format = pOpts->format;	183✔
483	if ((code = tEncodeI16(&encoder, format))) goto _err;	366!
484	if ((code = tEncodeI64(&encoder, reserved64))) goto _err;	183!
485
486	tEndEncode(&encoder);	183✔
487	int32_t tlen = encoder.pos;	183✔
488	tEncoderClear(&encoder);	183✔
489	return tlen;	183✔
490
491	_err:	×
492	tEncoderClear(&encoder);	×
493	return code;	×
494	}
495
496	int32_t tDeserializeTsdbRepOpts(void* buf, int32_t bufLen, STsdbRepOpts* pOpts) {	274✔
497	int32_t code;
498	SDecoder decoder = {0};	274✔
499	int64_t reserved64 = 0;	274✔
500	int8_t msgVer = 0;	274✔
501
502	tDecoderInit(&decoder, buf, bufLen);	274✔
503
504	if ((code = tStartDecode(&decoder))) goto _err;	274!
505	if ((code = tDecodeI8(&decoder, &msgVer))) goto _err;	274!
506	if (msgVer != TSDB_SNAP_MSG_VER) goto _err;	274!
507	int16_t format = 0;	274✔
508	if ((code = tDecodeI16(&decoder, &format))) goto _err;	274!
509	pOpts->format = format;	274✔
510	if ((code = tDecodeI64(&decoder, &reserved64))) goto _err;	274!
511
512	tEndDecode(&decoder);	274✔
513	tDecoderClear(&decoder);	274✔
514	return 0;	274✔
515
516	_err:	×
517	tDecoderClear(&decoder);	×
518	return code;	×
519	}
520
521	static int32_t tsdbRepOptsEstSize(STsdbRepOpts* pOpts) {	183✔
522	int32_t dataLen = 0;	183✔
523	dataLen += sizeof(SSyncTLV);	183✔
524	dataLen += tTsdbRepOptsDataLenCalc(pOpts);	183✔
525	return dataLen;	183✔
526	}
527
528	static int32_t tsdbRepOptsSerialize(STsdbRepOpts* pOpts, void* buf, int32_t bufLen) {	183✔
529	SSyncTLV* pSubHead = buf;	183✔
530	int32_t offset = 0;	183✔
531	int32_t tlen = 0;	183✔
532	if ((tlen = tSerializeTsdbRepOpts(pSubHead->val, bufLen, pOpts)) < 0) {	183!
533	return tlen;	×
534	}
535	pSubHead->typ = SNAP_DATA_RAW;	183✔
536	pSubHead->len = tlen;	183✔
537	offset += sizeof(*pSubHead) + tlen;	183✔
538	return offset;	183✔
539	}
540
541	// snap info
542	static int32_t tsdbSnapPrepDealWithSnapInfo(SVnode* pVnode, SSnapshot* pSnap, STsdbRepOpts* pInfo) {	91✔
543	if (!pSnap->data) {	91!
544	return 0;	×
545	}
546	int32_t code = 0;	91✔
547
548	SSyncTLV* pHead = (void*)pSnap->data;	91✔
549	int32_t offset = 0;	91✔
550
551	while (offset + sizeof(*pHead) < pHead->len) {	273✔
552	SSyncTLV* pField = (void*)(pHead->val + offset);	182✔
553	offset += sizeof(*pField) + pField->len;	182✔
554	void* buf = pField->val;	182✔
555	int32_t bufLen = pField->len;	182✔
556
557	switch (pField->typ) {	182!
558	case SNAP_DATA_TSDB:	91✔
559	case SNAP_DATA_RSMA1:
560	case SNAP_DATA_RSMA2: {
561	} break;	91✔
562	case SNAP_DATA_RAW: {	91✔
563	code = tDeserializeTsdbRepOpts(buf, bufLen, pInfo);	91✔
564	if (code < 0) {	91!
565	tsdbError("vgId:%d, failed to deserialize tsdb rep opts since %s", TD_VID(pVnode), terrstr());	×
566	return code;	×
567	}
568	} break;	91✔
569	default:	×
570	code = TSDB_CODE_INVALID_MSG;	×
571	tsdbError("vgId:%d, unexpected subfield type of snap info. typ:%d", TD_VID(pVnode), pField->typ);	×
572	return code;	×
573	}
574	}
575
576	return code;	91✔
577	}
578
579	int32_t tsdbSnapPrepDescription(SVnode* pVnode, SSnapshot* pSnap) {	183✔
580	STsdbPartitionInfo partitionInfo = {0};	183✔
581	int code = 0;	183✔
582	STsdbPartitionInfo* pInfo = &partitionInfo;	183✔
583
584	code = tsdbPartitionInfoInit(pVnode, pInfo);	183✔
585	if (code) {	183!
586	goto _out;	×
587	}
588
589	// deal with snap info for reply
590	STsdbRepOpts opts = {.format = TSDB_SNAP_REP_FMT_RAW};	183✔
591	if (pSnap->type == TDMT_SYNC_PREP_SNAPSHOT_REPLY) {	183✔
592	STsdbRepOpts leaderOpts = {0};	91✔
593	if ((code = tsdbSnapPrepDealWithSnapInfo(pVnode, pSnap, &leaderOpts)) < 0) {	91!
594	tsdbError("vgId:%d, failed to deal with snap info for reply since %s", TD_VID(pVnode), terrstr());	×
595	goto _out;	×
596	}
597	opts.format = TMIN(opts.format, leaderOpts.format);	91✔
598	}
599
600	// info data realloc
601	const int32_t headLen = sizeof(SSyncTLV);	183✔
602	int32_t bufLen = headLen;	183✔
603	bufLen += tsdbPartitionInfoEstSize(pInfo);	183✔
604	bufLen += tsdbRepOptsEstSize(&opts);	183✔
605	if ((code = syncSnapInfoDataRealloc(pSnap, bufLen)) != 0) {	183!
606	tsdbError("vgId:%d, failed to realloc memory for data of snap info. bytes:%d", TD_VID(pVnode), bufLen);	×
607	goto _out;	×
608	}
609
610	// serialization
611	char* buf = (void*)pSnap->data;	183✔
612	int32_t offset = headLen;	183✔
613	int32_t tlen = 0;	183✔
614
615	if ((tlen = tsdbPartitionInfoSerialize(pInfo, (uint8_t*)(buf + offset), bufLen - offset)) < 0) {	183!
616	code = tlen;	×
617	tsdbError("vgId:%d, failed to serialize tsdb partition info since %s", TD_VID(pVnode), terrstr());	×
618	goto _out;	×
619	}
620	offset += tlen;	183✔
621
622	if ((tlen = tsdbRepOptsSerialize(&opts, buf + offset, bufLen - offset)) < 0) {	183!
623	code = tlen;	×
624	tsdbError("vgId:%d, failed to serialize tsdb rep opts since %s", TD_VID(pVnode), terrstr());	×
625	goto _out;	×
626	}
627	offset += tlen;	183✔
628
629	// set header of info data
630	SSyncTLV* pHead = pSnap->data;	183✔
631	pHead->typ = pSnap->type;	183✔
632	pHead->len = offset - headLen;	183✔
633
634	tsdbInfo("vgId:%d, tsdb snap info prepared. type:%s, val length:%d", TD_VID(pVnode), TMSG_INFO(pHead->typ),	183!
635	pHead->len);
636
637	_out:	×
638	tsdbPartitionInfoClear(pInfo);	183✔
639	return code;	183✔
640	}

taosdata / TDengine / #4791

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