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

H
Hongze Cheng 已提交
16
#include "tsdb.h"
H
Hongze Cheng 已提交
17

H
Hongze Cheng 已提交
18
#define SL_MAX_LEVEL 5
H
Hongze Cheng 已提交
19

H
Hongze Cheng 已提交
20 21 22 23
#define SL_NODE_SIZE(l)        (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l)*2)
#define SL_NODE_FORWARD(n, l)  ((n)->forwards[l])
#define SL_NODE_BACKWARD(n, l) ((n)->forwards[(n)->level + (l)])
#define SL_NODE_DATA(n)        (&SL_NODE_BACKWARD(n, (n)->level))
H
Hongze Cheng 已提交
24

H
Hongze Cheng 已提交
25 26 27 28 29 30 31 32 33 34 35
#define SL_MOVE_BACKWARD 0x1
#define SL_MOVE_FROM_POS 0x2

static void    tbDataMovePosTo(STbData *pTbData, SMemSkipListNode **pos, TSDBKEY *pKey, int32_t flags);
static int32_t tsdbGetOrCreateTbData(SMemTable *pMemTable, tb_uid_t suid, tb_uid_t uid, STbData **ppTbData);
static int32_t tsdbInsertTableDataImpl(SMemTable *pMemTable, STbData *pTbData, int64_t version,
                                       SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlock, SSubmitBlkRsp *pRsp);

int32_t tsdbMemTableCreate(STsdb *pTsdb, SMemTable **ppMemTable) {
  int32_t    code = 0;
  SMemTable *pMemTable = NULL;
H
Hongze Cheng 已提交
36

H
refact  
Hongze Cheng 已提交
37
  pMemTable = (SMemTable *)taosMemoryCalloc(1, sizeof(*pMemTable));
H
Hongze Cheng 已提交
38
  if (pMemTable == NULL) {
H
Hongze Cheng 已提交
39 40
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
H
Hongze Cheng 已提交
41
  }
H
Hongze Cheng 已提交
42
  taosInitRWLatch(&pMemTable->latch);
H
Hongze Cheng 已提交
43 44
  pMemTable->pTsdb = pTsdb;
  pMemTable->nRef = 1;
H
Hongze Cheng 已提交
45
  pMemTable->info = tKEYINFOInit();
H
Hongze Cheng 已提交
46
  pMemTable->nRow = 0;
H
Hongze Cheng 已提交
47
  pMemTable->nDel = 0;
H
Hongze Cheng 已提交
48 49 50
  pMemTable->aTbData = taosArrayInit(128, sizeof(STbData *));
  if (pMemTable->aTbData == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
wafwerar's avatar
wafwerar 已提交
51
    taosMemoryFree(pMemTable);
H
Hongze Cheng 已提交
52
    goto _err;
H
Hongze Cheng 已提交
53 54
  }

H
Hongze Cheng 已提交
55 56 57 58 59 60 61 62 63 64 65
  *ppMemTable = pMemTable;
  return code;

_err:
  *ppMemTable = NULL;
  return code;
}

void tsdbMemTableDestroy(SMemTable *pMemTable) {
  if (pMemTable) {
    taosArrayDestroy(pMemTable->aTbData);
wafwerar's avatar
wafwerar 已提交
66
    taosMemoryFree(pMemTable);
H
Hongze Cheng 已提交
67 68 69 70 71 72 73 74
  }
}

static int32_t tbDataPCmprFn(const void *p1, const void *p2) {
  STbData *pTbData1 = *(STbData **)p1;
  STbData *pTbData2 = *(STbData **)p2;

  if (pTbData1->suid < pTbData2->suid) {
H
Hongze Cheng 已提交
75
    return -1;
H
Hongze Cheng 已提交
76 77 78 79 80 81 82 83
  } else if (pTbData1->suid > pTbData2->suid) {
    return 1;
  }

  if (pTbData1->uid < pTbData2->uid) {
    return -1;
  } else if (pTbData1->uid > pTbData2->uid) {
    return 1;
H
Hongze Cheng 已提交
84
  }
H
Hongze Cheng 已提交
85

H
Hongze Cheng 已提交
86
  return 0;
H
Hongze Cheng 已提交
87
}
H
Hongze Cheng 已提交
88 89 90 91 92
void tsdbGetTbDataFromMemTable(SMemTable *pMemTable, tb_uid_t suid, tb_uid_t uid, STbData **ppTbData) {
  STbData *pTbData = &(STbData){.suid = suid, .uid = uid};
  void    *p = taosArraySearch(pMemTable->aTbData, &pTbData, tbDataPCmprFn, TD_EQ);
  *ppTbData = p ? *(STbData **)p : NULL;
}
H
Hongze Cheng 已提交
93

H
Hongze Cheng 已提交
94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154
int32_t tsdbInsertTableData(STsdb *pTsdb, int64_t version, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlock,
                            SSubmitBlkRsp *pRsp) {
  int32_t    code = 0;
  SMemTable *pMemTable = pTsdb->mem;
  STbData   *pTbData = NULL;
  tb_uid_t   suid = pMsgIter->suid;
  tb_uid_t   uid = pMsgIter->uid;
  int32_t    sverNew;

  // check if table exists (todo: refact)
  SMetaReader mr = {0};
  SMetaEntry  me = {0};
  metaReaderInit(&mr, pTsdb->pVnode->pMeta, 0);
  if (metaGetTableEntryByUid(&mr, pMsgIter->uid) < 0) {
    metaReaderClear(&mr);
    code = TSDB_CODE_PAR_TABLE_NOT_EXIST;
    goto _err;
  }
  if (pRsp->tblFName) strcat(pRsp->tblFName, mr.me.name);

  if (mr.me.type == TSDB_NORMAL_TABLE) {
    sverNew = mr.me.ntbEntry.schemaRow.version;
  } else {
    metaGetTableEntryByUid(&mr, mr.me.ctbEntry.suid);
    sverNew = mr.me.stbEntry.schemaRow.version;
  }
  metaReaderClear(&mr);
  pRsp->sver = sverNew;

  // create/get STbData to op
  code = tsdbGetOrCreateTbData(pMemTable, suid, uid, &pTbData);
  if (code) {
    goto _err;
  }

  // do insert impl
  code = tsdbInsertTableDataImpl(pMemTable, pTbData, version, pMsgIter, pBlock, pRsp);
  if (code) {
    goto _err;
  }

  return code;

_err:
  return code;
}

int32_t tsdbDeleteTableData(STsdb *pTsdb, int64_t version, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKEY eKey) {
  int32_t    code = 0;
  SMemTable *pMemTable = pTsdb->mem;
  STbData   *pTbData = NULL;
  SVBufPool *pPool = pTsdb->pVnode->inUse;

  // check if table exists (todo)

  code = tsdbGetOrCreateTbData(pMemTable, suid, uid, &pTbData);
  if (code) {
    goto _err;
  }

  // do delete
H
Hongze Cheng 已提交
155 156
  SDelData *pDelData = (SDelData *)vnodeBufPoolMalloc(pPool, sizeof(*pDelData));
  if (pDelData == NULL) {
H
Hongze Cheng 已提交
157 158 159
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
H
Hongze Cheng 已提交
160 161 162 163
  pDelData->version = version;
  pDelData->sKey = sKey;
  pDelData->eKey = eKey;
  pDelData->pNext = NULL;
H
Hongze Cheng 已提交
164 165
  if (pTbData->pHead == NULL) {
    ASSERT(pTbData->pTail == NULL);
H
Hongze Cheng 已提交
166
    pTbData->pHead = pTbData->pTail = pDelData;
H
Hongze Cheng 已提交
167
  } else {
H
Hongze Cheng 已提交
168 169
    pTbData->pTail->pNext = pDelData;
    pTbData->pTail = pDelData;
H
Hongze Cheng 已提交
170 171 172 173
  }

  // update the state of pMemTable and other (todo)

H
Hongze Cheng 已提交
174
  pMemTable->nDel++;
H
Hongze Cheng 已提交
175

C
Cary Xu 已提交
176
  tsdbError("vgId:%d, delete data from table suid:%" PRId64 " uid:%" PRId64 " skey:%" PRId64 " eKey:%" PRId64
H
Hongze Cheng 已提交
177 178 179 180 181
            " since %s",
            TD_VID(pTsdb->pVnode), suid, uid, sKey, eKey, tstrerror(code));
  return code;

_err:
C
Cary Xu 已提交
182
  tsdbError("vgId:%d, failed to delete data from table suid:%" PRId64 " uid:%" PRId64 " skey:%" PRId64 " eKey:%" PRId64
H
Hongze Cheng 已提交
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 209 210
            " since %s",
            TD_VID(pTsdb->pVnode), suid, uid, sKey, eKey, tstrerror(code));
  return code;
}

int32_t tsdbTbDataIterCreate(STbData *pTbData, TSDBKEY *pFrom, int8_t backward, STbDataIter **ppIter) {
  int32_t code = 0;

  (*ppIter) = (STbDataIter *)taosMemoryCalloc(1, sizeof(STbDataIter));
  if ((*ppIter) == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

  tsdbTbDataIterOpen(pTbData, pFrom, backward, *ppIter);

_exit:
  return code;
}

void *tsdbTbDataIterDestroy(STbDataIter *pIter) {
  if (pIter) {
    taosMemoryFree(pIter);
  }

  return NULL;
}

H
Hongze Cheng 已提交
211
void tsdbTbDataIterOpen(STbData *pTbData, TSDBKEY *pFrom, int8_t backward, STbDataIter *pIter) {
H
Hongze Cheng 已提交
212
  SMemSkipListNode *pos[SL_MAX_LEVEL];
H
Hongze Cheng 已提交
213 214
  SMemSkipListNode *pHead;
  SMemSkipListNode *pTail;
H
Hongze Cheng 已提交
215

H
Hongze Cheng 已提交
216 217
  pHead = pTbData->sl.pHead;
  pTail = pTbData->sl.pTail;
H
Hongze Cheng 已提交
218 219
  pIter->pTbData = pTbData;
  pIter->backward = backward;
H
Hongze Cheng 已提交
220
  pIter->pRow = NULL;
H
Hongze Cheng 已提交
221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243
  if (pFrom == NULL) {
    // create from head or tail
    if (backward) {
      pIter->pNode = SL_NODE_BACKWARD(pTbData->sl.pTail, 0);
    } else {
      pIter->pNode = SL_NODE_FORWARD(pTbData->sl.pHead, 0);
    }
  } else {
    // create from a key
    if (backward) {
      tbDataMovePosTo(pTbData, pos, pFrom, SL_MOVE_BACKWARD);
      pIter->pNode = SL_NODE_BACKWARD(pos[0], 0);
    } else {
      tbDataMovePosTo(pTbData, pos, pFrom, 0);
      pIter->pNode = SL_NODE_FORWARD(pos[0], 0);
    }
  }
}

bool tsdbTbDataIterNext(STbDataIter *pIter) {
  SMemSkipListNode *pHead = pIter->pTbData->sl.pHead;
  SMemSkipListNode *pTail = pIter->pTbData->sl.pTail;

H
Hongze Cheng 已提交
244
  pIter->pRow = NULL;
H
Hongze Cheng 已提交
245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271
  if (pIter->backward) {
    ASSERT(pIter->pNode != pTail);

    if (pIter->pNode == pHead) {
      return false;
    }

    pIter->pNode = SL_NODE_BACKWARD(pIter->pNode, 0);
    if (pIter->pNode == pHead) {
      return false;
    }
  } else {
    ASSERT(pIter->pNode != pHead);

    if (pIter->pNode == pTail) {
      return false;
    }

    pIter->pNode = SL_NODE_FORWARD(pIter->pNode, 0);
    if (pIter->pNode == pTail) {
      return false;
    }
  }

  return true;
}

H
Hongze Cheng 已提交
272
TSDBROW *tsdbTbDataIterGet(STbDataIter *pIter) {
H
Hongze Cheng 已提交
273 274 275
  // we add here for commit usage
  if (pIter == NULL) return NULL;

H
Hongze Cheng 已提交
276 277
  if (pIter->pRow) {
    goto _exit;
H
Hongze Cheng 已提交
278 279 280
  }

  if (pIter->backward) {
H
Hongze Cheng 已提交
281 282
    if (pIter->pNode == pIter->pTbData->sl.pHead) {
      goto _exit;
H
Hongze Cheng 已提交
283 284
    }
  } else {
H
Hongze Cheng 已提交
285 286
    if (pIter->pNode == pIter->pTbData->sl.pTail) {
      goto _exit;
H
Hongze Cheng 已提交
287
    }
H
Hongze Cheng 已提交
288
  }
H
Hongze Cheng 已提交
289

H
Hongze Cheng 已提交
290 291 292 293 294
  tGetTSDBRow((uint8_t *)SL_NODE_DATA(pIter->pNode), &pIter->row);
  pIter->pRow = &pIter->row;

_exit:
  return pIter->pRow;
H
Hongze Cheng 已提交
295
}
H
Hongze Cheng 已提交
296

H
Hongze Cheng 已提交
297 298 299 300 301 302 303 304 305 306 307
static int32_t tsdbGetOrCreateTbData(SMemTable *pMemTable, tb_uid_t suid, tb_uid_t uid, STbData **ppTbData) {
  int32_t  code = 0;
  int32_t  idx = 0;
  STbData *pTbData = NULL;
  STbData *pTbDataT = &(STbData){.suid = suid, .uid = uid};

  // get
  idx = taosArraySearchIdx(pMemTable->aTbData, &pTbDataT, tbDataPCmprFn, TD_GE);
  if (idx >= 0) {
    pTbData = (STbData *)taosArrayGetP(pMemTable->aTbData, idx);
    if (tbDataPCmprFn(&pTbDataT, &pTbData) == 0) goto _exit;
H
Hongze Cheng 已提交
308
  }
309

H
Hongze Cheng 已提交
310 311 312 313 314 315 316 317
  // create
  SVBufPool *pPool = pMemTable->pTsdb->pVnode->inUse;
  int8_t     maxLevel = pMemTable->pTsdb->pVnode->config.tsdbCfg.slLevel;

  pTbData = vnodeBufPoolMalloc(pPool, sizeof(*pTbData) + SL_NODE_SIZE(maxLevel) * 2);
  if (pTbData == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
H
Hongze Cheng 已提交
318
  }
H
Hongze Cheng 已提交
319 320
  pTbData->suid = suid;
  pTbData->uid = uid;
H
Hongze Cheng 已提交
321
  pTbData->info = tKEYINFOInit();
H
Hongze Cheng 已提交
322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339
  pTbData->pHead = NULL;
  pTbData->pTail = NULL;
  pTbData->sl.seed = taosRand();
  pTbData->sl.size = 0;
  pTbData->sl.maxLevel = maxLevel;
  pTbData->sl.level = 0;
  pTbData->sl.pHead = (SMemSkipListNode *)&pTbData[1];
  pTbData->sl.pTail = (SMemSkipListNode *)POINTER_SHIFT(pTbData->sl.pHead, SL_NODE_SIZE(maxLevel));
  pTbData->sl.pHead->level = maxLevel;
  pTbData->sl.pTail->level = maxLevel;
  for (int8_t iLevel = 0; iLevel < maxLevel; iLevel++) {
    SL_NODE_FORWARD(pTbData->sl.pHead, iLevel) = pTbData->sl.pTail;
    SL_NODE_BACKWARD(pTbData->sl.pTail, iLevel) = pTbData->sl.pHead;

    SL_NODE_BACKWARD(pTbData->sl.pHead, iLevel) = NULL;
    SL_NODE_FORWARD(pTbData->sl.pTail, iLevel) = NULL;
  }

H
Hongze Cheng 已提交
340 341 342 343 344 345 346
  void *p;
  if (idx < 0) {
    p = taosArrayPush(pMemTable->aTbData, &pTbData);
  } else {
    p = taosArrayInsert(pMemTable->aTbData, idx, &pTbData);
  }
  if (p == NULL) {
H
Hongze Cheng 已提交
347 348 349 350 351 352 353
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }

_exit:
  *ppTbData = pTbData;
  return code;
H
Hongze Cheng 已提交
354

H
Hongze Cheng 已提交
355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372
_err:
  *ppTbData = NULL;
  return code;
}

static void tbDataMovePosTo(STbData *pTbData, SMemSkipListNode **pos, TSDBKEY *pKey, int32_t flags) {
  SMemSkipListNode *px;
  SMemSkipListNode *pn;
  TSDBKEY          *pTKey;
  int               c;
  int               backward = flags & SL_MOVE_BACKWARD;
  int               fromPos = flags & SL_MOVE_FROM_POS;

  if (backward) {
    px = pTbData->sl.pTail;

    for (int8_t iLevel = pTbData->sl.maxLevel - 1; iLevel >= pTbData->sl.level; iLevel--) {
      pos[iLevel] = px;
H
Hongze Cheng 已提交
373 374
    }

H
Hongze Cheng 已提交
375 376
    if (pTbData->sl.level) {
      if (fromPos) px = pos[pTbData->sl.level - 1];
H
Hongze Cheng 已提交
377

H
Hongze Cheng 已提交
378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394
      for (int8_t iLevel = pTbData->sl.level - 1; iLevel >= 0; iLevel--) {
        pn = SL_NODE_BACKWARD(px, iLevel);
        while (pn != pTbData->sl.pHead) {
          pTKey = (TSDBKEY *)SL_NODE_DATA(pn);

          c = tsdbKeyCmprFn(pTKey, pKey);
          if (c <= 0) {
            break;
          } else {
            px = pn;
            pn = SL_NODE_BACKWARD(px, iLevel);
          }
        }

        pos[iLevel] = px;
      }
    }
H
Hongze Cheng 已提交
395
  } else {
H
Hongze Cheng 已提交
396 397 398 399 400 401 402 403 404 405 406
    px = pTbData->sl.pHead;

    for (int8_t iLevel = pTbData->sl.maxLevel - 1; iLevel >= pTbData->sl.level; iLevel--) {
      pos[iLevel] = px;
    }

    if (pTbData->sl.level) {
      if (fromPos) px = pos[pTbData->sl.level - 1];

      for (int8_t iLevel = pTbData->sl.level - 1; iLevel >= 0; iLevel--) {
        pn = SL_NODE_FORWARD(px, iLevel);
H
Hongze Cheng 已提交
407
        while (pn != pTbData->sl.pTail) {
H
Hongze Cheng 已提交
408 409 410 411 412 413 414 415 416 417 418 419 420 421
          pTKey = (TSDBKEY *)SL_NODE_DATA(pn);

          c = tsdbKeyCmprFn(pTKey, pKey);
          if (c >= 0) {
            break;
          } else {
            px = pn;
            pn = SL_NODE_FORWARD(px, iLevel);
          }
        }

        pos[iLevel] = px;
      }
    }
H
Hongze Cheng 已提交
422
  }
H
Hongze Cheng 已提交
423
}
H
Hongze Cheng 已提交
424

H
Hongze Cheng 已提交
425 426 427 428
static FORCE_INLINE int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl) {
  int8_t         level = 1;
  int8_t         tlevel = TMIN(pSl->maxLevel, pSl->level + 1);
  const uint32_t factor = 4;
H
Hongze Cheng 已提交
429

H
Hongze Cheng 已提交
430 431 432
  while ((taosRandR(&pSl->seed) % factor) == 0 && level < tlevel) {
    level++;
  }
H
Hongze Cheng 已提交
433

H
Hongze Cheng 已提交
434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454
  return level;
}
static int32_t tbDataDoPut(SMemTable *pMemTable, STbData *pTbData, SMemSkipListNode **pos, TSDBROW *pRow,
                           int8_t forward) {
  int32_t           code = 0;
  int8_t            level;
  SMemSkipListNode *pNode;
  SVBufPool        *pPool = pMemTable->pTsdb->pVnode->inUse;

  // node
  level = tsdbMemSkipListRandLevel(&pTbData->sl);
  pNode = (SMemSkipListNode *)vnodeBufPoolMalloc(pPool, SL_NODE_SIZE(level) + tPutTSDBRow(NULL, pRow));
  if (pNode == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }
  pNode->level = level;
  for (int8_t iLevel = 0; iLevel < level; iLevel++) {
    SL_NODE_FORWARD(pNode, iLevel) = NULL;
    SL_NODE_BACKWARD(pNode, iLevel) = NULL;
  }
H
Hongze Cheng 已提交
455

H
Hongze Cheng 已提交
456
  tPutTSDBRow((uint8_t *)SL_NODE_DATA(pNode), pRow);
C
Cary Xu 已提交
457

H
Hongze Cheng 已提交
458 459 460
  // put
  for (int8_t iLevel = 0; iLevel < pNode->level; iLevel++) {
    SMemSkipListNode *px = pos[iLevel];
H
Hongze Cheng 已提交
461

H
Hongze Cheng 已提交
462 463
    if (forward) {
      SMemSkipListNode *pNext = SL_NODE_FORWARD(px, iLevel);
H
Hongze Cheng 已提交
464

H
Hongze Cheng 已提交
465 466
      SL_NODE_FORWARD(pNode, iLevel) = pNext;
      SL_NODE_BACKWARD(pNode, iLevel) = px;
H
Hongze Cheng 已提交
467

H
Hongze Cheng 已提交
468 469 470 471
      SL_NODE_BACKWARD(pNext, iLevel) = pNode;
      SL_NODE_FORWARD(px, iLevel) = pNode;
    } else {
      SMemSkipListNode *pPrev = SL_NODE_BACKWARD(px, iLevel);
C
Cary Xu 已提交
472

H
Hongze Cheng 已提交
473 474
      SL_NODE_FORWARD(pNode, iLevel) = px;
      SL_NODE_BACKWARD(pNode, iLevel) = pPrev;
H
Hongze Cheng 已提交
475

H
Hongze Cheng 已提交
476 477 478
      SL_NODE_FORWARD(pPrev, iLevel) = pNode;
      SL_NODE_BACKWARD(px, iLevel) = pNode;
    }
H
Hongze Cheng 已提交
479 480
  }

H
Hongze Cheng 已提交
481 482 483
  pTbData->sl.size++;
  if (pTbData->sl.level < pNode->level) {
    pTbData->sl.level = pNode->level;
H
Hongze Cheng 已提交
484 485
  }

H
Hongze Cheng 已提交
486 487
_exit:
  return code;
H
Hongze Cheng 已提交
488 489
}

H
Hongze Cheng 已提交
490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508
static int32_t tsdbInsertTableDataImpl(SMemTable *pMemTable, STbData *pTbData, int64_t version,
                                       SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlock, SSubmitBlkRsp *pRsp) {
  int32_t           code = 0;
  SSubmitBlkIter    blkIter = {0};
  TSDBKEY           key = {.version = version};
  SMemSkipListNode *pos[SL_MAX_LEVEL];
  TSDBROW           row = {.version = version, .pTSRow = NULL};
  int32_t           nRow = 0;

  tInitSubmitBlkIter(pMsgIter, pBlock, &blkIter);

  // backward put first data
  row.pTSRow = tGetSubmitBlkNext(&blkIter);
  key.ts = row.pTSRow->ts;
  nRow++;
  tbDataMovePosTo(pTbData, pos, &key, SL_MOVE_BACKWARD);
  code = tbDataDoPut(pMemTable, pTbData, pos, &row, 0);
  if (code) {
    goto _err;
H
Hongze Cheng 已提交
509 510
  }

H
Hongze Cheng 已提交
511 512
  if (tsdbKeyCmprFn(&key, &pTbData->info.minKey) < 0) {
    pTbData->info.minKey = key;
H
Hongze Cheng 已提交
513
  }
H
Hongze Cheng 已提交
514

H
Hongze Cheng 已提交
515 516
  if (tsdbKeyCmprFn(&key, &pMemTable->info.minKey) < 0) {
    pMemTable->info.minKey = key;
H
Hongze Cheng 已提交
517
  }
H
Hongze Cheng 已提交
518

H
Hongze Cheng 已提交
519 520 521 522 523 524 525
  // forward put rest data
  row.pTSRow = tGetSubmitBlkNext(&blkIter);
  if (row.pTSRow) {
    for (int8_t iLevel = 0; iLevel < pTbData->sl.maxLevel; iLevel++) {
      pos[iLevel] = SL_NODE_BACKWARD(pos[iLevel], iLevel);
    }
    do {
H
more  
Hongze Cheng 已提交
526
      key.ts = row.pTSRow->ts;
H
Hongze Cheng 已提交
527 528 529 530 531 532 533 534 535 536 537
      nRow++;
      tbDataMovePosTo(pTbData, pos, &key, SL_MOVE_FROM_POS);
      code = tbDataDoPut(pMemTable, pTbData, pos, &row, 1);
      if (code) {
        goto _err;
      }

      row.pTSRow = tGetSubmitBlkNext(&blkIter);
    } while (row.pTSRow);
  }

H
Hongze Cheng 已提交
538 539
  if (tsdbKeyCmprFn(&key, &pTbData->info.maxKey) > 0) {
    pTbData->info.maxKey = key;
H
Hongze Cheng 已提交
540
  }
H
Hongze Cheng 已提交
541

H
Hongze Cheng 已提交
542 543
  if (tsdbKeyCmprFn(&key, &pMemTable->info.maxKey) > 0) {
    pMemTable->info.maxKey = key;
H
Hongze Cheng 已提交
544
  }
H
Hongze Cheng 已提交
545 546 547
  if (pTbData->info.minVerion > version) pTbData->info.minVerion = version;
  if (pTbData->info.maxVersion < version) pTbData->info.maxVersion = version;

H
Hongze Cheng 已提交
548 549 550 551 552 553 554 555
  pMemTable->nRef++;
  pRsp->numOfRows = nRow;
  pRsp->affectedRows = nRow;

  return code;

_err:
  return code;
H
Hongze Cheng 已提交
556
}