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

taosdata / TDengine / #5060

17 May 2026 01:15AM UTC coverage: 73.425% (-0.02%) from 73.443%
#5060

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)

281800 of 383795 relevant lines covered (73.42%)

134332207.97 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) {
354,018,638✔
39
  int32_t code = 0;
354,018,638✔
40
  int32_t lino = 0;
354,018,638✔
41

42
  reader[0] = taosMemoryCalloc(1, sizeof(*reader[0]));
354,053,125✔
43
  if (reader[0] == NULL) {
353,837,271✔
44
    TAOS_CHECK_GOTO(terrno, &lino, _exit);
×
45
  }
46

47
  reader[0]->config[0] = config[0];
353,840,413✔
48
  reader[0]->buffers = config->buffers;
353,873,911✔
49
  if (reader[0]->buffers == NULL) {
354,036,246✔
50
    reader[0]->buffers = reader[0]->local;
353,989,245✔
51
  }
52

53
  // open file
54
  if (fname) {
353,994,452✔
55
    TAOS_CHECK_GOTO(tsdbOpenFile(fname, config->tsdb, TD_FILE_READ, &reader[0]->fd, 0), &lino, _exit);
353,994,452✔
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);
353,961,814✔
64
  if (offset < TSDB_FHDR_SIZE) {
353,915,985✔
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);
353,915,985✔
71

72
#if 1
73
  TAOS_CHECK_GOTO(
354,026,260✔
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:
353,995,718✔
95
  if (code) {
353,978,950✔
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;
353,978,950✔
101
}
102

103
void tsdbSttFileReaderClose(SSttFileReader **reader) {
354,075,748✔
104
  if (reader[0]) {
354,075,748✔
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);
354,058,365✔
109
    TARRAY2_DESTROY(reader[0]->tombBlkArray, NULL);
354,061,703✔
110
    TARRAY2_DESTROY(reader[0]->statisBlkArray, NULL);
354,030,639✔
111
    TARRAY2_DESTROY(reader[0]->sttBlkArray, NULL);
354,022,767✔
112
    taosMemoryFree(reader[0]);
354,044,551✔
113
    reader[0] = NULL;
354,060,677✔
114
  }
115
}
354,034,515✔
116

117
// SSttFSegReader
118
int32_t tsdbSttFileReadStatisBlk(SSttFileReader *reader, const TStatisBlkArray **statisBlkArray) {
350,586,777✔
119
  if (!reader->ctx->statisBlkLoaded) {
350,586,777✔
120
    if (reader->footer->statisBlkPtr->size > 0) {
350,623,790✔
121
      if (reader->footer->statisBlkPtr->size % sizeof(SStatisBlk) != 0) {
350,452,586✔
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);
350,367,937✔
128
      void   *data = taosMemoryMalloc(reader->footer->statisBlkPtr->size);
350,445,880✔
129
      if (!data) {
350,446,919✔
130
        return terrno;
×
131
      }
132

133
      SEncryptData *pEncryptData = &(reader->config->tsdb->pVnode->config.tsdbCfg.encryptData);
350,446,919✔
134

135
      int32_t code = tsdbReadFile(reader->fd, reader->footer->statisBlkPtr->offset, data,
350,497,332✔
136
                                  reader->footer->statisBlkPtr->size, 0, pEncryptData);
137
      if (code) {
350,499,351✔
138
        taosMemoryFree(data);
×
139
        return code;
×
140
      }
141

142
      TARRAY2_INIT_EX(reader->statisBlkArray, size, size, data);
350,499,351✔
143
    } else {
144
      TARRAY2_INIT(reader->statisBlkArray);
169,740✔
145
    }
146

147
    reader->ctx->statisBlkLoaded = true;
350,668,695✔
148
  }
149

150
  statisBlkArray[0] = reader->statisBlkArray;
350,643,435✔
151
  return 0;
350,665,971✔
152
}
153

154
int32_t tsdbSttFileReadTombBlk(SSttFileReader *reader, const TTombBlkArray **tombBlkArray) {
353,958,912✔
155
  if (!reader->ctx->tombBlkLoaded) {
353,958,912✔
156
    if (reader->footer->tombBlkPtr->size > 0) {
354,008,276✔
157
      if (reader->footer->tombBlkPtr->size % sizeof(STombBlk) != 0) {
13,187,848✔
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,187,848✔
164
      void   *data = taosMemoryMalloc(reader->footer->tombBlkPtr->size);
13,187,114✔
165
      if (!data) {
13,187,114✔
166
        return terrno;
×
167
      }
168

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

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

178
      TARRAY2_INIT_EX(reader->tombBlkArray, size, size, data);
13,187,114✔
179
    } else {
180
      TARRAY2_INIT(reader->tombBlkArray);
340,757,230✔
181
    }
182

183
    reader->ctx->tombBlkLoaded = true;
353,981,326✔
184
  }
185

186
  tombBlkArray[0] = reader->tombBlkArray;
353,946,149✔
187
  return 0;
353,953,862✔
188
}
189

190
int32_t tsdbSttFileReadSttBlk(SSttFileReader *reader, const TSttBlkArray **sttBlkArray) {
353,956,482✔
191
  if (!reader->ctx->sttBlkLoaded) {
353,956,482✔
192
    if (reader->footer->sttBlkPtr->size > 0) {
354,022,922✔
193
      if (reader->footer->sttBlkPtr->size % sizeof(SSttBlk) != 0) {
353,789,620✔
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);
353,654,519✔
200
      void   *data = taosMemoryMalloc(reader->footer->sttBlkPtr->size);
353,790,518✔
201
      if (!data) {
353,727,350✔
202
        return terrno;
×
203
      }
204

205
      SEncryptData *pEncryptData = &(reader->config->tsdb->pVnode->config.tsdbCfg.encryptData);
353,727,350✔
206

207
      int32_t code = tsdbReadFile(reader->fd, reader->footer->sttBlkPtr->offset, data, reader->footer->sttBlkPtr->size,
353,825,359✔
208
                                  0, pEncryptData);
209
      if (code) {
353,828,167✔
210
        taosMemoryFree(data);
×
211
        return code;
×
212
      }
213

214
      TARRAY2_INIT_EX(reader->sttBlkArray, size, size, data);
353,828,167✔
215
    } else {
216
      TARRAY2_INIT(reader->sttBlkArray);
223,752✔
217
    }
218

219
    reader->ctx->sttBlkLoaded = true;
354,017,923✔
220
  }
221

222
  sttBlkArray[0] = reader->sttBlkArray;
354,035,216✔
223
  return 0;
354,026,649✔
224
}
225

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

230
  SBuffer *buffer0 = reader->buffers + 0;
19,124,624✔
231
  SBuffer *assist = reader->buffers + 1;
19,125,351✔
232

233
  SEncryptData *pEncryptData = &(reader->config->tsdb->pVnode->config.tsdbCfg.encryptData);
19,124,624✔
234

235
  // load data
236
  tBufferClear(buffer0);
237
  TAOS_CHECK_GOTO(
19,124,624✔
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);
19,125,351✔
242
  TAOS_CHECK_GOTO(tBlockDataDecompress(&br, bData, assist), &lino, _exit);
19,125,351✔
243

244
_exit:
19,124,624✔
245
  if (code) {
19,125,351✔
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;
19,125,351✔
250
}
251

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

257
  SDiskDataHdr hdr;
229,123,166✔
258
  SBuffer     *buffer0 = reader->buffers + 0;
229,103,871✔
259
  SBuffer     *buffer1 = reader->buffers + 1;
229,081,567✔
260
  SBuffer     *assist = reader->buffers + 2;
229,108,706✔
261

262
  SEncryptData *pEncryptData = &(reader->config->tsdb->pVnode->config.tsdbCfg.encryptData);
229,097,790✔
263

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

269
  // decode header
270
  SBufferReader br = BUFFER_READER_INITIALIZER(0, buffer0);
229,179,621✔
271
  TAOS_CHECK_GOTO(tGetDiskDataHdr(&br, &hdr), &lino, _exit);
229,173,132✔
272

273
  if (hdr.delimiter != TSDB_FILE_DLMT) {
229,047,744✔
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);
229,047,744✔
281
  bData->suid = hdr.suid;
229,089,332✔
282
  bData->uid = (sttBlk->suid == 0) ? sttBlk->minUid : 0;
229,087,251✔
283
  bData->nRow = hdr.nRow;
229,072,747✔
284

285
  // key part
286
  TAOS_CHECK_GOTO(tBlockDataDecompressKeyPart(&hdr, &br, bData, assist), &lino, _exit);
229,023,993✔
287
  if (br.offset != buffer0->size) {
229,131,580✔
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;
229,117,942✔
295
  for (int i = 0; i < ncid; i++) {
244,040,644✔
296
    if (tBlockDataGetColData(bData, cids[i]) == NULL) {
204,241,248✔
297
      loadExtra = true;
189,323,537✔
298
      break;
189,323,537✔
299
    }
300
  }
301

302
  if (!loadExtra) {
229,122,933✔
303
    goto _exit;
39,809,997✔
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,
189,328,172✔
309
                                       pEncryptData),
310
                  &lino, _exit);
311

312
  // load each column
313
  SBlockCol blockCol = {
189,360,070✔
314
      .cid = 0,
315
  };
316
  br = BUFFER_READER_INITIALIZER(0, buffer0);
189,367,458✔
317
  for (int32_t i = 0; i < ncid; i++) {
748,593,221✔
318
    int16_t cid = cids[i];
559,168,147✔
319

320
    if (tBlockDataGetColData(bData, cid)) {  // already loaded
559,194,028✔
321
      continue;
9,386,988✔
322
    }
323

324
    while (cid > blockCol.cid) {
1,892,468,922✔
325
      if (br.offset >= buffer0->size) {
1,347,646,568✔
326
        blockCol.cid = INT16_MAX;
5,099,581✔
327
        break;
5,099,581✔
328
      }
329

330
      TAOS_CHECK_GOTO(tGetBlockCol(&br, &blockCol, hdr.fmtVer, hdr.cmprAlg), &lino, _exit);
1,342,553,535✔
331
    }
332

333
    if (cid < blockCol.cid) {
549,921,935✔
334
      const STColumn *tcol = tTSchemaSearchColumn(pTSchema, cid);
25,121,091✔
335
      TSDB_CHECK_NULL(tcol, code, lino, _exit, TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER);
25,121,091✔
336
      SBlockCol none = {
25,121,091✔
337
          .cid = cid,
338
          .type = tcol->type,
25,121,091✔
339
          .cflag = tcol->flags,
25,121,091✔
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);
25,121,091✔
348
    } else if (cid == blockCol.cid) {
524,800,844✔
349
      // load from file
350
      tBufferClear(buffer1);
351
      TAOS_CHECK_GOTO(
524,722,486✔
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);
524,766,459✔
358
      TAOS_CHECK_GOTO(tBlockDataDecompressColData(&hdr, &blockCol, &br1, bData, assist), &lino, _exit);
524,769,358✔
359
    }
360
  }
361

362
_exit:
229,319,627✔
363
  if (code) {
229,138,711✔
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;
229,138,711✔
368
}
369

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

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

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

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

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

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

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

418
  if (statisBlk->numOfPKs > TD_MAX_PK_COLS) {
283,479,020✔
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;
283,400,164✔
426
  SBuffer *assist = reader->buffers + 1;
283,440,316✔
427

428
  SEncryptData *pEncryptData = &(reader->config->tsdb->pVnode->config.tsdbCfg.encryptData);
283,524,133✔
429

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

436
  // decode data
437
  tStatisBlockClear(statisBlock);
283,498,694✔
438
  statisBlock->numOfPKs = statisBlk->numOfPKs;
283,437,484✔
439
  statisBlock->numOfRecords = statisBlk->numRec;
283,463,261✔
440
  SBufferReader br = BUFFER_READER_INITIALIZER(0, buffer0);
283,477,495✔
441
  for (int32_t i = 0; i < ARRAY_SIZE(statisBlock->buffers); ++i) {
1,701,173,820✔
442
    SCompressInfo info = {
1,417,613,813✔
443
        .dataType = TSDB_DATA_TYPE_BIGINT,
444
        .cmprAlg = statisBlk->cmprAlg,
1,417,597,199✔
445
        .compressedSize = statisBlk->size[i],
1,417,607,715✔
446
        .originalSize = statisBlk->numRec * sizeof(int64_t),
1,417,629,185✔
447
    };
448

449
    TAOS_CHECK_GOTO(tDecompressDataToBuffer(BR_PTR(&br), &info, &statisBlock->buffers[i], assist), &lino, _exit);
1,417,621,622✔
450
    br.offset += statisBlk->size[i];
1,417,641,194✔
451
  }
452

453
  if (statisBlk->numOfPKs > 0) {
283,560,007✔
454
    SValueColumnCompressInfo firstKeyInfos[TD_MAX_PK_COLS];
58,467,975✔
455
    SValueColumnCompressInfo lastKeyInfos[TD_MAX_PK_COLS];
58,467,241✔
456

457
    // decode compress info
458
    for (int32_t i = 0; i < statisBlk->numOfPKs; i++) {
116,935,216✔
459
      TAOS_CHECK_GOTO(tValueColumnCompressInfoDecode(&br, &firstKeyInfos[i]), &lino, _exit);
58,467,241✔
460
    }
461

462
    for (int32_t i = 0; i < statisBlk->numOfPKs; i++) {
116,935,216✔
463
      TAOS_CHECK_GOTO(tValueColumnCompressInfoDecode(&br, &lastKeyInfos[i]), &lino, _exit);
58,467,241✔
464
    }
465

466
    // decode value columns
467
    for (int32_t i = 0; i < statisBlk->numOfPKs; i++) {
116,933,014✔
468
      TAOS_CHECK_GOTO(tValueColumnDecompress(BR_PTR(&br), firstKeyInfos + i, &statisBlock->firstKeyPKs[i], assist),
58,467,975✔
469
                      &lino, _exit);
470
      br.offset += (firstKeyInfos[i].dataCompressedSize + firstKeyInfos[i].offsetCompressedSize);
58,467,241✔
471
    }
472

473
    for (int32_t i = 0; i < statisBlk->numOfPKs; i++) {
116,934,482✔
474
      TAOS_CHECK_GOTO(tValueColumnDecompress(BR_PTR(&br), &lastKeyInfos[i], &statisBlock->lastKeyPKs[i], assist), &lino,
58,467,241✔
475
                      _exit);
476
      br.offset += (lastKeyInfos[i].dataCompressedSize + lastKeyInfos[i].offsetCompressedSize);
58,467,975✔
477
    }
478
  }
479

480
  if (br.offset != buffer0->size) {
283,545,813✔
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:
283,508,995✔
488
  if (code) {
283,524,148✔
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;
283,524,148✔
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,926,754✔
524
                                           int64_t *fileSize, TSttBlkArray *sttBlkArray, SBuffer *buffers,
525
                                           SVersionRange *range, SEncryptData *encryptData) {
526
  if (blockData->nRow == 0) return 0;
48,926,754✔
527

528
  int32_t code = 0;
48,927,918✔
529

530
  SSttBlk sttBlk[1] = {{
48,931,466✔
531
      .suid = blockData->suid,
48,927,071✔
532
      .minUid = blockData->uid ? blockData->uid : blockData->aUid[0],
48,929,277✔
533
      .maxUid = blockData->uid ? blockData->uid : blockData->aUid[blockData->nRow - 1],
48,934,047✔
534
      .minKey = blockData->aTSKEY[0],
48,934,993✔
535
      .maxKey = blockData->aTSKEY[0],
48,930,982✔
536
      .minVer = blockData->aVersion[0],
48,934,108✔
537
      .maxVer = blockData->aVersion[0],
48,931,971✔
538
      .nRow = blockData->nRow,
48,932,119✔
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,936,646✔
549
  TAOS_CHECK_RETURN(tBlockDataCompress(blockData, info, buffers, buffers + 4));
48,932,651✔
550

551
  sttBlk->bInfo.offset = *fileSize;
48,930,508✔
552
  sttBlk->bInfo.szKey = buffers[0].size + buffers[1].size;
48,929,889✔
553
  sttBlk->bInfo.szBlock = buffers[2].size + buffers[3].size + sttBlk->bInfo.szKey;
48,934,128✔
554
  for (int i = 0; i < 4; i++) {
244,673,860✔
555
    if (buffers[i].size) {
195,733,573✔
556
      TAOS_CHECK_RETURN(tsdbWriteFile(fd, *fileSize, buffers[i].data, buffers[i].size, encryptData));
195,482,317✔
557
      *fileSize += buffers[i].size;
195,479,787✔
558
    }
559
  }
560

561
  TAOS_CHECK_RETURN(TARRAY2_APPEND_PTR(sttBlkArray, sttBlk));
97,877,784✔
562

563
  tBlockDataClear(blockData);
48,937,497✔
564
  return 0;
48,939,677✔
565
}
566

567
static int32_t tsdbSttFileDoWriteBlockData(SSttFileWriter *writer) {
71,253,118✔
568
  if (writer->blockData->nRow == 0) return 0;
71,253,118✔
569

570
  int32_t code = 0;
48,937,900✔
571
  int32_t lino = 0;
48,937,900✔
572

573
  tb_uid_t         uid = writer->blockData->suid == 0 ? writer->blockData->uid : writer->blockData->suid;
48,937,837✔
574
  SColCompressInfo info = {.defaultCmprAlg = writer->config->cmprAlg, .pColCmpr = NULL};
48,934,830✔
575
  code = metaGetColCmpr(writer->config->tsdb->pVnode->pMeta, uid, &(info.pColCmpr));
48,936,812✔
576

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

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

583
_exit:
48,935,424✔
584
  if (code) {
48,935,254✔
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,935,254✔
589
  return code;
48,932,998✔
590
}
591

592
static int32_t tsdbSttFileDoWriteStatisBlock(SSttFileWriter *writer) {
22,637,971✔
593
  if (writer->staticBlock->numOfRecords == 0) {
22,637,971✔
594
    return 0;
1,514,792✔
595
  }
596

597
  int32_t code = 0;
21,125,231✔
598
  int32_t lino = 0;
21,125,231✔
599

600
  SBuffer *buffer0 = writer->buffers + 0;
21,124,222✔
601
  SBuffer *buffer1 = writer->buffers + 1;
21,125,824✔
602
  SBuffer *assist = writer->buffers + 2;
21,126,542✔
603

604
  STbStatisRecord record;
21,123,977✔
605
  STbStatisBlock *statisBlock = writer->staticBlock;
21,125,024✔
606
  SStatisBlk      statisBlk = {0};
21,126,560✔
607

608
  statisBlk.dp->offset = writer->file->size;
21,126,804✔
609
  statisBlk.dp->size = 0;
21,124,422✔
610
  statisBlk.numRec = statisBlock->numOfRecords;
21,124,422✔
611
  statisBlk.cmprAlg = writer->config->cmprAlg;
21,123,782✔
612
  statisBlk.numOfPKs = statisBlock->numOfPKs;
21,125,330✔
613

614
  code = tStatisBlockGet(statisBlock, 0, &record);
21,122,439✔
615
  TSDB_CHECK_CODE(code, lino, _exit);
21,125,770✔
616
  statisBlk.minTbid.suid = record.suid;
21,125,770✔
617
  statisBlk.minTbid.uid = record.uid;
21,125,770✔
618

619
  code = tStatisBlockGet(statisBlock, statisBlock->numOfRecords - 1, &record);
21,125,770✔
620
  TSDB_CHECK_CODE(code, lino, _exit);
21,126,036✔
621
  statisBlk.maxTbid.suid = record.suid;
21,126,036✔
622
  statisBlk.maxTbid.uid = record.uid;
21,126,036✔
623

624
  SEncryptData *pEncryptData = &(writer->config->tsdb->pVnode->config.tsdbCfg.encryptData);
21,126,036✔
625

626
  // compress each column
627
  for (int32_t i = 0; i < ARRAY_SIZE(statisBlk.size); i++) {
126,761,342✔
628
    SCompressInfo info = {
105,632,823✔
629
        .dataType = TSDB_DATA_TYPE_BIGINT,
630
        .cmprAlg = statisBlk.cmprAlg,
105,630,644✔
631
        .originalSize = statisBlock->buffers[i].size,
105,630,644✔
632
    };
633

634
    tBufferClear(buffer0);
635
    TAOS_CHECK_GOTO(tCompressDataToBuffer(statisBlock->buffers[i].data, &info, buffer0, assist), &lino, _exit);
105,633,142✔
636
    TAOS_CHECK_GOTO(tsdbWriteFile(writer->fd, writer->file->size, buffer0->data, info.compressedSize, pEncryptData),
105,631,714✔
637
                    &lino, _exit);
638

639
    statisBlk.size[i] = info.compressedSize;
105,641,408✔
640
    statisBlk.dp->size += info.compressedSize;
105,640,209✔
641
    writer->file->size += info.compressedSize;
105,640,209✔
642
  }
643

644
  // compress primary keys
645
  if (statisBlk.numOfPKs > 0) {
21,128,519✔
646
    SValueColumnCompressInfo compressInfo = {.cmprAlg = statisBlk.cmprAlg};
307,584✔
647

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

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

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

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

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

672
  TAOS_CHECK_GOTO(TARRAY2_APPEND_PTR(writer->statisBlkArray, &statisBlk), &lino, _exit);
42,255,342✔
673

674
  tStatisBlockClear(writer->staticBlock);
21,126,823✔
675

676
_exit:
21,122,582✔
677
  if (code) {
21,125,532✔
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;
21,122,319✔
682
}
683

684
static int32_t tsdbSttFileDoWriteTombBlock(SSttFileWriter *writer) {
20,862,805✔
685
  if (TOMB_BLOCK_SIZE(writer->tombBlock) == 0) {
20,862,805✔
686
    return 0;
19,132,292✔
687
  }
688

689
  int32_t code = 0;
1,733,839✔
690
  int32_t lino = 0;
1,733,839✔
691

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

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

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

706
int32_t tsdbFileWriteSttBlk(STsdbFD *fd, const TSttBlkArray *sttBlkArray, SFDataPtr *ptr, int64_t *fileSize,
20,862,677✔
707
                            SEncryptData *encryptData) {
708
  ptr->size = TARRAY2_DATA_LEN(sttBlkArray);
20,862,677✔
709
  if (ptr->size > 0) {
20,864,760✔
710
    ptr->offset = *fileSize;
20,810,298✔
711

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

714
    *fileSize += ptr->size;
20,811,977✔
715
  }
716
  return 0;
20,867,493✔
717
}
718

719
static int32_t tsdbSttFileDoWriteSttBlk(SSttFileWriter *writer) {
20,860,489✔
720
  int32_t code = 0;
20,860,489✔
721
  int32_t lino;
20,859,978✔
722

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

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

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

737
static int32_t tsdbSttFileDoWriteStatisBlk(SSttFileWriter *writer) {
20,866,169✔
738
  int32_t code = 0;
20,866,169✔
739
  int32_t lino;
20,865,658✔
740
  SEncryptData *pEncryptData = &(writer->config->tsdb->pVnode->config.tsdbCfg.encryptData);
20,866,431✔
741

742
  writer->footer->statisBlkPtr->size = TARRAY2_DATA_LEN(writer->statisBlkArray);
20,866,660✔
743
  if (writer->footer->statisBlkPtr->size) {
20,867,575✔
744
    writer->footer->statisBlkPtr->offset = writer->file->size;
20,811,925✔
745
    TAOS_CHECK_GOTO(tsdbWriteFile(writer->fd, writer->file->size, (const uint8_t *)TARRAY2_DATA(writer->statisBlkArray),
20,809,562✔
746
                                  writer->footer->statisBlkPtr->size, pEncryptData),
747
                    &lino, _exit);
748
    writer->file->size += writer->footer->statisBlkPtr->size;
20,813,614✔
749
  }
750

751
_exit:
20,867,440✔
752
  if (code) {
20,864,701✔
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,864,701✔
757
}
758

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

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

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

769
_exit:
20,867,644✔
770
  if (code) {
20,867,644✔
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,867,644✔
775
}
776

777
int32_t tsdbFileWriteSttFooter(STsdbFD *fd, const SSttFooter *footer, int64_t *fileSize, SEncryptData *encryptData) {
20,862,314✔
778
  TAOS_CHECK_RETURN(tsdbWriteFile(fd, *fileSize, (const uint8_t *)footer, sizeof(*footer), encryptData));
20,862,314✔
779
  *fileSize += sizeof(*footer);
20,868,249✔
780
  return 0;
20,864,638✔
781
}
782

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

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

789
static int32_t tsdbSttFWriterDoOpen(SSttFileWriter *writer) {
20,831,395✔
790
  int32_t code = 0;
20,831,395✔
791
  int32_t lino = 0;
20,831,395✔
792
  STsdb  *tsdb = writer->config->tsdb;
20,837,110✔
793

794
  // set
795
  if (!writer->config->skmTb) writer->config->skmTb = writer->skmTb;
20,847,405✔
796
  if (!writer->config->skmRow) writer->config->skmRow = writer->skmRow;
20,845,973✔
797
  writer->buffers = writer->config->buffers;
20,846,665✔
798
  if (writer->buffers == NULL) {
20,848,811✔
799
    writer->buffers = writer->local;
×
800
  }
801

802
  // alloc disk id
803
  SDiskID diskId = {0};
20,846,059✔
804
  code = tsdbAllocateDisk(tsdb, tsdbFTypeLabel(TSDB_FTYPE_STT), writer->config->expLevel, &diskId);
20,840,147✔
805
  TSDB_CHECK_CODE(code, lino, _exit);
20,845,508✔
806

807
  writer->file[0] = (STFile){
20,848,051✔
808
      .type = TSDB_FTYPE_STT,
809
      .did = diskId,
810
      .fid = writer->config->fid,
20,845,508✔
811
      .cid = writer->config->cid,
20,853,431✔
812
      .size = 0,
813
      .minVer = VERSION_MAX,
814
      .maxVer = VERSION_MIN,
815
      .stt[0] =
816
          {
817
              .level = writer->config->level,
20,852,451✔
818
          },
819
  };
820

821
  // open file
822
  int32_t flag = TD_FILE_READ | TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
20,846,327✔
823
  char    fname[TSDB_FILENAME_LEN];
20,845,816✔
824

825
  tsdbTFileName(writer->config->tsdb, writer->file, fname);
20,848,825✔
826
  TAOS_CHECK_GOTO(tsdbOpenFile(fname, writer->config->tsdb, flag, &writer->fd, 0), &lino, _exit);
20,843,155✔
827

828
  uint8_t hdr[TSDB_FHDR_SIZE] = {0};
20,841,635✔
829
  SEncryptData *pEncryptData = &(writer->config->tsdb->pVnode->config.tsdbCfg.encryptData);
20,838,211✔
830

831
  TAOS_CHECK_GOTO(tsdbWriteFile(writer->fd, 0, hdr, sizeof(hdr), pEncryptData), &lino, _exit);
20,851,813✔
832
  writer->file->size += sizeof(hdr);
20,853,292✔
833

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

837
  writer->ctx->opened = true;
20,855,893✔
838

839
_exit:
20,856,752✔
840
  if (code) {
20,857,158✔
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,857,158✔
845
}
846

847
static void tsdbSttFWriterDoClose(SSttFileWriter *writer) {
20,853,606✔
848
  for (int32_t i = 0; i < ARRAY_SIZE(writer->local); ++i) {
229,102,134✔
849
    tBufferDestroy(writer->local + i);
208,274,562✔
850
  }
851
  tDestroyTSchema(writer->skmRow->pTSchema);
20,827,572✔
852
  tDestroyTSchema(writer->skmTb->pTSchema);
20,854,314✔
853
  tTombBlockDestroy(writer->tombBlock);
20,859,561✔
854
  tStatisBlockDestroy(writer->staticBlock);
20,850,590✔
855
  tBlockDataDestroy(writer->blockData);
20,859,687✔
856
  TARRAY2_DESTROY(writer->tombBlkArray, NULL);
20,868,147✔
857
  TARRAY2_DESTROY(writer->statisBlkArray, NULL);
20,867,752✔
858
  TARRAY2_DESTROY(writer->sttBlkArray, NULL);
20,861,797✔
859
}
20,864,520✔
860

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

866
static int32_t tsdbSttFWriterCloseCommit(SSttFileWriter *writer, TFileOpArray *opArray) {
20,863,412✔
867
  int32_t lino;
20,862,901✔
868
  int32_t code;
869

870
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteBlockData(writer), &lino, _exit);
20,862,884✔
871
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteStatisBlock(writer), &lino, _exit);
20,867,019✔
872
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteTombBlock(writer), &lino, _exit);
20,864,901✔
873
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteSttBlk(writer), &lino, _exit);
20,862,393✔
874
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteStatisBlk(writer), &lino, _exit);
20,862,683✔
875
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteTombBlk(writer), &lino, _exit);
20,862,223✔
876
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteFooter(writer), &lino, _exit);
20,863,333✔
877
  TAOS_CHECK_GOTO(tsdbSttFileDoUpdateHeader(writer), &lino, _exit);
20,864,173✔
878

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

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

883
  tsdbCloseFile(&writer->fd);
20,866,493✔
884

885
  STFileOp op = (STFileOp){
41,722,171✔
886
      .optype = TSDB_FOP_CREATE,
887
      .fid = writer->config->fid,
20,859,820✔
888
      .nf = writer->file[0],
889
  };
890
  tsdbTFileUpdVerRange(&op.nf, writer->ctx->range);
20,864,764✔
891

892
  TAOS_CHECK_GOTO(TARRAY2_APPEND(opArray, op), &lino, _exit);
41,702,534✔
893

894
_exit:
20,844,912✔
895
  if (code) {
20,844,861✔
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,844,861✔
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,722,359✔
911
  writer[0] = taosMemoryCalloc(1, sizeof(*writer[0]));
21,722,359✔
912
  if (writer[0] == NULL) {
21,704,736✔
913
    return terrno;
×
914
  }
915

916
  writer[0]->config[0] = config[0];
21,709,969✔
917
  writer[0]->ctx->opened = false;
21,712,411✔
918
  return 0;
21,721,080✔
919
}
920

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

924
  int32_t code = 0;
21,740,161✔
925
  int32_t lino = 0;
21,740,161✔
926

927
  if (writer[0]->ctx->opened) {
21,740,472✔
928
    if (abort) {
20,864,774✔
929
      TAOS_CHECK_GOTO(tsdbSttFWriterCloseAbort(writer[0]), &lino, _exit);
×
930
    } else {
931
      TAOS_CHECK_GOTO(tsdbSttFWriterCloseCommit(writer[0], opArray), &lino, _exit);
20,864,774✔
932
    }
933
    tsdbSttFWriterDoClose(writer[0]);
20,848,368✔
934
  }
935
  taosMemoryFree(writer[0]);
21,739,032✔
936
  writer[0] = NULL;
21,732,113✔
937

938
_exit:
21,733,627✔
939
  if (code) {
21,735,063✔
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,731,723✔
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,798,613✔
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,444,241✔
956

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

959
    TABLEID id = {
27,438,151✔
960
        .suid = row->suid,
27,439,019✔
961
        .uid = row->suid ? 0 : row->uid,
27,439,855✔
962
    };
963
    TAOS_CHECK_GOTO(tBlockDataInit(writer->blockData, &id, writer->config->skmTb->pTSchema, NULL, 0), &lino, _exit);
27,429,833✔
964
  }
965

966
  if (writer->ctx->tbid->uid != row->uid) {
2,147,483,647✔
967
    writer->ctx->tbid->suid = row->suid;
108,440,050✔
968
    writer->ctx->tbid->uid = row->uid;
108,450,625✔
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);
95,951✔
978
      continue;
92,501✔
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);
199,615,323✔
1001
  } else {
1002
    if (writer->blockData->nRow >= writer->config->maxRow) {
2,147,483,647✔
1003
      TAOS_CHECK_GOTO(tsdbSttFileDoWriteBlockData(writer), &lino, _exit);
21,273,155✔
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) {
21,264,406✔
1019
  int32_t code = 0;
21,264,406✔
1020
  int32_t lino = 0;
21,264,406✔
1021

1022
  SRowInfo row[1];
21,264,432✔
1023
  row->suid = bdata->suid;
21,266,550✔
1024
  for (int32_t i = 0; i < bdata->nRow; i++) {
170,401,745✔
1025
    row->uid = bdata->uid ? bdata->uid : bdata->aUid[i];
149,132,557✔
1026
    row->row = tsdbRowFromBlockData(bdata, i);
149,132,515✔
1027

1028
    TAOS_CHECK_GOTO(tsdbSttFileWriteRow(writer, row), &lino, _exit);
149,132,515✔
1029
  }
1030

1031
_exit:
21,266,797✔
1032
  if (code) {
21,266,797✔
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;
21,266,797✔
1037
}
1038

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

1043
  if (!writer->ctx->opened) {
6,961,779✔
1044
    TAOS_CHECK_GOTO(tsdbSttFWriterDoOpen(writer), &lino, _exit);
54,012✔
1045
  } else {
1046
    if (writer->blockData->nRow > 0) {
6,907,767✔
1047
      TAOS_CHECK_GOTO(tsdbSttFileDoWriteBlockData(writer), &lino, _exit);
1,679,827✔
1048
    }
1049

1050
    if (STATIS_BLOCK_SIZE(writer->staticBlock) > 0) {
6,907,767✔
1051
      TAOS_CHECK_GOTO(tsdbSttFileDoWriteStatisBlock(writer), &lino, _exit);
1,679,827✔
1052
    }
1053
  }
1054

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

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

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

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

© 2026 Coveralls, Inc