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

taosdata / TDengine / #5091

17 May 2026 01:15AM UTC coverage: 73.326% (-0.01%) from 73.34%
#5091

push

travis-ci

web-flow
feat (TDgpt): Dynamic Model Synchronization Enhancements (#35344)

* refactor: do some internal refactor.

* fix: fix multiprocess sync issue.

* feat: add dynamic anomaly detection and forecasting services

* fix: log error message for undeploying model in exception handling

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* fix: handle undeploy when model exists only on disk

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/286aafa0-c3ce-4c27-b803-2707571e9dc1

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: guard dynamic registry concurrent access

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/5e4db858-6458-40f4-ac28-d1b1b7f97c18

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: tighten service list locking scope

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/5e4db858-6458-40f4-ac28-d1b1b7f97c18

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: restore prophet support and update tests per review feedback

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/92298ae1-7da6-4d07-b20e-101c7cd0b26b

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: improve test name and move copy inside lock scope

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/92298ae1-7da6-4d07-b20e-101c7cd0b26b

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* Potential fix for pull request finding

Co-au... (continued)

281422 of 383795 relevant lines covered (73.33%)

136839186.86 hits per line

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

89.21
/source/dnode/vnode/src/tsdb/tsdbSttFileRW.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 "tsdbSttFileRW.h"
17
#include "meta.h"
18
#include "tsdbDataFileRW.h"
19

20
// SSttFReader ============================================================
21
struct SSttFileReader {
22
  SSttFileReaderConfig config[1];
23
  STsdbFD             *fd;
24
  SSttFooter           footer[1];
25
  struct {
26
    bool sttBlkLoaded;
27
    bool statisBlkLoaded;
28
    bool tombBlkLoaded;
29
  } ctx[1];
30
  TSttBlkArray    sttBlkArray[1];
31
  TStatisBlkArray statisBlkArray[1];
32
  TTombBlkArray   tombBlkArray[1];
33
  SBuffer         local[10];
34
  SBuffer        *buffers;
35
};
36

37
// SSttFileReader
38
int32_t tsdbSttFileReaderOpen(const char *fname, const SSttFileReaderConfig *config, SSttFileReader **reader) {
465,810,465✔
39
  int32_t code = 0;
465,810,465✔
40
  int32_t lino = 0;
465,810,465✔
41

42
  reader[0] = taosMemoryCalloc(1, sizeof(*reader[0]));
465,832,742✔
43
  if (reader[0] == NULL) {
465,548,077✔
44
    TAOS_CHECK_GOTO(terrno, &lino, _exit);
×
45
  }
46

47
  reader[0]->config[0] = config[0];
465,543,162✔
48
  reader[0]->buffers = config->buffers;
465,590,926✔
49
  if (reader[0]->buffers == NULL) {
465,800,398✔
50
    reader[0]->buffers = reader[0]->local;
465,829,298✔
51
  }
52

53
  // open file
54
  if (fname) {
465,808,208✔
55
    TAOS_CHECK_GOTO(tsdbOpenFile(fname, config->tsdb, TD_FILE_READ, &reader[0]->fd, 0), &lino, _exit);
465,808,208✔
56
  } else {
57
    char fname1[TSDB_FILENAME_LEN];
×
58
    tsdbTFileName(config->tsdb, config->file, fname1);
×
59
    TAOS_CHECK_GOTO(tsdbOpenFile(fname1, config->tsdb, TD_FILE_READ, &reader[0]->fd, 0), &lino, _exit);
×
60
  }
61

62
  // // open each segment reader
63
  int64_t offset = config->file->size - sizeof(SSttFooter);
465,815,698✔
64
  if (offset < TSDB_FHDR_SIZE) {
465,725,048✔
65
    tsdbError("vgId:%d %s failed at %s:%d since file size is too small: %" PRId64 " fname:%s",
×
66
              TD_VID(config->tsdb->pVnode), __func__, __FILE__, __LINE__, config->file->size, reader[0]->fd->path);
67
    TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
×
68
  }
69

70
  SEncryptData *pEncryptData = &(config->tsdb->pVnode->config.tsdbCfg.encryptData);
465,725,048✔
71

72
#if 1
73
  TAOS_CHECK_GOTO(
465,818,081✔
74
      tsdbReadFile(reader[0]->fd, offset, (uint8_t *)(reader[0]->footer), sizeof(SSttFooter), 0, pEncryptData), &lino,
75
      _exit);
76
#else
77
  int64_t size = config->file->size;
78

79
  for (; size > TSDB_FHDR_SIZE; size--) {
80
    code = tsdbReadFile(reader[0]->fd, size - sizeof(SSttFooter), (uint8_t *)(reader[0]->footer), sizeof(SSttFooter), 0,
81
                        encryptAlgoirthm, encryptKey);
82
    if (code) continue;
83
    if ((*reader)->footer->sttBlkPtr->offset + (*reader)->footer->sttBlkPtr->size + sizeof(SSttFooter) == size ||
84
        (*reader)->footer->statisBlkPtr->offset + (*reader)->footer->statisBlkPtr->size + sizeof(SSttFooter) == size ||
85
        (*reader)->footer->tombBlkPtr->offset + (*reader)->footer->tombBlkPtr->size + sizeof(SSttFooter) == size) {
86
      break;
87
    }
88
  }
89
  if (size <= TSDB_FHDR_SIZE) {
90
    TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
91
  }
92
#endif
93

94
_exit:
465,641,646✔
95
  if (code) {
465,620,675✔
96
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(config->tsdb->pVnode), __func__, __FILE__, lino,
×
97
              tstrerror(code));
98
    tsdbSttFileReaderClose(reader);
×
99
  }
100
  return code;
465,620,675✔
101
}
102

103
void tsdbSttFileReaderClose(SSttFileReader **reader) {
465,815,529✔
104
  if (reader[0]) {
465,815,529✔
105
    for (int32_t i = 0; i < ARRAY_SIZE(reader[0]->local); ++i) {
2,147,483,647✔
106
      tBufferDestroy(reader[0]->local + i);
2,147,483,647✔
107
    }
108
    tsdbCloseFile(&reader[0]->fd);
465,785,813✔
109
    TARRAY2_DESTROY(reader[0]->tombBlkArray, NULL);
465,764,545✔
110
    TARRAY2_DESTROY(reader[0]->statisBlkArray, NULL);
465,805,443✔
111
    TARRAY2_DESTROY(reader[0]->sttBlkArray, NULL);
465,740,791✔
112
    taosMemoryFree(reader[0]);
465,769,466✔
113
    reader[0] = NULL;
465,796,310✔
114
  }
115
}
465,739,784✔
116

117
// SSttFSegReader
118
int32_t tsdbSttFileReadStatisBlk(SSttFileReader *reader, const TStatisBlkArray **statisBlkArray) {
462,567,918✔
119
  if (!reader->ctx->statisBlkLoaded) {
462,567,918✔
120
    if (reader->footer->statisBlkPtr->size > 0) {
462,615,102✔
121
      if (reader->footer->statisBlkPtr->size % sizeof(SStatisBlk) != 0) {
462,453,701✔
122
        tsdbError("vgId:%d %s failed at %s:%d since stt file statis block size is not valid, fname:%s",
×
123
                  TD_VID(reader->config->tsdb->pVnode), __func__, __FILE__, __LINE__, reader->fd->path);
124
        return TSDB_CODE_FILE_CORRUPTED;
×
125
      }
126

127
      int32_t size = reader->footer->statisBlkPtr->size / sizeof(SStatisBlk);
462,349,852✔
128
      void   *data = taosMemoryMalloc(reader->footer->statisBlkPtr->size);
462,450,672✔
129
      if (!data) {
462,385,401✔
130
        return terrno;
×
131
      }
132

133
      SEncryptData *pEncryptData = &(reader->config->tsdb->pVnode->config.tsdbCfg.encryptData);
462,385,401✔
134

135
      int32_t code = tsdbReadFile(reader->fd, reader->footer->statisBlkPtr->offset, data,
462,485,845✔
136
                                  reader->footer->statisBlkPtr->size, 0, pEncryptData);
137
      if (code) {
462,509,867✔
138
        taosMemoryFree(data);
×
139
        return code;
×
140
      }
141

142
      TARRAY2_INIT_EX(reader->statisBlkArray, size, size, data);
462,509,867✔
143
    } else {
144
      TARRAY2_INIT(reader->statisBlkArray);
162,255✔
145
    }
146

147
    reader->ctx->statisBlkLoaded = true;
462,705,183✔
148
  }
149

150
  statisBlkArray[0] = reader->statisBlkArray;
462,690,411✔
151
  return 0;
462,682,897✔
152
}
153

154
int32_t tsdbSttFileReadTombBlk(SSttFileReader *reader, const TTombBlkArray **tombBlkArray) {
465,697,878✔
155
  if (!reader->ctx->tombBlkLoaded) {
465,697,878✔
156
    if (reader->footer->tombBlkPtr->size > 0) {
465,795,013✔
157
      if (reader->footer->tombBlkPtr->size % sizeof(STombBlk) != 0) {
13,011,123✔
158
        tsdbError("vgId:%d %s failed at %s:%d since stt file tomb block size is not valid, fname:%s",
×
159
                  TD_VID(reader->config->tsdb->pVnode), __func__, __FILE__, __LINE__, reader->fd->path);
160
        return TSDB_CODE_FILE_CORRUPTED;
×
161
      }
162

163
      int32_t size = reader->footer->tombBlkPtr->size / sizeof(STombBlk);
13,011,123✔
164
      void   *data = taosMemoryMalloc(reader->footer->tombBlkPtr->size);
13,011,123✔
165
      if (!data) {
13,011,123✔
166
        return terrno;
×
167
      }
168

169
      SEncryptData *pEncryptData = &(reader->config->tsdb->pVnode->config.tsdbCfg.encryptData);
13,011,123✔
170

171
      int32_t code = tsdbReadFile(reader->fd, reader->footer->tombBlkPtr->offset, data,
13,011,123✔
172
                                  reader->footer->tombBlkPtr->size, 0, pEncryptData);
173
      if (code) {
13,011,123✔
174
        taosMemoryFree(data);
×
175
        return code;
×
176
      }
177

178
      TARRAY2_INIT_EX(reader->tombBlkArray, size, size, data);
13,011,123✔
179
    } else {
180
      TARRAY2_INIT(reader->tombBlkArray);
452,708,113✔
181
    }
182

183
    reader->ctx->tombBlkLoaded = true;
465,781,781✔
184
  }
185

186
  tombBlkArray[0] = reader->tombBlkArray;
465,763,699✔
187
  return 0;
465,742,628✔
188
}
189

190
int32_t tsdbSttFileReadSttBlk(SSttFileReader *reader, const TSttBlkArray **sttBlkArray) {
465,629,134✔
191
  if (!reader->ctx->sttBlkLoaded) {
465,629,134✔
192
    if (reader->footer->sttBlkPtr->size > 0) {
465,743,562✔
193
      if (reader->footer->sttBlkPtr->size % sizeof(SSttBlk) != 0) {
465,563,305✔
194
        tsdbError("vgId:%d %s failed at %s:%d since stt file stt block size is not valid, fname:%s",
×
195
                  TD_VID(reader->config->tsdb->pVnode), __func__, __FILE__, __LINE__, reader->fd->path);
196
        return TSDB_CODE_FILE_CORRUPTED;
×
197
      }
198

199
      int32_t size = reader->footer->sttBlkPtr->size / sizeof(SSttBlk);
465,360,335✔
200
      void   *data = taosMemoryMalloc(reader->footer->sttBlkPtr->size);
465,491,580✔
201
      if (!data) {
465,456,405✔
202
        return terrno;
×
203
      }
204

205
      SEncryptData *pEncryptData = &(reader->config->tsdb->pVnode->config.tsdbCfg.encryptData);
465,456,405✔
206

207
      int32_t code = tsdbReadFile(reader->fd, reader->footer->sttBlkPtr->offset, data, reader->footer->sttBlkPtr->size,
465,582,377✔
208
                                  0, pEncryptData);
209
      if (code) {
465,593,414✔
210
        taosMemoryFree(data);
×
211
        return code;
×
212
      }
213

214
      TARRAY2_INIT_EX(reader->sttBlkArray, size, size, data);
465,593,414✔
215
    } else {
216
      TARRAY2_INIT(reader->sttBlkArray);
215,720✔
217
    }
218

219
    reader->ctx->sttBlkLoaded = true;
465,795,991✔
220
  }
221

222
  sttBlkArray[0] = reader->sttBlkArray;
465,821,416✔
223
  return 0;
465,824,901✔
224
}
225

226
int32_t tsdbSttFileReadBlockData(SSttFileReader *reader, const SSttBlk *sttBlk, SBlockData *bData) {
18,254,377✔
227
  int32_t code = 0;
18,254,377✔
228
  int32_t lino = 0;
18,254,377✔
229

230
  SBuffer *buffer0 = reader->buffers + 0;
18,254,377✔
231
  SBuffer *assist = reader->buffers + 1;
18,254,377✔
232

233
  SEncryptData *pEncryptData = &(reader->config->tsdb->pVnode->config.tsdbCfg.encryptData);
18,255,827✔
234

235
  // load data
236
  tBufferClear(buffer0);
237
  TAOS_CHECK_GOTO(
18,255,102✔
238
      tsdbReadFileToBuffer(reader->fd, sttBlk->bInfo.offset, sttBlk->bInfo.szBlock, buffer0, 0, pEncryptData), &lino,
239
      _exit);
240

241
  SBufferReader br = BUFFER_READER_INITIALIZER(0, buffer0);
18,255,827✔
242
  TAOS_CHECK_GOTO(tBlockDataDecompress(&br, bData, assist), &lino, _exit);
18,255,827✔
243

244
_exit:
18,255,224✔
245
  if (code) {
18,255,224✔
246
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(reader->config->tsdb->pVnode), __func__, __FILE__, lino,
×
247
              tstrerror(code));
248
  }
249
  return code;
18,255,224✔
250
}
251

252
int32_t tsdbSttFileReadBlockDataByColumn(SSttFileReader *reader, const SSttBlk *sttBlk, SBlockData *bData,
257,524,312✔
253
                                         STSchema *pTSchema, int16_t cids[], int32_t ncid) {
254
  int32_t code = 0;
257,524,312✔
255
  int32_t lino = 0;
257,524,312✔
256

257
  SDiskDataHdr hdr;
257,538,349✔
258
  SBuffer     *buffer0 = reader->buffers + 0;
257,534,822✔
259
  SBuffer     *buffer1 = reader->buffers + 1;
257,506,342✔
260
  SBuffer     *assist = reader->buffers + 2;
257,524,098✔
261

262
  SEncryptData *pEncryptData = &(reader->config->tsdb->pVnode->config.tsdbCfg.encryptData);
257,573,074✔
263

264
  // load key part
265
  tBufferClear(buffer0);
266
  TAOS_CHECK_GOTO(tsdbReadFileToBuffer(reader->fd, sttBlk->bInfo.offset, sttBlk->bInfo.szKey, buffer0, 0, pEncryptData),
257,539,395✔
267
                  &lino, _exit);
268

269
  // decode header
270
  SBufferReader br = BUFFER_READER_INITIALIZER(0, buffer0);
257,568,976✔
271
  TAOS_CHECK_GOTO(tGetDiskDataHdr(&br, &hdr), &lino, _exit);
257,545,653✔
272

273
  if (hdr.delimiter != TSDB_FILE_DLMT) {
257,474,078✔
274
    tsdbError("vgId:%d %s failed at %s:%d since disk data header delimiter is invalid, fname:%s",
×
275
              TD_VID(reader->config->tsdb->pVnode), __func__, __FILE__, __LINE__, reader->fd->path);
276
    TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
×
277
  }
278

279
  // set data container
280
  tBlockDataReset(bData);
257,474,078✔
281
  bData->suid = hdr.suid;
257,519,127✔
282
  bData->uid = (sttBlk->suid == 0) ? sttBlk->minUid : 0;
257,516,413✔
283
  bData->nRow = hdr.nRow;
257,501,873✔
284

285
  // key part
286
  TAOS_CHECK_GOTO(tBlockDataDecompressKeyPart(&hdr, &br, bData, assist), &lino, _exit);
257,506,192✔
287
  if (br.offset != buffer0->size) {
257,550,891✔
288
    tsdbError("vgId:%d %s failed at %s:%d since key part size mismatch, expected:%u, actual:%u, fname:%s",
×
289
              TD_VID(reader->config->tsdb->pVnode), __func__, __FILE__, __LINE__, buffer0->size, br.offset,
290
              reader->fd->path);
291
    TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
×
292
  }
293

294
  bool loadExtra = false;
257,543,134✔
295
  for (int i = 0; i < ncid; i++) {
306,523,582✔
296
    if (tBlockDataGetColData(bData, cids[i]) == NULL) {
251,712,391✔
297
      loadExtra = true;
202,735,667✔
298
      break;
202,735,667✔
299
    }
300
  }
301

302
  if (!loadExtra) {
257,546,858✔
303
    goto _exit;
54,811,740✔
304
  }
305

306
  // load SBlockCol part
307
  tBufferClear(buffer0);
308
  TAOS_CHECK_GOTO(tsdbReadFileToBuffer(reader->fd, sttBlk->bInfo.offset + sttBlk->bInfo.szKey, hdr.szBlkCol, buffer0, 0,
202,738,940✔
309
                                       pEncryptData),
310
                  &lino, _exit);
311

312
  // load each column
313
  SBlockCol blockCol = {
202,754,309✔
314
      .cid = 0,
315
  };
316
  br = BUFFER_READER_INITIALIZER(0, buffer0);
202,758,921✔
317
  for (int32_t i = 0; i < ncid; i++) {
886,268,987✔
318
    int16_t cid = cids[i];
683,482,510✔
319

320
    if (tBlockDataGetColData(bData, cid)) {  // already loaded
683,492,804✔
321
      continue;
26,683,655✔
322
    }
323

324
    while (cid > blockCol.cid) {
2,147,483,647✔
325
      if (br.offset >= buffer0->size) {
1,525,799,920✔
326
        blockCol.cid = INT16_MAX;
14,640,309✔
327
        break;
14,640,309✔
328
      }
329

330
      TAOS_CHECK_GOTO(tGetBlockCol(&br, &blockCol, hdr.fmtVer, hdr.cmprAlg), &lino, _exit);
1,511,152,691✔
331
    }
332

333
    if (cid < blockCol.cid) {
656,886,244✔
334
      const STColumn *tcol = tTSchemaSearchColumn(pTSchema, cid);
73,287,832✔
335
      TSDB_CHECK_NULL(tcol, code, lino, _exit, TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER);
73,285,108✔
336
      SBlockCol none = {
73,285,108✔
337
          .cid = cid,
338
          .type = tcol->type,
73,284,427✔
339
          .cflag = tcol->flags,
73,286,470✔
340
          .flag = HAS_NONE,
341
          .szOrigin = 0,
342
          .szBitmap = 0,
343
          .szOffset = 0,
344
          .szValue = 0,
345
          .offset = 0,
346
      };
347
      TAOS_CHECK_GOTO(tBlockDataDecompressColData(&hdr, &none, &br, bData, assist), &lino, _exit);
73,272,850✔
348
    } else if (cid == blockCol.cid) {
583,598,412✔
349
      // load from file
350
      tBufferClear(buffer1);
351
      TAOS_CHECK_GOTO(
583,561,621✔
352
          tsdbReadFileToBuffer(reader->fd, sttBlk->bInfo.offset + sttBlk->bInfo.szKey + hdr.szBlkCol + blockCol.offset,
353
                               blockCol.szBitmap + blockCol.szOffset + blockCol.szValue, buffer1, 0, pEncryptData),
354
          &lino, _exit);
355

356
      // decode the buffer
357
      SBufferReader br1 = BUFFER_READER_INITIALIZER(0, buffer1);
583,570,060✔
358
      TAOS_CHECK_GOTO(tBlockDataDecompressColData(&hdr, &blockCol, &br1, bData, assist), &lino, _exit);
583,571,394✔
359
    }
360
  }
361

362
_exit:
257,657,010✔
363
  if (code) {
257,558,164✔
364
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(reader->config->tsdb->pVnode), __func__, __FILE__, lino,
×
365
              tstrerror(code));
366
  }
367
  return code;
257,558,110✔
368
}
369

370
int32_t tsdbSttFileReadTombBlock(SSttFileReader *reader, const STombBlk *tombBlk, STombBlock *tombBlock) {
6,625,877✔
371
  int32_t code = 0;
6,625,877✔
372
  int32_t lino = 0;
6,625,877✔
373

374
  SBuffer *buffer0 = reader->buffers + 0;
6,625,877✔
375
  SBuffer *assist = reader->buffers + 1;
6,625,877✔
376

377
  SEncryptData *pEncryptData = &(reader->config->tsdb->pVnode->config.tsdbCfg.encryptData);
6,625,877✔
378

379
  // load
380
  tBufferClear(buffer0);
381
  TAOS_CHECK_GOTO(tsdbReadFileToBuffer(reader->fd, tombBlk->dp->offset, tombBlk->dp->size, buffer0, 0, pEncryptData),
6,625,877✔
382
                  &lino, _exit);
383

384
  // decode
385
  int32_t       size = 0;
6,625,877✔
386
  SBufferReader br = BUFFER_READER_INITIALIZER(0, buffer0);
6,625,877✔
387
  tTombBlockClear(tombBlock);
6,625,877✔
388
  tombBlock->numOfRecords = tombBlk->numRec;
6,625,259✔
389
  for (int32_t i = 0; i < ARRAY_SIZE(tombBlock->buffers); ++i) {
39,754,644✔
390
    SCompressInfo cinfo = {
33,128,767✔
391
        .cmprAlg = tombBlk->cmprAlg,
33,128,767✔
392
        .dataType = TSDB_DATA_TYPE_BIGINT,
393
        .originalSize = tombBlk->numRec * sizeof(int64_t),
33,128,767✔
394
        .compressedSize = tombBlk->size[i],
33,128,767✔
395
    };
396
    TAOS_CHECK_GOTO(tDecompressDataToBuffer(BR_PTR(&br), &cinfo, tombBlock->buffers + i, assist), &lino, _exit);
33,129,385✔
397
    br.offset += tombBlk->size[i];
33,128,767✔
398
  }
399

400
  if (br.offset != tombBlk->dp->size) {
6,625,877✔
401
    tsdbError("vgId:%d %s failed at %s:%d since tomb block size mismatch, fname:%s",
×
402
              TD_VID(reader->config->tsdb->pVnode), __func__, __FILE__, __LINE__, reader->fd->path);
403
    TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
×
404
  }
405

406
_exit:
6,625,877✔
407
  if (code) {
6,625,877✔
408
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(reader->config->tsdb->pVnode), __func__, __FILE__, lino,
×
409
              tstrerror(code));
410
  }
411
  return code;
6,625,877✔
412
}
413

414
int32_t tsdbSttFileReadStatisBlock(SSttFileReader *reader, const SStatisBlk *statisBlk, STbStatisBlock *statisBlock) {
284,097,418✔
415
  int32_t code = 0;
284,097,418✔
416
  int32_t lino = 0;
284,097,418✔
417

418
  if (statisBlk->numOfPKs > TD_MAX_PK_COLS) {
284,107,572✔
419
    tsdbError("vgId:%d %s failed at %s:%d since statis block numOfPKs %d exceeds max %d (file corrupted)",
×
420
              TD_VID(reader->config->tsdb->pVnode), __func__, __FILE__, __LINE__, (int)statisBlk->numOfPKs,
421
              (int)TD_MAX_PK_COLS);
422
    TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
×
423
  }
424

425
  SBuffer *buffer0 = reader->buffers + 0;
284,046,553✔
426
  SBuffer *assist = reader->buffers + 1;
284,118,985✔
427

428
  SEncryptData *pEncryptData = &(reader->config->tsdb->pVnode->config.tsdbCfg.encryptData);
284,130,204✔
429

430
  // load data
431
  tBufferClear(buffer0);
432
  TAOS_CHECK_GOTO(
284,116,053✔
433
      tsdbReadFileToBuffer(reader->fd, statisBlk->dp->offset, statisBlk->dp->size, buffer0, 0, pEncryptData), &lino,
434
      _exit);
435

436
  // decode data
437
  tStatisBlockClear(statisBlock);
284,136,723✔
438
  statisBlock->numOfPKs = statisBlk->numOfPKs;
284,102,310✔
439
  statisBlock->numOfRecords = statisBlk->numRec;
284,110,522✔
440
  SBufferReader br = BUFFER_READER_INITIALIZER(0, buffer0);
284,114,445✔
441
  for (int32_t i = 0; i < ARRAY_SIZE(statisBlock->buffers); ++i) {
1,704,822,606✔
442
    SCompressInfo info = {
1,420,665,715✔
443
        .dataType = TSDB_DATA_TYPE_BIGINT,
444
        .cmprAlg = statisBlk->cmprAlg,
1,420,666,976✔
445
        .compressedSize = statisBlk->size[i],
1,420,689,732✔
446
        .originalSize = statisBlk->numRec * sizeof(int64_t),
1,420,664,604✔
447
    };
448

449
    TAOS_CHECK_GOTO(tDecompressDataToBuffer(BR_PTR(&br), &info, &statisBlock->buffers[i], assist), &lino, _exit);
1,420,696,845✔
450
    br.offset += statisBlk->size[i];
1,420,681,114✔
451
  }
452

453
  if (statisBlk->numOfPKs > 0) {
284,156,891✔
454
    SValueColumnCompressInfo firstKeyInfos[TD_MAX_PK_COLS];
67,245,808✔
455
    SValueColumnCompressInfo lastKeyInfos[TD_MAX_PK_COLS];
67,245,808✔
456

457
    // decode compress info
458
    for (int32_t i = 0; i < statisBlk->numOfPKs; i++) {
134,489,573✔
459
      TAOS_CHECK_GOTO(tValueColumnCompressInfoDecode(&br, &firstKeyInfos[i]), &lino, _exit);
67,244,446✔
460
    }
461

462
    for (int32_t i = 0; i < statisBlk->numOfPKs; i++) {
134,489,573✔
463
      TAOS_CHECK_GOTO(tValueColumnCompressInfoDecode(&br, &lastKeyInfos[i]), &lino, _exit);
67,238,998✔
464
    }
465

466
    // decode value columns
467
    for (int32_t i = 0; i < statisBlk->numOfPKs; i++) {
134,483,444✔
468
      TAOS_CHECK_GOTO(tValueColumnDecompress(BR_PTR(&br), firstKeyInfos + i, &statisBlock->firstKeyPKs[i], assist),
67,244,446✔
469
                      &lino, _exit);
470
      br.offset += (firstKeyInfos[i].dataCompressedSize + firstKeyInfos[i].offsetCompressedSize);
67,242,361✔
471
    }
472

473
    for (int32_t i = 0; i < statisBlk->numOfPKs; i++) {
134,480,594✔
474
      TAOS_CHECK_GOTO(tValueColumnDecompress(BR_PTR(&br), &lastKeyInfos[i], &statisBlock->lastKeyPKs[i], assist), &lino,
67,241,722✔
475
                      _exit);
476
      br.offset += (lastKeyInfos[i].dataCompressedSize + lastKeyInfos[i].offsetCompressedSize);
67,245,808✔
477
    }
478
  }
479

480
  if (br.offset != buffer0->size) {
284,142,052✔
481
    tsdbError("vgId:%d %s failed at %s:%d since statis block size mismatch, expected: %u, actual: %u, fname:%s",
×
482
              TD_VID(reader->config->tsdb->pVnode), __func__, __FILE__, __LINE__, buffer0->size, br.offset,
483
              reader->fd->path);
484
    TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
×
485
  }
486

487
_exit:
284,148,365✔
488
  if (code) {
284,142,291✔
489
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(reader->config->tsdb->pVnode), __func__, __FILE__, lino,
×
490
              tstrerror(code));
491
  }
492
  return code;
284,142,291✔
493
}
494

495
// SSttFWriter ============================================================
496
struct SSttFileWriter {
497
  SSttFileWriterConfig config[1];
498
  struct {
499
    bool    opened;
500
    TABLEID tbid[1];
501
    // range
502
    SVersionRange range;
503
  } ctx[1];
504
  // file
505
  STsdbFD *fd;
506
  STFile   file[1];
507
  // data
508
  SSttFooter      footer[1];
509
  TTombBlkArray   tombBlkArray[1];
510
  TSttBlkArray    sttBlkArray[1];
511
  TStatisBlkArray statisBlkArray[1];
512
  STombBlock      tombBlock[1];
513
  STbStatisBlock  staticBlock[1];
514
  SBlockData      blockData[1];
515
  // helper data
516
  SSkmInfo skmTb[1];
517
  SSkmInfo skmRow[1];
518
  SBuffer  local[10];
519
  SBuffer *buffers;
520
  // SColCompressInfo2 pInfo;
521
};
522

523
static int32_t tsdbFileDoWriteSttBlockData(STsdbFD *fd, SBlockData *blockData, SColCompressInfo *info,
48,290,890✔
524
                                           int64_t *fileSize, TSttBlkArray *sttBlkArray, SBuffer *buffers,
525
                                           SVersionRange *range, SEncryptData *encryptData) {
526
  if (blockData->nRow == 0) return 0;
48,290,890✔
527

528
  int32_t code = 0;
48,307,312✔
529

530
  SSttBlk sttBlk[1] = {{
48,310,362✔
531
      .suid = blockData->suid,
48,293,982✔
532
      .minUid = blockData->uid ? blockData->uid : blockData->aUid[0],
48,297,612✔
533
      .maxUid = blockData->uid ? blockData->uid : blockData->aUid[blockData->nRow - 1],
48,320,511✔
534
      .minKey = blockData->aTSKEY[0],
48,314,834✔
535
      .maxKey = blockData->aTSKEY[0],
48,305,403✔
536
      .minVer = blockData->aVersion[0],
48,317,145✔
537
      .maxVer = blockData->aVersion[0],
48,318,564✔
538
      .nRow = blockData->nRow,
48,294,827✔
539
  }};
540

541
  for (int32_t iRow = 1; iRow < blockData->nRow; iRow++) {
2,147,483,647✔
542
    if (sttBlk->minKey > blockData->aTSKEY[iRow]) sttBlk->minKey = blockData->aTSKEY[iRow];
2,147,483,647✔
543
    if (sttBlk->maxKey < blockData->aTSKEY[iRow]) sttBlk->maxKey = blockData->aTSKEY[iRow];
2,147,483,647✔
544
    if (sttBlk->minVer > blockData->aVersion[iRow]) sttBlk->minVer = blockData->aVersion[iRow];
2,147,483,647✔
545
    if (sttBlk->maxVer < blockData->aVersion[iRow]) sttBlk->maxVer = blockData->aVersion[iRow];
2,147,483,647✔
546
  }
547

548
  tsdbWriterUpdVerRange(range, sttBlk->minVer, sttBlk->maxVer);
48,322,792✔
549
  TAOS_CHECK_RETURN(tBlockDataCompress(blockData, info, buffers, buffers + 4));
48,312,988✔
550

551
  sttBlk->bInfo.offset = *fileSize;
48,274,165✔
552
  sttBlk->bInfo.szKey = buffers[0].size + buffers[1].size;
48,274,306✔
553
  sttBlk->bInfo.szBlock = buffers[2].size + buffers[3].size + sttBlk->bInfo.szKey;
48,314,917✔
554
  for (int i = 0; i < 4; i++) {
241,556,331✔
555
    if (buffers[i].size) {
193,235,218✔
556
      TAOS_CHECK_RETURN(tsdbWriteFile(fd, *fileSize, buffers[i].data, buffers[i].size, encryptData));
193,018,042✔
557
      *fileSize += buffers[i].size;
193,026,716✔
558
    }
559
  }
560

561
  TAOS_CHECK_RETURN(TARRAY2_APPEND_PTR(sttBlkArray, sttBlk));
96,637,075✔
562

563
  tBlockDataClear(blockData);
48,315,962✔
564
  return 0;
48,323,238✔
565
}
566

567
static int32_t tsdbSttFileDoWriteBlockData(SSttFileWriter *writer) {
70,251,399✔
568
  if (writer->blockData->nRow == 0) return 0;
70,251,399✔
569

570
  int32_t code = 0;
48,309,875✔
571
  int32_t lino = 0;
48,309,875✔
572

573
  tb_uid_t         uid = writer->blockData->suid == 0 ? writer->blockData->uid : writer->blockData->suid;
48,316,462✔
574
  SColCompressInfo info = {.defaultCmprAlg = writer->config->cmprAlg, .pColCmpr = NULL};
48,321,103✔
575
  code = metaGetColCmpr(writer->config->tsdb->pVnode->pMeta, uid, &(info.pColCmpr));
48,319,400✔
576

577
  SEncryptData *pEncryptData = &(writer->config->tsdb->pVnode->config.tsdbCfg.encryptData);
48,308,689✔
578

579
  TAOS_CHECK_GOTO(tsdbFileDoWriteSttBlockData(writer->fd, writer->blockData, &info, &writer->file->size,
48,318,968✔
580
                                              writer->sttBlkArray, writer->buffers, &writer->ctx->range, pEncryptData),
581
                  &lino, _exit);
582

583
_exit:
48,291,664✔
584
  if (code) {
48,292,732✔
585
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
586
              tstrerror(code));
587
  }
588
  taosHashCleanup(info.pColCmpr);
48,292,732✔
589
  return code;
48,282,237✔
590
}
591

592
static int32_t tsdbSttFileDoWriteStatisBlock(SSttFileWriter *writer) {
22,323,127✔
593
  if (writer->staticBlock->numOfRecords == 0) {
22,323,127✔
594
    return 0;
1,498,970✔
595
  }
596

597
  int32_t code = 0;
20,826,886✔
598
  int32_t lino = 0;
20,826,886✔
599

600
  SBuffer *buffer0 = writer->buffers + 0;
20,825,121✔
601
  SBuffer *buffer1 = writer->buffers + 1;
20,827,611✔
602
  SBuffer *assist = writer->buffers + 2;
20,827,839✔
603

604
  STbStatisRecord record;
20,825,315✔
605
  STbStatisBlock *statisBlock = writer->staticBlock;
20,826,981✔
606
  SStatisBlk      statisBlk = {0};
20,827,372✔
607

608
  statisBlk.dp->offset = writer->file->size;
20,827,839✔
609
  statisBlk.dp->size = 0;
20,827,182✔
610
  statisBlk.numRec = statisBlock->numOfRecords;
20,827,182✔
611
  statisBlk.cmprAlg = writer->config->cmprAlg;
20,824,359✔
612
  statisBlk.numOfPKs = statisBlock->numOfPKs;
20,827,180✔
613

614
  code = tStatisBlockGet(statisBlock, 0, &record);
20,826,410✔
615
  TSDB_CHECK_CODE(code, lino, _exit);
20,828,284✔
616
  statisBlk.minTbid.suid = record.suid;
20,828,284✔
617
  statisBlk.minTbid.uid = record.uid;
20,828,284✔
618

619
  code = tStatisBlockGet(statisBlock, statisBlock->numOfRecords - 1, &record);
20,828,284✔
620
  TSDB_CHECK_CODE(code, lino, _exit);
20,827,866✔
621
  statisBlk.maxTbid.suid = record.suid;
20,827,866✔
622
  statisBlk.maxTbid.uid = record.uid;
20,827,866✔
623

624
  SEncryptData *pEncryptData = &(writer->config->tsdb->pVnode->config.tsdbCfg.encryptData);
20,827,866✔
625

626
  // compress each column
627
  for (int32_t i = 0; i < ARRAY_SIZE(statisBlk.size); i++) {
124,958,163✔
628
    SCompressInfo info = {
104,129,830✔
629
        .dataType = TSDB_DATA_TYPE_BIGINT,
630
        .cmprAlg = statisBlk.cmprAlg,
104,126,502✔
631
        .originalSize = statisBlock->buffers[i].size,
104,126,502✔
632
    };
633

634
    tBufferClear(buffer0);
635
    TAOS_CHECK_GOTO(tCompressDataToBuffer(statisBlock->buffers[i].data, &info, buffer0, assist), &lino, _exit);
104,132,161✔
636
    TAOS_CHECK_GOTO(tsdbWriteFile(writer->fd, writer->file->size, buffer0->data, info.compressedSize, pEncryptData),
104,128,690✔
637
                    &lino, _exit);
638

639
    statisBlk.size[i] = info.compressedSize;
104,135,506✔
640
    statisBlk.dp->size += info.compressedSize;
104,135,935✔
641
    writer->file->size += info.compressedSize;
104,135,935✔
642
  }
643

644
  // compress primary keys
645
  if (statisBlk.numOfPKs > 0) {
20,828,333✔
646
    SValueColumnCompressInfo compressInfo = {.cmprAlg = statisBlk.cmprAlg};
563,939✔
647

648
    tBufferClear(buffer0);
649
    tBufferClear(buffer1);
650

651
    for (int32_t i = 0; i < statisBlk.numOfPKs; i++) {
1,127,878✔
652
      TAOS_CHECK_GOTO(tValueColumnCompress(&statisBlock->firstKeyPKs[i], &compressInfo, buffer1, assist), &lino, _exit);
563,939✔
653
      TAOS_CHECK_GOTO(tValueColumnCompressInfoEncode(&compressInfo, buffer0), &lino, _exit);
563,939✔
654
    }
655

656
    for (int32_t i = 0; i < statisBlk.numOfPKs; i++) {
1,127,878✔
657
      TAOS_CHECK_GOTO(tValueColumnCompress(&statisBlock->lastKeyPKs[i], &compressInfo, buffer1, assist), &lino, _exit);
563,939✔
658
      TAOS_CHECK_GOTO(tValueColumnCompressInfoEncode(&compressInfo, buffer0), &lino, _exit);
563,939✔
659
    }
660

661
    TAOS_CHECK_GOTO(tsdbWriteFile(writer->fd, writer->file->size, buffer0->data, buffer0->size, pEncryptData), &lino,
563,939✔
662
                    _exit);
663
    writer->file->size += buffer0->size;
563,939✔
664
    statisBlk.dp->size += buffer0->size;
563,939✔
665

666
    TAOS_CHECK_GOTO(tsdbWriteFile(writer->fd, writer->file->size, buffer1->data, buffer1->size, pEncryptData), &lino,
563,939✔
667
                    _exit);
668
    writer->file->size += buffer1->size;
563,939✔
669
    statisBlk.dp->size += buffer1->size;
563,939✔
670
  }
671

672
  TAOS_CHECK_GOTO(TARRAY2_APPEND_PTR(writer->statisBlkArray, &statisBlk), &lino, _exit);
41,655,434✔
673

674
  tStatisBlockClear(writer->staticBlock);
20,827,101✔
675

676
_exit:
20,828,667✔
677
  if (code) {
20,827,114✔
678
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
679
              tstrerror(code));
680
  }
681
  return code;
20,823,577✔
682
}
683

684
static int32_t tsdbSttFileDoWriteTombBlock(SSttFileWriter *writer) {
20,518,156✔
685
  if (TOMB_BLOCK_SIZE(writer->tombBlock) == 0) {
20,518,156✔
686
    return 0;
18,804,875✔
687
  }
688

689
  int32_t code = 0;
1,715,222✔
690
  int32_t lino = 0;
1,715,222✔
691

692
  SEncryptData *pEncryptData = &(writer->config->tsdb->pVnode->config.tsdbCfg.encryptData);
1,715,222✔
693

694
  TAOS_CHECK_GOTO(tsdbFileWriteTombBlock(writer->fd, writer->tombBlock, writer->config->cmprAlg, &writer->file->size,
1,715,222✔
695
                                         writer->tombBlkArray, writer->buffers, &writer->ctx->range, pEncryptData),
696
                  &lino, _exit);
697

698
_exit:
1,715,222✔
699
  if (code) {
1,715,222✔
700
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
701
              tstrerror(code));
702
  }
703
  return code;
1,715,222✔
704
}
705

706
int32_t tsdbFileWriteSttBlk(STsdbFD *fd, const TSttBlkArray *sttBlkArray, SFDataPtr *ptr, int64_t *fileSize,
20,519,348✔
707
                            SEncryptData *encryptData) {
708
  ptr->size = TARRAY2_DATA_LEN(sttBlkArray);
20,519,348✔
709
  if (ptr->size > 0) {
20,520,314✔
710
    ptr->offset = *fileSize;
20,467,292✔
711

712
    TAOS_CHECK_RETURN(tsdbWriteFile(fd, *fileSize, (const uint8_t *)TARRAY2_DATA(sttBlkArray), ptr->size, encryptData));
20,468,049✔
713

714
    *fileSize += ptr->size;
20,464,830✔
715
  }
716
  return 0;
20,519,904✔
717
}
718

719
static int32_t tsdbSttFileDoWriteSttBlk(SSttFileWriter *writer) {
20,516,208✔
720
  int32_t code = 0;
20,516,208✔
721
  int32_t lino;
20,515,761✔
722

723
  SEncryptData *pEncryptData = &(writer->config->tsdb->pVnode->config.tsdbCfg.encryptData);
20,518,201✔
724

725
  TAOS_CHECK_GOTO(tsdbFileWriteSttBlk(writer->fd, writer->sttBlkArray, writer->footer->sttBlkPtr, &writer->file->size,
20,520,246✔
726
                                      pEncryptData),
727
                  &lino, _exit);
728

729
_exit:
20,519,206✔
730
  if (code) {
20,519,206✔
731
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
732
              tstrerror(code));
733
  }
734
  return code;
20,519,206✔
735
}
736

737
static int32_t tsdbSttFileDoWriteStatisBlk(SSttFileWriter *writer) {
20,516,042✔
738
  int32_t code = 0;
20,516,042✔
739
  int32_t lino;
20,515,595✔
740
  SEncryptData *pEncryptData = &(writer->config->tsdb->pVnode->config.tsdbCfg.encryptData);
20,517,489✔
741

742
  writer->footer->statisBlkPtr->size = TARRAY2_DATA_LEN(writer->statisBlkArray);
20,519,639✔
743
  if (writer->footer->statisBlkPtr->size) {
20,521,752✔
744
    writer->footer->statisBlkPtr->offset = writer->file->size;
20,467,555✔
745
    TAOS_CHECK_GOTO(tsdbWriteFile(writer->fd, writer->file->size, (const uint8_t *)TARRAY2_DATA(writer->statisBlkArray),
20,466,509✔
746
                                  writer->footer->statisBlkPtr->size, pEncryptData),
747
                    &lino, _exit);
748
    writer->file->size += writer->footer->statisBlkPtr->size;
20,466,705✔
749
  }
750

751
_exit:
20,520,355✔
752
  if (code) {
20,521,030✔
753
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
754
              tstrerror(code));
755
  }
756
  return code;
20,521,030✔
757
}
758

759
static int32_t tsdbSttFileDoWriteTombBlk(SSttFileWriter *writer) {
20,514,026✔
760
  int32_t code = 0;
20,514,026✔
761
  int32_t lino = 0;
20,514,026✔
762

763
  SEncryptData *pEncryptData = &(writer->config->tsdb->pVnode->config.tsdbCfg.encryptData);
20,517,864✔
764

765
  TAOS_CHECK_GOTO(tsdbFileWriteTombBlk(writer->fd, writer->tombBlkArray, writer->footer->tombBlkPtr,
20,520,321✔
766
                                       &writer->file->size, pEncryptData),
767
                  &lino, _exit);
768

769
_exit:
20,518,156✔
770
  if (code) {
20,517,100✔
771
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
772
              tstrerror(code));
773
  }
774
  return code;
20,517,100✔
775
}
776

777
int32_t tsdbFileWriteSttFooter(STsdbFD *fd, const SSttFooter *footer, int64_t *fileSize, SEncryptData *encryptData) {
20,517,667✔
778
  TAOS_CHECK_RETURN(tsdbWriteFile(fd, *fileSize, (const uint8_t *)footer, sizeof(*footer), encryptData));
20,517,667✔
779
  *fileSize += sizeof(*footer);
20,522,000✔
780
  return 0;
20,519,338✔
781
}
782

783
static int32_t tsdbSttFileDoWriteFooter(SSttFileWriter *writer) {
20,515,096✔
784
  SEncryptData *pEncryptData = &(writer->config->tsdb->pVnode->config.tsdbCfg.encryptData);
20,515,096✔
785

786
  return tsdbFileWriteSttFooter(writer->fd, writer->footer, &writer->file->size, pEncryptData);
20,518,886✔
787
}
788

789
static int32_t tsdbSttFWriterDoOpen(SSttFileWriter *writer) {
20,502,032✔
790
  int32_t code = 0;
20,502,032✔
791
  int32_t lino = 0;
20,502,032✔
792
  STsdb  *tsdb = writer->config->tsdb;
20,512,750✔
793

794
  // set
795
  if (!writer->config->skmTb) writer->config->skmTb = writer->skmTb;
20,512,321✔
796
  if (!writer->config->skmRow) writer->config->skmRow = writer->skmRow;
20,512,563✔
797
  writer->buffers = writer->config->buffers;
20,515,449✔
798
  if (writer->buffers == NULL) {
20,515,917✔
799
    writer->buffers = writer->local;
×
800
  }
801

802
  // alloc disk id
803
  SDiskID diskId = {0};
20,515,520✔
804
  code = tsdbAllocateDisk(tsdb, tsdbFTypeLabel(TSDB_FTYPE_STT), writer->config->expLevel, &diskId);
20,509,042✔
805
  TSDB_CHECK_CODE(code, lino, _exit);
20,505,398✔
806

807
  writer->file[0] = (STFile){
20,503,614✔
808
      .type = TSDB_FTYPE_STT,
809
      .did = diskId,
810
      .fid = writer->config->fid,
20,505,398✔
811
      .cid = writer->config->cid,
20,517,990✔
812
      .size = 0,
813
      .minVer = VERSION_MAX,
814
      .maxVer = VERSION_MIN,
815
      .stt[0] =
816
          {
817
              .level = writer->config->level,
20,516,149✔
818
          },
819
  };
820

821
  // open file
822
  int32_t flag = TD_FILE_READ | TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
20,513,207✔
823
  char    fname[TSDB_FILENAME_LEN];
20,512,760✔
824

825
  tsdbTFileName(writer->config->tsdb, writer->file, fname);
20,507,522✔
826
  TAOS_CHECK_GOTO(tsdbOpenFile(fname, writer->config->tsdb, flag, &writer->fd, 0), &lino, _exit);
20,504,544✔
827

828
  uint8_t hdr[TSDB_FHDR_SIZE] = {0};
20,510,896✔
829
  SEncryptData *pEncryptData = &(writer->config->tsdb->pVnode->config.tsdbCfg.encryptData);
20,507,624✔
830

831
  TAOS_CHECK_GOTO(tsdbWriteFile(writer->fd, 0, hdr, sizeof(hdr), pEncryptData), &lino, _exit);
20,516,650✔
832
  writer->file->size += sizeof(hdr);
20,502,475✔
833

834
  // range
835
  writer->ctx->range = (SVersionRange){.minVer = VERSION_MAX, .maxVer = VERSION_MIN};
20,518,278✔
836

837
  writer->ctx->opened = true;
20,517,614✔
838

839
_exit:
20,518,701✔
840
  if (code) {
20,517,978✔
841
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
842
              tstrerror(code));
843
  }
844
  return code;
20,517,978✔
845
}
846

847
static void tsdbSttFWriterDoClose(SSttFileWriter *writer) {
20,513,119✔
848
  for (int32_t i = 0; i < ARRAY_SIZE(writer->local); ++i) {
225,427,924✔
849
    tBufferDestroy(writer->local + i);
204,932,109✔
850
  }
851
  tDestroyTSchema(writer->skmRow->pTSchema);
20,495,815✔
852
  tDestroyTSchema(writer->skmTb->pTSchema);
20,511,469✔
853
  tTombBlockDestroy(writer->tombBlock);
20,513,799✔
854
  tStatisBlockDestroy(writer->staticBlock);
20,513,035✔
855
  tBlockDataDestroy(writer->blockData);
20,516,063✔
856
  TARRAY2_DESTROY(writer->tombBlkArray, NULL);
20,520,059✔
857
  TARRAY2_DESTROY(writer->statisBlkArray, NULL);
20,518,478✔
858
  TARRAY2_DESTROY(writer->sttBlkArray, NULL);
20,519,417✔
859
}
20,518,315✔
860

861
static int32_t tsdbSttFileDoUpdateHeader(SSttFileWriter *writer) {
20,516,713✔
862
  // TODO
863
  return 0;
20,516,713✔
864
}
865

866
static int32_t tsdbSttFWriterCloseCommit(SSttFileWriter *writer, TFileOpArray *opArray) {
20,519,233✔
867
  int32_t lino;
20,518,786✔
868
  int32_t code;
869

870
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteBlockData(writer), &lino, _exit);
20,518,992✔
871
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteStatisBlock(writer), &lino, _exit);
20,518,052✔
872
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteTombBlock(writer), &lino, _exit);
20,519,576✔
873
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteSttBlk(writer), &lino, _exit);
20,519,405✔
874
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteStatisBlk(writer), &lino, _exit);
20,516,816✔
875
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteTombBlk(writer), &lino, _exit);
20,517,696✔
876
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteFooter(writer), &lino, _exit);
20,517,380✔
877
  TAOS_CHECK_GOTO(tsdbSttFileDoUpdateHeader(writer), &lino, _exit);
20,519,873✔
878

879
  SEncryptData *pEncryptData = &(writer->config->tsdb->pVnode->config.tsdbCfg.encryptData);
20,516,713✔
880

881
  TAOS_CHECK_GOTO(tsdbFsyncFile(writer->fd, pEncryptData), &lino, _exit);
20,517,640✔
882

883
  tsdbCloseFile(&writer->fd);
20,515,477✔
884

885
  STFileOp op = (STFileOp){
41,026,435✔
886
      .optype = TSDB_FOP_CREATE,
887
      .fid = writer->config->fid,
20,513,541✔
888
      .nf = writer->file[0],
889
  };
890
  tsdbTFileUpdVerRange(&op.nf, writer->ctx->range);
20,517,205✔
891

892
  TAOS_CHECK_GOTO(TARRAY2_APPEND(opArray, op), &lino, _exit);
41,011,736✔
893

894
_exit:
20,499,009✔
895
  if (code) {
20,500,583✔
896
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
897
              tstrerror(code));
898
  }
899
  return code;
20,500,583✔
900
}
901

902
static int32_t tsdbSttFWriterCloseAbort(SSttFileWriter *writer) {
×
903
  char fname[TSDB_FILENAME_LEN];
×
904
  tsdbTFileName(writer->config->tsdb, writer->file, fname);
×
905
  tsdbCloseFile(&writer->fd);
×
906
  tsdbRemoveFile(fname);
×
907
  return 0;
×
908
}
909

910
int32_t tsdbSttFileWriterOpen(const SSttFileWriterConfig *config, SSttFileWriter **writer) {
21,299,475✔
911
  writer[0] = taosMemoryCalloc(1, sizeof(*writer[0]));
21,299,475✔
912
  if (writer[0] == NULL) {
21,288,741✔
913
    return terrno;
×
914
  }
915

916
  writer[0]->config[0] = config[0];
21,289,393✔
917
  writer[0]->ctx->opened = false;
21,292,097✔
918
  return 0;
21,304,957✔
919
}
920

921
int32_t tsdbSttFileWriterClose(SSttFileWriter **writer, int8_t abort, TFileOpArray *opArray) {
21,306,813✔
922
  if (writer == NULL || writer[0] == NULL) return 0;
21,306,813✔
923

924
  int32_t code = 0;
21,307,576✔
925
  int32_t lino = 0;
21,307,576✔
926

927
  if (writer[0]->ctx->opened) {
21,307,576✔
928
    if (abort) {
20,519,269✔
929
      TAOS_CHECK_GOTO(tsdbSttFWriterCloseAbort(writer[0]), &lino, _exit);
×
930
    } else {
931
      TAOS_CHECK_GOTO(tsdbSttFWriterCloseCommit(writer[0], opArray), &lino, _exit);
20,519,269✔
932
    }
933
    tsdbSttFWriterDoClose(writer[0]);
20,505,124✔
934
  }
935
  taosMemoryFree(writer[0]);
21,307,409✔
936
  writer[0] = NULL;
21,305,478✔
937

938
_exit:
21,306,162✔
939
  if (code) {
21,306,506✔
940
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer[0]->config->tsdb->pVnode), __func__, __FILE__, lino,
×
941
              tstrerror(code));
942
  }
943
  return code;
21,303,569✔
944
}
945

946
int32_t tsdbSttFileWriteRow(SSttFileWriter *writer, SRowInfo *row) {
2,147,483,647✔
947
  int32_t code = 0;
2,147,483,647✔
948
  int32_t lino = 0;
2,147,483,647✔
949

950
  if (!writer->ctx->opened) {
2,147,483,647✔
951
    TAOS_CHECK_GOTO(tsdbSttFWriterDoOpen(writer), &lino, _exit);
20,460,140✔
952
  }
953

954
  if (!TABLE_SAME_SCHEMA(row->suid, row->uid, writer->ctx->tbid->suid, writer->ctx->tbid->uid)) {
2,147,483,647✔
955
    TAOS_CHECK_GOTO(tsdbSttFileDoWriteBlockData(writer), &lino, _exit);
27,150,975✔
956

957
    TAOS_CHECK_GOTO(tsdbUpdateSkmTb(writer->config->tsdb, (TABLEID *)row, writer->config->skmTb), &lino, _exit);
27,116,747✔
958

959
    TABLEID id = {
27,141,313✔
960
        .suid = row->suid,
27,146,757✔
961
        .uid = row->suid ? 0 : row->uid,
27,150,372✔
962
    };
963
    TAOS_CHECK_GOTO(tBlockDataInit(writer->blockData, &id, writer->config->skmTb->pTSchema, NULL, 0), &lino, _exit);
27,124,691✔
964
  }
965

966
  if (writer->ctx->tbid->uid != row->uid) {
2,147,483,647✔
967
    writer->ctx->tbid->suid = row->suid;
97,708,058✔
968
    writer->ctx->tbid->uid = row->uid;
97,735,913✔
969
  }
970

971
  STsdbRowKey key;
2,147,483,647✔
972
  tsdbRowGetKey(&row->row, &key);
2,147,483,647✔
973

974
  for (;;) {
975
    code = tStatisBlockPut(writer->staticBlock, row, writer->config->maxRow);
2,147,483,647✔
976
    if (code == TSDB_CODE_INVALID_PARA) {
2,147,483,647✔
977
      TAOS_CHECK_GOTO(tsdbSttFileDoWriteStatisBlock(writer), &lino, _exit);
143,560✔
978
      continue;
143,546✔
979
    } else {
980
      TSDB_CHECK_CODE(code, lino, _exit);
2,147,483,647✔
981
    }
982
    break;
2,147,483,647✔
983
  }
984

985
  if (row->row.type == TSDBROW_ROW_FMT) {
2,147,483,647✔
986
    TAOS_CHECK_GOTO(tsdbUpdateSkmRow(writer->config->tsdb, writer->ctx->tbid,  //
2,147,483,647✔
987
                                     TSDBROW_SVERSION(&row->row), writer->config->skmRow),
988
                    &lino, _exit);
989
  }
990

991
  // row to col conversion
992
  if (key.version <= writer->config->compactVersion                                //
2,147,483,647✔
993
      && writer->blockData->nRow > 0                                               //
2,147,483,647✔
994
      && (writer->blockData->uid                                                   //
2,147,483,647✔
995
              ? writer->blockData->uid                                             //
996
              : writer->blockData->aUid[writer->blockData->nRow - 1]) == row->uid  //
2,147,483,647✔
997
      && tsdbRowCompareWithoutVersion(&row->row,
2,147,483,647✔
998
                                      &tsdbRowFromBlockData(writer->blockData, writer->blockData->nRow - 1)) == 0  //
2,147,483,647✔
999
  ) {
1000
    TAOS_CHECK_GOTO(tBlockDataUpdateRow(writer->blockData, &row->row, writer->config->skmRow->pTSchema), &lino, _exit);
192,282,857✔
1001
  } else {
1002
    if (writer->blockData->nRow >= writer->config->maxRow) {
2,147,483,647✔
1003
      TAOS_CHECK_GOTO(tsdbSttFileDoWriteBlockData(writer), &lino, _exit);
20,958,190✔
1004
    }
1005

1006
    TAOS_CHECK_GOTO(tBlockDataAppendRow(writer->blockData, &row->row, writer->config->skmRow->pTSchema, row->uid),
2,147,483,647✔
1007
                    &lino, _exit);
1008
  }
1009

1010
_exit:
2,147,483,647✔
1011
  if (code) {
2,147,483,647✔
1012
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
1013
              tstrerror(code));
1014
  }
1015
  return code;
2,147,483,647✔
1016
}
1017

1018
int32_t tsdbSttFileWriteBlockData(SSttFileWriter *writer, SBlockData *bdata) {
12,902,986✔
1019
  int32_t code = 0;
12,902,986✔
1020
  int32_t lino = 0;
12,902,986✔
1021

1022
  SRowInfo row[1];
12,902,581✔
1023
  row->suid = bdata->suid;
12,904,056✔
1024
  for (int32_t i = 0; i < bdata->nRow; i++) {
143,535,673✔
1025
    row->uid = bdata->uid ? bdata->uid : bdata->aUid[i];
130,629,477✔
1026
    row->row = tsdbRowFromBlockData(bdata, i);
130,630,012✔
1027

1028
    TAOS_CHECK_GOTO(tsdbSttFileWriteRow(writer, row), &lino, _exit);
130,630,012✔
1029
  }
1030

1031
_exit:
12,904,056✔
1032
  if (code) {
12,904,056✔
1033
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
1034
              tstrerror(code));
1035
  }
1036
  return code;
12,904,056✔
1037
}
1038

1039
int32_t tsdbSttFileWriteTombRecord(SSttFileWriter *writer, const STombRecord *record) {
6,879,616✔
1040
  int32_t code;
1041
  int32_t lino;
6,879,616✔
1042

1043
  if (!writer->ctx->opened) {
6,879,616✔
1044
    TAOS_CHECK_GOTO(tsdbSttFWriterDoOpen(writer), &lino, _exit);
53,465✔
1045
  } else {
1046
    if (writer->blockData->nRow > 0) {
6,826,151✔
1047
      TAOS_CHECK_GOTO(tsdbSttFileDoWriteBlockData(writer), &lino, _exit);
1,661,757✔
1048
    }
1049

1050
    if (STATIS_BLOCK_SIZE(writer->staticBlock) > 0) {
6,826,151✔
1051
      TAOS_CHECK_GOTO(tsdbSttFileDoWriteStatisBlock(writer), &lino, _exit);
1,661,757✔
1052
    }
1053
  }
1054

1055
  TAOS_CHECK_GOTO(tTombBlockPut(writer->tombBlock, record), &lino, _exit);
6,879,616✔
1056

1057
  if (TOMB_BLOCK_SIZE(writer->tombBlock) >= writer->config->maxRow) {
6,879,616✔
1058
    TAOS_CHECK_GOTO(tsdbSttFileDoWriteTombBlock(writer), &lino, _exit);
×
1059
  }
1060

1061
_exit:
6,879,616✔
1062
  if (code) {
6,879,616✔
1063
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
1064
              tstrerror(code));
1065
  } else {
1066
    tsdbTrace("vgId:%d write tomb record to stt file:%s, cid:%" PRId64 ", suid:%" PRId64 ", uid:%" PRId64
6,879,616✔
1067
              ", version:%" PRId64,
1068
              TD_VID(writer->config->tsdb->pVnode), writer->fd->path, writer->config->cid, record->suid, record->uid,
1069
              record->version);
1070
  }
1071
  return code;
6,879,616✔
1072
}
1073

1074
bool tsdbSttFileWriterIsOpened(SSttFileWriter *writer) { return writer->ctx->opened; }
4,394,713✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE TRIAL · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc