tsdbCommit.c 49.5 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

H
more  
Hongze Cheng 已提交
18 19
#define TSDB_MAX_SUBBLOCKS 8

H
more  
Hongze Cheng 已提交
20
typedef struct {
L
Liu Jicong 已提交
21
  STable            *pTable;
H
more  
Hongze Cheng 已提交
22 23 24 25 26 27 28 29 30 31 32 33 34 35 36
  SSkipListIterator *pIter;
} SCommitIter;

typedef struct {
  SRtn         rtn;     // retention snapshot
  SFSIter      fsIter;  // tsdb file iterator
  int          niters;  // memory iterators
  SCommitIter *iters;
  bool         isRFileSet;  // read and commit FSET
  SReadH       readh;
  SDFileSet    wSet;
  bool         isDFileSame;
  bool         isLFileSame;
  TSKEY        minKey;
  TSKEY        maxKey;
L
Liu Jicong 已提交
37 38 39 40 41
  SArray      *aBlkIdx;  // SBlockIdx array
  STable      *pTable;
  SArray      *aSupBlk;  // Table super-block array
  SArray      *aSubBlk;  // table sub-block array
  SDataCols   *pDataCols;
H
more  
Hongze Cheng 已提交
42 43
} SCommitH;

H
more  
Hongze Cheng 已提交
44 45
#define TSDB_DEFAULT_BLOCK_ROWS(maxRows) ((maxRows)*4 / 5)

C
Cary Xu 已提交
46 47 48 49 50 51 52 53 54 55 56 57
#define TSDB_COMMIT_REPO(ch)         TSDB_READ_REPO(&(ch->readh))
#define TSDB_COMMIT_REPO_ID(ch)      REPO_ID(TSDB_READ_REPO(&(ch->readh)))
#define TSDB_COMMIT_WRITE_FSET(ch)   (&((ch)->wSet))
#define TSDB_COMMIT_TABLE(ch)        ((ch)->pTable)
#define TSDB_COMMIT_HEAD_FILE(ch)    TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_HEAD)
#define TSDB_COMMIT_DATA_FILE(ch)    TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_DATA)
#define TSDB_COMMIT_LAST_FILE(ch)    TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_LAST)
#define TSDB_COMMIT_SMAD_FILE(ch)    TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_SMAD)
#define TSDB_COMMIT_SMAL_FILE(ch)    TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_SMAL)
#define TSDB_COMMIT_BUF(ch)          TSDB_READ_BUF(&((ch)->readh))
#define TSDB_COMMIT_COMP_BUF(ch)     TSDB_READ_COMP_BUF(&((ch)->readh))
#define TSDB_COMMIT_EXBUF(ch)        TSDB_READ_EXBUF(&((ch)->readh))
H
Hongze Cheng 已提交
58
#define TSDB_COMMIT_DEFAULT_ROWS(ch) TSDB_DEFAULT_BLOCK_ROWS(TSDB_COMMIT_REPO(ch)->pVnode->config.tsdbCfg.maxRows)
C
Cary Xu 已提交
59
#define TSDB_COMMIT_TXN_VERSION(ch)  FS_TXN_VERSION(REPO_FS(TSDB_COMMIT_REPO(ch)))
H
refact  
Hongze Cheng 已提交
60

H
more  
Hongze Cheng 已提交
61 62
static void tsdbStartCommit(STsdb *pRepo);
static void tsdbEndCommit(STsdb *pTsdb, int eno);
H
refact  
Hongze Cheng 已提交
63 64 65 66 67
static int  tsdbInitCommitH(SCommitH *pCommith, STsdb *pRepo);
static void tsdbSeekCommitIter(SCommitH *pCommith, TSKEY key);
static int  tsdbNextCommitFid(SCommitH *pCommith);
static void tsdbDestroyCommitH(SCommitH *pCommith);
static int  tsdbCreateCommitIters(SCommitH *pCommith);
H
Hongze Cheng 已提交
68
static void tsdbDestroyCommitIters(SCommitH *pCommith);
H
refact  
Hongze Cheng 已提交
69
static int  tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid);
H
more  
Hongze Cheng 已提交
70 71 72 73 74 75
static void tsdbResetCommitFile(SCommitH *pCommith);
static int  tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid);
// static int  tsdbCommitMeta(STsdbRepo *pRepo);
// static int  tsdbUpdateMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid, void *cont, int contLen, bool compact);
// static int  tsdbDropMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid);
// static int  tsdbCompactMetaFile(STsdbRepo *pRepo, STsdbFS *pfs, SMFile *pMFile);
H
more  
Hongze Cheng 已提交
76 77 78 79 80
static int  tsdbCommitToTable(SCommitH *pCommith, int tid);
static int  tsdbSetCommitTable(SCommitH *pCommith, STable *pTable);
static int  tsdbComparKeyBlock(const void *arg1, const void *arg2);
static int  tsdbWriteBlockInfo(SCommitH *pCommih);
static int  tsdbCommitMemData(SCommitH *pCommith, SCommitIter *pIter, TSKEY keyLimit, bool toData);
H
more  
Hongze Cheng 已提交
81
static int  tsdbMergeMemData(SCommitH *pCommith, SCommitIter *pIter, int bidx);
H
more  
Hongze Cheng 已提交
82 83 84 85 86 87
static int  tsdbMoveBlock(SCommitH *pCommith, int bidx);
static int  tsdbCommitAddBlock(SCommitH *pCommith, const SBlock *pSupBlock, const SBlock *pSubBlocks, int nSubBlocks);
static int  tsdbMergeBlockData(SCommitH *pCommith, SCommitIter *pIter, SDataCols *pDataCols, TSKEY keyLimit,
                               bool isLastOneBlock);
static void tsdbResetCommitTable(SCommitH *pCommith);
static void tsdbCloseCommitFile(SCommitH *pCommith, bool hasError);
H
more  
Hongze Cheng 已提交
88 89 90
static bool tsdbCanAddSubBlock(SCommitH *pCommith, SBlock *pBlock, SMergeInfo *pInfo);
static void tsdbLoadAndMergeFromCache(SDataCols *pDataCols, int *iter, SCommitIter *pCommitIter, SDataCols *pTarget,
                                      TSKEY maxKey, int maxRows, int8_t update);
H
more  
Hongze Cheng 已提交
91
int         tsdbWriteBlockIdx(SDFile *pHeadf, SArray *pIdxA, void **ppBuf);
H
refact  
Hongze Cheng 已提交
92 93 94

int tsdbApplyRtnOnFSet(STsdb *pRepo, SDFileSet *pSet, SRtn *pRtn) {
  SDiskID   did;
C
update  
Cary Xu 已提交
95
  SDFileSet nSet = {0};
L
Liu Jicong 已提交
96
  STsdbFS  *pfs = REPO_FS(pRepo);
H
refact  
Hongze Cheng 已提交
97 98 99 100 101 102
  int       level;

  ASSERT(pSet->fid >= pRtn->minFid);

  level = tsdbGetFidLevel(pSet->fid, pRtn);

H
Hongze Cheng 已提交
103
  if (tfsAllocDisk(pRepo->pVnode->pTfs, level, &did) < 0) {
H
refact  
Hongze Cheng 已提交
104 105 106 107 108 109
    terrno = TSDB_CODE_TDB_NO_AVAIL_DISK;
    return -1;
  }

  if (did.level > TSDB_FSET_LEVEL(pSet)) {
    // Need to move the FSET to higher level
S
Shengliang Guan 已提交
110
    tsdbInitDFileSet(pRepo, &nSet, did, pSet->fid, FS_TXN_VERSION(pfs));
H
refact  
Hongze Cheng 已提交
111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132

    if (tsdbCopyDFileSet(pSet, &nSet) < 0) {
      tsdbError("vgId:%d failed to copy FSET %d from level %d to level %d since %s", REPO_ID(pRepo), pSet->fid,
                TSDB_FSET_LEVEL(pSet), did.level, tstrerror(terrno));
      return -1;
    }

    if (tsdbUpdateDFileSet(pfs, &nSet) < 0) {
      return -1;
    }

    tsdbInfo("vgId:%d FSET %d is copied from level %d disk id %d to level %d disk id %d", REPO_ID(pRepo), pSet->fid,
             TSDB_FSET_LEVEL(pSet), TSDB_FSET_ID(pSet), did.level, did.id);
  } else {
    // On a correct level
    if (tsdbUpdateDFileSet(pfs, pSet) < 0) {
      return -1;
    }
  }

  return 0;
}
H
more  
Hongze Cheng 已提交
133

H
more  
Hongze Cheng 已提交
134 135 136 137 138 139 140
int tsdbPrepareCommit(STsdb *pTsdb) {
  if (pTsdb->mem == NULL) return 0;

  ASSERT(pTsdb->imem == NULL);

  pTsdb->imem = pTsdb->mem;
  pTsdb->mem = NULL;
L
Liu Jicong 已提交
141
  return 0;
H
more  
Hongze Cheng 已提交
142 143
}

H
refact  
Hongze Cheng 已提交
144
int tsdbCommit(STsdb *pRepo) {
H
Hongze Cheng 已提交
145 146 147
  SCommitH   commith = {0};
  SDFileSet *pSet = NULL;
  int        fid;
H
refact  
Hongze Cheng 已提交
148

H
Hongze Cheng 已提交
149 150 151
  // if (pRepo->imem == NULL) return 0;
  pRepo->imem = pRepo->mem;
  pRepo->mem = NULL;
H
refact  
Hongze Cheng 已提交
152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208

  tsdbStartCommit(pRepo);
  // Resource initialization
  if (tsdbInitCommitH(&commith, pRepo) < 0) {
    return -1;
  }

  // Skip expired memory data and expired FSET
  tsdbSeekCommitIter(&commith, commith.rtn.minKey);
  while ((pSet = tsdbFSIterNext(&(commith.fsIter)))) {
    if (pSet->fid < commith.rtn.minFid) {
      tsdbInfo("vgId:%d FSET %d on level %d disk id %d expires, remove it", REPO_ID(pRepo), pSet->fid,
               TSDB_FSET_LEVEL(pSet), TSDB_FSET_ID(pSet));
    } else {
      break;
    }
  }

  // Loop to commit to each file
  fid = tsdbNextCommitFid(&(commith));
  while (true) {
    // Loop over both on disk and memory
    if (pSet == NULL && fid == TSDB_IVLD_FID) break;

    if (pSet && (fid == TSDB_IVLD_FID || pSet->fid < fid)) {
      // Only has existing FSET but no memory data to commit in this
      // existing FSET, only check if file in correct retention
      if (tsdbApplyRtnOnFSet(pRepo, pSet, &(commith.rtn)) < 0) {
        tsdbDestroyCommitH(&commith);
        return -1;
      }

      pSet = tsdbFSIterNext(&(commith.fsIter));
    } else {
      // Has memory data to commit
      SDFileSet *pCSet;
      int        cfid;

      if (pSet == NULL || pSet->fid > fid) {
        // Commit to a new FSET with fid: fid
        pCSet = NULL;
        cfid = fid;
      } else {
        // Commit to an existing FSET
        pCSet = pSet;
        cfid = pSet->fid;
        pSet = tsdbFSIterNext(&(commith.fsIter));
      }

      if (tsdbCommitToFile(&commith, pCSet, cfid) < 0) {
        tsdbDestroyCommitH(&commith);
        return -1;
      }

      fid = tsdbNextCommitFid(&commith);
    }
  }
H
more  
Hongze Cheng 已提交
209

H
refact  
Hongze Cheng 已提交
210 211
  tsdbDestroyCommitH(&commith);
  tsdbEndCommit(pRepo, TSDB_CODE_SUCCESS);
H
more  
Hongze Cheng 已提交
212

H
Hongze Cheng 已提交
213
  return 0;
H
Hongze Cheng 已提交
214 215
}

H
Hongze Cheng 已提交
216 217 218 219 220
void tsdbGetRtnSnap(STsdb *pRepo, SRtn *pRtn) {
  STsdbCfg *pCfg = REPO_CFG(pRepo);
  TSKEY     minKey, midKey, maxKey, now;

  now = taosGetTimestamp(pCfg->precision);
H
Hongze Cheng 已提交
221 222 223
  minKey = now - pCfg->keep2 * tsTickPerDay[pCfg->precision];
  midKey = now - pCfg->keep1 * tsTickPerDay[pCfg->precision];
  maxKey = now - pCfg->keep0 * tsTickPerDay[pCfg->precision];
H
Hongze Cheng 已提交
224 225

  pRtn->minKey = minKey;
H
refact  
Hongze Cheng 已提交
226 227 228
  pRtn->minFid = (int)(TSDB_KEY_FID(minKey, pCfg->days, pCfg->precision));
  pRtn->midFid = (int)(TSDB_KEY_FID(midKey, pCfg->days, pCfg->precision));
  pRtn->maxFid = (int)(TSDB_KEY_FID(maxKey, pCfg->days, pCfg->precision));
H
Hongze Cheng 已提交
229 230 231 232
  tsdbDebug("vgId:%d now:%" PRId64 " minKey:%" PRId64 " minFid:%d, midFid:%d, maxFid:%d", REPO_ID(pRepo), now, minKey,
            pRtn->minFid, pRtn->midFid, pRtn->maxFid);
}

H
more  
Hongze Cheng 已提交
233 234 235 236 237 238 239 240 241 242
static void tsdbStartCommit(STsdb *pRepo) {
  STsdbMemTable *pMem = pRepo->imem;

  tsdbInfo("vgId:%d start to commit", REPO_ID(pRepo));

  tsdbStartFSTxn(pRepo, 0, 0);
}

static void tsdbEndCommit(STsdb *pTsdb, int eno) {
  tsdbEndFSTxn(pTsdb);
H
Hongze Cheng 已提交
243
  tsdbMemTableDestroy(pTsdb, pTsdb->imem);
H
more  
Hongze Cheng 已提交
244 245 246 247
  pTsdb->imem = NULL;
  tsdbInfo("vgId:%d commit over, %s", REPO_ID(pTsdb), (eno == TSDB_CODE_SUCCESS) ? "succeed" : "failed");
}

H
refact  
Hongze Cheng 已提交
248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289
static int tsdbInitCommitH(SCommitH *pCommith, STsdb *pRepo) {
  STsdbCfg *pCfg = REPO_CFG(pRepo);

  memset(pCommith, 0, sizeof(*pCommith));
  tsdbGetRtnSnap(pRepo, &(pCommith->rtn));

  TSDB_FSET_SET_CLOSED(TSDB_COMMIT_WRITE_FSET(pCommith));

  // Init read handle
  if (tsdbInitReadH(&(pCommith->readh), pRepo) < 0) {
    return -1;
  }

  // Init file iterator
  tsdbFSIterInit(&(pCommith->fsIter), REPO_FS(pRepo), TSDB_FS_ITER_FORWARD);

  if (tsdbCreateCommitIters(pCommith) < 0) {
    tsdbDestroyCommitH(pCommith);
    return -1;
  }

  pCommith->aBlkIdx = taosArrayInit(1024, sizeof(SBlockIdx));
  if (pCommith->aBlkIdx == NULL) {
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    tsdbDestroyCommitH(pCommith);
    return -1;
  }

  pCommith->aSupBlk = taosArrayInit(1024, sizeof(SBlock));
  if (pCommith->aSupBlk == NULL) {
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    tsdbDestroyCommitH(pCommith);
    return -1;
  }

  pCommith->aSubBlk = taosArrayInit(1024, sizeof(SBlock));
  if (pCommith->aSubBlk == NULL) {
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    tsdbDestroyCommitH(pCommith);
    return -1;
  }

H
refact  
Hongze Cheng 已提交
290
  pCommith->pDataCols = tdNewDataCols(0, pCfg->maxRows);
H
refact  
Hongze Cheng 已提交
291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309
  if (pCommith->pDataCols == NULL) {
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    tsdbDestroyCommitH(pCommith);
    return -1;
  }

  return 0;
}

// Skip all keys until key (not included)
static void tsdbSeekCommitIter(SCommitH *pCommith, TSKEY key) {
  for (int i = 0; i < pCommith->niters; i++) {
    SCommitIter *pIter = pCommith->iters + i;
    if (pIter->pTable == NULL || pIter->pIter == NULL) continue;

    tsdbLoadDataFromCache(pIter->pTable, pIter->pIter, key - 1, INT32_MAX, NULL, NULL, 0, true, NULL);
  }
}

H
more  
Hongze Cheng 已提交
310
static int tsdbNextCommitFid(SCommitH *pCommith) {
L
Liu Jicong 已提交
311
  STsdb    *pRepo = TSDB_COMMIT_REPO(pCommith);
H
more  
Hongze Cheng 已提交
312 313 314 315 316 317 318 319 320 321 322
  STsdbCfg *pCfg = REPO_CFG(pRepo);
  int       fid = TSDB_IVLD_FID;

  for (int i = 0; i < pCommith->niters; i++) {
    SCommitIter *pIter = pCommith->iters + i;
    // if (pIter->pTable == NULL || pIter->pIter == NULL) continue;

    TSKEY nextKey = tsdbNextIterKey(pIter->pIter);
    if (nextKey == TSDB_DATA_TIMESTAMP_NULL) {
      continue;
    } else {
H
refact  
Hongze Cheng 已提交
323
      int tfid = (int)(TSDB_KEY_FID(nextKey, pCfg->days, pCfg->precision));
H
more  
Hongze Cheng 已提交
324 325 326 327 328 329 330 331
      if (fid == TSDB_IVLD_FID || fid > tfid) {
        fid = tfid;
      }
    }
  }

  return fid;
}
H
refact  
Hongze Cheng 已提交
332 333 334 335 336 337 338 339 340 341 342

static void tsdbDestroyCommitH(SCommitH *pCommith) {
  pCommith->pDataCols = tdFreeDataCols(pCommith->pDataCols);
  pCommith->aSubBlk = taosArrayDestroy(pCommith->aSubBlk);
  pCommith->aSupBlk = taosArrayDestroy(pCommith->aSupBlk);
  pCommith->aBlkIdx = taosArrayDestroy(pCommith->aBlkIdx);
  tsdbDestroyCommitIters(pCommith);
  tsdbDestroyReadH(&(pCommith->readh));
  tsdbCloseDFileSet(TSDB_COMMIT_WRITE_FSET(pCommith));
}

H
more  
Hongze Cheng 已提交
343
static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid) {
L
Liu Jicong 已提交
344
  STsdb    *pRepo = TSDB_COMMIT_REPO(pCommith);
H
more  
Hongze Cheng 已提交
345
  STsdbCfg *pCfg = REPO_CFG(pRepo);
H
Hongze Cheng 已提交
346

H
more  
Hongze Cheng 已提交
347
  ASSERT(pSet == NULL || pSet->fid == fid);
H
Hongze Cheng 已提交
348

H
more  
Hongze Cheng 已提交
349
  tsdbResetCommitFile(pCommith);
H
refact  
Hongze Cheng 已提交
350
  tsdbGetFidKeyRange(pCfg->days, pCfg->precision, fid, &(pCommith->minKey), &(pCommith->maxKey));
H
Hongze Cheng 已提交
351

H
more  
Hongze Cheng 已提交
352 353 354 355
  // Set and open files
  if (tsdbSetAndOpenCommitFile(pCommith, pSet, fid) < 0) {
    return -1;
  }
H
Hongze Cheng 已提交
356

H
more  
Hongze Cheng 已提交
357
  // Loop to commit each table data
H
more  
Hongze Cheng 已提交
358
  for (int tid = 0; tid < pCommith->niters; tid++) {
H
more  
Hongze Cheng 已提交
359
    SCommitIter *pIter = pCommith->iters + tid;
H
Hongze Cheng 已提交
360

H
more  
Hongze Cheng 已提交
361
    if (pIter->pTable == NULL) continue;
H
Hongze Cheng 已提交
362

H
more  
Hongze Cheng 已提交
363 364 365 366 367 368 369
    if (tsdbCommitToTable(pCommith, tid) < 0) {
      tsdbCloseCommitFile(pCommith, true);
      // revert the file change
      tsdbApplyDFileSetChange(TSDB_COMMIT_WRITE_FSET(pCommith), pSet);
      return -1;
    }
  }
H
Hongze Cheng 已提交
370

H
more  
Hongze Cheng 已提交
371 372 373 374 375 376 377 378
  if (tsdbWriteBlockIdx(TSDB_COMMIT_HEAD_FILE(pCommith), pCommith->aBlkIdx, (void **)(&(TSDB_COMMIT_BUF(pCommith)))) <
      0) {
    tsdbError("vgId:%d failed to write SBlockIdx part to FSET %d since %s", REPO_ID(pRepo), fid, tstrerror(terrno));
    tsdbCloseCommitFile(pCommith, true);
    // revert the file change
    tsdbApplyDFileSetChange(TSDB_COMMIT_WRITE_FSET(pCommith), pSet);
    return -1;
  }
H
Hongze Cheng 已提交
379

H
more  
Hongze Cheng 已提交
380 381 382 383 384 385 386
  if (tsdbUpdateDFileSetHeader(&(pCommith->wSet)) < 0) {
    tsdbError("vgId:%d failed to update FSET %d header since %s", REPO_ID(pRepo), fid, tstrerror(terrno));
    tsdbCloseCommitFile(pCommith, true);
    // revert the file change
    tsdbApplyDFileSetChange(TSDB_COMMIT_WRITE_FSET(pCommith), pSet);
    return -1;
  }
H
Hongze Cheng 已提交
387

H
more  
Hongze Cheng 已提交
388 389
  // Close commit file
  tsdbCloseCommitFile(pCommith, false);
H
Hongze Cheng 已提交
390

H
more  
Hongze Cheng 已提交
391 392 393
  if (tsdbUpdateDFileSet(REPO_FS(pRepo), &(pCommith->wSet)) < 0) {
    return -1;
  }
H
Hongze Cheng 已提交
394

H
more  
Hongze Cheng 已提交
395 396
  return 0;
}
H
refact  
Hongze Cheng 已提交
397

H
Hongze Cheng 已提交
398
static int tsdbCreateCommitIters(SCommitH *pCommith) {
L
Liu Jicong 已提交
399 400
  STsdb             *pRepo = TSDB_COMMIT_REPO(pCommith);
  STsdbMemTable     *pMem = pRepo->imem;
H
Hongze Cheng 已提交
401
  SSkipListIterator *pSlIter;
L
Liu Jicong 已提交
402 403 404
  SCommitIter       *pCommitIter;
  SSkipListNode     *pNode;
  STbData           *pTbData;
H
Hongze Cheng 已提交
405 406

  pCommith->niters = SL_SIZE(pMem->pSlIdx);
wafwerar's avatar
wafwerar 已提交
407
  pCommith->iters = (SCommitIter *)taosMemoryCalloc(pCommith->niters, sizeof(SCommitIter));
H
Hongze Cheng 已提交
408 409
  if (pCommith->iters == NULL) {
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
H
refact  
Hongze Cheng 已提交
410 411 412
    return -1;
  }

H
Hongze Cheng 已提交
413 414 415 416
  // Loop to create iters for each skiplist
  pSlIter = tSkipListCreateIter(pMem->pSlIdx);
  if (pSlIter == NULL) {
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
H
refact  
Hongze Cheng 已提交
417 418
    return -1;
  }
H
Hongze Cheng 已提交
419 420 421 422
  for (int i = 0; i < pCommith->niters; i++) {
    tSkipListIterNext(pSlIter);
    pNode = tSkipListIterGet(pSlIter);
    pTbData = (STbData *)pNode->pData;
H
refact  
Hongze Cheng 已提交
423

H
Hongze Cheng 已提交
424 425 426
    pCommitIter = pCommith->iters + i;
    pCommitIter->pIter = tSkipListCreateIter(pTbData->pData);
    tSkipListIterNext(pCommitIter->pIter);
H
more  
Hongze Cheng 已提交
427

wafwerar's avatar
wafwerar 已提交
428
    pCommitIter->pTable = (STable *)taosMemoryMalloc(sizeof(STable));
H
more  
Hongze Cheng 已提交
429
    pCommitIter->pTable->uid = pTbData->uid;
H
more  
Hongze Cheng 已提交
430
    pCommitIter->pTable->tid = pTbData->uid;
H
Hongze Cheng 已提交
431
    pCommitIter->pTable->pSchema = metaGetTbTSchema(REPO_META(pRepo), pTbData->uid, 0);
H
refact  
Hongze Cheng 已提交
432 433 434 435 436
  }

  return 0;
}

H
Hongze Cheng 已提交
437 438
static void tsdbDestroyCommitIters(SCommitH *pCommith) {
  if (pCommith->iters == NULL) return;
H
refact  
Hongze Cheng 已提交
439

H
Hongze Cheng 已提交
440 441
  for (int i = 1; i < pCommith->niters; i++) {
    tSkipListDestroyIter(pCommith->iters[i].pIter);
H
more  
Hongze Cheng 已提交
442
    tdFreeSchema(pCommith->iters[i].pTable->pSchema);
wafwerar's avatar
wafwerar 已提交
443
    taosMemoryFree(pCommith->iters[i].pTable);
H
refact  
Hongze Cheng 已提交
444 445
  }

wafwerar's avatar
wafwerar 已提交
446
  taosMemoryFree(pCommith->iters);
H
Hongze Cheng 已提交
447 448
  pCommith->iters = NULL;
  pCommith->niters = 0;
H
refact  
Hongze Cheng 已提交
449 450
}

H
more  
Hongze Cheng 已提交
451 452 453 454 455 456
static void tsdbResetCommitFile(SCommitH *pCommith) {
  pCommith->isRFileSet = false;
  pCommith->isDFileSame = false;
  pCommith->isLFileSame = false;
  taosArrayClear(pCommith->aBlkIdx);
}
H
Hongze Cheng 已提交
457

H
more  
Hongze Cheng 已提交
458 459
static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid) {
  SDiskID    did;
L
Liu Jicong 已提交
460
  STsdb     *pRepo = TSDB_COMMIT_REPO(pCommith);
H
more  
Hongze Cheng 已提交
461
  SDFileSet *pWSet = TSDB_COMMIT_WRITE_FSET(pCommith);
H
Hongze Cheng 已提交
462

H
Hongze Cheng 已提交
463
  if (tfsAllocDisk(REPO_TFS(pRepo), tsdbGetFidLevel(fid, &(pCommith->rtn)), &did) < 0) {
H
more  
Hongze Cheng 已提交
464 465 466
    terrno = TSDB_CODE_TDB_NO_AVAIL_DISK;
    return -1;
  }
H
Hongze Cheng 已提交
467

H
more  
Hongze Cheng 已提交
468 469 470 471 472
  // Open read FSET
  if (pSet) {
    if (tsdbSetAndOpenReadFSet(&(pCommith->readh), pSet) < 0) {
      return -1;
    }
H
Hongze Cheng 已提交
473

H
more  
Hongze Cheng 已提交
474
    pCommith->isRFileSet = true;
H
Hongze Cheng 已提交
475

H
more  
Hongze Cheng 已提交
476 477 478 479
    if (tsdbLoadBlockIdx(&(pCommith->readh)) < 0) {
      tsdbCloseAndUnsetFSet(&(pCommith->readh));
      return -1;
    }
H
Hongze Cheng 已提交
480

H
more  
Hongze Cheng 已提交
481 482 483 484 485
    tsdbDebug("vgId:%d FSET %d at level %d disk id %d is opened to read to commit", REPO_ID(pRepo), TSDB_FSET_FID(pSet),
              TSDB_FSET_LEVEL(pSet), TSDB_FSET_ID(pSet));
  } else {
    pCommith->isRFileSet = false;
  }
H
Hongze Cheng 已提交
486

H
more  
Hongze Cheng 已提交
487 488 489
  // Set and open commit FSET
  if (pSet == NULL || did.level > TSDB_FSET_LEVEL(pSet)) {
    // Create a new FSET to write data
S
Shengliang Guan 已提交
490
    tsdbInitDFileSet(pRepo, pWSet, did, fid, FS_TXN_VERSION(REPO_FS(pRepo)));
H
Hongze Cheng 已提交
491

S
Shengliang Guan 已提交
492
    if (tsdbCreateDFileSet(pRepo, pWSet, true) < 0) {
H
more  
Hongze Cheng 已提交
493 494 495 496 497 498 499
      tsdbError("vgId:%d failed to create FSET %d at level %d disk id %d since %s", REPO_ID(pRepo),
                TSDB_FSET_FID(pWSet), TSDB_FSET_LEVEL(pWSet), TSDB_FSET_ID(pWSet), tstrerror(terrno));
      if (pCommith->isRFileSet) {
        tsdbCloseAndUnsetFSet(&(pCommith->readh));
      }
      return -1;
    }
H
Hongze Cheng 已提交
500

H
more  
Hongze Cheng 已提交
501 502
    pCommith->isDFileSame = false;
    pCommith->isLFileSame = false;
H
Hongze Cheng 已提交
503

H
more  
Hongze Cheng 已提交
504 505 506 507 508
    tsdbDebug("vgId:%d FSET %d at level %d disk id %d is created to commit", REPO_ID(pRepo), TSDB_FSET_FID(pWSet),
              TSDB_FSET_LEVEL(pWSet), TSDB_FSET_ID(pWSet));
  } else {
    did.level = TSDB_FSET_LEVEL(pSet);
    did.id = TSDB_FSET_ID(pSet);
H
Hongze Cheng 已提交
509

H
more  
Hongze Cheng 已提交
510 511
    pCommith->wSet.fid = fid;
    pCommith->wSet.state = 0;
H
Hongze Cheng 已提交
512

H
more  
Hongze Cheng 已提交
513 514
    // TSDB_FILE_HEAD
    SDFile *pWHeadf = TSDB_COMMIT_HEAD_FILE(pCommith);
S
Shengliang Guan 已提交
515
    tsdbInitDFile(pRepo, pWHeadf, did, fid, FS_TXN_VERSION(REPO_FS(pRepo)), TSDB_FILE_HEAD);
516
    if (tsdbCreateDFile(pRepo, pWHeadf, true, TSDB_FILE_HEAD) < 0) {
H
more  
Hongze Cheng 已提交
517 518
      tsdbError("vgId:%d failed to create file %s to commit since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pWHeadf),
                tstrerror(terrno));
H
Hongze Cheng 已提交
519

H
more  
Hongze Cheng 已提交
520 521 522
      if (pCommith->isRFileSet) {
        tsdbCloseAndUnsetFSet(&(pCommith->readh));
        return -1;
H
Hongze Cheng 已提交
523 524 525
      }
    }

H
more  
Hongze Cheng 已提交
526 527 528 529
    // TSDB_FILE_DATA
    SDFile *pRDataf = TSDB_READ_DATA_FILE(&(pCommith->readh));
    SDFile *pWDataf = TSDB_COMMIT_DATA_FILE(pCommith);
    tsdbInitDFileEx(pWDataf, pRDataf);
530 531
    // if (tsdbOpenDFile(pWDataf, O_WRONLY) < 0) {
    if (tsdbOpenDFile(pWDataf, TD_FILE_WRITE) < 0) {
H
more  
Hongze Cheng 已提交
532 533
      tsdbError("vgId:%d failed to open file %s to commit since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pWDataf),
                tstrerror(terrno));
H
Hongze Cheng 已提交
534

H
more  
Hongze Cheng 已提交
535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550
      tsdbCloseDFileSet(pWSet);
      tsdbRemoveDFile(pWHeadf);
      if (pCommith->isRFileSet) {
        tsdbCloseAndUnsetFSet(&(pCommith->readh));
        return -1;
      }
    }
    pCommith->isDFileSame = true;

    // TSDB_FILE_LAST
    SDFile *pRLastf = TSDB_READ_LAST_FILE(&(pCommith->readh));
    SDFile *pWLastf = TSDB_COMMIT_LAST_FILE(pCommith);
    if (pRLastf->info.size < 32 * 1024) {
      tsdbInitDFileEx(pWLastf, pRLastf);
      pCommith->isLFileSame = true;

551 552
      // if (tsdbOpenDFile(pWLastf, O_WRONLY) < 0) {
      if (tsdbOpenDFile(pWLastf, TD_FILE_WRITE) < 0) {
H
more  
Hongze Cheng 已提交
553 554
        tsdbError("vgId:%d failed to open file %s to commit since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pWLastf),
                  tstrerror(terrno));
H
Hongze Cheng 已提交
555

H
more  
Hongze Cheng 已提交
556 557 558 559 560 561 562 563
        tsdbCloseDFileSet(pWSet);
        tsdbRemoveDFile(pWHeadf);
        if (pCommith->isRFileSet) {
          tsdbCloseAndUnsetFSet(&(pCommith->readh));
          return -1;
        }
      }
    } else {
S
Shengliang Guan 已提交
564
      tsdbInitDFile(pRepo, pWLastf, did, fid, FS_TXN_VERSION(REPO_FS(pRepo)), TSDB_FILE_LAST);
H
more  
Hongze Cheng 已提交
565
      pCommith->isLFileSame = false;
H
Hongze Cheng 已提交
566

567
      if (tsdbCreateDFile(pRepo, pWLastf, true, TSDB_FILE_LAST) < 0) {
H
more  
Hongze Cheng 已提交
568 569
        tsdbError("vgId:%d failed to create file %s to commit since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pWLastf),
                  tstrerror(terrno));
H
Hongze Cheng 已提交
570

H
more  
Hongze Cheng 已提交
571 572 573 574 575 576 577 578
        tsdbCloseDFileSet(pWSet);
        (void)tsdbRemoveDFile(pWHeadf);
        if (pCommith->isRFileSet) {
          tsdbCloseAndUnsetFSet(&(pCommith->readh));
          return -1;
        }
      }
    }
579 580 581 582 583

    // TSDB_FILE_SMAD
    SDFile *pRSmadF = TSDB_READ_SMAD_FILE(&(pCommith->readh));
    SDFile *pWSmadF = TSDB_COMMIT_SMAD_FILE(pCommith);

584
    if (!taosCheckExistFile(TSDB_FILE_FULL_NAME(pRSmadF))) {
585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617
      tsdbDebug("vgId:%d create data file %s as not exist", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pRSmadF));
      tsdbInitDFile(pRepo, pWSmadF, did, fid, FS_TXN_VERSION(REPO_FS(pRepo)), TSDB_FILE_SMAD);

      if (tsdbCreateDFile(pRepo, pWSmadF, true, TSDB_FILE_SMAD) < 0) {
        tsdbError("vgId:%d failed to create file %s to commit since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pWSmadF),
                  tstrerror(terrno));

        tsdbCloseDFileSet(pWSet);
        (void)tsdbRemoveDFile(pWHeadf);
        if (pCommith->isRFileSet) {
          tsdbCloseAndUnsetFSet(&(pCommith->readh));
          return -1;
        }
      }
    } else {
      tsdbInitDFileEx(pWSmadF, pRSmadF);
      if (tsdbOpenDFile(pWSmadF, O_RDWR) < 0) {
        tsdbError("vgId:%d failed to open file %s to commit since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pWSmadF),
                  tstrerror(terrno));

        tsdbCloseDFileSet(pWSet);
        tsdbRemoveDFile(pWHeadf);
        if (pCommith->isRFileSet) {
          tsdbCloseAndUnsetFSet(&(pCommith->readh));
          return -1;
        }
      }
    }

    // TSDB_FILE_SMAL
    SDFile *pRSmalF = TSDB_READ_SMAL_FILE(&(pCommith->readh));
    SDFile *pWSmalF = TSDB_COMMIT_SMAL_FILE(pCommith);

618
    if ((pCommith->isLFileSame) && taosCheckExistFile(TSDB_FILE_FULL_NAME(pRSmalF))) {
619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646
      tsdbInitDFileEx(pWSmalF, pRSmalF);
      if (tsdbOpenDFile(pWSmalF, O_RDWR) < 0) {
        tsdbError("vgId:%d failed to open file %s to commit since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pWSmalF),
                  tstrerror(terrno));

        tsdbCloseDFileSet(pWSet);
        tsdbRemoveDFile(pWHeadf);
        if (pCommith->isRFileSet) {
          tsdbCloseAndUnsetFSet(&(pCommith->readh));
          return -1;
        }
      }
    } else {
      tsdbDebug("vgId:%d create data file %s as not exist", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pRSmalF));
      tsdbInitDFile(pRepo, pWSmalF, did, fid, FS_TXN_VERSION(REPO_FS(pRepo)), TSDB_FILE_SMAL);

      if (tsdbCreateDFile(pRepo, pWSmalF, true, TSDB_FILE_SMAL) < 0) {
        tsdbError("vgId:%d failed to create file %s to commit since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pWSmalF),
                  tstrerror(terrno));

        tsdbCloseDFileSet(pWSet);
        (void)tsdbRemoveDFile(pWHeadf);
        if (pCommith->isRFileSet) {
          tsdbCloseAndUnsetFSet(&(pCommith->readh));
          return -1;
        }
      }
    }
H
more  
Hongze Cheng 已提交
647
  }
H
Hongze Cheng 已提交
648 649 650 651

  return 0;
}

H
more  
Hongze Cheng 已提交
652
// extern int32_t tsTsdbMetaCompactRatio;
H
Hongze Cheng 已提交
653

H
more  
Hongze Cheng 已提交
654 655 656 657 658 659 660
int tsdbWriteBlockInfoImpl(SDFile *pHeadf, STable *pTable, SArray *pSupA, SArray *pSubA, void **ppBuf,
                           SBlockIdx *pIdx) {
  size_t      nSupBlocks;
  size_t      nSubBlocks;
  uint32_t    tlen;
  SBlockInfo *pBlkInfo;
  int64_t     offset;
L
Liu Jicong 已提交
661
  SBlock     *pBlock;
H
Hongze Cheng 已提交
662

H
more  
Hongze Cheng 已提交
663
  memset(pIdx, 0, sizeof(*pIdx));
H
Hongze Cheng 已提交
664

H
more  
Hongze Cheng 已提交
665 666
  nSupBlocks = taosArrayGetSize(pSupA);
  nSubBlocks = (pSubA == NULL) ? 0 : taosArrayGetSize(pSubA);
H
Hongze Cheng 已提交
667

H
more  
Hongze Cheng 已提交
668 669 670 671
  if (nSupBlocks <= 0) {
    // No data (data all deleted)
    return 0;
  }
H
Hongze Cheng 已提交
672

H
more  
Hongze Cheng 已提交
673 674 675
  tlen = (uint32_t)(sizeof(SBlockInfo) + sizeof(SBlock) * (nSupBlocks + nSubBlocks) + sizeof(TSCKSUM));
  if (tsdbMakeRoom(ppBuf, tlen) < 0) return -1;
  pBlkInfo = *ppBuf;
H
Hongze Cheng 已提交
676

H
more  
Hongze Cheng 已提交
677 678 679
  pBlkInfo->delimiter = TSDB_FILE_DELIMITER;
  pBlkInfo->tid = TABLE_TID(pTable);
  pBlkInfo->uid = TABLE_UID(pTable);
H
Hongze Cheng 已提交
680

H
more  
Hongze Cheng 已提交
681 682 683
  memcpy((void *)(pBlkInfo->blocks), taosArrayGet(pSupA, 0), nSupBlocks * sizeof(SBlock));
  if (nSubBlocks > 0) {
    memcpy((void *)(pBlkInfo->blocks + nSupBlocks), taosArrayGet(pSubA, 0), nSubBlocks * sizeof(SBlock));
H
Hongze Cheng 已提交
684

H
more  
Hongze Cheng 已提交
685 686
    for (int i = 0; i < nSupBlocks; i++) {
      pBlock = pBlkInfo->blocks + i;
H
Hongze Cheng 已提交
687

H
more  
Hongze Cheng 已提交
688 689 690 691 692
      if (pBlock->numOfSubBlocks > 1) {
        pBlock->offset += (sizeof(SBlockInfo) + sizeof(SBlock) * nSupBlocks);
      }
    }
  }
H
Hongze Cheng 已提交
693

H
more  
Hongze Cheng 已提交
694
  taosCalcChecksumAppend(0, (uint8_t *)pBlkInfo, tlen);
H
Hongze Cheng 已提交
695

H
more  
Hongze Cheng 已提交
696 697 698
  if (tsdbAppendDFile(pHeadf, (void *)pBlkInfo, tlen, &offset) < 0) {
    return -1;
  }
H
Hongze Cheng 已提交
699

H
more  
Hongze Cheng 已提交
700
  tsdbUpdateDFileMagic(pHeadf, POINTER_SHIFT(pBlkInfo, tlen - sizeof(TSCKSUM)));
H
Hongze Cheng 已提交
701

H
more  
Hongze Cheng 已提交
702 703
  // Set pIdx
  pBlock = taosArrayGetLast(pSupA);
H
Hongze Cheng 已提交
704

H
more  
Hongze Cheng 已提交
705 706 707 708 709 710
  pIdx->uid = TABLE_UID(pTable);
  pIdx->hasLast = pBlock->last ? 1 : 0;
  pIdx->maxKey = pBlock->keyLast;
  pIdx->numOfBlocks = (uint32_t)nSupBlocks;
  pIdx->len = tlen;
  pIdx->offset = (uint32_t)offset;
H
Hongze Cheng 已提交
711

H
more  
Hongze Cheng 已提交
712 713
  return 0;
}
H
Hongze Cheng 已提交
714

H
more  
Hongze Cheng 已提交
715 716 717 718 719
int tsdbWriteBlockIdx(SDFile *pHeadf, SArray *pIdxA, void **ppBuf) {
  SBlockIdx *pBlkIdx;
  size_t     nidx = taosArrayGetSize(pIdxA);
  int        tlen = 0, size;
  int64_t    offset;
H
more  
Hongze Cheng 已提交
720

H
more  
Hongze Cheng 已提交
721 722 723 724 725 726
  if (nidx <= 0) {
    // All data are deleted
    pHeadf->info.offset = 0;
    pHeadf->info.len = 0;
    return 0;
  }
H
Hongze Cheng 已提交
727

H
more  
Hongze Cheng 已提交
728 729
  for (size_t i = 0; i < nidx; i++) {
    pBlkIdx = (SBlockIdx *)taosArrayGet(pIdxA, i);
H
Hongze Cheng 已提交
730

H
more  
Hongze Cheng 已提交
731 732
    size = tsdbEncodeSBlockIdx(NULL, pBlkIdx);
    if (tsdbMakeRoom(ppBuf, tlen + size) < 0) return -1;
H
Hongze Cheng 已提交
733

H
more  
Hongze Cheng 已提交
734 735
    void *ptr = POINTER_SHIFT(*ppBuf, tlen);
    tsdbEncodeSBlockIdx(&ptr, pBlkIdx);
H
Hongze Cheng 已提交
736

H
more  
Hongze Cheng 已提交
737 738
    tlen += size;
  }
H
Hongze Cheng 已提交
739

H
more  
Hongze Cheng 已提交
740 741 742
  tlen += sizeof(TSCKSUM);
  if (tsdbMakeRoom(ppBuf, tlen) < 0) return -1;
  taosCalcChecksumAppend(0, (uint8_t *)(*ppBuf), tlen);
H
Hongze Cheng 已提交
743

H
more  
Hongze Cheng 已提交
744 745 746
  if (tsdbAppendDFile(pHeadf, *ppBuf, tlen, &offset) < tlen) {
    return -1;
  }
H
Hongze Cheng 已提交
747

H
more  
Hongze Cheng 已提交
748 749 750
  tsdbUpdateDFileMagic(pHeadf, POINTER_SHIFT(*ppBuf, tlen - sizeof(TSCKSUM)));
  pHeadf->info.offset = (uint32_t)offset;
  pHeadf->info.len = tlen;
H
Hongze Cheng 已提交
751

H
more  
Hongze Cheng 已提交
752 753
  return 0;
}
H
Hongze Cheng 已提交
754

H
Hongze Cheng 已提交
755
// =================== Commit Time-Series Data
H
Hongze Cheng 已提交
756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 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
static int tsdbCommitToTable(SCommitH *pCommith, int tid) {
  SCommitIter *pIter = pCommith->iters + tid;
  TSKEY        nextKey = tsdbNextIterKey(pIter->pIter);

  tsdbResetCommitTable(pCommith);

  // Set commit table
  if (tsdbSetCommitTable(pCommith, pIter->pTable) < 0) {
    return -1;
  }

  // No disk data and no memory data, just return
  if (pCommith->readh.pBlkIdx == NULL && (nextKey == TSDB_DATA_TIMESTAMP_NULL || nextKey > pCommith->maxKey)) {
    return 0;
  }

  // Must has disk data or has memory data
  int     nBlocks;
  int     bidx = 0;
  SBlock *pBlock;

  if (pCommith->readh.pBlkIdx) {
    if (tsdbLoadBlockInfo(&(pCommith->readh), NULL) < 0) {
      return -1;
    }

    nBlocks = pCommith->readh.pBlkIdx->numOfBlocks;
  } else {
    nBlocks = 0;
  }

  if (bidx < nBlocks) {
    pBlock = pCommith->readh.pBlkInfo->blocks + bidx;
  } else {
    pBlock = NULL;
  }

  while (true) {
    if (pBlock == NULL && (nextKey == TSDB_DATA_TIMESTAMP_NULL || nextKey > pCommith->maxKey)) break;

    if ((nextKey == TSDB_DATA_TIMESTAMP_NULL || nextKey > pCommith->maxKey) ||
        (pBlock && (!pBlock->last) && tsdbComparKeyBlock((void *)(&nextKey), pBlock) > 0)) {
      if (tsdbMoveBlock(pCommith, bidx) < 0) {
        return -1;
      }

      bidx++;
      if (bidx < nBlocks) {
        pBlock = pCommith->readh.pBlkInfo->blocks + bidx;
      } else {
        pBlock = NULL;
      }
    } else if (pBlock && (pBlock->last || tsdbComparKeyBlock((void *)(&nextKey), pBlock) == 0)) {
H
more  
Hongze Cheng 已提交
809 810 811 812
      // merge pBlock data and memory data
      if (tsdbMergeMemData(pCommith, pIter, bidx) < 0) {
        return -1;
      }
H
more  
Hongze Cheng 已提交
813

H
more  
Hongze Cheng 已提交
814 815 816 817 818 819 820
      bidx++;
      if (bidx < nBlocks) {
        pBlock = pCommith->readh.pBlkInfo->blocks + bidx;
      } else {
        pBlock = NULL;
      }
      nextKey = tsdbNextIterKey(pIter->pIter);
H
Hongze Cheng 已提交
821
    } else {
H
more  
Hongze Cheng 已提交
822 823 824 825 826 827 828 829 830 831 832
      // Only commit memory data
      if (pBlock == NULL) {
        if (tsdbCommitMemData(pCommith, pIter, pCommith->maxKey, false) < 0) {
          return -1;
        }
      } else {
        if (tsdbCommitMemData(pCommith, pIter, pBlock->keyFirst - 1, true) < 0) {
          return -1;
        }
      }
      nextKey = tsdbNextIterKey(pIter->pIter);
H
Hongze Cheng 已提交
833 834 835
    }
  }

H
more  
Hongze Cheng 已提交
836 837 838 839 840
  if (tsdbWriteBlockInfo(pCommith) < 0) {
    tsdbError("vgId:%d failed to write SBlockInfo part into file %s since %s", TSDB_COMMIT_REPO_ID(pCommith),
              TSDB_FILE_FULL_NAME(TSDB_COMMIT_HEAD_FILE(pCommith)), tstrerror(terrno));
    return -1;
  }
H
Hongze Cheng 已提交
841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877

  return 0;
}

static int tsdbSetCommitTable(SCommitH *pCommith, STable *pTable) {
  STSchema *pSchema = tsdbGetTableSchemaImpl(pTable, false, false, -1);

  pCommith->pTable = pTable;

  if (tdInitDataCols(pCommith->pDataCols, pSchema) < 0) {
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    return -1;
  }

  if (pCommith->isRFileSet) {
    if (tsdbSetReadTable(&(pCommith->readh), pTable) < 0) {
      return -1;
    }
  } else {
    pCommith->readh.pBlkIdx = NULL;
  }
  return 0;
}

static int tsdbComparKeyBlock(const void *arg1, const void *arg2) {
  TSKEY   key = *(TSKEY *)arg1;
  SBlock *pBlock = (SBlock *)arg2;

  if (key < pBlock->keyFirst) {
    return -1;
  } else if (key > pBlock->keyLast) {
    return 1;
  } else {
    return 0;
  }
}

C
Cary Xu 已提交
878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894
/**
 * @brief Write SDataCols to data file.
 *
 * @param pRepo
 * @param pTable
 * @param pDFile
 * @param pDFileAggr
 * @param pDataCols The pDataCols would be generated from mem/imem directly with 2 bits bitmap or from tsdbRead
 * interface with 1 bit bitmap.
 * @param pBlock
 * @param isLast
 * @param isSuper
 * @param ppBuf
 * @param ppCBuf
 * @param ppExBuf
 * @return int
 */
895 896
int tsdbWriteBlockImpl(STsdb *pRepo, STable *pTable, SDFile *pDFile, SDFile *pDFileAggr, SDataCols *pDataCols,
                       SBlock *pBlock, bool isLast, bool isSuper, void **ppBuf, void **ppCBuf, void **ppExBuf) {
C
Cary Xu 已提交
897 898
  STsdbCfg     *pCfg = REPO_CFG(pRepo);
  SBlockData   *pBlockData = NULL;
899
  SAggrBlkData *pAggrBlkData = NULL;
C
Cary Xu 已提交
900
  STSchema     *pSchema = pTable->pSchema;
901 902
  int64_t       offset = 0, offsetAggr = 0;
  int           rowsToWrite = pDataCols->numOfRows;
H
Hongze Cheng 已提交
903

H
refact  
Hongze Cheng 已提交
904 905
  ASSERT(rowsToWrite > 0 && rowsToWrite <= pCfg->maxRows);
  ASSERT((!isLast) || rowsToWrite < pCfg->minRows);
H
Hongze Cheng 已提交
906 907

  // Make buffer space
908
  if (tsdbMakeRoom(ppBuf, tsdbBlockStatisSize(pDataCols->numOfCols, SBlockVerLatest)) < 0) {
H
Hongze Cheng 已提交
909 910 911 912
    return -1;
  }
  pBlockData = (SBlockData *)(*ppBuf);

913 914 915 916 917
  if (tsdbMakeRoom(ppExBuf, tsdbBlockAggrSize(pDataCols->numOfCols, SBlockVerLatest)) < 0) {
    return -1;
  }
  pAggrBlkData = (SAggrBlkData *)(*ppExBuf);

H
Hongze Cheng 已提交
918
  // Get # of cols not all NULL(not including key column)
C
Cary Xu 已提交
919 920
  col_id_t nColsNotAllNull = 0;
  col_id_t nColsOfBlockSma = 0;
921
  for (int ncol = 1; ncol < pDataCols->numOfCols; ++ncol) {  // ncol from 1, we skip the timestamp column
C
Cary Xu 已提交
922 923 924
    STColumn    *pColumn = pSchema->columns + ncol;
    SDataCol    *pDataCol = pDataCols->cols + ncol;
    SBlockCol   *pBlockCol = pBlockData->cols + nColsNotAllNull;
C
Cary Xu 已提交
925
    SAggrBlkCol *pAggrBlkCol = (SAggrBlkCol *)pAggrBlkData + nColsOfBlockSma;
H
Hongze Cheng 已提交
926 927 928 929 930 931

    if (isAllRowsNull(pDataCol)) {  // all data to commit are NULL, just ignore it
      continue;
    }

    memset(pBlockCol, 0, sizeof(*pBlockCol));
932
    memset(pAggrBlkCol, 0, sizeof(*pAggrBlkCol));
H
Hongze Cheng 已提交
933 934 935

    pBlockCol->colId = pDataCol->colId;
    pBlockCol->type = pDataCol->type;
936 937
    pAggrBlkCol->colId = pDataCol->colId;

C
Cary Xu 已提交
938
    if (isSuper && IS_BSMA_ON(pColumn) && tDataTypes[pDataCol->type].statisFunc) {
939
#if 0
H
Hongze Cheng 已提交
940 941 942
      (*tDataTypes[pDataCol->type].statisFunc)(pDataCol->pData, rowsToWrite, &(pBlockCol->min), &(pBlockCol->max),
                                               &(pBlockCol->sum), &(pBlockCol->minIndex), &(pBlockCol->maxIndex),
                                               &(pBlockCol->numOfNull));
943
#endif
C
Cary Xu 已提交
944 945 946
      (*tDataTypes[pDataCol->type].statisFunc)(pDataCols->bitmapMode, pDataCol->pBitmap, pDataCol->pData, rowsToWrite,
                                               &(pAggrBlkCol->min), &(pAggrBlkCol->max), &(pAggrBlkCol->sum),
                                               &(pAggrBlkCol->minIndex), &(pAggrBlkCol->maxIndex),
947 948 949
                                               &(pAggrBlkCol->numOfNull));

      if (pAggrBlkCol->numOfNull == 0) {
C
Cary Xu 已提交
950
        TD_SET_COL_ROWS_NORM(pBlockCol);
C
Cary Xu 已提交
951
      } else {
C
Cary Xu 已提交
952
        TD_SET_COL_ROWS_MISC(pBlockCol);
C
Cary Xu 已提交
953
      }
C
Cary Xu 已提交
954
      ++nColsOfBlockSma;
C
Cary Xu 已提交
955 956 957
    } else if (tdIsBitmapBlkNorm(pDataCol->pBitmap, rowsToWrite, pDataCols->bitmapMode)) {
      // check if all rows normal
      TD_SET_COL_ROWS_NORM(pBlockCol);
C
Cary Xu 已提交
958
    } else {
C
Cary Xu 已提交
959
      TD_SET_COL_ROWS_MISC(pBlockCol);
H
Hongze Cheng 已提交
960
    }
C
Cary Xu 已提交
961 962

    ++nColsNotAllNull;
H
Hongze Cheng 已提交
963 964 965 966 967 968 969
  }

  ASSERT(nColsNotAllNull >= 0 && nColsNotAllNull <= pDataCols->numOfCols);

  // Compress the data if neccessary
  int      tcol = 0;  // counter of not all NULL and written columns
  uint32_t toffset = 0;
970
  int32_t  tsize = (int32_t)tsdbBlockStatisSize(nColsNotAllNull, SBlockVerLatest);
H
Hongze Cheng 已提交
971
  int32_t  lsize = tsize;
C
Cary Xu 已提交
972
  uint32_t tsizeAggr = (uint32_t)tsdbBlockAggrSize(nColsOfBlockSma, SBlockVerLatest);
H
Hongze Cheng 已提交
973
  int32_t  keyLen = 0;
C
Cary Xu 已提交
974
  int32_t  nBitmaps = (int32_t)TD_BITMAP_BYTES(rowsToWrite);
C
Cary Xu 已提交
975
  int32_t  sBitmaps = isSuper ? (int32_t)TD_BITMAP_BYTES_I(rowsToWrite) : nBitmaps;
C
Cary Xu 已提交
976

977
  for (int ncol = 0; ncol < pDataCols->numOfCols; ++ncol) {
H
Hongze Cheng 已提交
978 979 980
    // All not NULL columns finish
    if (ncol != 0 && tcol >= nColsNotAllNull) break;

L
Liu Jicong 已提交
981
    SDataCol  *pDataCol = pDataCols->cols + ncol;
H
Hongze Cheng 已提交
982 983 984 985 986
    SBlockCol *pBlockCol = pBlockData->cols + tcol;

    if (ncol != 0 && (pDataCol->colId != pBlockCol->colId)) continue;

    int32_t flen;  // final length
C
Cary Xu 已提交
987
    int32_t tlen = dataColGetNEleLen(pDataCol, rowsToWrite, pDataCols->bitmapMode);
C
Cary Xu 已提交
988

C
Cary Xu 已提交
989
#ifdef TD_SUPPORT_BITMAP
C
Cary Xu 已提交
990
    int32_t tBitmaps = 0;
991
    int32_t tBitmapsLen = 0;
C
Cary Xu 已提交
992
    if ((ncol != 0) && !TD_COL_ROWS_NORM(pBlockCol)) {
C
Cary Xu 已提交
993
      tBitmaps = isSuper ? sBitmaps : nBitmaps;
C
Cary Xu 已提交
994
    }
C
Cary Xu 已提交
995
#endif
C
Cary Xu 已提交
996

C
bug fix  
Cary Xu 已提交
997
    void *tptr, *bptr;
H
Hongze Cheng 已提交
998 999

    // Make room
1000
    if (tsdbMakeRoom(ppBuf, lsize + tlen + tBitmaps + 2 * COMP_OVERFLOW_BYTES + sizeof(TSCKSUM)) < 0) {
H
Hongze Cheng 已提交
1001 1002 1003 1004 1005 1006
      return -1;
    }
    pBlockData = (SBlockData *)(*ppBuf);
    pBlockCol = pBlockData->cols + tcol;
    tptr = POINTER_SHIFT(pBlockData, lsize);

H
more  
Hongze Cheng 已提交
1007
    if (pCfg->compression == TWO_STAGE_COMP && tsdbMakeRoom(ppCBuf, tlen + COMP_OVERFLOW_BYTES) < 0) {
H
Hongze Cheng 已提交
1008 1009 1010 1011 1012
      return -1;
    }

    // Compress or just copy
    if (pCfg->compression) {
1013
#if 0
C
Cary Xu 已提交
1014
      flen = (*(tDataTypes[pDataCol->type].compFunc))((char *)pDataCol->pData, tlen, rowsToWrite + tBitmaps, tptr,
H
Hongze Cheng 已提交
1015 1016
                                                      tlen + COMP_OVERFLOW_BYTES, pCfg->compression, *ppCBuf,
                                                      tlen + COMP_OVERFLOW_BYTES);
1017 1018 1019 1020 1021
#endif
      flen = (*(tDataTypes[pDataCol->type].compFunc))((char *)pDataCol->pData, tlen, rowsToWrite, tptr,
                                                      tlen + COMP_OVERFLOW_BYTES, pCfg->compression, *ppCBuf,
                                                      tlen + COMP_OVERFLOW_BYTES);
      if (tBitmaps > 0) {
C
bug fix  
Cary Xu 已提交
1022
        bptr = POINTER_SHIFT(pBlockData, lsize + flen);
C
Cary Xu 已提交
1023
        if (isSuper && !tdDataColsIsBitmapI(pDataCols)) {
C
Cary Xu 已提交
1024
          tdMergeBitmap((uint8_t *)pDataCol->pBitmap, rowsToWrite, (uint8_t *)pDataCol->pBitmap);
C
Cary Xu 已提交
1025
        }
1026
        tBitmapsLen =
C
bug fix  
Cary Xu 已提交
1027
            tsCompressTinyint((char *)pDataCol->pBitmap, tBitmaps, tBitmaps, bptr, tBitmaps + COMP_OVERFLOW_BYTES,
1028 1029 1030 1031
                              pCfg->compression, *ppCBuf, tBitmaps + COMP_OVERFLOW_BYTES);
        TASSERT((tBitmapsLen > 0) && (tBitmapsLen <= (tBitmaps + COMP_OVERFLOW_BYTES)));
        flen += tBitmapsLen;
      }
H
Hongze Cheng 已提交
1032 1033 1034
    } else {
      flen = tlen;
      memcpy(tptr, pDataCol->pData, flen);
1035
      if (tBitmaps > 0) {
C
bug fix  
Cary Xu 已提交
1036 1037
        bptr = POINTER_SHIFT(pBlockData, lsize + flen);
        memcpy(bptr, pDataCol->pBitmap, tBitmaps);
1038 1039 1040
        tBitmapsLen = tBitmaps;
        flen += tBitmapsLen;
      }
H
Hongze Cheng 已提交
1041 1042 1043 1044
    }

    // Add checksum
    ASSERT(flen > 0);
1045
    ASSERT(tBitmapsLen <= 1024);
H
Hongze Cheng 已提交
1046 1047 1048 1049 1050 1051
    flen += sizeof(TSCKSUM);
    taosCalcChecksumAppend(0, (uint8_t *)tptr, flen);
    tsdbUpdateDFileMagic(pDFile, POINTER_SHIFT(tptr, flen - sizeof(TSCKSUM)));

    if (ncol != 0) {
      tsdbSetBlockColOffset(pBlockCol, toffset);
1052 1053
      pBlockCol->len = flen;  // data + bitmaps
      pBlockCol->blen = tBitmapsLen;
1054
      ++tcol;
H
Hongze Cheng 已提交
1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074
    } else {
      keyLen = flen;
    }

    toffset += flen;
    lsize += flen;
  }

  pBlockData->delimiter = TSDB_FILE_DELIMITER;
  pBlockData->uid = TABLE_UID(pTable);
  pBlockData->numOfCols = nColsNotAllNull;

  taosCalcChecksumAppend(0, (uint8_t *)pBlockData, tsize);
  tsdbUpdateDFileMagic(pDFile, POINTER_SHIFT(pBlockData, tsize - sizeof(TSCKSUM)));

  // Write the whole block to file
  if (tsdbAppendDFile(pDFile, (void *)pBlockData, lsize, &offset) < lsize) {
    return -1;
  }

C
Cary Xu 已提交
1075
  uint32_t aggrStatus = nColsOfBlockSma > 0 ? 1 : 0;
1076 1077 1078 1079 1080 1081 1082 1083 1084 1085
  if (aggrStatus > 0) {
    taosCalcChecksumAppend(0, (uint8_t *)pAggrBlkData, tsizeAggr);
    tsdbUpdateDFileMagic(pDFileAggr, POINTER_SHIFT(pAggrBlkData, tsizeAggr - sizeof(TSCKSUM)));

    // Write the whole block to file
    if (tsdbAppendDFile(pDFileAggr, (void *)pAggrBlkData, tsizeAggr, &offsetAggr) < tsizeAggr) {
      return -1;
    }
  }

1086
  // Update pBlock membership variables
H
Hongze Cheng 已提交
1087 1088 1089 1090 1091 1092 1093 1094
  pBlock->last = isLast;
  pBlock->offset = offset;
  pBlock->algorithm = pCfg->compression;
  pBlock->numOfRows = rowsToWrite;
  pBlock->len = lsize;
  pBlock->keyLen = keyLen;
  pBlock->numOfSubBlocks = isSuper ? 1 : 0;
  pBlock->numOfCols = nColsNotAllNull;
C
Cary Xu 已提交
1095
  pBlock->numOfBSma = nColsOfBlockSma;
H
Hongze Cheng 已提交
1096 1097
  pBlock->keyFirst = dataColsKeyFirst(pDataCols);
  pBlock->keyLast = dataColsKeyLast(pDataCols);
1098 1099 1100
  pBlock->aggrStat = aggrStatus;
  pBlock->blkVer = SBlockVerLatest;
  pBlock->aggrOffset = (uint64_t)offsetAggr;
H
Hongze Cheng 已提交
1101

L
Liu Jicong 已提交
1102
  tsdbDebug("vgId:%d uid:%" PRId64 " a block of data is written to file %s, offset %" PRId64
H
Hongze Cheng 已提交
1103
            " numOfRows %d len %d numOfCols %" PRId16 " keyFirst %" PRId64 " keyLast %" PRId64,
L
Liu Jicong 已提交
1104
            REPO_ID(pRepo), TABLE_UID(pTable), TSDB_FILE_FULL_NAME(pDFile), offset, rowsToWrite, pBlock->len,
H
Hongze Cheng 已提交
1105 1106 1107 1108 1109 1110 1111
            pBlock->numOfCols, pBlock->keyFirst, pBlock->keyLast);

  return 0;
}

static int tsdbWriteBlock(SCommitH *pCommith, SDFile *pDFile, SDataCols *pDataCols, SBlock *pBlock, bool isLast,
                          bool isSuper) {
1112 1113 1114 1115
  return tsdbWriteBlockImpl(TSDB_COMMIT_REPO(pCommith), TSDB_COMMIT_TABLE(pCommith), pDFile,
                            isLast ? TSDB_COMMIT_SMAL_FILE(pCommith) : TSDB_COMMIT_SMAD_FILE(pCommith), pDataCols,
                            pBlock, isLast, isSuper, (void **)(&(TSDB_COMMIT_BUF(pCommith))),
                            (void **)(&(TSDB_COMMIT_COMP_BUF(pCommith))), (void **)(&(TSDB_COMMIT_EXBUF(pCommith))));
H
Hongze Cheng 已提交
1116 1117
}

H
more  
Hongze Cheng 已提交
1118
static int tsdbWriteBlockInfo(SCommitH *pCommih) {
L
Liu Jicong 已提交
1119
  SDFile   *pHeadf = TSDB_COMMIT_HEAD_FILE(pCommih);
H
more  
Hongze Cheng 已提交
1120
  SBlockIdx blkIdx;
L
Liu Jicong 已提交
1121
  STable   *pTable = TSDB_COMMIT_TABLE(pCommih);
H
Hongze Cheng 已提交
1122

H
more  
Hongze Cheng 已提交
1123 1124 1125 1126
  if (tsdbWriteBlockInfoImpl(pHeadf, pTable, pCommih->aSupBlk, pCommih->aSubBlk, (void **)(&(TSDB_COMMIT_BUF(pCommih))),
                             &blkIdx) < 0) {
    return -1;
  }
H
Hongze Cheng 已提交
1127

H
more  
Hongze Cheng 已提交
1128 1129 1130
  if (blkIdx.numOfBlocks == 0) {
    return 0;
  }
H
Hongze Cheng 已提交
1131

H
more  
Hongze Cheng 已提交
1132 1133 1134 1135
  if (taosArrayPush(pCommih->aBlkIdx, (void *)(&blkIdx)) == NULL) {
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    return -1;
  }
H
Hongze Cheng 已提交
1136

H
more  
Hongze Cheng 已提交
1137 1138
  return 0;
}
H
Hongze Cheng 已提交
1139

H
more  
Hongze Cheng 已提交
1140
static int tsdbCommitMemData(SCommitH *pCommith, SCommitIter *pIter, TSKEY keyLimit, bool toData) {
L
Liu Jicong 已提交
1141 1142
  STsdb     *pRepo = TSDB_COMMIT_REPO(pCommith);
  STsdbCfg  *pCfg = REPO_CFG(pRepo);
H
more  
Hongze Cheng 已提交
1143 1144
  SMergeInfo mInfo;
  int32_t    defaultRows = TSDB_COMMIT_DEFAULT_ROWS(pCommith);
L
Liu Jicong 已提交
1145
  SDFile    *pDFile;
H
more  
Hongze Cheng 已提交
1146 1147
  bool       isLast;
  SBlock     block;
H
more  
Hongze Cheng 已提交
1148

H
more  
Hongze Cheng 已提交
1149 1150 1151
  while (true) {
    tsdbLoadDataFromCache(pIter->pTable, pIter->pIter, keyLimit, defaultRows, pCommith->pDataCols, NULL, 0,
                          pCfg->update, &mInfo);
H
more  
Hongze Cheng 已提交
1152

H
more  
Hongze Cheng 已提交
1153
    if (pCommith->pDataCols->numOfRows <= 0) break;
H
more  
Hongze Cheng 已提交
1154

H
refact  
Hongze Cheng 已提交
1155
    if (toData || pCommith->pDataCols->numOfRows >= pCfg->minRows) {
H
more  
Hongze Cheng 已提交
1156 1157 1158 1159 1160 1161
      pDFile = TSDB_COMMIT_DATA_FILE(pCommith);
      isLast = false;
    } else {
      pDFile = TSDB_COMMIT_LAST_FILE(pCommith);
      isLast = true;
    }
H
Hongze Cheng 已提交
1162

H
more  
Hongze Cheng 已提交
1163
    if (tsdbWriteBlock(pCommith, pDFile, pCommith->pDataCols, &block, isLast, true) < 0) return -1;
H
Hongze Cheng 已提交
1164

H
more  
Hongze Cheng 已提交
1165 1166 1167 1168
    if (tsdbCommitAddBlock(pCommith, &block, NULL, 0) < 0) {
      return -1;
    }
  }
H
Hongze Cheng 已提交
1169

H
more  
Hongze Cheng 已提交
1170 1171
  return 0;
}
H
Hongze Cheng 已提交
1172

H
more  
Hongze Cheng 已提交
1173
static int tsdbMergeMemData(SCommitH *pCommith, SCommitIter *pIter, int bidx) {
L
Liu Jicong 已提交
1174 1175
  STsdb     *pRepo = TSDB_COMMIT_REPO(pCommith);
  STsdbCfg  *pCfg = REPO_CFG(pRepo);
H
more  
Hongze Cheng 已提交
1176
  int        nBlocks = pCommith->readh.pBlkIdx->numOfBlocks;
L
Liu Jicong 已提交
1177
  SBlock    *pBlock = pCommith->readh.pBlkInfo->blocks + bidx;
H
more  
Hongze Cheng 已提交
1178
  TSKEY      keyLimit;
1179
  int16_t    colId = PRIMARYKEY_TIMESTAMP_COL_ID;
H
more  
Hongze Cheng 已提交
1180 1181 1182
  SMergeInfo mInfo;
  SBlock     subBlocks[TSDB_MAX_SUBBLOCKS];
  SBlock     block, supBlock;
L
Liu Jicong 已提交
1183
  SDFile    *pDFile;
H
more  
Hongze Cheng 已提交
1184 1185 1186 1187 1188 1189

  if (bidx == nBlocks - 1) {
    keyLimit = pCommith->maxKey;
  } else {
    keyLimit = pBlock[1].keyFirst - 1;
  }
H
Hongze Cheng 已提交
1190

H
more  
Hongze Cheng 已提交
1191
  SSkipListIterator titer = *(pIter->pIter);
C
Cary Xu 已提交
1192
  if (tsdbLoadBlockDataCols(&(pCommith->readh), pBlock, NULL, &colId, 1, false) < 0) return -1;
H
Hongze Cheng 已提交
1193

H
more  
Hongze Cheng 已提交
1194 1195
  tsdbLoadDataFromCache(pIter->pTable, &titer, keyLimit, INT32_MAX, NULL, pCommith->readh.pDCols[0]->cols[0].pData,
                        pCommith->readh.pDCols[0]->numOfRows, pCfg->update, &mInfo);
H
Hongze Cheng 已提交
1196

H
more  
Hongze Cheng 已提交
1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216
  if (mInfo.nOperations == 0) {
    // no new data to insert (all updates denied)
    if (tsdbMoveBlock(pCommith, bidx) < 0) {
      return -1;
    }
    *(pIter->pIter) = titer;
  } else if (pBlock->numOfRows + mInfo.rowsInserted - mInfo.rowsDeleteSucceed == 0) {
    // Ignore the block
    ASSERT(0);
    *(pIter->pIter) = titer;
  } else if (tsdbCanAddSubBlock(pCommith, pBlock, &mInfo)) {
    // Add a sub-block
    tsdbLoadDataFromCache(pIter->pTable, pIter->pIter, keyLimit, INT32_MAX, pCommith->pDataCols,
                          pCommith->readh.pDCols[0]->cols[0].pData, pCommith->readh.pDCols[0]->numOfRows, pCfg->update,
                          &mInfo);
    if (pBlock->last) {
      pDFile = TSDB_COMMIT_LAST_FILE(pCommith);
    } else {
      pDFile = TSDB_COMMIT_DATA_FILE(pCommith);
    }
H
Hongze Cheng 已提交
1217

H
more  
Hongze Cheng 已提交
1218
    if (tsdbWriteBlock(pCommith, pDFile, pCommith->pDataCols, &block, pBlock->last, false) < 0) return -1;
H
Hongze Cheng 已提交
1219

H
more  
Hongze Cheng 已提交
1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239
    if (pBlock->numOfSubBlocks == 1) {
      subBlocks[0] = *pBlock;
      subBlocks[0].numOfSubBlocks = 0;
    } else {
      memcpy(subBlocks, POINTER_SHIFT(pCommith->readh.pBlkInfo, pBlock->offset),
             sizeof(SBlock) * pBlock->numOfSubBlocks);
    }
    subBlocks[pBlock->numOfSubBlocks] = block;
    supBlock = *pBlock;
    supBlock.keyFirst = mInfo.keyFirst;
    supBlock.keyLast = mInfo.keyLast;
    supBlock.numOfSubBlocks++;
    supBlock.numOfRows = pBlock->numOfRows + mInfo.rowsInserted - mInfo.rowsDeleteSucceed;
    supBlock.offset = taosArrayGetSize(pCommith->aSubBlk) * sizeof(SBlock);

    if (tsdbCommitAddBlock(pCommith, &supBlock, subBlocks, supBlock.numOfSubBlocks) < 0) return -1;
  } else {
    if (tsdbLoadBlockData(&(pCommith->readh), pBlock, NULL) < 0) return -1;
    if (tsdbMergeBlockData(pCommith, pIter, pCommith->readh.pDCols[0], keyLimit, bidx == (nBlocks - 1)) < 0) return -1;
  }
H
Hongze Cheng 已提交
1240

H
more  
Hongze Cheng 已提交
1241 1242
  return 0;
}
H
Hongze Cheng 已提交
1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296

static int tsdbMoveBlock(SCommitH *pCommith, int bidx) {
  SBlock *pBlock = pCommith->readh.pBlkInfo->blocks + bidx;
  SDFile *pDFile;
  SBlock  block;
  bool    isSameFile;

  ASSERT(pBlock->numOfSubBlocks > 0);

  if (pBlock->last) {
    pDFile = TSDB_COMMIT_LAST_FILE(pCommith);
    isSameFile = pCommith->isLFileSame;
  } else {
    pDFile = TSDB_COMMIT_DATA_FILE(pCommith);
    isSameFile = pCommith->isDFileSame;
  }

  if (isSameFile) {
    if (pBlock->numOfSubBlocks == 1) {
      if (tsdbCommitAddBlock(pCommith, pBlock, NULL, 0) < 0) {
        return -1;
      }
    } else {
      block = *pBlock;
      block.offset = sizeof(SBlock) * taosArrayGetSize(pCommith->aSubBlk);

      if (tsdbCommitAddBlock(pCommith, &block, POINTER_SHIFT(pCommith->readh.pBlkInfo, pBlock->offset),
                             pBlock->numOfSubBlocks) < 0) {
        return -1;
      }
    }
  } else {
    if (tsdbLoadBlockData(&(pCommith->readh), pBlock, NULL) < 0) return -1;
    if (tsdbWriteBlock(pCommith, pDFile, pCommith->readh.pDCols[0], &block, pBlock->last, true) < 0) return -1;
    if (tsdbCommitAddBlock(pCommith, &block, NULL, 0) < 0) return -1;
  }

  return 0;
}

static int tsdbCommitAddBlock(SCommitH *pCommith, const SBlock *pSupBlock, const SBlock *pSubBlocks, int nSubBlocks) {
  if (taosArrayPush(pCommith->aSupBlk, pSupBlock) == NULL) {
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    return -1;
  }

  if (pSubBlocks && taosArrayAddBatch(pCommith->aSubBlk, pSubBlocks, nSubBlocks) == NULL) {
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    return -1;
  }

  return 0;
}

H
more  
Hongze Cheng 已提交
1297 1298
static int tsdbMergeBlockData(SCommitH *pCommith, SCommitIter *pIter, SDataCols *pDataCols, TSKEY keyLimit,
                              bool isLastOneBlock) {
L
Liu Jicong 已提交
1299
  STsdb    *pRepo = TSDB_COMMIT_REPO(pCommith);
H
more  
Hongze Cheng 已提交
1300 1301
  STsdbCfg *pCfg = REPO_CFG(pRepo);
  SBlock    block;
L
Liu Jicong 已提交
1302
  SDFile   *pDFile;
H
more  
Hongze Cheng 已提交
1303 1304
  bool      isLast;
  int32_t   defaultRows = TSDB_COMMIT_DEFAULT_ROWS(pCommith);
H
more  
Hongze Cheng 已提交
1305

H
more  
Hongze Cheng 已提交
1306 1307 1308 1309
  int biter = 0;
  while (true) {
    tsdbLoadAndMergeFromCache(pCommith->readh.pDCols[0], &biter, pIter, pCommith->pDataCols, keyLimit, defaultRows,
                              pCfg->update);
H
Hongze Cheng 已提交
1310

H
more  
Hongze Cheng 已提交
1311
    if (pCommith->pDataCols->numOfRows == 0) break;
H
Hongze Cheng 已提交
1312

H
more  
Hongze Cheng 已提交
1313
    if (isLastOneBlock) {
H
refact  
Hongze Cheng 已提交
1314
      if (pCommith->pDataCols->numOfRows < pCfg->minRows) {
H
more  
Hongze Cheng 已提交
1315 1316 1317 1318 1319 1320 1321 1322 1323 1324
        pDFile = TSDB_COMMIT_LAST_FILE(pCommith);
        isLast = true;
      } else {
        pDFile = TSDB_COMMIT_DATA_FILE(pCommith);
        isLast = false;
      }
    } else {
      pDFile = TSDB_COMMIT_DATA_FILE(pCommith);
      isLast = false;
    }
H
Hongze Cheng 已提交
1325

H
more  
Hongze Cheng 已提交
1326 1327 1328
    if (tsdbWriteBlock(pCommith, pDFile, pCommith->pDataCols, &block, isLast, true) < 0) return -1;
    if (tsdbCommitAddBlock(pCommith, &block, NULL, 0) < 0) return -1;
  }
H
more  
Hongze Cheng 已提交
1329

H
more  
Hongze Cheng 已提交
1330 1331
  return 0;
}
H
more  
Hongze Cheng 已提交
1332

H
more  
Hongze Cheng 已提交
1333 1334 1335 1336 1337
static void tsdbLoadAndMergeFromCache(SDataCols *pDataCols, int *iter, SCommitIter *pCommitIter, SDataCols *pTarget,
                                      TSKEY maxKey, int maxRows, int8_t update) {
  TSKEY     key1 = INT64_MAX;
  TSKEY     key2 = INT64_MAX;
  STSchema *pSchema = NULL;
H
Hongze Cheng 已提交
1338

H
more  
Hongze Cheng 已提交
1339 1340
  ASSERT(maxRows > 0 && dataColsKeyLast(pDataCols) <= maxKey);
  tdResetDataCols(pTarget);
H
more  
Hongze Cheng 已提交
1341

C
Cary Xu 已提交
1342 1343
  pTarget->bitmapMode = pDataCols->bitmapMode;

H
more  
Hongze Cheng 已提交
1344 1345
  while (true) {
    key1 = (*iter >= pDataCols->numOfRows) ? INT64_MAX : dataColsKeyAt(pDataCols, *iter);
C
Cary Xu 已提交
1346 1347
    STSRow *row = tsdbNextIterRow(pCommitIter->pIter);
    if (row == NULL || TD_ROW_KEY(row) > maxKey) {
H
more  
Hongze Cheng 已提交
1348 1349
      key2 = INT64_MAX;
    } else {
C
Cary Xu 已提交
1350
      key2 = TD_ROW_KEY(row);
H
more  
Hongze Cheng 已提交
1351
    }
H
more  
Hongze Cheng 已提交
1352

H
more  
Hongze Cheng 已提交
1353
    if (key1 == INT64_MAX && key2 == INT64_MAX) break;
H
more  
Hongze Cheng 已提交
1354

H
more  
Hongze Cheng 已提交
1355
    if (key1 < key2) {
C
Cary Xu 已提交
1356
      for (int i = 0; i < pDataCols->numOfCols; ++i) {
H
more  
Hongze Cheng 已提交
1357
        // TODO: dataColAppendVal may fail
C
Cary Xu 已提交
1358
        SCellVal sVal = {0};
C
Cary Xu 已提交
1359
        if (tdGetColDataOfRow(&sVal, pDataCols->cols + i, *iter, pDataCols->bitmapMode) < 0) {
C
Cary Xu 已提交
1360 1361
          TASSERT(0);
        }
H
refact  
Hongze Cheng 已提交
1362 1363
        tdAppendValToDataCol(pTarget->cols + i, sVal.valType, sVal.val, pTarget->numOfRows, pTarget->maxPoints,
                             pTarget->bitmapMode);
H
more  
Hongze Cheng 已提交
1364
      }
H
more  
Hongze Cheng 已提交
1365

C
Cary Xu 已提交
1366 1367
      ++pTarget->numOfRows;
      ++(*iter);
H
more  
Hongze Cheng 已提交
1368
    } else if (key1 > key2) {
C
Cary Xu 已提交
1369 1370
      if (pSchema == NULL || schemaVersion(pSchema) != TD_ROW_SVER(row)) {
        pSchema = tsdbGetTableSchemaImpl(pCommitIter->pTable, false, false, TD_ROW_SVER(row));
H
more  
Hongze Cheng 已提交
1371 1372
        ASSERT(pSchema != NULL);
      }
H
Hongze Cheng 已提交
1373

C
Cary Xu 已提交
1374
      tdAppendSTSRowToDataCol(row, pSchema, pTarget);
H
more  
Hongze Cheng 已提交
1375 1376 1377

      tSkipListIterNext(pCommitIter->pIter);
    } else {
C
Cary Xu 已提交
1378
#if 0
H
more  
Hongze Cheng 已提交
1379 1380
      if (update != TD_ROW_OVERWRITE_UPDATE) {
        // copy disk data
C
Cary Xu 已提交
1381
        for (int i = 0; i < pDataCols->numOfCols; ++i) {
H
more  
Hongze Cheng 已提交
1382
          // TODO: dataColAppendVal may fail
C
Cary Xu 已提交
1383
          SCellVal sVal = {0};
C
Cary Xu 已提交
1384
          if (tdGetColDataOfRow(&sVal, pDataCols->cols + i, *iter, pDataCols->bitmapMode) < 0) {
C
Cary Xu 已提交
1385 1386
            TASSERT(0);
          }
C
Cary Xu 已提交
1387
          tdAppendValToDataCol(pTarget->cols + i, sVal.valType, sVal.val, pTarget->numOfRows, pTarget->maxPoints, pTarget->bitmapMode);
H
more  
Hongze Cheng 已提交
1388 1389 1390 1391 1392 1393
        }

        if (update == TD_ROW_DISCARD_UPDATE) pTarget->numOfRows++;
      }
      if (update != TD_ROW_DISCARD_UPDATE) {
        // copy mem data
C
Cary Xu 已提交
1394 1395
        if (pSchema == NULL || schemaVersion(pSchema) != TD_ROW_SVER(row)) {
          pSchema = tsdbGetTableSchemaImpl(pCommitIter->pTable, false, false, TD_ROW_SVER(row));
H
more  
Hongze Cheng 已提交
1396 1397 1398
          ASSERT(pSchema != NULL);
        }

C
Cary Xu 已提交
1399
        tdAppendSTSRowToDataCol(row, pSchema, pTarget, update == TD_ROW_OVERWRITE_UPDATE);
H
more  
Hongze Cheng 已提交
1400
      }
C
Cary Xu 已提交
1401
      ++(*iter);
H
more  
Hongze Cheng 已提交
1402
      tSkipListIterNext(pCommitIter->pIter);
C
Cary Xu 已提交
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
#endif
      // copy disk data
      for (int i = 0; i < pDataCols->numOfCols; ++i) {
        SCellVal sVal = {0};
        if (tdGetColDataOfRow(&sVal, pDataCols->cols + i, *iter, pDataCols->bitmapMode) < 0) {
          TASSERT(0);
        }
        // TODO: tdAppendValToDataCol may fail
        tdAppendValToDataCol(pTarget->cols + i, sVal.valType, sVal.val, pTarget->numOfRows, pTarget->maxPoints,
                             pTarget->bitmapMode);
      }

      if (TD_SUPPORT_UPDATE(update)) {
        // copy mem data(Multi-Version)
        if (pSchema == NULL || schemaVersion(pSchema) != TD_ROW_SVER(row)) {
          pSchema = tsdbGetTableSchemaImpl(pCommitIter->pTable, false, false, TD_ROW_SVER(row));
          ASSERT(pSchema != NULL);
        }

        // TODO: merge with Multi-Version
        STSRow *curRow = row;

        ++(*iter);
        tSkipListIterNext(pCommitIter->pIter);
        STSRow *nextRow = tsdbNextIterRow(pCommitIter->pIter);

        if (key2 < TD_ROW_KEY(nextRow)) {
          tdAppendSTSRowToDataCol(row, pSchema, pTarget);
        } else {
          tdAppendSTSRowToDataCol(row, pSchema, pTarget);
        }
        // TODO: merge with Multi-Version
      } else {
        ++pTarget->numOfRows;
        ++(*iter);
        tSkipListIterNext(pCommitIter->pIter);
      }
H
more  
Hongze Cheng 已提交
1440 1441 1442 1443 1444
    }

    if (pTarget->numOfRows >= maxRows) break;
  }
}
H
Hongze Cheng 已提交
1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462

static void tsdbResetCommitTable(SCommitH *pCommith) {
  taosArrayClear(pCommith->aSubBlk);
  taosArrayClear(pCommith->aSupBlk);
  pCommith->pTable = NULL;
}

static void tsdbCloseCommitFile(SCommitH *pCommith, bool hasError) {
  if (pCommith->isRFileSet) {
    tsdbCloseAndUnsetFSet(&(pCommith->readh));
  }

  if (!hasError) {
    TSDB_FSET_FSYNC(TSDB_COMMIT_WRITE_FSET(pCommith));
  }
  tsdbCloseDFileSet(TSDB_COMMIT_WRITE_FSET(pCommith));
}

H
more  
Hongze Cheng 已提交
1463
static bool tsdbCanAddSubBlock(SCommitH *pCommith, SBlock *pBlock, SMergeInfo *pInfo) {
L
Liu Jicong 已提交
1464
  STsdb    *pRepo = TSDB_COMMIT_REPO(pCommith);
H
more  
Hongze Cheng 已提交
1465 1466
  STsdbCfg *pCfg = REPO_CFG(pRepo);
  int       mergeRows = pBlock->numOfRows + pInfo->rowsInserted - pInfo->rowsDeleteSucceed;
H
Hongze Cheng 已提交
1467

H
more  
Hongze Cheng 已提交
1468
  ASSERT(mergeRows > 0);
H
Hongze Cheng 已提交
1469

H
refact  
Hongze Cheng 已提交
1470
  if (pBlock->numOfSubBlocks < TSDB_MAX_SUBBLOCKS && pInfo->nOperations <= pCfg->maxRows) {
H
more  
Hongze Cheng 已提交
1471
    if (pBlock->last) {
H
refact  
Hongze Cheng 已提交
1472
      if (pCommith->isLFileSame && mergeRows < pCfg->minRows) return true;
H
more  
Hongze Cheng 已提交
1473
    } else {
H
refact  
Hongze Cheng 已提交
1474
      if (pCommith->isDFileSame && mergeRows <= pCfg->maxRows) return true;
H
more  
Hongze Cheng 已提交
1475 1476
    }
  }
H
Hongze Cheng 已提交
1477

H
more  
Hongze Cheng 已提交
1478 1479
  return false;
}
H
Hongze Cheng 已提交
1480

H
more  
Hongze Cheng 已提交
1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496
// int tsdbApplyRtn(STsdbRepo *pRepo) {
//   SRtn       rtn;
//   SFSIter    fsiter;
//   STsdbFS *  pfs = REPO_FS(pRepo);
//   SDFileSet *pSet;

//   // Get retention snapshot
//   tsdbGetRtnSnap(pRepo, &rtn);

//   tsdbFSIterInit(&fsiter, pfs, TSDB_FS_ITER_FORWARD);
//   while ((pSet = tsdbFSIterNext(&fsiter))) {
//     if (pSet->fid < rtn.minFid) {
//       tsdbInfo("vgId:%d FSET %d at level %d disk id %d expires, remove it", REPO_ID(pRepo), pSet->fid,
//                TSDB_FSET_LEVEL(pSet), TSDB_FSET_ID(pSet));
//       continue;
//     }
H
Hongze Cheng 已提交
1497

H
more  
Hongze Cheng 已提交
1498 1499 1500 1501
//     if (tsdbApplyRtnOnFSet(pRepo, pSet, &rtn) < 0) {
//       return -1;
//     }
//   }
H
Hongze Cheng 已提交
1502

H
more  
Hongze Cheng 已提交
1503
//   return 0;
L
Liu Jicong 已提交
1504
// }