tsdbSnapshot.c 35.4 KB
Newer Older
H
Hongze Cheng 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
/*
 * 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 已提交
18
// STsdbSnapReader ========================================
H
Hongze Cheng 已提交
19
struct STsdbSnapReader {
H
more  
Hongze Cheng 已提交
20 21 22
  STsdb*  pTsdb;
  int64_t sver;
  int64_t ever;
H
Hongze Cheng 已提交
23
  STsdbFS fs;
C
Cary Xu 已提交
24
  int8_t  type;
H
more  
Hongze Cheng 已提交
25
  // for data file
H
Hongze Cheng 已提交
26 27
  int8_t        dataDone;
  int32_t       fid;
H
more  
Hongze Cheng 已提交
28
  SDataFReader* pDataFReader;
H
Hongze Cheng 已提交
29
  SArray*       aBlockIdx;  // SArray<SBlockIdx>
H
Hongze Cheng 已提交
30 31
  int32_t       iBlockIdx;
  SBlockIdx*    pBlockIdx;
H
Hongze Cheng 已提交
32
  SMapData      mBlock;  // SMapData<SBlock>
H
Hongze Cheng 已提交
33
  int32_t       iBlock;
H
Hongze Cheng 已提交
34 35
  SBlockData    oBlockData;
  SBlockData    nBlockData;
H
more  
Hongze Cheng 已提交
36
  // for del file
H
Hongze Cheng 已提交
37
  int8_t       delDone;
H
more  
Hongze Cheng 已提交
38
  SDelFReader* pDelFReader;
H
Hongze Cheng 已提交
39
  SArray*      aDelIdx;  // SArray<SDelIdx>
H
Hongze Cheng 已提交
40 41
  int32_t      iDelIdx;
  SArray*      aDelData;  // SArray<SDelData>
H
Hongze Cheng 已提交
42 43
};

H
Hongze Cheng 已提交
44 45
static int32_t tsdbSnapReadData(STsdbSnapReader* pReader, uint8_t** ppData) {
  int32_t code = 0;
H
Hongze Cheng 已提交
46
  STsdb*  pTsdb = pReader->pTsdb;
H
Hongze Cheng 已提交
47

H
Hongze Cheng 已提交
48 49
  while (true) {
    if (pReader->pDataFReader == NULL) {
H
Hongze Cheng 已提交
50 51
      SDFileSet* pSet =
          taosArraySearch(pReader->fs.aDFileSet, &(SDFileSet){.fid = pReader->fid}, tDFileSetCmprFn, TD_GT);
H
Hongze Cheng 已提交
52

H
Hongze Cheng 已提交
53
      if (pSet == NULL) goto _exit;
H
Hongze Cheng 已提交
54

H
Hongze Cheng 已提交
55
      pReader->fid = pSet->fid;
H
Hongze Cheng 已提交
56 57 58 59 60 61
      code = tsdbDataFReaderOpen(&pReader->pDataFReader, pReader->pTsdb, pSet);
      if (code) goto _err;

      // SBlockIdx
      code = tsdbReadBlockIdx(pReader->pDataFReader, pReader->aBlockIdx, NULL);
      if (code) goto _err;
H
Hongze Cheng 已提交
62

H
Hongze Cheng 已提交
63 64
      pReader->iBlockIdx = 0;
      pReader->pBlockIdx = NULL;
H
Hongze Cheng 已提交
65

S
Shengliang Guan 已提交
66
      tsdbInfo("vgId:%d, vnode snapshot tsdb open data file to read for %s, fid:%d", TD_VID(pTsdb->pVnode), pTsdb->path,
C
Cary Xu 已提交
67
               pReader->fid);
H
Hongze Cheng 已提交
68
    }
H
Hongze Cheng 已提交
69

H
Hongze Cheng 已提交
70 71 72 73 74 75
    while (true) {
      if (pReader->pBlockIdx == NULL) {
        if (pReader->iBlockIdx >= taosArrayGetSize(pReader->aBlockIdx)) {
          tsdbDataFReaderClose(&pReader->pDataFReader);
          break;
        }
H
Hongze Cheng 已提交
76

H
Hongze Cheng 已提交
77 78
        pReader->pBlockIdx = (SBlockIdx*)taosArrayGet(pReader->aBlockIdx, pReader->iBlockIdx);
        pReader->iBlockIdx++;
H
Hongze Cheng 已提交
79

H
Hongze Cheng 已提交
80 81 82 83 84 85
        code = tsdbReadBlock(pReader->pDataFReader, pReader->pBlockIdx, &pReader->mBlock, NULL);
        if (code) goto _err;

        pReader->iBlock = 0;
      }

H
Hongze Cheng 已提交
86 87
      SBlock  block;
      SBlock* pBlock = &block;
H
Hongze Cheng 已提交
88 89 90 91 92 93 94 95
      while (true) {
        if (pReader->iBlock >= pReader->mBlock.nItem) {
          pReader->pBlockIdx = NULL;
          break;
        }

        tMapDataGetItemByIdx(&pReader->mBlock, pReader->iBlock, pBlock, tGetBlock);
        pReader->iBlock++;
H
Hongze Cheng 已提交
96

H
Hongze Cheng 已提交
97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118
        if (pBlock->minVersion > pReader->ever || pBlock->maxVersion < pReader->sver) continue;

        code = tsdbReadBlockData(pReader->pDataFReader, pReader->pBlockIdx, pBlock, &pReader->oBlockData, NULL, NULL);
        if (code) goto _err;

        // filter
        tBlockDataReset(&pReader->nBlockData);
        for (int32_t iColData = 0; iColData < taosArrayGetSize(pReader->oBlockData.aIdx); iColData++) {
          SColData* pColDataO = tBlockDataGetColDataByIdx(&pReader->oBlockData, iColData);
          SColData* pColDataN = NULL;

          code = tBlockDataAddColData(&pReader->nBlockData, taosArrayGetSize(pReader->nBlockData.aIdx), &pColDataN);
          if (code) goto _err;

          tColDataInit(pColDataN, pColDataO->cid, pColDataO->type, pColDataO->smaOn);
        }

        for (int32_t iRow = 0; iRow < pReader->oBlockData.nRow; iRow++) {
          TSDBROW row = tsdbRowFromBlockData(&pReader->oBlockData, iRow);
          int64_t version = TSDBROW_VERSION(&row);

          if (version < pReader->sver || version > pReader->ever) continue;
H
Hongze Cheng 已提交
119

H
Hongze Cheng 已提交
120
          code = tBlockDataAppendRow(&pReader->nBlockData, &row, NULL);
H
Hongze Cheng 已提交
121
          if (code) goto _err;
H
Hongze Cheng 已提交
122
        }
H
Hongze Cheng 已提交
123

H
Hongze Cheng 已提交
124 125
        // org data
        // compress data (todo)
H
Hongze Cheng 已提交
126 127 128 129 130 131
        int32_t size = sizeof(TABLEID) + tPutBlockData(NULL, &pReader->nBlockData);

        *ppData = taosMemoryMalloc(sizeof(SSnapDataHdr) + size);
        if (*ppData == NULL) {
          code = TSDB_CODE_OUT_OF_MEMORY;
          goto _err;
H
Hongze Cheng 已提交
132
        }
H
Hongze Cheng 已提交
133 134

        SSnapDataHdr* pHdr = (SSnapDataHdr*)(*ppData);
C
Cary Xu 已提交
135
        pHdr->type = pReader->type;
H
Hongze Cheng 已提交
136 137 138 139 140 141 142 143
        pHdr->size = size;

        TABLEID* pId = (TABLEID*)(&pHdr[1]);
        pId->suid = pReader->pBlockIdx->suid;
        pId->uid = pReader->pBlockIdx->uid;

        tPutBlockData((uint8_t*)(&pId[1]), &pReader->nBlockData);

S
Shengliang Guan 已提交
144
        tsdbInfo("vgId:%d, vnode snapshot read data for %s, fid:%d suid:%" PRId64 " uid:%" PRId64
H
Hongze Cheng 已提交
145
                 " iBlock:%d minVersion:%d maxVersion:%d nRow:%d out of %d size:%d",
C
Cary Xu 已提交
146
                 TD_VID(pTsdb->pVnode), pTsdb->path, pReader->fid, pReader->pBlockIdx->suid, pReader->pBlockIdx->uid,
H
Hongze Cheng 已提交
147 148
                 pReader->iBlock - 1, pBlock->minVersion, pBlock->maxVersion, pReader->nBlockData.nRow, pBlock->nRow,
                 size);
H
Hongze Cheng 已提交
149 150

        goto _exit;
H
Hongze Cheng 已提交
151
      }
H
Hongze Cheng 已提交
152 153 154 155
    }
  }

_exit:
H
Hongze Cheng 已提交
156 157 158
  return code;

_err:
S
Shengliang Guan 已提交
159
  tsdbError("vgId:%d, vnode snapshot tsdb read data for %s failed since %s", TD_VID(pTsdb->pVnode), pTsdb->path,
C
Cary Xu 已提交
160
            tstrerror(code));
H
Hongze Cheng 已提交
161 162 163 164 165 166
  return code;
}

static int32_t tsdbSnapReadDel(STsdbSnapReader* pReader, uint8_t** ppData) {
  int32_t   code = 0;
  STsdb*    pTsdb = pReader->pTsdb;
H
Hongze Cheng 已提交
167
  SDelFile* pDelFile = pReader->fs.pDelFile;
H
Hongze Cheng 已提交
168 169 170 171 172 173 174 175 176 177

  if (pReader->pDelFReader == NULL) {
    if (pDelFile == NULL) {
      goto _exit;
    }

    // open
    code = tsdbDelFReaderOpen(&pReader->pDelFReader, pDelFile, pTsdb, NULL);
    if (code) goto _err;

H
Hongze Cheng 已提交
178
    // read index
H
Hongze Cheng 已提交
179 180 181 182 183 184
    code = tsdbReadDelIdx(pReader->pDelFReader, pReader->aDelIdx, NULL);
    if (code) goto _err;

    pReader->iDelIdx = 0;
  }

H
Hongze Cheng 已提交
185 186 187 188 189 190
  while (true) {
    if (pReader->iDelIdx >= taosArrayGetSize(pReader->aDelIdx)) {
      tsdbDelFReaderClose(&pReader->pDelFReader);
      break;
    }

H
Hongze Cheng 已提交
191 192
    SDelIdx* pDelIdx = (SDelIdx*)taosArrayGet(pReader->aDelIdx, pReader->iDelIdx);

H
Hongze Cheng 已提交
193 194
    pReader->iDelIdx++;

H
Hongze Cheng 已提交
195 196 197
    code = tsdbReadDelData(pReader->pDelFReader, pDelIdx, pReader->aDelData, NULL);
    if (code) goto _err;

H
Hongze Cheng 已提交
198
    int32_t size = 0;
H
Hongze Cheng 已提交
199 200 201 202
    for (int32_t iDelData = 0; iDelData < taosArrayGetSize(pReader->aDelData); iDelData++) {
      SDelData* pDelData = (SDelData*)taosArrayGet(pReader->aDelData, iDelData);

      if (pDelData->version >= pReader->sver && pDelData->version <= pReader->ever) {
H
Hongze Cheng 已提交
203
        size += tPutDelData(NULL, pDelData);
H
Hongze Cheng 已提交
204 205 206
      }
    }

H
Hongze Cheng 已提交
207
    if (size == 0) continue;
H
Hongze Cheng 已提交
208

H
Hongze Cheng 已提交
209 210 211 212 213 214 215
    // org data
    size = sizeof(TABLEID) + size;
    *ppData = taosMemoryMalloc(sizeof(SSnapDataHdr) + size);
    if (*ppData == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
    }
H
Hongze Cheng 已提交
216

H
Hongze Cheng 已提交
217
    SSnapDataHdr* pHdr = (SSnapDataHdr*)(*ppData);
C
Cary Xu 已提交
218
    pHdr->type = SNAP_DATA_DEL;
H
Hongze Cheng 已提交
219
    pHdr->size = size;
H
Hongze Cheng 已提交
220

H
Hongze Cheng 已提交
221 222 223 224 225 226
    TABLEID* pId = (TABLEID*)(&pHdr[1]);
    pId->suid = pDelIdx->suid;
    pId->uid = pDelIdx->uid;
    int32_t n = sizeof(SSnapDataHdr) + sizeof(TABLEID);
    for (int32_t iDelData = 0; iDelData < taosArrayGetSize(pReader->aDelData); iDelData++) {
      SDelData* pDelData = (SDelData*)taosArrayGet(pReader->aDelData, iDelData);
H
Hongze Cheng 已提交
227

H
Hongze Cheng 已提交
228 229
      if (pDelData->version < pReader->sver) continue;
      if (pDelData->version > pReader->ever) continue;
H
Hongze Cheng 已提交
230

H
Hongze Cheng 已提交
231
      n += tPutDelData((*ppData) + n, pDelData);
H
Hongze Cheng 已提交
232 233
    }

S
Shengliang Guan 已提交
234
    tsdbInfo("vgId:%d, vnode snapshot tsdb read del data for %s, suid:%" PRId64 " uid:%d" PRId64 " size:%d",
C
Cary Xu 已提交
235
             TD_VID(pTsdb->pVnode), pTsdb->path, pDelIdx->suid, pDelIdx->uid, size);
H
Hongze Cheng 已提交
236 237 238

    break;
  }
H
Hongze Cheng 已提交
239 240 241 242 243

_exit:
  return code;

_err:
S
Shengliang Guan 已提交
244
  tsdbError("vgId:%d, vnode snapshot tsdb read del for %s failed since %s", TD_VID(pTsdb->pVnode), pTsdb->pVnode,
C
Cary Xu 已提交
245
            tstrerror(code));
H
Hongze Cheng 已提交
246 247
  return code;
}
H
more  
Hongze Cheng 已提交
248

C
Cary Xu 已提交
249
int32_t tsdbSnapReaderOpen(STsdb* pTsdb, int64_t sver, int64_t ever, int8_t type, STsdbSnapReader** ppReader) {
H
Hongze Cheng 已提交
250 251
  int32_t          code = 0;
  STsdbSnapReader* pReader = NULL;
H
Hongze Cheng 已提交
252

H
more  
Hongze Cheng 已提交
253
  // alloc
H
Hongze Cheng 已提交
254
  pReader = (STsdbSnapReader*)taosMemoryCalloc(1, sizeof(*pReader));
H
more  
Hongze Cheng 已提交
255 256 257 258 259 260 261
  if (pReader == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
  pReader->pTsdb = pTsdb;
  pReader->sver = sver;
  pReader->ever = ever;
C
Cary Xu 已提交
262
  pReader->type = type;
H
more  
Hongze Cheng 已提交
263

H
Hongze Cheng 已提交
264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281
  code = taosThreadRwlockRdlock(&pTsdb->rwLock);
  if (code) {
    code = TAOS_SYSTEM_ERROR(code);
    goto _err;
  }

  code = tsdbFSRef(pTsdb, &pReader->fs);
  if (code) {
    taosThreadRwlockUnlock(&pTsdb->rwLock);
    goto _err;
  }

  code = taosThreadRwlockUnlock(&pTsdb->rwLock);
  if (code) {
    code = TAOS_SYSTEM_ERROR(code);
    goto _err;
  }

H
Hongze Cheng 已提交
282
  pReader->fid = INT32_MIN;
H
Hongze Cheng 已提交
283 284 285 286 287 288
  pReader->aBlockIdx = taosArrayInit(0, sizeof(SBlockIdx));
  if (pReader->aBlockIdx == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
  pReader->mBlock = tMapDataInit();
H
Hongze Cheng 已提交
289 290 291
  code = tBlockDataInit(&pReader->oBlockData);
  if (code) goto _err;
  code = tBlockDataInit(&pReader->nBlockData);
H
Hongze Cheng 已提交
292 293 294 295 296 297 298 299 300 301 302 303 304
  if (code) goto _err;

  pReader->aDelIdx = taosArrayInit(0, sizeof(SDelIdx));
  if (pReader->aDelIdx == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
  pReader->aDelData = taosArrayInit(0, sizeof(SDelData));
  if (pReader->aDelData == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }

S
Shengliang Guan 已提交
305
  tsdbInfo("vgId:%d, vnode snapshot tsdb reader opened for %s", TD_VID(pTsdb->pVnode), pTsdb->path);
H
more  
Hongze Cheng 已提交
306 307 308 309
  *ppReader = pReader;
  return code;

_err:
S
Shengliang Guan 已提交
310
  tsdbError("vgId:%d, vnode snapshot tsdb reader open for %s failed since %s", TD_VID(pTsdb->pVnode), pTsdb->path,
C
Cary Xu 已提交
311
            tstrerror(code));
H
more  
Hongze Cheng 已提交
312 313
  *ppReader = NULL;
  return code;
H
Hongze Cheng 已提交
314 315
}

H
Hongze Cheng 已提交
316
int32_t tsdbSnapReaderClose(STsdbSnapReader** ppReader) {
H
Hongze Cheng 已提交
317 318 319 320 321 322
  int32_t          code = 0;
  STsdbSnapReader* pReader = *ppReader;

  if (pReader->pDataFReader) {
    tsdbDataFReaderClose(&pReader->pDataFReader);
  }
H
Hongze Cheng 已提交
323 324
  taosArrayDestroy(pReader->aBlockIdx);
  tMapDataClear(&pReader->mBlock);
H
Hongze Cheng 已提交
325 326
  tBlockDataClear(&pReader->oBlockData, 1);
  tBlockDataClear(&pReader->nBlockData, 1);
H
Hongze Cheng 已提交
327 328 329 330 331 332 333

  if (pReader->pDelFReader) {
    tsdbDelFReaderClose(&pReader->pDelFReader);
  }
  taosArrayDestroy(pReader->aDelIdx);
  taosArrayDestroy(pReader->aDelData);

H
Hongze Cheng 已提交
334 335
  tsdbFSUnref(pReader->pTsdb, &pReader->fs);

S
Shengliang Guan 已提交
336
  tsdbInfo("vgId:%d, vnode snapshot tsdb reader closed for %s", TD_VID(pReader->pTsdb->pVnode), pReader->pTsdb->path);
H
Hongze Cheng 已提交
337

H
Hongze Cheng 已提交
338 339
  taosMemoryFree(pReader);
  *ppReader = NULL;
H
Hongze Cheng 已提交
340 341 342 343
  return code;
}

int32_t tsdbSnapRead(STsdbSnapReader* pReader, uint8_t** ppData) {
H
more  
Hongze Cheng 已提交
344
  int32_t code = 0;
H
Hongze Cheng 已提交
345

H
Hongze Cheng 已提交
346 347
  *ppData = NULL;

H
Hongze Cheng 已提交
348
  // read data file
H
Hongze Cheng 已提交
349 350 351
  if (!pReader->dataDone) {
    code = tsdbSnapReadData(pReader, ppData);
    if (code) {
H
Hongze Cheng 已提交
352 353 354 355
      goto _err;
    } else {
      if (*ppData) {
        goto _exit;
H
Hongze Cheng 已提交
356
      } else {
H
Hongze Cheng 已提交
357
        pReader->dataDone = 1;
H
Hongze Cheng 已提交
358 359 360
      }
    }
  }
H
Hongze Cheng 已提交
361 362

  // read del file
H
Hongze Cheng 已提交
363 364 365
  if (!pReader->delDone) {
    code = tsdbSnapReadDel(pReader, ppData);
    if (code) {
H
Hongze Cheng 已提交
366 367 368 369
      goto _err;
    } else {
      if (*ppData) {
        goto _exit;
H
Hongze Cheng 已提交
370
      } else {
H
Hongze Cheng 已提交
371
        pReader->delDone = 1;
H
Hongze Cheng 已提交
372 373 374
      }
    }
  }
H
Hongze Cheng 已提交
375

H
Hongze Cheng 已提交
376
_exit:
S
Shengliang Guan 已提交
377
  tsdbDebug("vgId:%d, vnode snapshot tsdb read for %s", TD_VID(pReader->pTsdb->pVnode), pReader->pTsdb->path);
H
more  
Hongze Cheng 已提交
378 379 380
  return code;

_err:
S
Shengliang Guan 已提交
381
  tsdbError("vgId:%d, vnode snapshot tsdb read for %s failed since %s", TD_VID(pReader->pTsdb->pVnode),
C
Cary Xu 已提交
382
            pReader->pTsdb->path, tstrerror(code));
H
more  
Hongze Cheng 已提交
383 384 385
  return code;
}

H
Hongze Cheng 已提交
386
// STsdbSnapWriter ========================================
H
Hongze Cheng 已提交
387 388 389 390
struct STsdbSnapWriter {
  STsdb*  pTsdb;
  int64_t sver;
  int64_t ever;
H
Hongze Cheng 已提交
391
  STsdbFS fs;
H
Hongze Cheng 已提交
392

H
Hongze Cheng 已提交
393 394 395
  // config
  int32_t minutes;
  int8_t  precision;
H
Hongze Cheng 已提交
396 397 398
  int32_t minRow;
  int32_t maxRow;
  int8_t  cmprAlg;
H
Hongze Cheng 已提交
399
  int64_t commitID;
H
Hongze Cheng 已提交
400

H
Hongze Cheng 已提交
401
  // for data file
H
Hongze Cheng 已提交
402 403
  SBlockData bData;

H
Hongze Cheng 已提交
404
  int32_t       fid;
H
Hongze Cheng 已提交
405
  SDataFReader* pDataFReader;
H
Hongze Cheng 已提交
406
  SArray*       aBlockIdx;  // SArray<SBlockIdx>
H
Hongze Cheng 已提交
407
  int32_t       iBlockIdx;
H
Hongze Cheng 已提交
408
  SBlockIdx*    pBlockIdx;
H
Hongze Cheng 已提交
409
  SMapData      mBlock;  // SMapData<SBlock>
H
Hongze Cheng 已提交
410
  int32_t       iBlock;
H
Hongze Cheng 已提交
411
  SBlockData*   pBlockData;
H
Hongze Cheng 已提交
412
  int32_t       iRow;
H
Hongze Cheng 已提交
413
  SBlockData    bDataR;
H
Hongze Cheng 已提交
414

H
Hongze Cheng 已提交
415
  SDataFWriter* pDataFWriter;
H
Hongze Cheng 已提交
416
  SBlockIdx*    pBlockIdxW;  // NULL when no committing table
H
Hongze Cheng 已提交
417 418
  SBlock        blockW;
  SBlockData    bDataW;
H
Hongze Cheng 已提交
419
  SBlockIdx     blockIdxW;
H
Hongze Cheng 已提交
420

H
Hongze Cheng 已提交
421 422 423
  SMapData mBlockW;     // SMapData<SBlock>
  SArray*  aBlockIdxW;  // SArray<SBlockIdx>

H
Hongze Cheng 已提交
424
  // for del file
H
Hongze Cheng 已提交
425
  SDelFReader* pDelFReader;
H
Hongze Cheng 已提交
426
  SDelFWriter* pDelFWriter;
H
Hongze Cheng 已提交
427
  int32_t      iDelIdx;
H
Hongze Cheng 已提交
428
  SArray*      aDelIdxR;
H
Hongze Cheng 已提交
429
  SArray*      aDelData;
H
Hongze Cheng 已提交
430
  SArray*      aDelIdxW;
H
Hongze Cheng 已提交
431 432
};

H
Hongze Cheng 已提交
433 434 435 436 437 438 439
static int32_t tsdbSnapWriteAppendData(STsdbSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
  int32_t     code = 0;
  int32_t     iRow = 0;           // todo
  int32_t     nRow = 0;           // todo
  SBlockData* pBlockData = NULL;  // todo

  while (iRow < nRow) {
H
Hongze Cheng 已提交
440
    code = tBlockDataAppendRow(&pWriter->bDataW, &tsdbRowFromBlockData(pBlockData, iRow), NULL);
H
Hongze Cheng 已提交
441 442 443 444 445 446
    if (code) goto _err;
  }

  return code;

_err:
S
Shengliang Guan 已提交
447
  tsdbError("vgId:%d, tsdb snapshot write append data for %s failed since %s", TD_VID(pWriter->pTsdb->pVnode),
C
Cary Xu 已提交
448
            pWriter->pTsdb->path, tstrerror(code));
H
Hongze Cheng 已提交
449 450 451
  return code;
}

H
Hongze Cheng 已提交
452
static int32_t tsdbSnapWriteTableDataEnd(STsdbSnapWriter* pWriter) {
H
more  
Hongze Cheng 已提交
453 454
  int32_t code = 0;

H
Hongze Cheng 已提交
455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481
  ASSERT(pWriter->pDataFWriter);

  if (pWriter->pBlockIdxW == NULL) goto _exit;

  // consume remain rows
  if (pWriter->pBlockData) {
    ASSERT(pWriter->iRow < pWriter->pBlockData->nRow);
    while (pWriter->iRow < pWriter->pBlockData->nRow) {
      code = tBlockDataAppendRow(&pWriter->bDataW, &tsdbRowFromBlockData(pWriter->pBlockData, pWriter->iRow), NULL);
      if (code) goto _err;

      if (pWriter->bDataW.nRow >= pWriter->maxRow * 4 / 5) {
        pWriter->blockW.last = 0;
        code = tsdbWriteBlockData(pWriter->pDataFWriter, &pWriter->bDataW, NULL, NULL, pWriter->pBlockIdxW,
                                  &pWriter->blockW, pWriter->cmprAlg);
        if (code) goto _err;

        code = tMapDataPutItem(&pWriter->mBlockW, &pWriter->blockW, tPutBlock);
        if (code) goto _err;

        tBlockReset(&pWriter->blockW);
        tBlockDataClearData(&pWriter->bDataW);
      }

      pWriter->iRow++;
    }
  }
H
Hongze Cheng 已提交
482

H
Hongze Cheng 已提交
483 484
  // write remain data if has
  if (pWriter->bDataW.nRow > 0) {
H
Hongze Cheng 已提交
485 486 487 488 489
    pWriter->blockW.last = 0;
    if (pWriter->bDataW.nRow < pWriter->minRow) {
      if (pWriter->iBlock > pWriter->mBlock.nItem) {
        pWriter->blockW.last = 1;
      }
H
Hongze Cheng 已提交
490 491
    }

H
Hongze Cheng 已提交
492 493 494
    code = tsdbWriteBlockData(pWriter->pDataFWriter, &pWriter->bDataW, NULL, NULL, pWriter->pBlockIdxW,
                              &pWriter->blockW, pWriter->cmprAlg);
    if (code) goto _err;
H
Hongze Cheng 已提交
495

H
Hongze Cheng 已提交
496 497 498
    code = tMapDataPutItem(&pWriter->mBlockW, &pWriter->blockW, tPutBlock);
    if (code) goto _err;
  }
H
Hongze Cheng 已提交
499

H
Hongze Cheng 已提交
500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523
  while (true) {
    if (pWriter->iBlock >= pWriter->mBlock.nItem) break;

    SBlock block;
    tMapDataGetItemByIdx(&pWriter->mBlock, pWriter->iBlock, &block, tGetBlock);

    if (block.last) {
      code = tsdbReadBlockData(pWriter->pDataFReader, pWriter->pBlockIdx, &block, &pWriter->bDataR, NULL, NULL);
      if (code) goto _err;

      tBlockReset(&block);
      block.last = 1;
      code = tsdbWriteBlockData(pWriter->pDataFWriter, &pWriter->bDataR, NULL, NULL, pWriter->pBlockIdxW, &block,
                                pWriter->cmprAlg);
      if (code) goto _err;
    }

    code = tMapDataPutItem(&pWriter->mBlockW, &block, tPutBlock);
    if (code) goto _err;

    pWriter->iBlock++;
  }

  // SBlock
H
Hongze Cheng 已提交
524 525
  code = tsdbWriteBlock(pWriter->pDataFWriter, &pWriter->mBlockW, NULL, pWriter->pBlockIdxW);
  if (code) goto _err;
H
Hongze Cheng 已提交
526

H
Hongze Cheng 已提交
527
  // SBlockIdx
H
Hongze Cheng 已提交
528 529 530
  if (taosArrayPush(pWriter->aBlockIdxW, pWriter->pBlockIdxW) == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
H
Hongze Cheng 已提交
531 532
  }

H
Hongze Cheng 已提交
533
_exit:
S
Shengliang Guan 已提交
534
  tsdbInfo("vgId:%d, tsdb snapshot write table data end for %s", TD_VID(pWriter->pTsdb->pVnode), pWriter->pTsdb->path);
H
Hongze Cheng 已提交
535 536 537
  return code;

_err:
S
Shengliang Guan 已提交
538
  tsdbError("vgId:%d, tsdb snapshot write table data end for %s failed since %s", TD_VID(pWriter->pTsdb->pVnode),
C
Cary Xu 已提交
539
            pWriter->pTsdb->path, tstrerror(code));
H
Hongze Cheng 已提交
540 541
  return code;
}
H
Hongze Cheng 已提交
542

H
Hongze Cheng 已提交
543
static int32_t tsdbSnapMoveWriteTableData(STsdbSnapWriter* pWriter, SBlockIdx* pBlockIdx) {
H
Hongze Cheng 已提交
544 545
  int32_t code = 0;

H
Hongze Cheng 已提交
546 547
  code = tsdbReadBlock(pWriter->pDataFReader, pBlockIdx, &pWriter->mBlock, NULL);
  if (code) goto _err;
H
Hongze Cheng 已提交
548

H
Hongze Cheng 已提交
549 550 551 552 553
  // SBlockData
  SBlock block;
  tMapDataReset(&pWriter->mBlockW);
  for (int32_t iBlock = 0; iBlock < pWriter->mBlock.nItem; iBlock++) {
    tMapDataGetItemByIdx(&pWriter->mBlock, iBlock, &block, tGetBlock);
H
Hongze Cheng 已提交
554

H
Hongze Cheng 已提交
555 556 557
    if (block.last) {
      code = tsdbReadBlockData(pWriter->pDataFReader, pBlockIdx, &block, &pWriter->bDataR, NULL, NULL);
      if (code) goto _err;
H
Hongze Cheng 已提交
558

H
Hongze Cheng 已提交
559 560 561 562
      tBlockReset(&block);
      block.last = 1;
      code =
          tsdbWriteBlockData(pWriter->pDataFWriter, &pWriter->bDataR, NULL, NULL, pBlockIdx, &block, pWriter->cmprAlg);
H
Hongze Cheng 已提交
563 564 565
      if (code) goto _err;
    }

H
Hongze Cheng 已提交
566
    code = tMapDataPutItem(&pWriter->mBlockW, &block, tPutBlock);
H
Hongze Cheng 已提交
567
    if (code) goto _err;
H
Hongze Cheng 已提交
568
  }
H
Hongze Cheng 已提交
569

H
Hongze Cheng 已提交
570 571 572 573 574 575 576 577 578
  // SBlock
  SBlockIdx blockIdx = {.suid = pBlockIdx->suid, .uid = pBlockIdx->uid};
  code = tsdbWriteBlock(pWriter->pDataFWriter, &pWriter->mBlockW, NULL, &blockIdx);
  if (code) goto _err;

  // SBlockIdx
  if (taosArrayPush(pWriter->aBlockIdxW, &blockIdx) == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
H
Hongze Cheng 已提交
579 580 581 582 583 584
  }

_exit:
  return code;

_err:
S
Shengliang Guan 已提交
585
  tsdbError("vgId:%d, tsdb snapshot move write table data for %s failed since %s", TD_VID(pWriter->pTsdb->pVnode),
C
Cary Xu 已提交
586
            pWriter->pTsdb->path, tstrerror(code));
H
Hongze Cheng 已提交
587 588 589
  return code;
}

H
Hongze Cheng 已提交
590
static int32_t tsdbSnapWriteTableDataImpl(STsdbSnapWriter* pWriter) {
H
Hongze Cheng 已提交
591 592
  int32_t     code = 0;
  SBlockData* pBlockData = &pWriter->bData;
H
Hongze Cheng 已提交
593 594 595
  int32_t     iRow = 0;
  TSDBROW     row;
  TSDBROW*    pRow = &row;
H
Hongze Cheng 已提交
596

H
Hongze Cheng 已提交
597 598 599
  // correct schema
  code = tBlockDataCorrectSchema(&pWriter->bDataW, pBlockData);
  if (code) goto _err;
H
Hongze Cheng 已提交
600

H
Hongze Cheng 已提交
601 602
  // loop to merge
  *pRow = tsdbRowFromBlockData(pBlockData, iRow);
H
Hongze Cheng 已提交
603 604
  while (true) {
    if (pRow == NULL) break;
H
Hongze Cheng 已提交
605

H
Hongze Cheng 已提交
606 607
    if (pWriter->pBlockData) {
      ASSERT(pWriter->iRow < pWriter->pBlockData->nRow);
H
Hongze Cheng 已提交
608

H
Hongze Cheng 已提交
609
      int32_t c = tsdbRowCmprFn(pRow, &tsdbRowFromBlockData(pWriter->pBlockData, pWriter->iRow));
H
Hongze Cheng 已提交
610

H
Hongze Cheng 已提交
611
      ASSERT(c);
H
Hongze Cheng 已提交
612

H
Hongze Cheng 已提交
613 614
      if (c < 0) {
        code = tBlockDataAppendRow(&pWriter->bDataW, pRow, NULL);
H
Hongze Cheng 已提交
615 616
        if (code) goto _err;

H
Hongze Cheng 已提交
617 618
        iRow++;
        if (iRow < pWriter->pBlockData->nRow) {
H
Hongze Cheng 已提交
619
          *pRow = tsdbRowFromBlockData(pBlockData, iRow);
H
Hongze Cheng 已提交
620 621 622 623 624
        } else {
          pRow = NULL;
        }
      } else if (c > 0) {
        code = tBlockDataAppendRow(&pWriter->bDataW, &tsdbRowFromBlockData(pWriter->pBlockData, pWriter->iRow), NULL);
H
Hongze Cheng 已提交
625 626
        if (code) goto _err;

H
Hongze Cheng 已提交
627 628 629 630
        pWriter->iRow++;
        if (pWriter->iRow >= pWriter->pBlockData->nRow) {
          pWriter->pBlockData = NULL;
        }
H
Hongze Cheng 已提交
631
      }
H
Hongze Cheng 已提交
632
    } else {
H
Hongze Cheng 已提交
633
      TSDBKEY key = TSDBROW_KEY(pRow);
H
Hongze Cheng 已提交
634

H
Hongze Cheng 已提交
635 636 637 638
      while (true) {
        if (pWriter->iBlock >= pWriter->mBlock.nItem) break;

        SBlock  block;
H
Hongze Cheng 已提交
639
        int32_t c;
H
Hongze Cheng 已提交
640

H
Hongze Cheng 已提交
641
        tMapDataGetItemByIdx(&pWriter->mBlock, pWriter->iBlock, &block, tGetBlock);
H
Hongze Cheng 已提交
642

H
Hongze Cheng 已提交
643 644
        if (block.last) {
          pWriter->pBlockData = &pWriter->bDataR;
H
Hongze Cheng 已提交
645

H
Hongze Cheng 已提交
646
          code = tsdbReadBlockData(pWriter->pDataFReader, pWriter->pBlockIdx, &block, pWriter->pBlockData, NULL, NULL);
H
Hongze Cheng 已提交
647
          if (code) goto _err;
H
Hongze Cheng 已提交
648
          pWriter->iRow = 0;
H
Hongze Cheng 已提交
649

H
Hongze Cheng 已提交
650
          pWriter->iBlock++;
H
Hongze Cheng 已提交
651
          break;
H
Hongze Cheng 已提交
652 653
        }

H
Hongze Cheng 已提交
654
        c = tsdbKeyCmprFn(&block.maxKey, &key);
H
Hongze Cheng 已提交
655 656 657

        ASSERT(c);

H
Hongze Cheng 已提交
658 659 660 661 662 663
        if (c < 0) {
          if (pWriter->bDataW.nRow) {
            pWriter->blockW.last = 0;
            code = tsdbWriteBlockData(pWriter->pDataFWriter, &pWriter->bDataW, NULL, NULL, pWriter->pBlockIdxW,
                                      &pWriter->blockW, pWriter->cmprAlg);
            if (code) goto _err;
H
Hongze Cheng 已提交
664

H
Hongze Cheng 已提交
665 666 667 668 669 670
            code = tMapDataPutItem(&pWriter->mBlockW, &pWriter->blockW, tPutBlock);
            if (code) goto _err;

            tBlockReset(&pWriter->blockW);
            tBlockDataClearData(&pWriter->bDataW);
          }
H
Hongze Cheng 已提交
671

H
Hongze Cheng 已提交
672
          code = tMapDataPutItem(&pWriter->mBlockW, &block, tPutBlock);
H
Hongze Cheng 已提交
673
          if (code) goto _err;
H
Hongze Cheng 已提交
674

H
Hongze Cheng 已提交
675 676 677
          pWriter->iBlock++;
        } else {
          c = tsdbKeyCmprFn(&tBlockDataLastKey(pBlockData), &block.minKey);
H
Hongze Cheng 已提交
678

H
Hongze Cheng 已提交
679 680
          ASSERT(c);

H
Hongze Cheng 已提交
681 682 683 684 685 686
          if (c > 0) {
            pWriter->pBlockData = &pWriter->bDataR;
            code =
                tsdbReadBlockData(pWriter->pDataFReader, pWriter->pBlockIdx, &block, pWriter->pBlockData, NULL, NULL);
            if (code) goto _err;
            pWriter->iRow = 0;
H
Hongze Cheng 已提交
687

H
Hongze Cheng 已提交
688 689
            pWriter->iBlock++;
          }
H
Hongze Cheng 已提交
690 691
          break;
        }
H
Hongze Cheng 已提交
692 693
      }

H
Hongze Cheng 已提交
694 695
      if (pWriter->pBlockData) continue;

H
Hongze Cheng 已提交
696 697
      code = tBlockDataAppendRow(&pWriter->bDataW, pRow, NULL);
      if (code) goto _err;
H
Hongze Cheng 已提交
698

H
Hongze Cheng 已提交
699
      iRow++;
H
Hongze Cheng 已提交
700 701
      if (iRow < pBlockData->nRow) {
        *pRow = tsdbRowFromBlockData(pBlockData, iRow);
H
Hongze Cheng 已提交
702 703 704
      } else {
        pRow = NULL;
      }
H
Hongze Cheng 已提交
705 706
    }

H
Hongze Cheng 已提交
707 708
  _check_write:
    if (pWriter->bDataW.nRow < pWriter->maxRow * 4 / 5) continue;
H
Hongze Cheng 已提交
709

H
Hongze Cheng 已提交
710
  _write_block:
H
Hongze Cheng 已提交
711 712
    code = tsdbWriteBlockData(pWriter->pDataFWriter, &pWriter->bDataW, NULL, NULL, pWriter->pBlockIdxW,
                              &pWriter->blockW, pWriter->cmprAlg);
H
Hongze Cheng 已提交
713
    if (code) goto _err;
H
Hongze Cheng 已提交
714

H
Hongze Cheng 已提交
715 716
    code = tMapDataPutItem(&pWriter->mBlockW, &pWriter->blockW, tPutBlock);
    if (code) goto _err;
H
Hongze Cheng 已提交
717

H
Hongze Cheng 已提交
718 719
    tBlockReset(&pWriter->blockW);
    tBlockDataClearData(&pWriter->bDataW);
H
Hongze Cheng 已提交
720 721
  }

H
Hongze Cheng 已提交
722 723 724
  return code;

_err:
S
Shengliang Guan 已提交
725
  tsdbError("vgId:%d, vnode snapshot tsdb write table data impl for %s failed since %s", TD_VID(pWriter->pTsdb->pVnode),
C
Cary Xu 已提交
726
            pWriter->pTsdb->path, tstrerror(code));
H
Hongze Cheng 已提交
727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768
  return code;
}

static int32_t tsdbSnapWriteTableData(STsdbSnapWriter* pWriter, TABLEID id) {
  int32_t     code = 0;
  SBlockData* pBlockData = &pWriter->bData;
  TSDBKEY     keyFirst = tBlockDataFirstKey(pBlockData);
  TSDBKEY     keyLast = tBlockDataLastKey(pBlockData);

  // end last table write if should
  if (pWriter->pBlockIdxW) {
    int32_t c = tTABLEIDCmprFn(pWriter->pBlockIdxW, &id);
    if (c < 0) {
      // end
      code = tsdbSnapWriteTableDataEnd(pWriter);
      if (code) goto _err;

      // reset
      pWriter->pBlockIdxW = NULL;
    } else if (c > 0) {
      ASSERT(0);
    }
  }

  // start new table data write if need
  if (pWriter->pBlockIdxW == NULL) {
    // write table data ahead
    while (true) {
      if (pWriter->iBlockIdx >= taosArrayGetSize(pWriter->aBlockIdx)) break;

      SBlockIdx* pBlockIdx = (SBlockIdx*)taosArrayGet(pWriter->aBlockIdx, pWriter->iBlockIdx);
      int32_t    c = tTABLEIDCmprFn(pBlockIdx, &id);

      if (c >= 0) break;

      code = tsdbSnapMoveWriteTableData(pWriter, pBlockIdx);
      if (code) goto _err;

      pWriter->iBlockIdx++;
    }

    // reader
H
Hongze Cheng 已提交
769
    pWriter->pBlockIdx = NULL;
H
Hongze Cheng 已提交
770 771 772
    if (pWriter->iBlockIdx < taosArrayGetSize(pWriter->aBlockIdx)) {
      ASSERT(pWriter->pDataFReader);

H
Hongze Cheng 已提交
773
      SBlockIdx* pBlockIdx = (SBlockIdx*)taosArrayGet(pWriter->aBlockIdx, pWriter->iBlockIdx);
H
Hongze Cheng 已提交
774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809
      int32_t    c = tTABLEIDCmprFn(pBlockIdx, &id);

      ASSERT(c >= 0);

      if (c == 0) {
        pWriter->pBlockIdx = pBlockIdx;
        pWriter->iBlockIdx++;
      }
    }

    if (pWriter->pBlockIdx) {
      code = tsdbReadBlock(pWriter->pDataFReader, pWriter->pBlockIdx, &pWriter->mBlock, NULL);
      if (code) goto _err;
    } else {
      tMapDataReset(&pWriter->mBlock);
    }
    pWriter->iBlock = 0;
    pWriter->pBlockData = NULL;
    pWriter->iRow = 0;

    // writer
    pWriter->pBlockIdxW = &pWriter->blockIdxW;
    pWriter->pBlockIdxW->suid = id.suid;
    pWriter->pBlockIdxW->uid = id.uid;

    tBlockReset(&pWriter->blockW);
    tBlockDataReset(&pWriter->bDataW);
    tMapDataReset(&pWriter->mBlockW);
  }

  ASSERT(pWriter->pBlockIdxW && pWriter->pBlockIdxW->suid == id.suid && pWriter->pBlockIdxW->uid == id.uid);
  ASSERT(pWriter->pBlockIdx == NULL || (pWriter->pBlockIdx->suid == id.suid && pWriter->pBlockIdx->uid == id.uid));

  code = tsdbSnapWriteTableDataImpl(pWriter);
  if (code) goto _err;

H
Hongze Cheng 已提交
810
_exit:
S
Shengliang Guan 已提交
811
  tsdbDebug("vgId:%d, vnode snapshot tsdb write data impl for %s", TD_VID(pWriter->pTsdb->pVnode), pWriter->pTsdb->path);
H
Hongze Cheng 已提交
812
  return code;
H
Hongze Cheng 已提交
813 814

_err:
S
Shengliang Guan 已提交
815
  tsdbError("vgId:%d, vnode snapshot tsdb write data impl for %s failed since %s", TD_VID(pWriter->pTsdb->pVnode),
C
Cary Xu 已提交
816
            pWriter->pTsdb->path, tstrerror(code));
H
Hongze Cheng 已提交
817
  return code;
H
Hongze Cheng 已提交
818
}
H
Hongze Cheng 已提交
819

H
Hongze Cheng 已提交
820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838
static int32_t tsdbSnapWriteDataEnd(STsdbSnapWriter* pWriter) {
  int32_t code = 0;
  STsdb*  pTsdb = pWriter->pTsdb;

  if (pWriter->pDataFWriter == NULL) goto _exit;

  code = tsdbSnapWriteTableDataEnd(pWriter);
  if (code) goto _err;

  while (pWriter->iBlockIdx < taosArrayGetSize(pWriter->aBlockIdx)) {
    code = tsdbSnapMoveWriteTableData(pWriter, (SBlockIdx*)taosArrayGet(pWriter->aBlockIdx, pWriter->iBlockIdx));
    if (code) goto _err;

    pWriter->iBlockIdx++;
  }

  code = tsdbWriteBlockIdx(pWriter->pDataFWriter, pWriter->aBlockIdxW, NULL);
  if (code) goto _err;

H
Hongze Cheng 已提交
839
  code = tsdbFSUpsertFSet(&pWriter->fs, &pWriter->pDataFWriter->wSet);
H
Hongze Cheng 已提交
840 841 842 843 844 845 846 847 848 849 850
  if (code) goto _err;

  code = tsdbDataFWriterClose(&pWriter->pDataFWriter, 1);
  if (code) goto _err;

  if (pWriter->pDataFReader) {
    code = tsdbDataFReaderClose(&pWriter->pDataFReader);
    if (code) goto _err;
  }

_exit:
S
Shengliang Guan 已提交
851
  tsdbInfo("vgId:%d, vnode snapshot tsdb writer data end for %s", TD_VID(pTsdb->pVnode), pTsdb->path);
H
Hongze Cheng 已提交
852 853 854
  return code;

_err:
S
Shengliang Guan 已提交
855
  tsdbError("vgId:%d, vnode snapshot tsdb writer data end for %s failed since %s", TD_VID(pTsdb->pVnode), pTsdb->path,
C
Cary Xu 已提交
856
            tstrerror(code));
H
Hongze Cheng 已提交
857 858 859
  return code;
}

H
Hongze Cheng 已提交
860
static int32_t tsdbSnapWriteData(STsdbSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
H
Hongze Cheng 已提交
861 862 863 864
  int32_t code = 0;
  STsdb*  pTsdb = pWriter->pTsdb;
  TABLEID id = *(TABLEID*)(pData + sizeof(SSnapDataHdr));
  int64_t n;
H
Hongze Cheng 已提交
865 866

  // decode
H
Hongze Cheng 已提交
867
  SBlockData* pBlockData = &pWriter->bData;
H
Hongze Cheng 已提交
868 869
  n = tGetBlockData(pData + sizeof(SSnapDataHdr) + sizeof(TABLEID), pBlockData);
  ASSERT(n + sizeof(SSnapDataHdr) + sizeof(TABLEID) == nData);
H
Hongze Cheng 已提交
870

H
Hongze Cheng 已提交
871 872 873
  // open file
  TSDBKEY keyFirst = tBlockDataFirstKey(pBlockData);
  TSDBKEY keyLast = tBlockDataLastKey(pBlockData);
H
Hongze Cheng 已提交
874

H
Hongze Cheng 已提交
875 876
  int32_t fid = tsdbKeyFid(keyFirst.ts, pWriter->minutes, pWriter->precision);
  ASSERT(fid == tsdbKeyFid(keyLast.ts, pWriter->minutes, pWriter->precision));
H
Hongze Cheng 已提交
877
  if (pWriter->pDataFWriter == NULL || pWriter->fid != fid) {
H
Hongze Cheng 已提交
878
    // end last file data write if need
H
Hongze Cheng 已提交
879
    code = tsdbSnapWriteDataEnd(pWriter);
H
Hongze Cheng 已提交
880
    if (code) goto _err;
H
Hongze Cheng 已提交
881 882

    pWriter->fid = fid;
H
Hongze Cheng 已提交
883 884

    // read
H
Hongze Cheng 已提交
885
    SDFileSet* pSet = taosArraySearch(pWriter->fs.aDFileSet, &(SDFileSet){.fid = fid}, tDFileSetCmprFn, TD_EQ);
H
Hongze Cheng 已提交
886 887 888 889 890 891 892
    if (pSet) {
      code = tsdbDataFReaderOpen(&pWriter->pDataFReader, pTsdb, pSet);
      if (code) goto _err;

      code = tsdbReadBlockIdx(pWriter->pDataFReader, pWriter->aBlockIdx, NULL);
      if (code) goto _err;
    } else {
H
Hongze Cheng 已提交
893
      ASSERT(pWriter->pDataFReader == NULL);
H
Hongze Cheng 已提交
894 895 896
      taosArrayClear(pWriter->aBlockIdx);
    }
    pWriter->iBlockIdx = 0;
H
Hongze Cheng 已提交
897 898 899
    pWriter->pBlockIdx = NULL;
    tMapDataReset(&pWriter->mBlock);
    pWriter->iBlock = 0;
H
Hongze Cheng 已提交
900
    pWriter->pBlockData = NULL;
H
Hongze Cheng 已提交
901
    pWriter->iRow = 0;
H
Hongze Cheng 已提交
902
    tBlockDataReset(&pWriter->bDataR);
H
Hongze Cheng 已提交
903

H
Hongze Cheng 已提交
904
    // write
H
Hongze Cheng 已提交
905 906 907 908 909
    SHeadFile fHead;
    SDataFile fData;
    SLastFile fLast;
    SSmaFile  fSma;
    SDFileSet wSet = {.pHeadF = &fHead, .pDataF = &fData, .pLastF = &fLast, .pSmaF = &fSma};
H
Hongze Cheng 已提交
910 911

    if (pSet) {
H
Hongze Cheng 已提交
912 913 914 915 916 917
      wSet.diskId = pSet->diskId;
      wSet.fid = fid;
      fHead = (SHeadFile){.commitID = pWriter->commitID, .offset = 0, .size = 0};
      fData = *pSet->pDataF;
      fLast = (SLastFile){.commitID = pWriter->commitID, .size = 0};
      fSma = *pSet->pSmaF;
H
Hongze Cheng 已提交
918
    } else {
H
Hongze Cheng 已提交
919 920 921 922 923 924
      wSet.diskId = (SDiskID){.level = 0, .id = 0};
      wSet.fid = fid;
      fHead = (SHeadFile){.commitID = pWriter->commitID, .offset = 0, .size = 0};
      fData = (SDataFile){.commitID = pWriter->commitID, .size = 0};
      fLast = (SLastFile){.commitID = pWriter->commitID, .size = 0};
      fSma = (SSmaFile){.commitID = pWriter->commitID, .size = 0};
H
Hongze Cheng 已提交
925 926 927 928
    }

    code = tsdbDataFWriterOpen(&pWriter->pDataFWriter, pTsdb, &wSet);
    if (code) goto _err;
H
Hongze Cheng 已提交
929

H
Hongze Cheng 已提交
930 931
    taosArrayClear(pWriter->aBlockIdxW);
    tMapDataReset(&pWriter->mBlockW);
H
Hongze Cheng 已提交
932
    pWriter->pBlockIdxW = NULL;
H
Hongze Cheng 已提交
933
    tBlockDataReset(&pWriter->bDataW);
H
Hongze Cheng 已提交
934 935
  }

H
Hongze Cheng 已提交
936
  code = tsdbSnapWriteTableData(pWriter, id);
H
Hongze Cheng 已提交
937
  if (code) goto _err;
H
Hongze Cheng 已提交
938

S
Shengliang Guan 已提交
939
  tsdbInfo("vgId:%d, vnode snapshot tsdb write data for %s, fid:%d suid:%" PRId64 " uid:%" PRId64 " nRow:%d",
C
Cary Xu 已提交
940
           TD_VID(pTsdb->pVnode), pTsdb->path, fid, id.suid, id.suid, pBlockData->nRow);
H
Hongze Cheng 已提交
941 942 943
  return code;

_err:
S
Shengliang Guan 已提交
944
  tsdbError("vgId:%d, vnode snapshot tsdb write data for %s failed since %s", TD_VID(pTsdb->pVnode), pTsdb->path,
C
Cary Xu 已提交
945
            tstrerror(code));
H
Hongze Cheng 已提交
946 947 948 949 950
  return code;
}

static int32_t tsdbSnapWriteDel(STsdbSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
  int32_t code = 0;
H
Hongze Cheng 已提交
951 952 953
  STsdb*  pTsdb = pWriter->pTsdb;

  if (pWriter->pDelFWriter == NULL) {
H
Hongze Cheng 已提交
954
    SDelFile* pDelFile = pWriter->fs.pDelFile;
H
Hongze Cheng 已提交
955 956 957 958 959 960

    // reader
    if (pDelFile) {
      code = tsdbDelFReaderOpen(&pWriter->pDelFReader, pDelFile, pTsdb, NULL);
      if (code) goto _err;

H
Hongze Cheng 已提交
961
      code = tsdbReadDelIdx(pWriter->pDelFReader, pWriter->aDelIdxR, NULL);
H
Hongze Cheng 已提交
962 963 964 965
      if (code) goto _err;
    }

    // writer
H
Hongze Cheng 已提交
966
    SDelFile delFile = {.commitID = pWriter->commitID, .offset = 0, .size = 0};
H
Hongze Cheng 已提交
967 968 969 970 971
    code = tsdbDelFWriterOpen(&pWriter->pDelFWriter, &delFile, pTsdb);
    if (code) goto _err;
  }

  // process the del data
H
Hongze Cheng 已提交
972
  TABLEID id = *(TABLEID*)(pData + sizeof(SSnapDataHdr));
H
Hongze Cheng 已提交
973 974 975

  while (true) {
    SDelIdx* pDelIdx = NULL;
H
Hongze Cheng 已提交
976
    int64_t  n = sizeof(SSnapDataHdr) + sizeof(TABLEID);
H
Hongze Cheng 已提交
977 978 979 980
    SDelData delData;
    SDelIdx  delIdx;
    int8_t   toBreak = 0;

H
Hongze Cheng 已提交
981
    if (pWriter->iDelIdx < taosArrayGetSize(pWriter->aDelIdxR)) {
H
Hongze Cheng 已提交
982
      pDelIdx = (SDelIdx*)taosArrayGet(pWriter->aDelIdxR, pWriter->iDelIdx);
H
Hongze Cheng 已提交
983 984 985 986 987 988 989 990 991 992 993 994 995
    }

    if (pDelIdx) {
      int32_t c = tTABLEIDCmprFn(&id, pDelIdx);
      if (c < 0) {
        goto _new_del;
      } else {
        code = tsdbReadDelData(pWriter->pDelFReader, pDelIdx, pWriter->aDelData, NULL);
        if (code) goto _err;

        pWriter->iDelIdx++;
        if (c == 0) {
          toBreak = 1;
H
Hongze Cheng 已提交
996
          delIdx = (SDelIdx){.suid = id.suid, .uid = id.uid};
H
Hongze Cheng 已提交
997 998
          goto _merge_del;
        } else {
H
Hongze Cheng 已提交
999
          delIdx = (SDelIdx){.suid = pDelIdx->suid, .uid = pDelIdx->uid};
H
Hongze Cheng 已提交
1000 1001 1002 1003 1004 1005 1006
          goto _write_del;
        }
      }
    }

  _new_del:
    toBreak = 1;
H
Hongze Cheng 已提交
1007
    delIdx = (SDelIdx){.suid = id.suid, .uid = id.uid};
H
Hongze Cheng 已提交
1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022
    taosArrayClear(pWriter->aDelData);

  _merge_del:
    while (n < nData) {
      n += tGetDelData(pData + n, &delData);
      if (taosArrayPush(pWriter->aDelData, &delData) == NULL) {
        code = TSDB_CODE_OUT_OF_MEMORY;
        goto _err;
      }
    }

  _write_del:
    code = tsdbWriteDelData(pWriter->pDelFWriter, pWriter->aDelData, NULL, &delIdx);
    if (code) goto _err;

H
Hongze Cheng 已提交
1023
    if (taosArrayPush(pWriter->aDelIdxW, &delIdx) == NULL) {
H
Hongze Cheng 已提交
1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
    }

    if (toBreak) break;
  }

_exit:
  return code;

_err:
S
Shengliang Guan 已提交
1035
  tsdbError("vgId:%d, vnode snapshot tsdb write del for %s failed since %s", TD_VID(pTsdb->pVnode), pTsdb->path,
C
Cary Xu 已提交
1036
            tstrerror(code));
H
Hongze Cheng 已提交
1037 1038 1039
  return code;
}

H
Hongze Cheng 已提交
1040 1041
static int32_t tsdbSnapWriteDelEnd(STsdbSnapWriter* pWriter) {
  int32_t code = 0;
H
Hongze Cheng 已提交
1042 1043 1044
  STsdb*  pTsdb = pWriter->pTsdb;

  if (pWriter->pDelFWriter == NULL) goto _exit;
H
Hongze Cheng 已提交
1045

H
Hongze Cheng 已提交
1046 1047
  for (; pWriter->iDelIdx < taosArrayGetSize(pWriter->aDelIdxR); pWriter->iDelIdx++) {
    SDelIdx* pDelIdx = (SDelIdx*)taosArrayGet(pWriter->aDelIdxR, pWriter->iDelIdx);
H
Hongze Cheng 已提交
1048 1049 1050 1051 1052 1053 1054 1055

    code = tsdbReadDelData(pWriter->pDelFReader, pDelIdx, pWriter->aDelData, NULL);
    if (code) goto _err;

    SDelIdx delIdx = (SDelIdx){.suid = pDelIdx->suid, .uid = pDelIdx->uid};
    code = tsdbWriteDelData(pWriter->pDelFWriter, pWriter->aDelData, NULL, &delIdx);
    if (code) goto _err;

H
Hongze Cheng 已提交
1056
    if (taosArrayPush(pWriter->aDelIdxR, &delIdx) == NULL) {
H
Hongze Cheng 已提交
1057 1058 1059 1060 1061 1062 1063 1064
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
    }
  }

  code = tsdbUpdateDelFileHdr(pWriter->pDelFWriter);
  if (code) goto _err;

H
Hongze Cheng 已提交
1065
  code = tsdbFSUpsertDelFile(&pWriter->fs, &pWriter->pDelFWriter->fDel);
H
Hongze Cheng 已提交
1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076
  if (code) goto _err;

  code = tsdbDelFWriterClose(&pWriter->pDelFWriter, 1);
  if (code) goto _err;

  if (pWriter->pDelFReader) {
    code = tsdbDelFReaderClose(&pWriter->pDelFReader);
    if (code) goto _err;
  }

_exit:
S
Shengliang Guan 已提交
1077
  tsdbInfo("vgId:%d, vnode snapshot tsdb write del for %s end", TD_VID(pTsdb->pVnode), pTsdb->path);
H
Hongze Cheng 已提交
1078 1079 1080
  return code;

_err:
S
Shengliang Guan 已提交
1081
  tsdbError("vgId:%d, vnode snapshot tsdb write del end for %s failed since %s", TD_VID(pTsdb->pVnode), pTsdb->path,
C
Cary Xu 已提交
1082
            tstrerror(code));
H
Hongze Cheng 已提交
1083 1084 1085
  return code;
}

H
Hongze Cheng 已提交
1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099
int32_t tsdbSnapWriterOpen(STsdb* pTsdb, int64_t sver, int64_t ever, STsdbSnapWriter** ppWriter) {
  int32_t          code = 0;
  STsdbSnapWriter* pWriter = NULL;

  // alloc
  pWriter = (STsdbSnapWriter*)taosMemoryCalloc(1, sizeof(*pWriter));
  if (pWriter == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
  pWriter->pTsdb = pTsdb;
  pWriter->sver = sver;
  pWriter->ever = ever;

H
Hongze Cheng 已提交
1100 1101 1102
  code = tsdbFSCopy(pTsdb, &pWriter->fs);
  if (code) goto _err;

H
Hongze Cheng 已提交
1103
  // config
H
Hongze Cheng 已提交
1104 1105 1106 1107 1108
  pWriter->minutes = pTsdb->keepCfg.days;
  pWriter->precision = pTsdb->keepCfg.precision;
  pWriter->minRow = pTsdb->pVnode->config.tsdbCfg.minRows;
  pWriter->maxRow = pTsdb->pVnode->config.tsdbCfg.maxRows;
  pWriter->cmprAlg = pTsdb->pVnode->config.tsdbCfg.compression;
H
Hongze Cheng 已提交
1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146
  pWriter->commitID = pTsdb->pVnode->state.commitID;

  // for data file
  code = tBlockDataInit(&pWriter->bData);

  if (code) goto _err;
  pWriter->aBlockIdx = taosArrayInit(0, sizeof(SBlockIdx));
  if (pWriter->aBlockIdx == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
  code = tBlockDataInit(&pWriter->bDataR);
  if (code) goto _err;

  pWriter->aBlockIdxW = taosArrayInit(0, sizeof(SBlockIdx));
  if (pWriter->aBlockIdxW == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
  code = tBlockDataInit(&pWriter->bDataW);
  if (code) goto _err;

  // for del file
  pWriter->aDelIdxR = taosArrayInit(0, sizeof(SDelIdx));
  if (pWriter->aDelIdxR == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
  pWriter->aDelData = taosArrayInit(0, sizeof(SDelData));
  if (pWriter->aDelData == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
  pWriter->aDelIdxW = taosArrayInit(0, sizeof(SDelIdx));
  if (pWriter->aDelIdxW == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
H
Hongze Cheng 已提交
1147

H
Hongze Cheng 已提交
1148 1149
  *ppWriter = pWriter;

S
Shengliang Guan 已提交
1150
  tsdbInfo("vgId:%d, tsdb snapshot writer open for %s succeed", TD_VID(pTsdb->pVnode), pTsdb->path);
C
Cary Xu 已提交
1151
  return code;
H
Hongze Cheng 已提交
1152
_err:
S
Shengliang Guan 已提交
1153
  tsdbError("vgId:%d, tsdb snapshot writer open for %s failed since %s", TD_VID(pTsdb->pVnode), pTsdb->path,
C
Cary Xu 已提交
1154
            tstrerror(code));
H
Hongze Cheng 已提交
1155 1156 1157 1158
  *ppWriter = NULL;
  return code;
}

H
Hongze Cheng 已提交
1159
int32_t tsdbSnapWriterClose(STsdbSnapWriter** ppWriter, int8_t rollback) {
H
Hongze Cheng 已提交
1160 1161 1162 1163
  int32_t          code = 0;
  STsdbSnapWriter* pWriter = *ppWriter;

  if (rollback) {
H
Hongze Cheng 已提交
1164 1165 1166
    ASSERT(0);
    // code = tsdbFSRollback(pWriter->pTsdb->pFS);
    // if (code) goto _err;
H
Hongze Cheng 已提交
1167
  } else {
H
Hongze Cheng 已提交
1168 1169 1170 1171 1172 1173
    code = tsdbSnapWriteDataEnd(pWriter);
    if (code) goto _err;

    code = tsdbSnapWriteDelEnd(pWriter);
    if (code) goto _err;

H
Hongze Cheng 已提交
1174 1175 1176 1177
    code = tsdbFSCommit1(pWriter->pTsdb, &pWriter->fs);
    if (code) goto _err;

    code = tsdbFSCommit2(pWriter->pTsdb, &pWriter->fs);
H
Hongze Cheng 已提交
1178 1179 1180
    if (code) goto _err;
  }

S
Shengliang Guan 已提交
1181
  tsdbInfo("vgId:%d, vnode snapshot tsdb writer close for %s", TD_VID(pWriter->pTsdb->pVnode), pWriter->pTsdb->path);
H
Hongze Cheng 已提交
1182 1183 1184 1185 1186
  taosMemoryFree(pWriter);
  *ppWriter = NULL;
  return code;

_err:
S
Shengliang Guan 已提交
1187
  tsdbError("vgId:%d, vnode snapshot tsdb writer close for %s failed since %s", TD_VID(pWriter->pTsdb->pVnode),
C
Cary Xu 已提交
1188 1189 1190
            pWriter->pTsdb->path, tstrerror(code));
  taosMemoryFree(pWriter);
  *ppWriter = NULL;
H
Hongze Cheng 已提交
1191 1192 1193 1194
  return code;
}

int32_t tsdbSnapWrite(STsdbSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
H
Hongze Cheng 已提交
1195 1196
  int32_t       code = 0;
  SSnapDataHdr* pHdr = (SSnapDataHdr*)pData;
H
Hongze Cheng 已提交
1197 1198

  // ts data
C
Cary Xu 已提交
1199
  if (pHdr->type == SNAP_DATA_TSDB) {
H
Hongze Cheng 已提交
1200
    code = tsdbSnapWriteData(pWriter, pData, nData);
H
Hongze Cheng 已提交
1201
    if (code) goto _err;
H
Hongze Cheng 已提交
1202 1203

    goto _exit;
H
Hongze Cheng 已提交
1204
  } else {
H
Hongze Cheng 已提交
1205 1206 1207 1208
    if (pWriter->pDataFWriter) {
      code = tsdbSnapWriteDataEnd(pWriter);
      if (code) goto _err;
    }
H
Hongze Cheng 已提交
1209 1210 1211
  }

  // del data
C
Cary Xu 已提交
1212
  if (pHdr->type == SNAP_DATA_DEL) {
H
Hongze Cheng 已提交
1213
    code = tsdbSnapWriteDel(pWriter, pData, nData);
H
Hongze Cheng 已提交
1214 1215 1216
    if (code) goto _err;
  }

H
Hongze Cheng 已提交
1217
_exit:
C
Cary Xu 已提交
1218
  tsdbDebug("vgId:%d, tsdb snapshot write for %s succeed", TD_VID(pWriter->pTsdb->pVnode), pWriter->pTsdb->path);
H
Hongze Cheng 已提交
1219 1220 1221
  return code;

_err:
C
Cary Xu 已提交
1222
  tsdbError("vgId:%d, tsdb snapshot write for %s failed since %s", TD_VID(pWriter->pTsdb->pVnode), pWriter->pTsdb->path,
C
Cary Xu 已提交
1223
            tstrerror(code));
H
Hongze Cheng 已提交
1224 1225
  return code;
}