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

taosdata / TDengine / #4548

22 Jul 2025 02:37AM UTC coverage: 54.273% (-3.0%) from 57.287%
#4548

push

travis-ci

GitHub
Merge pull request #32061 from taosdata/new_testcases

132738 of 315239 branches covered (42.11%)

Branch coverage included in aggregate %.

201371 of 300373 relevant lines covered (67.04%)

3475977.14 hits per line

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

65.77
/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) {
406,521✔
404
  int32_t code;
405

406
  if ((code = tBufferPutI16v(buffer, pBlockCol->cid))) return code;
813,041!
407
  if ((code = tBufferPutI8(buffer, pBlockCol->type))) return code;
813,040!
408
  if ((code = tBufferPutI8(buffer, pBlockCol->cflag))) return code;
813,040!
409
  if ((code = tBufferPutI8(buffer, pBlockCol->flag))) return code;
813,040!
410
  if ((code = tBufferPutI32v(buffer, pBlockCol->szOrigin))) return code;
813,040!
411

412
  if (pBlockCol->flag != HAS_NULL) {
406,520✔
413
    if (pBlockCol->flag != HAS_VALUE) {
394,412✔
414
      if ((code = tBufferPutI32v(buffer, pBlockCol->szBitmap))) return code;
216,384!
415
    }
416

417
    if (IS_VAR_DATA_TYPE(pBlockCol->type)) {
394,412!
418
      if ((code = tBufferPutI32v(buffer, pBlockCol->szOffset))) return code;
138,354!
419
    }
420

421
    if (pBlockCol->flag != (HAS_NULL | HAS_NONE)) {
394,412✔
422
      if ((code = tBufferPutI32v(buffer, pBlockCol->szValue))) return code;
787,654!
423
    }
424

425
    if ((code = tBufferPutI32v(buffer, pBlockCol->offset))) return code;
788,824!
426
  }
427
  if (ver >= BLOCK_WITH_ALG_VER) {
406,520!
428
    if ((code = tBufferPutU32(buffer, pBlockCol->alg))) return code;
813,198!
429
  } else {
430
    if ((code = tBufferPutU32(buffer, defaultCmprAlg))) return code;
×
431
  }
432
  return 0;
406,520✔
433
}
434

435
int32_t tGetBlockCol(SBufferReader *br, SBlockCol *pBlockCol, int32_t ver, uint32_t defaultCmprAlg) {
9,537,822✔
436
  int32_t code;
437

438
  if ((code = tBufferGetI16v(br, &pBlockCol->cid))) return code;
9,537,822!
439
  if ((code = tBufferGetI8(br, &pBlockCol->type))) return code;
9,539,774!
440
  if ((code = tBufferGetI8(br, &pBlockCol->cflag))) return code;
9,537,783!
441
  if ((code = tBufferGetI8(br, &pBlockCol->flag))) return code;
9,531,249!
442
  if ((code = tBufferGetI32v(br, &pBlockCol->szOrigin))) return code;
9,529,183!
443

444
  pBlockCol->szBitmap = 0;
9,532,543✔
445
  pBlockCol->szOffset = 0;
9,532,543✔
446
  pBlockCol->szValue = 0;
9,532,543✔
447
  pBlockCol->offset = 0;
9,532,543✔
448

449
  if (pBlockCol->flag != HAS_NULL) {
9,532,543✔
450
    if (pBlockCol->flag != HAS_VALUE) {
8,794,780✔
451
      if ((code = tBufferGetI32v(br, &pBlockCol->szBitmap))) return code;
2,401,234!
452
    }
453

454
    if (IS_VAR_DATA_TYPE(pBlockCol->type)) {
8,795,393!
455
      if ((code = tBufferGetI32v(br, &pBlockCol->szOffset))) return code;
1,379,434!
456
    }
457

458
    if (pBlockCol->flag != (HAS_NULL | HAS_NONE)) {
8,800,420✔
459
      if ((code = tBufferGetI32v(br, &pBlockCol->szValue))) return code;
8,795,228!
460
    }
461

462
    if ((code = tBufferGetI32v(br, &pBlockCol->offset))) return code;
8,803,100!
463
  }
464

465
  if (ver >= BLOCK_WITH_ALG_VER) {
9,533,471!
466
    if ((code = tBufferGetU32(br, &pBlockCol->alg))) return code;
9,534,333!
467
  } else {
468
    pBlockCol->alg = defaultCmprAlg;
×
469
  }
470

471
  return 0;
9,533,018✔
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) {
159,578✔
556
  int64_t fid;
557
  if (key < 0) {
159,578✔
558
    fid = ((key + 1) / tsTickPerMin[precision] / minutes - 1);
6,229✔
559
    return (fid < INT32_MIN) ? INT32_MIN : (int32_t)fid;
6,229✔
560
  } else {
561
    fid = ((key / tsTickPerMin[precision] / minutes));
153,349✔
562
    return (fid > INT32_MAX) ? INT32_MAX : (int32_t)fid;
153,349✔
563
  }
564
}
565

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

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

575
  if (pKeepCfg->precision == TSDB_TIME_PRECISION_MILLI) {
15,738✔
576
    nowSec = nowSec * 1000;
15,610✔
577
  } else if (pKeepCfg->precision == TSDB_TIME_PRECISION_MICRO) {
128✔
578
    nowSec = nowSec * 1000000l;
62✔
579
  } else if (pKeepCfg->precision == TSDB_TIME_PRECISION_NANO) {
66!
580
    nowSec = nowSec * 1000000000l;
66✔
581
  } else {
582
    tsdbError("invalid time precision:%d", pKeepCfg->precision);
×
583
    return 0;
×
584
  }
585

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

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

595
  if (fid >= aFid[0]) {
15,738✔
596
    return 0;
15,300✔
597
  } else if (fid >= aFid[1]) {
438✔
598
    return 1;
138✔
599
  } else if (fid >= aFid[2]) {
300✔
600
    return 2;
294✔
601
  } else {
602
    return -1;
6✔
603
  }
604
}
605

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

612
  if (pRow->type == TSDBROW_ROW_FMT) {
428,021,143✔
613
    int32_t ret = tRowGet(pRow->pTSRow, pTSchema, iCol, pColVal);
77,049,135✔
614
    if (ret != 0) {
76,984,978!
615
      tsdbError("failed to get column value, code:%d", ret);
×
616
    }
617
  } else if (pRow->type == TSDBROW_COL_FMT) {
350,972,008!
618
    if (iCol == 0) {
360,569,044!
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);
360,569,044✔
624
      if (pColData) {
359,812,169✔
625
        if (tColDataGetValue(pColData, pRow->iRow, pColVal) != 0) {
98,067,209!
626
          tsdbError("failed to tColDataGetValue");
×
627
        }
628
      } else {
629
        *pColVal = COL_VAL_NONE(pTColumn->colId, pTColumn->type);
261,744,960✔
630
      }
631
    }
632
  }
633
}
427,799,383✔
634

635
void tsdbRowGetKey(TSDBROW *row, STsdbRowKey *key) {
386,404,480✔
636
  if (row->type == TSDBROW_ROW_FMT) {
386,404,480✔
637
    key->version = row->version;
182,732,528✔
638
    tRowGetKey(row->pTSRow, &key->key);
365,465,056✔
639
  } else {
640
    key->version = row->pBlockData->aVersion[row->iRow];
203,671,952✔
641
    tColRowGetKey(row->pBlockData, row->iRow, &key->key);
203,671,952!
642
  }
643
}
386,948,744✔
644

645
void tColRowGetPrimaryKey(SBlockData *pBlock, int32_t irow, SRowKey *key) {
309,168,242✔
646
  for (int32_t i = 0; i < pBlock->nColData; i++) {
309,280,199✔
647
    SColData *pColData = &pBlock->aColData[i];
257,059,536✔
648
    if (pColData->cflag & COL_IS_KEY) {
257,059,536✔
649
      SColVal cv;
650
      if (tColDataGetValue(pColData, irow, &cv) != 0) {
111,792!
651
        break;
×
652
      }
653
      key->pks[key->numOfPKs] = cv.value;
111,957✔
654
      key->numOfPKs++;
111,957✔
655
    } else {
656
      break;
256,947,744✔
657
    }
658
  }
659
}
309,168,407✔
660

661
int32_t tsdbRowKeyCmpr(const STsdbRowKey *key1, const STsdbRowKey *key2) {
29,115,200✔
662
  int32_t c = tRowKeyCompare(&key1->key, &key2->key);
29,115,200✔
663

664
  if (c) {
29,115,252✔
665
    return c;
27,466,866✔
666
  }
667

668
  if (key1->version < key2->version) {
1,648,386✔
669
    return -1;
1,324,291✔
670
  } else if (key1->version > key2->version) {
324,095!
671
    return 1;
342,016✔
672
  }
673
  return 0;
×
674
}
675

676
int32_t tsdbRowCompare(const void *p1, const void *p2) {
5,739,302✔
677
  STsdbRowKey key1, key2;
678

679
  tsdbRowGetKey((TSDBROW *)p1, &key1);
5,739,302✔
680
  tsdbRowGetKey((TSDBROW *)p2, &key2);
5,727,841✔
681
  return tsdbRowKeyCmpr(&key1, &key2);
5,691,749✔
682
}
683

684
int32_t tsdbRowCompareWithoutVersion(const void *p1, const void *p2) {
120,300,610✔
685
  STsdbRowKey key1, key2;
686

687
  tsdbRowGetKey((TSDBROW *)p1, &key1);
120,300,610✔
688
  tsdbRowGetKey((TSDBROW *)p2, &key2);
120,218,772✔
689
  return tRowKeyCompare(&key1.key, &key2.key);
120,274,961✔
690
}
691

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

702
  return 0;
21,002✔
703
}
704

705
void tsdbRowClose(STSDBRowIter *pIter) {
20,939✔
706
  if (pIter->pRow && pIter->pRow->type == TSDBROW_ROW_FMT) {
20,939!
707
    tRowIterClose(&pIter->pIter);
20,939✔
708
  }
709
  pIter->pRow = NULL;
21,033✔
710
  pIter->pIter = NULL;
21,033✔
711
}
21,033✔
712

713
SColVal *tsdbRowIterNext(STSDBRowIter *pIter) {
384,072✔
714
  if (pIter->pRow->type == TSDBROW_ROW_FMT) {
384,072!
715
    return tRowIterNext(pIter->pIter);
384,433✔
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) {
45,711,893✔
743
  int32_t   code = 0;
45,711,893✔
744
  TSDBKEY   key = TSDBROW_KEY(pRow);
45,711,893✔
745
  SColVal  *pColVal = &(SColVal){0};
45,711,893✔
746
  STColumn *pTColumn;
747
  int32_t   iCol, jCol = 1;
45,711,893✔
748
  pRow->arg = pMerger->arg;
45,711,893✔
749

750
  if (NULL == pTSchema) {
45,711,893✔
751
    pTSchema = pMerger->pTSchema;
39,617,384✔
752
  }
753

754
  if (taosArrayGetSize(pMerger->pArray) == 0) {
45,711,893✔
755
    // ts
756
    jCol = 0;
37,436,096✔
757
    pTColumn = &pTSchema->columns[jCol++];
37,436,096✔
758
    SValue val = {.type = pTColumn->type};
37,436,096✔
759
    VALUE_SET_TRIVIAL_DATUM(&val, key.ts);
37,436,096✔
760
    *pColVal = COL_VAL_VALUE(pTColumn->colId, val);
37,436,096✔
761
    if (taosArrayPush(pMerger->pArray, pColVal) == NULL) {
75,432,591!
762
      code = terrno;
×
763
      return code;
×
764
    }
765

766
    // other
767
    for (iCol = 1; jCol < pTSchema->numOfCols && iCol < pMerger->pTSchema->numOfCols; ++iCol) {
366,157,618✔
768
      pTColumn = &pMerger->pTSchema->columns[iCol];
326,980,310✔
769
      if (pTSchema->columns[jCol].colId < pTColumn->colId) {
326,980,310!
770
        ++jCol;
×
771
        --iCol;
×
772
        continue;
×
773
      } else if (pTSchema->columns[jCol].colId > pTColumn->colId) {
326,980,310!
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);
326,980,310✔
781
      bool usepData = IS_VAR_DATA_TYPE(pColVal->value.type) || pColVal->value.type == TSDB_DATA_TYPE_DECIMAL;
325,247,099!
782
      if ((!COL_VAL_IS_NONE(pColVal)) && (!COL_VAL_IS_NULL(pColVal)) && usepData) {
325,247,099✔
783
        uint8_t *pVal = pColVal->value.pData;
24,119,107✔
784

785
        pColVal->value.pData = NULL;
24,119,107✔
786
        code = tRealloc(&pColVal->value.pData, pColVal->value.nData);
24,119,107!
787
        if (code) {
24,107,065!
788
          return code;
×
789
        }
790

791
        if (pColVal->value.nData) {
24,107,065✔
792
          memcpy(pColVal->value.pData, pVal, pColVal->value.nData);
24,064,258✔
793
        }
794
      }
795

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

801
    for (; iCol < pMerger->pTSchema->numOfCols; ++iCol) {
39,177,317✔
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;
39,177,308✔
809
    return 0;
39,177,308✔
810
  } else {
811
    for (iCol = 1; iCol < pMerger->pTSchema->numOfCols && jCol < pTSchema->numOfCols; ++iCol) {
120,281,369✔
812
      pTColumn = &pMerger->pTSchema->columns[iCol];
111,938,735✔
813
      if (pTSchema->columns[jCol].colId < pTColumn->colId) {
111,938,735!
814
        ++jCol;
×
815
        --iCol;
×
816
        continue;
×
817
      } else if (pTSchema->columns[jCol].colId > pTColumn->colId) {
111,938,735!
818
        continue;
×
819
      }
820

821
      tsdbRowGetColVal(pRow, pTSchema, jCol++, pColVal);
111,938,735✔
822

823
      if (key.version > pMerger->version) {
111,918,268✔
824
        if (!COL_VAL_IS_NONE(pColVal)) {
111,904,027✔
825
          if (IS_VAR_DATA_TYPE(pColVal->value.type) || pColVal->value.type == TSDB_DATA_TYPE_DECIMAL) {
75,002,197!
826
            SColVal *pTColVal = taosArrayGet(pMerger->pArray, iCol);
11,242,194✔
827
            if (!pTColVal) return terrno;
11,493,974!
828
            if (!COL_VAL_IS_NULL(pColVal)) {
11,495,076✔
829
              code = tRealloc(&pTColVal->value.pData, pColVal->value.nData);
11,381,223✔
830
              if (code) return code;
11,386,869!
831

832
              pTColVal->value.nData = pColVal->value.nData;
11,386,869✔
833
              if (pTColVal->value.nData) {
11,386,869✔
834
                memcpy(pTColVal->value.pData, pColVal->value.pData, pTColVal->value.nData);
10,678,707✔
835
              }
836
              pTColVal->flag = 0;
11,386,869✔
837
            } else {
838
              tFree(pTColVal->value.pData);
113,853!
839
              taosArraySet(pMerger->pArray, iCol, pColVal);
113,885✔
840
            }
841
          } else {
842
            taosArraySet(pMerger->pArray, iCol, pColVal);
52,272,445✔
843
          }
844
        }
845
      } else if (key.version < pMerger->version) {
14,241!
846
        SColVal *tColVal = (SColVal *)taosArrayGet(pMerger->pArray, iCol);
14,241✔
847
        if (COL_VAL_IS_NONE(tColVal) && !COL_VAL_IS_NONE(pColVal)) {
14,239!
848
          bool usepData = IS_VAR_DATA_TYPE(pColVal->value.type) || pColVal->value.type == TSDB_DATA_TYPE_DECIMAL;
×
849
          if ((!COL_VAL_IS_NULL(pColVal)) && usepData) {
×
850
            code = tRealloc(&tColVal->value.pData, pColVal->value.nData);
×
851
            if (code) return code;
×
852

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

867
    pMerger->version = key.version;
8,342,634✔
868
    return code;
8,342,634✔
869
  }
870
}
871

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

882
void tsdbRowMergerClear(SRowMerger *pMerger) {
38,152,850✔
883
  for (int32_t iCol = 1; iCol < pMerger->pTSchema->numOfCols; iCol++) {
367,194,076✔
884
    SColVal *pTColVal = taosArrayGet(pMerger->pArray, iCol);
329,148,412✔
885
    if (IS_VAR_DATA_TYPE(pTColVal->value.type) || pTColVal->value.type == TSDB_DATA_TYPE_DECIMAL) {
329,015,133!
886
      tFree(pTColVal->value.pData);
88,753,181!
887
    }
888
  }
889

890
  taosArrayClear(pMerger->pArray);
38,045,664✔
891
}
37,177,958✔
892

893
void tsdbRowMergerCleanup(SRowMerger *pMerger) {
1,025,509✔
894
  int32_t numOfCols = taosArrayGetSize(pMerger->pArray);
1,025,509✔
895
  for (int32_t iCol = 1; iCol < numOfCols; iCol++) {
1,025,604!
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,025,604✔
903
}
1,025,595✔
904

905
int32_t tsdbRowMergerGetRow(SRowMerger *pMerger, SRow **ppRow) {
37,516,497✔
906
  SRowBuildScanInfo scanInfo = {.hasBlob = 0};
37,516,497✔
907
  return tRowBuild(pMerger->pArray, pMerger->pTSchema, ppRow, &scanInfo);
37,516,497✔
908
  // tRowBuildWithMerge(pMerger->pArray, pMerger->pTSchema, ppRow);
909
}
910

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

922
  taosArrayClear(pSkyline);
288✔
923
  TSDBKEY **pItem = TARRAY_GET_ELEM(pSkyline, 0);
288✔
924

925
  while (i1 < n1 && i2 < n2) {
1,054✔
926
    pKey1 = (TSDBKEY *)taosArrayGetP(pSkyline1, i1);
766✔
927
    pKey2 = (TSDBKEY *)taosArrayGetP(pSkyline2, i2);
766✔
928

929
    if (pKey1->ts < pKey2->ts) {
766✔
930
      version1 = pKey1->version;
348✔
931
      *pItem = pKey1;
348✔
932
      i1++;
348✔
933
    } else if (pKey1->ts > pKey2->ts) {
418✔
934
      version2 = pKey2->version;
186✔
935
      *pItem = pKey2;
186✔
936
      i2++;
186✔
937
    } else {
938
      version1 = pKey1->version;
232✔
939
      version2 = pKey2->version;
232✔
940
      *pItem = pKey1;
232✔
941
      i1++;
232✔
942
      i2++;
232✔
943
    }
944

945
    (*pItem)->version = TMAX(version1, version2);
766✔
946
    pItem++;
766✔
947
  }
948

949
  while (i1 < n1) {
480✔
950
    pKey1 = (TSDBKEY *)taosArrayGetP(pSkyline1, i1);
192✔
951
    *pItem = pKey1;
192✔
952
    pItem++;
192✔
953
    i1++;
192✔
954
  }
955

956
  while (i2 < n2) {
580✔
957
    pKey2 = (TSDBKEY *)taosArrayGetP(pSkyline2, i2);
292✔
958
    *pItem = pKey2;
292✔
959
    pItem++;
292✔
960
    i2++;
292✔
961
  }
962

963
  pSkyline->size = TARRAY_ELEM_IDX(pSkyline, pItem);
288✔
964
}
288✔
965

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

971
  taosArrayClear(pSkyline);
10,172✔
972
  if (sidx == eidx) {
10,172✔
973
    TSDBKEY *pItem1 = taosArrayGet(aSkyline, sidx * 2);
9,884✔
974
    TSDBKEY *pItem2 = taosArrayGet(aSkyline, sidx * 2 + 1);
9,884✔
975
    if (taosArrayPush(pSkyline, &pItem1) == NULL) {
9,884!
976
      return terrno;
×
977
    }
978

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

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

997
    code = tsdbBuildDeleteSkylineImpl(aSkyline, sidx, midx, pSkyline1);
288✔
998
    if (code) goto _clear;
288!
999

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

1003
    tsdbMergeSkyline(pSkyline1, pSkyline2, pSkyline);
288✔
1004

1005
  _clear:
288✔
1006
    taosArrayDestroy(pSkyline1);
288✔
1007
    taosArrayDestroy(pSkyline2);
288✔
1008
  }
1009

1010
  return code;
10,172✔
1011
}
1012

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

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

1028
  taosArrayClear(aSkyline);
9,596✔
1029
  for (int32_t i = sidx; i <= eidx; ++i) {
19,480✔
1030
    pDelData = (SDelData *)taosArrayGet(aDelData, i);
9,884✔
1031
    if (taosArrayPush(aTmpSkyline, &(TSDBKEY){.ts = pDelData->sKey, .version = pDelData->version}) == NULL) {
19,767!
1032
      code = terrno;
×
1033
      goto _clear;
×
1034
    }
1035

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

1042
  code = tsdbBuildDeleteSkylineImpl(aTmpSkyline, sidx, eidx, pSkyline);
9,596✔
1043
  if (code) goto _clear;
9,596!
1044

1045
  int32_t skylineNum = taosArrayGetSize(pSkyline);
9,596✔
1046
  for (int32_t i = 0; i < skylineNum; ++i) {
29,129✔
1047
    TSDBKEY *p = taosArrayGetP(pSkyline, i);
19,533✔
1048
    if (taosArrayPush(aSkyline, p) == NULL) {
19,533!
1049
      code = terrno;
×
1050
      goto _clear;
×
1051
    }
1052
  }
1053

1054
_clear:
9,596✔
1055
  taosArrayDestroy(aTmpSkyline);
9,596✔
1056
  taosArrayDestroy(pSkyline);
9,595✔
1057

1058
  return code;
9,595✔
1059
}
1060

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

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

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

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

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

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

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

1097
  return code;
1098
}
1099
*/
1100

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

1114
void tBlockDataDestroy(SBlockData *pBlockData) {
5,531,895✔
1115
  tFree(pBlockData->aUid);
5,531,895!
1116
  tFree(pBlockData->aVersion);
5,531,898!
1117
  tFree(pBlockData->aTSKEY);
5,531,891✔
1118

1119
  for (int32_t i = 0; i < pBlockData->nColData; i++) {
9,193,163✔
1120
    tColDataDestroy(&pBlockData->aColData[i]);
3,661,230✔
1121
  }
1122

1123
  if (pBlockData->aColData) {
5,531,933✔
1124
    taosMemoryFree(pBlockData->aColData);
952,859!
1125
    pBlockData->aColData = NULL;
952,881✔
1126
  }
1127
}
5,531,955✔
1128

1129
static int32_t tBlockDataAdjustColData(SBlockData *pBlockData, int32_t nColData) {
48,331✔
1130
  int32_t code = 0;
48,331✔
1131

1132
  if (pBlockData->nColData > nColData) {
48,331✔
1133
    for (int32_t i = nColData; i < pBlockData->nColData; i++) {
2,355✔
1134
      tColDataDestroy(&pBlockData->aColData[i]);
2,080✔
1135
    }
1136
  } else if (pBlockData->nColData < nColData) {
48,055✔
1137
    SColData *aColData = taosMemoryRealloc(pBlockData->aColData, sizeof(SBlockData) * nColData);
38,579!
1138
    if (aColData == NULL) {
38,580!
1139
      code = terrno;
×
1140
      goto _exit;
×
1141
    }
1142

1143
    pBlockData->aColData = aColData;
38,580✔
1144
    memset(&pBlockData->aColData[pBlockData->nColData], 0, sizeof(SBlockData) * (nColData - pBlockData->nColData));
38,580✔
1145
  }
1146
  pBlockData->nColData = nColData;
48,331✔
1147

1148
_exit:
48,331✔
1149
  return code;
48,331✔
1150
}
1151
int32_t tBlockDataInit(SBlockData *pBlockData, TABLEID *pId, STSchema *pTSchema, int16_t *aCid, int32_t nCid) {
48,324✔
1152
  int32_t code = 0;
48,324✔
1153

1154
  if (!pId->suid && !pId->uid) {
48,324!
1155
    return TSDB_CODE_INVALID_PARA;
×
1156
  }
1157

1158
  pBlockData->suid = pId->suid;
48,324✔
1159
  pBlockData->uid = pId->uid;
48,324✔
1160
  pBlockData->nRow = 0;
48,324✔
1161

1162
  if (aCid) {
48,324!
1163
    code = tBlockDataAdjustColData(pBlockData, nCid);
×
1164
    if (code) goto _exit;
×
1165

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

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

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

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

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

1196
    for (int32_t iColData = 0; iColData < pBlockData->nColData; iColData++) {
565,294✔
1197
      STColumn *pTColumn = &pTSchema->columns[iColData + 1];
516,971✔
1198
      tColDataInit(&pBlockData->aColData[iColData], pTColumn->colId, pTColumn->type, pTColumn->flags);
516,971✔
1199
    }
1200
  }
1201

1202
_exit:
48,323✔
1203
  return code;
48,323✔
1204
}
1205

1206
void tBlockDataReset(SBlockData *pBlockData) {
4,199,106✔
1207
  pBlockData->suid = 0;
4,199,106✔
1208
  pBlockData->uid = 0;
4,199,106✔
1209
  pBlockData->nRow = 0;
4,199,106✔
1210
  for (int32_t i = 0; i < pBlockData->nColData; i++) {
5,480,990✔
1211
    tColDataDestroy(&pBlockData->aColData[i]);
1,281,862✔
1212
  }
1213
  pBlockData->nColData = 0;
4,199,128✔
1214
  taosMemoryFreeClear(pBlockData->aColData);
4,199,128✔
1215
}
4,199,141✔
1216

1217
void tBlockDataClear(SBlockData *pBlockData) {
164,794✔
1218
  pBlockData->nRow = 0;
164,794✔
1219
  for (int32_t iColData = 0; iColData < pBlockData->nColData; iColData++) {
1,019,651✔
1220
    tColDataClear(tBlockDataGetColDataByIdx(pBlockData, iColData));
854,866✔
1221
  }
1222
}
164,785✔
1223

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

1229
  SColData *newColData = taosMemoryRealloc(pBlockData->aColData, sizeof(SColData) * (pBlockData->nColData + 1));
4,552,940!
1230
  if (newColData == NULL) {
4,556,051!
1231
    return terrno;
×
1232
  }
1233

1234
  pBlockData->aColData = newColData;
4,556,051✔
1235
  pBlockData->nColData++;
4,556,051✔
1236

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

1241
  return 0;
4,554,658✔
1242
}
1243

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

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

1256
  for (int32_t iColDataTo = 0; iColDataTo < pBlockData->nColData; iColDataTo++) {
327,150,476✔
1257
    SColData *pColDataTo = &pBlockData->aColData[iColDataTo];
298,081,413✔
1258

1259
    while (pColDataFrom && pColDataFrom->cid < pColDataTo->cid) {
298,081,413!
1260
      pColDataFrom = (++iColDataFrom < pBlockDataFrom->nColData) ? &pBlockDataFrom->aColData[iColDataFrom] : NULL;
×
1261
    }
1262

1263
    if (pColDataFrom == NULL || pColDataFrom->cid > pColDataTo->cid) {
298,732,039!
1264
      cv = COL_VAL_NONE(pColDataTo->cid, pColDataTo->type);
650,636✔
1265
      if (flag == 0 && (code = tColDataAppendValue(pColDataTo, &cv))) goto _exit;
650,636!
1266
    } else {
1267
      code = tColDataGetValue(pColDataFrom, iRow, &cv);
297,430,777✔
1268
      if (code) goto _exit;
301,161,187!
1269

1270
      if (flag) {
301,161,187✔
1271
        code = tColDataUpdateValue(pColDataTo, &cv, flag > 0);
13,029,667✔
1272
      } else {
1273
        code = tColDataAppendValue(pColDataTo, &cv);
288,131,520✔
1274
      }
1275
      if (code) goto _exit;
296,922,665!
1276

1277
      pColDataFrom = (++iColDataFrom < pBlockDataFrom->nColData) ? &pBlockDataFrom->aColData[iColDataFrom] : NULL;
296,922,665✔
1278
    }
1279
  }
1280

1281
_exit:
29,069,063✔
1282
  return code;
29,069,063✔
1283
}
1284

1285
int32_t tBlockDataAppendRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema, int64_t uid) {
79,322,015✔
1286
  int32_t code = 0;
79,322,015✔
1287

1288
  if (!(pBlockData->suid || pBlockData->uid)) {
79,322,015!
1289
    return TSDB_CODE_INVALID_PARA;
×
1290
  }
1291

1292
  // uid
1293
  if (pBlockData->uid == 0) {
79,322,015✔
1294
    if (!uid) {
29,601,811!
1295
      return TSDB_CODE_INVALID_PARA;
×
1296
    }
1297
    code = tRealloc((uint8_t **)&pBlockData->aUid, sizeof(int64_t) * (pBlockData->nRow + 1));
29,601,811!
1298
    if (code) goto _exit;
29,672,189!
1299
    pBlockData->aUid[pBlockData->nRow] = uid;
29,672,189✔
1300
  }
1301
  // version
1302
  code = tRealloc((uint8_t **)&pBlockData->aVersion, sizeof(int64_t) * (pBlockData->nRow + 1));
79,392,393!
1303
  if (code) goto _exit;
79,420,914!
1304
  pBlockData->aVersion[pBlockData->nRow] = TSDBROW_VERSION(pRow);
79,420,914✔
1305
  // timestamp
1306
  code = tRealloc((uint8_t **)&pBlockData->aTSKEY, sizeof(TSKEY) * (pBlockData->nRow + 1));
79,420,914✔
1307
  if (code) goto _exit;
79,446,380!
1308
  pBlockData->aTSKEY[pBlockData->nRow] = TSDBROW_TS(pRow);
79,446,380✔
1309

1310
  if (pRow->type == TSDBROW_ROW_FMT) {
79,446,380✔
1311
    code = tRowUpsertColData(pRow->pTSRow, pTSchema, pBlockData->aColData, pBlockData->nColData, 0 /* append */);
50,963,853✔
1312
    if (code) goto _exit;
51,079,881!
1313
  } else if (pRow->type == TSDBROW_COL_FMT) {
28,482,527!
1314
    code = tBlockDataUpsertBlockRow(pBlockData, pRow->pBlockData, pRow->iRow, 0 /* append */);
28,595,896✔
1315
    if (code) goto _exit;
28,729,393!
1316
  } else {
1317
    return TSDB_CODE_INVALID_PARA;
×
1318
  }
1319
  pBlockData->nRow++;
79,809,274✔
1320

1321
_exit:
79,809,274✔
1322
  return code;
79,809,274✔
1323
}
1324
int32_t tBlockDataUpdateRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema) {
1,187,680✔
1325
  int32_t code = 0;
1,187,680✔
1326

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

1337
  // update other rows
1338
  if (pRow->type == TSDBROW_ROW_FMT) {
1,187,680✔
1339
    code = tRowUpsertColData(pRow->pTSRow, pTSchema, pBlockData->aColData, pBlockData->nColData,
147,815!
1340
                             (rversion > lversion) ? 1 : -1 /* update */);
1341
    if (code) goto _exit;
151,526!
1342
  } else if (pRow->type == TSDBROW_COL_FMT) {
1,039,865!
1343
    code = tBlockDataUpsertBlockRow(pBlockData, pRow->pBlockData, pRow->iRow, (rversion > lversion) ? 1 : -1);
1,040,940!
1344
    if (code) goto _exit;
1,052,964!
1345
  } else {
1346
    code = TSDB_CODE_INVALID_PARA;
×
1347
    goto _exit;
×
1348
  }
1349

1350
_exit:
1,203,415✔
1351
  return code;
1,203,415✔
1352
}
1353

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

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

1374
SColData *tBlockDataGetColData(SBlockData *pBlockData, int16_t cid) {
366,296,697✔
1375
  int32_t lidx = 0;
366,296,697✔
1376
  int32_t ridx = pBlockData->nColData - 1;
366,296,697✔
1377

1378
  while (lidx <= ridx) {
682,695,522✔
1379
    int32_t   midx = (lidx + ridx) >> 1;
415,311,759✔
1380
    SColData *pColData = tBlockDataGetColDataByIdx(pBlockData, midx);
415,311,759✔
1381
    int32_t   c = (pColData->cid == cid) ? 0 : ((pColData->cid > cid) ? 1 : -1);
415,311,759✔
1382

1383
    if (c == 0) {
415,311,759✔
1384
      return pColData;
98,912,934✔
1385
    } else if (c < 0) {
316,398,825✔
1386
      lidx = midx + 1;
171,607,710✔
1387
    } else {
1388
      ridx = midx - 1;
144,791,115✔
1389
    }
1390
  }
1391

1392
  return NULL;
267,383,763✔
1393
}
1394

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

1404
  SColCompressInfo *pInfo = pCompr;
41,228✔
1405
  code = tsdbGetColCmprAlgFromSet(pInfo->pColCmpr, 1, &pInfo->defaultCmprAlg);
41,228✔
1406
  TAOS_UNUSED(code);
1407

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

1423
  tBufferClear(&buffers[1]);
41,222✔
1424
  code = tBlockDataCompressKeyPart(bData, &hdr, &buffers[1], assist, (SColCompressInfo *)pInfo);
41,222✔
1425
  TSDB_CHECK_CODE(code, lino, _exit);
41,232!
1426

1427
  // Regulart column part
1428
  tBufferClear(&buffers[2]);
41,232✔
1429
  tBufferClear(&buffers[3]);
41,232✔
1430
  for (int i = 0; i < bData->nColData; i++) {
505,553✔
1431
    SColData *colData = tBlockDataGetColDataByIdx(bData, i);
464,296✔
1432

1433
    if (colData->cflag & COL_IS_KEY) {
464,296✔
1434
      continue;
61,711✔
1435
    }
1436
    if (colData->flag == HAS_NONE) {
460,332✔
1437
      continue;
57,747✔
1438
    }
1439

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

1448
    int32_t offset = buffers[3].size;
402,513✔
1449
    code = tColDataCompress(colData, &cinfo, &buffers[3], assist);
402,513✔
1450
    TSDB_CHECK_CODE(code, lino, _exit);
402,556!
1451

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

1463
    code = tPutBlockCol(&buffers[2], &blockCol, hdr.fmtVer, hdr.cmprAlg);
402,556✔
1464
    TSDB_CHECK_CODE(code, lino, _exit);
402,610!
1465
  }
1466
  hdr.szBlkCol = buffers[2].size;
41,257✔
1467

1468
  // SDiskDataHdr part
1469
  tBufferClear(&buffers[0]);
1470
  code = tPutDiskDataHdr(&buffers[0], &hdr);
41,257✔
1471
  TSDB_CHECK_CODE(code, lino, _exit);
41,199!
1472

1473
_exit:
41,199✔
1474
  return code;
41,199✔
1475
}
1476

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

1483
  // SDiskDataHdr
1484
  code = tGetDiskDataHdr(br, &hdr);
11,961✔
1485
  TSDB_CHECK_CODE(code, lino, _exit);
11,958!
1486

1487
  tBlockDataReset(blockData);
11,958✔
1488
  blockData->suid = hdr.suid;
11,959✔
1489
  blockData->uid = hdr.uid;
11,959✔
1490
  blockData->nRow = hdr.nRow;
11,959✔
1491

1492
  // Key part
1493
  code = tBlockDataDecompressKeyPart(&hdr, br, blockData, assist);
11,959✔
1494
  TSDB_CHECK_CODE(code, lino, _exit);
11,960!
1495

1496
  // Column part
1497
  SBufferReader br2 = *br;
11,960✔
1498
  br->offset += hdr.szBlkCol;
11,960✔
1499
  for (uint32_t startOffset = br2.offset; br2.offset - startOffset < hdr.szBlkCol;) {
131,611✔
1500
    SBlockCol blockCol;
1501

1502
    code = tGetBlockCol(&br2, &blockCol, hdr.fmtVer, hdr.cmprAlg);
119,645✔
1503
    TSDB_CHECK_CODE(code, lino, _exit);
119,573!
1504
    if (blockCol.alg == 0) blockCol.alg = hdr.cmprAlg;
119,573✔
1505
    code = tBlockDataDecompressColData(&hdr, &blockCol, br, blockData, assist);
119,573✔
1506
    TSDB_CHECK_CODE(code, lino, _exit);
119,651!
1507
  }
1508

1509
_exit:
11,966✔
1510
  return code;
11,966✔
1511
}
1512

1513
// SDiskDataHdr ==============================
1514
int32_t tPutDiskDataHdr(SBuffer *buffer, const SDiskDataHdr *pHdr) {
41,224✔
1515
  int32_t code;
1516

1517
  if ((code = tBufferPutU32(buffer, pHdr->delimiter))) return code;
82,462!
1518
  if ((code = tBufferPutU32v(buffer, pHdr->fmtVer))) return code;
82,476!
1519
  if ((code = tBufferPutI64(buffer, pHdr->suid))) return code;
82,476!
1520
  if ((code = tBufferPutI64(buffer, pHdr->uid))) return code;
82,476!
1521
  if ((code = tBufferPutI32v(buffer, pHdr->szUid))) return code;
82,476!
1522
  if ((code = tBufferPutI32v(buffer, pHdr->szVer))) return code;
82,476!
1523
  if ((code = tBufferPutI32v(buffer, pHdr->szKey))) return code;
82,476!
1524
  if ((code = tBufferPutI32v(buffer, pHdr->szBlkCol))) return code;
82,476!
1525
  if ((code = tBufferPutI32v(buffer, pHdr->nRow))) return code;
82,476!
1526
  if (pHdr->fmtVer < 2) {
41,238!
1527
    if ((code = tBufferPutI8(buffer, pHdr->cmprAlg))) return code;
×
1528
  } else if (pHdr->fmtVer == 2) {
41,238✔
1529
    if ((code = tBufferPutU32(buffer, pHdr->cmprAlg))) return code;
82,402!
1530
  } else {
1531
    // more data fmt ver
1532
  }
1533
  if (pHdr->fmtVer >= 1) {
41,238✔
1534
    if ((code = tBufferPutI8(buffer, pHdr->numOfPKs))) return code;
82,330!
1535
    for (int i = 0; i < pHdr->numOfPKs; i++) {
45,129✔
1536
      if ((code = tPutBlockCol(buffer, &pHdr->primaryBlockCols[i], pHdr->fmtVer, pHdr->cmprAlg))) return code;
3,963!
1537
    }
1538
  }
1539

1540
  return 0;
41,239✔
1541
}
1542

1543
int32_t tGetDiskDataHdr(SBufferReader *br, SDiskDataHdr *pHdr) {
1,529,880✔
1544
  int32_t code;
1545

1546
  if ((code = tBufferGetU32(br, &pHdr->delimiter))) return code;
1,529,880!
1547
  if ((code = tBufferGetU32v(br, &pHdr->fmtVer))) return code;
1,530,620!
1548
  if ((code = tBufferGetI64(br, &pHdr->suid))) return code;
1,530,961!
1549
  if ((code = tBufferGetI64(br, &pHdr->uid))) return code;
1,529,960!
1550
  if ((code = tBufferGetI32v(br, &pHdr->szUid))) return code;
1,530,660!
1551
  if ((code = tBufferGetI32v(br, &pHdr->szVer))) return code;
1,531,067!
1552
  if ((code = tBufferGetI32v(br, &pHdr->szKey))) return code;
1,531,027!
1553
  if ((code = tBufferGetI32v(br, &pHdr->szBlkCol))) return code;
1,530,706!
1554
  if ((code = tBufferGetI32v(br, &pHdr->nRow))) return code;
1,531,140!
1555
  if (pHdr->fmtVer < 2) {
1,531,090!
1556
    int8_t cmprAlg = 0;
×
1557
    if ((code = tBufferGetI8(br, &cmprAlg))) return code;
×
1558
    pHdr->cmprAlg = cmprAlg;
×
1559
  } else if (pHdr->fmtVer == 2) {
1,531,090✔
1560
    if ((code = tBufferGetU32(br, &pHdr->cmprAlg))) return code;
1,530,230!
1561
  } else {
1562
    // more data fmt ver
1563
  }
1564
  if (pHdr->fmtVer >= 1) {
1,531,647✔
1565
    if ((code = tBufferGetI8(br, &pHdr->numOfPKs))) return code;
1,530,900!
1566
    for (int i = 0; i < pHdr->numOfPKs; i++) {
1,726,361✔
1567
      if ((code = tGetBlockCol(br, &pHdr->primaryBlockCols[i], pHdr->fmtVer, pHdr->cmprAlg))) {
196,014!
1568
        return code;
×
1569
      }
1570
    }
1571
  } else {
1572
    pHdr->numOfPKs = 0;
747✔
1573
  }
1574

1575
  return 0;
1,531,094✔
1576
}
1577

1578
// ALGORITHM ==============================
1579
int32_t tPutColumnDataAgg(SBuffer *buffer, SColumnDataAgg *pColAgg) {
117,075✔
1580
  int32_t code;
1581

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

1600
  return 0;
117,075✔
1601
}
1602

1603
int32_t tGetColumnDataAgg(SBufferReader *br, SColumnDataAgg *pColAgg) {
260,990✔
1604
  int32_t code;
1605

1606
  if ((code = tBufferGetI32v(br, &pColAgg->colId))) return code;
260,990!
1607
  if ((code = tBufferGetI16v(br, &pColAgg->numOfNull))) return code;
260,946!
1608
  if (pColAgg->colId & DECIMAL_AGG_FLAG) {
260,863✔
1609
    pColAgg->colId &= 0xFFFF;
40✔
1610
    if ((code = tBufferGetU64(br, &pColAgg->decimal128Sum[0]))) return code;
40!
1611
    if ((code = tBufferGetU64(br, &pColAgg->decimal128Sum[1]))) return code;
40!
1612
    if ((code = tBufferGetU64(br, &pColAgg->decimal128Max[0]))) return code;
40!
1613
    if ((code = tBufferGetU64(br, &pColAgg->decimal128Max[1]))) return code;
40!
1614
    if ((code = tBufferGetU64(br, &pColAgg->decimal128Min[0]))) return code;
40!
1615
    if ((code = tBufferGetU64(br, &pColAgg->decimal128Min[1]))) return code;
40!
1616
    if ((code = tBufferGetU8(br, &pColAgg->overflow))) return code;
40!
1617
  } else {
1618
    if ((code = tBufferGetI64(br, &pColAgg->sum))) return code;
260,823!
1619
    if ((code = tBufferGetI64(br, &pColAgg->max))) return code;
260,821!
1620
    if ((code = tBufferGetI64(br, &pColAgg->min))) return code;
260,745!
1621
  }
1622

1623
  return 0;
260,690✔
1624
}
1625

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

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

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

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

1661
  code = tCompressDataToBuffer((uint8_t *)bData->aTSKEY, &cinfo, buffer, assist);
41,229✔
1662
  TSDB_CHECK_CODE(code, lino, _exit);
41,235!
1663
  hdr->szKey = cinfo.compressedSize;
41,235✔
1664

1665
  // primary keys
1666
  for (hdr->numOfPKs = 0; hdr->numOfPKs < bData->nColData; hdr->numOfPKs++) {
45,199✔
1667
    if (!(hdr->numOfPKs <= TD_MAX_PK_COLS)) {
45,190!
1668
      return TSDB_CODE_INVALID_PARA;
×
1669
    }
1670

1671
    SBlockCol *blockCol = &hdr->primaryBlockCols[hdr->numOfPKs];
45,190✔
1672
    SColData  *colData = tBlockDataGetColDataByIdx(bData, hdr->numOfPKs);
45,190✔
1673

1674
    if ((colData->cflag & COL_IS_KEY) == 0) {
45,190✔
1675
      break;
41,226✔
1676
    }
1677

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

1687
    code = tColDataCompress(colData, &info, buffer, assist);
3,963✔
1688
    TSDB_CHECK_CODE(code, lino, _exit);
3,964!
1689

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

1704
_exit:
9✔
1705
  return code;
41,235✔
1706
}
1707

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

1713
  SColData *colData;
1714

1715
  code = tBlockDataAddColData(blockData, blockCol->cid, blockCol->type, blockCol->cflag, &colData);
4,552,862✔
1716
  TSDB_CHECK_CODE(code, lino, _exit);
4,554,587!
1717

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

1733
  switch (blockCol->flag) {
4,554,587✔
1734
    case (HAS_NONE | HAS_NULL | HAS_VALUE):
4✔
1735
      info.bitmapOriginalSize = BIT2_SIZE(hdr->nRow);
4✔
1736
      break;
4✔
1737
    case (HAS_NONE | HAS_NULL):
1,058,781✔
1738
    case (HAS_NONE | HAS_VALUE):
1739
    case (HAS_NULL | HAS_VALUE):
1740
      info.bitmapOriginalSize = BIT1_SIZE(hdr->nRow);
1,058,781✔
1741
      break;
1,058,781✔
1742
  }
1743

1744
  code = tColDataDecompress(BR_PTR(br), &info, colData, assist);
4,554,587✔
1745
  TSDB_CHECK_CODE(code, lino, _exit);
4,553,604!
1746
  br->offset += blockCol->szBitmap + blockCol->szOffset + blockCol->szValue;
4,553,604✔
1747

1748
_exit:
4,553,604✔
1749
  return code;
4,553,604✔
1750
}
1751

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

1758
  // uid
1759
  if (hdr->szUid > 0) {
1,528,794✔
1760
    cinfo = (SCompressInfo){
1,132,351✔
1761
        .cmprAlg = hdr->cmprAlg,
1,132,351✔
1762
        .dataType = TSDB_DATA_TYPE_BIGINT,
1763
        .compressedSize = hdr->szUid,
1,132,351✔
1764
        .originalSize = sizeof(int64_t) * hdr->nRow,
1,132,351✔
1765
    };
1766

1767
    code = tRealloc((uint8_t **)&blockData->aUid, cinfo.originalSize);
1,132,351✔
1768
    TSDB_CHECK_CODE(code, lino, _exit);
1,133,679!
1769
    code = tDecompressData(BR_PTR(br), &cinfo, blockData->aUid, cinfo.originalSize, assist);
1,133,679✔
1770
    TSDB_CHECK_CODE(code, lino, _exit);
1,133,021!
1771
    br->offset += cinfo.compressedSize;
1,133,021✔
1772
  }
1773

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

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

1800
  // primary keys
1801
  for (int i = 0; i < hdr->numOfPKs; i++) {
1,728,120✔
1802
    const SBlockCol *blockCol = &hdr->primaryBlockCols[i];
196,401✔
1803

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

1811
    code = tBlockDataDecompressColData(hdr, blockCol, br, blockData, assist);
196,401✔
1812
    TSDB_CHECK_CODE(code, lino, _exit);
196,274!
1813
  }
1814

1815
_exit:
1,531,719✔
1816
  return code;
1,531,719✔
1817
}
1818

1819
int32_t tsdbGetColCmprAlgFromSet(SHashObj *set, int16_t colId, uint32_t *alg) {
447,772✔
1820
  if (set == NULL) return -1;
447,772✔
1821

1822
  uint32_t *ret = taosHashGet(set, &colId, sizeof(colId));
447,566✔
1823
  if (ret == NULL) {
447,439!
1824
    return TSDB_CODE_NOT_FOUND;
×
1825
  }
1826

1827
  *alg = *ret;
447,439✔
1828
  return 0;
447,439✔
1829
}
1830
uint32_t tsdbCvtTimestampAlg(uint32_t alg) {
×
1831
  DEFINE_VAR(alg)
×
1832

1833
  return 0;
×
1834
}
1835

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

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

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

1854
  if (diskId) {
20,780✔
1855
    *diskId = did;
20,779✔
1856
  }
1857
  return code;
20,780✔
1858
}
1859

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

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

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

1875
  if (diskId) {
36!
1876
    *diskId = did;
36✔
1877
  }
1878
  return 0;
36✔
1879
}
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

© 2026 Coveralls, Inc