• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

taosdata / TDengine / #4720

08 Sep 2025 08:43AM UTC coverage: 58.139% (-0.6%) from 58.762%
#4720

push

travis-ci

web-flow
Merge pull request #32881 from taosdata/enh/add-new-windows-ci

fix(ci): update workflow reference to use new Windows CI YAML

133181 of 292179 branches covered (45.58%)

Branch coverage included in aggregate %.

201691 of 283811 relevant lines covered (71.07%)

5442780.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

67.17
/source/dnode/vnode/src/tsdb/tsdbUtil.c
1
/*
2
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3
 *
4
 * This program is free software: you can use, redistribute, and/or modify
5
 * it under the terms of the GNU Affero General Public License, version 3
6
 * or later ("AGPL"), as published by the Free Software Foundation.
7
 *
8
 * This program is distributed in the hope that it will be useful, but WITHOUT
9
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10
 * FITNESS FOR A PARTICULAR PURPOSE.
11
 *
12
 * You should have received a copy of the GNU Affero General Public License
13
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
14
 */
15

16
#include "tcompression.h"
17
#include "tdataformat.h"
18
#include "tsdb.h"
19
#include "tsdbDef.h"
20

21
int32_t tsdbGetColCmprAlgFromSet(SHashObj *set, int16_t colId, uint32_t *alg);
22

23
static int32_t tBlockDataCompressKeyPart(SBlockData *bData, SDiskDataHdr *hdr, SBuffer *buffer, SBuffer *assist,
24
                                         SColCompressInfo *pCompressExt);
25

26
// SMapData =======================================================================
27
void tMapDataReset(SMapData *pMapData) {
×
28
  pMapData->nItem = 0;
×
29
  pMapData->nData = 0;
×
30
}
×
31

32
void tMapDataClear(SMapData *pMapData) {
×
33
  tFree(pMapData->aOffset);
×
34
  tFree(pMapData->pData);
×
35
  pMapData->pData = NULL;
×
36
  pMapData->aOffset = NULL;
×
37
}
×
38

39
#ifdef BUILD_NO_CALL
40
int32_t tMapDataPutItem(SMapData *pMapData, void *pItem, int32_t (*tPutItemFn)(uint8_t *, void *)) {
41
  int32_t code = 0;
42
  int32_t offset = pMapData->nData;
43
  int32_t nItem = pMapData->nItem;
44

45
  pMapData->nItem++;
46
  pMapData->nData += tPutItemFn(NULL, pItem);
47

48
  // alloc
49
  code = tRealloc((uint8_t **)&pMapData->aOffset, sizeof(int32_t) * pMapData->nItem);
50
  if (code) goto _exit;
51
  code = tRealloc(&pMapData->pData, pMapData->nData);
52
  if (code) goto _exit;
53

54
  // put
55
  pMapData->aOffset[nItem] = offset;
56
  tPutItemFn(pMapData->pData + offset, pItem);
57

58
_exit:
59
  return code;
60
}
61

62
int32_t tMapDataCopy(SMapData *pFrom, SMapData *pTo) {
63
  int32_t code = 0;
64

65
  pTo->nItem = pFrom->nItem;
66
  pTo->nData = pFrom->nData;
67
  code = tRealloc((uint8_t **)&pTo->aOffset, sizeof(int32_t) * pFrom->nItem);
68
  if (code) goto _exit;
69
  code = tRealloc(&pTo->pData, pFrom->nData);
70
  if (code) goto _exit;
71
  memcpy(pTo->aOffset, pFrom->aOffset, sizeof(int32_t) * pFrom->nItem);
72
  memcpy(pTo->pData, pFrom->pData, pFrom->nData);
73

74
_exit:
75
  return code;
76
}
77

78
int32_t tMapDataSearch(SMapData *pMapData, void *pSearchItem, int32_t (*tGetItemFn)(uint8_t *, void *),
79
                       int32_t (*tItemCmprFn)(const void *, const void *), void *pItem) {
80
  int32_t code = 0;
81
  int32_t lidx = 0;
82
  int32_t ridx = pMapData->nItem - 1;
83
  int32_t midx;
84
  int32_t c;
85

86
  while (lidx <= ridx) {
87
    midx = (lidx + ridx) / 2;
88

89
    tMapDataGetItemByIdx(pMapData, midx, pItem, tGetItemFn);
90

91
    c = tItemCmprFn(pSearchItem, pItem);
92
    if (c == 0) {
93
      goto _exit;
94
    } else if (c < 0) {
95
      ridx = midx - 1;
96
    } else {
97
      lidx = midx + 1;
98
    }
99
  }
100

101
  code = TSDB_CODE_NOT_FOUND;
102

103
_exit:
104
  return code;
105
}
106
#endif
107

108
void tMapDataGetItemByIdx(SMapData *pMapData, int32_t idx, void *pItem, int32_t (*tGetItemFn)(uint8_t *, void *)) {
×
109
  int32_t r = tGetItemFn(pMapData->pData + pMapData->aOffset[idx], pItem);
×
110
}
×
111

112
#ifdef BUILD_NO_CALL
113
int32_t tMapDataToArray(SMapData *pMapData, int32_t itemSize, int32_t (*tGetItemFn)(uint8_t *, void *),
114
                        SArray **ppArray) {
115
  int32_t code = 0;
116

117
  SArray *pArray = taosArrayInit(pMapData->nItem, itemSize);
118
  if (pArray == NULL) {
119
    code = terrno;
120
    goto _exit;
121
  }
122

123
  for (int32_t i = 0; i < pMapData->nItem; i++) {
124
    tMapDataGetItemByIdx(pMapData, i, taosArrayReserve(pArray, 1), tGetItemFn);
125
  }
126

127
_exit:
128
  *ppArray = pArray;
129
  return code;
130
}
131

132
int32_t tPutMapData(uint8_t *p, SMapData *pMapData) {
133
  int32_t n = 0;
134

135
  n += tPutI32v(p ? p + n : p, pMapData->nItem);
136
  if (pMapData->nItem) {
137
    int32_t lOffset = 0;
138
    for (int32_t iItem = 0; iItem < pMapData->nItem; iItem++) {
139
      n += tPutI32v(p ? p + n : p, pMapData->aOffset[iItem] - lOffset);
140
      lOffset = pMapData->aOffset[iItem];
141
    }
142

143
    n += tPutI32v(p ? p + n : p, pMapData->nData);
144
    if (p) {
145
      memcpy(p + n, pMapData->pData, pMapData->nData);
146
    }
147
    n += pMapData->nData;
148
  }
149

150
  return n;
151
}
152
#endif
153

154
int32_t tGetMapData(uint8_t *p, SMapData *pMapData, int32_t *decodedSize) {
×
155
  int32_t n = 0;
×
156
  int32_t code;
157
  int32_t offset;
158

159
  tMapDataReset(pMapData);
×
160

161
  n += tGetI32v(p + n, &pMapData->nItem);
×
162
  if (pMapData->nItem) {
×
163
    code = tRealloc((uint8_t **)&pMapData->aOffset, sizeof(int32_t) * pMapData->nItem);
×
164
    if (code) {
×
165
      return code;
×
166
    }
167

168
    int32_t lOffset = 0;
×
169
    for (int32_t iItem = 0; iItem < pMapData->nItem; iItem++) {
×
170
      n += tGetI32v(p + n, &pMapData->aOffset[iItem]);
×
171
      pMapData->aOffset[iItem] += lOffset;
×
172
      lOffset = pMapData->aOffset[iItem];
×
173
    }
174

175
    n += tGetI32v(p + n, &pMapData->nData);
×
176
    code = tRealloc(&pMapData->pData, pMapData->nData);
×
177
    if (code) {
×
178
      return code;
×
179
    }
180
    memcpy(pMapData->pData, p + n, pMapData->nData);
×
181
    n += pMapData->nData;
×
182
  }
183

184
  if (decodedSize) {
×
185
    *decodedSize = n;
×
186
  }
187

188
  return 0;
×
189
}
190

191
#ifdef BUILD_NO_CALL
192
// TABLEID =======================================================================
193
int32_t tTABLEIDCmprFn(const void *p1, const void *p2) {
194
  TABLEID *pId1 = (TABLEID *)p1;
195
  TABLEID *pId2 = (TABLEID *)p2;
196

197
  if (pId1->suid < pId2->suid) {
198
    return -1;
199
  } else if (pId1->suid > pId2->suid) {
200
    return 1;
201
  }
202

203
  if (pId1->uid < pId2->uid) {
204
    return -1;
205
  } else if (pId1->uid > pId2->uid) {
206
    return 1;
207
  }
208

209
  return 0;
210
}
211

212
// SBlockIdx ======================================================
213
int32_t tPutBlockIdx(uint8_t *p, void *ph) {
214
  int32_t    n = 0;
215
  SBlockIdx *pBlockIdx = (SBlockIdx *)ph;
216

217
  n += tPutI64(p ? p + n : p, pBlockIdx->suid);
218
  n += tPutI64(p ? p + n : p, pBlockIdx->uid);
219
  n += tPutI64v(p ? p + n : p, pBlockIdx->offset);
220
  n += tPutI64v(p ? p + n : p, pBlockIdx->size);
221

222
  return n;
223
}
224
#endif
225

226
int32_t tGetBlockIdx(uint8_t *p, void *ph) {
×
227
  int32_t    n = 0;
×
228
  SBlockIdx *pBlockIdx = (SBlockIdx *)ph;
×
229

230
  n += tGetI64(p + n, &pBlockIdx->suid);
×
231
  n += tGetI64(p + n, &pBlockIdx->uid);
×
232
  n += tGetI64v(p + n, &pBlockIdx->offset);
×
233
  n += tGetI64v(p + n, &pBlockIdx->size);
×
234

235
  return n;
×
236
}
237

238
#ifdef BUILD_NO_CALL
239
int32_t tCmprBlockIdx(void const *lhs, void const *rhs) {
240
  SBlockIdx *lBlockIdx = (SBlockIdx *)lhs;
241
  SBlockIdx *rBlockIdx = (SBlockIdx *)rhs;
242

243
  if (lBlockIdx->suid < rBlockIdx->suid) {
244
    return -1;
245
  } else if (lBlockIdx->suid > rBlockIdx->suid) {
246
    return 1;
247
  }
248

249
  if (lBlockIdx->uid < rBlockIdx->uid) {
250
    return -1;
251
  } else if (lBlockIdx->uid > rBlockIdx->uid) {
252
    return 1;
253
  }
254

255
  return 0;
256
}
257

258
int32_t tCmprBlockL(void const *lhs, void const *rhs) {
259
  SBlockIdx *lBlockIdx = (SBlockIdx *)lhs;
260
  SSttBlk   *rBlockL = (SSttBlk *)rhs;
261

262
  if (lBlockIdx->suid < rBlockL->suid) {
263
    return -1;
264
  } else if (lBlockIdx->suid > rBlockL->suid) {
265
    return 1;
266
  }
267

268
  if (lBlockIdx->uid < rBlockL->minUid) {
269
    return -1;
270
  } else if (lBlockIdx->uid > rBlockL->maxUid) {
271
    return 1;
272
  }
273

274
  return 0;
275
}
276

277
// SDataBlk ======================================================
278
void tDataBlkReset(SDataBlk *pDataBlk) {
279
  *pDataBlk = (SDataBlk){.minKey = TSDBKEY_MAX, .maxKey = TSDBKEY_MIN, .minVer = VERSION_MAX, .maxVer = VERSION_MIN};
280
}
281

282
int32_t tPutDataBlk(uint8_t *p, void *ph) {
283
  int32_t   n = 0;
284
  SDataBlk *pDataBlk = (SDataBlk *)ph;
285

286
  n += tPutI64v(p ? p + n : p, pDataBlk->minKey.version);
287
  n += tPutI64v(p ? p + n : p, pDataBlk->minKey.ts);
288
  n += tPutI64v(p ? p + n : p, pDataBlk->maxKey.version);
289
  n += tPutI64v(p ? p + n : p, pDataBlk->maxKey.ts);
290
  n += tPutI64v(p ? p + n : p, pDataBlk->minVer);
291
  n += tPutI64v(p ? p + n : p, pDataBlk->maxVer);
292
  n += tPutI32v(p ? p + n : p, pDataBlk->nRow);
293
  n += tPutI8(p ? p + n : p, pDataBlk->hasDup);
294
  n += tPutI8(p ? p + n : p, pDataBlk->nSubBlock);
295
  for (int8_t iSubBlock = 0; iSubBlock < pDataBlk->nSubBlock; iSubBlock++) {
296
    n += tPutI64v(p ? p + n : p, pDataBlk->aSubBlock[iSubBlock].offset);
297
    n += tPutI32v(p ? p + n : p, pDataBlk->aSubBlock[iSubBlock].szBlock);
298
    n += tPutI32v(p ? p + n : p, pDataBlk->aSubBlock[iSubBlock].szKey);
299
  }
300
  if (pDataBlk->nSubBlock == 1 && !pDataBlk->hasDup) {
301
    n += tPutI64v(p ? p + n : p, pDataBlk->smaInfo.offset);
302
    n += tPutI32v(p ? p + n : p, pDataBlk->smaInfo.size);
303
  }
304

305
  return n;
306
}
307
#endif
308

309
int32_t tGetDataBlk(uint8_t *p, void *ph) {
×
310
  int32_t   n = 0;
×
311
  SDataBlk *pDataBlk = (SDataBlk *)ph;
×
312

313
  n += tGetI64v(p + n, &pDataBlk->minKey.version);
×
314
  n += tGetI64v(p + n, &pDataBlk->minKey.ts);
×
315
  n += tGetI64v(p + n, &pDataBlk->maxKey.version);
×
316
  n += tGetI64v(p + n, &pDataBlk->maxKey.ts);
×
317
  n += tGetI64v(p + n, &pDataBlk->minVer);
×
318
  n += tGetI64v(p + n, &pDataBlk->maxVer);
×
319
  n += tGetI32v(p + n, &pDataBlk->nRow);
×
320
  n += tGetI8(p + n, &pDataBlk->hasDup);
×
321
  n += tGetI8(p + n, &pDataBlk->nSubBlock);
×
322
  for (int8_t iSubBlock = 0; iSubBlock < pDataBlk->nSubBlock; iSubBlock++) {
×
323
    n += tGetI64v(p + n, &pDataBlk->aSubBlock[iSubBlock].offset);
×
324
    n += tGetI32v(p + n, &pDataBlk->aSubBlock[iSubBlock].szBlock);
×
325
    n += tGetI32v(p + n, &pDataBlk->aSubBlock[iSubBlock].szKey);
×
326
  }
327
  if (pDataBlk->nSubBlock == 1 && !pDataBlk->hasDup) {
×
328
    n += tGetI64v(p + n, &pDataBlk->smaInfo.offset);
×
329
    n += tGetI32v(p + n, &pDataBlk->smaInfo.size);
×
330
  } else {
331
    pDataBlk->smaInfo.offset = 0;
×
332
    pDataBlk->smaInfo.size = 0;
×
333
  }
334

335
  return n;
×
336
}
337

338
#ifdef BUILD_NO_CALL
339
int32_t tDataBlkCmprFn(const void *p1, const void *p2) {
340
  SDataBlk *pBlock1 = (SDataBlk *)p1;
341
  SDataBlk *pBlock2 = (SDataBlk *)p2;
342

343
  if (tsdbKeyCmprFn(&pBlock1->maxKey, &pBlock2->minKey) < 0) {
344
    return -1;
345
  } else if (tsdbKeyCmprFn(&pBlock1->minKey, &pBlock2->maxKey) > 0) {
346
    return 1;
347
  }
348

349
  return 0;
350
}
351

352
bool tDataBlkHasSma(SDataBlk *pDataBlk) {
353
  if (pDataBlk->nSubBlock > 1) return false;
354
  if (pDataBlk->hasDup) return false;
355

356
  return pDataBlk->smaInfo.size > 0;
357
}
358

359
// SSttBlk ======================================================
360
int32_t tPutSttBlk(uint8_t *p, void *ph) {
361
  int32_t  n = 0;
362
  SSttBlk *pSttBlk = (SSttBlk *)ph;
363

364
  n += tPutI64(p ? p + n : p, pSttBlk->suid);
365
  n += tPutI64(p ? p + n : p, pSttBlk->minUid);
366
  n += tPutI64(p ? p + n : p, pSttBlk->maxUid);
367
  n += tPutI64v(p ? p + n : p, pSttBlk->minKey);
368
  n += tPutI64v(p ? p + n : p, pSttBlk->maxKey);
369
  n += tPutI64v(p ? p + n : p, pSttBlk->minVer);
370
  n += tPutI64v(p ? p + n : p, pSttBlk->maxVer);
371
  n += tPutI32v(p ? p + n : p, pSttBlk->nRow);
372
  n += tPutI64v(p ? p + n : p, pSttBlk->bInfo.offset);
373
  n += tPutI32v(p ? p + n : p, pSttBlk->bInfo.szBlock);
374
  n += tPutI32v(p ? p + n : p, pSttBlk->bInfo.szKey);
375

376
  return n;
377
}
378
#endif
379

380
int32_t tGetSttBlk(uint8_t *p, void *ph) {
×
381
  int32_t  n = 0;
×
382
  SSttBlk *pSttBlk = (SSttBlk *)ph;
×
383

384
  n += tGetI64(p + n, &pSttBlk->suid);
×
385
  n += tGetI64(p + n, &pSttBlk->minUid);
×
386
  n += tGetI64(p + n, &pSttBlk->maxUid);
×
387
  n += tGetI64v(p + n, &pSttBlk->minKey);
×
388
  n += tGetI64v(p + n, &pSttBlk->maxKey);
×
389
  n += tGetI64v(p + n, &pSttBlk->minVer);
×
390
  n += tGetI64v(p + n, &pSttBlk->maxVer);
×
391
  n += tGetI32v(p + n, &pSttBlk->nRow);
×
392
  n += tGetI64v(p + n, &pSttBlk->bInfo.offset);
×
393
  n += tGetI32v(p + n, &pSttBlk->bInfo.szBlock);
×
394
  n += tGetI32v(p + n, &pSttBlk->bInfo.szKey);
×
395

396
  return n;
×
397
}
398

399
// SBlockCol ======================================================
400

401
static const int32_t BLOCK_WITH_ALG_VER = 2;
402

403
int32_t tPutBlockCol(SBuffer *buffer, const SBlockCol *pBlockCol, int32_t ver, uint32_t defaultCmprAlg) {
773,372✔
404
  int32_t code;
405

406
  if ((code = tBufferPutI16v(buffer, pBlockCol->cid))) return code;
1,546,745!
407
  if ((code = tBufferPutI8(buffer, pBlockCol->type))) return code;
1,546,746!
408
  if ((code = tBufferPutI8(buffer, pBlockCol->cflag))) return code;
1,546,746!
409
  if ((code = tBufferPutI8(buffer, pBlockCol->flag))) return code;
1,546,746!
410
  if ((code = tBufferPutI32v(buffer, pBlockCol->szOrigin))) return code;
1,546,746!
411

412
  if (pBlockCol->flag != HAS_NULL) {
773,373✔
413
    if (pBlockCol->flag != HAS_VALUE) {
758,164✔
414
      if ((code = tBufferPutI32v(buffer, pBlockCol->szBitmap))) return code;
386,194!
415
    }
416

417
    if (IS_VAR_DATA_TYPE(pBlockCol->type)) {
758,164!
418
      if ((code = tBufferPutI32v(buffer, pBlockCol->szOffset))) return code;
201,926!
419
    }
420

421
    if (pBlockCol->flag != (HAS_NULL | HAS_NONE)) {
758,164✔
422
      if ((code = tBufferPutI32v(buffer, pBlockCol->szValue))) return code;
1,499,650!
423
    }
424

425
    if ((code = tBufferPutI32v(buffer, pBlockCol->offset))) return code;
1,516,328!
426
  }
427
  if (ver >= BLOCK_WITH_ALG_VER) {
773,373!
428
    if ((code = tBufferPutU32(buffer, pBlockCol->alg))) return code;
1,547,166!
429
  } else {
430
    if ((code = tBufferPutU32(buffer, defaultCmprAlg))) return code;
×
431
  }
432
  return 0;
773,373✔
433
}
434

435
int32_t tGetBlockCol(SBufferReader *br, SBlockCol *pBlockCol, int32_t ver, uint32_t defaultCmprAlg) {
10,263,275✔
436
  int32_t code;
437

438
  if ((code = tBufferGetI16v(br, &pBlockCol->cid))) return code;
10,263,275!
439
  if ((code = tBufferGetI8(br, &pBlockCol->type))) return code;
10,263,551!
440
  if ((code = tBufferGetI8(br, &pBlockCol->cflag))) return code;
10,262,923!
441
  if ((code = tBufferGetI8(br, &pBlockCol->flag))) return code;
10,261,369!
442
  if ((code = tBufferGetI32v(br, &pBlockCol->szOrigin))) return code;
10,260,801!
443

444
  pBlockCol->szBitmap = 0;
10,259,061✔
445
  pBlockCol->szOffset = 0;
10,259,061✔
446
  pBlockCol->szValue = 0;
10,259,061✔
447
  pBlockCol->offset = 0;
10,259,061✔
448

449
  if (pBlockCol->flag != HAS_NULL) {
10,259,061✔
450
    if (pBlockCol->flag != HAS_VALUE) {
9,151,277✔
451
      if ((code = tBufferGetI32v(br, &pBlockCol->szBitmap))) return code;
1,696,669!
452
    }
453

454
    if (IS_VAR_DATA_TYPE(pBlockCol->type)) {
9,151,134!
455
      if ((code = tBufferGetI32v(br, &pBlockCol->szOffset))) return code;
1,437,080!
456
    }
457

458
    if (pBlockCol->flag != (HAS_NULL | HAS_NONE)) {
9,155,093✔
459
      if ((code = tBufferGetI32v(br, &pBlockCol->szValue))) return code;
9,151,522!
460
    }
461

462
    if ((code = tBufferGetI32v(br, &pBlockCol->offset))) return code;
9,155,364!
463
  }
464

465
  if (ver >= BLOCK_WITH_ALG_VER) {
10,259,561!
466
    if ((code = tBufferGetU32(br, &pBlockCol->alg))) return code;
10,261,127!
467
  } else {
468
    pBlockCol->alg = defaultCmprAlg;
×
469
  }
470

471
  return 0;
10,259,342✔
472
}
473

474
#ifdef BUILD_NO_CALL
475
int32_t tBlockColCmprFn(const void *p1, const void *p2) {
476
  if (((SBlockCol *)p1)->cid < ((SBlockCol *)p2)->cid) {
477
    return -1;
478
  } else if (((SBlockCol *)p1)->cid > ((SBlockCol *)p2)->cid) {
479
    return 1;
480
  }
481

482
  return 0;
483
}
484

485
// SDelIdx ======================================================
486
int32_t tCmprDelIdx(void const *lhs, void const *rhs) {
487
  SDelIdx *lDelIdx = (SDelIdx *)lhs;
488
  SDelIdx *rDelIdx = (SDelIdx *)rhs;
489

490
  if (lDelIdx->suid < rDelIdx->suid) {
491
    return -1;
492
  } else if (lDelIdx->suid > rDelIdx->suid) {
493
    return 1;
494
  }
495

496
  if (lDelIdx->uid < rDelIdx->uid) {
497
    return -1;
498
  } else if (lDelIdx->uid > rDelIdx->uid) {
499
    return 1;
500
  }
501

502
  return 0;
503
}
504

505
int32_t tPutDelIdx(uint8_t *p, void *ph) {
506
  SDelIdx *pDelIdx = (SDelIdx *)ph;
507
  int32_t  n = 0;
508

509
  n += tPutI64(p ? p + n : p, pDelIdx->suid);
510
  n += tPutI64(p ? p + n : p, pDelIdx->uid);
511
  n += tPutI64v(p ? p + n : p, pDelIdx->offset);
512
  n += tPutI64v(p ? p + n : p, pDelIdx->size);
513

514
  return n;
515
}
516
#endif
517

518
int32_t tGetDelIdx(uint8_t *p, void *ph) {
×
519
  SDelIdx *pDelIdx = (SDelIdx *)ph;
×
520
  int32_t  n = 0;
×
521

522
  n += tGetI64(p + n, &pDelIdx->suid);
×
523
  n += tGetI64(p + n, &pDelIdx->uid);
×
524
  n += tGetI64v(p + n, &pDelIdx->offset);
×
525
  n += tGetI64v(p + n, &pDelIdx->size);
×
526

527
  return n;
×
528
}
529

530
#ifdef BUILD_NO_CALL
531
// SDelData ======================================================
532
int32_t tPutDelData(uint8_t *p, void *ph) {
533
  SDelData *pDelData = (SDelData *)ph;
534
  int32_t   n = 0;
535

536
  n += tPutI64v(p ? p + n : p, pDelData->version);
537
  n += tPutI64(p ? p + n : p, pDelData->sKey);
538
  n += tPutI64(p ? p + n : p, pDelData->eKey);
539

540
  return n;
541
}
542
#endif
543

544
int32_t tGetDelData(uint8_t *p, void *ph) {
×
545
  SDelData *pDelData = (SDelData *)ph;
×
546
  int32_t   n = 0;
×
547

548
  n += tGetI64v(p + n, &pDelData->version);
×
549
  n += tGetI64(p + n, &pDelData->sKey);
×
550
  n += tGetI64(p + n, &pDelData->eKey);
×
551

552
  return n;
×
553
}
554

555
int32_t tsdbKeyFid(TSKEY key, int32_t minutes, int8_t precision) {
208,310✔
556
  int64_t fid;
557
  if (key < 0) {
208,310✔
558
    fid = ((key + 1) / tsTickPerMin[precision] / minutes - 1);
6,069✔
559
    return (fid < INT32_MIN) ? INT32_MIN : (int32_t)fid;
6,069✔
560
  } else {
561
    fid = ((key / tsTickPerMin[precision] / minutes));
202,241✔
562
    return (fid > INT32_MAX) ? INT32_MAX : (int32_t)fid;
202,241✔
563
  }
564
}
565

566
void tsdbFidKeyRange(int32_t fid, int32_t minutes, int8_t precision, TSKEY *minKey, TSKEY *maxKey) {
2,546,656✔
567
  *minKey = tsTickPerMin[precision] * fid * minutes;
2,546,656✔
568
  *maxKey = *minKey + tsTickPerMin[precision] * minutes - 1;
2,546,656✔
569
}
2,546,656✔
570

571
int32_t tsdbFidLevel(int32_t fid, STsdbKeepCfg *pKeepCfg, int64_t nowSec) {
25,006✔
572
  int32_t aFid[3];
573
  TSKEY   key;
574

575
  if (pKeepCfg->precision == TSDB_TIME_PRECISION_MILLI) {
25,006✔
576
    nowSec = nowSec * 1000;
24,826✔
577
  } else if (pKeepCfg->precision == TSDB_TIME_PRECISION_MICRO) {
180✔
578
    nowSec = nowSec * 1000000l;
80✔
579
  } else if (pKeepCfg->precision == TSDB_TIME_PRECISION_NANO) {
100!
580
    nowSec = nowSec * 1000000000l;
100✔
581
  } else {
582
    tsdbError("invalid time precision:%d", pKeepCfg->precision);
×
583
    return 0;
×
584
  }
585

586
  nowSec = nowSec - pKeepCfg->keepTimeOffset * tsTickPerHour[pKeepCfg->precision];
25,006✔
587

588
  key = nowSec - pKeepCfg->keep0 * tsTickPerMin[pKeepCfg->precision];
25,006✔
589
  aFid[0] = tsdbKeyFid(key, pKeepCfg->days, pKeepCfg->precision);
25,006✔
590
  key = nowSec - pKeepCfg->keep1 * tsTickPerMin[pKeepCfg->precision];
25,006✔
591
  aFid[1] = tsdbKeyFid(key, pKeepCfg->days, pKeepCfg->precision);
25,006✔
592
  key = nowSec - pKeepCfg->keep2 * tsTickPerMin[pKeepCfg->precision];
25,006✔
593
  aFid[2] = tsdbKeyFid(key, pKeepCfg->days, pKeepCfg->precision);
25,006✔
594

595
  if (fid >= aFid[0]) {
25,006✔
596
    return 0;
24,415✔
597
  } else if (fid >= aFid[1]) {
591✔
598
    return 1;
272✔
599
  } else if (fid >= aFid[2]) {
319✔
600
    return 2;
316✔
601
  } else {
602
    return -1;
3✔
603
  }
604
}
605

606
// TSDBROW ======================================================
607
void tsdbRowGetColVal(TSDBROW *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal) {
549,077,815✔
608
  int32_t   code = 0;
549,077,815✔
609
  STColumn *pTColumn = &pTSchema->columns[iCol];
549,077,815✔
610
  SValue    value;
611

612
  if (pRow->type == TSDBROW_ROW_FMT) {
549,077,815✔
613
    int32_t ret = tRowGet(pRow->pTSRow, pTSchema, iCol, pColVal);
90,552,861✔
614
    if (ret != 0) {
90,484,661!
615
      tsdbError("failed to get column value, code:%d", ret);
×
616
    }
617
  } else if (pRow->type == TSDBROW_COL_FMT) {
458,524,954!
618
    if (iCol == 0) {
461,204,368!
619
      SValue val = {.type = TSDB_DATA_TYPE_TIMESTAMP};
×
620
      VALUE_SET_TRIVIAL_DATUM(&val, pRow->pBlockData->aTSKEY[pRow->iRow]);
×
621
      *pColVal = COL_VAL_VALUE(PRIMARYKEY_TIMESTAMP_COL_ID, val);
×
622
    } else {
623
      SColData *pColData = tBlockDataGetColData(pRow->pBlockData, pTColumn->colId);
461,204,368✔
624
      if (pColData) {
459,953,195✔
625
        if (tColDataGetValue(pColData, pRow->iRow, pColVal) != 0) {
115,758,393!
626
          tsdbError("failed to tColDataGetValue");
×
627
        }
628
      } else {
629
        *pColVal = COL_VAL_NONE(pTColumn->colId, pTColumn->type);
344,194,802✔
630
      }
631
    }
632
  }
633
}
548,478,864✔
634

635
void tsdbRowGetKey(TSDBROW *row, STsdbRowKey *key) {
683,515,858✔
636
  if (row->type == TSDBROW_ROW_FMT) {
683,515,858✔
637
    key->version = row->version;
361,250,747✔
638
    tRowGetKey(row->pTSRow, &key->key);
722,501,494✔
639
  } else {
640
    key->version = row->pBlockData->aVersion[row->iRow];
322,265,111✔
641
    tColRowGetKey(row->pBlockData, row->iRow, &key->key);
322,265,111!
642
  }
643
}
684,348,278✔
644

645
void tColRowGetPrimaryKey(SBlockData *pBlock, int32_t irow, SRowKey *key) {
471,276,349✔
646
  for (int32_t i = 0; i < pBlock->nColData; i++) {
471,457,926✔
647
    SColData *pColData = &pBlock->aColData[i];
397,605,349✔
648
    if (pColData->cflag & COL_IS_KEY) {
397,605,349✔
649
      SColVal cv;
650
      if (tColDataGetValue(pColData, irow, &cv) != 0) {
181,392!
651
        break;
×
652
      }
653
      key->pks[key->numOfPKs] = cv.value;
181,577✔
654
      key->numOfPKs++;
181,577✔
655
    } else {
656
      break;
397,423,957✔
657
    }
658
  }
659
}
471,276,534✔
660

661
int32_t tsdbRowKeyCmpr(const STsdbRowKey *key1, const STsdbRowKey *key2) {
41,954,605✔
662
  int32_t c = tRowKeyCompare(&key1->key, &key2->key);
41,954,605✔
663

664
  if (c) {
41,960,209✔
665
    return c;
36,707,456✔
666
  }
667

668
  if (key1->version < key2->version) {
5,252,753✔
669
    return -1;
4,588,704✔
670
  } else if (key1->version > key2->version) {
664,049!
671
    return 1;
723,801✔
672
  }
673
  return 0;
×
674
}
675

676
int32_t tsdbRowCompare(const void *p1, const void *p2) {
9,051,327✔
677
  STsdbRowKey key1, key2;
678

679
  tsdbRowGetKey((TSDBROW *)p1, &key1);
9,051,327✔
680
  tsdbRowGetKey((TSDBROW *)p2, &key2);
9,038,676✔
681
  return tsdbRowKeyCmpr(&key1, &key2);
8,992,612✔
682
}
683

684
int32_t tsdbRowCompareWithoutVersion(const void *p1, const void *p2) {
201,492,460✔
685
  STsdbRowKey key1, key2;
686

687
  tsdbRowGetKey((TSDBROW *)p1, &key1);
201,492,460✔
688
  tsdbRowGetKey((TSDBROW *)p2, &key2);
201,148,153✔
689
  return tRowKeyCompare(&key1.key, &key2.key);
201,753,549✔
690
}
691

692
// STSDBRowIter ======================================================
693
int32_t tsdbRowIterOpen(STSDBRowIter *pIter, TSDBROW *pRow, STSchema *pTSchema) {
39,155✔
694
  pIter->pRow = pRow;
39,155✔
695
  if (pRow->type == TSDBROW_ROW_FMT) {
39,155!
696
    int32_t code = tRowIterOpen(pRow->pTSRow, pTSchema, &pIter->pIter);
39,156✔
697
    if (code) return code;
39,225!
698
  } else if (pRow->type == TSDBROW_COL_FMT) {
×
699
    pIter->iColData = 0;
×
700
  }
701

702
  return 0;
39,224✔
703
}
704

705
void tsdbRowClose(STSDBRowIter *pIter) {
39,051✔
706
  if (pIter->pRow && pIter->pRow->type == TSDBROW_ROW_FMT) {
39,051!
707
    tRowIterClose(&pIter->pIter);
39,052✔
708
  }
709
  pIter->pRow = NULL;
39,271✔
710
  pIter->pIter = NULL;
39,271✔
711
}
39,271✔
712

713
SColVal *tsdbRowIterNext(STSDBRowIter *pIter) {
840,661✔
714
  if (pIter->pRow->type == TSDBROW_ROW_FMT) {
840,661!
715
    return tRowIterNext(pIter->pIter);
841,215✔
716
  } else if (pIter->pRow->type == TSDBROW_COL_FMT) {
×
717
    if (pIter->iColData == 0) {
×
718
      SValue val = {.type = TSDB_DATA_TYPE_TIMESTAMP};
×
719
      VALUE_SET_TRIVIAL_DATUM(&val, pIter->pRow->pBlockData->aTSKEY[pIter->pRow->iRow]);
×
720
      pIter->cv = COL_VAL_VALUE(PRIMARYKEY_TIMESTAMP_COL_ID, val);
×
721
      ++pIter->iColData;
×
722
      return &pIter->cv;
×
723
    }
724

725
    if (pIter->iColData <= pIter->pRow->pBlockData->nColData) {
×
726
      if (tColDataGetValue(&pIter->pRow->pBlockData->aColData[pIter->iColData - 1], pIter->pRow->iRow, &pIter->cv) !=
×
727
          0) {
728
        return NULL;
×
729
      }
730
      ++pIter->iColData;
×
731
      return &pIter->cv;
×
732
    } else {
733
      return NULL;
×
734
    }
735
  } else {
736
    tsdbError("invalid row type:%d", pIter->pRow->type);
×
737
    return NULL;
×
738
  }
739
}
740

741
// SRowMerger ======================================================
742
int32_t tsdbRowMergerAdd(SRowMerger *pMerger, TSDBROW *pRow, STSchema *pTSchema) {
70,542,291✔
743
  int32_t   code = 0;
70,542,291✔
744
  TSDBKEY   key = TSDBROW_KEY(pRow);
70,542,291✔
745
  SColVal  *pColVal = &(SColVal){0};
70,542,291✔
746
  STColumn *pTColumn;
747
  int32_t   iCol, jCol = 1;
70,542,291✔
748
  pRow->arg = pMerger->arg;
70,542,291✔
749

750
  if (NULL == pTSchema) {
70,542,291✔
751
    pTSchema = pMerger->pTSchema;
61,998,442✔
752
  }
753

754
  if (taosArrayGetSize(pMerger->pArray) == 0) {
70,542,291✔
755
    // ts
756
    jCol = 0;
59,438,755✔
757
    pTColumn = &pTSchema->columns[jCol++];
59,438,755✔
758
    SValue val = {.type = pTColumn->type};
59,438,755✔
759
    VALUE_SET_TRIVIAL_DATUM(&val, key.ts);
59,438,755✔
760
    *pColVal = COL_VAL_VALUE(pTColumn->colId, val);
59,438,755✔
761
    if (taosArrayPush(pMerger->pArray, pColVal) == NULL) {
120,576,672!
762
      code = terrno;
×
763
      return code;
×
764
    }
765

766
    // other
767
    for (iCol = 1; jCol < pTSchema->numOfCols && iCol < pMerger->pTSchema->numOfCols; ++iCol) {
489,806,219✔
768
      pTColumn = &pMerger->pTSchema->columns[iCol];
428,727,098✔
769
      if (pTSchema->columns[jCol].colId < pTColumn->colId) {
428,727,098!
770
        ++jCol;
×
771
        --iCol;
×
772
        continue;
×
773
      } else if (pTSchema->columns[jCol].colId > pTColumn->colId) {
428,727,098!
774
        if (taosArrayPush(pMerger->pArray, &COL_VAL_NONE(pTColumn->colId, pTColumn->type)) == NULL) {
×
775
          return terrno;
×
776
        }
777
        continue;
×
778
      }
779

780
      tsdbRowGetColVal(pRow, pTSchema, jCol++, pColVal);
428,727,098✔
781
      bool usepData = IS_VAR_DATA_TYPE(pColVal->value.type) || pColVal->value.type == TSDB_DATA_TYPE_DECIMAL;
425,148,060!
782
      if ((!COL_VAL_IS_NONE(pColVal)) && (!COL_VAL_IS_NULL(pColVal)) && usepData) {
425,148,060✔
783
        uint8_t *pVal = pColVal->value.pData;
28,915,815✔
784

785
        pColVal->value.pData = NULL;
28,915,815✔
786
        code = tRealloc(&pColVal->value.pData, pColVal->value.nData);
28,915,815!
787
        if (code) {
28,890,748!
788
          return code;
×
789
        }
790

791
        if (pColVal->value.nData) {
28,890,748✔
792
          memcpy(pColVal->value.pData, pVal, pColVal->value.nData);
28,820,378✔
793
        }
794
      }
795

796
      if (taosArrayPush(pMerger->pArray, pColVal) == NULL) {
853,791,295!
797
        return terrno;
×
798
      }
799
    }
800

801
    for (; iCol < pMerger->pTSchema->numOfCols; ++iCol) {
61,079,130✔
802
      pTColumn = &pMerger->pTSchema->columns[iCol];
9✔
803
      if (taosArrayPush(pMerger->pArray, &COL_VAL_NONE(pTColumn->colId, pTColumn->type)) == NULL) {
18!
804
        return terrno;
×
805
      }
806
    }
807

808
    pMerger->version = key.version;
61,079,121✔
809
    return 0;
61,079,121✔
810
  } else {
811
    for (iCol = 1; iCol < pMerger->pTSchema->numOfCols && jCol < pTSchema->numOfCols; ++iCol) {
137,113,021✔
812
      pTColumn = &pMerger->pTSchema->columns[iCol];
126,105,582✔
813
      if (pTSchema->columns[jCol].colId < pTColumn->colId) {
126,105,582!
814
        ++jCol;
×
815
        --iCol;
×
816
        continue;
×
817
      } else if (pTSchema->columns[jCol].colId > pTColumn->colId) {
126,105,582!
818
        continue;
×
819
      }
820

821
      tsdbRowGetColVal(pRow, pTSchema, jCol++, pColVal);
126,105,582✔
822

823
      if (key.version > pMerger->version) {
126,159,980✔
824
        if (!COL_VAL_IS_NONE(pColVal)) {
124,876,199✔
825
          if (IS_VAR_DATA_TYPE(pColVal->value.type) || pColVal->value.type == TSDB_DATA_TYPE_DECIMAL) {
89,504,216!
826
            SColVal *pTColVal = taosArrayGet(pMerger->pArray, iCol);
15,033,565✔
827
            if (!pTColVal) return terrno;
15,192,059!
828
            if (!COL_VAL_IS_NULL(pColVal)) {
15,194,138✔
829
              code = tRealloc(&pTColVal->value.pData, pColVal->value.nData);
15,081,008✔
830
              if (code) return code;
15,089,719!
831

832
              pTColVal->value.nData = pColVal->value.nData;
15,089,719✔
833
              if (pTColVal->value.nData) {
15,089,719✔
834
                memcpy(pTColVal->value.pData, pColVal->value.pData, pTColVal->value.nData);
14,347,446✔
835
              }
836
              pTColVal->flag = 0;
15,089,719✔
837
            } else {
838
              tFree(pTColVal->value.pData);
113,130!
839
              taosArraySet(pMerger->pArray, iCol, pColVal);
113,158✔
840
            }
841
          } else {
842
            taosArraySet(pMerger->pArray, iCol, pColVal);
59,276,995✔
843
          }
844
        }
845
      } else if (key.version < pMerger->version) {
1,283,781!
846
        SColVal *tColVal = (SColVal *)taosArrayGet(pMerger->pArray, iCol);
1,283,781✔
847
        if (COL_VAL_IS_NONE(tColVal) && !COL_VAL_IS_NONE(pColVal)) {
1,283,735✔
848
          bool usepData = IS_VAR_DATA_TYPE(pColVal->value.type) || pColVal->value.type == TSDB_DATA_TYPE_DECIMAL;
252!
849
          if ((!COL_VAL_IS_NULL(pColVal)) && usepData) {
252!
850
            code = tRealloc(&tColVal->value.pData, pColVal->value.nData);
84!
851
            if (code) return code;
84!
852

853
            tColVal->value.nData = pColVal->value.nData;
84✔
854
            if (pColVal->value.nData) {
84!
855
              memcpy(tColVal->value.pData, pColVal->value.pData, pColVal->value.nData);
84✔
856
            }
857
            tColVal->flag = 0;
84✔
858
          } else {
859
            taosArraySet(pMerger->pArray, iCol, pColVal);
168✔
860
          }
861
        }
862
      } else {
863
        return TSDB_CODE_SUCCESS;  // same version, no need to merge
×
864
      }
865
    }
866

867
    pMerger->version = key.version;
11,007,439✔
868
    return code;
11,007,439✔
869
  }
870
}
871

872
int32_t tsdbRowMergerInit(SRowMerger *pMerger, STSchema *pSchema) {
1,191,734✔
873
  pMerger->pTSchema = pSchema;
1,191,734✔
874
  pMerger->pArray = taosArrayInit(pSchema->numOfCols, sizeof(SColVal));
1,191,734✔
875
  if (pMerger->pArray == NULL) {
1,192,500!
876
    return terrno;
×
877
  } else {
878
    return TSDB_CODE_SUCCESS;
1,192,535✔
879
  }
880
}
881

882
void tsdbRowMergerClear(SRowMerger *pMerger) {
61,741,067✔
883
  for (int32_t iCol = 1; iCol < pMerger->pTSchema->numOfCols; iCol++) {
484,573,825✔
884
    SColVal *pTColVal = taosArrayGet(pMerger->pArray, iCol);
422,956,594✔
885
    if (IS_VAR_DATA_TYPE(pTColVal->value.type) || pTColVal->value.type == TSDB_DATA_TYPE_DECIMAL) {
422,786,408!
886
      tFree(pTColVal->value.pData);
128,586,557!
887
    }
888
  }
889

890
  taosArrayClear(pMerger->pArray);
61,617,231✔
891
}
58,527,769✔
892

893
void tsdbRowMergerCleanup(SRowMerger *pMerger) {
1,213,414✔
894
  int32_t numOfCols = taosArrayGetSize(pMerger->pArray);
1,213,414✔
895
  for (int32_t iCol = 1; iCol < numOfCols; iCol++) {
1,213,479!
896
    SColVal *pTColVal = taosArrayGet(pMerger->pArray, iCol);
×
897
    if (IS_VAR_DATA_TYPE(pTColVal->value.type) || pTColVal->value.type == TSDB_DATA_TYPE_DECIMAL) {
×
898
      tFree(pTColVal->value.pData);
×
899
    }
900
  }
901

902
  taosArrayDestroy(pMerger->pArray);
1,213,479✔
903
}
1,213,489✔
904

905
int32_t tsdbRowMergerGetRow(SRowMerger *pMerger, SRow **ppRow) {
59,529,194✔
906
  SRowBuildScanInfo scanInfo = {.hasBlob = 0};
59,529,194✔
907
  return tRowBuild(pMerger->pArray, pMerger->pTSchema, ppRow, &scanInfo);
59,529,194✔
908
}
909

910
// delete skyline ======================================================
911
static void tsdbMergeSkyline(SArray *pSkyline1, SArray *pSkyline2, SArray *pSkyline) {
341,581✔
912
  int32_t  i1 = 0;
341,581✔
913
  int32_t  n1 = taosArrayGetSize(pSkyline1);
341,581✔
914
  int32_t  i2 = 0;
341,581✔
915
  int32_t  n2 = taosArrayGetSize(pSkyline2);
341,581✔
916
  TSDBKEY *pKey1;
917
  TSDBKEY *pKey2;
918
  int64_t  version1 = 0;
341,581✔
919
  int64_t  version2 = 0;
341,581✔
920

921
  taosArrayClear(pSkyline);
341,581✔
922
  TSDBKEY **pItem = TARRAY_GET_ELEM(pSkyline, 0);
341,581✔
923

924
  while (i1 < n1 && i2 < n2) {
2,130,556✔
925
    pKey1 = (TSDBKEY *)taosArrayGetP(pSkyline1, i1);
1,788,975✔
926
    pKey2 = (TSDBKEY *)taosArrayGetP(pSkyline2, i2);
1,788,975✔
927

928
    if (pKey1->ts < pKey2->ts) {
1,788,975✔
929
      version1 = pKey1->version;
949,753✔
930
      *pItem = pKey1;
949,753✔
931
      i1++;
949,753✔
932
    } else if (pKey1->ts > pKey2->ts) {
839,222✔
933
      version2 = pKey2->version;
430,300✔
934
      *pItem = pKey2;
430,300✔
935
      i2++;
430,300✔
936
    } else {
937
      version1 = pKey1->version;
408,922✔
938
      version2 = pKey2->version;
408,922✔
939
      *pItem = pKey1;
408,922✔
940
      i1++;
408,922✔
941
      i2++;
408,922✔
942
    }
943

944
    (*pItem)->version = TMAX(version1, version2);
1,788,975✔
945
    pItem++;
1,788,975✔
946
  }
947

948
  while (i1 < n1) {
421,795✔
949
    pKey1 = (TSDBKEY *)taosArrayGetP(pSkyline1, i1);
80,214✔
950
    *pItem = pKey1;
80,214✔
951
    pItem++;
80,214✔
952
    i1++;
80,214✔
953
  }
954

955
  while (i2 < n2) {
788,335✔
956
    pKey2 = (TSDBKEY *)taosArrayGetP(pSkyline2, i2);
446,754✔
957
    *pItem = pKey2;
446,754✔
958
    pItem++;
446,754✔
959
    i2++;
446,754✔
960
  }
961

962
  pSkyline->size = TARRAY_ELEM_IDX(pSkyline, pItem);
341,581✔
963
}
341,581✔
964

965
int32_t tsdbBuildDeleteSkylineImpl(SArray *aSkyline, int32_t sidx, int32_t eidx, SArray *pSkyline) {
714,656✔
966
  int32_t   code = 0;
714,656✔
967
  SDelData *pDelData;
968
  int32_t   midx;
969

970
  taosArrayClear(pSkyline);
714,656✔
971
  if (sidx == eidx) {
714,659✔
972
    TSDBKEY *pItem1 = taosArrayGet(aSkyline, sidx * 2);
373,080✔
973
    TSDBKEY *pItem2 = taosArrayGet(aSkyline, sidx * 2 + 1);
373,079✔
974
    if (taosArrayPush(pSkyline, &pItem1) == NULL) {
373,080!
975
      return terrno;
×
976
    }
977

978
    if (taosArrayPush(pSkyline, &pItem2) == NULL) {
373,081!
979
      return terrno;
×
980
    }
981
  } else {
982
    SArray *pSkyline1 = NULL;
341,579✔
983
    SArray *pSkyline2 = NULL;
341,579✔
984
    midx = (sidx + eidx) / 2;
341,579✔
985

986
    pSkyline1 = taosArrayInit((midx - sidx + 1) * 2, POINTER_BYTES);
341,579✔
987
    if (pSkyline1 == NULL) {
341,581!
988
      return terrno;
×
989
    }
990
    pSkyline2 = taosArrayInit((eidx - midx) * 2, POINTER_BYTES);
341,581✔
991
    if (pSkyline2 == NULL) {
341,581!
992
      taosArrayDestroy(pSkyline1);
×
993
      return terrno;
×
994
    }
995

996
    code = tsdbBuildDeleteSkylineImpl(aSkyline, sidx, midx, pSkyline1);
341,581✔
997
    if (code) goto _clear;
341,581!
998

999
    code = tsdbBuildDeleteSkylineImpl(aSkyline, midx + 1, eidx, pSkyline2);
341,581✔
1000
    if (code) goto _clear;
341,581!
1001

1002
    tsdbMergeSkyline(pSkyline1, pSkyline2, pSkyline);
341,581✔
1003

1004
  _clear:
341,581✔
1005
    taosArrayDestroy(pSkyline1);
341,581✔
1006
    taosArrayDestroy(pSkyline2);
341,581✔
1007
  }
1008

1009
  return code;
714,657✔
1010
}
1011

1012
int32_t tsdbBuildDeleteSkyline(SArray *aDelData, int32_t sidx, int32_t eidx, SArray *aSkyline) {
31,497✔
1013
  SDelData *pDelData;
1014
  int32_t   code = 0;
31,497✔
1015
  int32_t   dataNum = eidx - sidx + 1;
31,497✔
1016
  SArray   *aTmpSkyline = taosArrayInit(dataNum * 2, sizeof(TSDBKEY));
31,497✔
1017
  if (aTmpSkyline == NULL) {
31,500!
1018
    return terrno;
×
1019
  }
1020

1021
  SArray *pSkyline = taosArrayInit(dataNum * 2, POINTER_BYTES);
31,500✔
1022
  if (pSkyline == NULL) {
31,500!
1023
    taosArrayDestroy(aTmpSkyline);
×
1024
    return terrno;
×
1025
  }
1026

1027
  taosArrayClear(aSkyline);
31,500✔
1028
  for (int32_t i = sidx; i <= eidx; ++i) {
404,572✔
1029
    pDelData = (SDelData *)taosArrayGet(aDelData, i);
373,070✔
1030
    if (taosArrayPush(aTmpSkyline, &(TSDBKEY){.ts = pDelData->sKey, .version = pDelData->version}) == NULL) {
746,152!
1031
      code = terrno;
×
1032
      goto _clear;
×
1033
    }
1034

1035
    if (taosArrayPush(aTmpSkyline, &(TSDBKEY){.ts = pDelData->eKey, .version = 0}) == NULL) {
746,154!
1036
      code = terrno;
×
1037
      goto _clear;
×
1038
    }
1039
  }
1040

1041
  code = tsdbBuildDeleteSkylineImpl(aTmpSkyline, sidx, eidx, pSkyline);
31,502✔
1042
  if (code) goto _clear;
31,495!
1043

1044
  int32_t skylineNum = taosArrayGetSize(pSkyline);
31,495✔
1045
  for (int32_t i = 0; i < skylineNum; ++i) {
368,735✔
1046
    TSDBKEY *p = taosArrayGetP(pSkyline, i);
337,237✔
1047
    if (taosArrayPush(aSkyline, p) == NULL) {
337,238!
1048
      code = terrno;
×
1049
      goto _clear;
×
1050
    }
1051
  }
1052

1053
_clear:
31,498✔
1054
  taosArrayDestroy(aTmpSkyline);
31,498✔
1055
  taosArrayDestroy(pSkyline);
31,498✔
1056

1057
  return code;
31,500✔
1058
}
1059

1060
/*
1061
int32_t tsdbBuildDeleteSkyline2(SArray *aDelData, int32_t sidx, int32_t eidx, SArray *aSkyline) {
1062
  int32_t   code = 0;
1063
  SDelData *pDelData;
1064
  int32_t   midx;
1065

1066
  taosArrayClear(aSkyline);
1067
  if (sidx == eidx) {
1068
    pDelData = (SDelData *)taosArrayGet(aDelData, sidx);
1069
    taosArrayPush(aSkyline, &(TSDBKEY){.ts = pDelData->sKey, .version = pDelData->version});
1070
    taosArrayPush(aSkyline, &(TSDBKEY){.ts = pDelData->eKey, .version = 0});
1071
  } else {
1072
    SArray *aSkyline1 = NULL;
1073
    SArray *aSkyline2 = NULL;
1074

1075
    aSkyline1 = taosArrayInit(0, sizeof(TSDBKEY));
1076
    aSkyline2 = taosArrayInit(0, sizeof(TSDBKEY));
1077
    if (aSkyline1 == NULL || aSkyline2 == NULL) {
1078
      code = TSDB_CODE_OUT_OF_MEMORY;
1079
      goto _clear;
1080
    }
1081
    midx = (sidx + eidx) / 2;
1082

1083
    code = tsdbBuildDeleteSkyline(aDelData, sidx, midx, aSkyline1);
1084
    if (code) goto _clear;
1085

1086
    code = tsdbBuildDeleteSkyline(aDelData, midx + 1, eidx, aSkyline2);
1087
    if (code) goto _clear;
1088

1089
    code = tsdbMergeSkyline(aSkyline1, aSkyline2, aSkyline);
1090

1091
  _clear:
1092
    taosArrayDestroy(aSkyline1);
1093
    taosArrayDestroy(aSkyline2);
1094
  }
1095

1096
  return code;
1097
}
1098
*/
1099

1100
// SBlockData ======================================================
1101
int32_t tBlockDataCreate(SBlockData *pBlockData) {
5,749,564✔
1102
  pBlockData->suid = 0;
5,749,564✔
1103
  pBlockData->uid = 0;
5,749,564✔
1104
  pBlockData->nRow = 0;
5,749,564✔
1105
  pBlockData->aUid = NULL;
5,749,564✔
1106
  pBlockData->aVersion = NULL;
5,749,564✔
1107
  pBlockData->aTSKEY = NULL;
5,749,564✔
1108
  pBlockData->nColData = 0;
5,749,564✔
1109
  pBlockData->aColData = NULL;
5,749,564✔
1110
  return 0;
5,749,564✔
1111
}
1112

1113
void tBlockDataDestroy(SBlockData *pBlockData) {
5,832,269✔
1114
  tFree(pBlockData->aUid);
5,832,269!
1115
  tFree(pBlockData->aVersion);
5,832,339!
1116
  tFree(pBlockData->aTSKEY);
5,832,332!
1117

1118
  for (int32_t i = 0; i < pBlockData->nColData; i++) {
9,745,507✔
1119
    tColDataDestroy(&pBlockData->aColData[i]);
3,913,164✔
1120
  }
1121

1122
  if (pBlockData->aColData) {
5,832,343✔
1123
    taosMemoryFree(pBlockData->aColData);
1,061,162!
1124
    pBlockData->aColData = NULL;
1,061,188✔
1125
  }
1126
}
5,832,369✔
1127

1128
static int32_t tBlockDataAdjustColData(SBlockData *pBlockData, int32_t nColData) {
119,824✔
1129
  int32_t code = 0;
119,824✔
1130

1131
  if (pBlockData->nColData > nColData) {
119,824✔
1132
    for (int32_t i = nColData; i < pBlockData->nColData; i++) {
24,870✔
1133
      tColDataDestroy(&pBlockData->aColData[i]);
24,291✔
1134
    }
1135
  } else if (pBlockData->nColData < nColData) {
119,245✔
1136
    SColData *aColData = taosMemoryRealloc(pBlockData->aColData, sizeof(SBlockData) * nColData);
95,357!
1137
    if (aColData == NULL) {
95,361!
1138
      code = terrno;
×
1139
      goto _exit;
×
1140
    }
1141

1142
    pBlockData->aColData = aColData;
95,361✔
1143
    memset(&pBlockData->aColData[pBlockData->nColData], 0, sizeof(SBlockData) * (nColData - pBlockData->nColData));
95,361✔
1144
  }
1145
  pBlockData->nColData = nColData;
119,828✔
1146

1147
_exit:
119,828✔
1148
  return code;
119,828✔
1149
}
1150
int32_t tBlockDataInit(SBlockData *pBlockData, TABLEID *pId, STSchema *pTSchema, int16_t *aCid, int32_t nCid) {
119,866✔
1151
  int32_t code = 0;
119,866✔
1152

1153
  if (!pId->suid && !pId->uid) {
119,866!
1154
    return TSDB_CODE_INVALID_PARA;
×
1155
  }
1156

1157
  pBlockData->suid = pId->suid;
119,866✔
1158
  pBlockData->uid = pId->uid;
119,866✔
1159
  pBlockData->nRow = 0;
119,866✔
1160

1161
  if (aCid) {
119,866!
1162
    code = tBlockDataAdjustColData(pBlockData, nCid);
×
1163
    if (code) goto _exit;
×
1164

1165
    int32_t   iColumn = 1;
×
1166
    STColumn *pTColumn = &pTSchema->columns[iColumn];
×
1167
    for (int32_t iCid = 0; iCid < nCid; iCid++) {
×
1168
      // aCid array (from taos client catalog) contains columns that does not exist in the pTSchema. the pTSchema is
1169
      // newer
1170
      if (pTColumn == NULL) {
×
1171
        continue;
×
1172
      }
1173

1174
      while (pTColumn->colId < aCid[iCid]) {
×
1175
        iColumn++;
×
1176
        if (!(iColumn < pTSchema->numOfCols)) {
×
1177
          return TSDB_CODE_INVALID_PARA;
×
1178
        }
1179
        pTColumn = &pTSchema->columns[iColumn];
×
1180
      }
1181

1182
      if (pTColumn->colId != aCid[iCid]) {
×
1183
        continue;
×
1184
      }
1185

1186
      tColDataInit(&pBlockData->aColData[iCid], pTColumn->colId, pTColumn->type, pTColumn->flags);
×
1187

1188
      iColumn++;
×
1189
      pTColumn = (iColumn < pTSchema->numOfCols) ? &pTSchema->columns[iColumn] : NULL;
×
1190
    }
1191
  } else {
1192
    code = tBlockDataAdjustColData(pBlockData, pTSchema->numOfCols - 1);
119,866✔
1193
    if (code) goto _exit;
119,830!
1194

1195
    for (int32_t iColData = 0; iColData < pBlockData->nColData; iColData++) {
1,198,256✔
1196
      STColumn *pTColumn = &pTSchema->columns[iColData + 1];
1,078,342✔
1197
      tColDataInit(&pBlockData->aColData[iColData], pTColumn->colId, pTColumn->type, pTColumn->flags);
1,078,342✔
1198
    }
1199
  }
1200

1201
_exit:
119,914✔
1202
  return code;
119,914✔
1203
}
1204

1205
void tBlockDataReset(SBlockData *pBlockData) {
4,488,108✔
1206
  pBlockData->suid = 0;
4,488,108✔
1207
  pBlockData->uid = 0;
4,488,108✔
1208
  pBlockData->nRow = 0;
4,488,108✔
1209
  for (int32_t i = 0; i < pBlockData->nColData; i++) {
6,075,919✔
1210
    tColDataDestroy(&pBlockData->aColData[i]);
1,587,805✔
1211
  }
1212
  pBlockData->nColData = 0;
4,488,114✔
1213
  taosMemoryFreeClear(pBlockData->aColData);
4,488,114!
1214
}
4,488,122✔
1215

1216
void tBlockDataClear(SBlockData *pBlockData) {
198,674✔
1217
  pBlockData->nRow = 0;
198,674✔
1218
  for (int32_t iColData = 0; iColData < pBlockData->nColData; iColData++) {
1,425,524✔
1219
    tColDataClear(tBlockDataGetColDataByIdx(pBlockData, iColData));
1,226,825✔
1220
  }
1221
}
198,699✔
1222

1223
int32_t tBlockDataAddColData(SBlockData *pBlockData, int16_t cid, int8_t type, int8_t cflag, SColData **ppColData) {
4,700,758✔
1224
  if (pBlockData->nColData != 0 && pBlockData->aColData[pBlockData->nColData - 1].cid >= cid) {
4,700,758!
1225
    return TSDB_CODE_INVALID_PARA;
×
1226
  }
1227

1228
  SColData *newColData = taosMemoryRealloc(pBlockData->aColData, sizeof(SColData) * (pBlockData->nColData + 1));
4,700,758!
1229
  if (newColData == NULL) {
4,703,316!
1230
    return terrno;
×
1231
  }
1232

1233
  pBlockData->aColData = newColData;
4,703,316✔
1234
  pBlockData->nColData++;
4,703,316✔
1235

1236
  *ppColData = &pBlockData->aColData[pBlockData->nColData - 1];
4,703,316✔
1237
  memset(*ppColData, 0, sizeof(SColData));
4,703,316✔
1238
  tColDataInit(*ppColData, cid, type, cflag);
4,703,316✔
1239

1240
  return 0;
4,703,179✔
1241
}
1242

1243
/* flag > 0: forward update
1244
 * flag == 0: insert
1245
 * flag < 0: backward update
1246
 */
1247
static int32_t tBlockDataUpsertBlockRow(SBlockData *pBlockData, SBlockData *pBlockDataFrom, int32_t iRow,
37,674,128✔
1248
                                        int32_t flag) {
1249
  int32_t code = 0;
37,674,128✔
1250

1251
  SColVal   cv = {0};
37,674,128✔
1252
  int32_t   iColDataFrom = 0;
37,674,128✔
1253
  SColData *pColDataFrom = (iColDataFrom < pBlockDataFrom->nColData) ? &pBlockDataFrom->aColData[iColDataFrom] : NULL;
37,674,128!
1254

1255
  for (int32_t iColDataTo = 0; iColDataTo < pBlockData->nColData; iColDataTo++) {
373,918,215✔
1256
    SColData *pColDataTo = &pBlockData->aColData[iColDataTo];
336,643,108✔
1257

1258
    while (pColDataFrom && pColDataFrom->cid < pColDataTo->cid) {
336,643,108!
1259
      pColDataFrom = (++iColDataFrom < pBlockDataFrom->nColData) ? &pBlockDataFrom->aColData[iColDataFrom] : NULL;
×
1260
    }
1261

1262
    if (pColDataFrom == NULL || pColDataFrom->cid > pColDataTo->cid) {
337,454,821✔
1263
      cv = COL_VAL_NONE(pColDataTo->cid, pColDataTo->type);
811,658✔
1264
      if (flag == 0 && (code = tColDataAppendValue(pColDataTo, &cv))) goto _exit;
811,658!
1265
    } else {
1266
      code = tColDataGetValue(pColDataFrom, iRow, &cv);
335,831,450✔
1267
      if (code) goto _exit;
340,922,657!
1268

1269
      if (flag) {
340,922,657✔
1270
        code = tColDataUpdateValue(pColDataTo, &cv, flag > 0);
13,730,804✔
1271
      } else {
1272
        code = tColDataAppendValue(pColDataTo, &cv);
327,191,853✔
1273
      }
1274
      if (code) goto _exit;
335,432,374!
1275

1276
      pColDataFrom = (++iColDataFrom < pBlockDataFrom->nColData) ? &pBlockDataFrom->aColData[iColDataFrom] : NULL;
335,432,374✔
1277
    }
1278
  }
1279

1280
_exit:
37,275,107✔
1281
  return code;
37,275,107✔
1282
}
1283

1284
int32_t tBlockDataAppendRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema, int64_t uid) {
134,316,975✔
1285
  int32_t code = 0;
134,316,975✔
1286

1287
  if (!(pBlockData->suid || pBlockData->uid)) {
134,316,975!
1288
    return TSDB_CODE_INVALID_PARA;
×
1289
  }
1290

1291
  // uid
1292
  if (pBlockData->uid == 0) {
134,316,975✔
1293
    if (!uid) {
56,774,160!
1294
      return TSDB_CODE_INVALID_PARA;
×
1295
    }
1296
    code = tRealloc((uint8_t **)&pBlockData->aUid, sizeof(int64_t) * (pBlockData->nRow + 1));
56,774,160✔
1297
    if (code) goto _exit;
56,823,582!
1298
    pBlockData->aUid[pBlockData->nRow] = uid;
56,823,582✔
1299
  }
1300
  // version
1301
  code = tRealloc((uint8_t **)&pBlockData->aVersion, sizeof(int64_t) * (pBlockData->nRow + 1));
134,366,397!
1302
  if (code) goto _exit;
134,435,712!
1303
  pBlockData->aVersion[pBlockData->nRow] = TSDBROW_VERSION(pRow);
134,435,712✔
1304
  // timestamp
1305
  code = tRealloc((uint8_t **)&pBlockData->aTSKEY, sizeof(TSKEY) * (pBlockData->nRow + 1));
134,435,712!
1306
  if (code) goto _exit;
134,787,207!
1307
  pBlockData->aTSKEY[pBlockData->nRow] = TSDBROW_TS(pRow);
134,787,207✔
1308

1309
  if (pRow->type == TSDBROW_ROW_FMT) {
134,787,207✔
1310
    code = tRowUpsertColData(pRow->pTSRow, pTSchema, pBlockData->aColData, pBlockData->nColData, 0 /* append */);
98,783,368✔
1311
    if (code) goto _exit;
100,528,423!
1312
  } else if (pRow->type == TSDBROW_COL_FMT) {
36,003,839!
1313
    code = tBlockDataUpsertBlockRow(pBlockData, pRow->pBlockData, pRow->iRow, 0 /* append */);
36,532,334✔
1314
    if (code) goto _exit;
36,768,311!
1315
  } else {
1316
    return TSDB_CODE_INVALID_PARA;
×
1317
  }
1318
  pBlockData->nRow++;
137,296,734✔
1319

1320
_exit:
137,296,734✔
1321
  return code;
137,296,734✔
1322
}
1323
int32_t tBlockDataUpdateRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema) {
1,519,555✔
1324
  int32_t code = 0;
1,519,555✔
1325

1326
  // version
1327
  int64_t lversion = pBlockData->aVersion[pBlockData->nRow - 1];
1,519,555✔
1328
  int64_t rversion = TSDBROW_VERSION(pRow);
1,519,555✔
1329
  if (lversion == rversion) {
1,519,555!
1330
    return TSDB_CODE_SUCCESS;
×
1331
  }
1332
  if (rversion > lversion) {
1,519,555!
1333
    pBlockData->aVersion[pBlockData->nRow - 1] = rversion;
1,520,644✔
1334
  }
1335

1336
  // update other rows
1337
  if (pRow->type == TSDBROW_ROW_FMT) {
1,519,555✔
1338
    code = tRowUpsertColData(pRow->pTSRow, pTSchema, pBlockData->aColData, pBlockData->nColData,
351,590!
1339
                             (rversion > lversion) ? 1 : -1 /* update */);
1340
    if (code) goto _exit;
366,811!
1341
  } else if (pRow->type == TSDBROW_COL_FMT) {
1,167,965!
1342
    code = tBlockDataUpsertBlockRow(pBlockData, pRow->pBlockData, pRow->iRow, (rversion > lversion) ? 1 : -1);
1,172,105!
1343
    if (code) goto _exit;
1,187,789!
1344
  } else {
1345
    code = TSDB_CODE_INVALID_PARA;
×
1346
    goto _exit;
×
1347
  }
1348

1349
_exit:
1,550,460✔
1350
  return code;
1,550,460✔
1351
}
1352

1353
#ifdef BUILD_NO_CALL
1354
int32_t tBlockDataTryUpsertRow(SBlockData *pBlockData, TSDBROW *pRow, int64_t uid) {
1355
  if (pBlockData->nRow == 0) {
1356
    return 1;
1357
  } else if (pBlockData->aTSKEY[pBlockData->nRow - 1] == TSDBROW_TS(pRow)) {
1358
    return pBlockData->nRow;
1359
  } else {
1360
    return pBlockData->nRow + 1;
1361
  }
1362
}
1363

1364
int32_t tBlockDataUpsertRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema, int64_t uid) {
1365
  if (pBlockData->nRow > 0 && pBlockData->aTSKEY[pBlockData->nRow - 1] == TSDBROW_TS(pRow)) {
1366
    return tBlockDataUpdateRow(pBlockData, pRow, pTSchema);
1367
  } else {
1368
    return tBlockDataAppendRow(pBlockData, pRow, pTSchema, uid);
1369
  }
1370
}
1371
#endif
1372

1373
SColData *tBlockDataGetColData(SBlockData *pBlockData, int16_t cid) {
467,284,537✔
1374
  int32_t lidx = 0;
467,284,537✔
1375
  int32_t ridx = pBlockData->nColData - 1;
467,284,537✔
1376

1377
  while (lidx <= ridx) {
833,265,493✔
1378
    int32_t   midx = (lidx + ridx) >> 1;
482,461,195✔
1379
    SColData *pColData = tBlockDataGetColDataByIdx(pBlockData, midx);
482,461,195✔
1380
    int32_t   c = (pColData->cid == cid) ? 0 : ((pColData->cid > cid) ? 1 : -1);
482,461,195✔
1381

1382
    if (c == 0) {
482,461,195✔
1383
      return pColData;
116,480,239✔
1384
    } else if (c < 0) {
365,980,956✔
1385
      lidx = midx + 1;
203,782,442✔
1386
    } else {
1387
      ridx = midx - 1;
162,198,514✔
1388
    }
1389
  }
1390

1391
  return NULL;
350,804,298✔
1392
}
1393

1394
/* buffers[0]: SDiskDataHdr
1395
 * buffers[1]: key part: uid + version + ts + primary keys
1396
 * buffers[2]: SBlockCol part
1397
 * buffers[3]: regular column part
1398
 */
1399
int32_t tBlockDataCompress(SBlockData *bData, void *pCompr, SBuffer *buffers, SBuffer *assist) {
74,169✔
1400
  int32_t code = 0;
74,169✔
1401
  int32_t lino = 0;
74,169✔
1402

1403
  SColCompressInfo *pInfo = pCompr;
74,169✔
1404
  code = tsdbGetColCmprAlgFromSet(pInfo->pColCmpr, 1, &pInfo->defaultCmprAlg);
74,169✔
1405
  TAOS_UNUSED(code);
1406

1407
  SDiskDataHdr hdr = {
74,179✔
1408
      .delimiter = TSDB_FILE_DLMT,
1409
      .fmtVer = 2,
1410
      .suid = bData->suid,
74,179✔
1411
      .uid = bData->uid,
74,179✔
1412
      .szUid = 0,     // filled by compress key
1413
      .szVer = 0,     // filled by compress key
1414
      .szKey = 0,     // filled by compress key
1415
      .szBlkCol = 0,  // filled by this func
1416
      .nRow = bData->nRow,
74,179✔
1417
      .cmprAlg = pInfo->defaultCmprAlg,
74,179✔
1418
      .numOfPKs = 0,  // filled by compress key
1419
  };
1420
  // Key part
1421

1422
  tBufferClear(&buffers[1]);
74,179✔
1423
  code = tBlockDataCompressKeyPart(bData, &hdr, &buffers[1], assist, (SColCompressInfo *)pInfo);
74,179✔
1424
  TSDB_CHECK_CODE(code, lino, _exit);
74,188!
1425

1426
  // Regulart column part
1427
  tBufferClear(&buffers[2]);
74,188✔
1428
  tBufferClear(&buffers[3]);
74,188✔
1429
  for (int i = 0; i < bData->nColData; i++) {
902,971✔
1430
    SColData *colData = tBlockDataGetColDataByIdx(bData, i);
828,652✔
1431

1432
    if (colData->cflag & COL_IS_KEY) {
828,652✔
1433
      continue;
59,178✔
1434
    }
1435
    if (colData->flag == HAS_NONE) {
824,703✔
1436
      continue;
55,229✔
1437
    }
1438

1439
    SColDataCompressInfo cinfo = {
769,474✔
1440
        .cmprAlg = pInfo->defaultCmprAlg,
769,474✔
1441
    };
1442
    code = tsdbGetColCmprAlgFromSet(pInfo->pColCmpr, colData->cid, &cinfo.cmprAlg);
769,474✔
1443
    if (code < 0) {
1444
      //
1445
    }
1446

1447
    int32_t offset = buffers[3].size;
769,235✔
1448
    code = tColDataCompress(colData, &cinfo, &buffers[3], assist);
769,235✔
1449
    TSDB_CHECK_CODE(code, lino, _exit);
769,548!
1450

1451
    SBlockCol blockCol = (SBlockCol){.cid = cinfo.columnId,
769,548✔
1452
                                     .type = cinfo.dataType,
769,548✔
1453
                                     .cflag = cinfo.columnFlag,
769,548✔
1454
                                     .flag = cinfo.flag,
769,548✔
1455
                                     .szOrigin = cinfo.dataOriginalSize,
769,548✔
1456
                                     .szBitmap = cinfo.bitmapCompressedSize,
769,548✔
1457
                                     .szOffset = cinfo.offsetCompressedSize,
769,548✔
1458
                                     .szValue = cinfo.dataCompressedSize,
769,548✔
1459
                                     .offset = offset,
1460
                                     .alg = cinfo.cmprAlg};
769,548✔
1461

1462
    code = tPutBlockCol(&buffers[2], &blockCol, hdr.fmtVer, hdr.cmprAlg);
769,548✔
1463
    TSDB_CHECK_CODE(code, lino, _exit);
769,605!
1464
  }
1465
  hdr.szBlkCol = buffers[2].size;
74,319✔
1466

1467
  // SDiskDataHdr part
1468
  tBufferClear(&buffers[0]);
1469
  code = tPutDiskDataHdr(&buffers[0], &hdr);
74,319✔
1470
  TSDB_CHECK_CODE(code, lino, _exit);
74,094!
1471

1472
_exit:
74,094✔
1473
  return code;
74,094✔
1474
}
1475

1476
int32_t tBlockDataDecompress(SBufferReader *br, SBlockData *blockData, SBuffer *assist) {
20,258✔
1477
  int32_t       code = 0;
20,258✔
1478
  int32_t       lino = 0;
20,258✔
1479
  SDiskDataHdr  hdr = {0};
20,258✔
1480
  SCompressInfo cinfo;
1481

1482
  // SDiskDataHdr
1483
  code = tGetDiskDataHdr(br, &hdr);
20,258✔
1484
  TSDB_CHECK_CODE(code, lino, _exit);
20,256!
1485

1486
  tBlockDataReset(blockData);
20,256✔
1487
  blockData->suid = hdr.suid;
20,257✔
1488
  blockData->uid = hdr.uid;
20,257✔
1489
  blockData->nRow = hdr.nRow;
20,257✔
1490

1491
  // Key part
1492
  code = tBlockDataDecompressKeyPart(&hdr, br, blockData, assist);
20,257✔
1493
  TSDB_CHECK_CODE(code, lino, _exit);
20,257!
1494

1495
  // Column part
1496
  SBufferReader br2 = *br;
20,257✔
1497
  br->offset += hdr.szBlkCol;
20,257✔
1498
  for (uint32_t startOffset = br2.offset; br2.offset - startOffset < hdr.szBlkCol;) {
159,038✔
1499
    SBlockCol blockCol;
1500

1501
    code = tGetBlockCol(&br2, &blockCol, hdr.fmtVer, hdr.cmprAlg);
138,779✔
1502
    TSDB_CHECK_CODE(code, lino, _exit);
138,756!
1503
    if (blockCol.alg == 0) blockCol.alg = hdr.cmprAlg;
138,756✔
1504
    code = tBlockDataDecompressColData(&hdr, &blockCol, br, blockData, assist);
138,756✔
1505
    TSDB_CHECK_CODE(code, lino, _exit);
138,781!
1506
  }
1507

1508
_exit:
20,259✔
1509
  return code;
20,259✔
1510
}
1511

1512
// SDiskDataHdr ==============================
1513
int32_t tPutDiskDataHdr(SBuffer *buffer, const SDiskDataHdr *pHdr) {
74,147✔
1514
  int32_t code;
1515

1516
  if ((code = tBufferPutU32(buffer, pHdr->delimiter))) return code;
148,303!
1517
  if ((code = tBufferPutU32v(buffer, pHdr->fmtVer))) return code;
148,312!
1518
  if ((code = tBufferPutI64(buffer, pHdr->suid))) return code;
148,312!
1519
  if ((code = tBufferPutI64(buffer, pHdr->uid))) return code;
148,312!
1520
  if ((code = tBufferPutI32v(buffer, pHdr->szUid))) return code;
148,312!
1521
  if ((code = tBufferPutI32v(buffer, pHdr->szVer))) return code;
148,312!
1522
  if ((code = tBufferPutI32v(buffer, pHdr->szKey))) return code;
148,312!
1523
  if ((code = tBufferPutI32v(buffer, pHdr->szBlkCol))) return code;
148,312!
1524
  if ((code = tBufferPutI32v(buffer, pHdr->nRow))) return code;
148,312!
1525
  if (pHdr->fmtVer < 2) {
74,156!
1526
    if ((code = tBufferPutI8(buffer, pHdr->cmprAlg))) return code;
×
1527
  } else if (pHdr->fmtVer == 2) {
74,156✔
1528
    if ((code = tBufferPutU32(buffer, pHdr->cmprAlg))) return code;
148,046!
1529
  } else {
1530
    // more data fmt ver
1531
  }
1532
  if (pHdr->fmtVer >= 1) {
74,156✔
1533
    if ((code = tBufferPutI8(buffer, pHdr->numOfPKs))) return code;
148,064!
1534
    for (int i = 0; i < pHdr->numOfPKs; i++) {
77,980✔
1535
      if ((code = tPutBlockCol(buffer, &pHdr->primaryBlockCols[i], pHdr->fmtVer, pHdr->cmprAlg))) return code;
3,949!
1536
    }
1537
  }
1538

1539
  return 0;
74,155✔
1540
}
1541

1542
int32_t tGetDiskDataHdr(SBufferReader *br, SDiskDataHdr *pHdr) {
1,737,496✔
1543
  int32_t code;
1544

1545
  if ((code = tBufferGetU32(br, &pHdr->delimiter))) return code;
1,737,496!
1546
  if ((code = tBufferGetU32v(br, &pHdr->fmtVer))) return code;
1,738,828!
1547
  if ((code = tBufferGetI64(br, &pHdr->suid))) return code;
1,738,611!
1548
  if ((code = tBufferGetI64(br, &pHdr->uid))) return code;
1,738,170!
1549
  if ((code = tBufferGetI32v(br, &pHdr->szUid))) return code;
1,738,293!
1550
  if ((code = tBufferGetI32v(br, &pHdr->szVer))) return code;
1,737,596!
1551
  if ((code = tBufferGetI32v(br, &pHdr->szKey))) return code;
1,738,652!
1552
  if ((code = tBufferGetI32v(br, &pHdr->szBlkCol))) return code;
1,738,663!
1553
  if ((code = tBufferGetI32v(br, &pHdr->nRow))) return code;
1,738,864!
1554
  if (pHdr->fmtVer < 2) {
1,738,419!
1555
    int8_t cmprAlg = 0;
×
1556
    if ((code = tBufferGetI8(br, &cmprAlg))) return code;
×
1557
    pHdr->cmprAlg = cmprAlg;
×
1558
  } else if (pHdr->fmtVer == 2) {
1,738,419!
1559
    if ((code = tBufferGetU32(br, &pHdr->cmprAlg))) return code;
1,738,516!
1560
  } else {
1561
    // more data fmt ver
1562
  }
1563
  if (pHdr->fmtVer >= 1) {
1,738,646✔
1564
    if ((code = tBufferGetI8(br, &pHdr->numOfPKs))) return code;
1,738,255!
1565
    for (int i = 0; i < pHdr->numOfPKs; i++) {
1,811,395✔
1566
      if ((code = tGetBlockCol(br, &pHdr->primaryBlockCols[i], pHdr->fmtVer, pHdr->cmprAlg))) {
72,874!
1567
        return code;
×
1568
      }
1569
    }
1570
  } else {
1571
    pHdr->numOfPKs = 0;
391✔
1572
  }
1573

1574
  return 0;
1,738,912✔
1575
}
1576

1577
// ALGORITHM ==============================
1578
int32_t tPutColumnDataAgg(SBuffer *buffer, SColumnDataAgg *pColAgg) {
231,529✔
1579
  int32_t code;
1580

1581
  if (pColAgg->colId & DECIMAL_AGG_FLAG) {
231,529✔
1582
    if ((code = tBufferPutI32v(buffer, pColAgg->colId))) return code;
2,788!
1583
    if ((code = tBufferPutI16v(buffer, pColAgg->numOfNull))) return code;
2,788!
1584
    if ((code = tBufferPutU64(buffer, pColAgg->decimal128Sum[0]))) return code;
2,788!
1585
    if ((code = tBufferPutU64(buffer, pColAgg->decimal128Sum[1]))) return code;
2,788!
1586
    if ((code = tBufferPutU64(buffer, pColAgg->decimal128Max[0]))) return code;
2,788!
1587
    if ((code = tBufferPutU64(buffer, pColAgg->decimal128Max[1]))) return code;
2,788!
1588
    if ((code = tBufferPutU64(buffer, pColAgg->decimal128Min[0]))) return code;
2,788!
1589
    if ((code = tBufferPutU64(buffer, pColAgg->decimal128Min[1]))) return code;
2,788!
1590
    if ((code = tBufferPutU8(buffer, pColAgg->overflow))) return code;
2,788!
1591
  } else {
1592
    if ((code = tBufferPutI32v(buffer, pColAgg->colId))) return code;
460,270!
1593
    if ((code = tBufferPutI16v(buffer, pColAgg->numOfNull))) return code;
460,270!
1594
    if ((code = tBufferPutI64(buffer, pColAgg->sum))) return code;
460,270!
1595
    if ((code = tBufferPutI64(buffer, pColAgg->max))) return code;
460,270!
1596
    if ((code = tBufferPutI64(buffer, pColAgg->min))) return code;
460,270!
1597
  }
1598

1599
  return 0;
231,529✔
1600
}
1601

1602
int32_t tGetColumnDataAgg(SBufferReader *br, SColumnDataAgg *pColAgg) {
231,010✔
1603
  int32_t code;
1604

1605
  if ((code = tBufferGetI32v(br, &pColAgg->colId))) return code;
231,010!
1606
  if ((code = tBufferGetI16v(br, &pColAgg->numOfNull))) return code;
231,010!
1607
  if (pColAgg->colId & DECIMAL_AGG_FLAG) {
231,012✔
1608
    pColAgg->colId &= 0xFFFF;
46✔
1609
    if ((code = tBufferGetU64(br, &pColAgg->decimal128Sum[0]))) return code;
46!
1610
    if ((code = tBufferGetU64(br, &pColAgg->decimal128Sum[1]))) return code;
46!
1611
    if ((code = tBufferGetU64(br, &pColAgg->decimal128Max[0]))) return code;
46!
1612
    if ((code = tBufferGetU64(br, &pColAgg->decimal128Max[1]))) return code;
46!
1613
    if ((code = tBufferGetU64(br, &pColAgg->decimal128Min[0]))) return code;
46!
1614
    if ((code = tBufferGetU64(br, &pColAgg->decimal128Min[1]))) return code;
46!
1615
    if ((code = tBufferGetU8(br, &pColAgg->overflow))) return code;
46!
1616
  } else {
1617
    if ((code = tBufferGetI64(br, &pColAgg->sum))) return code;
230,966!
1618
    if ((code = tBufferGetI64(br, &pColAgg->max))) return code;
230,965!
1619
    if ((code = tBufferGetI64(br, &pColAgg->min))) return code;
230,964!
1620
  }
1621

1622
  return 0;
231,009✔
1623
}
1624

1625
static int32_t tBlockDataCompressKeyPart(SBlockData *bData, SDiskDataHdr *hdr, SBuffer *buffer, SBuffer *assist,
74,124✔
1626
                                         SColCompressInfo *compressInfo) {
1627
  int32_t       code = 0;
74,124✔
1628
  int32_t       lino = 0;
74,124✔
1629
  SCompressInfo cinfo;
1630

1631
  // uid
1632
  if (bData->uid == 0) {
74,124✔
1633
    cinfo = (SCompressInfo){
33,284✔
1634
        .cmprAlg = hdr->cmprAlg,
33,284✔
1635
        .dataType = TSDB_DATA_TYPE_BIGINT,
1636
        .originalSize = sizeof(int64_t) * bData->nRow,
33,284✔
1637
    };
1638
    code = tCompressDataToBuffer(bData->aUid, &cinfo, buffer, assist);
33,284✔
1639
    TSDB_CHECK_CODE(code, lino, _exit);
33,277!
1640
    hdr->szUid = cinfo.compressedSize;
33,277✔
1641
  }
1642

1643
  // version
1644
  cinfo = (SCompressInfo){
74,117✔
1645
      .cmprAlg = hdr->cmprAlg,
74,117✔
1646
      .dataType = TSDB_DATA_TYPE_BIGINT,
1647
      .originalSize = sizeof(int64_t) * bData->nRow,
74,117✔
1648
  };
1649
  code = tCompressDataToBuffer((uint8_t *)bData->aVersion, &cinfo, buffer, assist);
74,117✔
1650
  TSDB_CHECK_CODE(code, lino, _exit);
74,122!
1651
  hdr->szVer = cinfo.compressedSize;
74,122✔
1652

1653
  // ts
1654
  cinfo = (SCompressInfo){
74,122✔
1655
      .cmprAlg = hdr->cmprAlg,
74,122✔
1656
      .dataType = TSDB_DATA_TYPE_TIMESTAMP,
1657
      .originalSize = sizeof(TSKEY) * bData->nRow,
74,122✔
1658
  };
1659

1660
  code = tCompressDataToBuffer((uint8_t *)bData->aTSKEY, &cinfo, buffer, assist);
74,122✔
1661
  TSDB_CHECK_CODE(code, lino, _exit);
74,153!
1662
  hdr->szKey = cinfo.compressedSize;
74,153✔
1663

1664
  // primary keys
1665
  for (hdr->numOfPKs = 0; hdr->numOfPKs < bData->nColData; hdr->numOfPKs++) {
78,102!
1666
    if (!(hdr->numOfPKs <= TD_MAX_PK_COLS)) {
78,129!
1667
      return TSDB_CODE_INVALID_PARA;
×
1668
    }
1669

1670
    SBlockCol *blockCol = &hdr->primaryBlockCols[hdr->numOfPKs];
78,129✔
1671
    SColData  *colData = tBlockDataGetColDataByIdx(bData, hdr->numOfPKs);
78,129✔
1672

1673
    if ((colData->cflag & COL_IS_KEY) == 0) {
78,129✔
1674
      break;
74,182✔
1675
    }
1676

1677
    SColDataCompressInfo info = {
3,947✔
1678
        .cmprAlg = hdr->cmprAlg,
3,947✔
1679
    };
1680
    code = tsdbGetColCmprAlgFromSet(compressInfo->pColCmpr, colData->cid, &info.cmprAlg);
3,947✔
1681
    if (code < 0) {
1682
      // do nothing
1683
    } else {
1684
    }
1685

1686
    code = tColDataCompress(colData, &info, buffer, assist);
3,948✔
1687
    TSDB_CHECK_CODE(code, lino, _exit);
3,949!
1688

1689
    *blockCol = (SBlockCol){
3,949✔
1690
        .cid = info.columnId,
3,949✔
1691
        .type = info.dataType,
3,949✔
1692
        .cflag = info.columnFlag,
3,949✔
1693
        .flag = info.flag,
3,949✔
1694
        .szOrigin = info.dataOriginalSize,
3,949✔
1695
        .szBitmap = info.bitmapCompressedSize,
3,949✔
1696
        .szOffset = info.offsetCompressedSize,
3,949✔
1697
        .szValue = info.dataCompressedSize,
3,949✔
1698
        .offset = 0,
1699
        .alg = info.cmprAlg,
3,949✔
1700
    };
1701
  }
1702

1703
_exit:
×
1704
  return code;
74,155✔
1705
}
1706

1707
int32_t tBlockDataDecompressColData(const SDiskDataHdr *hdr, const SBlockCol *blockCol, SBufferReader *br,
4,701,131✔
1708
                                    SBlockData *blockData, SBuffer *assist) {
1709
  int32_t code = 0;
4,701,131✔
1710
  int32_t lino = 0;
4,701,131✔
1711

1712
  SColData *colData;
1713

1714
  code = tBlockDataAddColData(blockData, blockCol->cid, blockCol->type, blockCol->cflag, &colData);
4,701,131✔
1715
  TSDB_CHECK_CODE(code, lino, _exit);
4,703,097!
1716

1717
  SColDataCompressInfo info = {
3,751,735✔
1718
      .cmprAlg = blockCol->alg,
4,703,097✔
1719
      .columnFlag = blockCol->cflag,
4,703,097✔
1720
      .flag = blockCol->flag,
4,703,097✔
1721
      .dataType = blockCol->type,
4,703,097✔
1722
      .columnId = blockCol->cid,
4,703,097✔
1723
      .numOfData = hdr->nRow,
4,703,097✔
1724
      .bitmapOriginalSize = 0,
1725
      .bitmapCompressedSize = blockCol->szBitmap,
4,703,097✔
1726
      .offsetOriginalSize = blockCol->szOffset ? sizeof(int32_t) * hdr->nRow : 0,
4,703,097✔
1727
      .offsetCompressedSize = blockCol->szOffset,
4,703,097✔
1728
      .dataOriginalSize = blockCol->szOrigin,
4,703,097✔
1729
      .dataCompressedSize = blockCol->szValue,
4,703,097✔
1730
  };
1731

1732
  switch (blockCol->flag) {
4,703,097✔
1733
    case (HAS_NONE | HAS_NULL | HAS_VALUE):
2✔
1734
      info.bitmapOriginalSize = BIT2_SIZE(hdr->nRow);
2✔
1735
      break;
2✔
1736
    case (HAS_NONE | HAS_NULL):
814,322✔
1737
    case (HAS_NONE | HAS_VALUE):
1738
    case (HAS_NULL | HAS_VALUE):
1739
      info.bitmapOriginalSize = BIT1_SIZE(hdr->nRow);
814,322✔
1740
      break;
814,322✔
1741
  }
1742

1743
  code = tColDataDecompress(BR_PTR(br), &info, colData, assist);
4,703,097✔
1744
  TSDB_CHECK_CODE(code, lino, _exit);
4,700,906!
1745
  br->offset += blockCol->szBitmap + blockCol->szOffset + blockCol->szValue;
4,700,906✔
1746

1747
_exit:
4,700,906✔
1748
  return code;
4,700,906✔
1749
}
1750

1751
int32_t tBlockDataDecompressKeyPart(const SDiskDataHdr *hdr, SBufferReader *br, SBlockData *blockData,
1,735,988✔
1752
                                    SBuffer *assist) {
1753
  int32_t       code = 0;
1,735,988✔
1754
  int32_t       lino = 0;
1,735,988✔
1755
  SCompressInfo cinfo;
1756

1757
  // uid
1758
  if (hdr->szUid > 0) {
1,735,988✔
1759
    cinfo = (SCompressInfo){
1,309,773✔
1760
        .cmprAlg = hdr->cmprAlg,
1,309,773✔
1761
        .dataType = TSDB_DATA_TYPE_BIGINT,
1762
        .compressedSize = hdr->szUid,
1,309,773✔
1763
        .originalSize = sizeof(int64_t) * hdr->nRow,
1,309,773✔
1764
    };
1765

1766
    code = tRealloc((uint8_t **)&blockData->aUid, cinfo.originalSize);
1,309,773✔
1767
    TSDB_CHECK_CODE(code, lino, _exit);
1,311,128!
1768
    code = tDecompressData(BR_PTR(br), &cinfo, blockData->aUid, cinfo.originalSize, assist);
1,311,128✔
1769
    TSDB_CHECK_CODE(code, lino, _exit);
1,309,785!
1770
    br->offset += cinfo.compressedSize;
1,309,785✔
1771
  }
1772

1773
  // version
1774
  cinfo = (SCompressInfo){
1,736,000✔
1775
      .cmprAlg = hdr->cmprAlg,
1,736,000✔
1776
      .dataType = TSDB_DATA_TYPE_BIGINT,
1777
      .compressedSize = hdr->szVer,
1,736,000✔
1778
      .originalSize = sizeof(int64_t) * hdr->nRow,
1,736,000✔
1779
  };
1780
  code = tRealloc((uint8_t **)&blockData->aVersion, cinfo.originalSize);
1,736,000✔
1781
  TSDB_CHECK_CODE(code, lino, _exit);
1,738,596!
1782
  code = tDecompressData(BR_PTR(br), &cinfo, blockData->aVersion, cinfo.originalSize, assist);
1,738,596✔
1783
  TSDB_CHECK_CODE(code, lino, _exit);
1,739,390!
1784
  br->offset += cinfo.compressedSize;
1,739,390✔
1785

1786
  // ts
1787
  cinfo = (SCompressInfo){
1,739,390✔
1788
      .cmprAlg = hdr->cmprAlg,
1,739,390✔
1789
      .dataType = TSDB_DATA_TYPE_TIMESTAMP,
1790
      .compressedSize = hdr->szKey,
1,739,390✔
1791
      .originalSize = sizeof(TSKEY) * hdr->nRow,
1,739,390✔
1792
  };
1793
  code = tRealloc((uint8_t **)&blockData->aTSKEY, cinfo.originalSize);
1,739,390✔
1794
  TSDB_CHECK_CODE(code, lino, _exit);
1,739,175!
1795
  code = tDecompressData(BR_PTR(br), &cinfo, blockData->aTSKEY, cinfo.originalSize, assist);
1,739,175✔
1796
  TSDB_CHECK_CODE(code, lino, _exit);
1,739,394!
1797
  br->offset += cinfo.compressedSize;
1,739,394✔
1798

1799
  // primary keys
1800
  for (int i = 0; i < hdr->numOfPKs; i++) {
1,812,286✔
1801
    const SBlockCol *blockCol = &hdr->primaryBlockCols[i];
72,967✔
1802

1803
    if (!(blockCol->flag == HAS_VALUE)) {
72,967!
1804
      TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
×
1805
    }
1806
    if (!(blockCol->cflag & COL_IS_KEY)) {
72,967!
1807
      TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
×
1808
    }
1809

1810
    code = tBlockDataDecompressColData(hdr, blockCol, br, blockData, assist);
72,967✔
1811
    TSDB_CHECK_CODE(code, lino, _exit);
72,892!
1812
  }
1813

1814
_exit:
1,739,319✔
1815
  return code;
1,739,319✔
1816
}
1817

1818
int32_t tsdbGetColCmprAlgFromSet(SHashObj *set, int16_t colId, uint32_t *alg) {
847,475✔
1819
  if (set == NULL) return -1;
847,475✔
1820

1821
  uint32_t *ret = taosHashGet(set, &colId, sizeof(colId));
847,269✔
1822
  if (ret == NULL) {
847,006!
1823
    return TSDB_CODE_NOT_FOUND;
×
1824
  }
1825

1826
  *alg = *ret;
847,006✔
1827
  return 0;
847,006✔
1828
}
1829
uint32_t tsdbCvtTimestampAlg(uint32_t alg) {
×
1830
  DEFINE_VAR(alg)
×
1831

1832
  return 0;
×
1833
}
1834

1835
int32_t tsdbAllocateDisk(STsdb *tsdb, const char *label, int32_t expLevel, SDiskID *diskId) {
33,069✔
1836
  int32_t code = 0;
33,069✔
1837
  int32_t lino = 0;
33,069✔
1838
  SDiskID did = {0};
33,069✔
1839
  STfs   *tfs = tsdb->pVnode->pTfs;
33,069✔
1840

1841
  code = tfsAllocDisk(tfs, expLevel, label, &did);
33,069✔
1842
  if (code) {
33,082!
1843
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, __LINE__,
×
1844
              tstrerror(code));
1845
    return code;
×
1846
  }
1847

1848
  if (tfsMkdirRecurAt(tfs, tsdb->path, did) != 0) {
33,082!
1849
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, __LINE__,
×
1850
              tstrerror(code));
1851
  }
1852

1853
  if (diskId) {
33,080✔
1854
    *diskId = did;
33,079✔
1855
  }
1856
  return code;
33,080✔
1857
}
1858

1859
int32_t tsdbAllocateDiskAtLevel(STsdb *tsdb, int32_t level, const char *label, SDiskID *diskId) {
170✔
1860
  int32_t code = 0;
170✔
1861
  SDiskID did = {0};
170✔
1862
  STfs   *tfs = tsdb->pVnode->pTfs;
170✔
1863

1864
  code = tfsAllocDiskAtLevel(tfs, level, label, &did);
170✔
1865
  if (code) {
170!
1866
    return code;
×
1867
  }
1868

1869
  if (tfsMkdirRecurAt(tfs, tsdb->path, did) != 0) {
170!
1870
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(tsdb->pVnode), __func__, __FILE__, __LINE__,
×
1871
              tstrerror(code));
1872
  }
1873

1874
  if (diskId) {
170!
1875
    *diskId = did;
170✔
1876
  }
1877
  return 0;
170✔
1878
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc