#3523

Committed 06 Nov 2024 02:29AM UTC coverage: 55.861% (-2.4%) from 58.216%

Build # #3523

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #28551 from taosdata/feat/TS-5215-2

test(blob): testing & fixes for blob

Run Details

106075 of 245834 branches covered (43.15%)

Branch coverage included in aggregate %.

0 of 15 new or added lines in 2 files covered. (0.0%)

17003 existing lines in 254 files now uncovered.

181910 of 269703 relevant lines covered (67.45%)

1527639.59 hits per line

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

37.27

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

static void tsdbSnapRAWReadFileSetCloseReader(STsdbSnapRAWReader* reader);

// reader
typedef struct SDataFileRAWReaderIter {
  int32_t count;
  int32_t idx;
} SDataFileRAWReaderIter;

typedef struct STsdbSnapRAWReader {
  STsdb*  tsdb;
  int64_t ever;
  int8_t  type;

  TFileSetArray* fsetArr;

  // context
  struct {
    int32_t    fsetArrIdx;
    STFileSet* fset;
    bool       isDataDone;
  } ctx[1];

  // reader
  SDataFileRAWReaderArray dataReaderArr[1];

  // iter
  SDataFileRAWReaderIter dataIter[1];
} STsdbSnapRAWReader;

int32_t tsdbSnapRAWReaderOpen(STsdb* tsdb, int64_t ever, int8_t type, STsdbSnapRAWReader** reader) {
  int32_t code = 0;
  int32_t lino = 0;

  reader[0] = taosMemoryCalloc(1, sizeof(STsdbSnapRAWReader));
  if (reader[0] == NULL) return terrno;

  reader[0]->tsdb = tsdb;
  reader[0]->ever = ever;
  reader[0]->type = type;

  code = tsdbFSCreateRefSnapshot(tsdb->pFS, &reader[0]->fsetArr);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  if (code) {
    tsdbError("vgId:%d %s failed at line %d since %s, sver:0, ever:%" PRId64 " type:%d", TD_VID(tsdb->pVnode), __func__,
              lino, tstrerror(code), ever, type);
    tsdbFSDestroyRefSnapshot(&reader[0]->fsetArr);
    taosMemoryFree(reader[0]);
    reader[0] = NULL;
  } else {
    tsdbInfo("vgId:%d, tsdb snapshot raw reader opened. sver:0, ever:%" PRId64 " type:%d", TD_VID(tsdb->pVnode), ever,
             type);
  }
  return code;
}

void tsdbSnapRAWReaderClose(STsdbSnapRAWReader** reader) {
  if (reader[0] == NULL) return;

  int32_t code = 0;
  int32_t lino = 0;

  STsdb* tsdb = reader[0]->tsdb;

  TARRAY2_DESTROY(reader[0]->dataReaderArr, tsdbDataFileRAWReaderClose);
  tsdbFSDestroyRefSnapshot(&reader[0]->fsetArr);
  taosMemoryFree(reader[0]);
  reader[0] = NULL;
  return;
}

static int32_t tsdbSnapRAWReadFileSetOpenReader(STsdbSnapRAWReader* reader) {
  int32_t code = 0;
  int32_t lino = 0;

  // data
  for (int32_t ftype = 0; ftype < TSDB_FTYPE_MAX; ftype++) {
    if (reader->ctx->fset->farr[ftype] == NULL) {
      continue;
    }
    STFileObj*               fobj = reader->ctx->fset->farr[ftype];
    SDataFileRAWReader*      dataReader;
    SDataFileRAWReaderConfig config = {
        .tsdb = reader->tsdb,
        .szPage = reader->tsdb->pVnode->config.tsdbPageSize,
        .file = fobj->f[0],
    };
    code = tsdbDataFileRAWReaderOpen(NULL, &config, &dataReader);
    TSDB_CHECK_CODE(code, lino, _exit);

    code = TARRAY2_APPEND(reader->dataReaderArr, dataReader);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  // stt
  SSttLvl* lvl;
  TARRAY2_FOREACH(reader->ctx->fset->lvlArr, lvl) {
    STFileObj* fobj;
    TARRAY2_FOREACH(lvl->fobjArr, fobj) {
      SDataFileRAWReader*      dataReader;
      SDataFileRAWReaderConfig config = {
          .tsdb = reader->tsdb,
          .szPage = reader->tsdb->pVnode->config.tsdbPageSize,
          .file = fobj->f[0],
      };
      code = tsdbDataFileRAWReaderOpen(NULL, &config, &dataReader);
      TSDB_CHECK_CODE(code, lino, _exit);

      code = TARRAY2_APPEND(reader->dataReaderArr, dataReader);
      TSDB_CHECK_CODE(code, lino, _exit);
    }
  }

_exit:
  if (code) {
    tsdbSnapRAWReadFileSetCloseReader(reader);
    TSDB_ERROR_LOG(TD_VID(reader->tsdb->pVnode), code, lino);
  }
  return code;
}

static void tsdbSnapRAWReadFileSetCloseReader(STsdbSnapRAWReader* reader) {
  TARRAY2_CLEAR(reader->dataReaderArr, tsdbDataFileRAWReaderClose);
}

static int32_t tsdbSnapRAWReadFileSetOpenIter(STsdbSnapRAWReader* reader) {
  reader->dataIter->count = TARRAY2_SIZE(reader->dataReaderArr);
  reader->dataIter->idx = 0;
  return 0;
}

static void tsdbSnapRAWReadFileSetCloseIter(STsdbSnapRAWReader* reader) {
  reader->dataIter->count = 0;
  reader->dataIter->idx = 0;
}

static int64_t tsdbSnapRAWReadPeek(SDataFileRAWReader* reader) {
  int64_t size = TMIN(reader->config->file.size - reader->ctx->offset, TSDB_SNAP_DATA_PAYLOAD_SIZE);
  return size;
}

static SDataFileRAWReader* tsdbSnapRAWReaderIterNext(STsdbSnapRAWReader* reader) {
  while (reader->dataIter->idx < reader->dataIter->count) {
    SDataFileRAWReader* dataReader = TARRAY2_GET(reader->dataReaderArr, reader->dataIter->idx);
    if (dataReader->ctx->offset < dataReader->config->file.size) {
      return dataReader;
    }
    reader->dataIter->idx++;
  }
  return NULL;
}

static int32_t tsdbSnapRAWReadNext(STsdbSnapRAWReader* reader, SSnapDataHdr** ppData) {
  int32_t code = 0;
  int32_t lino = 0;
  int8_t  type = reader->type;
  ppData[0] = NULL;

  SDataFileRAWReader* dataReader = tsdbSnapRAWReaderIterNext(reader);
  if (dataReader == NULL) {
    return 0;
  }

  // prepare
  int64_t dataLength = tsdbSnapRAWReadPeek(dataReader);

  void* pBuf = taosMemoryCalloc(1, sizeof(SSnapDataHdr) + sizeof(STsdbDataRAWBlockHeader) + dataLength);
  if (pBuf == NULL) {
    code = terrno;
    TSDB_CHECK_CODE(code, lino, _exit);
  }
  SSnapDataHdr* pHdr = pBuf;
  pHdr->type = type;
  pHdr->size = sizeof(STsdbDataRAWBlockHeader) + dataLength;

  // read
  STsdbDataRAWBlockHeader* pBlock = (void*)pHdr->data;
  pBlock->offset = dataReader->ctx->offset;
  pBlock->dataLength = dataLength;

  code = tsdbDataFileRAWReadBlockData(dataReader, pBlock);
  TSDB_CHECK_CODE(code, lino, _exit);

  // finish
  dataReader->ctx->offset += pBlock->dataLength;
  ppData[0] = pBuf;

_exit:
  if (code) {
    taosMemoryFree(pBuf);
    pBuf = NULL;
    TSDB_ERROR_LOG(TD_VID(reader->tsdb->pVnode), code, lino);
  }
  return code;
}

static int32_t tsdbSnapRAWReadData(STsdbSnapRAWReader* reader, uint8_t** ppData) {
  int32_t code = 0;
  int32_t lino = 0;

  code = tsdbSnapRAWReadNext(reader, (SSnapDataHdr**)ppData);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  if (code) {
    TSDB_ERROR_LOG(TD_VID(reader->tsdb->pVnode), code, lino);
  }
  return code;
}

static int32_t tsdbSnapRAWReadBegin(STsdbSnapRAWReader* reader) {
  int32_t code = 0;
  int32_t lino = 0;

  if (reader->ctx->fsetArrIdx < TARRAY2_SIZE(reader->fsetArr)) {
    reader->ctx->fset = TARRAY2_GET(reader->fsetArr, reader->ctx->fsetArrIdx++);
    reader->ctx->isDataDone = false;

    code = tsdbSnapRAWReadFileSetOpenReader(reader);
    TSDB_CHECK_CODE(code, lino, _exit);

    code = tsdbSnapRAWReadFileSetOpenIter(reader);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

_exit:
  if (code) {
    TSDB_ERROR_LOG(TD_VID(reader->tsdb->pVnode), code, lino);
  }
  return code;
}

static int32_t tsdbSnapRAWReadEnd(STsdbSnapRAWReader* reader) {
  tsdbSnapRAWReadFileSetCloseIter(reader);
  tsdbSnapRAWReadFileSetCloseReader(reader);
  reader->ctx->fset = NULL;
  return 0;
}

int32_t tsdbSnapRAWRead(STsdbSnapRAWReader* reader, uint8_t** data) {
  int32_t code = 0;
  int32_t lino = 0;

  data[0] = NULL;

  for (;;) {
    if (reader->ctx->fset == NULL) {
      code = tsdbSnapRAWReadBegin(reader);
      TSDB_CHECK_CODE(code, lino, _exit);

      if (reader->ctx->fset == NULL) {
        break;
      }
    }

    if (!reader->ctx->isDataDone) {
      code = tsdbSnapRAWReadData(reader, data);
      TSDB_CHECK_CODE(code, lino, _exit);
      if (data[0]) {
        goto _exit;
      } else {
        reader->ctx->isDataDone = true;
      }
    }

    code = tsdbSnapRAWReadEnd(reader);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

_exit:
  if (code) {
    TSDB_ERROR_LOG(TD_VID(reader->tsdb->pVnode), code, lino);
  } else {
    tsdbDebug("vgId:%d %s done", TD_VID(reader->tsdb->pVnode), __func__);
  }
  return code;
}

// writer
struct STsdbSnapRAWWriter {
  STsdb*  tsdb;
  int64_t sver;
  int64_t ever;
  int32_t minutes;
  int8_t  precision;
  int32_t minRow;
  int32_t maxRow;
  int8_t  cmprAlg;
  int64_t commitID;
  int32_t szPage;
  int64_t compactVersion;
  int64_t now;

  TFileSetArray* fsetArr;
  TFileOpArray   fopArr[1];

  struct {
    bool       fsetWriteBegin;
    int32_t    fid;
    STFileSet* fset;
    SDiskID    did;
    int64_t    cid;
    int64_t    level;

    // writer
    SFSetRAWWriter* fsetWriter;
  } ctx[1];
};

int32_t tsdbSnapRAWWriterOpen(STsdb* pTsdb, int64_t ever, STsdbSnapRAWWriter** writer) {
  int32_t code = 0;
  int32_t lino = 0;

  // start to write
  writer[0] = taosMemoryCalloc(1, sizeof(*writer[0]));
  if (writer[0] == NULL) return terrno;

  writer[0]->tsdb = pTsdb;
  writer[0]->ever = ever;
  writer[0]->minutes = pTsdb->keepCfg.days;
  writer[0]->precision = pTsdb->keepCfg.precision;
  writer[0]->minRow = pTsdb->pVnode->config.tsdbCfg.minRows;
  writer[0]->maxRow = pTsdb->pVnode->config.tsdbCfg.maxRows;
  writer[0]->cmprAlg = pTsdb->pVnode->config.tsdbCfg.compression;
  writer[0]->commitID = tsdbFSAllocEid(pTsdb->pFS);
  writer[0]->szPage = pTsdb->pVnode->config.tsdbPageSize;
  writer[0]->compactVersion = INT64_MAX;
  writer[0]->now = taosGetTimestampMs();

  code = tsdbFSCreateCopySnapshot(pTsdb->pFS, &writer[0]->fsetArr);
  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));
  } else {
    tsdbInfo("vgId:%d %s done, sver:0, ever:%" PRId64, TD_VID(pTsdb->pVnode), __func__, ever);
  }
  return code;
}

static int32_t tsdbSnapRAWWriteFileSetCloseIter(STsdbSnapRAWWriter* writer) { return 0; }

static int32_t tsdbSnapRAWWriteFileSetOpenWriter(STsdbSnapRAWWriter* writer) {
  int32_t code = 0;
  int32_t lino = 0;

  SFSetRAWWriterConfig config = {
      .tsdb = writer->tsdb,
      .szPage = writer->szPage,
      .fid = writer->ctx->fid,
      .cid = writer->commitID,
      .did = writer->ctx->did,
      .level = writer->ctx->level,
  };

  code = tsdbFSetRAWWriterOpen(&config, &writer->ctx->fsetWriter);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  if (code) {
    TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);
  }
  return code;
}

static int32_t tsdbSnapRAWWriteFileSetCloseWriter(STsdbSnapRAWWriter* writer) {
  return tsdbFSetRAWWriterClose(&writer->ctx->fsetWriter, 0, writer->fopArr);
}

static int32_t tsdbSnapRAWWriteFileSetBegin(STsdbSnapRAWWriter* writer, int32_t fid) {
  int32_t code = 0;
  int32_t lino = 0;

  STFileSet* fset = &(STFileSet){.fid = fid};

  writer->ctx->fid = fid;
  STFileSet** fsetPtr = TARRAY2_SEARCH(writer->fsetArr, &fset, tsdbTFileSetCmprFn, TD_EQ);
  writer->ctx->fset = (fsetPtr == NULL) ? NULL : *fsetPtr;

  int32_t level = tsdbFidLevel(fid, &writer->tsdb->keepCfg, taosGetTimestampSec());
  code = tfsAllocDisk(writer->tsdb->pVnode->pTfs, level, &writer->ctx->did);
  TSDB_CHECK_CODE(code, lino, _exit);

  code = tfsMkdirRecurAt(writer->tsdb->pVnode->pTfs, writer->tsdb->path, writer->ctx->did);
  TSDB_CHECK_CODE(code, lino, _exit);

  code = tsdbSnapRAWWriteFileSetOpenWriter(writer);
  TSDB_CHECK_CODE(code, lino, _exit);

  writer->ctx->level = level;
  writer->ctx->fsetWriteBegin = true;

_exit:
  if (code) {
    TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);
  }
  return code;
}

static int32_t tsdbSnapRAWWriteFileSetEnd(STsdbSnapRAWWriter* writer) {
  if (!writer->ctx->fsetWriteBegin) return 0;

  int32_t code = 0;
  int32_t lino = 0;

  // close write
  code = tsdbSnapRAWWriteFileSetCloseWriter(writer);
  TSDB_CHECK_CODE(code, lino, _exit);

  writer->ctx->fsetWriteBegin = false;

_exit:
  if (code) {
    TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);
  }
  return code;
}

int32_t tsdbSnapRAWWriterPrepareClose(STsdbSnapRAWWriter* writer) {
  int32_t code = 0;
  int32_t lino = 0;

  code = tsdbSnapRAWWriteFileSetEnd(writer);
  TSDB_CHECK_CODE(code, lino, _exit);

  code = tsdbFSEditBegin(writer->tsdb->pFS, writer->fopArr, TSDB_FEDIT_COMMIT);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  if (code) {
    TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);
  } else {
    tsdbDebug("vgId:%d %s done", TD_VID(writer->tsdb->pVnode), __func__);
  }
  return code;
}

int32_t tsdbSnapRAWWriterClose(STsdbSnapRAWWriter** writer, int8_t rollback) {
  if (writer[0] == NULL) return 0;

  int32_t code = 0;
  int32_t lino = 0;

  STsdb* tsdb = writer[0]->tsdb;

  if (rollback) {
    code = tsdbFSEditAbort(writer[0]->tsdb->pFS);
    TSDB_CHECK_CODE(code, lino, _exit);
  } else {
    (void)taosThreadMutexLock(&writer[0]->tsdb->mutex);

    code = tsdbFSEditCommit(writer[0]->tsdb->pFS);
    if (code) {
      (void)taosThreadMutexUnlock(&writer[0]->tsdb->mutex);
      TSDB_CHECK_CODE(code, lino, _exit);
    }

    writer[0]->tsdb->pFS->fsstate = TSDB_FS_STATE_NORMAL;

    (void)taosThreadMutexUnlock(&writer[0]->tsdb->mutex);
  }

  TARRAY2_DESTROY(writer[0]->fopArr, NULL);
  tsdbFSDestroyCopySnapshot(&writer[0]->fsetArr);

  taosMemoryFree(writer[0]);
  writer[0] = NULL;

_exit:
  if (code) {
    TSDB_ERROR_LOG(TD_VID(tsdb->pVnode), lino, code);
  } else {
    tsdbInfo("vgId:%d %s done", TD_VID(tsdb->pVnode), __func__);
  }
  return code;
}

static int32_t tsdbSnapRAWWriteTimeSeriesData(STsdbSnapRAWWriter* writer, STsdbDataRAWBlockHeader* bHdr) {
  int32_t code = 0;
  int32_t lino = 0;

  int32_t encryptAlgorithm = writer->tsdb->pVnode->config.tsdbCfg.encryptAlgorithm;
  char*   encryptKey = writer->tsdb->pVnode->config.tsdbCfg.encryptKey;

  code = tsdbFSetRAWWriteBlockData(writer->ctx->fsetWriter, bHdr, encryptAlgorithm, encryptKey);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  if (code) {
    TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);
  }
  return code;
}

static int32_t tsdbSnapRAWWriteData(STsdbSnapRAWWriter* writer, SSnapDataHdr* hdr) {
  int32_t code = 0;
  int32_t lino = 0;

  STsdbDataRAWBlockHeader* bHdr = (void*)hdr->data;
  int32_t                  fid = bHdr->file.fid;
  if (!writer->ctx->fsetWriteBegin || fid != writer->ctx->fid) {
    code = tsdbSnapRAWWriteFileSetEnd(writer);
    TSDB_CHECK_CODE(code, lino, _exit);

    code = tsdbSnapRAWWriteFileSetBegin(writer, fid);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  code = tsdbSnapRAWWriteTimeSeriesData(writer, bHdr);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  if (code) {
    TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);
  }
  return code;
}

int32_t tsdbSnapRAWWrite(STsdbSnapRAWWriter* writer, SSnapDataHdr* hdr) {
  int32_t code = 0;
  int32_t lino = 0;

  code = tsdbSnapRAWWriteData(writer, hdr);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  if (code) {
    tsdbError("vgId:%d %s failed at line %d since %s, type:%d index:%" PRId64 " size:%" PRId64,
              TD_VID(writer->tsdb->pVnode), __func__, lino, tstrerror(code), hdr->type, hdr->index, hdr->size);
  } else {
    tsdbDebug("vgId:%d %s done, type:%d index:%" PRId64 " size:%" PRId64, TD_VID(writer->tsdb->pVnode), __func__,
              hdr->type, hdr->index, hdr->size);
  }
  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 "tsdbDataFileRAW.h"
18	#include "tsdbFS2.h"
19	#include "tsdbFSetRAW.h"
20
21	static void tsdbSnapRAWReadFileSetCloseReader(STsdbSnapRAWReader* reader);
22
23	// reader
24	typedef struct SDataFileRAWReaderIter {
25	int32_t count;
26	int32_t idx;
27	} SDataFileRAWReaderIter;
28
29	typedef struct STsdbSnapRAWReader {
30	STsdb* tsdb;
31	int64_t ever;
32	int8_t type;
33
34	TFileSetArray* fsetArr;
35
36	// context
37	struct {
38	int32_t fsetArrIdx;
39	STFileSet* fset;
40	bool isDataDone;
41	} ctx[1];
42
43	// reader
44	SDataFileRAWReaderArray dataReaderArr[1];
45
46	// iter
47	SDataFileRAWReaderIter dataIter[1];
48	} STsdbSnapRAWReader;
49
50	int32_t tsdbSnapRAWReaderOpen(STsdb* tsdb, int64_t ever, int8_t type, STsdbSnapRAWReader** reader) {	1✔
51	int32_t code = 0;	1✔
52	int32_t lino = 0;	1✔
53
54	reader[0] = taosMemoryCalloc(1, sizeof(STsdbSnapRAWReader));	1✔
55	if (reader[0] == NULL) return terrno;	1!
56
57	reader[0]->tsdb = tsdb;	1✔
58	reader[0]->ever = ever;	1✔
59	reader[0]->type = type;	1✔
60
61	code = tsdbFSCreateRefSnapshot(tsdb->pFS, &reader[0]->fsetArr);	1✔
62	TSDB_CHECK_CODE(code, lino, _exit);	1!
63
64	_exit:	1✔
65	if (code) {	1!
66	tsdbError("vgId:%d %s failed at line %d since %s, sver:0, ever:%" PRId64 " type:%d", TD_VID(tsdb->pVnode), __func__,	×
67	lino, tstrerror(code), ever, type);
68	tsdbFSDestroyRefSnapshot(&reader[0]->fsetArr);	×
69	taosMemoryFree(reader[0]);	×
70	reader[0] = NULL;	×
71	} else {
72	tsdbInfo("vgId:%d, tsdb snapshot raw reader opened. sver:0, ever:%" PRId64 " type:%d", TD_VID(tsdb->pVnode), ever,	1!
73	type);
74	}
75	return code;	1✔
76	}
77
78	void tsdbSnapRAWReaderClose(STsdbSnapRAWReader** reader) {	1✔
79	if (reader[0] == NULL) return;	1!
80
81	int32_t code = 0;	1✔
82	int32_t lino = 0;	1✔
83
84	STsdb* tsdb = reader[0]->tsdb;	1✔
85
86	TARRAY2_DESTROY(reader[0]->dataReaderArr, tsdbDataFileRAWReaderClose);	1!
87	tsdbFSDestroyRefSnapshot(&reader[0]->fsetArr);	1✔
88	taosMemoryFree(reader[0]);	1✔
89	reader[0] = NULL;	1✔
90	return;	1✔
91	}
92
93	static int32_t tsdbSnapRAWReadFileSetOpenReader(STsdbSnapRAWReader* reader) {	1✔
94	int32_t code = 0;	1✔
95	int32_t lino = 0;	1✔
96
97	// data
98	for (int32_t ftype = 0; ftype < TSDB_FTYPE_MAX; ftype++) {	5✔
99	if (reader->ctx->fset->farr[ftype] == NULL) {	4!
100	continue;	4✔
101	}
UNCOV 102	STFileObj* fobj = reader->ctx->fset->farr[ftype];	×
103	SDataFileRAWReader* dataReader;
UNCOV 104	SDataFileRAWReaderConfig config = {	×
UNCOV 105	.tsdb = reader->tsdb,	×
UNCOV 106	.szPage = reader->tsdb->pVnode->config.tsdbPageSize,	×
107	.file = fobj->f[0],
108	};
UNCOV 109	code = tsdbDataFileRAWReaderOpen(NULL, &config, &dataReader);	×
UNCOV 110	TSDB_CHECK_CODE(code, lino, _exit);	×
111
UNCOV 112	code = TARRAY2_APPEND(reader->dataReaderArr, dataReader);	×
UNCOV 113	TSDB_CHECK_CODE(code, lino, _exit);	×
114	}
115
116	// stt
117	SSttLvl* lvl;
118	TARRAY2_FOREACH(reader->ctx->fset->lvlArr, lvl) {	2✔
119	STFileObj* fobj;
120	TARRAY2_FOREACH(lvl->fobjArr, fobj) {	2✔
121	SDataFileRAWReader* dataReader;
122	SDataFileRAWReaderConfig config = {	1✔
123	.tsdb = reader->tsdb,	1✔
124	.szPage = reader->tsdb->pVnode->config.tsdbPageSize,	1✔
125	.file = fobj->f[0],
126	};
127	code = tsdbDataFileRAWReaderOpen(NULL, &config, &dataReader);	1✔
128	TSDB_CHECK_CODE(code, lino, _exit);	1!
129
130	code = TARRAY2_APPEND(reader->dataReaderArr, dataReader);	1!
131	TSDB_CHECK_CODE(code, lino, _exit);	1!
132	}
133	}
134
135	_exit:	1✔
136	if (code) {	1!
137	tsdbSnapRAWReadFileSetCloseReader(reader);	×
138	TSDB_ERROR_LOG(TD_VID(reader->tsdb->pVnode), code, lino);	×
139	}
140	return code;	1✔
141	}
142
143	static void tsdbSnapRAWReadFileSetCloseReader(STsdbSnapRAWReader* reader) {	1✔
144	TARRAY2_CLEAR(reader->dataReaderArr, tsdbDataFileRAWReaderClose);	2!
145	}	1✔
146
147	static int32_t tsdbSnapRAWReadFileSetOpenIter(STsdbSnapRAWReader* reader) {	1✔
148	reader->dataIter->count = TARRAY2_SIZE(reader->dataReaderArr);	1✔
149	reader->dataIter->idx = 0;	1✔
150	return 0;	1✔
151	}
152
153	static void tsdbSnapRAWReadFileSetCloseIter(STsdbSnapRAWReader* reader) {	1✔
154	reader->dataIter->count = 0;	1✔
155	reader->dataIter->idx = 0;	1✔
156	}	1✔
157
158	static int64_t tsdbSnapRAWReadPeek(SDataFileRAWReader* reader) {	1✔
159	int64_t size = TMIN(reader->config->file.size - reader->ctx->offset, TSDB_SNAP_DATA_PAYLOAD_SIZE);	1✔
160	return size;	1✔
161	}
162
163	static SDataFileRAWReader* tsdbSnapRAWReaderIterNext(STsdbSnapRAWReader* reader) {	2✔
164	while (reader->dataIter->idx < reader->dataIter->count) {	3✔
165	SDataFileRAWReader* dataReader = TARRAY2_GET(reader->dataReaderArr, reader->dataIter->idx);	2✔
166	if (dataReader->ctx->offset < dataReader->config->file.size) {	2✔
167	return dataReader;	1✔
168	}
169	reader->dataIter->idx++;	1✔
170	}
171	return NULL;	1✔
172	}
173
174	static int32_t tsdbSnapRAWReadNext(STsdbSnapRAWReader* reader, SSnapDataHdr** ppData) {	2✔
175	int32_t code = 0;	2✔
176	int32_t lino = 0;	2✔
177	int8_t type = reader->type;	2✔
178	ppData[0] = NULL;	2✔
179
180	SDataFileRAWReader* dataReader = tsdbSnapRAWReaderIterNext(reader);	2✔
181	if (dataReader == NULL) {	2✔
182	return 0;	1✔
183	}
184
185	// prepare
186	int64_t dataLength = tsdbSnapRAWReadPeek(dataReader);	1✔
187
188	void* pBuf = taosMemoryCalloc(1, sizeof(SSnapDataHdr) + sizeof(STsdbDataRAWBlockHeader) + dataLength);	1✔
189	if (pBuf == NULL) {	1!
190	code = terrno;	×
191	TSDB_CHECK_CODE(code, lino, _exit);	×
192	}
193	SSnapDataHdr* pHdr = pBuf;	1✔
194	pHdr->type = type;	1✔
195	pHdr->size = sizeof(STsdbDataRAWBlockHeader) + dataLength;	1✔
196
197	// read
198	STsdbDataRAWBlockHeader* pBlock = (void*)pHdr->data;	1✔
199	pBlock->offset = dataReader->ctx->offset;	1✔
200	pBlock->dataLength = dataLength;	1✔
201
202	code = tsdbDataFileRAWReadBlockData(dataReader, pBlock);	1✔
203	TSDB_CHECK_CODE(code, lino, _exit);	1!
204
205	// finish
206	dataReader->ctx->offset += pBlock->dataLength;	1✔
207	ppData[0] = pBuf;	1✔
208
209	_exit:	1✔
210	if (code) {	1!
211	taosMemoryFree(pBuf);	×
212	pBuf = NULL;	×
213	TSDB_ERROR_LOG(TD_VID(reader->tsdb->pVnode), code, lino);	×
214	}
215	return code;	1✔
216	}
217
218	static int32_t tsdbSnapRAWReadData(STsdbSnapRAWReader* reader, uint8_t** ppData) {	2✔
219	int32_t code = 0;	2✔
220	int32_t lino = 0;	2✔
221
222	code = tsdbSnapRAWReadNext(reader, (SSnapDataHdr**)ppData);	2✔
223	TSDB_CHECK_CODE(code, lino, _exit);	2!
224
225	_exit:	2✔
226	if (code) {	2!
227	TSDB_ERROR_LOG(TD_VID(reader->tsdb->pVnode), code, lino);	×
228	}
229	return code;	2✔
230	}
231
232	static int32_t tsdbSnapRAWReadBegin(STsdbSnapRAWReader* reader) {	2✔
233	int32_t code = 0;	2✔
234	int32_t lino = 0;	2✔
235
236	if (reader->ctx->fsetArrIdx < TARRAY2_SIZE(reader->fsetArr)) {	2✔
237	reader->ctx->fset = TARRAY2_GET(reader->fsetArr, reader->ctx->fsetArrIdx++);	1✔
238	reader->ctx->isDataDone = false;	1✔
239
240	code = tsdbSnapRAWReadFileSetOpenReader(reader);	1✔
241	TSDB_CHECK_CODE(code, lino, _exit);	1!
242
243	code = tsdbSnapRAWReadFileSetOpenIter(reader);	1✔
244	TSDB_CHECK_CODE(code, lino, _exit);	1!
245	}
246
247	_exit:	2✔
248	if (code) {	2!
249	TSDB_ERROR_LOG(TD_VID(reader->tsdb->pVnode), code, lino);	×
250	}
251	return code;	2✔
252	}
253
254	static int32_t tsdbSnapRAWReadEnd(STsdbSnapRAWReader* reader) {	1✔
255	tsdbSnapRAWReadFileSetCloseIter(reader);	1✔
256	tsdbSnapRAWReadFileSetCloseReader(reader);	1✔
257	reader->ctx->fset = NULL;	1✔
258	return 0;	1✔
259	}
260
261	int32_t tsdbSnapRAWRead(STsdbSnapRAWReader* reader, uint8_t** data) {	2✔
262	int32_t code = 0;	2✔
263	int32_t lino = 0;	2✔
264
265	data[0] = NULL;	2✔
266
267	for (;;) {
268	if (reader->ctx->fset == NULL) {	3✔
269	code = tsdbSnapRAWReadBegin(reader);	2✔
270	TSDB_CHECK_CODE(code, lino, _exit);	2!
271
272	if (reader->ctx->fset == NULL) {	2✔
273	break;	1✔
274	}
275	}
276
277	if (!reader->ctx->isDataDone) {	2!
278	code = tsdbSnapRAWReadData(reader, data);	2✔
279	TSDB_CHECK_CODE(code, lino, _exit);	2!
280	if (data[0]) {	2✔
281	goto _exit;	1✔
282	} else {
283	reader->ctx->isDataDone = true;	1✔
284	}
285	}
286
287	code = tsdbSnapRAWReadEnd(reader);	1✔
288	TSDB_CHECK_CODE(code, lino, _exit);	1!
289	}
290
291	_exit:	2✔
292	if (code) {	2!
293	TSDB_ERROR_LOG(TD_VID(reader->tsdb->pVnode), code, lino);	×
294	} else {
295	tsdbDebug("vgId:%d %s done", TD_VID(reader->tsdb->pVnode), __func__);	2!
296	}
297	return code;	2✔
298	}
299
300	// writer
301	struct STsdbSnapRAWWriter {
302	STsdb* tsdb;
303	int64_t sver;
304	int64_t ever;
305	int32_t minutes;
306	int8_t precision;
307	int32_t minRow;
308	int32_t maxRow;
309	int8_t cmprAlg;
310	int64_t commitID;
311	int32_t szPage;
312	int64_t compactVersion;
313	int64_t now;
314
315	TFileSetArray* fsetArr;
316	TFileOpArray fopArr[1];
317
318	struct {
319	bool fsetWriteBegin;
320	int32_t fid;
321	STFileSet* fset;
322	SDiskID did;
323	int64_t cid;
324	int64_t level;
325
326	// writer
327	SFSetRAWWriter* fsetWriter;
328	} ctx[1];
329	};
330
UNCOV 331	int32_t tsdbSnapRAWWriterOpen(STsdb* pTsdb, int64_t ever, STsdbSnapRAWWriter** writer) {	×
UNCOV 332	int32_t code = 0;	×
UNCOV 333	int32_t lino = 0;	×
334
335	// start to write
UNCOV 336	writer[0] = taosMemoryCalloc(1, sizeof(*writer[0]));	×
UNCOV 337	if (writer[0] == NULL) return terrno;	×
338
UNCOV 339	writer[0]->tsdb = pTsdb;	×
UNCOV 340	writer[0]->ever = ever;	×
UNCOV 341	writer[0]->minutes = pTsdb->keepCfg.days;	×
UNCOV 342	writer[0]->precision = pTsdb->keepCfg.precision;	×
UNCOV 343	writer[0]->minRow = pTsdb->pVnode->config.tsdbCfg.minRows;	×
UNCOV 344	writer[0]->maxRow = pTsdb->pVnode->config.tsdbCfg.maxRows;	×
UNCOV 345	writer[0]->cmprAlg = pTsdb->pVnode->config.tsdbCfg.compression;	×
UNCOV 346	writer[0]->commitID = tsdbFSAllocEid(pTsdb->pFS);	×
UNCOV 347	writer[0]->szPage = pTsdb->pVnode->config.tsdbPageSize;	×
UNCOV 348	writer[0]->compactVersion = INT64_MAX;	×
UNCOV 349	writer[0]->now = taosGetTimestampMs();	×
350
UNCOV 351	code = tsdbFSCreateCopySnapshot(pTsdb->pFS, &writer[0]->fsetArr);	×
UNCOV 352	TSDB_CHECK_CODE(code, lino, _exit);	×
353
UNCOV 354	_exit:	×
UNCOV 355	if (code) {	×
356	tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));	×
357	} else {
UNCOV 358	tsdbInfo("vgId:%d %s done, sver:0, ever:%" PRId64, TD_VID(pTsdb->pVnode), __func__, ever);	×
359	}
UNCOV 360	return code;	×
361	}
362
363	static int32_t tsdbSnapRAWWriteFileSetCloseIter(STsdbSnapRAWWriter* writer) { return 0; }	×
364
UNCOV 365	static int32_t tsdbSnapRAWWriteFileSetOpenWriter(STsdbSnapRAWWriter* writer) {	×
UNCOV 366	int32_t code = 0;	×
UNCOV 367	int32_t lino = 0;	×
368
UNCOV 369	SFSetRAWWriterConfig config = {	×
UNCOV 370	.tsdb = writer->tsdb,	×
UNCOV 371	.szPage = writer->szPage,	×
UNCOV 372	.fid = writer->ctx->fid,	×
UNCOV 373	.cid = writer->commitID,	×
374	.did = writer->ctx->did,
UNCOV 375	.level = writer->ctx->level,	×
376	};
377
UNCOV 378	code = tsdbFSetRAWWriterOpen(&config, &writer->ctx->fsetWriter);	×
UNCOV 379	TSDB_CHECK_CODE(code, lino, _exit);	×
380
UNCOV 381	_exit:	×
UNCOV 382	if (code) {	×
383	TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);	×
384	}
UNCOV 385	return code;	×
386	}
387
UNCOV 388	static int32_t tsdbSnapRAWWriteFileSetCloseWriter(STsdbSnapRAWWriter* writer) {	×
UNCOV 389	return tsdbFSetRAWWriterClose(&writer->ctx->fsetWriter, 0, writer->fopArr);	×
390	}
391
UNCOV 392	static int32_t tsdbSnapRAWWriteFileSetBegin(STsdbSnapRAWWriter* writer, int32_t fid) {	×
UNCOV 393	int32_t code = 0;	×
UNCOV 394	int32_t lino = 0;	×
395
UNCOV 396	STFileSet* fset = &(STFileSet){.fid = fid};	×
397
UNCOV 398	writer->ctx->fid = fid;	×
UNCOV 399	STFileSet** fsetPtr = TARRAY2_SEARCH(writer->fsetArr, &fset, tsdbTFileSetCmprFn, TD_EQ);	×
UNCOV 400	writer->ctx->fset = (fsetPtr == NULL) ? NULL : *fsetPtr;	×
401
UNCOV 402	int32_t level = tsdbFidLevel(fid, &writer->tsdb->keepCfg, taosGetTimestampSec());	×
UNCOV 403	code = tfsAllocDisk(writer->tsdb->pVnode->pTfs, level, &writer->ctx->did);	×
UNCOV 404	TSDB_CHECK_CODE(code, lino, _exit);	×
405
UNCOV 406	code = tfsMkdirRecurAt(writer->tsdb->pVnode->pTfs, writer->tsdb->path, writer->ctx->did);	×
UNCOV 407	TSDB_CHECK_CODE(code, lino, _exit);	×
408
UNCOV 409	code = tsdbSnapRAWWriteFileSetOpenWriter(writer);	×
UNCOV 410	TSDB_CHECK_CODE(code, lino, _exit);	×
411
UNCOV 412	writer->ctx->level = level;	×
UNCOV 413	writer->ctx->fsetWriteBegin = true;	×
414
UNCOV 415	_exit:	×
UNCOV 416	if (code) {	×
417	TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);	×
418	}
UNCOV 419	return code;	×
420	}
421
UNCOV 422	static int32_t tsdbSnapRAWWriteFileSetEnd(STsdbSnapRAWWriter* writer) {	×
UNCOV 423	if (!writer->ctx->fsetWriteBegin) return 0;	×
424
UNCOV 425	int32_t code = 0;	×
UNCOV 426	int32_t lino = 0;	×
427
428	// close write
UNCOV 429	code = tsdbSnapRAWWriteFileSetCloseWriter(writer);	×
UNCOV 430	TSDB_CHECK_CODE(code, lino, _exit);	×
431
UNCOV 432	writer->ctx->fsetWriteBegin = false;	×
433
UNCOV 434	_exit:	×
UNCOV 435	if (code) {	×
436	TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);	×
437	}
UNCOV 438	return code;	×
439	}
440
UNCOV 441	int32_t tsdbSnapRAWWriterPrepareClose(STsdbSnapRAWWriter* writer) {	×
UNCOV 442	int32_t code = 0;	×
UNCOV 443	int32_t lino = 0;	×
444
UNCOV 445	code = tsdbSnapRAWWriteFileSetEnd(writer);	×
UNCOV 446	TSDB_CHECK_CODE(code, lino, _exit);	×
447
UNCOV 448	code = tsdbFSEditBegin(writer->tsdb->pFS, writer->fopArr, TSDB_FEDIT_COMMIT);	×
UNCOV 449	TSDB_CHECK_CODE(code, lino, _exit);	×
450
UNCOV 451	_exit:	×
UNCOV 452	if (code) {	×
453	TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);	×
454	} else {
UNCOV 455	tsdbDebug("vgId:%d %s done", TD_VID(writer->tsdb->pVnode), __func__);	×
456	}
UNCOV 457	return code;	×
458	}
459
UNCOV 460	int32_t tsdbSnapRAWWriterClose(STsdbSnapRAWWriter** writer, int8_t rollback) {	×
UNCOV 461	if (writer[0] == NULL) return 0;	×
462
UNCOV 463	int32_t code = 0;	×
UNCOV 464	int32_t lino = 0;	×
465
UNCOV 466	STsdb* tsdb = writer[0]->tsdb;	×
467
UNCOV 468	if (rollback) {	×
469	code = tsdbFSEditAbort(writer[0]->tsdb->pFS);	×
470	TSDB_CHECK_CODE(code, lino, _exit);	×
471	} else {
UNCOV 472	(void)taosThreadMutexLock(&writer[0]->tsdb->mutex);	×
473
UNCOV 474	code = tsdbFSEditCommit(writer[0]->tsdb->pFS);	×
UNCOV 475	if (code) {	×
476	(void)taosThreadMutexUnlock(&writer[0]->tsdb->mutex);	×
477	TSDB_CHECK_CODE(code, lino, _exit);	×
478	}
479
UNCOV 480	writer[0]->tsdb->pFS->fsstate = TSDB_FS_STATE_NORMAL;	×
481
UNCOV 482	(void)taosThreadMutexUnlock(&writer[0]->tsdb->mutex);	×
483	}
484
UNCOV 485	TARRAY2_DESTROY(writer[0]->fopArr, NULL);	×
UNCOV 486	tsdbFSDestroyCopySnapshot(&writer[0]->fsetArr);	×
487
UNCOV 488	taosMemoryFree(writer[0]);	×
UNCOV 489	writer[0] = NULL;	×
490
UNCOV 491	_exit:	×
UNCOV 492	if (code) {	×
493	TSDB_ERROR_LOG(TD_VID(tsdb->pVnode), lino, code);	×
494	} else {
UNCOV 495	tsdbInfo("vgId:%d %s done", TD_VID(tsdb->pVnode), __func__);	×
496	}
UNCOV 497	return code;	×
498	}
499
UNCOV 500	static int32_t tsdbSnapRAWWriteTimeSeriesData(STsdbSnapRAWWriter* writer, STsdbDataRAWBlockHeader* bHdr) {	×
UNCOV 501	int32_t code = 0;	×
UNCOV 502	int32_t lino = 0;	×
503
UNCOV 504	int32_t encryptAlgorithm = writer->tsdb->pVnode->config.tsdbCfg.encryptAlgorithm;	×
UNCOV 505	char* encryptKey = writer->tsdb->pVnode->config.tsdbCfg.encryptKey;	×
506
UNCOV 507	code = tsdbFSetRAWWriteBlockData(writer->ctx->fsetWriter, bHdr, encryptAlgorithm, encryptKey);	×
UNCOV 508	TSDB_CHECK_CODE(code, lino, _exit);	×
509
UNCOV 510	_exit:	×
UNCOV 511	if (code) {	×
512	TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);	×
513	}
UNCOV 514	return code;	×
515	}
516
UNCOV 517	static int32_t tsdbSnapRAWWriteData(STsdbSnapRAWWriter* writer, SSnapDataHdr* hdr) {	×
UNCOV 518	int32_t code = 0;	×
UNCOV 519	int32_t lino = 0;	×
520
UNCOV 521	STsdbDataRAWBlockHeader* bHdr = (void*)hdr->data;	×
UNCOV 522	int32_t fid = bHdr->file.fid;	×
UNCOV 523	if (!writer->ctx->fsetWriteBegin \|\| fid != writer->ctx->fid) {	×
UNCOV 524	code = tsdbSnapRAWWriteFileSetEnd(writer);	×
UNCOV 525	TSDB_CHECK_CODE(code, lino, _exit);	×
526
UNCOV 527	code = tsdbSnapRAWWriteFileSetBegin(writer, fid);	×
UNCOV 528	TSDB_CHECK_CODE(code, lino, _exit);	×
529	}
530
UNCOV 531	code = tsdbSnapRAWWriteTimeSeriesData(writer, bHdr);	×
UNCOV 532	TSDB_CHECK_CODE(code, lino, _exit);	×
533
UNCOV 534	_exit:	×
UNCOV 535	if (code) {	×
536	TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);	×
537	}
UNCOV 538	return code;	×
539	}
540
UNCOV 541	int32_t tsdbSnapRAWWrite(STsdbSnapRAWWriter* writer, SSnapDataHdr* hdr) {	×
UNCOV 542	int32_t code = 0;	×
UNCOV 543	int32_t lino = 0;	×
544
UNCOV 545	code = tsdbSnapRAWWriteData(writer, hdr);	×
UNCOV 546	TSDB_CHECK_CODE(code, lino, _exit);	×
547
UNCOV 548	_exit:	×
UNCOV 549	if (code) {	×
550	tsdbError("vgId:%d %s failed at line %d since %s, type:%d index:%" PRId64 " size:%" PRId64,	×
551	TD_VID(writer->tsdb->pVnode), __func__, lino, tstrerror(code), hdr->type, hdr->index, hdr->size);
552	} else {
UNCOV 553	tsdbDebug("vgId:%d %s done, type:%d index:%" PRId64 " size:%" PRId64, TD_VID(writer->tsdb->pVnode), __func__,	×
554	hdr->type, hdr->index, hdr->size);
555	}
UNCOV 556	return code;	×
557	}

taosdata / TDengine / #3523

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