tsdbCommit.c 42.4 KB
Newer Older
H
Hongze Cheng 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/*
 * 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/>.
 */

H
Hongze Cheng 已提交
16
#include "tsdb.h"
H
Hongze Cheng 已提交
17 18 19 20 21
typedef struct {
  int64_t   suid;
  int64_t   uid;
  STSchema *pTSchema;
} SSkmInfo;
H
Hongze Cheng 已提交
22

H
Hongze Cheng 已提交
23 24 25 26 27 28
typedef struct {
  int64_t suid;
  int64_t uid;
  TSDBROW row;
} SRowInfo;

H
Hongze Cheng 已提交
29
typedef struct {
H
Hongze Cheng 已提交
30
  STsdb *pTsdb;
H
Hongze Cheng 已提交
31
  /* commit data */
H
Hongze Cheng 已提交
32
  int64_t commitID;
H
Hongze Cheng 已提交
33 34
  int32_t minutes;
  int8_t  precision;
H
Hongze Cheng 已提交
35 36
  int32_t minRow;
  int32_t maxRow;
H
Hongze Cheng 已提交
37
  int8_t  cmprAlg;
H
Hongze Cheng 已提交
38
  STsdbFS fs;
H
Hongze Cheng 已提交
39
  // --------------
H
Hongze Cheng 已提交
40
  TSKEY   nextKey;  // reset by each table commit
H
Hongze Cheng 已提交
41 42 43
  int32_t commitFid;
  TSKEY   minKey;
  TSKEY   maxKey;
H
Hongze Cheng 已提交
44
  // commit file data
H
Hongze Cheng 已提交
45 46
  struct {
    SDataFReader *pReader;
H
Hongze Cheng 已提交
47 48 49 50 51 52 53 54 55 56 57
    // data
    SArray    *aBlockIdx;  // SArray<SBlockIdx>
    int32_t    iBlockIdx;
    SBlockIdx *pBlockIdx;
    SMapData   mBlock;  // SMapData<SBlock>
    SBlockData bData;
    // last
    SArray    *aBlockL;  // SArray<SBlockL>
    int32_t    iBlockL;
    SBlockData bDatal;
    int32_t    iRow;
H
Hongze Cheng 已提交
58 59
    SRowInfo  *pRowInfo;
    SRowInfo   rowInfo;
H
Hongze Cheng 已提交
60 61 62 63
  } dReader;
  struct {
    SDataFWriter *pWriter;
    SArray       *aBlockIdx;  // SArray<SBlockIdx>
H
Hongze Cheng 已提交
64
    SArray       *aBlockL;    // SArray<SBlockL>
H
Hongze Cheng 已提交
65 66
    SMapData      mBlock;     // SMapData<SBlock>
    SBlockData    bData;
H
Hongze Cheng 已提交
67
    SBlockData    bDatal;
H
Hongze Cheng 已提交
68 69 70
  } dWriter;
  SSkmInfo skmTable;
  SSkmInfo skmRow;
H
Hongze Cheng 已提交
71
  /* commit del */
H
Hongze Cheng 已提交
72 73
  SDelFReader *pDelFReader;
  SDelFWriter *pDelFWriter;
H
Hongze Cheng 已提交
74 75 76
  SArray      *aDelIdx;   // SArray<SDelIdx>
  SArray      *aDelIdxN;  // SArray<SDelIdx>
  SArray      *aDelData;  // SArray<SDelData>
H
Hongze Cheng 已提交
77
} SCommitter;
H
refact  
Hongze Cheng 已提交
78

H
Hongze Cheng 已提交
79 80 81 82 83
static int32_t tsdbStartCommit(STsdb *pTsdb, SCommitter *pCommitter);
static int32_t tsdbCommitData(SCommitter *pCommitter);
static int32_t tsdbCommitDel(SCommitter *pCommitter);
static int32_t tsdbCommitCache(SCommitter *pCommitter);
static int32_t tsdbEndCommit(SCommitter *pCommitter, int32_t eno);
H
refact  
Hongze Cheng 已提交
84

H
refact  
Hongze Cheng 已提交
85
int32_t tsdbBegin(STsdb *pTsdb) {
H
Hongze Cheng 已提交
86
  int32_t code = 0;
H
Hongze Cheng 已提交
87

88 89
  if (!pTsdb) return code;

H
Hongze Cheng 已提交
90 91
  SMemTable *pMemTable;
  code = tsdbMemTableCreate(pTsdb, &pMemTable);
H
Hongze Cheng 已提交
92
  if (code) goto _err;
H
Hongze Cheng 已提交
93

H
Hongze Cheng 已提交
94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109
  // lock
  code = taosThreadRwlockWrlock(&pTsdb->rwLock);
  if (code) {
    code = TAOS_SYSTEM_ERROR(code);
    goto _err;
  }

  pTsdb->mem = pMemTable;

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

H
Hongze Cheng 已提交
110 111 112
  return code;

_err:
S
Shengliang Guan 已提交
113
  tsdbError("vgId:%d, tsdb begin failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
114
  return code;
H
Hongze Cheng 已提交
115 116
}

H
more  
Hongze Cheng 已提交
117
int32_t tsdbCommit(STsdb *pTsdb) {
118
  if (!pTsdb) return 0;
H
Hongze Cheng 已提交
119

H
more  
Hongze Cheng 已提交
120
  int32_t    code = 0;
H
Hongze Cheng 已提交
121 122 123 124
  SCommitter commith;
  SMemTable *pMemTable = pTsdb->mem;

  // check
H
Hongze Cheng 已提交
125
  if (pMemTable->nRow == 0 && pMemTable->nDel == 0) {
H
Hongze Cheng 已提交
126
    taosThreadRwlockWrlock(&pTsdb->rwLock);
H
Hongze Cheng 已提交
127
    pTsdb->mem = NULL;
H
Hongze Cheng 已提交
128 129 130
    taosThreadRwlockUnlock(&pTsdb->rwLock);

    tsdbUnrefMemTable(pMemTable);
H
Hongze Cheng 已提交
131 132
    goto _exit;
  }
H
refact  
Hongze Cheng 已提交
133

H
more  
Hongze Cheng 已提交
134
  // start commit
H
more  
Hongze Cheng 已提交
135
  code = tsdbStartCommit(pTsdb, &commith);
H
Hongze Cheng 已提交
136
  if (code) goto _err;
H
refact  
Hongze Cheng 已提交
137

H
refact  
Hongze Cheng 已提交
138 139
  // commit impl
  code = tsdbCommitData(&commith);
H
Hongze Cheng 已提交
140
  if (code) goto _err;
H
refact  
Hongze Cheng 已提交
141 142

  code = tsdbCommitDel(&commith);
H
Hongze Cheng 已提交
143
  if (code) goto _err;
H
refact  
Hongze Cheng 已提交
144 145

  // end commit
H
more  
Hongze Cheng 已提交
146
  code = tsdbEndCommit(&commith, 0);
H
Hongze Cheng 已提交
147
  if (code) goto _err;
H
refact  
Hongze Cheng 已提交
148

H
Hongze Cheng 已提交
149
_exit:
H
refact  
Hongze Cheng 已提交
150 151 152
  return code;

_err:
H
Hongze Cheng 已提交
153
  tsdbEndCommit(&commith, code);
C
Cary Xu 已提交
154
  tsdbError("vgId:%d, failed to commit since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
H
refact  
Hongze Cheng 已提交
155 156 157
  return code;
}

H
Hongze Cheng 已提交
158
static int32_t tsdbCommitDelStart(SCommitter *pCommitter) {
H
Hongze Cheng 已提交
159 160 161 162
  int32_t    code = 0;
  STsdb     *pTsdb = pCommitter->pTsdb;
  SMemTable *pMemTable = pTsdb->imem;

H
Hongze Cheng 已提交
163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179
  pCommitter->aDelIdx = taosArrayInit(0, sizeof(SDelIdx));
  if (pCommitter->aDelIdx == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }

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

  pCommitter->aDelIdxN = taosArrayInit(0, sizeof(SDelIdx));
  if (pCommitter->aDelIdxN == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
H
Hongze Cheng 已提交
180

H
Hongze Cheng 已提交
181
  SDelFile *pDelFileR = pCommitter->fs.pDelFile;
H
Hongze Cheng 已提交
182
  if (pDelFileR) {
H
Hongze Cheng 已提交
183
    code = tsdbDelFReaderOpen(&pCommitter->pDelFReader, pDelFileR, pTsdb);
H
Hongze Cheng 已提交
184
    if (code) goto _err;
H
Hongze Cheng 已提交
185

H
Hongze Cheng 已提交
186
    code = tsdbReadDelIdx(pCommitter->pDelFReader, pCommitter->aDelIdx);
H
Hongze Cheng 已提交
187
    if (code) goto _err;
H
Hongze Cheng 已提交
188 189
  }

H
Hongze Cheng 已提交
190
  // prepare new
H
Hongze Cheng 已提交
191 192
  SDelFile wDelFile = {.commitID = pCommitter->commitID, .size = 0, .offset = 0};
  code = tsdbDelFWriterOpen(&pCommitter->pDelFWriter, &wDelFile, pTsdb);
H
Hongze Cheng 已提交
193
  if (code) goto _err;
H
Hongze Cheng 已提交
194 195

_exit:
S
Shengliang Guan 已提交
196
  tsdbDebug("vgId:%d, commit del start", TD_VID(pTsdb->pVnode));
H
Hongze Cheng 已提交
197 198 199
  return code;

_err:
S
Shengliang Guan 已提交
200
  tsdbError("vgId:%d, commit del start failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
201 202 203
  return code;
}

H
Hongze Cheng 已提交
204
static int32_t tsdbCommitTableDel(SCommitter *pCommitter, STbData *pTbData, SDelIdx *pDelIdx) {
H
Hongze Cheng 已提交
205
  int32_t   code = 0;
H
Hongze Cheng 已提交
206
  SDelData *pDelData;
H
Hongze Cheng 已提交
207 208
  tb_uid_t  suid;
  tb_uid_t  uid;
H
Hongze Cheng 已提交
209 210

  if (pTbData) {
H
Hongze Cheng 已提交
211 212
    suid = pTbData->suid;
    uid = pTbData->uid;
H
Hongze Cheng 已提交
213

H
Hongze Cheng 已提交
214 215 216 217
    if (pTbData->pHead == NULL) {
      pTbData = NULL;
    }
  }
H
Hongze Cheng 已提交
218 219

  if (pDelIdx) {
H
Hongze Cheng 已提交
220 221 222
    suid = pDelIdx->suid;
    uid = pDelIdx->uid;

H
Hongze Cheng 已提交
223
    code = tsdbReadDelData(pCommitter->pDelFReader, pDelIdx, pCommitter->aDelData);
H
Hongze Cheng 已提交
224
    if (code) goto _err;
225 226
  } else {
    taosArrayClear(pCommitter->aDelData);
H
Hongze Cheng 已提交
227 228
  }

H
Hongze Cheng 已提交
229
  if (pTbData == NULL && pDelIdx == NULL) goto _exit;
H
Hongze Cheng 已提交
230

H
Hongze Cheng 已提交
231
  SDelIdx delIdx = {.suid = suid, .uid = uid};
H
Hongze Cheng 已提交
232 233

  // memory
H
Hongze Cheng 已提交
234 235
  pDelData = pTbData ? pTbData->pHead : NULL;
  for (; pDelData; pDelData = pDelData->pNext) {
H
Hongze Cheng 已提交
236 237 238 239
    if (taosArrayPush(pCommitter->aDelData, pDelData) == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
    }
H
Hongze Cheng 已提交
240 241 242
  }

  // write
H
Hongze Cheng 已提交
243
  code = tsdbWriteDelData(pCommitter->pDelFWriter, pCommitter->aDelData, &delIdx);
H
Hongze Cheng 已提交
244 245 246
  if (code) goto _err;

  // put delIdx
247
  if (taosArrayPush(pCommitter->aDelIdxN, &delIdx) == NULL) {
H
Hongze Cheng 已提交
248 249 250
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
H
Hongze Cheng 已提交
251 252 253 254 255

_exit:
  return code;

_err:
S
Shengliang Guan 已提交
256
  tsdbError("vgId:%d, commit table del failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
257 258 259
  return code;
}

H
Hongze Cheng 已提交
260 261
static int32_t tsdbCommitDelEnd(SCommitter *pCommitter) {
  int32_t code = 0;
H
Hongze Cheng 已提交
262
  STsdb  *pTsdb = pCommitter->pTsdb;
H
Hongze Cheng 已提交
263

H
Hongze Cheng 已提交
264
  code = tsdbWriteDelIdx(pCommitter->pDelFWriter, pCommitter->aDelIdxN);
H
Hongze Cheng 已提交
265
  if (code) goto _err;
H
Hongze Cheng 已提交
266

H
Hongze Cheng 已提交
267 268 269
  code = tsdbUpdateDelFileHdr(pCommitter->pDelFWriter);
  if (code) goto _err;

H
Hongze Cheng 已提交
270
  code = tsdbFSUpsertDelFile(&pCommitter->fs, &pCommitter->pDelFWriter->fDel);
H
Hongze Cheng 已提交
271
  if (code) goto _err;
H
Hongze Cheng 已提交
272

H
Hongze Cheng 已提交
273
  code = tsdbDelFWriterClose(&pCommitter->pDelFWriter, 1);
H
Hongze Cheng 已提交
274
  if (code) goto _err;
H
Hongze Cheng 已提交
275 276

  if (pCommitter->pDelFReader) {
H
Hongze Cheng 已提交
277
    code = tsdbDelFReaderClose(&pCommitter->pDelFReader);
H
Hongze Cheng 已提交
278 279 280
    if (code) goto _err;
  }

H
Hongze Cheng 已提交
281 282 283 284
  taosArrayDestroy(pCommitter->aDelIdx);
  taosArrayDestroy(pCommitter->aDelData);
  taosArrayDestroy(pCommitter->aDelIdxN);

H
Hongze Cheng 已提交
285 286 287
  return code;

_err:
S
Shengliang Guan 已提交
288
  tsdbError("vgId:%d, commit del end failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
289 290 291
  return code;
}

H
Hongze Cheng 已提交
292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339
static int32_t tsdbCommitterUpdateTableSchema(SCommitter *pCommitter, int64_t suid, int64_t uid, int32_t sver) {
  int32_t code = 0;

  if (pCommitter->skmTable.pTSchema) {
    if (pCommitter->skmTable.suid == suid) {
      if (suid == 0) {
        if (pCommitter->skmTable.uid == uid && sver == pCommitter->skmTable.pTSchema->version) goto _exit;
      } else {
        if (sver == pCommitter->skmTable.pTSchema->version) goto _exit;
      }
    }
  }

  pCommitter->skmTable.suid = suid;
  pCommitter->skmTable.uid = uid;
  tTSchemaDestroy(pCommitter->skmTable.pTSchema);
  code = metaGetTbTSchemaEx(pCommitter->pTsdb->pVnode->pMeta, suid, uid, sver, &pCommitter->skmTable.pTSchema);
  if (code) goto _exit;

_exit:
  return code;
}

static int32_t tsdbCommitterUpdateRowSchema(SCommitter *pCommitter, int64_t suid, int64_t uid, int32_t sver) {
  int32_t code = 0;

  if (pCommitter->skmRow.pTSchema) {
    if (pCommitter->skmRow.suid == suid) {
      if (suid == 0) {
        if (pCommitter->skmRow.uid == uid && sver == pCommitter->skmRow.pTSchema->version) goto _exit;
      } else {
        if (sver == pCommitter->skmRow.pTSchema->version) goto _exit;
      }
    }
  }

  pCommitter->skmRow.suid = suid;
  pCommitter->skmRow.uid = uid;
  tTSchemaDestroy(pCommitter->skmRow.pTSchema);
  code = metaGetTbTSchemaEx(pCommitter->pTsdb->pVnode->pMeta, suid, uid, sver, &pCommitter->skmRow.pTSchema);
  if (code) {
    goto _exit;
  }

_exit:
  return code;
}

H
Hongze Cheng 已提交
340
static int32_t tsdbCommitterNextLastRow(SCommitter *pCommitter) {
H
Hongze Cheng 已提交
341 342 343 344 345
  int32_t code = 0;

  ASSERT(pCommitter->dReader.pReader);
  ASSERT(pCommitter->dReader.pRowInfo);

H
Hongze Cheng 已提交
346
  SBlockData *pBlockDatal = &pCommitter->dReader.bDatal;
H
Hongze Cheng 已提交
347
  pCommitter->dReader.iRow++;
H
Hongze Cheng 已提交
348
  if (pCommitter->dReader.iRow < pBlockDatal->nRow) {
H
Hongze Cheng 已提交
349 350 351
    if (pBlockDatal->uid) {
      pCommitter->dReader.pRowInfo->uid = pBlockDatal->uid;
    } else {
H
Hongze Cheng 已提交
352 353 354
      pCommitter->dReader.pRowInfo->uid = pBlockDatal->aUid[pCommitter->dReader.iRow];
    }
    pCommitter->dReader.pRowInfo->row = tsdbRowFromBlockData(pBlockDatal, pCommitter->dReader.iRow);
H
Hongze Cheng 已提交
355 356 357
  } else {
    pCommitter->dReader.iBlockL++;
    if (pCommitter->dReader.iBlockL < taosArrayGetSize(pCommitter->dReader.aBlockL)) {
H
Hongze Cheng 已提交
358 359 360 361 362 363 364 365 366 367 368
      SBlockL *pBlockL = (SBlockL *)taosArrayGet(pCommitter->dReader.aBlockL, pCommitter->dReader.iBlockL);
      int64_t  suid = pBlockL->suid;
      int64_t  uid = pBlockL->maxUid;

      code = tsdbCommitterUpdateTableSchema(pCommitter, suid, uid, 1 /*TODO*/);
      if (code) goto _exit;

      code = tBlockDataInit(pBlockDatal, suid, suid ? 0 : uid, pCommitter->skmTable.pTSchema);
      if (code) goto _exit;

      code = tsdbReadLastBlock(pCommitter->dReader.pReader, pBlockL, pBlockDatal);
H
Hongze Cheng 已提交
369 370 371
      if (code) goto _exit;

      pCommitter->dReader.iRow = 0;
H
Hongze Cheng 已提交
372 373 374 375 376 377 378
      pCommitter->dReader.pRowInfo->suid = pBlockDatal->suid;
      if (pBlockDatal->uid) {
        pCommitter->dReader.pRowInfo->uid = pBlockDatal->uid;
      } else {
        pCommitter->dReader.pRowInfo->uid = pBlockDatal->aUid[0];
      }
      pCommitter->dReader.pRowInfo->row = tsdbRowFromBlockData(pBlockDatal, pCommitter->dReader.iRow);
H
Hongze Cheng 已提交
379 380 381 382 383 384 385 386 387
    } else {
      pCommitter->dReader.pRowInfo = NULL;
    }
  }

_exit:
  return code;
}

H
Hongze Cheng 已提交
388 389 390 391 392 393 394 395 396 397
static int32_t tsdbCommitterNextTableData(SCommitter *pCommitter) {
  int32_t code = 0;

  ASSERT(pCommitter->dReader.pBlockIdx);

  pCommitter->dReader.iBlockIdx++;
  if (pCommitter->dReader.iBlockIdx < taosArrayGetSize(pCommitter->dReader.aBlockIdx)) {
    pCommitter->dReader.pBlockIdx =
        (SBlockIdx *)taosArrayGet(pCommitter->dReader.aBlockIdx, pCommitter->dReader.iBlockIdx);

H
Hongze Cheng 已提交
398
    code = tsdbReadBlock(pCommitter->dReader.pReader, pCommitter->dReader.pBlockIdx, &pCommitter->dReader.mBlock);
H
Hongze Cheng 已提交
399 400 401 402 403 404 405 406 407 408 409
    if (code) goto _exit;

    ASSERT(pCommitter->dReader.mBlock.nItem > 0);
  } else {
    pCommitter->dReader.pBlockIdx = NULL;
  }

_exit:
  return code;
}

H
Hongze Cheng 已提交
410 411 412 413
static int32_t tsdbCommitFileDataStart(SCommitter *pCommitter) {
  int32_t    code = 0;
  STsdb     *pTsdb = pCommitter->pTsdb;
  SDFileSet *pRSet = NULL;
H
Hongze Cheng 已提交
414

H
Hongze Cheng 已提交
415
  // memory
H
Hongze Cheng 已提交
416 417 418
  pCommitter->commitFid = tsdbKeyFid(pCommitter->nextKey, pCommitter->minutes, pCommitter->precision);
  tsdbFidKeyRange(pCommitter->commitFid, pCommitter->minutes, pCommitter->precision, &pCommitter->minKey,
                  &pCommitter->maxKey);
H
Hongze Cheng 已提交
419
  pCommitter->nextKey = TSKEY_MAX;
H
Hongze Cheng 已提交
420

H
Hongze Cheng 已提交
421
  // Reader
H
Hongze Cheng 已提交
422 423
  pRSet = (SDFileSet *)taosArraySearch(pCommitter->fs.aDFileSet, &(SDFileSet){.fid = pCommitter->commitFid},
                                       tDFileSetCmprFn, TD_EQ);
H
Hongze Cheng 已提交
424
  if (pRSet) {
H
Hongze Cheng 已提交
425
    code = tsdbDataFReaderOpen(&pCommitter->dReader.pReader, pTsdb, pRSet);
H
Hongze Cheng 已提交
426 427
    if (code) goto _err;

H
Hongze Cheng 已提交
428
    // data
H
Hongze Cheng 已提交
429
    code = tsdbReadBlockIdx(pCommitter->dReader.pReader, pCommitter->dReader.aBlockIdx);
H
Hongze Cheng 已提交
430
    if (code) goto _err;
H
Hongze Cheng 已提交
431

H
Hongze Cheng 已提交
432 433 434 435 436
    pCommitter->dReader.iBlockIdx = 0;
    if (pCommitter->dReader.iBlockIdx < taosArrayGetSize(pCommitter->dReader.aBlockIdx)) {
      pCommitter->dReader.pBlockIdx =
          (SBlockIdx *)taosArrayGet(pCommitter->dReader.aBlockIdx, pCommitter->dReader.iBlockIdx);

H
Hongze Cheng 已提交
437
      code = tsdbReadBlock(pCommitter->dReader.pReader, pCommitter->dReader.pBlockIdx, &pCommitter->dReader.mBlock);
H
Hongze Cheng 已提交
438 439 440 441
      if (code) goto _err;
    } else {
      pCommitter->dReader.pBlockIdx = NULL;
    }
H
Hongze Cheng 已提交
442
    tBlockDataReset(&pCommitter->dReader.bData);
H
Hongze Cheng 已提交
443 444

    // last
H
Hongze Cheng 已提交
445
    code = tsdbReadBlockL(pCommitter->dReader.pReader, pCommitter->dReader.aBlockL);
H
Hongze Cheng 已提交
446 447
    if (code) goto _err;

H
Hongze Cheng 已提交
448 449 450
    pCommitter->dReader.iBlockL = -1;
    pCommitter->dReader.iRow = -1;
    pCommitter->dReader.pRowInfo = &pCommitter->dReader.rowInfo;
H
Hongze Cheng 已提交
451
    tBlockDataReset(&pCommitter->dReader.bDatal);
H
Hongze Cheng 已提交
452
    code = tsdbCommitterNextLastRow(pCommitter);
H
Hongze Cheng 已提交
453
    if (code) goto _err;
H
Hongze Cheng 已提交
454
  } else {
H
Hongze Cheng 已提交
455
    pCommitter->dReader.pBlockIdx = NULL;
H
Hongze Cheng 已提交
456
    pCommitter->dReader.pRowInfo = NULL;
H
Hongze Cheng 已提交
457
  }
H
Hongze Cheng 已提交
458

H
Hongze Cheng 已提交
459
  // Writer
H
Hongze Cheng 已提交
460 461 462 463 464
  SHeadFile fHead;
  SDataFile fData;
  SLastFile fLast;
  SSmaFile  fSma;
  SDFileSet wSet = {.pHeadF = &fHead, .pDataF = &fData, .pLastF = &fLast, .pSmaF = &fSma};
H
Hongze Cheng 已提交
465
  if (pRSet) {
H
Hongze Cheng 已提交
466 467
    wSet.diskId = pRSet->diskId;
    wSet.fid = pCommitter->commitFid;
H
Hongze Cheng 已提交
468
    fHead = (SHeadFile){.commitID = pCommitter->commitID, .offset = 0, .size = 0, .loffset = 0};
H
Hongze Cheng 已提交
469 470 471
    fData = *pRSet->pDataF;
    fLast = (SLastFile){.commitID = pCommitter->commitID, .size = 0};
    fSma = *pRSet->pSmaF;
H
Hongze Cheng 已提交
472
  } else {
473 474 475 476
    SDiskID did = {0};

    tfsAllocDisk(pTsdb->pVnode->pTfs, 0, &did);

H
Hongze Cheng 已提交
477 478
    tfsMkdirRecurAt(pTsdb->pVnode->pTfs, pTsdb->path, did);

479
    wSet.diskId = did;
H
Hongze Cheng 已提交
480
    wSet.fid = pCommitter->commitFid;
H
Hongze Cheng 已提交
481
    fHead = (SHeadFile){.commitID = pCommitter->commitID, .offset = 0, .size = 0, .loffset = 0};
H
Hongze Cheng 已提交
482 483 484
    fData = (SDataFile){.commitID = pCommitter->commitID, .size = 0};
    fLast = (SLastFile){.commitID = pCommitter->commitID, .size = 0};
    fSma = (SSmaFile){.commitID = pCommitter->commitID, .size = 0};
H
Hongze Cheng 已提交
485
  }
H
Hongze Cheng 已提交
486
  code = tsdbDataFWriterOpen(&pCommitter->dWriter.pWriter, pTsdb, &wSet);
H
Hongze Cheng 已提交
487
  if (code) goto _err;
H
Hongze Cheng 已提交
488

H
Hongze Cheng 已提交
489 490 491 492 493 494
  taosArrayClear(pCommitter->dWriter.aBlockIdx);
  taosArrayClear(pCommitter->dWriter.aBlockL);
  tMapDataReset(&pCommitter->dWriter.mBlock);
  tBlockDataReset(&pCommitter->dWriter.bData);
  tBlockDataReset(&pCommitter->dWriter.bDatal);

H
Hongze Cheng 已提交
495
_exit:
H
Hongze Cheng 已提交
496 497 498
  return code;

_err:
S
Shengliang Guan 已提交
499
  tsdbError("vgId:%d, commit file data start failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
500
  return code;
H
Hongze Cheng 已提交
501 502
}

H
Hongze Cheng 已提交
503
static int32_t tsdbCommitDataBlock(SCommitter *pCommitter, SBlock *pBlock) {
H
Hongze Cheng 已提交
504 505 506
  int32_t     code = 0;
  SBlockData *pBlockData = &pCommitter->dWriter.bData;
  SBlock      block;
H
Hongze Cheng 已提交
507

H
Hongze Cheng 已提交
508
  ASSERT(pBlockData->nRow > 0);
H
Hongze Cheng 已提交
509

H
Hongze Cheng 已提交
510 511 512 513 514 515
  if (pBlock) {
    block = *pBlock;  // as a subblock
  } else {
    tBlockReset(&block);  // as a new block
  }

H
Hongze Cheng 已提交
516
  // info
H
Hongze Cheng 已提交
517 518 519
  block.nRow += pBlockData->nRow;
  for (int32_t iRow = 0; iRow < pBlockData->nRow; iRow++) {
    TSDBKEY key = {.ts = pBlockData->aTSKEY[iRow], .version = pBlockData->aVersion[iRow]};
H
Hongze Cheng 已提交
520

H
Hongze Cheng 已提交
521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539
    if (iRow == 0) {
      if (tsdbKeyCmprFn(&block.minKey, &key) > 0) {
        block.minKey = key;
      }
    } else {
      if (pBlockData->aTSKEY[iRow] == pBlockData->aTSKEY[iRow - 1]) {
        block.hasDup = 1;
      }
    }

    if (iRow == pBlockData->nRow - 1 && tsdbKeyCmprFn(&block.maxKey, &key) < 0) {
      block.maxKey = key;
    }

    block.minVer = TMIN(block.minVer, key.version);
    block.maxVer = TMAX(block.maxVer, key.version);
  }

  // write
H
Hongze Cheng 已提交
540 541
  block.nSubBlock++;
  code = tsdbWriteBlockData(pCommitter->dWriter.pWriter, pBlockData, &block.aSubBlock[block.nSubBlock - 1],
H
Hongze Cheng 已提交
542
                            ((block.nSubBlock == 1) && !block.hasDup) ? &block.smaInfo : NULL, pCommitter->cmprAlg, 0);
H
Hongze Cheng 已提交
543 544 545
  if (code) goto _err;

  // put SBlock
H
Hongze Cheng 已提交
546
  code = tMapDataPutItem(&pCommitter->dWriter.mBlock, &block, tPutBlock);
H
Hongze Cheng 已提交
547
  if (code) goto _err;
H
Hongze Cheng 已提交
548

H
Hongze Cheng 已提交
549
  // clear
H
Hongze Cheng 已提交
550
  tBlockDataClear(pBlockData);
H
Hongze Cheng 已提交
551

H
Hongze Cheng 已提交
552 553 554 555
  return code;

_err:
  tsdbError("vgId:%d tsdb commit data block failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
556 557 558 559
  return code;
}

static int32_t tsdbCommitLastBlock(SCommitter *pCommitter) {
H
Hongze Cheng 已提交
560 561 562 563 564 565
  int32_t     code = 0;
  SBlockL     blockL;
  SBlockData *pBlockData = &pCommitter->dWriter.bDatal;

  ASSERT(pBlockData->nRow > 0);

H
Hongze Cheng 已提交
566
  // info
H
Hongze Cheng 已提交
567 568 569 570 571 572
  blockL.suid = pBlockData->suid;
  blockL.nRow = pBlockData->nRow;
  blockL.minVer = VERSION_MAX;
  blockL.maxVer = VERSION_MIN;
  for (int32_t iRow = 0; iRow < pBlockData->nRow; iRow++) {
    blockL.minVer = TMIN(blockL.minVer, pBlockData->aVersion[iRow]);
H
Hongze Cheng 已提交
573
    blockL.maxVer = TMAX(blockL.maxVer, pBlockData->aVersion[iRow]);
H
Hongze Cheng 已提交
574 575 576
  }
  blockL.minUid = pBlockData->uid ? pBlockData->uid : pBlockData->aUid[0];
  blockL.maxUid = pBlockData->uid ? pBlockData->uid : pBlockData->aUid[pBlockData->nRow - 1];
H
Hongze Cheng 已提交
577

H
Hongze Cheng 已提交
578
  // write
H
Hongze Cheng 已提交
579
  code = tsdbWriteBlockData(pCommitter->dWriter.pWriter, pBlockData, &blockL.bInfo, NULL, pCommitter->cmprAlg, 1);
H
Hongze Cheng 已提交
580
  if (code) goto _err;
H
Hongze Cheng 已提交
581

H
Hongze Cheng 已提交
582
  // push SBlockL
H
Hongze Cheng 已提交
583 584
  if (taosArrayPush(pCommitter->dWriter.aBlockL, &blockL) == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
H
Hongze Cheng 已提交
585
    goto _err;
H
Hongze Cheng 已提交
586 587
  }

H
Hongze Cheng 已提交
588
  // clear
H
Hongze Cheng 已提交
589
  tBlockDataClear(pBlockData);
H
Hongze Cheng 已提交
590

H
Hongze Cheng 已提交
591 592 593 594
  return code;

_err:
  tsdbError("vgId:%d tsdb commit last block failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
595 596 597
  return code;
}

H
Hongze Cheng 已提交
598
static int32_t tsdbMergeCommitData(SCommitter *pCommitter, STbDataIter *pIter, SBlock *pBlock) {
H
Hongze Cheng 已提交
599
  int32_t     code = 0;
H
Hongze Cheng 已提交
600 601 602
  STbData    *pTbData = pIter->pTbData;
  SBlockData *pBlockDataR = &pCommitter->dReader.bData;
  SBlockData *pBlockDataW = &pCommitter->dWriter.bData;
H
Hongze Cheng 已提交
603

H
Hongze Cheng 已提交
604
  code = tsdbReadDataBlock(pCommitter->dReader.pReader, pBlock, pBlockDataR);
H
Hongze Cheng 已提交
605 606
  if (code) goto _err;

H
Hongze Cheng 已提交
607
  tBlockDataClear(pBlockDataW);
H
Hongze Cheng 已提交
608 609 610 611 612 613 614
  int32_t  iRow = 0;
  TSDBROW  row;
  TSDBROW *pRow1 = tsdbTbDataIterGet(pIter);
  TSDBROW *pRow2 = &row;
  *pRow2 = tsdbRowFromBlockData(pBlockDataR, iRow);
  while (pRow1 && pRow2) {
    int32_t c = tsdbRowCmprFn(pRow1, pRow2);
H
Hongze Cheng 已提交
615

H
Hongze Cheng 已提交
616 617 618
    if (c < 0) {
      code = tsdbCommitterUpdateRowSchema(pCommitter, pTbData->suid, pTbData->uid, TSDBROW_SVERSION(pRow1));
      if (code) goto _err;
H
Hongze Cheng 已提交
619

H
Hongze Cheng 已提交
620
      code = tBlockDataAppendRow(pBlockDataW, pRow1, pCommitter->skmRow.pTSchema, pTbData->uid);
H
Hongze Cheng 已提交
621
      if (code) goto _err;
H
Hongze Cheng 已提交
622

H
Hongze Cheng 已提交
623 624 625 626
      // next
      tsdbTbDataIterNext(pIter);
      pRow1 = tsdbTbDataIterGet(pIter);
    } else if (c > 0) {
H
Hongze Cheng 已提交
627
      code = tBlockDataAppendRow(pBlockDataW, pRow2, NULL, pTbData->uid);
H
Hongze Cheng 已提交
628
      if (code) goto _err;
H
Hongze Cheng 已提交
629

H
Hongze Cheng 已提交
630 631 632
      iRow++;
      if (iRow < pBlockDataR->nRow) {
        *pRow2 = tsdbRowFromBlockData(pBlockDataR, iRow);
H
Hongze Cheng 已提交
633
      } else {
H
Hongze Cheng 已提交
634
        pRow2 = NULL;
H
Hongze Cheng 已提交
635
      }
H
Hongze Cheng 已提交
636 637
    } else {
      ASSERT(0);
H
Hongze Cheng 已提交
638 639
    }

H
Hongze Cheng 已提交
640 641
    // check
    if (pBlockDataW->nRow >= pCommitter->maxRow * 4 / 5) {
H
Hongze Cheng 已提交
642
      code = tsdbCommitDataBlock(pCommitter, NULL);
H
Hongze Cheng 已提交
643
      if (code) goto _err;
H
Hongze Cheng 已提交
644
    }
H
Hongze Cheng 已提交
645
  }
H
Hongze Cheng 已提交
646

H
Hongze Cheng 已提交
647
  while (pRow2) {
H
Hongze Cheng 已提交
648
    code = tBlockDataAppendRow(pBlockDataW, pRow2, NULL, pTbData->uid);
H
Hongze Cheng 已提交
649 650
    if (code) goto _err;

H
Hongze Cheng 已提交
651 652 653
    iRow++;
    if (iRow < pBlockDataR->nRow) {
      *pRow2 = tsdbRowFromBlockData(pBlockDataR, iRow);
H
Hongze Cheng 已提交
654
    } else {
H
Hongze Cheng 已提交
655
      pRow2 = NULL;
H
Hongze Cheng 已提交
656 657
    }

H
Hongze Cheng 已提交
658 659
    // check
    if (pBlockDataW->nRow >= pCommitter->maxRow * 4 / 5) {
H
Hongze Cheng 已提交
660
      code = tsdbCommitDataBlock(pCommitter, NULL);
H
Hongze Cheng 已提交
661
      if (code) goto _err;
H
Hongze Cheng 已提交
662
    }
H
Hongze Cheng 已提交
663
  }
H
Hongze Cheng 已提交
664

H
Hongze Cheng 已提交
665 666
  // check
  if (pBlockDataW->nRow > 0) {
H
Hongze Cheng 已提交
667
    code = tsdbCommitDataBlock(pCommitter, NULL);
H
Hongze Cheng 已提交
668
    if (code) goto _err;
H
Hongze Cheng 已提交
669 670 671 672 673
  }

  return code;

_err:
H
Hongze Cheng 已提交
674
  tsdbError("vgId:%d, tsdb merge commit data failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
675 676 677
  return code;
}

H
Hongze Cheng 已提交
678 679 680
static int32_t tsdbCommitTableMemData(SCommitter *pCommitter, STbDataIter *pIter, TSDBKEY toKey) {
  int32_t     code = 0;
  STbData    *pTbData = pIter->pTbData;
H
Hongze Cheng 已提交
681
  SBlockData *pBlockData = &pCommitter->dWriter.bData;
H
Hongze Cheng 已提交
682

H
Hongze Cheng 已提交
683
  tBlockDataClear(pBlockData);
H
Hongze Cheng 已提交
684
  TSDBROW *pRow = tsdbTbDataIterGet(pIter);
H
Hongze Cheng 已提交
685
  while (true) {
H
Hongze Cheng 已提交
686
    if (pRow == NULL) {
H
Hongze Cheng 已提交
687 688 689 690 691 692 693
      if (pBlockData->nRow > 0) {
        goto _write_block;
      } else {
        break;
      }
    }

H
Hongze Cheng 已提交
694
    // update schema
H
Hongze Cheng 已提交
695
    code = tsdbCommitterUpdateRowSchema(pCommitter, pTbData->suid, pTbData->uid, TSDBROW_SVERSION(pRow));
H
Hongze Cheng 已提交
696 697
    if (code) goto _err;

H
Hongze Cheng 已提交
698
    // append
H
Hongze Cheng 已提交
699
    code = tBlockDataAppendRow(pBlockData, pRow, pCommitter->skmRow.pTSchema, pTbData->uid);
H
Hongze Cheng 已提交
700 701 702 703
    if (code) goto _err;

    tsdbTbDataIterNext(pIter);
    pRow = tsdbTbDataIterGet(pIter);
704
    if (pRow) {
H
Hongze Cheng 已提交
705 706
      TSDBKEY rowKey = TSDBROW_KEY(pRow);
      if (tsdbKeyCmprFn(&rowKey, &toKey) >= 0) {
707 708 709
        pRow = NULL;
      }
    }
H
Hongze Cheng 已提交
710

H
Hongze Cheng 已提交
711 712
    if (pBlockData->nRow >= pCommitter->maxRow * 4 / 5) {
    _write_block:
H
Hongze Cheng 已提交
713
      code = tsdbCommitDataBlock(pCommitter, NULL);
H
Hongze Cheng 已提交
714 715
      if (code) goto _err;
    }
H
Hongze Cheng 已提交
716 717 718 719 720
  }

  return code;

_err:
S
Shengliang Guan 已提交
721
  tsdbError("vgId:%d, tsdb commit table mem data failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
722 723 724
  return code;
}

H
Hongze Cheng 已提交
725 726
static int32_t tsdbGetNumOfRowsLessThan(STbDataIter *pIter, TSDBKEY key) {
  int32_t nRow = 0;
H
Hongze Cheng 已提交
727

H
Hongze Cheng 已提交
728
  STbDataIter iter = *pIter;
H
Hongze Cheng 已提交
729
  while (true) {
H
Hongze Cheng 已提交
730
    TSDBROW *pRow = tsdbTbDataIterGet(&iter);
H
Hongze Cheng 已提交
731 732
    if (pRow == NULL) break;

H
Hongze Cheng 已提交
733 734
    int32_t c = tsdbKeyCmprFn(&TSDBROW_KEY(pRow), &key);
    if (c < 0) {
H
Hongze Cheng 已提交
735
      nRow++;
H
Hongze Cheng 已提交
736
      tsdbTbDataIterNext(&iter);
H
Hongze Cheng 已提交
737 738 739 740 741 742 743 744 745 746 747 748
    } else if (c > 0) {
      break;
    } else {
      ASSERT(0);
    }
  }

  return nRow;
}

static int32_t tsdbMergeAsSubBlock(SCommitter *pCommitter, STbDataIter *pIter, SBlock *pBlock) {
  int32_t     code = 0;
H
Hongze Cheng 已提交
749
  STbData    *pTbData = pIter->pTbData;
H
Hongze Cheng 已提交
750
  SBlockData *pBlockData = &pCommitter->dWriter.bData;
H
Hongze Cheng 已提交
751

H
Hongze Cheng 已提交
752
  tBlockDataClear(pBlockData);
H
Hongze Cheng 已提交
753
  TSDBROW *pRow = tsdbTbDataIterGet(pIter);
H
Hongze Cheng 已提交
754
  while (true) {
H
Hongze Cheng 已提交
755 756 757 758 759 760 761 762 763 764
    if (pRow == NULL) break;

    code = tsdbCommitterUpdateRowSchema(pCommitter, pTbData->suid, pTbData->uid, TSDBROW_SVERSION(pRow));
    if (code) goto _err;

    code = tBlockDataAppendRow(pBlockData, pRow, pCommitter->skmRow.pTSchema, pTbData->uid);
    if (code) goto _err;

    tsdbTbDataIterNext(pIter);
    pRow = tsdbTbDataIterGet(pIter);
H
Hongze Cheng 已提交
765
    if (pRow) {
H
Hongze Cheng 已提交
766 767
      TSDBKEY rowKey = TSDBROW_KEY(pRow);
      if (tsdbKeyCmprFn(&rowKey, &pBlock->maxKey) > 0) {
H
Hongze Cheng 已提交
768 769 770 771 772
        pRow = NULL;
      }
    }
  }

H
Hongze Cheng 已提交
773
  ASSERT(pBlockData->nRow > 0 && pBlock->nRow + pBlockData->nRow <= pCommitter->maxRow);
H
Hongze Cheng 已提交
774

H
Hongze Cheng 已提交
775
  code = tsdbCommitDataBlock(pCommitter, pBlock);
H
Hongze Cheng 已提交
776 777 778 779 780
  if (code) goto _err;

  return code;

_err:
S
Shengliang Guan 已提交
781
  tsdbError("vgId:%d, tsdb merge as subblock failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
782 783 784
  return code;
}

H
Hongze Cheng 已提交
785 786 787 788
static int32_t tsdbMergeCommitLast(SCommitter *pCommitter, STbDataIter *pIter) {
  int32_t  code = 0;
  STbData *pTbData = pIter->pTbData;
  int32_t  nRow = tsdbGetNumOfRowsLessThan(pIter, (TSDBKEY){.ts = pCommitter->maxKey + 1, .version = VERSION_MIN});
H
Hongze Cheng 已提交
789

H
Hongze Cheng 已提交
790 791
  if (pCommitter->dReader.pRowInfo && tTABLEIDCmprFn(pTbData, pCommitter->dReader.pRowInfo) == 0) {
    if (pCommitter->dReader.pRowInfo->suid) {  // super table
H
Hongze Cheng 已提交
792 793 794 795
      for (int32_t iRow = pCommitter->dReader.iRow; iRow < pCommitter->dReader.bDatal.nRow; iRow++) {
        if (pTbData->uid != pCommitter->dReader.bDatal.aUid[iRow]) break;
        nRow++;
      }
H
Hongze Cheng 已提交
796
    } else {  // normal table
H
Hongze Cheng 已提交
797 798
      ASSERT(pCommitter->dReader.iRow == 0);
      nRow += pCommitter->dReader.bDatal.nRow;
H
Hongze Cheng 已提交
799 800 801 802 803
    }
  }

  if (nRow == 0) goto _exit;

H
Hongze Cheng 已提交
804
  TSDBROW *pRow = tsdbTbDataIterGet(pIter);
H
Hongze Cheng 已提交
805 806 807 808
  if (pRow && TSDBROW_TS(pRow) > pCommitter->maxKey) {
    pRow = NULL;
  }

H
Hongze Cheng 已提交
809 810
  SRowInfo *pRowInfo = pCommitter->dReader.pRowInfo;
  if (pRowInfo && pRowInfo->uid != pTbData->uid) {
H
Hongze Cheng 已提交
811 812 813 814
    pRowInfo = NULL;
  }

  while (nRow) {
H
Hongze Cheng 已提交
815 816 817
    SBlockData *pBlockData;
    int8_t      toData;

H
Hongze Cheng 已提交
818
    if (nRow < pCommitter->minRow) {  // to .last
H
Hongze Cheng 已提交
819
      toData = 0;
H
Hongze Cheng 已提交
820 821
      pBlockData = &pCommitter->dWriter.bDatal;

H
Hongze Cheng 已提交
822
      // commit and reset block data schema if need
H
Hongze Cheng 已提交
823 824
      // QUESTION: Is there a case that pBlockData->nRow == 0 but need to change schema ?
      if (pBlockData->suid || pBlockData->uid) {
H
Hongze Cheng 已提交
825
        if (pBlockData->suid != pTbData->suid || pBlockData->suid == 0) {
H
Hongze Cheng 已提交
826 827 828 829
          if (pBlockData->nRow > 0) {
            code = tsdbCommitLastBlock(pCommitter);
            if (code) goto _err;
          }
H
Hongze Cheng 已提交
830

H
Hongze Cheng 已提交
831 832 833 834 835 836
          tBlockDataReset(pBlockData);
        }
      }

      // set block data schema if need
      if (pBlockData->suid == 0 && pBlockData->uid == 0) {
H
Hongze Cheng 已提交
837 838
        code =
            tBlockDataInit(pBlockData, pTbData->suid, pTbData->suid ? 0 : pTbData->uid, pCommitter->skmTable.pTSchema);
H
Hongze Cheng 已提交
839 840 841 842 843 844 845 846
        if (code) goto _err;
      }

      if (pBlockData->nRow + nRow > pCommitter->maxRow) {
        code = tsdbCommitLastBlock(pCommitter);
        if (code) goto _err;
      }
    } else {  // to .data
H
Hongze Cheng 已提交
847
      toData = 1;
H
Hongze Cheng 已提交
848
      pBlockData = &pCommitter->dWriter.bData;
H
Hongze Cheng 已提交
849
      ASSERT(pBlockData->nRow == 0);
H
Hongze Cheng 已提交
850 851 852 853 854 855 856 857
    }

    while (pRow && pRowInfo) {
      int32_t c = tsdbRowCmprFn(pRow, &pRowInfo->row);
      if (c < 0) {
        code = tsdbCommitterUpdateRowSchema(pCommitter, pTbData->suid, pTbData->uid, TSDBROW_SVERSION(pRow));
        if (code) goto _err;

H
Hongze Cheng 已提交
858
        code = tBlockDataAppendRow(pBlockData, pRow, pCommitter->skmRow.pTSchema, pTbData->uid);
H
Hongze Cheng 已提交
859 860 861 862 863 864 865 866
        if (code) goto _err;

        tsdbTbDataIterNext(pIter);
        pRow = tsdbTbDataIterGet(pIter);
        if (pRow && TSDBROW_TS(pRow) > pCommitter->maxKey) {
          pRow = NULL;
        }
      } else if (c > 0) {
H
Hongze Cheng 已提交
867
        code = tBlockDataAppendRow(pBlockData, &pRowInfo->row, NULL, pTbData->uid);
H
Hongze Cheng 已提交
868 869 870 871 872 873
        if (code) goto _err;

        code = tsdbCommitterNextLastRow(pCommitter);
        if (code) goto _err;

        pRowInfo = pCommitter->dReader.pRowInfo;
H
Hongze Cheng 已提交
874
        if (pRowInfo && pRowInfo->uid != pTbData->uid) {
H
Hongze Cheng 已提交
875 876 877 878 879 880 881
          pRowInfo = NULL;
        }
      } else {
        ASSERT(0);
      }

      nRow--;
H
Hongze Cheng 已提交
882 883 884
      if (toData) {
        if (nRow == 0 || pBlockData->nRow >= pCommitter->maxRow * 4 / 5) {
          code = tsdbCommitDataBlock(pCommitter, NULL);
H
Hongze Cheng 已提交
885 886 887 888 889 890 891 892 893 894
          if (code) goto _err;
          goto _outer_break;
        }
      }
    }

    while (pRow) {
      code = tsdbCommitterUpdateRowSchema(pCommitter, pTbData->suid, pTbData->uid, TSDBROW_SVERSION(pRow));
      if (code) goto _err;

H
Hongze Cheng 已提交
895
      code = tBlockDataAppendRow(pBlockData, pRow, pCommitter->skmRow.pTSchema, pTbData->uid);
H
Hongze Cheng 已提交
896 897 898 899 900 901 902 903
      if (code) goto _err;

      tsdbTbDataIterNext(pIter);
      pRow = tsdbTbDataIterGet(pIter);
      if (pRow && TSDBROW_TS(pRow) > pCommitter->maxKey) {
        pRow = NULL;
      }

H
Hongze Cheng 已提交
904
      nRow--;
H
Hongze Cheng 已提交
905 906 907
      if (toData) {
        if (nRow == 0 || pBlockData->nRow >= pCommitter->maxRow * 4 / 5) {
          code = tsdbCommitDataBlock(pCommitter, NULL);
H
Hongze Cheng 已提交
908 909
          if (code) goto _err;
          goto _outer_break;
H
Hongze Cheng 已提交
910 911
        }
      }
H
Hongze Cheng 已提交
912 913 914
    }

    while (pRowInfo) {
H
Hongze Cheng 已提交
915
      code = tBlockDataAppendRow(pBlockData, &pRowInfo->row, NULL, pTbData->uid);
H
Hongze Cheng 已提交
916 917 918 919
      if (code) goto _err;

      code = tsdbCommitterNextLastRow(pCommitter);
      if (code) goto _err;
H
Hongze Cheng 已提交
920 921

      pRowInfo = pCommitter->dReader.pRowInfo;
H
Hongze Cheng 已提交
922
      if (pRowInfo && pRowInfo->uid != pTbData->uid) {
H
Hongze Cheng 已提交
923 924 925
        pRowInfo = NULL;
      }

H
Hongze Cheng 已提交
926
      nRow--;
H
Hongze Cheng 已提交
927 928 929
      if (toData) {
        if (nRow == 0 || pBlockData->nRow >= pCommitter->maxRow * 4 / 5) {
          code = tsdbCommitDataBlock(pCommitter, NULL);
H
Hongze Cheng 已提交
930 931 932
          if (code) goto _err;
          goto _outer_break;
        }
H
Hongze Cheng 已提交
933 934 935
      }
    }

H
Hongze Cheng 已提交
936 937
  _outer_break:
    ASSERT(nRow >= 0);
H
Hongze Cheng 已提交
938 939
  }

H
Hongze Cheng 已提交
940
_exit:
H
Hongze Cheng 已提交
941 942 943
  return code;

_err:
H
Hongze Cheng 已提交
944
  tsdbError("vgId:%d tsdb merge commit last failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
945 946 947
  return code;
}

H
Hongze Cheng 已提交
948 949 950 951
static int32_t tsdbCommitTableData(SCommitter *pCommitter, STbData *pTbData) {
  int32_t code = 0;

  ASSERT(pCommitter->dReader.pBlockIdx == NULL || tTABLEIDCmprFn(pCommitter->dReader.pBlockIdx, pTbData) >= 0);
H
Hongze Cheng 已提交
952
  ASSERT(pCommitter->dReader.pRowInfo == NULL || tTABLEIDCmprFn(pCommitter->dReader.pRowInfo, pTbData) >= 0);
H
Hongze Cheng 已提交
953 954 955 956 957 958 959 960 961 962 963 964 965 966

  // merge commit table data
  STbDataIter  iter = {0};
  STbDataIter *pIter = &iter;
  TSDBROW     *pRow;

  tsdbTbDataIterOpen(pTbData, &(TSDBKEY){.ts = pCommitter->minKey, .version = VERSION_MIN}, 0, pIter);
  pRow = tsdbTbDataIterGet(pIter);
  if (pRow && TSDBROW_TS(pRow) > pCommitter->maxKey) {
    pRow = NULL;
  }

  if (pRow == NULL) goto _exit;

H
Hongze Cheng 已提交
967 968 969
  int32_t iBlock = 0;
  SBlock  block;
  SBlock *pBlock = &block;
H
Hongze Cheng 已提交
970
  if (pCommitter->dReader.pBlockIdx && tTABLEIDCmprFn(pTbData, pCommitter->dReader.pBlockIdx) == 0) {
H
Hongze Cheng 已提交
971 972 973 974 975
    tMapDataGetItemByIdx(&pCommitter->dReader.mBlock, iBlock, pBlock, tGetBlock);
  } else {
    pBlock = NULL;
  }

H
Hongze Cheng 已提交
976
  code = tsdbCommitterUpdateTableSchema(pCommitter, pTbData->suid, pTbData->uid, pTbData->maxSkmVer /*TODO*/);
H
Hongze Cheng 已提交
977 978
  if (code) goto _err;

H
Hongze Cheng 已提交
979
  tMapDataReset(&pCommitter->dWriter.mBlock);
H
Hongze Cheng 已提交
980 981
  code = tBlockDataInit(&pCommitter->dReader.bData, pTbData->suid, pTbData->uid, pCommitter->skmTable.pTSchema);
  if (code) goto _err;
H
Hongze Cheng 已提交
982
  code = tBlockDataInit(&pCommitter->dWriter.bData, pTbData->suid, pTbData->uid, pCommitter->skmTable.pTSchema);
H
Hongze Cheng 已提交
983
  if (code) goto _err;
H
Hongze Cheng 已提交
984 985

  // .data merge
H
Hongze Cheng 已提交
986 987 988 989 990 991 992 993 994 995 996 997 998
  while (pBlock && pRow) {
    int32_t c = tBlockCmprFn(pBlock, &(SBlock){.minKey = TSDBROW_KEY(pRow), .maxKey = TSDBROW_KEY(pRow)});
    if (c < 0) {  // disk
      code = tMapDataPutItem(&pCommitter->dWriter.mBlock, pBlock, tPutBlock);
      if (code) goto _err;

      // next
      iBlock++;
      if (iBlock < pCommitter->dReader.mBlock.nItem) {
        tMapDataGetItemByIdx(&pCommitter->dReader.mBlock, iBlock, pBlock, tGetBlock);
      } else {
        pBlock = NULL;
      }
H
Hongze Cheng 已提交
999 1000
    } else if (c > 0) {  // memory
      code = tsdbCommitTableMemData(pCommitter, pIter, pBlock->minKey);
H
Hongze Cheng 已提交
1001 1002 1003 1004 1005 1006 1007 1008
      if (code) goto _err;

      // next
      pRow = tsdbTbDataIterGet(pIter);
      if (pRow && TSDBROW_TS(pRow) > pCommitter->maxKey) {
        pRow = NULL;
      }
    } else {  // merge
H
Hongze Cheng 已提交
1009
      int32_t nOvlp = tsdbGetNumOfRowsLessThan(pIter, pBlock->maxKey);
H
Hongze Cheng 已提交
1010 1011 1012 1013 1014 1015 1016

      ASSERT(nOvlp > 0);

      if (pBlock->nRow + nOvlp <= pCommitter->maxRow && pBlock->nSubBlock < TSDB_MAX_SUBBLOCKS) {
        code = tsdbMergeAsSubBlock(pCommitter, pIter, pBlock);
        if (code) goto _err;
      } else {
H
Hongze Cheng 已提交
1017
        code = tsdbMergeCommitData(pCommitter, pIter, pBlock);
H
Hongze Cheng 已提交
1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038
        if (code) goto _err;
      }

      // next
      pRow = tsdbTbDataIterGet(pIter);
      if (pRow && TSDBROW_TS(pRow) > pCommitter->maxKey) {
        pRow = NULL;
      }
      iBlock++;
      if (iBlock < pCommitter->dReader.mBlock.nItem) {
        tMapDataGetItemByIdx(&pCommitter->dReader.mBlock, iBlock, pBlock, tGetBlock);
      } else {
        pBlock = NULL;
      }
    }
  }

  while (pBlock) {
    code = tMapDataPutItem(&pCommitter->dWriter.mBlock, pBlock, tPutBlock);
    if (code) goto _err;

H
Hongze Cheng 已提交
1039
    // next
H
Hongze Cheng 已提交
1040 1041 1042 1043 1044 1045 1046 1047
    iBlock++;
    if (iBlock < pCommitter->dReader.mBlock.nItem) {
      tMapDataGetItemByIdx(&pCommitter->dReader.mBlock, iBlock, pBlock, tGetBlock);
    } else {
      pBlock = NULL;
    }
  }

H
Hongze Cheng 已提交
1048
  // .data append and .last merge
H
Hongze Cheng 已提交
1049 1050
  code = tsdbMergeCommitLast(pCommitter, pIter);
  if (code) goto _err;
H
Hongze Cheng 已提交
1051 1052 1053 1054

  // end
  if (pCommitter->dWriter.mBlock.nItem > 0) {
    SBlockIdx blockIdx = {.suid = pTbData->suid, .uid = pTbData->uid};
H
Hongze Cheng 已提交
1055
    code = tsdbWriteBlock(pCommitter->dWriter.pWriter, &pCommitter->dWriter.mBlock, &blockIdx);
H
Hongze Cheng 已提交
1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076
    if (code) goto _err;

    if (taosArrayPush(pCommitter->dWriter.aBlockIdx, &blockIdx) == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
    }
  }

_exit:
  pRow = tsdbTbDataIterGet(pIter);
  if (pRow) {
    pCommitter->nextKey = TMIN(pCommitter->nextKey, TSDBROW_TS(pRow));
  }

  return code;

_err:
  tsdbError("vgId:%d tsdb commit table data failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
  return code;
}

H
Hongze Cheng 已提交
1077 1078 1079
static int32_t tsdbCommitFileDataEnd(SCommitter *pCommitter) {
  int32_t code = 0;

H
Hongze Cheng 已提交
1080
  // write aBlockIdx
H
Hongze Cheng 已提交
1081
  code = tsdbWriteBlockIdx(pCommitter->dWriter.pWriter, pCommitter->dWriter.aBlockIdx);
H
Hongze Cheng 已提交
1082 1083
  if (code) goto _err;

H
Hongze Cheng 已提交
1084
  // write aBlockL
H
Hongze Cheng 已提交
1085
  code = tsdbWriteBlockL(pCommitter->dWriter.pWriter, pCommitter->dWriter.aBlockL);
H
Hongze Cheng 已提交
1086 1087
  if (code) goto _err;

H
Hongze Cheng 已提交
1088
  // update file header
H
Hongze Cheng 已提交
1089
  code = tsdbUpdateDFileSetHeader(pCommitter->dWriter.pWriter);
H
Hongze Cheng 已提交
1090 1091 1092
  if (code) goto _err;

  // upsert SDFileSet
H
Hongze Cheng 已提交
1093
  code = tsdbFSUpsertFSet(&pCommitter->fs, &pCommitter->dWriter.pWriter->wSet);
H
Hongze Cheng 已提交
1094 1095 1096
  if (code) goto _err;

  // close and sync
H
Hongze Cheng 已提交
1097
  code = tsdbDataFWriterClose(&pCommitter->dWriter.pWriter, 1);
H
Hongze Cheng 已提交
1098 1099
  if (code) goto _err;

H
Hongze Cheng 已提交
1100 1101
  if (pCommitter->dReader.pReader) {
    code = tsdbDataFReaderClose(&pCommitter->dReader.pReader);
H
Hongze Cheng 已提交
1102
    if (code) goto _err;
H
Hongze Cheng 已提交
1103 1104 1105 1106 1107 1108
  }

_exit:
  return code;

_err:
S
Shengliang Guan 已提交
1109
  tsdbError("vgId:%d, commit file data end failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1110 1111 1112
  return code;
}

H
Hongze Cheng 已提交
1113 1114 1115
static int32_t tsdbMoveCommitData(SCommitter *pCommitter, TABLEID toTable) {
  int32_t code = 0;

H
Hongze Cheng 已提交
1116
  // .data
H
Hongze Cheng 已提交
1117 1118 1119 1120
  while (true) {
    if (pCommitter->dReader.pBlockIdx == NULL || tTABLEIDCmprFn(pCommitter->dReader.pBlockIdx, &toTable) >= 0) break;

    SBlockIdx blockIdx = *pCommitter->dReader.pBlockIdx;
H
Hongze Cheng 已提交
1121
    code = tsdbWriteBlock(pCommitter->dWriter.pWriter, &pCommitter->dReader.mBlock, &blockIdx);
H
Hongze Cheng 已提交
1122 1123 1124 1125 1126 1127 1128
    if (code) goto _err;

    if (taosArrayPush(pCommitter->dWriter.aBlockIdx, &blockIdx) == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
    }

H
Hongze Cheng 已提交
1129 1130
    code = tsdbCommitterNextTableData(pCommitter);
    if (code) goto _err;
H
Hongze Cheng 已提交
1131 1132
  }

H
Hongze Cheng 已提交
1133
  // .last
H
Hongze Cheng 已提交
1134
  while (true) {
H
Hongze Cheng 已提交
1135
    if (pCommitter->dReader.pRowInfo == NULL || tTABLEIDCmprFn(pCommitter->dReader.pRowInfo, &toTable) >= 0) break;
H
Hongze Cheng 已提交
1136

H
Hongze Cheng 已提交
1137 1138 1139 1140 1141 1142 1143 1144 1145
    SBlockData *pBlockDataR = &pCommitter->dReader.bDatal;
    SBlockData *pBlockDataW = &pCommitter->dWriter.bDatal;
    tb_uid_t    suid = pCommitter->dReader.pRowInfo->suid;
    tb_uid_t    uid = pCommitter->dReader.pRowInfo->uid;

    ASSERT((pBlockDataR->suid && !pBlockDataR->uid) || (!pBlockDataR->suid && pBlockDataR->uid));
    ASSERT(pBlockDataR->nRow > 0);

    // commit and reset block data schema if need
H
Hongze Cheng 已提交
1146 1147 1148 1149 1150 1151
    if (pBlockDataW->suid || pBlockDataW->uid) {
      if (pBlockDataW->suid != suid || pBlockDataW->suid == 0) {
        if (pBlockDataW->nRow > 0) {
          code = tsdbCommitLastBlock(pCommitter);
          if (code) goto _err;
        }
H
Hongze Cheng 已提交
1152
        tBlockDataReset(pBlockDataW);
H
Hongze Cheng 已提交
1153
      }
H
Hongze Cheng 已提交
1154 1155
    }

H
Hongze Cheng 已提交
1156 1157 1158
    // set block data schema if need
    if (pBlockDataW->suid == 0 && pBlockDataW->uid == 0) {
      code = tsdbCommitterUpdateTableSchema(pCommitter, suid, uid, 1 /*TOOD*/);
H
Hongze Cheng 已提交
1159 1160
      if (code) goto _err;

H
Hongze Cheng 已提交
1161
      code = tBlockDataInit(pBlockDataW, suid, suid ? 0 : uid, pCommitter->skmTable.pTSchema);
H
Hongze Cheng 已提交
1162
      if (code) goto _err;
H
Hongze Cheng 已提交
1163 1164
    }

H
Hongze Cheng 已提交
1165 1166
    // check if it can make sure that one table data in one block
    int32_t nRow = 0;
H
Hongze Cheng 已提交
1167
    if (pBlockDataR->suid) {
H
Hongze Cheng 已提交
1168 1169
      int32_t iRow = pCommitter->dReader.iRow;
      while ((iRow < pBlockDataR->nRow) && (pBlockDataR->aUid[iRow] == uid)) {
H
Hongze Cheng 已提交
1170
        nRow++;
H
Hongze Cheng 已提交
1171
        iRow++;
H
Hongze Cheng 已提交
1172 1173 1174 1175
      }
    } else {
      ASSERT(pCommitter->dReader.iRow == 0);
      nRow = pBlockDataR->nRow;
H
Hongze Cheng 已提交
1176
    }
H
Hongze Cheng 已提交
1177

H
Hongze Cheng 已提交
1178
    ASSERT(nRow > 0 && nRow < pCommitter->minRow);
H
Hongze Cheng 已提交
1179

H
Hongze Cheng 已提交
1180 1181
    if (pBlockDataW->nRow + nRow > pCommitter->maxRow) {
      ASSERT(pBlockDataW->nRow > 0);
H
Hongze Cheng 已提交
1182

H
Hongze Cheng 已提交
1183 1184
      code = tsdbCommitLastBlock(pCommitter);
      if (code) goto _err;
H
Hongze Cheng 已提交
1185 1186
    }

H
Hongze Cheng 已提交
1187
    while (nRow > 0) {
H
Hongze Cheng 已提交
1188
      code = tBlockDataAppendRow(pBlockDataW, &pCommitter->dReader.pRowInfo->row, NULL, uid);
H
Hongze Cheng 已提交
1189 1190
      if (code) goto _err;

H
Hongze Cheng 已提交
1191
      code = tsdbCommitterNextLastRow(pCommitter);
H
Hongze Cheng 已提交
1192
      if (code) goto _err;
H
Hongze Cheng 已提交
1193 1194

      nRow--;
H
Hongze Cheng 已提交
1195 1196 1197 1198 1199 1200 1201 1202 1203 1204
    }
  }

  return code;

_err:
  tsdbError("vgId:%d tsdb move commit data failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
  return code;
}

H
Hongze Cheng 已提交
1205
static int32_t tsdbCommitFileData(SCommitter *pCommitter) {
H
Hongze Cheng 已提交
1206 1207 1208
  int32_t    code = 0;
  STsdb     *pTsdb = pCommitter->pTsdb;
  SMemTable *pMemTable = pTsdb->imem;
H
Hongze Cheng 已提交
1209 1210 1211 1212 1213 1214

  // commit file data start
  code = tsdbCommitFileDataStart(pCommitter);
  if (code) goto _err;

  // commit file data impl
H
Hongze Cheng 已提交
1215 1216
  for (int32_t iTbData = 0; iTbData < taosArrayGetSize(pMemTable->aTbData); iTbData++) {
    STbData *pTbData = (STbData *)taosArrayGetP(pMemTable->aTbData, iTbData);
H
Hongze Cheng 已提交
1217

H
Hongze Cheng 已提交
1218
    // move commit until current (suid, uid)
H
Hongze Cheng 已提交
1219
    code = tsdbMoveCommitData(pCommitter, *(TABLEID *)pTbData);
H
Hongze Cheng 已提交
1220 1221
    if (code) goto _err;

H
Hongze Cheng 已提交
1222
    // commit current table data
H
Hongze Cheng 已提交
1223
    code = tsdbCommitTableData(pCommitter, pTbData);
H
Hongze Cheng 已提交
1224
    if (code) goto _err;
H
Hongze Cheng 已提交
1225 1226 1227 1228 1229 1230

    // move next reader table data if need
    if (pCommitter->dReader.pBlockIdx && tTABLEIDCmprFn(pTbData, pCommitter->dReader.pBlockIdx) == 0) {
      code = tsdbCommitterNextTableData(pCommitter);
      if (code) goto _err;
    }
H
Hongze Cheng 已提交
1231
  }
H
Hongze Cheng 已提交
1232

H
Hongze Cheng 已提交
1233 1234
  code = tsdbMoveCommitData(pCommitter, (TABLEID){.suid = INT64_MAX, .uid = INT64_MAX});
  if (code) goto _err;
H
Hongze Cheng 已提交
1235

H
Hongze Cheng 已提交
1236
  if (pCommitter->dWriter.bDatal.nRow > 0) {
H
Hongze Cheng 已提交
1237
    code = tsdbCommitLastBlock(pCommitter);
H
Hongze Cheng 已提交
1238
    if (code) goto _err;
H
Hongze Cheng 已提交
1239 1240
  }

H
Hongze Cheng 已提交
1241 1242
  // commit file data end
  code = tsdbCommitFileDataEnd(pCommitter);
H
Hongze Cheng 已提交
1243
  if (code) goto _err;
H
Hongze Cheng 已提交
1244 1245 1246 1247

  return code;

_err:
S
Shengliang Guan 已提交
1248
  tsdbError("vgId:%d, commit file data failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1249 1250
  tsdbDataFReaderClose(&pCommitter->dReader.pReader);
  tsdbDataFWriterClose(&pCommitter->dWriter.pWriter, 0);
H
Hongze Cheng 已提交
1251 1252 1253
  return code;
}

H
Hongze Cheng 已提交
1254 1255
// ----------------------------------------------------------------------------
static int32_t tsdbStartCommit(STsdb *pTsdb, SCommitter *pCommitter) {
H
Hongze Cheng 已提交
1256
  int32_t code = 0;
H
Hongze Cheng 已提交
1257

H
Hongze Cheng 已提交
1258 1259
  memset(pCommitter, 0, sizeof(*pCommitter));
  ASSERT(pTsdb->mem && pTsdb->imem == NULL);
H
Hongze Cheng 已提交
1260

H
more  
Hongze Cheng 已提交
1261
  taosThreadRwlockWrlock(&pTsdb->rwLock);
H
Hongze Cheng 已提交
1262 1263
  pTsdb->imem = pTsdb->mem;
  pTsdb->mem = NULL;
H
more  
Hongze Cheng 已提交
1264
  taosThreadRwlockUnlock(&pTsdb->rwLock);
H
Hongze Cheng 已提交
1265

H
Hongze Cheng 已提交
1266
  pCommitter->pTsdb = pTsdb;
H
Hongze Cheng 已提交
1267
  pCommitter->commitID = pTsdb->pVnode->state.commitID;
H
Hongze Cheng 已提交
1268 1269 1270 1271
  pCommitter->minutes = pTsdb->keepCfg.days;
  pCommitter->precision = pTsdb->keepCfg.precision;
  pCommitter->minRow = pTsdb->pVnode->config.tsdbCfg.minRows;
  pCommitter->maxRow = pTsdb->pVnode->config.tsdbCfg.maxRows;
H
Hongze Cheng 已提交
1272
  pCommitter->cmprAlg = pTsdb->pVnode->config.tsdbCfg.compression;
H
Hongze Cheng 已提交
1273

H
Hongze Cheng 已提交
1274
  code = tsdbFSCopy(pTsdb, &pCommitter->fs);
H
Hongze Cheng 已提交
1275 1276 1277 1278 1279
  if (code) goto _err;

  return code;

_err:
S
Shengliang Guan 已提交
1280
  tsdbError("vgId:%d, tsdb start commit failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1281 1282 1283
  return code;
}

H
Hongze Cheng 已提交
1284 1285 1286
static int32_t tsdbCommitDataStart(SCommitter *pCommitter) {
  int32_t code = 0;

H
Hongze Cheng 已提交
1287
  // Reader
H
Hongze Cheng 已提交
1288 1289
  pCommitter->dReader.aBlockIdx = taosArrayInit(0, sizeof(SBlockIdx));
  if (pCommitter->dReader.aBlockIdx == NULL) {
H
Hongze Cheng 已提交
1290 1291 1292 1293
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

H
Hongze Cheng 已提交
1294
  code = tBlockDataCreate(&pCommitter->dReader.bData);
H
Hongze Cheng 已提交
1295 1296
  if (code) goto _exit;

H
Hongze Cheng 已提交
1297 1298
  pCommitter->dReader.aBlockL = taosArrayInit(0, sizeof(SBlockL));
  if (pCommitter->dReader.aBlockL == NULL) {
H
Hongze Cheng 已提交
1299 1300 1301 1302
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

H
Hongze Cheng 已提交
1303
  code = tBlockDataCreate(&pCommitter->dReader.bDatal);
H
Hongze Cheng 已提交
1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318
  if (code) goto _exit;

  // Writer
  pCommitter->dWriter.aBlockIdx = taosArrayInit(0, sizeof(SBlockIdx));
  if (pCommitter->dWriter.aBlockIdx == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

  pCommitter->dWriter.aBlockL = taosArrayInit(0, sizeof(SBlockL));
  if (pCommitter->dWriter.aBlockL == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

H
Hongze Cheng 已提交
1319
  code = tBlockDataCreate(&pCommitter->dWriter.bData);
H
Hongze Cheng 已提交
1320
  if (code) goto _exit;
H
Hongze Cheng 已提交
1321

H
Hongze Cheng 已提交
1322
  code = tBlockDataCreate(&pCommitter->dWriter.bDatal);
H
Hongze Cheng 已提交
1323 1324
  if (code) goto _exit;

H
Hongze Cheng 已提交
1325 1326 1327 1328 1329
_exit:
  return code;
}

static void tsdbCommitDataEnd(SCommitter *pCommitter) {
H
Hongze Cheng 已提交
1330
  // Reader
H
Hongze Cheng 已提交
1331 1332
  taosArrayDestroy(pCommitter->dReader.aBlockIdx);
  tMapDataClear(&pCommitter->dReader.mBlock);
H
Hongze Cheng 已提交
1333
  tBlockDataDestroy(&pCommitter->dReader.bData, 1);
H
Hongze Cheng 已提交
1334
  taosArrayDestroy(pCommitter->dReader.aBlockL);
H
Hongze Cheng 已提交
1335
  tBlockDataDestroy(&pCommitter->dReader.bDatal, 1);
H
Hongze Cheng 已提交
1336 1337

  // Writer
H
Hongze Cheng 已提交
1338
  taosArrayDestroy(pCommitter->dWriter.aBlockIdx);
H
Hongze Cheng 已提交
1339
  taosArrayDestroy(pCommitter->dWriter.aBlockL);
H
Hongze Cheng 已提交
1340
  tMapDataClear(&pCommitter->dWriter.mBlock);
H
Hongze Cheng 已提交
1341 1342
  tBlockDataDestroy(&pCommitter->dWriter.bData, 1);
  tBlockDataDestroy(&pCommitter->dWriter.bDatal, 1);
H
Hongze Cheng 已提交
1343 1344
  tTSchemaDestroy(pCommitter->skmTable.pTSchema);
  tTSchemaDestroy(pCommitter->skmRow.pTSchema);
H
Hongze Cheng 已提交
1345 1346
}

H
Hongze Cheng 已提交
1347 1348 1349 1350
static int32_t tsdbCommitData(SCommitter *pCommitter) {
  int32_t    code = 0;
  STsdb     *pTsdb = pCommitter->pTsdb;
  SMemTable *pMemTable = pTsdb->imem;
H
Hongze Cheng 已提交
1351

H
Hongze Cheng 已提交
1352
  // check
H
Hongze Cheng 已提交
1353
  if (pMemTable->nRow == 0) goto _exit;
H
Hongze Cheng 已提交
1354

H
Hongze Cheng 已提交
1355 1356
  // start ====================
  code = tsdbCommitDataStart(pCommitter);
H
Hongze Cheng 已提交
1357
  if (code) goto _err;
H
Hongze Cheng 已提交
1358 1359 1360

  // impl ====================
  pCommitter->nextKey = pMemTable->minKey;
H
Hongze Cheng 已提交
1361 1362
  while (pCommitter->nextKey < TSKEY_MAX) {
    code = tsdbCommitFileData(pCommitter);
H
Hongze Cheng 已提交
1363
    if (code) goto _err;
H
Hongze Cheng 已提交
1364
  }
H
Hongze Cheng 已提交
1365

H
Hongze Cheng 已提交
1366 1367 1368
  // end ====================
  tsdbCommitDataEnd(pCommitter);

H
Hongze Cheng 已提交
1369
_exit:
S
Shengliang Guan 已提交
1370
  tsdbDebug("vgId:%d, commit data done, nRow:%" PRId64, TD_VID(pTsdb->pVnode), pMemTable->nRow);
H
Hongze Cheng 已提交
1371
  return code;
H
Hongze Cheng 已提交
1372

H
Hongze Cheng 已提交
1373
_err:
H
Hongze Cheng 已提交
1374
  tsdbCommitDataEnd(pCommitter);
S
Shengliang Guan 已提交
1375
  tsdbError("vgId:%d, commit data failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1376 1377
  return code;
}
H
Hongze Cheng 已提交
1378

H
Hongze Cheng 已提交
1379 1380 1381 1382
static int32_t tsdbCommitDel(SCommitter *pCommitter) {
  int32_t    code = 0;
  STsdb     *pTsdb = pCommitter->pTsdb;
  SMemTable *pMemTable = pTsdb->imem;
H
Hongze Cheng 已提交
1383

H
Hongze Cheng 已提交
1384 1385
  if (pMemTable->nDel == 0) {
    goto _exit;
H
Hongze Cheng 已提交
1386
  }
H
Hongze Cheng 已提交
1387

H
Hongze Cheng 已提交
1388 1389 1390 1391 1392
  // start
  code = tsdbCommitDelStart(pCommitter);
  if (code) {
    goto _err;
  }
H
Hongze Cheng 已提交
1393

H
Hongze Cheng 已提交
1394
  // impl
H
Hongze Cheng 已提交
1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449
  int32_t  iDelIdx = 0;
  int32_t  nDelIdx = taosArrayGetSize(pCommitter->aDelIdx);
  int32_t  iTbData = 0;
  int32_t  nTbData = taosArrayGetSize(pMemTable->aTbData);
  STbData *pTbData;
  SDelIdx *pDelIdx;

  ASSERT(nTbData > 0);

  pTbData = (STbData *)taosArrayGetP(pMemTable->aTbData, iTbData);
  pDelIdx = (iDelIdx < nDelIdx) ? (SDelIdx *)taosArrayGet(pCommitter->aDelIdx, iDelIdx) : NULL;
  while (true) {
    if (pTbData == NULL && pDelIdx == NULL) break;

    if (pTbData && pDelIdx) {
      int32_t c = tTABLEIDCmprFn(pTbData, pDelIdx);

      if (c == 0) {
        goto _commit_mem_and_disk_del;
      } else if (c < 0) {
        goto _commit_mem_del;
      } else {
        goto _commit_disk_del;
      }
    } else if (pTbData) {
      goto _commit_mem_del;
    } else {
      goto _commit_disk_del;
    }

  _commit_mem_del:
    code = tsdbCommitTableDel(pCommitter, pTbData, NULL);
    if (code) goto _err;

    iTbData++;
    pTbData = (iTbData < nTbData) ? (STbData *)taosArrayGetP(pMemTable->aTbData, iTbData) : NULL;
    continue;

  _commit_disk_del:
    code = tsdbCommitTableDel(pCommitter, NULL, pDelIdx);
    if (code) goto _err;

    iDelIdx++;
    pDelIdx = (iDelIdx < nDelIdx) ? (SDelIdx *)taosArrayGet(pCommitter->aDelIdx, iDelIdx) : NULL;
    continue;

  _commit_mem_and_disk_del:
    code = tsdbCommitTableDel(pCommitter, pTbData, pDelIdx);
    if (code) goto _err;

    iTbData++;
    pTbData = (iTbData < nTbData) ? (STbData *)taosArrayGetP(pMemTable->aTbData, iTbData) : NULL;
    iDelIdx++;
    pDelIdx = (iDelIdx < nDelIdx) ? (SDelIdx *)taosArrayGet(pCommitter->aDelIdx, iDelIdx) : NULL;
    continue;
H
Hongze Cheng 已提交
1450
  }
H
Hongze Cheng 已提交
1451

H
Hongze Cheng 已提交
1452 1453 1454 1455 1456
  // end
  code = tsdbCommitDelEnd(pCommitter);
  if (code) {
    goto _err;
  }
H
Hongze Cheng 已提交
1457

H
Hongze Cheng 已提交
1458
_exit:
S
Shengliang Guan 已提交
1459
  tsdbDebug("vgId:%d, commit del done, nDel:%" PRId64, TD_VID(pTsdb->pVnode), pMemTable->nDel);
H
Hongze Cheng 已提交
1460 1461 1462
  return code;

_err:
S
Shengliang Guan 已提交
1463
  tsdbError("vgId:%d, commit del failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1464
  return code;
H
Hongze Cheng 已提交
1465 1466
}

H
Hongze Cheng 已提交
1467
static int32_t tsdbEndCommit(SCommitter *pCommitter, int32_t eno) {
H
Hongze Cheng 已提交
1468 1469 1470 1471
  int32_t    code = 0;
  STsdb     *pTsdb = pCommitter->pTsdb;
  SMemTable *pMemTable = pTsdb->imem;

H
Hongze Cheng 已提交
1472
  ASSERT(eno == 0);
H
Hongze Cheng 已提交
1473 1474 1475

  code = tsdbFSCommit1(pTsdb, &pCommitter->fs);
  if (code) goto _err;
H
Hongze Cheng 已提交
1476

H
Hongze Cheng 已提交
1477
  // lock
H
Hongze Cheng 已提交
1478
  taosThreadRwlockWrlock(&pTsdb->rwLock);
H
Hongze Cheng 已提交
1479 1480 1481 1482 1483 1484 1485 1486

  // commit or rollback
  code = tsdbFSCommit2(pTsdb, &pCommitter->fs);
  if (code) {
    taosThreadRwlockUnlock(&pTsdb->rwLock);
    goto _err;
  }

H
Hongze Cheng 已提交
1487
  pTsdb->imem = NULL;
H
Hongze Cheng 已提交
1488 1489

  // unlock
H
Hongze Cheng 已提交
1490 1491 1492
  taosThreadRwlockUnlock(&pTsdb->rwLock);

  tsdbUnrefMemTable(pMemTable);
H
Hongze Cheng 已提交
1493
  tsdbFSDestroy(&pCommitter->fs);
H
Hongze Cheng 已提交
1494

S
Shengliang Guan 已提交
1495
  tsdbInfo("vgId:%d, tsdb end commit", TD_VID(pTsdb->pVnode));
H
Hongze Cheng 已提交
1496 1497 1498
  return code;

_err:
S
Shengliang Guan 已提交
1499
  tsdbError("vgId:%d, tsdb end commit failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1500 1501
  return code;
}