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

taosdata / TDengine / #4380

25 Jun 2025 06:58AM UTC coverage: 62.307% (-0.09%) from 62.393%
#4380

push

travis-ci

web-flow
feat(mqtt): mqtt subscription (#30127)

* feat(mqtt): Initial commit for mqtt

* chore(xnode/mnd): xnode message handlers for mnode

* chore(mnd/xnode): mnode part for xnode

* chore(xnode/translater): fix show commands

* fix(ast/creater): fix xnode create option

* fix(xnode/ci): fix ci & doc's error codes

* chore(xnode/sql): make create/drop/show work properly

* fix(xnode/sql): commit new files

* fix(xnode/sql): commit cmake files

* fix: fix testing cases

* fix(xnode/tsc): fix tokens

* fix(ast/anode): fix anode update decl.

* fix(xnode/error): fix xnode error codes

* fix: xnode make/destroy

* chore: xnode with option & dnode id

* chore: use taosmqtt for xnode

* chore: new error code for xnode launching

* chore(xnode): new error code

* chore: header for _xnode_mgmt_mqtt

* chore: source for _xnode_mgmt_mqtt

* chore: remove test directory from cmake

* chore: remove taosmqtt for ci to compile

* chore: remove taosudf header from xnode

* chore: new window macro

* chore: remove xnode mgmt mqtt for windows compilation

* Revert "chore: remove xnode mgmt mqtt for windows compilation"

This reverts commit 197e1640c.

* chore: cleanup code

* chore: xnode mgmt comment windows part out

* chore: mgmt/mqtt, move uv head toppest

* xnode/mnode: create xnode once per dnode

* fix(xnode/systable/test): fix column count

* xnode/sdb: renumber sdb type for xnode to make start/stop order correct

* xnode/mqtt: new param mqttPort

* fix SXnode's struct type

* transfer dnode id to mqtt subscription

* tmqtt: remove uv_a linking

* tmqtt/tools: sources for tools

* tools: fix windows compilation

* tools/producer: fix windows sleep param

* tools/producer: fix uninited var rc

* make tools only for linux

* test/mnodes: wail 1 or 2 seconds for offline to be leader

* update topic producer tool for geometry data type testing

* format tool sql statements

* show xnodes' ep

* make shell auto complete xnodes

* use usleep... (continued)

156642 of 320746 branches covered (48.84%)

Branch coverage included in aggregate %.

61 of 1020 new or added lines in 21 files covered. (5.98%)

1736 existing lines in 172 files now uncovered.

242538 of 319922 relevant lines covered (75.81%)

6277604.48 hits per line

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

76.52
/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) {
3,483,162✔
39
  int32_t code = 0;
3,483,162✔
40
  int32_t lino = 0;
3,483,162✔
41

42
  reader[0] = taosMemoryCalloc(1, sizeof(*reader[0]));
3,483,162!
43
  if (reader[0] == NULL) {
3,484,284!
44
    TAOS_CHECK_GOTO(terrno, &lino, _exit);
×
45
  }
46

47
  reader[0]->config[0] = config[0];
3,484,284✔
48
  reader[0]->buffers = config->buffers;
3,484,284✔
49
  if (reader[0]->buffers == NULL) {
3,484,284✔
50
    reader[0]->buffers = reader[0]->local;
3,484,195✔
51
  }
52

53
  // open file
54
  if (fname) {
3,484,284!
55
    TAOS_CHECK_GOTO(tsdbOpenFile(fname, config->tsdb, TD_FILE_READ, &reader[0]->fd, 0), &lino, _exit);
3,484,284!
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);
3,484,464✔
64
  if (offset < TSDB_FHDR_SIZE) {
3,484,464!
65
    TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
×
66
  }
67

68
  int32_t encryptAlgoirthm = config->tsdb->pVnode->config.tsdbCfg.encryptAlgorithm;
3,484,464✔
69
  char   *encryptKey = config->tsdb->pVnode->config.tsdbCfg.encryptKey;
3,484,464✔
70
#if 1
71
  TAOS_CHECK_GOTO(tsdbReadFile(reader[0]->fd, offset, (uint8_t *)(reader[0]->footer), sizeof(SSttFooter), 0,
3,484,464!
72
                               encryptAlgoirthm, encryptKey),
73
                  &lino, _exit);
74
#else
75
  int64_t size = config->file->size;
76

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

92
_exit:
3,480,704✔
93
  if (code) {
3,480,704!
94
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(config->tsdb->pVnode), __func__, __FILE__, lino,
×
95
              tstrerror(code));
96
    tsdbSttFileReaderClose(reader);
×
97
  }
98
  return code;
3,482,335✔
99
}
100

101
void tsdbSttFileReaderClose(SSttFileReader **reader) {
3,477,514✔
102
  if (reader[0]) {
3,477,514!
103
    for (int32_t i = 0; i < ARRAY_SIZE(reader[0]->local); ++i) {
38,222,030✔
104
      tBufferDestroy(reader[0]->local + i);
34,742,921✔
105
    }
106
    tsdbCloseFile(&reader[0]->fd);
3,479,109✔
107
    TARRAY2_DESTROY(reader[0]->tombBlkArray, NULL);
3,488,004!
108
    TARRAY2_DESTROY(reader[0]->statisBlkArray, NULL);
3,488,004!
109
    TARRAY2_DESTROY(reader[0]->sttBlkArray, NULL);
3,487,045!
110
    taosMemoryFree(reader[0]);
3,486,645✔
111
    reader[0] = NULL;
3,488,072✔
112
  }
113
}
3,487,437✔
114

115
// SSttFSegReader
116
int32_t tsdbSttFileReadStatisBlk(SSttFileReader *reader, const TStatisBlkArray **statisBlkArray) {
3,470,778✔
117
  if (!reader->ctx->statisBlkLoaded) {
3,470,778!
118
    if (reader->footer->statisBlkPtr->size > 0) {
3,472,822!
119
      if (reader->footer->statisBlkPtr->size % sizeof(SStatisBlk) != 0) {
3,474,449!
120
        return TSDB_CODE_FILE_CORRUPTED;
×
121
      }
122

123
      int32_t size = reader->footer->statisBlkPtr->size / sizeof(SStatisBlk);
3,474,449✔
124
      void   *data = taosMemoryMalloc(reader->footer->statisBlkPtr->size);
3,474,449✔
125
      if (!data) {
3,474,038!
126
        return terrno;
×
127
      }
128

129
      int32_t encryptAlgorithm = reader->config->tsdb->pVnode->config.tsdbCfg.encryptAlgorithm;
3,474,038✔
130
      char   *encryptKey = reader->config->tsdb->pVnode->config.tsdbCfg.encryptKey;
3,474,038✔
131
      int32_t code = tsdbReadFile(reader->fd, reader->footer->statisBlkPtr->offset, data,
3,474,038✔
132
                                  reader->footer->statisBlkPtr->size, 0, encryptAlgorithm, encryptKey);
133
      if (code) {
3,476,450✔
134
        taosMemoryFree(data);
263!
135
        return code;
×
136
      }
137

138
      TARRAY2_INIT_EX(reader->statisBlkArray, size, size, data);
3,476,187✔
139
    } else {
UNCOV
140
      TARRAY2_INIT(reader->statisBlkArray);
×
141
    }
142

143
    reader->ctx->statisBlkLoaded = true;
3,474,560✔
144
  }
145

146
  statisBlkArray[0] = reader->statisBlkArray;
3,472,516✔
147
  return 0;
3,472,516✔
148
}
149

150
int32_t tsdbSttFileReadTombBlk(SSttFileReader *reader, const TTombBlkArray **tombBlkArray) {
3,479,541✔
151
  if (!reader->ctx->tombBlkLoaded) {
3,479,541!
152
    if (reader->footer->tombBlkPtr->size > 0) {
3,480,193✔
153
      if (reader->footer->tombBlkPtr->size % sizeof(STombBlk) != 0) {
15,417!
154
        return TSDB_CODE_FILE_CORRUPTED;
×
155
      }
156

157
      int32_t size = reader->footer->tombBlkPtr->size / sizeof(STombBlk);
15,417✔
158
      void   *data = taosMemoryMalloc(reader->footer->tombBlkPtr->size);
15,417!
159
      if (!data) {
15,417!
160
        return terrno;
×
161
      }
162

163
      int32_t encryptAlgorithm = reader->config->tsdb->pVnode->config.tsdbCfg.encryptAlgorithm;
15,417✔
164
      char   *encryptKey = reader->config->tsdb->pVnode->config.tsdbCfg.encryptKey;
15,417✔
165
      int32_t code = tsdbReadFile(reader->fd, reader->footer->tombBlkPtr->offset, data,
15,417✔
166
                                  reader->footer->tombBlkPtr->size, 0, encryptAlgorithm, encryptKey);
167
      if (code) {
15,417!
168
        taosMemoryFree(data);
×
169
        return code;
×
170
      }
171

172
      TARRAY2_INIT_EX(reader->tombBlkArray, size, size, data);
15,417✔
173
    } else {
174
      TARRAY2_INIT(reader->tombBlkArray);
3,464,776✔
175
    }
176

177
    reader->ctx->tombBlkLoaded = true;
3,480,193✔
178
  }
179

180
  tombBlkArray[0] = reader->tombBlkArray;
3,479,541✔
181
  return 0;
3,479,541✔
182
}
183

184
int32_t tsdbSttFileReadSttBlk(SSttFileReader *reader, const TSttBlkArray **sttBlkArray) {
3,477,660✔
185
  if (!reader->ctx->sttBlkLoaded) {
3,477,660!
186
    if (reader->footer->sttBlkPtr->size > 0) {
3,480,227!
187
      if (reader->footer->sttBlkPtr->size % sizeof(SSttBlk) != 0) {
3,481,328!
188
        return TSDB_CODE_FILE_CORRUPTED;
×
189
      }
190

191
      int32_t size = reader->footer->sttBlkPtr->size / sizeof(SSttBlk);
3,481,328✔
192
      void   *data = taosMemoryMalloc(reader->footer->sttBlkPtr->size);
3,481,328✔
193
      if (!data) {
3,481,003!
194
        return terrno;
×
195
      }
196

197
      int32_t encryptAlgorithm = reader->config->tsdb->pVnode->config.tsdbCfg.encryptAlgorithm;
3,481,003✔
198
      char   *encryptKey = reader->config->tsdb->pVnode->config.tsdbCfg.encryptKey;
3,481,003✔
199
      int32_t code = tsdbReadFile(reader->fd, reader->footer->sttBlkPtr->offset, data, reader->footer->sttBlkPtr->size,
3,481,003✔
200
                                  0, encryptAlgorithm, encryptKey);
201
      if (code) {
3,482,835✔
202
        taosMemoryFree(data);
75!
203
        return code;
×
204
      }
205

206
      TARRAY2_INIT_EX(reader->sttBlkArray, size, size, data);
3,482,760✔
207
    } else {
UNCOV
208
      TARRAY2_INIT(reader->sttBlkArray);
×
209
    }
210

211
    reader->ctx->sttBlkLoaded = true;
3,481,659✔
212
  }
213

214
  sttBlkArray[0] = reader->sttBlkArray;
3,479,092✔
215
  return 0;
3,479,092✔
216
}
217

218
int32_t tsdbSttFileReadBlockData(SSttFileReader *reader, const SSttBlk *sttBlk, SBlockData *bData) {
21,640✔
219
  int32_t code = 0;
21,640✔
220
  int32_t lino = 0;
21,640✔
221

222
  SBuffer *buffer0 = reader->buffers + 0;
21,640✔
223
  SBuffer *assist = reader->buffers + 1;
21,640✔
224

225
  int32_t encryptAlgorithm = reader->config->tsdb->pVnode->config.tsdbCfg.encryptAlgorithm;
21,640✔
226
  char   *encryptKey = reader->config->tsdb->pVnode->config.tsdbCfg.encryptKey;
21,640✔
227
  // load data
228
  tBufferClear(buffer0);
229
  TAOS_CHECK_GOTO(tsdbReadFileToBuffer(reader->fd, sttBlk->bInfo.offset, sttBlk->bInfo.szBlock, buffer0, 0,
21,640!
230
                                       encryptAlgorithm, encryptKey),
231
                  &lino, _exit);
232

233
  SBufferReader br = BUFFER_READER_INITIALIZER(0, buffer0);
21,639✔
234
  TAOS_CHECK_GOTO(tBlockDataDecompress(&br, bData, assist), &lino, _exit);
21,639!
235

236
_exit:
21,640✔
237
  if (code) {
21,640!
238
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(reader->config->tsdb->pVnode), __func__, __FILE__, lino,
×
239
              tstrerror(code));
240
  }
241
  return code;
21,640✔
242
}
243

244
int32_t tsdbSttFileReadBlockDataByColumn(SSttFileReader *reader, const SSttBlk *sttBlk, SBlockData *bData,
1,654,580✔
245
                                         STSchema *pTSchema, int16_t cids[], int32_t ncid) {
246
  int32_t code = 0;
1,654,580✔
247
  int32_t lino = 0;
1,654,580✔
248

249
  SDiskDataHdr hdr;
250
  SBuffer     *buffer0 = reader->buffers + 0;
1,654,580✔
251
  SBuffer     *buffer1 = reader->buffers + 1;
1,654,580✔
252
  SBuffer     *assist = reader->buffers + 2;
1,654,580✔
253

254
  int32_t encryptAlgorithm = reader->config->tsdb->pVnode->config.tsdbCfg.encryptAlgorithm;
1,654,580✔
255
  char   *encryptKey = reader->config->tsdb->pVnode->config.tsdbCfg.encryptKey;
1,654,580✔
256
  // load key part
257
  tBufferClear(buffer0);
258
  TAOS_CHECK_GOTO(tsdbReadFileToBuffer(reader->fd, sttBlk->bInfo.offset, sttBlk->bInfo.szKey, buffer0, 0,
1,654,580!
259
                                       encryptAlgorithm, encryptKey),
260
                  &lino, _exit);
261

262
  // decode header
263
  SBufferReader br = BUFFER_READER_INITIALIZER(0, buffer0);
1,656,960✔
264
  TAOS_CHECK_GOTO(tGetDiskDataHdr(&br, &hdr), &lino, _exit);
1,656,960!
265

266
  if (hdr.delimiter != TSDB_FILE_DLMT) {
1,656,072!
267
    TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
×
268
  }
269

270
  // set data container
271
  tBlockDataReset(bData);
1,656,072✔
272
  bData->suid = hdr.suid;
1,655,205✔
273
  bData->uid = (sttBlk->suid == 0) ? sttBlk->minUid : 0;
1,655,205✔
274
  bData->nRow = hdr.nRow;
1,655,205✔
275

276
  // key part
277
  TAOS_CHECK_GOTO(tBlockDataDecompressKeyPart(&hdr, &br, bData, assist), &lino, _exit);
1,655,205!
278
  if (br.offset != buffer0->size) {
1,656,012!
279
    TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
×
280
  }
281

282
  bool loadExtra = false;
1,656,012✔
283
  for (int i = 0; i < ncid; i++) {
2,019,906✔
284
    if (tBlockDataGetColData(bData, cids[i]) == NULL) {
1,550,809✔
285
      loadExtra = true;
1,186,302✔
286
      break;
1,186,302✔
287
    }
288
  }
289

290
  if (!loadExtra) {
1,655,399✔
291
    goto _exit;
469,893✔
292
  }
293

294
  // load SBlockCol part
295
  tBufferClear(buffer0);
296
  TAOS_CHECK_GOTO(tsdbReadFileToBuffer(reader->fd, sttBlk->bInfo.offset + sttBlk->bInfo.szKey, hdr.szBlkCol, buffer0, 0,
1,185,506!
297
                                       encryptAlgorithm, encryptKey),
298
                  &lino, _exit);
299

300
  // load each column
301
  SBlockCol blockCol = {
1,186,660✔
302
      .cid = 0,
303
  };
304
  br = BUFFER_READER_INITIALIZER(0, buffer0);
1,186,660✔
305
  for (int32_t i = 0; i < ncid; i++) {
5,417,258✔
306
    int16_t cid = cids[i];
4,233,154✔
307

308
    if (tBlockDataGetColData(bData, cid)) {  // already loaded
4,233,154✔
309
      continue;
174,839✔
310
    }
311

312
    while (cid > blockCol.cid) {
13,094,658✔
313
      if (br.offset >= buffer0->size) {
9,138,425✔
314
        blockCol.cid = INT16_MAX;
97,501✔
315
        break;
97,501✔
316
      }
317

318
      TAOS_CHECK_GOTO(tGetBlockCol(&br, &blockCol, hdr.fmtVer, hdr.cmprAlg), &lino, _exit);
9,040,924!
319
    }
320

321
    if (cid < blockCol.cid) {
4,053,734✔
322
      const STColumn *tcol = tTSchemaSearchColumn(pTSchema, cid);
438,651✔
323
      TSDB_CHECK_NULL(tcol, code, lino, _exit, TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER);
438,158!
324
      SBlockCol none = {
438,158✔
325
          .cid = cid,
326
          .type = tcol->type,
438,158✔
327
          .cflag = tcol->flags,
438,158✔
328
          .flag = HAS_NONE,
329
          .szOrigin = 0,
330
          .szBitmap = 0,
331
          .szOffset = 0,
332
          .szValue = 0,
333
          .offset = 0,
334
      };
335
      TAOS_CHECK_GOTO(tBlockDataDecompressColData(&hdr, &none, &br, bData, assist), &lino, _exit);
438,158!
336
    } else if (cid == blockCol.cid) {
3,615,083!
337
      // load from file
338
      tBufferClear(buffer1);
339
      TAOS_CHECK_GOTO(
3,619,916!
340
          tsdbReadFileToBuffer(reader->fd, sttBlk->bInfo.offset + sttBlk->bInfo.szKey + hdr.szBlkCol + blockCol.offset,
341
                               blockCol.szBitmap + blockCol.szOffset + blockCol.szValue, buffer1, 0, encryptAlgorithm,
342
                               encryptKey),
343
          &lino, _exit);
344

345
      // decode the buffer
346
      SBufferReader br1 = BUFFER_READER_INITIALIZER(0, buffer1);
3,623,294✔
347
      TAOS_CHECK_GOTO(tBlockDataDecompressColData(&hdr, &blockCol, &br1, bData, assist), &lino, _exit);
3,623,294!
348
    }
349
  }
350

351
_exit:
1,184,104✔
352
  if (code) {
1,653,997!
353
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(reader->config->tsdb->pVnode), __func__, __FILE__, lino,
×
354
              tstrerror(code));
355
  }
356
  return code;
1,656,583✔
357
}
358

359
int32_t tsdbSttFileReadTombBlock(SSttFileReader *reader, const STombBlk *tombBlk, STombBlock *tombBlock) {
14,734✔
360
  int32_t code = 0;
14,734✔
361
  int32_t lino = 0;
14,734✔
362

363
  SBuffer *buffer0 = reader->buffers + 0;
14,734✔
364
  SBuffer *assist = reader->buffers + 1;
14,734✔
365

366
  int32_t encryptAlgorithm = reader->config->tsdb->pVnode->config.tsdbCfg.encryptAlgorithm;
14,734✔
367
  char   *encryptKey = reader->config->tsdb->pVnode->config.tsdbCfg.encryptKey;
14,734✔
368
  // load
369
  tBufferClear(buffer0);
370
  TAOS_CHECK_GOTO(tsdbReadFileToBuffer(reader->fd, tombBlk->dp->offset, tombBlk->dp->size, buffer0, 0, encryptAlgorithm,
14,734!
371
                                       encryptKey),
372
                  &lino, _exit);
373

374
  // decode
375
  int32_t       size = 0;
14,734✔
376
  SBufferReader br = BUFFER_READER_INITIALIZER(0, buffer0);
14,734✔
377
  tTombBlockClear(tombBlock);
14,734✔
378
  tombBlock->numOfRecords = tombBlk->numRec;
14,734✔
379
  for (int32_t i = 0; i < ARRAY_SIZE(tombBlock->buffers); ++i) {
88,403✔
380
    SCompressInfo cinfo = {
73,667✔
381
        .cmprAlg = tombBlk->cmprAlg,
73,667✔
382
        .dataType = TSDB_DATA_TYPE_BIGINT,
383
        .originalSize = tombBlk->numRec * sizeof(int64_t),
73,667✔
384
        .compressedSize = tombBlk->size[i],
73,667✔
385
    };
386
    TAOS_CHECK_GOTO(tDecompressDataToBuffer(BR_PTR(&br), &cinfo, tombBlock->buffers + i, assist), &lino, _exit);
73,667!
387
    br.offset += tombBlk->size[i];
73,669✔
388
  }
389

390
  if (br.offset != tombBlk->dp->size) {
14,736!
391
    TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
×
392
  }
393

394
_exit:
14,736✔
395
  if (code) {
14,736!
396
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(reader->config->tsdb->pVnode), __func__, __FILE__, lino,
×
397
              tstrerror(code));
398
  }
399
  return code;
14,734✔
400
}
401

402
int32_t tsdbSttFileReadStatisBlock(SSttFileReader *reader, const SStatisBlk *statisBlk, STbStatisBlock *statisBlock) {
2,024,003✔
403
  int32_t code = 0;
2,024,003✔
404
  int32_t lino = 0;
2,024,003✔
405

406
  SBuffer *buffer0 = reader->buffers + 0;
2,024,003✔
407
  SBuffer *assist = reader->buffers + 1;
2,024,003✔
408

409
  int32_t encryptAlgorithm = reader->config->tsdb->pVnode->config.tsdbCfg.encryptAlgorithm;
2,024,003✔
410
  char   *encryptKey = reader->config->tsdb->pVnode->config.tsdbCfg.encryptKey;
2,024,003✔
411
  // load data
412
  tBufferClear(buffer0);
413
  TAOS_CHECK_GOTO(tsdbReadFileToBuffer(reader->fd, statisBlk->dp->offset, statisBlk->dp->size, buffer0, 0,
2,024,003!
414
                                       encryptAlgorithm, encryptKey),
415
                  &lino, _exit);
416

417
  // decode data
418
  tStatisBlockClear(statisBlock);
2,026,131✔
419
  statisBlock->numOfPKs = statisBlk->numOfPKs;
2,025,834✔
420
  statisBlock->numOfRecords = statisBlk->numRec;
2,025,834✔
421
  SBufferReader br = BUFFER_READER_INITIALIZER(0, buffer0);
2,025,834✔
422
  for (int32_t i = 0; i < ARRAY_SIZE(statisBlock->buffers); ++i) {
12,151,269✔
423
    SCompressInfo info = {
10,126,064✔
424
        .dataType = TSDB_DATA_TYPE_BIGINT,
425
        .cmprAlg = statisBlk->cmprAlg,
10,126,064✔
426
        .compressedSize = statisBlk->size[i],
10,126,064✔
427
        .originalSize = statisBlk->numRec * sizeof(int64_t),
10,126,064✔
428
    };
429

430
    TAOS_CHECK_GOTO(tDecompressDataToBuffer(BR_PTR(&br), &info, &statisBlock->buffers[i], assist), &lino, _exit);
10,126,064!
431
    br.offset += statisBlk->size[i];
10,125,435✔
432
  }
433

434
  if (statisBlk->numOfPKs > 0) {
2,025,205✔
435
    SValueColumnCompressInfo firstKeyInfos[TD_MAX_PK_COLS];
436
    SValueColumnCompressInfo lastKeyInfos[TD_MAX_PK_COLS];
437

438
    // decode compress info
439
    for (int32_t i = 0; i < statisBlk->numOfPKs; i++) {
982,652✔
440
      TAOS_CHECK_GOTO(tValueColumnCompressInfoDecode(&br, &firstKeyInfos[i]), &lino, _exit);
491,309!
441
    }
442

443
    for (int32_t i = 0; i < statisBlk->numOfPKs; i++) {
982,486✔
444
      TAOS_CHECK_GOTO(tValueColumnCompressInfoDecode(&br, &lastKeyInfos[i]), &lino, _exit);
491,155!
445
    }
446

447
    // decode value columns
448
    for (int32_t i = 0; i < statisBlk->numOfPKs; i++) {
982,652✔
449
      TAOS_CHECK_GOTO(tValueColumnDecompress(BR_PTR(&br), firstKeyInfos + i, &statisBlock->firstKeyPKs[i], assist),
490,874!
450
                      &lino, _exit);
451
      br.offset += (firstKeyInfos[i].dataCompressedSize + firstKeyInfos[i].offsetCompressedSize);
491,321✔
452
    }
453

454
    for (int32_t i = 0; i < statisBlk->numOfPKs; i++) {
983,228✔
455
      TAOS_CHECK_GOTO(tValueColumnDecompress(BR_PTR(&br), &lastKeyInfos[i], &statisBlock->lastKeyPKs[i], assist), &lino,
491,289!
456
                      _exit);
457
      br.offset += (lastKeyInfos[i].dataCompressedSize + lastKeyInfos[i].offsetCompressedSize);
491,450✔
458
    }
459
  }
460

461
  if (br.offset != buffer0->size) {
2,025,671!
462
    TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
×
463
  }
464

465
_exit:
2,025,671✔
466
  if (code) {
2,025,671!
467
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(reader->config->tsdb->pVnode), __func__, __FILE__, lino,
×
468
              tstrerror(code));
469
  }
470
  return code;
2,025,639✔
471
}
472

473
// SSttFWriter ============================================================
474
struct SSttFileWriter {
475
  SSttFileWriterConfig config[1];
476
  struct {
477
    bool    opened;
478
    TABLEID tbid[1];
479
    // range
480
    SVersionRange range;
481
  } ctx[1];
482
  // file
483
  STsdbFD *fd;
484
  STFile   file[1];
485
  // data
486
  SSttFooter      footer[1];
487
  TTombBlkArray   tombBlkArray[1];
488
  TSttBlkArray    sttBlkArray[1];
489
  TStatisBlkArray statisBlkArray[1];
490
  STombBlock      tombBlock[1];
491
  STbStatisBlock  staticBlock[1];
492
  SBlockData      blockData[1];
493
  // helper data
494
  SSkmInfo skmTb[1];
495
  SSkmInfo skmRow[1];
496
  SBuffer  local[10];
497
  SBuffer *buffers;
498
  // SColCompressInfo2 pInfo;
499
};
500

501
static int32_t tsdbFileDoWriteSttBlockData(STsdbFD *fd, SBlockData *blockData, SColCompressInfo *info,
622,705✔
502
                                           int64_t *fileSize, TSttBlkArray *sttBlkArray, SBuffer *buffers,
503
                                           SVersionRange *range, int32_t encryptAlgorithm, char *encryptKey) {
504
  if (blockData->nRow == 0) return 0;
622,705!
505

506
  int32_t code = 0;
622,705✔
507

508
  SSttBlk sttBlk[1] = {{
1,868,115✔
509
      .suid = blockData->suid,
622,705✔
510
      .minUid = blockData->uid ? blockData->uid : blockData->aUid[0],
622,705✔
511
      .maxUid = blockData->uid ? blockData->uid : blockData->aUid[blockData->nRow - 1],
622,705✔
512
      .minKey = blockData->aTSKEY[0],
622,705✔
513
      .maxKey = blockData->aTSKEY[0],
622,705✔
514
      .minVer = blockData->aVersion[0],
622,705✔
515
      .maxVer = blockData->aVersion[0],
622,705✔
516
      .nRow = blockData->nRow,
622,705✔
517
  }};
518

519
  for (int32_t iRow = 1; iRow < blockData->nRow; iRow++) {
83,849,073✔
520
    if (sttBlk->minKey > blockData->aTSKEY[iRow]) sttBlk->minKey = blockData->aTSKEY[iRow];
83,226,368✔
521
    if (sttBlk->maxKey < blockData->aTSKEY[iRow]) sttBlk->maxKey = blockData->aTSKEY[iRow];
83,226,368✔
522
    if (sttBlk->minVer > blockData->aVersion[iRow]) sttBlk->minVer = blockData->aVersion[iRow];
83,226,368✔
523
    if (sttBlk->maxVer < blockData->aVersion[iRow]) sttBlk->maxVer = blockData->aVersion[iRow];
83,226,368✔
524
  }
525

526
  tsdbWriterUpdVerRange(range, sttBlk->minVer, sttBlk->maxVer);
622,705✔
527
  TAOS_CHECK_RETURN(tBlockDataCompress(blockData, info, buffers, buffers + 4));
622,734!
528

529
  sttBlk->bInfo.offset = *fileSize;
622,794✔
530
  sttBlk->bInfo.szKey = buffers[0].size + buffers[1].size;
622,794✔
531
  sttBlk->bInfo.szBlock = buffers[2].size + buffers[3].size + sttBlk->bInfo.szKey;
622,794✔
532
  for (int i = 0; i < 4; i++) {
3,113,576✔
533
    if (buffers[i].size) {
2,490,884✔
534
      TAOS_CHECK_RETURN(tsdbWriteFile(fd, *fileSize, buffers[i].data, buffers[i].size, encryptAlgorithm, encryptKey));
2,489,807✔
535
      *fileSize += buffers[i].size;
2,489,705✔
536
    }
537
  }
538

539
  TAOS_CHECK_RETURN(TARRAY2_APPEND_PTR(sttBlkArray, sttBlk));
1,245,426!
540

541
  tBlockDataClear(blockData);
622,734✔
542
  return 0;
622,752✔
543
}
544

545
static int32_t tsdbSttFileDoWriteBlockData(SSttFileWriter *writer) {
967,273✔
546
  if (writer->blockData->nRow == 0) return 0;
967,273✔
547

548
  int32_t code = 0;
622,754✔
549
  int32_t lino = 0;
622,754✔
550

551
  tb_uid_t         uid = writer->blockData->suid == 0 ? writer->blockData->uid : writer->blockData->suid;
622,754✔
552
  SColCompressInfo info = {.defaultCmprAlg = writer->config->cmprAlg, .pColCmpr = NULL};
622,754✔
553
  code = metaGetColCmpr(writer->config->tsdb->pVnode->pMeta, uid, &(info.pColCmpr));
622,754✔
554

555
  int32_t encryptAlgorithm = writer->config->tsdb->pVnode->config.tsdbCfg.encryptAlgorithm;
622,714✔
556
  char   *encryptKey = writer->config->tsdb->pVnode->config.tsdbCfg.encryptKey;
622,714✔
557
  TAOS_CHECK_GOTO(
622,714!
558
      tsdbFileDoWriteSttBlockData(writer->fd, writer->blockData, &info, &writer->file->size, writer->sttBlkArray,
559
                                  writer->buffers, &writer->ctx->range, encryptAlgorithm, encryptKey),
560
      &lino, _exit);
561

562
_exit:
622,744✔
563
  if (code) {
622,744!
564
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
565
              tstrerror(code));
566
  }
567
  taosHashCleanup(info.pColCmpr);
622,744✔
568
  return code;
622,853✔
569
}
570

571
static int32_t tsdbSttFileDoWriteStatisBlock(SSttFileWriter *writer) {
346,240✔
572
  if (writer->staticBlock->numOfRecords == 0) {
346,240✔
573
    return 0;
4,107✔
574
  }
575

576
  int32_t code = 0;
342,133✔
577
  int32_t lino = 0;
342,133✔
578

579
  SBuffer *buffer0 = writer->buffers + 0;
342,133✔
580
  SBuffer *buffer1 = writer->buffers + 1;
342,133✔
581
  SBuffer *assist = writer->buffers + 2;
342,133✔
582

583
  STbStatisRecord record;
584
  STbStatisBlock *statisBlock = writer->staticBlock;
342,133✔
585
  SStatisBlk      statisBlk = {0};
342,133✔
586

587
  statisBlk.dp->offset = writer->file->size;
342,133✔
588
  statisBlk.dp->size = 0;
342,133✔
589
  statisBlk.numRec = statisBlock->numOfRecords;
342,133✔
590
  statisBlk.cmprAlg = writer->config->cmprAlg;
342,133✔
591
  statisBlk.numOfPKs = statisBlock->numOfPKs;
342,133✔
592

593
  code = tStatisBlockGet(statisBlock, 0, &record);
342,133✔
594
  TSDB_CHECK_CODE(code, lino, _exit);
342,122!
595
  statisBlk.minTbid.suid = record.suid;
342,122✔
596
  statisBlk.minTbid.uid = record.uid;
342,122✔
597

598
  code = tStatisBlockGet(statisBlock, statisBlock->numOfRecords - 1, &record);
342,122✔
599
  TSDB_CHECK_CODE(code, lino, _exit);
342,127!
600
  statisBlk.maxTbid.suid = record.suid;
342,127✔
601
  statisBlk.maxTbid.uid = record.uid;
342,127✔
602

603
  int32_t encryptAlgorithm = writer->config->tsdb->pVnode->config.tsdbCfg.encryptAlgorithm;
342,127✔
604
  char   *encryptKey = writer->config->tsdb->pVnode->config.tsdbCfg.encryptKey;
342,127✔
605

606
  // compress each column
607
  for (int32_t i = 0; i < ARRAY_SIZE(statisBlk.size); i++) {
2,052,633✔
608
    SCompressInfo info = {
1,710,385✔
609
        .dataType = TSDB_DATA_TYPE_BIGINT,
610
        .cmprAlg = statisBlk.cmprAlg,
1,710,385✔
611
        .originalSize = statisBlock->buffers[i].size,
1,710,385✔
612
    };
613

614
    tBufferClear(buffer0);
615
    TAOS_CHECK_GOTO(tCompressDataToBuffer(statisBlock->buffers[i].data, &info, buffer0, assist), &lino, _exit);
1,710,385!
616
    TAOS_CHECK_GOTO(
1,710,556!
617
        tsdbWriteFile(writer->fd, writer->file->size, buffer0->data, info.compressedSize, encryptAlgorithm, encryptKey),
618
        &lino, _exit);
619

620
    statisBlk.size[i] = info.compressedSize;
1,710,506✔
621
    statisBlk.dp->size += info.compressedSize;
1,710,506✔
622
    writer->file->size += info.compressedSize;
1,710,506✔
623
  }
624

625
  // compress primary keys
626
  if (statisBlk.numOfPKs > 0) {
342,248✔
627
    SValueColumnCompressInfo compressInfo = {.cmprAlg = statisBlk.cmprAlg};
4,662✔
628

629
    tBufferClear(buffer0);
630
    tBufferClear(buffer1);
631

632
    for (int32_t i = 0; i < statisBlk.numOfPKs; i++) {
9,322✔
633
      TAOS_CHECK_GOTO(tValueColumnCompress(&statisBlock->firstKeyPKs[i], &compressInfo, buffer1, assist), &lino, _exit);
4,661!
634
      TAOS_CHECK_GOTO(tValueColumnCompressInfoEncode(&compressInfo, buffer0), &lino, _exit);
4,662!
635
    }
636

637
    for (int32_t i = 0; i < statisBlk.numOfPKs; i++) {
9,323✔
638
      TAOS_CHECK_GOTO(tValueColumnCompress(&statisBlock->lastKeyPKs[i], &compressInfo, buffer1, assist), &lino, _exit);
4,659!
639
      TAOS_CHECK_GOTO(tValueColumnCompressInfoEncode(&compressInfo, buffer0), &lino, _exit);
4,661!
640
    }
641

642
    TAOS_CHECK_GOTO(
4,664!
643
        tsdbWriteFile(writer->fd, writer->file->size, buffer0->data, buffer0->size, encryptAlgorithm, encryptKey),
644
        &lino, _exit);
645
    writer->file->size += buffer0->size;
4,660✔
646
    statisBlk.dp->size += buffer0->size;
4,660✔
647

648
    TAOS_CHECK_GOTO(
4,660!
649
        tsdbWriteFile(writer->fd, writer->file->size, buffer1->data, buffer1->size, encryptAlgorithm, encryptKey),
650
        &lino, _exit);
651
    writer->file->size += buffer1->size;
4,659✔
652
    statisBlk.dp->size += buffer1->size;
4,659✔
653
  }
654

655
  TAOS_CHECK_GOTO(TARRAY2_APPEND_PTR(writer->statisBlkArray, &statisBlk), &lino, _exit);
684,499!
656

657
  tStatisBlockClear(writer->staticBlock);
342,254✔
658

659
_exit:
342,090✔
660
  if (code) {
342,090!
661
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
662
              tstrerror(code));
663
  }
664
  return code;
342,087✔
665
}
666

667
static int32_t tsdbSttFileDoWriteTombBlock(SSttFileWriter *writer) {
340,504✔
668
  if (TOMB_BLOCK_SIZE(writer->tombBlock) == 0) {
340,504✔
669
    return 0;
336,064✔
670
  }
671

672
  int32_t code = 0;
4,440✔
673
  int32_t lino = 0;
4,440✔
674

675
  int32_t encryptAlgorithm = writer->config->tsdb->pVnode->config.tsdbCfg.encryptAlgorithm;
4,440✔
676
  char   *encryptKey = writer->config->tsdb->pVnode->config.tsdbCfg.encryptKey;
4,440✔
677

678
  TAOS_CHECK_GOTO(
4,440!
679
      tsdbFileWriteTombBlock(writer->fd, writer->tombBlock, writer->config->cmprAlg, &writer->file->size,
680
                             writer->tombBlkArray, writer->buffers, &writer->ctx->range, encryptAlgorithm, encryptKey),
681
      &lino, _exit);
682

683
_exit:
4,447✔
684
  if (code) {
4,447!
685
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
686
              tstrerror(code));
687
  }
688
  return code;
4,447✔
689
}
690

691
int32_t tsdbFileWriteSttBlk(STsdbFD *fd, const TSttBlkArray *sttBlkArray, SFDataPtr *ptr, int64_t *fileSize,
340,501✔
692
                            int32_t encryptAlgorithm, char *encryptKey) {
693
  ptr->size = TARRAY2_DATA_LEN(sttBlkArray);
340,501✔
694
  if (ptr->size > 0) {
340,501✔
695
    ptr->offset = *fileSize;
340,398✔
696

697
    TAOS_CHECK_RETURN(tsdbWriteFile(fd, *fileSize, (const uint8_t *)TARRAY2_DATA(sttBlkArray), ptr->size,
340,398!
698
                                    encryptAlgorithm, encryptKey));
699

700
    *fileSize += ptr->size;
340,442✔
701
  }
702
  return 0;
340,545✔
703
}
704

705
static int32_t tsdbSttFileDoWriteSttBlk(SSttFileWriter *writer) {
340,499✔
706
  int32_t code = 0;
340,499✔
707
  int32_t lino;
708

709
  int32_t encryptAlgorithm = writer->config->tsdb->pVnode->config.tsdbCfg.encryptAlgorithm;
340,499✔
710
  char   *encryptKey = writer->config->tsdb->pVnode->config.tsdbCfg.encryptKey;
340,499✔
711

712
  TAOS_CHECK_GOTO(tsdbFileWriteSttBlk(writer->fd, writer->sttBlkArray, writer->footer->sttBlkPtr, &writer->file->size,
340,499!
713
                                      encryptAlgorithm, encryptKey),
714
                  &lino, _exit);
715

716
_exit:
340,547✔
717
  if (code) {
340,547!
718
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
719
              tstrerror(code));
720
  }
721
  return code;
340,548✔
722
}
723

724
static int32_t tsdbSttFileDoWriteStatisBlk(SSttFileWriter *writer) {
340,538✔
725
  int32_t code = 0;
340,538✔
726
  int32_t lino;
727
  int32_t encryptAlgorithm = writer->config->tsdb->pVnode->config.tsdbCfg.encryptAlgorithm;
340,538✔
728
  char   *encryptKey = writer->config->tsdb->pVnode->config.tsdbCfg.encryptKey;
340,538✔
729

730
  writer->footer->statisBlkPtr->size = TARRAY2_DATA_LEN(writer->statisBlkArray);
340,538✔
731
  if (writer->footer->statisBlkPtr->size) {
340,538✔
732
    writer->footer->statisBlkPtr->offset = writer->file->size;
340,437✔
733
    TAOS_CHECK_GOTO(tsdbWriteFile(writer->fd, writer->file->size, (const uint8_t *)TARRAY2_DATA(writer->statisBlkArray),
340,437!
734
                                  writer->footer->statisBlkPtr->size, encryptAlgorithm, encryptKey),
735
                    &lino, _exit);
736
    writer->file->size += writer->footer->statisBlkPtr->size;
340,428✔
737
  }
738

739
_exit:
101✔
740
  if (code) {
340,529!
741
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
742
              tstrerror(code));
743
  }
744
  return code;
340,540✔
745
}
746

747
static int32_t tsdbSttFileDoWriteTombBlk(SSttFileWriter *writer) {
340,518✔
748
  int32_t code = 0;
340,518✔
749
  int32_t lino = 0;
340,518✔
750

751
  int32_t encryptAlgorithm = writer->config->tsdb->pVnode->config.tsdbCfg.encryptAlgorithm;
340,518✔
752
  char   *encryptKey = writer->config->tsdb->pVnode->config.tsdbCfg.encryptKey;
340,518✔
753

754
  TAOS_CHECK_GOTO(tsdbFileWriteTombBlk(writer->fd, writer->tombBlkArray, writer->footer->tombBlkPtr,
340,518!
755
                                       &writer->file->size, encryptAlgorithm, encryptKey),
756
                  &lino, _exit);
757

758
_exit:
340,523✔
759
  if (code) {
340,523!
760
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
761
              tstrerror(code));
762
  }
763
  return code;
340,523✔
764
}
765

766
int32_t tsdbFileWriteSttFooter(STsdbFD *fd, const SSttFooter *footer, int64_t *fileSize, int32_t encryptAlgorithm,
340,514✔
767
                               char *encryptKey) {
768
  TAOS_CHECK_RETURN(
340,514!
769
      tsdbWriteFile(fd, *fileSize, (const uint8_t *)footer, sizeof(*footer), encryptAlgorithm, encryptKey));
770
  *fileSize += sizeof(*footer);
340,539✔
771
  return 0;
340,539✔
772
}
773

774
static int32_t tsdbSttFileDoWriteFooter(SSttFileWriter *writer) {
340,508✔
775
  int32_t encryptAlgorithm = writer->config->tsdb->pVnode->config.tsdbCfg.encryptAlgorithm;
340,508✔
776
  char   *encryptKey = writer->config->tsdb->pVnode->config.tsdbCfg.encryptKey;
340,508✔
777

778
  return tsdbFileWriteSttFooter(writer->fd, writer->footer, &writer->file->size, encryptAlgorithm, encryptKey);
340,508✔
779
}
780

781
static int32_t tsdbSttFWriterDoOpen(SSttFileWriter *writer) {
340,526✔
782
  int32_t code = 0;
340,526✔
783
  int32_t lino = 0;
340,526✔
784
  STsdb  *tsdb = writer->config->tsdb;
340,526✔
785

786
  // set
787
  if (!writer->config->skmTb) writer->config->skmTb = writer->skmTb;
340,526!
788
  if (!writer->config->skmRow) writer->config->skmRow = writer->skmRow;
340,526!
789
  writer->buffers = writer->config->buffers;
340,526✔
790
  if (writer->buffers == NULL) {
340,526!
791
    writer->buffers = writer->local;
×
792
  }
793

794
  // alloc disk id
795
  SDiskID diskId = {0};
340,526✔
796
  code = tsdbAllocateDisk(tsdb, tsdbFTypeLabel(TSDB_FTYPE_STT), writer->config->expLevel, &diskId);
340,526✔
797
  TSDB_CHECK_CODE(code, lino, _exit);
340,525!
798

799
  writer->file[0] = (STFile){
340,525✔
800
      .type = TSDB_FTYPE_STT,
801
      .did = diskId,
802
      .fid = writer->config->fid,
340,525✔
803
      .cid = writer->config->cid,
340,525✔
804
      .size = 0,
805
      .minVer = VERSION_MAX,
806
      .maxVer = VERSION_MIN,
807
      .stt[0] =
808
          {
809
              .level = writer->config->level,
340,525✔
810
          },
811
  };
812

813
  // open file
814
  int32_t flag = TD_FILE_READ | TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
340,525✔
815
  char    fname[TSDB_FILENAME_LEN];
816

817
  tsdbTFileName(writer->config->tsdb, writer->file, fname);
340,525✔
818
  TAOS_CHECK_GOTO(tsdbOpenFile(fname, writer->config->tsdb, flag, &writer->fd, 0), &lino, _exit);
340,558!
819

820
  uint8_t hdr[TSDB_FHDR_SIZE] = {0};
340,546✔
821
  int32_t encryptAlgorithm = writer->config->tsdb->pVnode->config.tsdbCfg.encryptAlgorithm;
340,546✔
822
  char   *encryptKey = writer->config->tsdb->pVnode->config.tsdbCfg.encryptKey;
340,546✔
823

824
  TAOS_CHECK_GOTO(tsdbWriteFile(writer->fd, 0, hdr, sizeof(hdr), encryptAlgorithm, encryptKey), &lino, _exit);
340,546!
825
  writer->file->size += sizeof(hdr);
340,491✔
826

827
  // range
828
  writer->ctx->range = (SVersionRange){.minVer = VERSION_MAX, .maxVer = VERSION_MIN};
340,491✔
829

830
  writer->ctx->opened = true;
340,491✔
831

832
_exit:
340,491✔
833
  if (code) {
340,491!
834
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
835
              tstrerror(code));
836
  }
837
  return code;
340,524✔
838
}
839

840
static void tsdbSttFWriterDoClose(SSttFileWriter *writer) {
340,578✔
841
  for (int32_t i = 0; i < ARRAY_SIZE(writer->local); ++i) {
3,746,224✔
842
    tBufferDestroy(writer->local + i);
3,405,646!
843
  }
844
  tDestroyTSchema(writer->skmRow->pTSchema);
340,578!
845
  tDestroyTSchema(writer->skmTb->pTSchema);
340,578!
846
  tTombBlockDestroy(writer->tombBlock);
340,578✔
847
  tStatisBlockDestroy(writer->staticBlock);
340,581✔
848
  tBlockDataDestroy(writer->blockData);
340,577✔
849
  TARRAY2_DESTROY(writer->tombBlkArray, NULL);
340,567!
850
  TARRAY2_DESTROY(writer->statisBlkArray, NULL);
340,567!
851
  TARRAY2_DESTROY(writer->sttBlkArray, NULL);
340,579!
852
}
340,581✔
853

854
static int32_t tsdbSttFileDoUpdateHeader(SSttFileWriter *writer) {
340,513✔
855
  // TODO
856
  return 0;
340,513✔
857
}
858

859
static int32_t tsdbSttFWriterCloseCommit(SSttFileWriter *writer, TFileOpArray *opArray) {
340,538✔
860
  int32_t lino;
861
  int32_t code;
862

863
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteBlockData(writer), &lino, _exit);
340,538!
864
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteStatisBlock(writer), &lino, _exit);
340,563!
865
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteTombBlock(writer), &lino, _exit);
340,515!
866
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteSttBlk(writer), &lino, _exit);
340,508!
867
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteStatisBlk(writer), &lino, _exit);
340,547!
868
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteTombBlk(writer), &lino, _exit);
340,532!
869
  TAOS_CHECK_GOTO(tsdbSttFileDoWriteFooter(writer), &lino, _exit);
340,525!
870
  TAOS_CHECK_GOTO(tsdbSttFileDoUpdateHeader(writer), &lino, _exit);
340,537!
871

872
  int32_t encryptAlgorithm = writer->config->tsdb->pVnode->config.tsdbCfg.encryptAlgorithm;
340,514✔
873
  char   *encryptKey = writer->config->tsdb->pVnode->config.tsdbCfg.encryptKey;
340,514✔
874

875
  TAOS_CHECK_GOTO(tsdbFsyncFile(writer->fd, encryptAlgorithm, encryptKey), &lino, _exit);
340,514!
876

877
  tsdbCloseFile(&writer->fd);
340,553✔
878

879
  STFileOp op = (STFileOp){
340,584✔
880
      .optype = TSDB_FOP_CREATE,
881
      .fid = writer->config->fid,
340,584✔
882
      .nf = writer->file[0],
883
  };
884
  tsdbTFileUpdVerRange(&op.nf, writer->ctx->range);
340,584✔
885

886
  TAOS_CHECK_GOTO(TARRAY2_APPEND(opArray, op), &lino, _exit);
681,142!
887

888
_exit:
340,571✔
889
  if (code) {
340,571!
890
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
891
              tstrerror(code));
892
  }
893
  return code;
340,568✔
894
}
895

896
static int32_t tsdbSttFWriterCloseAbort(SSttFileWriter *writer) {
×
897
  char fname[TSDB_FILENAME_LEN];
898
  tsdbTFileName(writer->config->tsdb, writer->file, fname);
×
899
  tsdbCloseFile(&writer->fd);
×
900
  tsdbRemoveFile(fname);
×
901
  return 0;
×
902
}
903

904
int32_t tsdbSttFileWriterOpen(const SSttFileWriterConfig *config, SSttFileWriter **writer) {
461,169✔
905
  writer[0] = taosMemoryCalloc(1, sizeof(*writer[0]));
461,169!
906
  if (writer[0] == NULL) {
461,213!
907
    return terrno;
×
908
  }
909

910
  writer[0]->config[0] = config[0];
461,213✔
911
  writer[0]->ctx->opened = false;
461,213✔
912
  return 0;
461,213✔
913
}
914

915
int32_t tsdbSttFileWriterClose(SSttFileWriter **writer, int8_t abort, TFileOpArray *opArray) {
461,176✔
916
  int32_t code = 0;
461,176✔
917
  int32_t lino = 0;
461,176✔
918

919
  if (writer[0]->ctx->opened) {
461,176✔
920
    if (abort) {
340,542!
921
      TAOS_CHECK_GOTO(tsdbSttFWriterCloseAbort(writer[0]), &lino, _exit);
×
922
    } else {
923
      TAOS_CHECK_GOTO(tsdbSttFWriterCloseCommit(writer[0], opArray), &lino, _exit);
340,542!
924
    }
925
    tsdbSttFWriterDoClose(writer[0]);
340,561✔
926
  }
927
  taosMemoryFree(writer[0]);
461,210!
928
  writer[0] = NULL;
461,220✔
929

930
_exit:
461,220✔
931
  if (code) {
461,220!
932
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer[0]->config->tsdb->pVnode), __func__, __FILE__, lino,
×
933
              tstrerror(code));
934
  }
935
  return code;
461,221✔
936
}
937

938
int32_t tsdbSttFileWriteRow(SSttFileWriter *writer, SRowInfo *row) {
82,057,306✔
939
  int32_t code = 0;
82,057,306✔
940
  int32_t lino = 0;
82,057,306✔
941

942
  if (!writer->ctx->opened) {
82,057,306✔
943
    TAOS_CHECK_GOTO(tsdbSttFWriterDoOpen(writer), &lino, _exit);
340,423!
944
  }
945

946
  if (!TABLE_SAME_SCHEMA(row->suid, row->uid, writer->ctx->tbid->suid, writer->ctx->tbid->uid)) {
82,057,267✔
947
    TAOS_CHECK_GOTO(tsdbSttFileDoWriteBlockData(writer), &lino, _exit);
603,224!
948

949
    TAOS_CHECK_GOTO(tsdbUpdateSkmTb(writer->config->tsdb, (TABLEID *)row, writer->config->skmTb), &lino, _exit);
603,202!
950

951
    TABLEID id = {
1,206,426✔
952
        .suid = row->suid,
603,213✔
953
        .uid = row->suid ? 0 : row->uid,
603,213✔
954
    };
955
    TAOS_CHECK_GOTO(tBlockDataInit(writer->blockData, &id, writer->config->skmTb->pTSchema, NULL, 0), &lino, _exit);
603,213!
956
  }
957

958
  if (writer->ctx->tbid->uid != row->uid) {
82,057,213✔
959
    writer->ctx->tbid->suid = row->suid;
1,450,316✔
960
    writer->ctx->tbid->uid = row->uid;
1,450,316✔
961
  }
962

963
  STsdbRowKey key;
964
  tsdbRowGetKey(&row->row, &key);
82,057,213✔
965

966
  for (;;) {
967
    code = tStatisBlockPut(writer->staticBlock, row, writer->config->maxRow);
82,025,963✔
968
    if (code == TSDB_CODE_INVALID_PARA) {
79,536,252!
969
      TAOS_CHECK_GOTO(tsdbSttFileDoWriteStatisBlock(writer), &lino, _exit);
×
970
      continue;
1,337✔
971
    } else {
972
      TSDB_CHECK_CODE(code, lino, _exit);
79,548,114!
973
    }
974
    break;
79,548,114✔
975
  }
976

977
  if (row->row.type == TSDBROW_ROW_FMT) {
79,548,114✔
978
    TAOS_CHECK_GOTO(tsdbUpdateSkmRow(writer->config->tsdb, writer->ctx->tbid,  //
79,389,805!
979
                                     TSDBROW_SVERSION(&row->row), writer->config->skmRow),
980
                    &lino, _exit);
981
  }
982

983
  // row to col conversion
984
  if (key.version <= writer->config->compactVersion                                //
79,915,606✔
985
      && writer->blockData->nRow > 0                                               //
79,907,593✔
986
      && (writer->blockData->uid                                                   //
158,646,918✔
987
              ? writer->blockData->uid                                             //
988
              : writer->blockData->aUid[writer->blockData->nRow - 1]) == row->uid  //
79,323,459✔
989
      && tsdbRowCompareWithoutVersion(&row->row,
79,180,366✔
990
                                      &tsdbRowFromBlockData(writer->blockData, writer->blockData->nRow - 1)) == 0  //
78,565,199✔
991
  ) {
992
    TAOS_CHECK_GOTO(tBlockDataUpdateRow(writer->blockData, &row->row, writer->config->skmRow->pTSchema), &lino, _exit);
4,260,443!
993
  } else {
994
    if (writer->blockData->nRow >= writer->config->maxRow) {
76,270,330✔
995
      TAOS_CHECK_GOTO(tsdbSttFileDoWriteBlockData(writer), &lino, _exit);
19,263!
996
    }
997

998
    TAOS_CHECK_GOTO(tBlockDataAppendRow(writer->blockData, &row->row, writer->config->skmRow->pTSchema, row->uid),
76,270,328!
999
                    &lino, _exit);
1000
  }
1001

1002
_exit:
82,856,313✔
1003
  if (code) {
82,856,313!
1004
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
1005
              tstrerror(code));
1006
  }
1007
  return code;
82,845,214✔
1008
}
1009

1010
int32_t tsdbSttFileWriteBlockData(SSttFileWriter *writer, SBlockData *bdata) {
9,832✔
1011
  int32_t code = 0;
9,832✔
1012
  int32_t lino = 0;
9,832✔
1013

1014
  SRowInfo row[1];
1015
  row->suid = bdata->suid;
9,832✔
1016
  for (int32_t i = 0; i < bdata->nRow; i++) {
100,388✔
1017
    row->uid = bdata->uid ? bdata->uid : bdata->aUid[i];
90,555!
1018
    row->row = tsdbRowFromBlockData(bdata, i);
90,555✔
1019

1020
    TAOS_CHECK_GOTO(tsdbSttFileWriteRow(writer, row), &lino, _exit);
90,555!
1021
  }
1022

1023
_exit:
9,833✔
1024
  if (code) {
9,833!
1025
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
1026
              tstrerror(code));
1027
  }
1028
  return code;
9,833✔
1029
}
1030

1031
int32_t tsdbSttFileWriteTombRecord(SSttFileWriter *writer, const STombRecord *record) {
15,430✔
1032
  int32_t code;
1033
  int32_t lino;
1034

1035
  if (!writer->ctx->opened) {
15,430✔
1036
    TAOS_CHECK_GOTO(tsdbSttFWriterDoOpen(writer), &lino, _exit);
107!
1037
  } else {
1038
    if (writer->blockData->nRow > 0) {
15,323✔
1039
      TAOS_CHECK_GOTO(tsdbSttFileDoWriteBlockData(writer), &lino, _exit);
4,341!
1040
    }
1041

1042
    if (STATIS_BLOCK_SIZE(writer->staticBlock) > 0) {
15,323✔
1043
      TAOS_CHECK_GOTO(tsdbSttFileDoWriteStatisBlock(writer), &lino, _exit);
4,341!
1044
    }
1045
  }
1046

1047
  TAOS_CHECK_GOTO(tTombBlockPut(writer->tombBlock, record), &lino, _exit);
15,430!
1048

1049
  if (TOMB_BLOCK_SIZE(writer->tombBlock) >= writer->config->maxRow) {
15,432!
1050
    TAOS_CHECK_GOTO(tsdbSttFileDoWriteTombBlock(writer), &lino, _exit);
×
1051
  }
1052

1053
_exit:
15,432✔
1054
  if (code) {
15,432!
1055
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, __FILE__, lino,
×
1056
              tstrerror(code));
1057
  } else {
1058
    tsdbTrace("vgId:%d write tomb record to stt file:%s, cid:%" PRId64 ", suid:%" PRId64 ", uid:%" PRId64
15,432✔
1059
              ", version:%" PRId64,
1060
              TD_VID(writer->config->tsdb->pVnode), writer->fd->path, writer->config->cid, record->suid, record->uid,
1061
              record->version);
1062
  }
1063
  return code;
15,432✔
1064
}
1065

1066
bool tsdbSttFileWriterIsOpened(SSttFileWriter *writer) { return writer->ctx->opened; }
8,573✔
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