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

taosdata / TDengine / #5095

17 May 2026 01:15AM UTC coverage: 73.282% (-0.005%) from 73.287%
#5095

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)

281251 of 383795 relevant lines covered (73.28%)

132492491.26 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) {
644,645,266✔
39
  int32_t code = 0;
644,645,266✔
40
  int32_t lino = 0;
644,645,266✔
41

42
  reader[0] = taosMemoryCalloc(1, sizeof(*reader[0]));
644,721,819✔
43
  if (reader[0] == NULL) {
644,281,856✔
44
    TAOS_CHECK_GOTO(terrno, &lino, _exit);
×
45
  }
46

47
  reader[0]->config[0] = config[0];
644,244,035✔
48
  reader[0]->buffers = config->buffers;
644,346,242✔
49
  if (reader[0]->buffers == NULL) {
644,744,428✔
50
    reader[0]->buffers = reader[0]->local;
644,744,171✔
51
  }
52

53
  // open file
54
  if (fname) {
644,759,220✔
55
    TAOS_CHECK_GOTO(tsdbOpenFile(fname, config->tsdb, TD_FILE_READ, &reader[0]->fd, 0), &lino, _exit);
644,759,220✔
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);
644,684,324✔
64
  if (offset < TSDB_FHDR_SIZE) {
644,597,356✔
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);
644,597,356✔
71

72
#if 1
73
  TAOS_CHECK_GOTO(
644,745,797✔
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:
644,304,035✔
95
  if (code) {
644,259,961✔
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;
644,259,961✔
101
}
102

103
void tsdbSttFileReaderClose(SSttFileReader **reader) {
644,608,662✔
104
  if (reader[0]) {
644,608,662✔
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);
644,962,353✔
109
    TARRAY2_DESTROY(reader[0]->tombBlkArray, NULL);
644,521,883✔
110
    TARRAY2_DESTROY(reader[0]->statisBlkArray, NULL);
644,565,039✔
111
    TARRAY2_DESTROY(reader[0]->sttBlkArray, NULL);
644,446,740✔
112
    taosMemoryFree(reader[0]);
644,581,353✔
113
    reader[0] = NULL;
644,697,334✔
114
  }
115
}
644,553,196✔
116

117
// SSttFSegReader
118
int32_t tsdbSttFileReadStatisBlk(SSttFileReader *reader, const TStatisBlkArray **statisBlkArray) {
641,145,819✔
119
  if (!reader->ctx->statisBlkLoaded) {
641,145,819✔
120
    if (reader->footer->statisBlkPtr->size > 0) {
641,344,314✔
121
      if (reader->footer->statisBlkPtr->size % sizeof(SStatisBlk) != 0) {
641,208,921✔
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);
640,955,376✔
128
      void   *data = taosMemoryMalloc(reader->footer->statisBlkPtr->size);
641,135,729✔
129
      if (!data) {
641,054,285✔
130
        return terrno;
×
131
      }
132

133
      SEncryptData *pEncryptData = &(reader->config->tsdb->pVnode->config.tsdbCfg.encryptData);
641,054,285✔
134

135
      int32_t code = tsdbReadFile(reader->fd, reader->footer->statisBlkPtr->offset, data,
641,229,516✔
136
                                  reader->footer->statisBlkPtr->size, 0, pEncryptData);
137
      if (code) {
641,283,159✔
138
        taosMemoryFree(data);
×
139
        return code;
×
140
      }
141

142
      TARRAY2_INIT_EX(reader->statisBlkArray, size, size, data);
641,283,159✔
143
    } else {
144
      TARRAY2_INIT(reader->statisBlkArray);
170,129✔
145
    }
146

147
    reader->ctx->statisBlkLoaded = true;
641,482,179✔
148
  }
149

150
  statisBlkArray[0] = reader->statisBlkArray;
641,399,461✔
151
  return 0;
641,444,230✔
152
}
153

154
int32_t tsdbSttFileReadTombBlk(SSttFileReader *reader, const TTombBlkArray **tombBlkArray) {
644,513,274✔
155
  if (!reader->ctx->tombBlkLoaded) {
644,513,274✔
156
    if (reader->footer->tombBlkPtr->size > 0) {
644,660,912✔
157
      if (reader->footer->tombBlkPtr->size % sizeof(STombBlk) != 0) {
13,049,927✔
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,049,927✔
164
      void   *data = taosMemoryMalloc(reader->footer->tombBlkPtr->size);
13,049,927✔
165
      if (!data) {
13,049,927✔
166
        return terrno;
×
167
      }
168

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

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

178
      TARRAY2_INIT_EX(reader->tombBlkArray, size, size, data);
13,049,927✔
179
    } else {
180
      TARRAY2_INIT(reader->tombBlkArray);
631,494,942✔
181
    }
182

183
    reader->ctx->tombBlkLoaded = true;
644,702,430✔
184
  }
185

186
  tombBlkArray[0] = reader->tombBlkArray;
644,659,229✔
187
  return 0;
644,587,442✔
188
}
189

190
int32_t tsdbSttFileReadSttBlk(SSttFileReader *reader, const TSttBlkArray **sttBlkArray) {
644,330,101✔
191
  if (!reader->ctx->sttBlkLoaded) {
644,330,101✔
192
    if (reader->footer->sttBlkPtr->size > 0) {
644,597,559✔
193
      if (reader->footer->sttBlkPtr->size % sizeof(SSttBlk) != 0) {
644,467,017✔
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);
644,009,691✔
200
      void   *data = taosMemoryMalloc(reader->footer->sttBlkPtr->size);
644,300,522✔
201
      if (!data) {
644,231,503✔
202
        return terrno;
×
203
      }
204

205
      SEncryptData *pEncryptData = &(reader->config->tsdb->pVnode->config.tsdbCfg.encryptData);
644,231,503✔
206

207
      int32_t code = tsdbReadFile(reader->fd, reader->footer->sttBlkPtr->offset, data, reader->footer->sttBlkPtr->size,
644,477,950✔
208
                                  0, pEncryptData);
209
      if (code) {
644,481,934✔
210
        taosMemoryFree(data);
×
211
        return code;
×
212
      }
213

214
      TARRAY2_INIT_EX(reader->sttBlkArray, size, size, data);
644,481,934✔
215
    } else {
216
      TARRAY2_INIT(reader->sttBlkArray);
223,644✔
217
    }
218

219
    reader->ctx->sttBlkLoaded = true;
644,698,677✔
220
  }
221

222
  sttBlkArray[0] = reader->sttBlkArray;
644,678,274✔
223
  return 0;
644,731,303✔
224
}
225

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

230
  SBuffer *buffer0 = reader->buffers + 0;
18,133,400✔
231
  SBuffer *assist = reader->buffers + 1;
18,133,392✔
232

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

235
  // load data
236
  tBufferClear(buffer0);
237
  TAOS_CHECK_GOTO(
18,133,400✔
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,134,115✔
242
  TAOS_CHECK_GOTO(tBlockDataDecompress(&br, bData, assist), &lino, _exit);
18,134,115✔
243

244
_exit:
18,131,225✔
245
  if (code) {
18,131,225✔
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,131,225✔
250
}
251

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

257
  SDiskDataHdr hdr;
297,539,222✔
258
  SBuffer     *buffer0 = reader->buffers + 0;
297,526,799✔
259
  SBuffer     *buffer1 = reader->buffers + 1;
297,505,100✔
260
  SBuffer     *assist = reader->buffers + 2;
297,524,519✔
261

262
  SEncryptData *pEncryptData = &(reader->config->tsdb->pVnode->config.tsdbCfg.encryptData);
297,566,745✔
263

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

269
  // decode header
270
  SBufferReader br = BUFFER_READER_INITIALIZER(0, buffer0);
297,567,704✔
271
  TAOS_CHECK_GOTO(tGetDiskDataHdr(&br, &hdr), &lino, _exit);
297,546,759✔
272

273
  if (hdr.delimiter != TSDB_FILE_DLMT) {
297,493,948✔
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);
297,493,948✔
281
  bData->suid = hdr.suid;
297,516,981✔
282
  bData->uid = (sttBlk->suid == 0) ? sttBlk->minUid : 0;
297,517,755✔
283
  bData->nRow = hdr.nRow;
297,507,768✔
284

285
  // key part
286
  TAOS_CHECK_GOTO(tBlockDataDecompressKeyPart(&hdr, &br, bData, assist), &lino, _exit);
297,504,813✔
287
  if (br.offset != buffer0->size) {
297,549,393✔
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;
297,540,796✔
295
  for (int i = 0; i < ncid; i++) {
386,753,774✔
296
    if (tBlockDataGetColData(bData, cids[i]) == NULL) {
308,888,593✔
297
      loadExtra = true;
219,684,037✔
298
      break;
219,684,037✔
299
    }
300
  }
301

302
  if (!loadExtra) {
297,549,218✔
303
    goto _exit;
77,864,460✔
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,
219,683,868✔
309
                                       pEncryptData),
310
                  &lino, _exit);
311

312
  // load each column
313
  SBlockCol blockCol = {
219,709,690✔
314
      .cid = 0,
315
  };
316
  br = BUFFER_READER_INITIALIZER(0, buffer0);
219,712,757✔
317
  for (int32_t i = 0; i < ncid; i++) {
991,085,466✔
318
    int16_t cid = cids[i];
771,339,890✔
319

320
    if (tBlockDataGetColData(bData, cid)) {  // already loaded
771,358,040✔
321
      continue;
44,128,933✔
322
    }
323

324
    while (cid > blockCol.cid) {
2,147,483,647✔
325
      if (br.offset >= buffer0->size) {
1,715,631,099✔
326
        blockCol.cid = INT16_MAX;
23,494,590✔
327
        break;
23,494,590✔
328
      }
329

330
      TAOS_CHECK_GOTO(tGetBlockCol(&br, &blockCol, hdr.fmtVer, hdr.cmprAlg), &lino, _exit);
1,692,157,018✔
331
    }
332

333
    if (cid < blockCol.cid) {
727,312,160✔
334
      const STColumn *tcol = tTSchemaSearchColumn(pTSchema, cid);
104,499,060✔
335
      TSDB_CHECK_NULL(tcol, code, lino, _exit, TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER);
104,492,881✔
336
      SBlockCol none = {
104,492,881✔
337
          .cid = cid,
338
          .type = tcol->type,
104,493,614✔
339
          .cflag = tcol->flags,
104,492,883✔
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);
104,492,883✔
348
    } else if (cid == blockCol.cid) {
622,813,100✔
349
      // load from file
350
      tBufferClear(buffer1);
351
      TAOS_CHECK_GOTO(
622,780,052✔
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);
622,765,768✔
358
      TAOS_CHECK_GOTO(tBlockDataDecompressColData(&hdr, &blockCol, &br1, bData, assist), &lino, _exit);
622,782,558✔
359
    }
360
  }
361

362
_exit:
297,684,481✔
363
  if (code) {
297,558,299✔
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;
297,558,299✔
368
}
369

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

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

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

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

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

400
  if (br.offset != tombBlk->dp->size) {
6,636,582✔
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,636,582✔
407
  if (code) {
6,636,582✔
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,636,582✔
412
}
413

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

418
  if (statisBlk->numOfPKs > TD_MAX_PK_COLS) {
335,251,047✔
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;
335,219,712✔
426
  SBuffer *assist = reader->buffers + 1;
335,263,380✔
427

428
  SEncryptData *pEncryptData = &(reader->config->tsdb->pVnode->config.tsdbCfg.encryptData);
335,281,976✔
429

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

436
  // decode data
437
  tStatisBlockClear(statisBlock);
335,287,522✔
438
  statisBlock->numOfPKs = statisBlk->numOfPKs;
335,248,578✔
439
  statisBlock->numOfRecords = statisBlk->numRec;
335,268,044✔
440
  SBufferReader br = BUFFER_READER_INITIALIZER(0, buffer0);
335,268,305✔
441
  for (int32_t i = 0; i < ARRAY_SIZE(statisBlock->buffers); ++i) {
2,011,685,729✔
442
    SCompressInfo info = {
1,676,377,563✔
443
        .dataType = TSDB_DATA_TYPE_BIGINT,
444
        .cmprAlg = statisBlk->cmprAlg,
1,676,394,588✔
445
        .compressedSize = statisBlk->size[i],
1,676,423,473✔
446
        .originalSize = statisBlk->numRec * sizeof(int64_t),
1,676,401,328✔
447
    };
448

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

453
  if (statisBlk->numOfPKs > 0) {
335,308,166✔
454
    SValueColumnCompressInfo firstKeyInfos[TD_MAX_PK_COLS];
118,533,251✔
455
    SValueColumnCompressInfo lastKeyInfos[TD_MAX_PK_COLS];
118,533,251✔
456

457
    // decode compress info
458
    for (int32_t i = 0; i < statisBlk->numOfPKs; i++) {
237,060,280✔
459
      TAOS_CHECK_GOTO(tValueColumnCompressInfoDecode(&br, &firstKeyInfos[i]), &lino, _exit);
118,531,181✔
460
    }
461

462
    for (int32_t i = 0; i < statisBlk->numOfPKs; i++) {
237,060,284✔
463
      TAOS_CHECK_GOTO(tValueColumnCompressInfoDecode(&br, &lastKeyInfos[i]), &lino, _exit);
118,504,178✔
464
    }
465

466
    // decode value columns
467
    for (int32_t i = 0; i < statisBlk->numOfPKs; i++) {
237,049,908✔
468
      TAOS_CHECK_GOTO(tValueColumnDecompress(BR_PTR(&br), firstKeyInfos + i, &statisBlock->firstKeyPKs[i], assist),
118,525,647✔
469
                      &lino, _exit);
470
      br.offset += (firstKeyInfos[i].dataCompressedSize + firstKeyInfos[i].offsetCompressedSize);
118,524,953✔
471
    }
472

473
    for (int32_t i = 0; i < statisBlk->numOfPKs; i++) {
237,049,189✔
474
      TAOS_CHECK_GOTO(tValueColumnDecompress(BR_PTR(&br), &lastKeyInfos[i], &statisBlock->lastKeyPKs[i], assist), &lino,
118,515,279✔
475
                      _exit);
476
      br.offset += (lastKeyInfos[i].dataCompressedSize + lastKeyInfos[i].offsetCompressedSize);
118,533,941✔
477
    }
478
  }
479

480
  if (br.offset != buffer0->size) {
335,296,305✔
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:
335,293,159✔
488
  if (code) {
335,287,074✔
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;
335,287,074✔
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,
44,887,166✔
524
                                           int64_t *fileSize, TSttBlkArray *sttBlkArray, SBuffer *buffers,
525
                                           SVersionRange *range, SEncryptData *encryptData) {
526
  if (blockData->nRow == 0) return 0;
44,887,166✔
527

528
  int32_t code = 0;
44,887,176✔
529

530
  SSttBlk sttBlk[1] = {{
44,890,226✔
531
      .suid = blockData->suid,
44,889,951✔
532
      .minUid = blockData->uid ? blockData->uid : blockData->aUid[0],
44,889,602✔
533
      .maxUid = blockData->uid ? blockData->uid : blockData->aUid[blockData->nRow - 1],
44,891,625✔
534
      .minKey = blockData->aTSKEY[0],
44,892,455✔
535
      .maxKey = blockData->aTSKEY[0],
44,887,255✔
536
      .minVer = blockData->aVersion[0],
44,891,881✔
537
      .maxVer = blockData->aVersion[0],
44,890,706✔
538
      .nRow = blockData->nRow,
44,889,412✔
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);
44,895,592✔
549
  TAOS_CHECK_RETURN(tBlockDataCompress(blockData, info, buffers, buffers + 4));
44,892,815✔
550

551
  sttBlk->bInfo.offset = *fileSize;
44,889,680✔
552
  sttBlk->bInfo.szKey = buffers[0].size + buffers[1].size;
44,889,581✔
553
  sttBlk->bInfo.szBlock = buffers[2].size + buffers[3].size + sttBlk->bInfo.szKey;
44,890,798✔
554
  for (int i = 0; i < 4; i++) {
224,470,120✔
555
    if (buffers[i].size) {
179,574,443✔
556
      TAOS_CHECK_RETURN(tsdbWriteFile(fd, *fileSize, buffers[i].data, buffers[i].size, encryptData));
179,331,584✔
557
      *fileSize += buffers[i].size;
179,327,716✔
558
    }
559
  }
560

561
  TAOS_CHECK_RETURN(TARRAY2_APPEND_PTR(sttBlkArray, sttBlk));
89,790,760✔
562

563
  tBlockDataClear(blockData);
44,895,083✔
564
  return 0;
44,895,214✔
565
}
566

567
static int32_t tsdbSttFileDoWriteBlockData(SSttFileWriter *writer) {
66,804,174✔
568
  if (writer->blockData->nRow == 0) return 0;
66,804,174✔
569

570
  int32_t code = 0;
44,893,595✔
571
  int32_t lino = 0;
44,893,595✔
572

573
  tb_uid_t         uid = writer->blockData->suid == 0 ? writer->blockData->uid : writer->blockData->suid;
44,895,281✔
574
  SColCompressInfo info = {.defaultCmprAlg = writer->config->cmprAlg, .pColCmpr = NULL};
44,895,967✔
575
  code = metaGetColCmpr(writer->config->tsdb->pVnode->pMeta, uid, &(info.pColCmpr));
44,894,586✔
576

577
  SEncryptData *pEncryptData = &(writer->config->tsdb->pVnode->config.tsdbCfg.encryptData);
44,892,192✔
578

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

583
_exit:
44,890,432✔
584
  if (code) {
44,890,856✔
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);
44,890,856✔
589
  return code;
44,889,345✔
590
}
591

592
static int32_t tsdbSttFileDoWriteStatisBlock(SSttFileWriter *writer) {
22,270,680✔
593
  if (writer->staticBlock->numOfRecords == 0) {
22,270,680✔
594
    return 0;
1,502,880✔
595
  }
596

597
  int32_t code = 0;
20,771,738✔
598
  int32_t lino = 0;
20,771,738✔
599

600
  SBuffer *buffer0 = writer->buffers + 0;
20,769,114✔
601
  SBuffer *buffer1 = writer->buffers + 1;
20,770,723✔
602
  SBuffer *assist = writer->buffers + 2;
20,772,194✔
603

604
  STbStatisRecord record;
20,770,797✔
605
  STbStatisBlock *statisBlock = writer->staticBlock;
20,773,115✔
606
  SStatisBlk      statisBlk = {0};
20,773,995✔
607

608
  statisBlk.dp->offset = writer->file->size;
20,774,530✔
609
  statisBlk.dp->size = 0;
20,771,816✔
610
  statisBlk.numRec = statisBlock->numOfRecords;
20,771,816✔
611
  statisBlk.cmprAlg = writer->config->cmprAlg;
20,768,335✔
612
  statisBlk.numOfPKs = statisBlock->numOfPKs;
20,772,065✔
613

614
  code = tStatisBlockGet(statisBlock, 0, &record);
20,772,729✔
615
  TSDB_CHECK_CODE(code, lino, _exit);
20,773,258✔
616
  statisBlk.minTbid.suid = record.suid;
20,773,258✔
617
  statisBlk.minTbid.uid = record.uid;
20,773,258✔
618

619
  code = tStatisBlockGet(statisBlock, statisBlock->numOfRecords - 1, &record);
20,773,258✔
620
  TSDB_CHECK_CODE(code, lino, _exit);
20,771,790✔
621
  statisBlk.maxTbid.suid = record.suid;
20,771,790✔
622
  statisBlk.maxTbid.uid = record.uid;
20,771,790✔
623

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

626
  // compress each column
627
  for (int32_t i = 0; i < ARRAY_SIZE(statisBlk.size); i++) {
124,627,971✔
628
    SCompressInfo info = {
103,853,441✔
629
        .dataType = TSDB_DATA_TYPE_BIGINT,
630
        .cmprAlg = statisBlk.cmprAlg,
103,859,455✔
631
        .originalSize = statisBlock->buffers[i].size,
103,859,455✔
632
    };
633

634
    tBufferClear(buffer0);
635
    TAOS_CHECK_GOTO(tCompressDataToBuffer(statisBlock->buffers[i].data, &info, buffer0, assist), &lino, _exit);
103,859,829✔
636
    TAOS_CHECK_GOTO(tsdbWriteFile(writer->fd, writer->file->size, buffer0->data, info.compressedSize, pEncryptData),
103,865,878✔
637
                    &lino, _exit);
638

639
    statisBlk.size[i] = info.compressedSize;
103,870,829✔
640
    statisBlk.dp->size += info.compressedSize;
103,868,364✔
641
    writer->file->size += info.compressedSize;
103,868,364✔
642
  }
643

644
  // compress primary keys
645
  if (statisBlk.numOfPKs > 0) {
20,774,530✔
646
    SValueColumnCompressInfo compressInfo = {.cmprAlg = statisBlk.cmprAlg};
519,607✔
647

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

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

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

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

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

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

674
  tStatisBlockClear(writer->staticBlock);
20,773,258✔
675

676
_exit:
20,775,020✔
677
  if (code) {
20,773,609✔
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,769,035✔
682
}
683

684
static int32_t tsdbSttFileDoWriteTombBlock(SSttFileWriter *writer) {
20,467,632✔
685
  if (TOMB_BLOCK_SIZE(writer->tombBlock) == 0) {
20,467,632✔
686
    return 0;
18,749,128✔
687
  }
688

689
  int32_t code = 0;
1,721,086✔
690
  int32_t lino = 0;
1,721,086✔
691

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

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

698
_exit:
1,721,086✔
699
  if (code) {
1,721,086✔
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,721,086✔
704
}
705

706
int32_t tsdbFileWriteSttBlk(STsdbFD *fd, const TSttBlkArray *sttBlkArray, SFDataPtr *ptr, int64_t *fileSize,
20,468,120✔
707
                            SEncryptData *encryptData) {
708
  ptr->size = TARRAY2_DATA_LEN(sttBlkArray);
20,468,120✔
709
  if (ptr->size > 0) {
20,469,801✔
710
    ptr->offset = *fileSize;
20,417,076✔
711

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

714
    *fileSize += ptr->size;
20,412,608✔
715
  }
716
  return 0;
20,469,488✔
717
}
718

719
static int32_t tsdbSttFileDoWriteSttBlk(SSttFileWriter *writer) {
20,464,147✔
720
  int32_t code = 0;
20,464,147✔
721
  int32_t lino;
20,463,700✔
722

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

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

729
_exit:
20,468,967✔
730
  if (code) {
20,468,967✔
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,468,967✔
735
}
736

737
static int32_t tsdbSttFileDoWriteStatisBlk(SSttFileWriter *writer) {
20,465,106✔
738
  int32_t code = 0;
20,465,106✔
739
  int32_t lino;
20,464,659✔
740
  SEncryptData *pEncryptData = &(writer->config->tsdb->pVnode->config.tsdbCfg.encryptData);
20,466,162✔
741

742
  writer->footer->statisBlkPtr->size = TARRAY2_DATA_LEN(writer->statisBlkArray);
20,467,975✔
743
  if (writer->footer->statisBlkPtr->size) {
20,469,725✔
744
    writer->footer->statisBlkPtr->offset = writer->file->size;
20,416,268✔
745
    TAOS_CHECK_GOTO(tsdbWriteFile(writer->fd, writer->file->size, (const uint8_t *)TARRAY2_DATA(writer->statisBlkArray),
20,414,839✔
746
                                  writer->footer->statisBlkPtr->size, pEncryptData),
747
                    &lino, _exit);
748
    writer->file->size += writer->footer->statisBlkPtr->size;
20,416,002✔
749
  }
750

751
_exit:
20,470,144✔
752
  if (code) {
20,471,647✔
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,471,647✔
757
}
758

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

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

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

769
_exit:
20,467,632✔
770
  if (code) {
20,465,839✔
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,465,839✔
775
}
776

777
int32_t tsdbFileWriteSttFooter(STsdbFD *fd, const SSttFooter *footer, int64_t *fileSize, SEncryptData *encryptData) {
20,467,714✔
778
  TAOS_CHECK_RETURN(tsdbWriteFile(fd, *fileSize, (const uint8_t *)footer, sizeof(*footer), encryptData));
20,467,714✔
779
  *fileSize += sizeof(*footer);
20,469,398✔
780
  return 0;
20,468,465✔
781
}
782

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

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

789
static int32_t tsdbSttFWriterDoOpen(SSttFileWriter *writer) {
20,455,021✔
790
  int32_t code = 0;
20,455,021✔
791
  int32_t lino = 0;
20,455,021✔
792
  STsdb  *tsdb = writer->config->tsdb;
20,457,618✔
793

794
  // set
795
  if (!writer->config->skmTb) writer->config->skmTb = writer->skmTb;
20,461,296✔
796
  if (!writer->config->skmRow) writer->config->skmRow = writer->skmRow;
20,458,354✔
797
  writer->buffers = writer->config->buffers;
20,462,774✔
798
  if (writer->buffers == NULL) {
20,465,317✔
799
    writer->buffers = writer->local;
×
800
  }
801

802
  // alloc disk id
803
  SDiskID diskId = {0};
20,465,713✔
804
  code = tsdbAllocateDisk(tsdb, tsdbFTypeLabel(TSDB_FTYPE_STT), writer->config->expLevel, &diskId);
20,457,077✔
805
  TSDB_CHECK_CODE(code, lino, _exit);
20,458,246✔
806

807
  writer->file[0] = (STFile){
20,454,872✔
808
      .type = TSDB_FTYPE_STT,
809
      .did = diskId,
810
      .fid = writer->config->fid,
20,458,246✔
811
      .cid = writer->config->cid,
20,467,571✔
812
      .size = 0,
813
      .minVer = VERSION_MAX,
814
      .maxVer = VERSION_MIN,
815
      .stt[0] =
816
          {
817
              .level = writer->config->level,
20,464,004✔
818
          },
819
  };
820

821
  // open file
822
  int32_t flag = TD_FILE_READ | TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
20,460,065✔
823
  char    fname[TSDB_FILENAME_LEN];
20,459,618✔
824

825
  tsdbTFileName(writer->config->tsdb, writer->file, fname);
20,454,905✔
826
  TAOS_CHECK_GOTO(tsdbOpenFile(fname, writer->config->tsdb, flag, &writer->fd, 0), &lino, _exit);
20,449,477✔
827

828
  uint8_t hdr[TSDB_FHDR_SIZE] = {0};
20,459,701✔
829
  SEncryptData *pEncryptData = &(writer->config->tsdb->pVnode->config.tsdbCfg.encryptData);
20,457,284✔
830

831
  TAOS_CHECK_GOTO(tsdbWriteFile(writer->fd, 0, hdr, sizeof(hdr), pEncryptData), &lino, _exit);
20,464,249✔
832
  writer->file->size += sizeof(hdr);
20,454,742✔
833

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

837
  writer->ctx->opened = true;
20,469,455✔
838

839
_exit:
20,468,152✔
840
  if (code) {
20,468,763✔
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,468,763✔
845
}
846

847
static void tsdbSttFWriterDoClose(SSttFileWriter *writer) {
20,468,234✔
848
  for (int32_t i = 0; i < ARRAY_SIZE(writer->local); ++i) {
225,096,238✔
849
    tBufferDestroy(writer->local + i);
204,634,552✔
850
  }
851
  tDestroyTSchema(writer->skmRow->pTSchema);
20,461,686✔
852
  tDestroyTSchema(writer->skmTb->pTSchema);
20,468,036✔
853
  tTombBlockDestroy(writer->tombBlock);
20,468,460✔
854
  tStatisBlockDestroy(writer->staticBlock);
20,468,683✔
855
  tBlockDataDestroy(writer->blockData);
20,469,746✔
856
  TARRAY2_DESTROY(writer->tombBlkArray, NULL);
20,470,911✔
857
  TARRAY2_DESTROY(writer->statisBlkArray, NULL);
20,470,533✔
858
  TARRAY2_DESTROY(writer->sttBlkArray, NULL);
20,468,938✔
859
}
20,467,776✔
860

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

866
static int32_t tsdbSttFWriterCloseCommit(SSttFileWriter *writer, TFileOpArray *opArray) {
20,467,865✔
867
  int32_t lino;
20,467,418✔
868
  int32_t code;
869

870
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteBlockData(writer), &lino, _exit);
20,468,526✔
871
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteStatisBlock(writer), &lino, _exit);
20,464,687✔
872
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteTombBlock(writer), &lino, _exit);
20,468,056✔
873
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteSttBlk(writer), &lino, _exit);
20,468,920✔
874
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteStatisBlk(writer), &lino, _exit);
20,464,682✔
875
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteTombBlk(writer), &lino, _exit);
20,469,190✔
876
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteFooter(writer), &lino, _exit);
20,465,994✔
877
  TAOS_CHECK_GOTO(tsdbSttFileDoUpdateHeader(writer), &lino, _exit);
20,468,278✔
878

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

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

883
  tsdbCloseFile(&writer->fd);
20,466,870✔
884

885
  STFileOp op = (STFileOp){
40,928,252✔
886
      .optype = TSDB_FOP_CREATE,
887
      .fid = writer->config->fid,
20,463,137✔
888
      .nf = writer->file[0],
889
  };
890
  tsdbTFileUpdVerRange(&op.nf, writer->ctx->range);
20,468,389✔
891

892
  TAOS_CHECK_GOTO(TARRAY2_APPEND(opArray, op), &lino, _exit);
40,928,153✔
893

894
_exit:
20,465,965✔
895
  if (code) {
20,466,195✔
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,466,195✔
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,256,413✔
911
  writer[0] = taosMemoryCalloc(1, sizeof(*writer[0]));
21,256,413✔
912
  if (writer[0] == NULL) {
21,243,293✔
913
    return terrno;
×
914
  }
915

916
  writer[0]->config[0] = config[0];
21,241,110✔
917
  writer[0]->ctx->opened = false;
21,243,954✔
918
  return 0;
21,264,043✔
919
}
920

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

924
  int32_t code = 0;
21,264,661✔
925
  int32_t lino = 0;
21,264,661✔
926

927
  if (writer[0]->ctx->opened) {
21,265,418✔
928
    if (abort) {
20,466,808✔
929
      TAOS_CHECK_GOTO(tsdbSttFWriterCloseAbort(writer[0]), &lino, _exit);
×
930
    } else {
931
      TAOS_CHECK_GOTO(tsdbSttFWriterCloseCommit(writer[0], opArray), &lino, _exit);
20,466,808✔
932
    }
933
    tsdbSttFWriterDoClose(writer[0]);
20,463,695✔
934
  }
935
  taosMemoryFree(writer[0]);
21,265,064✔
936
  writer[0] = NULL;
21,265,603✔
937

938
_exit:
21,266,138✔
939
  if (code) {
21,264,219✔
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,261,863✔
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,405,619✔
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);
23,899,908✔
956

957
    TAOS_CHECK_GOTO(tsdbUpdateSkmTb(writer->config->tsdb, (TABLEID *)row, writer->config->skmTb), &lino, _exit);
23,902,444✔
958

959
    TABLEID id = {
23,902,809✔
960
        .suid = row->suid,
23,904,476✔
961
        .uid = row->suid ? 0 : row->uid,
23,906,563✔
962
    };
963
    TAOS_CHECK_GOTO(tBlockDataInit(writer->blockData, &id, writer->config->skmTb->pTSchema, NULL, 0), &lino, _exit);
23,894,933✔
964
  }
965

966
  if (writer->ctx->tbid->uid != row->uid) {
2,147,483,647✔
967
    writer->ctx->tbid->suid = row->suid;
102,013,118✔
968
    writer->ctx->tbid->uid = row->uid;
102,022,426✔
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);
137,768✔
978
      continue;
137,768✔
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);
191,032,435✔
1001
  } else {
1002
    if (writer->blockData->nRow >= writer->config->maxRow) {
2,147,483,647✔
1003
      TAOS_CHECK_GOTO(tsdbSttFileDoWriteBlockData(writer), &lino, _exit);
20,778,229✔
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) {
20,943,006✔
1019
  int32_t code = 0;
20,943,006✔
1020
  int32_t lino = 0;
20,943,006✔
1021

1022
  SRowInfo row[1];
20,943,136✔
1023
  row->suid = bdata->suid;
20,943,541✔
1024
  for (int32_t i = 0; i < bdata->nRow; i++) {
167,989,338✔
1025
    row->uid = bdata->uid ? bdata->uid : bdata->aUid[i];
147,047,937✔
1026
    row->row = tsdbRowFromBlockData(bdata, i);
147,050,345✔
1027

1028
    TAOS_CHECK_GOTO(tsdbSttFileWriteRow(writer, row), &lino, _exit);
147,050,345✔
1029
  }
1030

1031
_exit:
20,942,471✔
1032
  if (code) {
20,943,541✔
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;
20,943,541✔
1037
}
1038

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

1043
  if (!writer->ctx->opened) {
6,930,723✔
1044
    TAOS_CHECK_GOTO(tsdbSttFWriterDoOpen(writer), &lino, _exit);
53,515✔
1045
  } else {
1046
    if (writer->blockData->nRow > 0) {
6,877,208✔
1047
      TAOS_CHECK_GOTO(tsdbSttFileDoWriteBlockData(writer), &lino, _exit);
1,667,571✔
1048
    }
1049

1050
    if (STATIS_BLOCK_SIZE(writer->staticBlock) > 0) {
6,877,208✔
1051
      TAOS_CHECK_GOTO(tsdbSttFileDoWriteStatisBlock(writer), &lino, _exit);
1,667,571✔
1052
    }
1053
  }
1054

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

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

1061
_exit:
6,930,723✔
1062
  if (code) {
6,930,723✔
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,930,723✔
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,930,723✔
1072
}
1073

1074
bool tsdbSttFileWriterIsOpened(SSttFileWriter *writer) { return writer->ctx->opened; }
4,417,933✔
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