tsdbSnapshot.c 29.7 KB
Newer Older
H
Hongze Cheng 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
/*
 * 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"
H
Hongze Cheng 已提交
17 18
#include "tsdbDataFileRW.h"
#include "tsdbFS2.h"
H
Hongze Cheng 已提交
19
#include "tsdbFSetRW.h"
H
Hongze Cheng 已提交
20 21
#include "tsdbIter.h"
#include "tsdbSttFileRW.h"
H
Hongze Cheng 已提交
22

H
Hongze Cheng 已提交
23 24 25
extern int32_t tsdbUpdateTableSchema(SMeta* pMeta, int64_t suid, int64_t uid, SSkmInfo* pSkmInfo);
extern int32_t tsdbWriteDataBlock(SDataFWriter* pWriter, SBlockData* pBlockData, SMapData* mDataBlk, int8_t cmprAlg);
extern int32_t tsdbWriteSttBlock(SDataFWriter* pWriter, SBlockData* pBlockData, SArray* aSttBlk, int8_t cmprAlg);
H
Hongze Cheng 已提交
26

H
Hongze Cheng 已提交
27
// STsdbSnapReader ========================================
H
Hongze Cheng 已提交
28
struct STsdbSnapReader {
H
Hongze Cheng 已提交
29 30 31 32 33
  STsdb*  tsdb;
  int64_t sver;
  int64_t ever;
  int8_t  type;

H
Hongze Cheng 已提交
34
  uint8_t* aBuf[5];
H
Hongze Cheng 已提交
35 36 37 38 39 40 41 42 43 44 45
  SSkmInfo skmTb[1];

  TFileSetArray* fsetArr;

  // context
  struct {
    int32_t    fsetArrIdx;
    STFileSet* fset;
    bool       isDataDone;
    bool       isTombDone;
  } ctx[1];
H
Hongze Cheng 已提交
46

H
Hongze Cheng 已提交
47 48 49 50 51 52 53 54 55 56 57 58 59
  // reader
  SDataFileReader*    dataReader;
  TSttFileReaderArray sttReaderArr[1];

  // iter
  TTsdbIterArray dataIterArr[1];
  SIterMerger*   dataIterMerger;
  TTsdbIterArray tombIterArr[1];
  SIterMerger*   tombIterMerger;

  // data
  SBlockData blockData[1];
  STombBlock tombBlock[1];
H
Hongze Cheng 已提交
60
};
H
Hongze Cheng 已提交
61

H
Hongze Cheng 已提交
62
static int32_t tsdbSnapReadFileSetOpenReader(STsdbSnapReader* reader) {
H
Hongze Cheng 已提交
63
  int32_t code = 0;
H
add log  
Hongze Cheng 已提交
64
  int32_t lino = 0;
H
Hongze Cheng 已提交
65

H
Hongze Cheng 已提交
66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81
  ASSERT(reader->dataReader == NULL);
  ASSERT(TARRAY2_SIZE(reader->sttReaderArr) == 0);

  // data
  SDataFileReaderConfig config = {
      .tsdb = reader->tsdb,
      .szPage = reader->tsdb->pVnode->config.tsdbPageSize,
      .bufArr = reader->aBuf,
  };
  bool hasDataFile = false;
  for (int32_t ftype = 0; ftype < TSDB_FTYPE_MAX; ftype++) {
    if (reader->ctx->fset->farr[ftype] != NULL) {
      hasDataFile = true;
      config.files[ftype].exist = true;
      config.files[ftype].file = reader->ctx->fset->farr[ftype]->f[0];
    }
H
Hongze Cheng 已提交
82
  }
H
Hongze Cheng 已提交
83

H
Hongze Cheng 已提交
84 85 86 87
  if (hasDataFile) {
    code = tsdbDataFileReaderOpen(NULL, &config, &reader->dataReader);
    TSDB_CHECK_CODE(code, lino, _exit);
  }
H
Hongze Cheng 已提交
88

H
Hongze Cheng 已提交
89 90 91 92 93 94 95 96 97 98 99 100 101 102 103
  // stt
  SSttLvl* lvl;
  TARRAY2_FOREACH(reader->ctx->fset->lvlArr, lvl) {
    STFileObj* fobj;
    TARRAY2_FOREACH(lvl->fobjArr, fobj) {
      SSttFileReader*      sttReader;
      SSttFileReaderConfig config = {
          .tsdb = reader->tsdb,
          .szPage = reader->tsdb->pVnode->config.tsdbPageSize,
          .file = fobj->f[0],
          .bufArr = reader->aBuf,
      };

      code = tsdbSttFileReaderOpen(fobj->fname, &config, &sttReader);
      TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
104

H
Hongze Cheng 已提交
105 106 107 108
      code = TARRAY2_APPEND(reader->sttReaderArr, sttReader);
      TSDB_CHECK_CODE(code, lino, _exit);
    }
  }
H
Hongze Cheng 已提交
109

H
Hongze Cheng 已提交
110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154
_exit:
  if (code) {
    TSDB_ERROR_LOG(TD_VID(reader->tsdb->pVnode), code, lino);
  }
  return code;
}

static int32_t tsdbSnapReadFileSetCloseReader(STsdbSnapReader* reader) {
  int32_t code = 0;
  int32_t lino = 0;

  TARRAY2_CLEAR(reader->sttReaderArr, tsdbSttFileReaderClose);
  tsdbDataFileReaderClose(&reader->dataReader);

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

static int32_t tsdbSnapReadFileSetOpenIter(STsdbSnapReader* reader) {
  int32_t code = 0;
  int32_t lino = 0;

  ASSERT(reader->dataIterMerger == NULL);
  ASSERT(reader->tombIterMerger == NULL);
  ASSERT(TARRAY2_SIZE(reader->dataIterArr) == 0);
  ASSERT(TARRAY2_SIZE(reader->tombIterArr) == 0);

  STsdbIter*      iter;
  STsdbIterConfig config = {
      .filterByVersion = true,
      .verRange[0] = reader->sver,
      .verRange[1] = reader->ever,
  };

  // data file
  if (reader->dataReader) {
    // data
    config.type = TSDB_ITER_TYPE_DATA;
    config.dataReader = reader->dataReader;

    code = tsdbIterOpen(&config, &iter);
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
155

H
Hongze Cheng 已提交
156
    code = TARRAY2_APPEND(reader->dataIterArr, iter);
H
add log  
Hongze Cheng 已提交
157
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
158

H
Hongze Cheng 已提交
159 160 161
    // tomb
    config.type = TSDB_ITER_TYPE_DATA_TOMB;
    config.dataReader = reader->dataReader;
H
Hongze Cheng 已提交
162

H
Hongze Cheng 已提交
163 164 165 166 167
    code = tsdbIterOpen(&config, &iter);
    TSDB_CHECK_CODE(code, lino, _exit);

    code = TARRAY2_APPEND(reader->tombIterArr, iter);
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
168 169
  }

H
Hongze Cheng 已提交
170 171 172 173 174 175 176 177
  // stt file
  SSttFileReader* sttReader;
  TARRAY2_FOREACH(reader->sttReaderArr, sttReader) {
    // data
    config.type = TSDB_ITER_TYPE_STT;
    config.sttReader = sttReader;

    code = tsdbIterOpen(&config, &iter);
H
add log  
Hongze Cheng 已提交
178
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
179

H
Hongze Cheng 已提交
180 181
    code = TARRAY2_APPEND(reader->dataIterArr, iter);
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
182

H
Hongze Cheng 已提交
183 184 185
    // tomb
    config.type = TSDB_ITER_TYPE_STT_TOMB;
    config.sttReader = sttReader;
H
Hongze Cheng 已提交
186

H
Hongze Cheng 已提交
187 188 189 190 191
    code = tsdbIterOpen(&config, &iter);
    TSDB_CHECK_CODE(code, lino, _exit);

    code = TARRAY2_APPEND(reader->tombIterArr, iter);
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
192 193
  }

H
Hongze Cheng 已提交
194 195 196 197
  // merger
  code = tsdbIterMergerOpen(reader->dataIterArr, &reader->dataIterMerger, false);
  TSDB_CHECK_CODE(code, lino, _exit);

H
Hongze Cheng 已提交
198
  code = tsdbIterMergerOpen(reader->tombIterArr, &reader->tombIterMerger, true);
H
Hongze Cheng 已提交
199
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
200

H
add log  
Hongze Cheng 已提交
201 202
_exit:
  if (code) {
H
Hongze Cheng 已提交
203
    TSDB_ERROR_LOG(TD_VID(reader->tsdb->pVnode), code, lino);
H
add log  
Hongze Cheng 已提交
204
  }
H
Hongze Cheng 已提交
205 206 207
  return code;
}

H
Hongze Cheng 已提交
208 209 210 211 212 213
static int32_t tsdbSnapReadFileSetCloseIter(STsdbSnapReader* reader) {
  tsdbIterMergerClose(&reader->dataIterMerger);
  tsdbIterMergerClose(&reader->tombIterMerger);
  TARRAY2_CLEAR(reader->dataIterArr, tsdbIterClose);
  TARRAY2_CLEAR(reader->tombIterArr, tsdbIterClose);
  return 0;
H
Hongze Cheng 已提交
214
}
H
Hongze Cheng 已提交
215

H
Hongze Cheng 已提交
216
static int32_t tsdbSnapReadFileSetBegin(STsdbSnapReader* reader) {
H
Hongze Cheng 已提交
217 218
  int32_t code = 0;
  int32_t lino = 0;
H
Hongze Cheng 已提交
219

H
Hongze Cheng 已提交
220 221
  ASSERT(reader->ctx->fset == NULL);

H
Hongze Cheng 已提交
222 223 224 225
  if (reader->ctx->fsetArrIdx < TARRAY2_SIZE(reader->fsetArr)) {
    reader->ctx->fset = TARRAY2_GET(reader->fsetArr, reader->ctx->fsetArrIdx++);
    reader->ctx->isDataDone = false;
    reader->ctx->isTombDone = false;
H
Hongze Cheng 已提交
226

H
Hongze Cheng 已提交
227 228
    code = tsdbSnapReadFileSetOpenReader(reader);
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
229

H
Hongze Cheng 已提交
230 231
    code = tsdbSnapReadFileSetOpenIter(reader);
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
232
  }
H
Hongze Cheng 已提交
233

H
Hongze Cheng 已提交
234 235
_exit:
  if (code) {
H
Hongze Cheng 已提交
236
    TSDB_ERROR_LOG(TD_VID(reader->tsdb->pVnode), code, lino);
H
Hongze Cheng 已提交
237
  }
H
Hongze Cheng 已提交
238 239 240
  return code;
}

H
Hongze Cheng 已提交
241 242 243 244 245
static int32_t tsdbSnapReadFileSetEnd(STsdbSnapReader* reader) {
  tsdbSnapReadFileSetCloseIter(reader);
  tsdbSnapReadFileSetCloseReader(reader);
  reader->ctx->fset = NULL;
  return 0;
H
Hongze Cheng 已提交
246 247
}

H
Hongze Cheng 已提交
248
static int32_t tsdbSnapCmprData(STsdbSnapReader* reader, uint8_t** data) {
H
Hongze Cheng 已提交
249
  int32_t code = 0;
H
Hongze Cheng 已提交
250
  int32_t lino = 0;
H
Hongze Cheng 已提交
251 252

  int32_t aBufN[5] = {0};
H
Hongze Cheng 已提交
253 254
  code = tCmprBlockData(reader->blockData, NO_COMPRESSION, NULL, NULL, reader->aBuf, aBufN);
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
255 256

  int32_t size = aBufN[0] + aBufN[1] + aBufN[2] + aBufN[3];
H
Hongze Cheng 已提交
257 258
  *data = taosMemoryMalloc(sizeof(SSnapDataHdr) + size);
  if (*data == NULL) {
H
Hongze Cheng 已提交
259
    code = TSDB_CODE_OUT_OF_MEMORY;
H
Hongze Cheng 已提交
260
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
261 262
  }

H
Hongze Cheng 已提交
263 264
  SSnapDataHdr* pHdr = (SSnapDataHdr*)*data;
  pHdr->type = reader->type;
H
Hongze Cheng 已提交
265 266
  pHdr->size = size;

H
Hongze Cheng 已提交
267 268
  memcpy(pHdr->data, reader->aBuf[3], aBufN[3]);
  memcpy(pHdr->data + aBufN[3], reader->aBuf[2], aBufN[2]);
H
Hongze Cheng 已提交
269
  if (aBufN[1]) {
H
Hongze Cheng 已提交
270
    memcpy(pHdr->data + aBufN[3] + aBufN[2], reader->aBuf[1], aBufN[1]);
H
Hongze Cheng 已提交
271 272
  }
  if (aBufN[0]) {
H
Hongze Cheng 已提交
273
    memcpy(pHdr->data + aBufN[3] + aBufN[2] + aBufN[1], reader->aBuf[0], aBufN[0]);
H
Hongze Cheng 已提交
274 275 276
  }

_exit:
H
Hongze Cheng 已提交
277 278 279
  if (code) {
    TSDB_ERROR_LOG(TD_VID(reader->tsdb->pVnode), lino, code);
  }
H
Hongze Cheng 已提交
280 281 282
  return code;
}

H
Hongze Cheng 已提交
283
static int32_t tsdbSnapReadTimeSeriesData(STsdbSnapReader* reader, uint8_t** data) {
H
Hongze Cheng 已提交
284
  int32_t code = 0;
H
add log  
Hongze Cheng 已提交
285 286
  int32_t lino = 0;

H
Hongze Cheng 已提交
287
  tBlockDataReset(reader->blockData);
H
Hongze Cheng 已提交
288

H
Hongze Cheng 已提交
289 290 291 292
  for (SRowInfo* row; (row = tsdbIterMergerGetData(reader->dataIterMerger));) {
    if (reader->blockData->suid == 0 && reader->blockData->uid == 0) {
      code = tsdbUpdateSkmTb(reader->tsdb, (TABLEID*)row, reader->skmTb);
      TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
293

H
Hongze Cheng 已提交
294 295 296 297 298
      TABLEID tbid = {
          .suid = row->suid,
          .uid = row->suid ? 0 : row->uid,
      };
      code = tBlockDataInit(reader->blockData, &tbid, reader->skmTb->pTSchema, NULL, 0);
H
add log  
Hongze Cheng 已提交
299
      TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
300 301
    }

H
Hongze Cheng 已提交
302 303
    if (!TABLE_SAME_SCHEMA(reader->blockData->suid, reader->blockData->uid, row->suid, row->uid)) {
      break;
H
Hongze Cheng 已提交
304 305
    }

H
Hongze Cheng 已提交
306
    code = tBlockDataAppendRow(reader->blockData, &row->row, NULL, row->uid);
H
add log  
Hongze Cheng 已提交
307
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
308

H
Hongze Cheng 已提交
309
    code = tsdbIterMergerNext(reader->dataIterMerger);
H
add log  
Hongze Cheng 已提交
310
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
311

H
Hongze Cheng 已提交
312 313 314
    if (reader->blockData->nRow >= 81920) {
      break;
    }
H
Hongze Cheng 已提交
315
  }
H
Hongze Cheng 已提交
316

H
Hongze Cheng 已提交
317 318 319
  if (reader->blockData->nRow > 0) {
    ASSERT(reader->blockData->suid || reader->blockData->uid);
    code = tsdbSnapCmprData(reader, data);
H
add log  
Hongze Cheng 已提交
320
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
321 322
  }

H
add log  
Hongze Cheng 已提交
323 324
_exit:
  if (code) {
H
Hongze Cheng 已提交
325
    TSDB_ERROR_LOG(TD_VID(reader->tsdb->pVnode), code, lino);
H
add log  
Hongze Cheng 已提交
326
  }
H
Hongze Cheng 已提交
327 328 329
  return code;
}

H
Hongze Cheng 已提交
330
static int32_t tsdbSnapCmprTombData(STsdbSnapReader* reader, uint8_t** data) {
H
add log  
Hongze Cheng 已提交
331 332 333
  int32_t code = 0;
  int32_t lino = 0;

H
Hongze Cheng 已提交
334 335 336
  int64_t size = sizeof(SSnapDataHdr);
  for (int32_t i = 0; i < ARRAY_SIZE(reader->tombBlock->dataArr); i++) {
    size += TARRAY2_DATA_LEN(reader->tombBlock->dataArr + i);
H
Hongze Cheng 已提交
337
  }
H
Hongze Cheng 已提交
338

H
Hongze Cheng 已提交
339 340
  data[0] = taosMemoryMalloc(size);
  if (data[0] == NULL) {
H
Hongze Cheng 已提交
341
    code = TSDB_CODE_OUT_OF_MEMORY;
H
add log  
Hongze Cheng 已提交
342
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
343
  }
H
Hongze Cheng 已提交
344

H
Hongze Cheng 已提交
345 346 347
  SSnapDataHdr* hdr = (SSnapDataHdr*)data[0];
  hdr->type = SNAP_DATA_DEL;
  hdr->size = size;
H
Hongze Cheng 已提交
348

H
Hongze Cheng 已提交
349 350 351 352
  uint8_t* tdata = hdr->data;
  for (int32_t i = 0; i < TARRAY_SIZE(reader->tombBlock->dataArr); i++) {
    memcpy(tdata, TARRAY2_DATA(reader->tombBlock->dataArr + i), TARRAY2_DATA_LEN(reader->tombBlock->dataArr + i));
    tdata += TARRAY2_DATA_LEN(reader->tombBlock->dataArr + i);
H
Hongze Cheng 已提交
353 354
  }

H
Hongze Cheng 已提交
355 356
_exit:
  if (code) {
H
Hongze Cheng 已提交
357
    TSDB_ERROR_LOG(TD_VID(reader->tsdb->pVnode), code, lino);
H
Hongze Cheng 已提交
358 359 360
  }
  return code;
}
H
Hongze Cheng 已提交
361

H
Hongze Cheng 已提交
362
static int32_t tsdbSnapReadTombData(STsdbSnapReader* reader, uint8_t** data) {
H
Hongze Cheng 已提交
363 364
  int32_t code = 0;
  int32_t lino = 0;
H
Hongze Cheng 已提交
365

H
Hongze Cheng 已提交
366
  tTombBlockClear(reader->tombBlock);
H
Hongze Cheng 已提交
367

H
Hongze Cheng 已提交
368 369
  for (STombRecord* record; (record = tsdbIterMergerGetTombRecord(reader->tombIterMerger)) != NULL;) {
    code = tTombBlockPut(reader->tombBlock, record);
H
add log  
Hongze Cheng 已提交
370
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
371

H
Hongze Cheng 已提交
372 373
    if (TOMB_BLOCK_SIZE(reader->tombBlock) >= 81920) {
      break;
H
Hongze Cheng 已提交
374
    }
H
Hongze Cheng 已提交
375 376
  }

H
Hongze Cheng 已提交
377 378
  if (TOMB_BLOCK_SIZE(reader->tombBlock) > 0) {
    code = tsdbSnapCmprTombData(reader, data);
H
Hongze Cheng 已提交
379
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
380
  }
H
Hongze Cheng 已提交
381 382

_exit:
H
add log  
Hongze Cheng 已提交
383
  if (code) {
H
Hongze Cheng 已提交
384
    TSDB_ERROR_LOG(TD_VID(reader->tsdb->pVnode), code, lino);
H
add log  
Hongze Cheng 已提交
385
  }
H
Hongze Cheng 已提交
386 387
  return code;
}
H
more  
Hongze Cheng 已提交
388

H
Hongze Cheng 已提交
389
int32_t tsdbSnapReaderOpen(STsdb* tsdb, int64_t sver, int64_t ever, int8_t type, STsdbSnapReader** reader) {
H
Hongze Cheng 已提交
390 391
  int32_t code = 0;
  int32_t lino = 0;
H
Hongze Cheng 已提交
392

H
Hongze Cheng 已提交
393 394
  reader[0] = (STsdbSnapReader*)taosMemoryCalloc(1, sizeof(*reader[0]));
  if (reader[0] == NULL) return TSDB_CODE_OUT_OF_MEMORY;
H
more  
Hongze Cheng 已提交
395

H
Hongze Cheng 已提交
396 397 398 399
  reader[0]->tsdb = tsdb;
  reader[0]->sver = sver;
  reader[0]->ever = ever;
  reader[0]->type = type;
H
Hongze Cheng 已提交
400

H
Hongze Cheng 已提交
401
  code = tsdbFSCreateRefSnapshot(tsdb->pFS, &reader[0]->fsetArr);
H
Hongze Cheng 已提交
402
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
403

H
Hongze Cheng 已提交
404 405
_exit:
  if (code) {
H
Hongze Cheng 已提交
406
    tsdbError("vgId:%d %s failed at line %d since %s, sver:%" PRId64 " ever:%" PRId64 " type:%d", TD_VID(tsdb->pVnode),
H
Hongze Cheng 已提交
407
              __func__, lino, tstrerror(code), sver, ever, type);
H
Hongze Cheng 已提交
408 409 410
    tsdbFSDestroyRefSnapshot(&reader[0]->fsetArr);
    taosMemoryFree(reader[0]);
    reader[0] = NULL;
H
Hongze Cheng 已提交
411
  } else {
H
Hongze Cheng 已提交
412
    tsdbInfo("vgId:%d %s done, sver:%" PRId64 " ever:%" PRId64 " type:%d", TD_VID(tsdb->pVnode), __func__, sver, ever,
H
Hongze Cheng 已提交
413
             type);
H
Hongze Cheng 已提交
414
  }
H
more  
Hongze Cheng 已提交
415
  return code;
H
Hongze Cheng 已提交
416 417
}

H
Hongze Cheng 已提交
418 419 420
int32_t tsdbSnapReaderClose(STsdbSnapReader** reader) {
  if (reader[0] == NULL) return 0;

H
Hongze Cheng 已提交
421 422
  int32_t code = 0;
  int32_t lino = 0;
H
Hongze Cheng 已提交
423

H
Hongze Cheng 已提交
424
  STsdb* tsdb = reader[0]->tsdb;
H
Hongze Cheng 已提交
425

H
Hongze Cheng 已提交
426 427
  tTombBlockDestroy(reader[0]->tombBlock);
  tBlockDataDestroy(reader[0]->blockData);
H
Hongze Cheng 已提交
428

H
Hongze Cheng 已提交
429 430 431 432 433 434
  tsdbIterMergerClose(&reader[0]->dataIterMerger);
  tsdbIterMergerClose(&reader[0]->tombIterMerger);
  TARRAY2_DESTROY(reader[0]->dataIterArr, tsdbIterClose);
  TARRAY2_DESTROY(reader[0]->tombIterArr, tsdbIterClose);
  TARRAY2_DESTROY(reader[0]->sttReaderArr, tsdbSttFileReaderClose);
  tsdbDataFileReaderClose(&reader[0]->dataReader);
H
Hongze Cheng 已提交
435

H
Hongze Cheng 已提交
436 437 438
  tsdbFSDestroyRefSnapshot(&reader[0]->fsetArr);
  tDestroyTSchema(reader[0]->skmTb->pTSchema);

H
Hongze Cheng 已提交
439
  for (int32_t i = 0; i < ARRAY_SIZE(reader[0]->aBuf); ++i) {
H
Hongze Cheng 已提交
440
    tFree(reader[0]->aBuf[i]);
H
Hongze Cheng 已提交
441
  }
H
Hongze Cheng 已提交
442 443 444

  taosMemoryFree(reader[0]);
  reader[0] = NULL;
H
Hongze Cheng 已提交
445 446 447

_exit:
  if (code) {
H
Hongze Cheng 已提交
448
    TSDB_ERROR_LOG(TD_VID(tsdb->pVnode), lino, code);
H
Hongze Cheng 已提交
449
  } else {
H
Hongze Cheng 已提交
450
    tsdbDebug("vgId:%d %s done", TD_VID(tsdb->pVnode), __func__);
H
Hongze Cheng 已提交
451
  }
H
Hongze Cheng 已提交
452 453 454
  return code;
}

H
Hongze Cheng 已提交
455
int32_t tsdbSnapRead(STsdbSnapReader* reader, uint8_t** data) {
H
more  
Hongze Cheng 已提交
456
  int32_t code = 0;
H
add log  
Hongze Cheng 已提交
457
  int32_t lino = 0;
H
Hongze Cheng 已提交
458

H
Hongze Cheng 已提交
459
  data[0] = NULL;
H
Hongze Cheng 已提交
460

H
Hongze Cheng 已提交
461 462 463 464 465 466 467 468
  for (;;) {
    if (reader->ctx->fset == NULL) {
      code = tsdbSnapReadFileSetBegin(reader);
      TSDB_CHECK_CODE(code, lino, _exit);

      if (reader->ctx->fset == NULL) {
        break;
      }
H
Hongze Cheng 已提交
469
    }
H
Hongze Cheng 已提交
470

H
Hongze Cheng 已提交
471 472 473 474 475 476 477 478
    if (!reader->ctx->isDataDone) {
      code = tsdbSnapReadTimeSeriesData(reader, data);
      TSDB_CHECK_CODE(code, lino, _exit);
      if (data[0]) {
        goto _exit;
      } else {
        reader->ctx->isDataDone = true;
      }
H
Hongze Cheng 已提交
479
    }
H
Hongze Cheng 已提交
480 481 482 483 484 485 486 487 488 489 490 491 492

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

    code = tsdbSnapReadFileSetEnd(reader);
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
493
  }
H
Hongze Cheng 已提交
494

H
Hongze Cheng 已提交
495
_exit:
H
add log  
Hongze Cheng 已提交
496
  if (code) {
H
Hongze Cheng 已提交
497
    TSDB_ERROR_LOG(TD_VID(reader->tsdb->pVnode), code, lino);
H
add log  
Hongze Cheng 已提交
498
  } else {
H
Hongze Cheng 已提交
499
    tsdbDebug("vgId:%d %s done", TD_VID(reader->tsdb->pVnode), __func__);
H
add log  
Hongze Cheng 已提交
500
  }
H
more  
Hongze Cheng 已提交
501 502 503
  return code;
}

H
Hongze Cheng 已提交
504
// STsdbSnapWriter ========================================
H
Hongze Cheng 已提交
505
struct STsdbSnapWriter {
H
Hongze Cheng 已提交
506
  STsdb*   tsdb;
H
Hongze Cheng 已提交
507 508
  int64_t  sver;
  int64_t  ever;
H
Hongze Cheng 已提交
509 510 511 512 513 514
  int32_t  minutes;
  int8_t   precision;
  int32_t  minRow;
  int32_t  maxRow;
  int8_t   cmprAlg;
  int64_t  commitID;
H
Hongze Cheng 已提交
515 516 517
  int32_t  szPage;
  int64_t  compactVersion;
  int64_t  now;
H
Hongze Cheng 已提交
518
  uint8_t* aBuf[5];
H
Hongze Cheng 已提交
519

H
Hongze Cheng 已提交
520 521 522 523
  TFileSetArray* fsetArr;
  TFileOpArray   fopArr[1];

  struct {
H
Hongze Cheng 已提交
524
    bool       fsetWriteBegin;
H
Hongze Cheng 已提交
525 526
    int32_t    fid;
    STFileSet* fset;
H
Hongze Cheng 已提交
527 528 529
    SDiskID    did;
    bool       hasData;
    bool       hasTomb;
H
Hongze Cheng 已提交
530 531 532 533 534 535 536 537 538 539 540

    // reader
    SDataFileReader*    dataReader;
    TSttFileReaderArray sttReaderArr[1];

    // iter/merger
    TTsdbIterArray dataIterArr[1];
    SIterMerger*   dataIterMerger;
    TTsdbIterArray tombIterArr[1];
    SIterMerger*   tombIterMerger;

H
Hongze Cheng 已提交
541 542 543
    // writer
    SFSetWriter* fsetWriter;
  } ctx[1];
H
Hongze Cheng 已提交
544 545
};

H
Hongze Cheng 已提交
546 547
// APIs
static int32_t tsdbSnapWriteTimeSeriesRow(STsdbSnapWriter* writer, SRowInfo* row) {
H
Hongze Cheng 已提交
548
  int32_t code = 0;
H
Hongze Cheng 已提交
549 550
  int32_t lino = 0;

H
Hongze Cheng 已提交
551 552 553 554 555
  while (writer->ctx->hasData) {
    SRowInfo* row1 = tsdbIterMergerGetData(writer->ctx->dataIterMerger);
    if (row1 == NULL) {
      writer->ctx->hasData = false;
      break;
H
Hongze Cheng 已提交
556
    }
H
Hongze Cheng 已提交
557

H
Hongze Cheng 已提交
558 559 560
    int32_t c = tRowInfoCmprFn(row1, row);
    if (c <= 0) {
      code = tsdbFSetWriteRow(writer->ctx->fsetWriter, row1);
H
Hongze Cheng 已提交
561
      TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
562

H
Hongze Cheng 已提交
563
      code = tsdbIterMergerNext(writer->ctx->dataIterMerger);
H
Hongze Cheng 已提交
564
      TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
565 566
    } else {
      break;
H
Hongze Cheng 已提交
567
    }
H
Hongze Cheng 已提交
568
  }
H
Hongze Cheng 已提交
569

H
Hongze Cheng 已提交
570 571 572
  if (row->suid == INT64_MAX) {
    ASSERT(writer->ctx->hasData == false);
    goto _exit;
H
Hongze Cheng 已提交
573
  }
H
Hongze Cheng 已提交
574

H
Hongze Cheng 已提交
575
  code = tsdbFSetWriteRow(writer->ctx->fsetWriter, row);
H
Hongze Cheng 已提交
576 577
  TSDB_CHECK_CODE(code, lino, _exit);

H
Hongze Cheng 已提交
578
_exit:
H
Hongze Cheng 已提交
579
  if (code) {
H
Hongze Cheng 已提交
580
    TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);
H
Hongze Cheng 已提交
581
  }
H
Hongze Cheng 已提交
582 583 584
  return code;
}

H
Hongze Cheng 已提交
585
static int32_t tsdbSnapWriteFileSetOpenReader(STsdbSnapWriter* writer) {
H
Hongze Cheng 已提交
586
  int32_t code = 0;
H
Hongze Cheng 已提交
587 588
  int32_t lino = 0;

H
Hongze Cheng 已提交
589 590
  ASSERT(writer->ctx->dataReader == NULL);
  ASSERT(TARRAY2_SIZE(writer->ctx->sttReaderArr) == 0);
H
Hongze Cheng 已提交
591

H
Hongze Cheng 已提交
592 593 594 595 596 597 598
  if (writer->ctx->fset) {
    // open data reader
    SDataFileReaderConfig dataFileReaderConfig = {
        .tsdb = writer->tsdb,
        .bufArr = writer->aBuf,
        .szPage = writer->szPage,
    };
H
Hongze Cheng 已提交
599

H
Hongze Cheng 已提交
600 601 602
    for (int32_t ftype = 0; ftype < TSDB_FTYPE_MAX; ++ftype) {
      if (writer->ctx->fset->farr[ftype] == NULL) {
        continue;
H
Hongze Cheng 已提交
603
      }
H
Hongze Cheng 已提交
604 605 606

      dataFileReaderConfig.files[ftype].exist = true;
      dataFileReaderConfig.files[ftype].file = writer->ctx->fset->farr[ftype]->f[0];
H
Hongze Cheng 已提交
607 608
    }

H
Hongze Cheng 已提交
609
    code = tsdbDataFileReaderOpen(NULL, &dataFileReaderConfig, &writer->ctx->dataReader);
H
Hongze Cheng 已提交
610
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
611

H
Hongze Cheng 已提交
612 613 614 615 616 617 618 619 620 621 622 623 624 625
    // open stt reader array
    SSttLvl* lvl;
    TARRAY2_FOREACH(writer->ctx->fset->lvlArr, lvl) {
      STFileObj* fobj;
      TARRAY2_FOREACH(lvl->fobjArr, fobj) {
        SSttFileReader*      reader;
        SSttFileReaderConfig sttFileReaderConfig = {
            .tsdb = writer->tsdb,
            .szPage = writer->szPage,
            .bufArr = writer->aBuf,
            .file = fobj->f[0],
        };

        code = tsdbSttFileReaderOpen(fobj->fname, &sttFileReaderConfig, &reader);
H
Hongze Cheng 已提交
626
        TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
627

H
Hongze Cheng 已提交
628 629
        code = TARRAY2_APPEND(writer->ctx->sttReaderArr, reader);
        TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
630 631
      }
    }
H
Hongze Cheng 已提交
632
  }
H
Hongze Cheng 已提交
633

H
Hongze Cheng 已提交
634 635
_exit:
  if (code) {
H
Hongze Cheng 已提交
636
    TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);
H
Hongze Cheng 已提交
637
  }
H
Hongze Cheng 已提交
638 639 640
  return code;
}

H
Hongze Cheng 已提交
641 642 643 644 645 646 647
static int32_t tsdbSnapWriteFileSetCloseReader(STsdbSnapWriter* writer) {
  TARRAY2_CLEAR(writer->ctx->sttReaderArr, tsdbSttFileReaderClose);
  tsdbDataFileReaderClose(&writer->ctx->dataReader);
  return 0;
}

static int32_t tsdbSnapWriteFileSetOpenIter(STsdbSnapWriter* writer) {
H
Hongze Cheng 已提交
648
  int32_t code = 0;
H
Hongze Cheng 已提交
649
  int32_t lino = 0;
H
Hongze Cheng 已提交
650

H
Hongze Cheng 已提交
651 652 653 654
  // data ieter
  if (writer->ctx->dataReader) {
    STsdbIter*      iter;
    STsdbIterConfig config = {0};
H
Hongze Cheng 已提交
655

H
Hongze Cheng 已提交
656 657 658
    // data
    config.type = TSDB_ITER_TYPE_DATA;
    config.dataReader = writer->ctx->dataReader;
H
Hongze Cheng 已提交
659

H
Hongze Cheng 已提交
660
    code = tsdbIterOpen(&config, &iter);
H
Hongze Cheng 已提交
661
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
662

H
Hongze Cheng 已提交
663
    code = TARRAY2_APPEND(writer->ctx->dataIterArr, iter);
H
Hongze Cheng 已提交
664
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
665

H
Hongze Cheng 已提交
666 667 668
    // tome
    config.type = TSDB_ITER_TYPE_DATA_TOMB;
    config.dataReader = writer->ctx->dataReader;
H
Hongze Cheng 已提交
669

H
Hongze Cheng 已提交
670 671
    code = tsdbIterOpen(&config, &iter);
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
672

H
Hongze Cheng 已提交
673 674 675
    code = TARRAY2_APPEND(writer->ctx->tombIterArr, iter);
    TSDB_CHECK_CODE(code, lino, _exit);
  }
H
Hongze Cheng 已提交
676

H
Hongze Cheng 已提交
677 678 679 680 681
  // stt iter
  SSttFileReader* sttFileReader;
  TARRAY2_FOREACH(writer->ctx->sttReaderArr, sttFileReader) {
    STsdbIter*      iter;
    STsdbIterConfig config = {0};
H
Hongze Cheng 已提交
682

H
Hongze Cheng 已提交
683 684 685
    // data
    config.type = TSDB_ITER_TYPE_STT;
    config.sttReader = sttFileReader;
H
Hongze Cheng 已提交
686

H
Hongze Cheng 已提交
687 688
    code = tsdbIterOpen(&config, &iter);
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
689

H
Hongze Cheng 已提交
690
    code = TARRAY2_APPEND(writer->ctx->dataIterArr, iter);
H
Hongze Cheng 已提交
691
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
692

H
Hongze Cheng 已提交
693 694 695
    // tomb
    config.type = TSDB_ITER_TYPE_STT_TOMB;
    config.sttReader = sttFileReader;
H
Hongze Cheng 已提交
696

H
Hongze Cheng 已提交
697 698 699 700
    code = tsdbIterOpen(&config, &iter);
    TSDB_CHECK_CODE(code, lino, _exit);

    code = TARRAY2_APPEND(writer->ctx->tombIterArr, iter);
H
Hongze Cheng 已提交
701
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
702 703
  }

H
Hongze Cheng 已提交
704 705 706 707
  // open merger
  code = tsdbIterMergerOpen(writer->ctx->dataIterArr, &writer->ctx->dataIterMerger, false);
  TSDB_CHECK_CODE(code, lino, _exit);

H
Hongze Cheng 已提交
708
  code = tsdbIterMergerOpen(writer->ctx->tombIterArr, &writer->ctx->tombIterMerger, true);
H
Hongze Cheng 已提交
709
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
710

H
Hongze Cheng 已提交
711 712
_exit:
  if (code) {
H
Hongze Cheng 已提交
713
    TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);
H
Hongze Cheng 已提交
714
  }
H
Hongze Cheng 已提交
715 716 717
  return code;
}

H
Hongze Cheng 已提交
718 719 720 721 722 723 724 725 726
static int32_t tsdbSnapWriteFileSetCloseIter(STsdbSnapWriter* writer) {
  tsdbIterMergerClose(&writer->ctx->dataIterMerger);
  tsdbIterMergerClose(&writer->ctx->tombIterMerger);
  TARRAY2_CLEAR(writer->ctx->dataIterArr, tsdbIterClose);
  TARRAY2_CLEAR(writer->ctx->tombIterArr, tsdbIterClose);
  return 0;
}

static int32_t tsdbSnapWriteFileSetOpenWriter(STsdbSnapWriter* writer) {
H
Hongze Cheng 已提交
727 728 729
  int32_t code = 0;
  int32_t lino = 0;

H
Hongze Cheng 已提交
730 731 732 733 734 735 736 737 738 739 740 741 742
  SFSetWriterConfig config = {
      .tsdb = writer->tsdb,
      .toSttOnly = false,
      .compactVersion = writer->compactVersion,
      .minRow = writer->minRow,
      .maxRow = writer->maxRow,
      .szPage = writer->szPage,
      .cmprAlg = writer->cmprAlg,
      .fid = writer->ctx->fid,
      .cid = writer->commitID,
      .did = writer->ctx->did,
      .level = 0,
  };
H
Hongze Cheng 已提交
743

H
Hongze Cheng 已提交
744
  code = tsdbFSetWriterOpen(&config, &writer->ctx->fsetWriter);
H
Hongze Cheng 已提交
745
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
746

H
Hongze Cheng 已提交
747 748
_exit:
  if (code) {
H
Hongze Cheng 已提交
749
    TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);
H
Hongze Cheng 已提交
750
  }
H
Hongze Cheng 已提交
751 752 753
  return code;
}

H
Hongze Cheng 已提交
754 755 756 757 758
static int32_t tsdbSnapWriteFileSetCloseWriter(STsdbSnapWriter* writer) {
  return tsdbFSetWriterClose(&writer->ctx->fsetWriter, 0, writer->fopArr);
}

static int32_t tsdbSnapWriteFileSetBegin(STsdbSnapWriter* writer, int32_t fid) {
H
Hongze Cheng 已提交
759
  int32_t code = 0;
H
Hongze Cheng 已提交
760
  int32_t lino = 0;
H
Hongze Cheng 已提交
761

H
Hongze Cheng 已提交
762
  ASSERT(writer->ctx->fsetWriteBegin == false);
H
Hongze Cheng 已提交
763

H
Hongze Cheng 已提交
764
  STFileSet* fset = &(STFileSet){.fid = fid};
H
Hongze Cheng 已提交
765

H
Hongze Cheng 已提交
766
  writer->ctx->fid = fid;
H
Hongze Cheng 已提交
767 768
  STFileSet** fsetPtr = TARRAY2_SEARCH(writer->fsetArr, &fset, tsdbTFileSetCmprFn, TD_EQ);
  writer->ctx->fset = (fsetPtr == NULL) ? NULL : *fsetPtr;
H
Hongze Cheng 已提交
769

H
Hongze Cheng 已提交
770
  int32_t level = tsdbFidLevel(fid, &writer->tsdb->keepCfg, taosGetTimestampSec());
H
Hongze Cheng 已提交
771 772 773
  if (tfsAllocDisk(writer->tsdb->pVnode->pTfs, level, &writer->ctx->did)) {
    code = TSDB_CODE_NO_AVAIL_DISK;
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
774
  }
H
Hongze Cheng 已提交
775
  tfsMkdirRecurAt(writer->tsdb->pVnode->pTfs, writer->tsdb->path, writer->ctx->did);
H
Hongze Cheng 已提交
776

H
Hongze Cheng 已提交
777 778 779 780 781 782 783 784 785 786 787 788 789 790
  writer->ctx->hasData = true;
  writer->ctx->hasTomb = true;

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

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

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

  writer->ctx->fsetWriteBegin = true;

H
Hongze Cheng 已提交
791 792
_exit:
  if (code) {
H
Hongze Cheng 已提交
793
    TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);
H
Hongze Cheng 已提交
794
  }
H
Hongze Cheng 已提交
795 796 797
  return code;
}

H
Hongze Cheng 已提交
798
static int32_t tsdbSnapWriteTombRecord(STsdbSnapWriter* writer, const STombRecord* record) {
H
Hongze Cheng 已提交
799
  int32_t code = 0;
H
Hongze Cheng 已提交
800
  int32_t lino = 0;
H
Hongze Cheng 已提交
801

H
Hongze Cheng 已提交
802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819
  while (writer->ctx->hasTomb) {
    STombRecord* record1 = tsdbIterMergerGetTombRecord(writer->ctx->tombIterMerger);
    if (record1 == NULL) {
      writer->ctx->hasTomb = false;
      break;
    }

    int32_t c = tTombRecordCompare(record1, record);
    if (c <= 0) {
      code = tsdbFSetWriteTombRecord(writer->ctx->fsetWriter, record1);
      TSDB_CHECK_CODE(code, lino, _exit);
    } else {
      break;
    }
  }

  if (record->suid == INT64_MAX) {
    ASSERT(writer->ctx->hasTomb == false);
H
Hongze Cheng 已提交
820
    goto _exit;
H
Hongze Cheng 已提交
821 822
  }

H
Hongze Cheng 已提交
823
  code = tsdbFSetWriteTombRecord(writer->ctx->fsetWriter, record);
H
Hongze Cheng 已提交
824
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
825

H
Hongze Cheng 已提交
826
_exit:
H
Hongze Cheng 已提交
827
  if (code) {
H
Hongze Cheng 已提交
828
    TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);
H
Hongze Cheng 已提交
829
  }
H
Hongze Cheng 已提交
830
  return code;
H
Hongze Cheng 已提交
831 832
}

H
Hongze Cheng 已提交
833 834 835
static int32_t tsdbSnapWriteFileSetEnd(STsdbSnapWriter* writer) {
  if (!writer->ctx->fsetWriteBegin) return 0;

H
Hongze Cheng 已提交
836
  int32_t code = 0;
H
Hongze Cheng 已提交
837
  int32_t lino = 0;
H
Hongze Cheng 已提交
838

H
Hongze Cheng 已提交
839 840 841 842
  SRowInfo row = {
      .suid = INT64_MAX,
      .uid = INT64_MAX,
  };
H
Hongze Cheng 已提交
843

H
Hongze Cheng 已提交
844
  code = tsdbSnapWriteTimeSeriesRow(writer, &row);
H
Hongze Cheng 已提交
845
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
846

H
Hongze Cheng 已提交
847 848 849 850
  STombRecord record = {
      .suid = INT64_MAX,
      .uid = INT64_MAX,
  };
H
Hongze Cheng 已提交
851

H
Hongze Cheng 已提交
852
  code = tsdbSnapWriteTombRecord(writer, &record);
H
Hongze Cheng 已提交
853
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
854

H
Hongze Cheng 已提交
855 856
  // close write
  code = tsdbSnapWriteFileSetCloseWriter(writer);
H
Hongze Cheng 已提交
857
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
858

H
Hongze Cheng 已提交
859
  code = tsdbSnapWriteFileSetCloseIter(writer);
H
Hongze Cheng 已提交
860
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
861

H
Hongze Cheng 已提交
862
  code = tsdbSnapWriteFileSetCloseReader(writer);
H
Hongze Cheng 已提交
863
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
864

H
Hongze Cheng 已提交
865
  writer->ctx->fsetWriteBegin = false;
H
Hongze Cheng 已提交
866

H
Hongze Cheng 已提交
867 868
_exit:
  if (code) {
H
Hongze Cheng 已提交
869
    TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);
H
Hongze Cheng 已提交
870
  }
H
Hongze Cheng 已提交
871 872 873
  return code;
}

H
Hongze Cheng 已提交
874
static int32_t tsdbSnapWriteTimeSeriesData(STsdbSnapWriter* writer, SSnapDataHdr* hdr) {
H
Hongze Cheng 已提交
875
  int32_t code = 0;
H
Hongze Cheng 已提交
876
  int32_t lino = 0;
H
Hongze Cheng 已提交
877

H
Hongze Cheng 已提交
878
  SBlockData blockData[1] = {0};
H
Hongze Cheng 已提交
879

H
Hongze Cheng 已提交
880
  code = tDecmprBlockData(hdr->data, hdr->size - sizeof(*hdr), blockData, writer->aBuf);
H
Hongze Cheng 已提交
881
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
882

H
Hongze Cheng 已提交
883 884 885 886
  int32_t fid = tsdbKeyFid(blockData->aTSKEY[0], writer->minutes, writer->precision);
  if (!writer->ctx->fsetWriteBegin || fid != writer->ctx->fid) {
    code = tsdbSnapWriteFileSetEnd(writer);
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
887

H
Hongze Cheng 已提交
888
    code = tsdbSnapWriteFileSetBegin(writer, fid);
H
Hongze Cheng 已提交
889
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
890 891
  }

H
Hongze Cheng 已提交
892 893 894 895 896 897 898 899 900
  for (int32_t i = 0; i < blockData->nRow; ++i) {
    SRowInfo rowInfo = {
        .suid = blockData->suid,
        .uid = blockData->uid ? blockData->uid : blockData->aUid[i],
        .row = tsdbRowFromBlockData(blockData, i),
    };

    code = tsdbSnapWriteTimeSeriesRow(writer, &rowInfo);
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
901 902
  }

H
Hongze Cheng 已提交
903
_exit:
H
Hongze Cheng 已提交
904
  if (code) {
H
Hongze Cheng 已提交
905
    TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);
H
Hongze Cheng 已提交
906
  } else {
H
Hongze Cheng 已提交
907 908
    tsdbDebug("vgId:%d %s done, suid:%" PRId64 " uid:%" PRId64 " nRow:%d", TD_VID(writer->tsdb->pVnode), __func__,
              blockData->suid, blockData->uid, blockData->nRow);
H
Hongze Cheng 已提交
909
  }
H
Hongze Cheng 已提交
910
  tBlockDataDestroy(blockData);
H
Hongze Cheng 已提交
911 912 913
  return code;
}

H
Hongze Cheng 已提交
914
static int32_t tsdbSnapWriteDecmprTombBlock(SSnapDataHdr* hdr, STombBlock* tombBlock) {
H
Hongze Cheng 已提交
915
  int32_t code = 0;
H
Hongze Cheng 已提交
916
  int32_t lino = 0;
H
Hongze Cheng 已提交
917

H
Hongze Cheng 已提交
918 919 920 921
  int64_t size = hdr->size - sizeof(*hdr);
  ASSERT(size % TOMB_RECORD_ELEM_NUM == 0);
  size = size / TOMB_RECORD_ELEM_NUM;
  ASSERT(size % sizeof(int64_t) == 0);
H
Hongze Cheng 已提交
922

H
Hongze Cheng 已提交
923 924 925 926 927
  int64_t* data = (int64_t*)hdr->data;
  for (int32_t i = 0; i < TOMB_RECORD_ELEM_NUM; ++i) {
    code = TARRAY2_APPEND_BATCH(&tombBlock->dataArr[i], hdr->data + i * size, size / sizeof(int64_t));
    TSDB_CHECK_CODE(code, lino, _exit);
  }
H
Hongze Cheng 已提交
928

H
Hongze Cheng 已提交
929 930 931
_exit:
  return code;
}
H
Hongze Cheng 已提交
932

H
Hongze Cheng 已提交
933 934 935
static int32_t tsdbSnapWriteTombData(STsdbSnapWriter* writer, SSnapDataHdr* hdr) {
  int32_t code = 0;
  int32_t lino = 0;
H
Hongze Cheng 已提交
936

H
Hongze Cheng 已提交
937 938
  STombRecord record;
  STombBlock  tombBlock[1] = {0};
H
Hongze Cheng 已提交
939

H
Hongze Cheng 已提交
940 941
  code = tsdbSnapWriteDecmprTombBlock(hdr, tombBlock);
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
942

H
Hongze Cheng 已提交
943 944 945 946 947
  tTombBlockGet(tombBlock, 0, &record);
  int32_t fid = tsdbKeyFid(record.skey, writer->minutes, writer->precision);
  if (!writer->ctx->fsetWriteBegin || fid != writer->ctx->fid) {
    code = tsdbSnapWriteFileSetEnd(writer);
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
948

H
Hongze Cheng 已提交
949 950 951
    code = tsdbSnapWriteFileSetBegin(writer, fid);
    TSDB_CHECK_CODE(code, lino, _exit);
  }
H
Hongze Cheng 已提交
952

H
Hongze Cheng 已提交
953 954 955 956 957
  if (writer->ctx->hasData) {
    SRowInfo row = {
        .suid = INT64_MAX,
        .uid = INT64_MAX,
    };
H
Hongze Cheng 已提交
958

H
Hongze Cheng 已提交
959
    code = tsdbSnapWriteTimeSeriesRow(writer, &row);
H
Hongze Cheng 已提交
960
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
961 962
  }

H
Hongze Cheng 已提交
963
  ASSERT(writer->ctx->hasData == false);
H
Hongze Cheng 已提交
964

H
Hongze Cheng 已提交
965 966
  for (int32_t i = 0; i < TOMB_BLOCK_SIZE(tombBlock); ++i) {
    tTombBlockGet(tombBlock, i, &record);
H
Hongze Cheng 已提交
967

H
Hongze Cheng 已提交
968
    code = tsdbSnapWriteTombRecord(writer, &record);
H
Hongze Cheng 已提交
969 970 971
    TSDB_CHECK_CODE(code, lino, _exit);
  }

H
Hongze Cheng 已提交
972
  tTombBlockDestroy(tombBlock);
H
Hongze Cheng 已提交
973 974 975

_exit:
  if (code) {
H
Hongze Cheng 已提交
976
    TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);
H
Hongze Cheng 已提交
977
  }
H
Hongze Cheng 已提交
978 979 980
  return code;
}

H
Hongze Cheng 已提交
981
int32_t tsdbSnapWriterOpen(STsdb* pTsdb, int64_t sver, int64_t ever, STsdbSnapWriter** writer) {
H
Hongze Cheng 已提交
982 983
  int32_t code = 0;
  int32_t lino = 0;
H
Hongze Cheng 已提交
984

H
Hongze Cheng 已提交
985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000
  writer[0] = taosMemoryCalloc(1, sizeof(*writer[0]));
  if (writer[0] == NULL) return TSDB_CODE_OUT_OF_MEMORY;

  writer[0]->tsdb = pTsdb;
  writer[0]->sver = sver;
  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]->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);
H
Hongze Cheng 已提交
1001
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
1002

H
Hongze Cheng 已提交
1003 1004
_exit:
  if (code) {
H
Hongze Cheng 已提交
1005
    tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
H
Hongze Cheng 已提交
1006
  } else {
H
Hongze Cheng 已提交
1007
    tsdbInfo("vgId:%d %s done, sver:%" PRId64 " ever:%" PRId64, TD_VID(pTsdb->pVnode), __func__, sver, ever);
H
Hongze Cheng 已提交
1008
  }
H
Hongze Cheng 已提交
1009 1010 1011
  return code;
}

H
Hongze Cheng 已提交
1012
int32_t tsdbSnapWriterPrepareClose(STsdbSnapWriter* writer) {
H
Hongze Cheng 已提交
1013
  int32_t code = 0;
H
Hongze Cheng 已提交
1014 1015
  int32_t lino = 0;

H
Hongze Cheng 已提交
1016 1017
  code = tsdbSnapWriteFileSetEnd(writer);
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
1018

H
Hongze Cheng 已提交
1019
  code = tsdbFSEditBegin(writer->tsdb->pFS, writer->fopArr, TSDB_FEDIT_COMMIT);
H
Hongze Cheng 已提交
1020
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
1021 1022 1023

_exit:
  if (code) {
H
Hongze Cheng 已提交
1024
    TSDB_ERROR_LOG(TD_VID(writer->tsdb->pVnode), lino, code);
H
Hongze Cheng 已提交
1025
  } else {
H
Hongze Cheng 已提交
1026
    tsdbDebug("vgId:%d %s done", TD_VID(writer->tsdb->pVnode), __func__);
H
Hongze Cheng 已提交
1027 1028 1029 1030
  }
  return code;
}

H
Hongze Cheng 已提交
1031
int32_t tsdbSnapWriterClose(STsdbSnapWriter** writer, int8_t rollback) {
H
Hongze Cheng 已提交
1032 1033
  if (writer[0] == NULL) return 0;

H
Hongze Cheng 已提交
1034 1035 1036
  int32_t code = 0;
  int32_t lino = 0;

H
Hongze Cheng 已提交
1037
  STsdb* tsdb = writer[0]->tsdb;
H
Hongze Cheng 已提交
1038 1039

  if (rollback) {
H
Hongze Cheng 已提交
1040 1041
    code = tsdbFSEditAbort(writer[0]->tsdb->pFS);
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
1042
  } else {
H
Hongze Cheng 已提交
1043
    taosThreadRwlockWrlock(&writer[0]->tsdb->rwLock);
H
Hongze Cheng 已提交
1044

H
Hongze Cheng 已提交
1045
    code = tsdbFSEditCommit(writer[0]->tsdb->pFS);
H
Hongze Cheng 已提交
1046
    if (code) {
H
Hongze Cheng 已提交
1047
      taosThreadRwlockUnlock(&writer[0]->tsdb->rwLock);
H
Hongze Cheng 已提交
1048 1049 1050
      TSDB_CHECK_CODE(code, lino, _exit);
    }

H
Hongze Cheng 已提交
1051
    taosThreadRwlockUnlock(&writer[0]->tsdb->rwLock);
H
Hongze Cheng 已提交
1052 1053
  }

H
Hongze Cheng 已提交
1054 1055 1056 1057 1058 1059
  tsdbIterMergerClose(&writer[0]->ctx->tombIterMerger);
  tsdbIterMergerClose(&writer[0]->ctx->dataIterMerger);
  TARRAY2_DESTROY(writer[0]->ctx->tombIterArr, tsdbIterClose);
  TARRAY2_DESTROY(writer[0]->ctx->dataIterArr, tsdbIterClose);
  TARRAY2_DESTROY(writer[0]->ctx->sttReaderArr, tsdbSttFileReaderClose);
  tsdbDataFileReaderClose(&writer[0]->ctx->dataReader);
H
Hongze Cheng 已提交
1060

H
Hongze Cheng 已提交
1061 1062
  TARRAY2_DESTROY(writer[0]->fopArr, NULL);
  tsdbFSDestroyCopySnapshot(&writer[0]->fsetArr);
H
Hongze Cheng 已提交
1063

H
Hongze Cheng 已提交
1064 1065
  for (int32_t i = 0; i < ARRAY_SIZE(writer[0]->aBuf); ++i) {
    tFree(writer[0]->aBuf[i]);
H
Hongze Cheng 已提交
1066 1067
  }

H
Hongze Cheng 已提交
1068 1069
  taosMemoryFree(writer[0]);
  writer[0] = NULL;
H
Hongze Cheng 已提交
1070 1071 1072

_exit:
  if (code) {
H
Hongze Cheng 已提交
1073
    TSDB_ERROR_LOG(TD_VID(tsdb->pVnode), lino, code);
H
Hongze Cheng 已提交
1074
  } else {
H
Hongze Cheng 已提交
1075
    tsdbInfo("vgId:%d %s done", TD_VID(tsdb->pVnode), __func__);
H
Hongze Cheng 已提交
1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091
  }
  return code;
}

int32_t tsdbSnapWrite(STsdbSnapWriter* writer, SSnapDataHdr* hdr) {
  int32_t code = 0;
  int32_t lino = 0;

  if (hdr->type == SNAP_DATA_TSDB) {
    code = tsdbSnapWriteTimeSeriesData(writer, hdr);
    TSDB_CHECK_CODE(code, lino, _exit);
  } else if (hdr->type == SNAP_DATA_DEL) {
    code = tsdbSnapWriteTombData(writer, hdr);
    TSDB_CHECK_CODE(code, lino, _exit);
  } else {
    ASSERT(0);
H
Hongze Cheng 已提交
1092 1093
  }

H
Hongze Cheng 已提交
1094
_exit:
H
Hongze Cheng 已提交
1095 1096
  if (code) {
    tsdbError("vgId:%d %s failed at line %d since %s, type:%d index:%" PRId64 " size:%" PRId64,
H
Hongze Cheng 已提交
1097
              TD_VID(writer->tsdb->pVnode), __func__, lino, tstrerror(code), hdr->type, hdr->index, hdr->size);
H
Hongze Cheng 已提交
1098
  } else {
H
Hongze Cheng 已提交
1099 1100
    tsdbDebug("vgId:%d %s done, type:%d index:%" PRId64 " size:%" PRId64, TD_VID(writer->tsdb->pVnode), __func__,
              hdr->type, hdr->index, hdr->size);
H
Hongze Cheng 已提交
1101
  }
H
Hongze Cheng 已提交
1102 1103
  return code;
}