tsdbSttFileRW.c 27.1 KB
Newer Older
H
Hongze Cheng 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

H
Hongze Cheng 已提交
16
#include "inc/tsdbSttFileRW.h"
H
Hongze Cheng 已提交
17

H
Hongze Cheng 已提交
18 19
typedef struct {
  int64_t   prevFooter;
H
Hongze Cheng 已提交
20 21 22 23 24
  SFDataPtr sttBlkPtr[1];
  SFDataPtr delBlkPtr[1];
  SFDataPtr statisBlkPtr[1];
  SFDataPtr rsrvd[2];
} SSttFooter;
H
Hongze Cheng 已提交
25

H
Hongze Cheng 已提交
26
// SSttFReader ============================================================
H
Hongze Cheng 已提交
27
struct SSttFileReader {
H
Hongze Cheng 已提交
28
  SSttFileReaderConfig config[1];
H
Hongze Cheng 已提交
29
  TSttSegReaderArray   readerArray[1];
H
Hongze Cheng 已提交
30
  STsdbFD             *fd;
H
Hongze Cheng 已提交
31
  uint8_t             *bufArr[5];
H
Hongze Cheng 已提交
32 33 34 35
};

struct SSttSegReader {
  SSttFileReader *reader;
H
Hongze Cheng 已提交
36
  SSttFooter      footer[1];
H
Hongze Cheng 已提交
37 38 39 40
  struct {
    bool sttBlkLoaded;
    bool delBlkLoaded;
    bool statisBlkLoaded;
H
Hongze Cheng 已提交
41
  } ctx[1];
H
Hongze Cheng 已提交
42 43
  TSttBlkArray    sttBlkArray[1];
  TStatisBlkArray statisBlkArray[1];
H
Hongze Cheng 已提交
44
  TDelBlkArray    delBlkArray[1];
H
Hongze Cheng 已提交
45 46
};

H
Hongze Cheng 已提交
47
// SSttFileReader
H
Hongze Cheng 已提交
48
static int32_t tsdbSttSegReaderOpen(SSttFileReader *reader, int64_t offset, SSttSegReader **segReader) {
H
Hongze Cheng 已提交
49 50
  ASSERT(offset >= TSDB_FHDR_SIZE);

H
Hongze Cheng 已提交
51
  int32_t code = 0;
H
Hongze Cheng 已提交
52
  int32_t lino = 0;
H
Hongze Cheng 已提交
53
  int32_t vid = TD_VID(reader->config->tsdb->pVnode);
H
Hongze Cheng 已提交
54 55

  segReader[0] = taosMemoryCalloc(1, sizeof(*segReader[0]));
H
Hongze Cheng 已提交
56
  if (!segReader[0]) return TSDB_CODE_OUT_OF_MEMORY;
H
Hongze Cheng 已提交
57 58

  segReader[0]->reader = reader;
H
Hongze Cheng 已提交
59
  code = tsdbReadFile(reader->fd, offset, (uint8_t *)(segReader[0]->footer), sizeof(SSttFooter));
H
Hongze Cheng 已提交
60 61 62 63
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  if (code) {
H
Hongze Cheng 已提交
64
    TSDB_ERROR_LOG(vid, lino, code);
H
Hongze Cheng 已提交
65 66 67
    taosMemoryFree(segReader[0]);
    segReader[0] = NULL;
  }
H
Hongze Cheng 已提交
68 69 70
  return code;
}

H
Hongze Cheng 已提交
71
static int32_t tsdbSttSegReaderClose(SSttSegReader **reader) {
H
Hongze Cheng 已提交
72
  if (reader[0]) {
H
Hongze Cheng 已提交
73 74 75
    TARRAY2_FREE(reader[0]->sttBlkArray);
    TARRAY2_FREE(reader[0]->delBlkArray);
    TARRAY2_FREE(reader[0]->statisBlkArray);
H
Hongze Cheng 已提交
76 77
    taosMemoryFree(reader[0]);
    reader[0] = NULL;
H
Hongze Cheng 已提交
78 79
  }
  return 0;
H
Hongze Cheng 已提交
80 81
}

H
Hongze Cheng 已提交
82
int32_t tsdbSttFileReaderOpen(const char *fname, const SSttFileReaderConfig *config, SSttFileReader **reader) {
H
Hongze Cheng 已提交
83
  int32_t code = 0;
H
Hongze Cheng 已提交
84 85 86 87 88 89
  int32_t lino = 0;
  int32_t vid = TD_VID(config->tsdb->pVnode);

  reader[0] = taosMemoryCalloc(1, sizeof(*reader[0]));
  if (reader[0] == NULL) return TSDB_CODE_OUT_OF_MEMORY;

H
Hongze Cheng 已提交
90
  reader[0]->config[0] = config[0];
H
Hongze Cheng 已提交
91
  if (!reader[0]->config->bufArr) reader[0]->config->bufArr = reader[0]->bufArr;
H
Hongze Cheng 已提交
92 93 94 95 96 97

  // open file
  code = tsdbOpenFile(fname, config->szPage, TD_FILE_READ, &reader[0]->fd);
  TSDB_CHECK_CODE(code, lino, _exit);

  // open each segment reader
H
Hongze Cheng 已提交
98
  int64_t size = config->file->size;
H
Hongze Cheng 已提交
99
  while (size > 0) {
H
Hongze Cheng 已提交
100
    SSttSegReader *reader1;
H
Hongze Cheng 已提交
101

H
Hongze Cheng 已提交
102
    code = tsdbSttSegReaderOpen(reader[0], size - sizeof(SSttFooter), &reader1);
H
Hongze Cheng 已提交
103 104
    TSDB_CHECK_CODE(code, lino, _exit);

H
Hongze Cheng 已提交
105
    code = TARRAY2_APPEND(reader[0]->readerArray, reader1);
H
Hongze Cheng 已提交
106 107
    TSDB_CHECK_CODE(code, lino, _exit);

H
Hongze Cheng 已提交
108
    size = reader1->footer->prevFooter;
H
Hongze Cheng 已提交
109 110
  }

H
Hongze Cheng 已提交
111
  ASSERT(TARRAY2_SIZE(reader[0]->readerArray) == config->file->stt->nseg);
H
Hongze Cheng 已提交
112 113 114

_exit:
  if (code) {
H
Hongze Cheng 已提交
115
    TSDB_ERROR_LOG(vid, lino, code);
H
Hongze Cheng 已提交
116
    tsdbSttFileReaderClose(reader);
H
Hongze Cheng 已提交
117
  }
H
Hongze Cheng 已提交
118 119 120
  return code;
}

H
Hongze Cheng 已提交
121
int32_t tsdbSttFileReaderClose(SSttFileReader **reader) {
H
Hongze Cheng 已提交
122
  if (reader[0]) {
H
Hongze Cheng 已提交
123 124 125
    for (int32_t i = 0; i < ARRAY_SIZE(reader[0]->bufArr); ++i) {
      tFree(reader[0]->bufArr[i]);
    }
H
Hongze Cheng 已提交
126 127 128 129 130
    tsdbCloseFile(&reader[0]->fd);
    TARRAY2_CLEAR_FREE(reader[0]->readerArray, tsdbSttSegReaderClose);
    taosMemoryFree(reader[0]);
    reader[0] = NULL;
  }
H
Hongze Cheng 已提交
131
  return 0;
H
Hongze Cheng 已提交
132 133
}

H
Hongze Cheng 已提交
134
int32_t tsdbSttFileReaderGetSegReader(SSttFileReader *reader, const TSttSegReaderArray **readerArray) {
H
Hongze Cheng 已提交
135
  readerArray[0] = reader->readerArray;
H
Hongze Cheng 已提交
136
  return 0;
H
Hongze Cheng 已提交
137 138
}

H
Hongze Cheng 已提交
139
// SSttFSegReader
H
Hongze Cheng 已提交
140
int32_t tsdbSttFileReadStatisBlk(SSttSegReader *reader, const TStatisBlkArray **statisBlkArray) {
H
Hongze Cheng 已提交
141
  if (!reader->ctx->statisBlkLoaded) {
H
Hongze Cheng 已提交
142
    if (reader->footer->statisBlkPtr->size > 0) {
H
Hongze Cheng 已提交
143
      ASSERT(reader->footer->statisBlkPtr->size % sizeof(SStatisBlk) == 0);
H
Hongze Cheng 已提交
144

H
Hongze Cheng 已提交
145
      int32_t size = reader->footer->statisBlkPtr->size / sizeof(SStatisBlk);
H
Hongze Cheng 已提交
146
      void   *data = taosMemoryMalloc(reader->footer->statisBlkPtr->size);
H
Hongze Cheng 已提交
147 148
      if (!data) return TSDB_CODE_OUT_OF_MEMORY;

H
Hongze Cheng 已提交
149 150
      int32_t code = tsdbReadFile(reader->reader->fd, reader->footer->statisBlkPtr->offset, data,
                                  reader->footer->statisBlkPtr->size);
H
Hongze Cheng 已提交
151 152 153 154
      if (code) {
        taosMemoryFree(data);
        return code;
      }
H
Hongze Cheng 已提交
155

H
Hongze Cheng 已提交
156
      TARRAY2_INIT_EX(reader->statisBlkArray, size, size, data);
H
Hongze Cheng 已提交
157
    } else {
H
Hongze Cheng 已提交
158
      TARRAY2_INIT(reader->statisBlkArray);
H
Hongze Cheng 已提交
159 160
    }

H
Hongze Cheng 已提交
161
    reader->ctx->statisBlkLoaded = true;
H
Hongze Cheng 已提交
162 163
  }

H
Hongze Cheng 已提交
164
  statisBlkArray[0] = reader->statisBlkArray;
H
Hongze Cheng 已提交
165
  return 0;
H
Hongze Cheng 已提交
166 167
}

H
Hongze Cheng 已提交
168
int32_t tsdbSttFileReadDelBlk(SSttSegReader *reader, const TDelBlkArray **delBlkArray) {
H
Hongze Cheng 已提交
169
  if (!reader->ctx->delBlkLoaded) {
H
Hongze Cheng 已提交
170 171
    if (reader->footer->delBlkPtr->size > 0) {
      ASSERT(reader->footer->delBlkPtr->size % sizeof(SDelBlk) == 0);
H
Hongze Cheng 已提交
172

H
Hongze Cheng 已提交
173 174
      int32_t size = reader->footer->delBlkPtr->size / sizeof(SDelBlk);
      void   *data = taosMemoryMalloc(reader->footer->delBlkPtr->size);
H
Hongze Cheng 已提交
175 176
      if (!data) return TSDB_CODE_OUT_OF_MEMORY;

H
Hongze Cheng 已提交
177 178
      int32_t code =
          tsdbReadFile(reader->reader->fd, reader->footer->delBlkPtr->offset, data, reader->footer->delBlkPtr->size);
H
Hongze Cheng 已提交
179 180 181 182
      if (code) {
        taosMemoryFree(data);
        return code;
      }
H
Hongze Cheng 已提交
183

H
Hongze Cheng 已提交
184
      TARRAY2_INIT_EX(reader->delBlkArray, size, size, data);
H
Hongze Cheng 已提交
185
    } else {
H
Hongze Cheng 已提交
186
      TARRAY2_INIT(reader->delBlkArray);
H
Hongze Cheng 已提交
187 188
    }

H
Hongze Cheng 已提交
189
    reader->ctx->delBlkLoaded = true;
H
Hongze Cheng 已提交
190 191
  }

H
Hongze Cheng 已提交
192
  delBlkArray[0] = reader->delBlkArray;
H
Hongze Cheng 已提交
193 194 195
  return 0;
}

H
Hongze Cheng 已提交
196
int32_t tsdbSttFileReadSttBlk(SSttSegReader *reader, const TSttBlkArray **sttBlkArray) {
H
Hongze Cheng 已提交
197
  if (!reader->ctx->sttBlkLoaded) {
H
Hongze Cheng 已提交
198 199
    if (reader->footer->sttBlkPtr->size > 0) {
      ASSERT(reader->footer->sttBlkPtr->size % sizeof(SSttBlk) == 0);
H
Hongze Cheng 已提交
200

H
Hongze Cheng 已提交
201 202
      int32_t size = reader->footer->sttBlkPtr->size / sizeof(SSttBlk);
      void   *data = taosMemoryMalloc(reader->footer->sttBlkPtr->size);
H
Hongze Cheng 已提交
203 204
      if (!data) return TSDB_CODE_OUT_OF_MEMORY;

H
Hongze Cheng 已提交
205 206
      int32_t code =
          tsdbReadFile(reader->reader->fd, reader->footer->sttBlkPtr->offset, data, reader->footer->sttBlkPtr->size);
H
Hongze Cheng 已提交
207 208 209 210
      if (code) {
        taosMemoryFree(data);
        return code;
      }
H
Hongze Cheng 已提交
211

H
Hongze Cheng 已提交
212
      TARRAY2_INIT_EX(reader->sttBlkArray, size, size, data);
H
Hongze Cheng 已提交
213
    } else {
H
Hongze Cheng 已提交
214
      TARRAY2_INIT(reader->sttBlkArray);
H
Hongze Cheng 已提交
215 216
    }

H
Hongze Cheng 已提交
217
    reader->ctx->sttBlkLoaded = true;
H
Hongze Cheng 已提交
218 219
  }

H
Hongze Cheng 已提交
220
  sttBlkArray[0] = reader->sttBlkArray;
H
Hongze Cheng 已提交
221
  return 0;
H
Hongze Cheng 已提交
222 223
}

H
Hongze Cheng 已提交
224
int32_t tsdbSttFileReadDataBlock(SSttSegReader *reader, const SSttBlk *sttBlk, SBlockData *bData) {
H
Hongze Cheng 已提交
225
  int32_t code = 0;
H
Hongze Cheng 已提交
226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242
  int32_t lino = 0;

  code = tRealloc(&reader->reader->config->bufArr[0], sttBlk->bInfo.szBlock);
  TSDB_CHECK_CODE(code, lino, _exit);

  code =
      tsdbReadFile(reader->reader->fd, sttBlk->bInfo.offset, reader->reader->config->bufArr[0], sttBlk->bInfo.szBlock);
  TSDB_CHECK_CODE(code, lino, _exit);

  code = tDecmprBlockData(reader->reader->config->bufArr[0], sttBlk->bInfo.szBlock, bData,
                          &reader->reader->config->bufArr[1]);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  if (code) {
    TSDB_ERROR_LOG(TD_VID(reader->reader->config->tsdb->pVnode), lino, code);
  }
H
Hongze Cheng 已提交
243 244 245
  return code;
}

H
Hongze Cheng 已提交
246
int32_t tsdbSttFileReadDelBlock(SSttSegReader *reader, const SDelBlk *delBlk, SDelBlock *dData) {
H
Hongze Cheng 已提交
247
  int32_t code = 0;
H
Hongze Cheng 已提交
248 249 250
  int32_t lino = 0;

  tDelBlockClear(dData);
H
Hongze Cheng 已提交
251 252

  code = tRealloc(&reader->reader->config->bufArr[0], delBlk->dp->size);
H
Hongze Cheng 已提交
253 254
  TSDB_CHECK_CODE(code, lino, _exit);

H
Hongze Cheng 已提交
255
  code = tsdbReadFile(reader->reader->fd, delBlk->dp->offset, reader->reader->config->bufArr[0], delBlk->dp->size);
H
Hongze Cheng 已提交
256 257 258
  if (code) TSDB_CHECK_CODE(code, lino, _exit);

  int64_t size = 0;
H
Hongze Cheng 已提交
259
  for (int32_t i = 0; i < ARRAY_SIZE(dData->dataArr); ++i) {
H
Hongze Cheng 已提交
260 261 262
    code = tsdbDecmprData(reader->reader->config->bufArr[0] + size, delBlk->size[i], TSDB_DATA_TYPE_BIGINT,
                          TWO_STAGE_COMP, &reader->reader->config->bufArr[1], sizeof(int64_t) * delBlk->numRec,
                          &reader->reader->config->bufArr[2]);
H
Hongze Cheng 已提交
263 264 265 266 267 268 269 270
    TSDB_CHECK_CODE(code, lino, _exit);

    size += delBlk->size[i];
  }

  ASSERT(size == delBlk->dp->size);
_exit:
  if (code) {
H
Hongze Cheng 已提交
271
    TSDB_ERROR_LOG(TD_VID(reader->reader->config->tsdb->pVnode), lino, code);
H
Hongze Cheng 已提交
272
  }
H
Hongze Cheng 已提交
273 274 275
  return code;
}

H
Hongze Cheng 已提交
276
int32_t tsdbSttFileReadStatisBlock(SSttSegReader *reader, const SStatisBlk *statisBlk, STbStatisBlock *sData) {
H
Hongze Cheng 已提交
277
  int32_t code = 0;
H
Hongze Cheng 已提交
278 279 280
  int32_t lino = 0;

  tStatisBlockClear(sData);
H
Hongze Cheng 已提交
281 282

  code = tRealloc(&reader->reader->config->bufArr[0], statisBlk->dp->size);
H
Hongze Cheng 已提交
283 284
  TSDB_CHECK_CODE(code, lino, _exit);

H
Hongze Cheng 已提交
285 286
  code =
      tsdbReadFile(reader->reader->fd, statisBlk->dp->offset, reader->reader->config->bufArr[0], statisBlk->dp->size);
H
Hongze Cheng 已提交
287 288 289
  if (code) TSDB_CHECK_CODE(code, lino, _exit);

  int64_t size = 0;
H
Hongze Cheng 已提交
290
  for (int32_t i = 0; i < ARRAY_SIZE(sData->dataArr); ++i) {
H
Hongze Cheng 已提交
291 292 293
    code = tsdbDecmprData(reader->reader->config->bufArr[0] + size, statisBlk->size[i], TSDB_DATA_TYPE_BIGINT,
                          TWO_STAGE_COMP, &reader->reader->config->bufArr[1], sizeof(int64_t) * statisBlk->numRec,
                          &reader->reader->config->bufArr[2]);
H
Hongze Cheng 已提交
294 295
    TSDB_CHECK_CODE(code, lino, _exit);

H
Hongze Cheng 已提交
296 297 298 299 300
    for (int32_t j = 0; j < statisBlk->numRec; ++j) {
      code = TARRAY2_APPEND(sData->dataArr + i, ((int64_t *)reader->reader->config->bufArr[1])[j]);
      TSDB_CHECK_CODE(code, lino, _exit);
    }

H
Hongze Cheng 已提交
301 302 303 304 305 306 307
    size += statisBlk->size[i];
  }

  ASSERT(size == statisBlk->dp->size);

_exit:
  if (code) {
H
Hongze Cheng 已提交
308
    TSDB_ERROR_LOG(TD_VID(reader->reader->config->tsdb->pVnode), lino, code);
H
Hongze Cheng 已提交
309
  }
H
Hongze Cheng 已提交
310 311 312
  return code;
}

H
Hongze Cheng 已提交
313
// SSttFWriter ============================================================
H
Hongze Cheng 已提交
314
struct SSttFileWriter {
H
Hongze Cheng 已提交
315
  SSttFileWriterConfig config[1];
H
Hongze Cheng 已提交
316
  struct {
H
Hongze Cheng 已提交
317 318
    bool    opened;
    TABLEID tbid[1];
H
Hongze Cheng 已提交
319
  } ctx[1];
H
Hongze Cheng 已提交
320
  // file
H
Hongze Cheng 已提交
321
  STFile file[1];
H
Hongze Cheng 已提交
322
  // data
H
Hongze Cheng 已提交
323 324
  TSttBlkArray    sttBlkArray[1];
  TStatisBlkArray statisBlkArray[1];
H
Hongze Cheng 已提交
325
  TDelBlkArray    delBlkArray[1];
H
Hongze Cheng 已提交
326
  SSttFooter      footer[1];
H
Hongze Cheng 已提交
327 328
  SBlockData      bData[1];
  STbStatisBlock  sData[1];
H
Hongze Cheng 已提交
329
  SDelBlock       dData[1];
H
Hongze Cheng 已提交
330
  // helper data
H
Hongze Cheng 已提交
331 332
  SSkmInfo skmTb[1];
  SSkmInfo skmRow[1];
H
Hongze Cheng 已提交
333 334
  int32_t  sizeArr[5];
  uint8_t *bufArr[5];
H
Hongze Cheng 已提交
335
  STsdbFD *fd;
H
Hongze Cheng 已提交
336
};
H
Hongze Cheng 已提交
337

H
Hongze Cheng 已提交
338 339 340
static int32_t tsdbSttFileDoWriteTSDataBlock(SSttFileWriter *writer) {
  if (writer->bData->nRow == 0) return 0;

H
Hongze Cheng 已提交
341
  int32_t code = 0;
H
Hongze Cheng 已提交
342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357
  int32_t lino = 0;
  SSttBlk sttBlk[1];

  sttBlk->suid = writer->bData->suid;
  sttBlk->minUid = writer->bData->uid ? writer->bData->uid : writer->bData->aUid[0];
  sttBlk->maxUid = writer->bData->uid ? writer->bData->uid : writer->bData->aUid[writer->bData->nRow - 1];
  sttBlk->minKey = sttBlk->maxKey = writer->bData->aTSKEY[0];
  sttBlk->minVer = sttBlk->maxVer = writer->bData->aVersion[0];
  sttBlk->nRow = writer->bData->nRow;
  for (int32_t iRow = 1; iRow < writer->bData->nRow; iRow++) {
    if (sttBlk->minKey > writer->bData->aTSKEY[iRow]) sttBlk->minKey = writer->bData->aTSKEY[iRow];
    if (sttBlk->maxKey < writer->bData->aTSKEY[iRow]) sttBlk->maxKey = writer->bData->aTSKEY[iRow];
    if (sttBlk->minVer > writer->bData->aVersion[iRow]) sttBlk->minVer = writer->bData->aVersion[iRow];
    if (sttBlk->maxVer < writer->bData->aVersion[iRow]) sttBlk->maxVer = writer->bData->aVersion[iRow];
  }

H
Hongze Cheng 已提交
358
  code = tCmprBlockData(writer->bData, writer->config->cmprAlg, NULL, NULL, writer->config->aBuf, writer->sizeArr);
H
Hongze Cheng 已提交
359
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
360

H
Hongze Cheng 已提交
361
  sttBlk->bInfo.offset = writer->file->size;
H
Hongze Cheng 已提交
362 363
  sttBlk->bInfo.szKey = writer->sizeArr[2] + writer->sizeArr[3];
  sttBlk->bInfo.szBlock = writer->sizeArr[0] + writer->sizeArr[1] + sttBlk->bInfo.szKey;
H
Hongze Cheng 已提交
364

H
Hongze Cheng 已提交
365
  for (int32_t i = 3; i >= 0; i--) {
H
Hongze Cheng 已提交
366 367
    if (writer->sizeArr[i]) {
      code = tsdbWriteFile(writer->fd, writer->file->size, writer->config->aBuf[i], writer->sizeArr[i]);
H
Hongze Cheng 已提交
368
      TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
369
      writer->file->size += writer->sizeArr[i];
H
Hongze Cheng 已提交
370 371 372
    }
  }

H
Hongze Cheng 已提交
373
  code = TARRAY2_APPEND_PTR(writer->sttBlkArray, sttBlk);
H
Hongze Cheng 已提交
374
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
375

H
Hongze Cheng 已提交
376 377
  tBlockDataClear(writer->bData);

H
Hongze Cheng 已提交
378 379
_exit:
  if (code) {
H
Hongze Cheng 已提交
380
    TSDB_ERROR_LOG(TD_VID(writer->config->tsdb->pVnode), lino, code);
H
Hongze Cheng 已提交
381
  }
H
Hongze Cheng 已提交
382 383 384
  return code;
}

H
Hongze Cheng 已提交
385
static int32_t tsdbSttFileDoWriteStatisBlock(SSttFileWriter *writer) {
H
Hongze Cheng 已提交
386
  if (STATIS_BLOCK_SIZE(writer->sData)) return 0;
H
Hongze Cheng 已提交
387

H
Hongze Cheng 已提交
388
  int32_t code = 0;
H
Hongze Cheng 已提交
389
  int32_t lino = 0;
H
Hongze Cheng 已提交
390

H
Hongze Cheng 已提交
391
  SStatisBlk statisBlk[1] = {{
H
Hongze Cheng 已提交
392
      .numRec = STATIS_BLOCK_SIZE(writer->sData),
H
Hongze Cheng 已提交
393 394 395 396 397 398 399 400 401 402
      .minTbid =
          {
              .suid = TARRAY2_FIRST(writer->sData->suid),
              .uid = TARRAY2_FIRST(writer->sData->uid),
          },
      .maxTbid =
          {
              .suid = TARRAY2_LAST(writer->sData->suid),
              .uid = TARRAY2_LAST(writer->sData->uid),
          },
H
Hongze Cheng 已提交
403 404
      .minVer = TARRAY2_FIRST(writer->sData->minVer),
      .maxVer = TARRAY2_FIRST(writer->sData->maxVer),
H
Hongze Cheng 已提交
405 406
  }};

H
Hongze Cheng 已提交
407 408 409 410
  for (int32_t i = 1; i < STATIS_BLOCK_SIZE(writer->sData); i++) {
    statisBlk->minVer = TMIN(statisBlk->minVer, TARRAY2_GET(writer->sData->minVer, i));
    statisBlk->maxVer = TMAX(statisBlk->maxVer, TARRAY2_GET(writer->sData->maxVer, i));
  }
H
Hongze Cheng 已提交
411

H
Hongze Cheng 已提交
412 413
  statisBlk->dp->offset = writer->file->size;
  statisBlk->dp->size = 0;
H
Hongze Cheng 已提交
414

H
Hongze Cheng 已提交
415
  for (int32_t i = 0; i < STATIS_RECORD_NUM_ELEM; i++) {
H
Hongze Cheng 已提交
416
    int32_t size;
H
Hongze Cheng 已提交
417 418 419
    code = tsdbCmprData((uint8_t *)TARRAY2_DATA(&writer->sData->dataArr[i]),
                        TARRAY2_DATA_LEN(&writer->sData->dataArr[i]), TSDB_DATA_TYPE_BIGINT, TWO_STAGE_COMP,
                        &writer->config->aBuf[0], 0, &size, &writer->config->aBuf[1]);
H
Hongze Cheng 已提交
420
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
421 422 423 424

    code = tsdbWriteFile(writer->fd, writer->file->size, writer->config->aBuf[0], size);
    TSDB_CHECK_CODE(code, lino, _exit);

H
Hongze Cheng 已提交
425 426
    statisBlk->size[i] = size;
    statisBlk->dp->size += size;
H
Hongze Cheng 已提交
427
    writer->file->size += size;
H
Hongze Cheng 已提交
428
  }
H
Hongze Cheng 已提交
429

H
Hongze Cheng 已提交
430
  code = TARRAY2_APPEND_PTR(writer->statisBlkArray, statisBlk);
H
Hongze Cheng 已提交
431
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
432

H
Hongze Cheng 已提交
433 434
  tStatisBlockClear(writer->sData);

H
Hongze Cheng 已提交
435 436
_exit:
  if (code) {
H
Hongze Cheng 已提交
437
    TSDB_ERROR_LOG(TD_VID(writer->config->tsdb->pVnode), lino, code);
H
Hongze Cheng 已提交
438
  }
H
Hongze Cheng 已提交
439 440 441
  return code;
}

H
Hongze Cheng 已提交
442
static int32_t tsdbSttFileDoWriteDelBlock(SSttFileWriter *writer) {
H
Hongze Cheng 已提交
443
  return 0;
H
Hongze Cheng 已提交
444
#if 0
H
Hongze Cheng 已提交
445 446
  if (writer->dData->nRow == 0) return 0;

H
Hongze Cheng 已提交
447
  int32_t code = 0;
H
Hongze Cheng 已提交
448 449
  int32_t lino;

H
Hongze Cheng 已提交
450
  SDelBlk delBlk[1];
H
Hongze Cheng 已提交
451

H
Hongze Cheng 已提交
452 453 454 455 456 457 458 459 460
  delBlk->nRow = writer->sData->nRow;
  delBlk->minTid.suid = writer->sData->aData[0][0];
  delBlk->minTid.uid = writer->sData->aData[1][0];
  delBlk->maxTid.suid = writer->sData->aData[0][writer->sData->nRow - 1];
  delBlk->maxTid.uid = writer->sData->aData[1][writer->sData->nRow - 1];
  delBlk->minVer = delBlk->maxVer = delBlk->maxVer = writer->sData->aData[2][0];
  for (int32_t iRow = 1; iRow < writer->sData->nRow; iRow++) {
    if (delBlk->minVer > writer->sData->aData[2][iRow]) delBlk->minVer = writer->sData->aData[2][iRow];
    if (delBlk->maxVer < writer->sData->aData[2][iRow]) delBlk->maxVer = writer->sData->aData[2][iRow];
H
Hongze Cheng 已提交
461 462
  }

H
Hongze Cheng 已提交
463
  delBlk->dp.offset = writer->file->size;
H
Hongze Cheng 已提交
464
  delBlk->dp.size = 0;  // TODO
H
Hongze Cheng 已提交
465

H
Hongze Cheng 已提交
466 467
  int64_t tsize = sizeof(int64_t) * writer->dData->nRow;
  for (int32_t i = 0; i < ARRAY_SIZE(writer->dData->aData); i++) {
H
Hongze Cheng 已提交
468
    code = tsdbWriteFile(writer->fd, writer->file->size, (const uint8_t *)writer->dData->aData[i], tsize);
H
Hongze Cheng 已提交
469 470
    TSDB_CHECK_CODE(code, lino, _exit);

H
Hongze Cheng 已提交
471
    delBlk->dp.size += tsize;
H
Hongze Cheng 已提交
472
    writer->file->size += tsize;
H
Hongze Cheng 已提交
473
  }
H
Hongze Cheng 已提交
474
  tDelBlockDestroy(writer->dData);
H
Hongze Cheng 已提交
475

H
Hongze Cheng 已提交
476
  code = TARRAY2_APPEND_PTR(writer->delBlkArray, delBlk);
H
Hongze Cheng 已提交
477
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
478 479 480

_exit:
  if (code) {
H
Hongze Cheng 已提交
481
    tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, lino,
H
Hongze Cheng 已提交
482 483 484 485
              tstrerror(code));
  } else {
    // tsdbTrace();
  }
H
Hongze Cheng 已提交
486
  return code;
H
Hongze Cheng 已提交
487
#endif
H
Hongze Cheng 已提交
488 489
}

H
Hongze Cheng 已提交
490
static int32_t tsdbSttFileDoWriteSttBlk(SSttFileWriter *writer) {
H
Hongze Cheng 已提交
491
  int32_t code = 0;
H
Hongze Cheng 已提交
492 493
  int32_t lino;

H
Hongze Cheng 已提交
494
  writer->footer->sttBlkPtr->offset = writer->file->size;
H
Hongze Cheng 已提交
495
  writer->footer->sttBlkPtr->size = TARRAY2_DATA_LEN(writer->sttBlkArray);
H
Hongze Cheng 已提交
496

H
Hongze Cheng 已提交
497
  if (writer->footer->sttBlkPtr->size) {
H
Hongze Cheng 已提交
498
    code = tsdbWriteFile(writer->fd, writer->file->size, (const uint8_t *)TARRAY2_DATA(writer->sttBlkArray),
H
Hongze Cheng 已提交
499
                         writer->footer->sttBlkPtr->size);
H
Hongze Cheng 已提交
500
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
501
    writer->file->size += writer->footer->sttBlkPtr->size;
H
Hongze Cheng 已提交
502 503 504 505
  }

_exit:
  if (code) {
H
Hongze Cheng 已提交
506
    TSDB_ERROR_LOG(TD_VID(writer->config->tsdb->pVnode), lino, code);
H
Hongze Cheng 已提交
507
  }
H
Hongze Cheng 已提交
508 509 510
  return code;
}

H
Hongze Cheng 已提交
511
static int32_t tsdbSttFileDoWriteStatisBlk(SSttFileWriter *writer) {
H
Hongze Cheng 已提交
512
  int32_t code = 0;
H
Hongze Cheng 已提交
513
  int32_t lino;
H
Hongze Cheng 已提交
514

H
Hongze Cheng 已提交
515
  writer->footer->statisBlkPtr->offset = writer->file->size;
H
Hongze Cheng 已提交
516
  writer->footer->statisBlkPtr->size = TARRAY2_DATA_LEN(writer->statisBlkArray);
H
Hongze Cheng 已提交
517

H
Hongze Cheng 已提交
518
  if (writer->footer->statisBlkPtr->size) {
H
Hongze Cheng 已提交
519
    code = tsdbWriteFile(writer->fd, writer->file->size, (const uint8_t *)TARRAY2_DATA(writer->statisBlkArray),
H
Hongze Cheng 已提交
520
                         writer->footer->statisBlkPtr->size);
H
Hongze Cheng 已提交
521
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
522
    writer->file->size += writer->footer->statisBlkPtr->size;
H
Hongze Cheng 已提交
523 524 525 526
  }

_exit:
  if (code) {
H
Hongze Cheng 已提交
527
    TSDB_ERROR_LOG(TD_VID(writer->config->tsdb->pVnode), lino, code);
H
Hongze Cheng 已提交
528
  }
H
Hongze Cheng 已提交
529 530 531
  return code;
}

H
Hongze Cheng 已提交
532
static int32_t tsdbSttFileDoWriteDelBlk(SSttFileWriter *writer) {
H
Hongze Cheng 已提交
533
  int32_t code = 0;
H
Hongze Cheng 已提交
534 535
  int32_t lino;

H
Hongze Cheng 已提交
536
  writer->footer->delBlkPtr->offset = writer->file->size;
H
Hongze Cheng 已提交
537
  writer->footer->delBlkPtr->size = TARRAY2_DATA_LEN(writer->delBlkArray);
H
Hongze Cheng 已提交
538

H
Hongze Cheng 已提交
539
  if (writer->footer->delBlkPtr->size) {
H
Hongze Cheng 已提交
540
    code = tsdbWriteFile(writer->fd, writer->file->size, (const uint8_t *)TARRAY2_DATA(writer->delBlkArray),
H
Hongze Cheng 已提交
541
                         writer->footer->delBlkPtr->size);
H
Hongze Cheng 已提交
542
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
543
    writer->file->size += writer->footer->delBlkPtr->size;
H
Hongze Cheng 已提交
544 545 546 547
  }

_exit:
  if (code) {
H
Hongze Cheng 已提交
548
    tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(writer->config->tsdb->pVnode), __func__, lino,
H
Hongze Cheng 已提交
549
              tstrerror(code));
H
Hongze Cheng 已提交
550 551 552 553
  }
  return code;
}

H
Hongze Cheng 已提交
554
static int32_t tsdbSttFileDoWriteFooter(SSttFileWriter *writer) {
H
Hongze Cheng 已提交
555 556
  writer->footer->prevFooter = writer->config->file.size;
  int32_t code = tsdbWriteFile(writer->fd, writer->file->size, (const uint8_t *)writer->footer, sizeof(writer->footer));
H
Hongze Cheng 已提交
557
  writer->file->size += sizeof(writer->footer);
H
Hongze Cheng 已提交
558 559 560
  return code;
}

H
Hongze Cheng 已提交
561
static int32_t tsdbSttFWriterDoOpen(SSttFileWriter *writer) {
H
Hongze Cheng 已提交
562
  int32_t code = 0;
H
Hongze Cheng 已提交
563
  int32_t lino = 0;
H
Hongze Cheng 已提交
564
  int32_t vid = TD_VID(writer->config->tsdb->pVnode);
H
Hongze Cheng 已提交
565

H
Hongze Cheng 已提交
566
  // set
H
Hongze Cheng 已提交
567 568 569 570
  writer->file[0] = writer->config->file;
  writer->file->stt->nseg++;
  if (!writer->config->skmTb) writer->config->skmTb = writer->skmTb;
  if (!writer->config->skmRow) writer->config->skmRow = writer->skmRow;
H
Hongze Cheng 已提交
571
  if (!writer->config->aBuf) writer->config->aBuf = writer->bufArr;
H
Hongze Cheng 已提交
572

H
Hongze Cheng 已提交
573 574 575
  // open file
  int32_t flag;
  char    fname[TSDB_FILENAME_LEN];
H
Hongze Cheng 已提交
576

H
Hongze Cheng 已提交
577
  if (writer->file->size) {
H
Hongze Cheng 已提交
578 579 580
    flag = TD_FILE_READ | TD_FILE_WRITE;
  } else {
    flag = TD_FILE_READ | TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
H
Hongze Cheng 已提交
581
  }
H
Hongze Cheng 已提交
582

H
Hongze Cheng 已提交
583
  tsdbTFileName(writer->config->tsdb, writer->file, fname);
H
Hongze Cheng 已提交
584
  code = tsdbOpenFile(fname, writer->config->szPage, flag, &writer->fd);
H
Hongze Cheng 已提交
585 586
  TSDB_CHECK_CODE(code, lino, _exit);

H
Hongze Cheng 已提交
587
  if (!writer->file->size) {
H
Hongze Cheng 已提交
588 589 590
    uint8_t hdr[TSDB_FHDR_SIZE] = {0};
    code = tsdbWriteFile(writer->fd, 0, hdr, sizeof(hdr));
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
591
    writer->file->size += sizeof(hdr);
H
Hongze Cheng 已提交
592
  }
H
Hongze Cheng 已提交
593

H
Hongze Cheng 已提交
594
_exit:
H
Hongze Cheng 已提交
595
  if (code) {
H
Hongze Cheng 已提交
596
    TSDB_ERROR_LOG(vid, lino, code);
H
Hongze Cheng 已提交
597
  } else {
H
Hongze Cheng 已提交
598
    writer->ctx->opened = true;
H
Hongze Cheng 已提交
599
  }
H
Hongze Cheng 已提交
600
  return code;
H
Hongze Cheng 已提交
601 602
}

H
Hongze Cheng 已提交
603 604
static void tsdbSttFWriterDoClose(SSttFileWriter *writer) {
  ASSERT(!writer->fd);
H
Hongze Cheng 已提交
605

H
Hongze Cheng 已提交
606 607
  for (int32_t i = 0; i < ARRAY_SIZE(writer->sizeArr); ++i) {
    tFree(writer->bufArr[i]);
H
Hongze Cheng 已提交
608 609 610 611 612 613 614
  }
  tDestroyTSchema(writer->skmRow->pTSchema);
  tDestroyTSchema(writer->skmTb->pTSchema);
  tStatisBlockFree(writer->sData);
  tDelBlockFree(writer->dData);
  tBlockDataDestroy(writer->bData);
  TARRAY2_FREE(writer->delBlkArray);
H
Hongze Cheng 已提交
615
  TARRAY2_FREE(writer->statisBlkArray);
H
Hongze Cheng 已提交
616
  TARRAY2_FREE(writer->sttBlkArray);
H
Hongze Cheng 已提交
617
}
H
Hongze Cheng 已提交
618

H
Hongze Cheng 已提交
619 620 621 622
static int32_t tsdbSttFileDoUpdateHeader(SSttFileWriter *writer) {
  // TODO
  return 0;
}
H
Hongze Cheng 已提交
623

H
Hongze Cheng 已提交
624 625 626
static int32_t tsdbSttFWriterCloseCommit(SSttFileWriter *writer, STFileOp *op) {
  int32_t lino;
  int32_t code;
H
Hongze Cheng 已提交
627
  int32_t vid = TD_VID(writer->config->tsdb->pVnode);
H
Hongze Cheng 已提交
628

H
Hongze Cheng 已提交
629 630
  code = tsdbSttFileDoWriteTSDataBlock(writer);
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
631

H
Hongze Cheng 已提交
632
  code = tsdbSttFileDoWriteStatisBlock(writer);
H
Hongze Cheng 已提交
633 634
  TSDB_CHECK_CODE(code, lino, _exit);

H
Hongze Cheng 已提交
635 636
  code = tsdbSttFileDoWriteDelBlock(writer);
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
637

H
Hongze Cheng 已提交
638 639
  code = tsdbSttFileDoWriteSttBlk(writer);
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
640

H
Hongze Cheng 已提交
641 642
  code = tsdbSttFileDoWriteStatisBlk(writer);
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
643

H
Hongze Cheng 已提交
644 645
  code = tsdbSttFileDoWriteDelBlk(writer);
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
646

H
Hongze Cheng 已提交
647 648
  code = tsdbSttFileDoWriteFooter(writer);
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
649

H
Hongze Cheng 已提交
650
  code = tsdbSttFileDoUpdateHeader(writer);
H
Hongze Cheng 已提交
651 652
  TSDB_CHECK_CODE(code, lino, _exit);

H
Hongze Cheng 已提交
653
  code = tsdbFsyncFile(writer->fd);
H
Hongze Cheng 已提交
654 655
  TSDB_CHECK_CODE(code, lino, _exit);

H
Hongze Cheng 已提交
656
  tsdbCloseFile(&writer->fd);
H
Hongze Cheng 已提交
657

H
Hongze Cheng 已提交
658
  ASSERT(writer->config->file.size < writer->file->size);
H
Hongze Cheng 已提交
659 660 661
  op->optype = writer->config->file.size ? TSDB_FOP_MODIFY : TSDB_FOP_CREATE;
  op->fid = writer->config->file.fid;
  op->of = writer->config->file;
H
Hongze Cheng 已提交
662
  op->nf = writer->file[0];
H
Hongze Cheng 已提交
663

H
Hongze Cheng 已提交
664 665
_exit:
  if (code) {
H
Hongze Cheng 已提交
666
    TSDB_ERROR_LOG(vid, lino, code);
H
Hongze Cheng 已提交
667
  }
H
Hongze Cheng 已提交
668
  return code;
H
Hongze Cheng 已提交
669
}
H
Hongze Cheng 已提交
670

H
Hongze Cheng 已提交
671
static int32_t tsdbSttFWriterCloseAbort(SSttFileWriter *writer) {
H
Hongze Cheng 已提交
672
  if (writer->config->file.size) {  // truncate the file to the original size
H
Hongze Cheng 已提交
673 674
    ASSERT(writer->config->file.size <= writer->file->size);
    if (writer->config->file.size < writer->file->size) {
H
Hongze Cheng 已提交
675
      taosFtruncateFile(writer->fd->pFD, writer->config->file.size);
H
Hongze Cheng 已提交
676 677 678
      tsdbCloseFile(&writer->fd);
    }
  } else {  // remove the file
H
Hongze Cheng 已提交
679 680
    char fname[TSDB_FILENAME_LEN];
    tsdbTFileName(writer->config->tsdb, &writer->config->file, fname);
H
Hongze Cheng 已提交
681 682 683 684
    tsdbCloseFile(&writer->fd);
    taosRemoveFile(fname);
  }

H
Hongze Cheng 已提交
685 686
  return 0;
}
H
Hongze Cheng 已提交
687

H
Hongze Cheng 已提交
688
int32_t tsdbSttFileWriterOpen(const SSttFileWriterConfig *config, SSttFileWriter **writer) {
H
Hongze Cheng 已提交
689
  writer[0] = taosMemoryCalloc(1, sizeof(*writer[0]));
H
Hongze Cheng 已提交
690 691
  if (writer[0] == NULL) return TSDB_CODE_OUT_OF_MEMORY;

H
Hongze Cheng 已提交
692
  writer[0]->config[0] = config[0];
H
Hongze Cheng 已提交
693
  writer[0]->ctx->opened = false;
H
Hongze Cheng 已提交
694 695 696
  return 0;
}

H
Hongze Cheng 已提交
697
int32_t tsdbSttFileWriterClose(SSttFileWriter **writer, int8_t abort, STFileOp *op) {
H
Hongze Cheng 已提交
698
  int32_t code = 0;
H
Hongze Cheng 已提交
699
  int32_t lino = 0;
H
Hongze Cheng 已提交
700
  int32_t vid = TD_VID(writer[0]->config->tsdb->pVnode);
H
Hongze Cheng 已提交
701

H
Hongze Cheng 已提交
702
  if (!writer[0]->ctx->opened) {
H
Hongze Cheng 已提交
703
    if (op) op->optype = TSDB_FOP_NONE;
H
Hongze Cheng 已提交
704
  } else {
H
Hongze Cheng 已提交
705 706
    if (abort) {
      code = tsdbSttFWriterCloseAbort(writer[0]);
H
Hongze Cheng 已提交
707
      TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
708
    } else {
H
Hongze Cheng 已提交
709
      code = tsdbSttFWriterCloseCommit(writer[0], op);
H
Hongze Cheng 已提交
710
      TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
711
    }
H
Hongze Cheng 已提交
712
    tsdbSttFWriterDoClose(writer[0]);
H
Hongze Cheng 已提交
713
  }
H
Hongze Cheng 已提交
714 715
  taosMemoryFree(writer[0]);
  writer[0] = NULL;
H
Hongze Cheng 已提交
716 717 718

_exit:
  if (code) {
H
Hongze Cheng 已提交
719
    TSDB_ERROR_LOG(vid, lino, code);
H
Hongze Cheng 已提交
720
  }
H
Hongze Cheng 已提交
721
  return code;
H
Hongze Cheng 已提交
722 723
}

H
Hongze Cheng 已提交
724
int32_t tsdbSttFileWriteTSData(SSttFileWriter *writer, SRowInfo *row) {
H
Hongze Cheng 已提交
725
  int32_t code = 0;
H
Hongze Cheng 已提交
726
  int32_t lino = 0;
H
Hongze Cheng 已提交
727
  int32_t vid = TD_VID(writer->config->tsdb->pVnode);
H
Hongze Cheng 已提交
728

H
Hongze Cheng 已提交
729
  if (!writer->ctx->opened) {
H
Hongze Cheng 已提交
730 731 732
    code = tsdbSttFWriterDoOpen(writer);
    TSDB_CHECK_CODE(code, lino, _exit);
  }
H
Hongze Cheng 已提交
733

H
Hongze Cheng 已提交
734 735
  TSDBKEY key[1] = {TSDBROW_KEY(&row->row)};
  if (!TABLE_SAME_SCHEMA(row->suid, row->uid, writer->ctx->tbid->suid, writer->ctx->tbid->uid)) {
H
Hongze Cheng 已提交
736 737
    code = tsdbSttFileDoWriteTSDataBlock(writer);
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
738

H
Hongze Cheng 已提交
739 740 741 742 743 744 745 746 747 748 749 750
    code = tsdbUpdateSkmTb(writer->config->tsdb, (TABLEID *)row, writer->config->skmTb);
    TSDB_CHECK_CODE(code, lino, _exit);

    TABLEID id = {.suid = row->suid, .uid = row->suid ? 0 : row->uid};
    code = tBlockDataInit(writer->bData, &id, writer->config->skmTb->pTSchema, NULL, 0);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  if (writer->ctx->tbid->uid != row->uid) {
    writer->ctx->tbid->suid = row->suid;
    writer->ctx->tbid->uid = row->uid;

H
Hongze Cheng 已提交
751
    if (STATIS_BLOCK_SIZE(writer->sData) >= writer->config->maxRow) {
H
Hongze Cheng 已提交
752
      code = tsdbSttFileDoWriteStatisBlock(writer);
H
Hongze Cheng 已提交
753
      TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
754 755
    }

H
Hongze Cheng 已提交
756 757 758
    STbStatisRecord record[1] = {{
        .suid = row->suid,
        .uid = row->uid,
H
Hongze Cheng 已提交
759 760 761 762 763 764
        .firstKey = key->ts,
        .firstKeyVer = key->version,
        .lastKey = key->ts,
        .lastKeyVer = key->version,
        .minVer = key->version,
        .maxVer = key->version,
H
Hongze Cheng 已提交
765 766 767 768
        .count = 1,
    }};
    code = tStatisBlockPut(writer->sData, record);
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
769 770
  }

H
Hongze Cheng 已提交
771
  if (row->row.type == TSDBROW_ROW_FMT) {
H
Hongze Cheng 已提交
772 773
    code = tsdbUpdateSkmRow(writer->config->tsdb, writer->ctx->tbid,  //
                            TSDBROW_SVERSION(&row->row), writer->config->skmRow);
H
Hongze Cheng 已提交
774
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
775 776
  }

H
Hongze Cheng 已提交
777
  // row to col conversion
H
Hongze Cheng 已提交
778 779 780 781 782 783 784 785 786
  if (key->version <= writer->config->compactVersion                       //
      && writer->bData->nRow > 0                                           //
      && (writer->bData->uid                                               //
              ? writer->bData->uid                                         //
              : writer->bData->aUid[writer->bData->nRow - 1]) == row->uid  //
      && writer->bData->aTSKEY[writer->bData->nRow - 1] == key->ts         //
  ) {
    code = tBlockDataUpdateRow(writer->bData, &row->row, writer->config->skmRow->pTSchema);
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
787 788 789 790 791
  } else {
    if (writer->bData->nRow >= writer->config->maxRow) {
      code = tsdbSttFileDoWriteTSDataBlock(writer);
      TSDB_CHECK_CODE(code, lino, _exit);
    }
H
Hongze Cheng 已提交
792

H
Hongze Cheng 已提交
793
    code = tBlockDataAppendRow(writer->bData, &row->row, writer->config->skmRow->pTSchema, row->uid);
H
Hongze Cheng 已提交
794
    TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
795 796
  }

H
Hongze Cheng 已提交
797 798 799 800 801 802 803 804
  TARRAY2_LAST(writer->sData->minVer) = TMIN(TARRAY2_LAST(writer->sData->minVer), key->version);
  TARRAY2_LAST(writer->sData->maxVer) = TMAX(TARRAY2_LAST(writer->sData->maxVer), key->version);
  if (key->ts > TARRAY2_LAST(writer->sData->lastKey)) {
    TARRAY2_LAST(writer->sData->lastKey) = key->ts;
    TARRAY2_LAST(writer->sData->lastKeyVer) = key->version;
    TARRAY2_LAST(writer->sData->count)++;
  } else if (key->ts == TARRAY2_LAST(writer->sData->lastKey)) {
    TARRAY2_LAST(writer->sData->lastKeyVer) = key->version;
H
Hongze Cheng 已提交
805 806 807 808
  } else {
    ASSERTS(0, "timestamp should be in ascending order");
  }

H
Hongze Cheng 已提交
809 810
_exit:
  if (code) {
H
Hongze Cheng 已提交
811
    TSDB_ERROR_LOG(vid, lino, code);
H
Hongze Cheng 已提交
812
  }
H
Hongze Cheng 已提交
813
  return code;
H
Hongze Cheng 已提交
814 815
}

H
Hongze Cheng 已提交
816
int32_t tsdbSttFileWriteTSDataBlock(SSttFileWriter *writer, SBlockData *bdata) {
H
Hongze Cheng 已提交
817 818 819
  int32_t code = 0;
  int32_t lino = 0;

H
Hongze Cheng 已提交
820 821
  SRowInfo row[1];
  row->suid = bdata->suid;
H
Hongze Cheng 已提交
822
  for (int32_t i = 0; i < bdata->nRow; i++) {
H
Hongze Cheng 已提交
823 824
    row->uid = bdata->uid ? bdata->uid : bdata->aUid[i];
    row->row = tsdbRowFromBlockData(bdata, i);
H
Hongze Cheng 已提交
825

H
Hongze Cheng 已提交
826
    code = tsdbSttFileWriteTSData(writer, row);
H
Hongze Cheng 已提交
827 828 829 830 831
    TSDB_CHECK_CODE(code, lino, _exit);
  }

_exit:
  if (code) {
H
Hongze Cheng 已提交
832
    TSDB_ERROR_LOG(TD_VID(writer->config->tsdb->pVnode), lino, code);
H
Hongze Cheng 已提交
833
  }
H
Hongze Cheng 已提交
834
  return code;
H
Hongze Cheng 已提交
835 836
}

H
Hongze Cheng 已提交
837
int32_t tsdbSttFileWriteDLData(SSttFileWriter *writer, TABLEID *tbid, SDelData *pDelData) {
H
Hongze Cheng 已提交
838 839
  ASSERTS(0, "TODO: Not implemented yet");

H
Hongze Cheng 已提交
840
  int32_t code;
H
Hongze Cheng 已提交
841 842 843
  int32_t lino;
  int32_t vid = TD_VID(writer->config->tsdb->pVnode);

H
Hongze Cheng 已提交
844
  if (!writer->ctx->opened) {
H
Hongze Cheng 已提交
845 846 847 848
    code = tsdbSttFWriterDoOpen(writer);
    return code;
  }

H
Hongze Cheng 已提交
849 850
  code = tsdbSttFileDoWriteTSDataBlock(writer);
  TSDB_CHECK_CODE(code, lino, _exit);
H
Hongze Cheng 已提交
851

H
Hongze Cheng 已提交
852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874
  code = tsdbSttFileDoWriteStatisBlock(writer);
  TSDB_CHECK_CODE(code, lino, _exit);

#if 0
  writer->dData[0].aData[0][writer->dData[0].nRow] = tbid->suid;         // suid
  writer->dData[0].aData[1][writer->dData[0].nRow] = tbid->uid;          // uid
  writer->dData[0].aData[2][writer->dData[0].nRow] = pDelData->version;  // version
  writer->dData[0].aData[3][writer->dData[0].nRow] = pDelData->sKey;     // skey
  writer->dData[0].aData[4][writer->dData[0].nRow] = pDelData->eKey;     // ekey
  writer->dData[0].nRow++;

  if (writer->dData[0].nRow >= writer->config->maxRow) {
    return tsdbSttFileDoWriteDelBlock(writer);
  } else {
    return 0;
  }
#endif

_exit:
  if (code) {
    TSDB_ERROR_LOG(vid, lino, code);
  }
  return code;
H
Hongze Cheng 已提交
875
}