tsdbReaderWriter.c 40.5 KB
Newer Older
H
refact  
Hongze Cheng 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
/*
 * 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/>.
 */

#include "tsdb.h"

H
Hongze Cheng 已提交
18
// SDelFWriter ====================================================
H
Hongze Cheng 已提交
19 20
int32_t tsdbDelFWriterOpen(SDelFWriter **ppWriter, SDelFile *pFile, STsdb *pTsdb) {
  int32_t      code = 0;
H
Hongze Cheng 已提交
21 22
  char         fname[TSDB_FILENAME_LEN];
  char         hdr[TSDB_FHDR_SIZE] = {0};
H
Hongze Cheng 已提交
23
  SDelFWriter *pDelFWriter;
H
Hongze Cheng 已提交
24
  int64_t      n;
H
Hongze Cheng 已提交
25

H
Hongze Cheng 已提交
26
  // alloc
H
Hongze Cheng 已提交
27 28 29 30 31 32
  pDelFWriter = (SDelFWriter *)taosMemoryCalloc(1, sizeof(*pDelFWriter));
  if (pDelFWriter == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
  pDelFWriter->pTsdb = pTsdb;
H
Hongze Cheng 已提交
33 34 35
  pDelFWriter->fDel = *pFile;

  tsdbDelFileName(pTsdb, pFile, fname);
H
Hongze Cheng 已提交
36 37 38 39 40 41
  pDelFWriter->pWriteH = taosOpenFile(fname, TD_FILE_WRITE | TD_FILE_CREATE);
  if (pDelFWriter->pWriteH == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
42 43 44
  // update header
  n = taosWriteFile(pDelFWriter->pWriteH, &hdr, TSDB_FHDR_SIZE);
  if (n < 0) {
H
Hongze Cheng 已提交
45 46 47 48
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
49
  pDelFWriter->fDel.size = TSDB_FHDR_SIZE;
H
Hongze Cheng 已提交
50
  pDelFWriter->fDel.offset = 0;
H
more  
Hongze Cheng 已提交
51

H
Hongze Cheng 已提交
52
  *ppWriter = pDelFWriter;
H
Hongze Cheng 已提交
53 54 55
  return code;

_err:
S
Shengliang Guan 已提交
56
  tsdbError("vgId:%d, failed to open del file writer since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
57
  *ppWriter = NULL;
H
Hongze Cheng 已提交
58 59 60
  return code;
}

H
Hongze Cheng 已提交
61 62 63
int32_t tsdbDelFWriterClose(SDelFWriter **ppWriter, int8_t sync) {
  int32_t      code = 0;
  SDelFWriter *pWriter = *ppWriter;
H
Hongze Cheng 已提交
64
  STsdb       *pTsdb = pWriter->pTsdb;
H
Hongze Cheng 已提交
65 66 67 68 69 70 71 72 73 74 75 76 77

  // sync
  if (sync && taosFsyncFile(pWriter->pWriteH) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  // close
  if (taosCloseFile(&pWriter->pWriteH) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
78 79 80
  for (int32_t iBuf = 0; iBuf < sizeof(pWriter->aBuf) / sizeof(uint8_t *); iBuf++) {
    tFree(pWriter->aBuf[iBuf]);
  }
H
Hongze Cheng 已提交
81 82
  taosMemoryFree(pWriter);

H
Hongze Cheng 已提交
83
  *ppWriter = NULL;
H
Hongze Cheng 已提交
84 85 86
  return code;

_err:
H
Hongze Cheng 已提交
87
  tsdbError("vgId:%d, failed to close del file writer since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
88 89 90
  return code;
}

H
Hongze Cheng 已提交
91 92 93 94
int32_t tsdbWriteDelData(SDelFWriter *pWriter, SArray *aDelData, SDelIdx *pDelIdx) {
  int32_t code = 0;
  int64_t size;
  int64_t n;
H
Hongze Cheng 已提交
95

H
Hongze Cheng 已提交
96
  // prepare
H
Hongze Cheng 已提交
97
  size = sizeof(uint32_t);
H
Hongze Cheng 已提交
98 99 100 101
  for (int32_t iDelData = 0; iDelData < taosArrayGetSize(aDelData); iDelData++) {
    size += tPutDelData(NULL, taosArrayGet(aDelData, iDelData));
  }
  size += sizeof(TSCKSUM);
H
Hongze Cheng 已提交
102 103

  // alloc
H
Hongze Cheng 已提交
104
  code = tRealloc(&pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
105 106 107
  if (code) goto _err;

  // build
H
Hongze Cheng 已提交
108
  n = 0;
H
Hongze Cheng 已提交
109
  n += tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT);
H
Hongze Cheng 已提交
110
  for (int32_t iDelData = 0; iDelData < taosArrayGetSize(aDelData); iDelData++) {
H
Hongze Cheng 已提交
111
    n += tPutDelData(pWriter->aBuf[0] + n, taosArrayGet(aDelData, iDelData));
H
Hongze Cheng 已提交
112
  }
H
Hongze Cheng 已提交
113
  taosCalcChecksumAppend(0, pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
114 115 116 117

  ASSERT(n + sizeof(TSCKSUM) == size);

  // write
H
Hongze Cheng 已提交
118
  n = taosWriteFile(pWriter->pWriteH, pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
119 120 121 122 123 124 125 126
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  ASSERT(n == size);

  // update
H
Hongze Cheng 已提交
127
  pDelIdx->offset = pWriter->fDel.size;
H
Hongze Cheng 已提交
128
  pDelIdx->size = size;
H
Hongze Cheng 已提交
129
  pWriter->fDel.size += size;
H
Hongze Cheng 已提交
130 131 132 133

  return code;

_err:
S
Shengliang Guan 已提交
134
  tsdbError("vgId:%d, failed to write del data since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
135 136 137
  return code;
}

H
Hongze Cheng 已提交
138
int32_t tsdbWriteDelIdx(SDelFWriter *pWriter, SArray *aDelIdx) {
H
Hongze Cheng 已提交
139
  int32_t  code = 0;
H
Hongze Cheng 已提交
140 141 142 143
  int64_t  size;
  int64_t  n;
  SDelIdx *pDelIdx;

H
Hongze Cheng 已提交
144
  // prepare
H
Hongze Cheng 已提交
145
  size = sizeof(uint32_t);
H
Hongze Cheng 已提交
146 147 148 149
  for (int32_t iDelIdx = 0; iDelIdx < taosArrayGetSize(aDelIdx); iDelIdx++) {
    size += tPutDelIdx(NULL, taosArrayGet(aDelIdx, iDelIdx));
  }
  size += sizeof(TSCKSUM);
H
Hongze Cheng 已提交
150 151

  // alloc
H
Hongze Cheng 已提交
152
  code = tRealloc(&pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
153
  if (code) goto _err;
H
Hongze Cheng 已提交
154

H
Hongze Cheng 已提交
155
  // build
H
Hongze Cheng 已提交
156
  n = 0;
H
Hongze Cheng 已提交
157
  n += tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT);
H
Hongze Cheng 已提交
158
  for (int32_t iDelIdx = 0; iDelIdx < taosArrayGetSize(aDelIdx); iDelIdx++) {
H
Hongze Cheng 已提交
159
    n += tPutDelIdx(pWriter->aBuf[0] + n, taosArrayGet(aDelIdx, iDelIdx));
H
Hongze Cheng 已提交
160
  }
H
Hongze Cheng 已提交
161
  taosCalcChecksumAppend(0, pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
162

H
Hongze Cheng 已提交
163 164
  ASSERT(n + sizeof(TSCKSUM) == size);

H
Hongze Cheng 已提交
165
  // write
H
Hongze Cheng 已提交
166
  n = taosWriteFile(pWriter->pWriteH, pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
167
  if (n < 0) {
H
Hongze Cheng 已提交
168 169 170 171
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
172
  // update
H
Hongze Cheng 已提交
173 174
  pWriter->fDel.offset = pWriter->fDel.size;
  pWriter->fDel.size += size;
H
Hongze Cheng 已提交
175

H
Hongze Cheng 已提交
176 177 178
  return code;

_err:
S
Shengliang Guan 已提交
179
  tsdbError("vgId:%d, write del idx failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
180 181 182
  return code;
}

H
Hongze Cheng 已提交
183 184 185 186 187
int32_t tsdbUpdateDelFileHdr(SDelFWriter *pWriter) {
  int32_t code = 0;
  char    hdr[TSDB_FHDR_SIZE];
  int64_t size = TSDB_FHDR_SIZE;
  int64_t n;
H
Hongze Cheng 已提交
188 189

  // build
H
Hongze Cheng 已提交
190 191 192
  memset(hdr, 0, size);
  tPutDelFile(hdr, &pWriter->fDel);
  taosCalcChecksumAppend(0, hdr, size);
H
Hongze Cheng 已提交
193 194 195 196 197 198 199 200

  // seek
  if (taosLSeekFile(pWriter->pWriteH, 0, SEEK_SET) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  // write
H
Hongze Cheng 已提交
201 202
  n = taosWriteFile(pWriter->pWriteH, hdr, size);
  if (n < 0) {
H
Hongze Cheng 已提交
203 204 205 206
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
207 208 209
  return code;

_err:
S
Shengliang Guan 已提交
210
  tsdbError("vgId:%d, update del file hdr failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
211 212 213 214 215
  return code;
}

// SDelFReader ====================================================
struct SDelFReader {
H
Hongze Cheng 已提交
216
  STsdb    *pTsdb;
H
Hongze Cheng 已提交
217
  SDelFile  fDel;
H
Hongze Cheng 已提交
218
  TdFilePtr pReadH;
H
Hongze Cheng 已提交
219

H
Hongze Cheng 已提交
220
  uint8_t *aBuf[1];
H
refact  
Hongze Cheng 已提交
221
};
H
Hongze Cheng 已提交
222

H
Hongze Cheng 已提交
223
int32_t tsdbDelFReaderOpen(SDelFReader **ppReader, SDelFile *pFile, STsdb *pTsdb) {
H
Hongze Cheng 已提交
224
  int32_t      code = 0;
H
Hongze Cheng 已提交
225
  char         fname[TSDB_FILENAME_LEN];
H
Hongze Cheng 已提交
226
  SDelFReader *pDelFReader;
H
Hongze Cheng 已提交
227
  int64_t      n;
H
Hongze Cheng 已提交
228 229 230 231 232 233 234 235 236 237

  // alloc
  pDelFReader = (SDelFReader *)taosMemoryCalloc(1, sizeof(*pDelFReader));
  if (pDelFReader == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }

  // open impl
  pDelFReader->pTsdb = pTsdb;
H
Hongze Cheng 已提交
238 239 240
  pDelFReader->fDel = *pFile;

  tsdbDelFileName(pTsdb, pFile, fname);
H
Hongze Cheng 已提交
241
  pDelFReader->pReadH = taosOpenFile(fname, TD_FILE_READ);
H
Hongze Cheng 已提交
242
  if (pDelFReader->pReadH == NULL) {
H
Hongze Cheng 已提交
243 244 245 246 247 248 249 250 251 252
    code = TAOS_SYSTEM_ERROR(errno);
    taosMemoryFree(pDelFReader);
    goto _err;
  }

_exit:
  *ppReader = pDelFReader;
  return code;

_err:
S
Shengliang Guan 已提交
253
  tsdbError("vgId:%d, del file reader open failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
254
  *ppReader = NULL;
H
Hongze Cheng 已提交
255 256 257
  return code;
}

H
Hongze Cheng 已提交
258 259 260
int32_t tsdbDelFReaderClose(SDelFReader **ppReader) {
  int32_t      code = 0;
  SDelFReader *pReader = *ppReader;
H
Hongze Cheng 已提交
261 262

  if (pReader) {
H
Hongze Cheng 已提交
263 264 265 266
    if (taosCloseFile(&pReader->pReadH) < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _exit;
    }
H
Hongze Cheng 已提交
267 268 269
    for (int32_t iBuf = 0; iBuf < sizeof(pReader->aBuf) / sizeof(uint8_t *); iBuf++) {
      tFree(pReader->aBuf[iBuf]);
    }
H
Hongze Cheng 已提交
270 271
    taosMemoryFree(pReader);
  }
H
Hongze Cheng 已提交
272
  *ppReader = NULL;
H
Hongze Cheng 已提交
273

H
Hongze Cheng 已提交
274
_exit:
H
Hongze Cheng 已提交
275 276 277
  return code;
}

H
Hongze Cheng 已提交
278 279 280 281 282
int32_t tsdbReadDelData(SDelFReader *pReader, SDelIdx *pDelIdx, SArray *aDelData) {
  int32_t code = 0;
  int64_t offset = pDelIdx->offset;
  int64_t size = pDelIdx->size;
  int64_t n;
H
Hongze Cheng 已提交
283

H
Hongze Cheng 已提交
284
  taosArrayClear(aDelData);
H
Hongze Cheng 已提交
285 286

  // seek
H
Hongze Cheng 已提交
287
  if (taosLSeekFile(pReader->pReadH, offset, SEEK_SET) < 0) {
H
Hongze Cheng 已提交
288 289 290 291 292
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  // alloc
H
Hongze Cheng 已提交
293
  code = tRealloc(&pReader->aBuf[0], size);
H
Hongze Cheng 已提交
294 295 296
  if (code) goto _err;

  // read
H
Hongze Cheng 已提交
297
  n = taosReadFile(pReader->pReadH, pReader->aBuf[0], size);
H
Hongze Cheng 已提交
298 299 300
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
H
Hongze Cheng 已提交
301
  } else if (n < size) {
H
Hongze Cheng 已提交
302 303
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
H
Hongze Cheng 已提交
304 305 306
  }

  // check
H
Hongze Cheng 已提交
307
  if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) {
H
Hongze Cheng 已提交
308 309 310 311
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

H
Hongze Cheng 已提交
312 313
  // // decode
  n = 0;
H
Hongze Cheng 已提交
314 315

  uint32_t delimiter;
H
Hongze Cheng 已提交
316
  n += tGetU32(pReader->aBuf[0] + n, &delimiter);
H
Hongze Cheng 已提交
317
  while (n < size - sizeof(TSCKSUM)) {
H
Hongze Cheng 已提交
318
    SDelData delData;
H
Hongze Cheng 已提交
319
    n += tGetDelData(pReader->aBuf[0] + n, &delData);
H
Hongze Cheng 已提交
320

H
Hongze Cheng 已提交
321
    if (taosArrayPush(aDelData, &delData) == NULL) {
H
Hongze Cheng 已提交
322 323 324
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
    }
H
Hongze Cheng 已提交
325
  }
H
Hongze Cheng 已提交
326

H
Hongze Cheng 已提交
327 328
  ASSERT(n == size - sizeof(TSCKSUM));

H
Hongze Cheng 已提交
329 330 331
  return code;

_err:
S
Shengliang Guan 已提交
332
  tsdbError("vgId:%d, read del data failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
333 334 335
  return code;
}

H
Hongze Cheng 已提交
336 337 338 339 340
int32_t tsdbReadDelIdx(SDelFReader *pReader, SArray *aDelIdx) {
  int32_t code = 0;
  int32_t n;
  int64_t offset = pReader->fDel.offset;
  int64_t size = pReader->fDel.size - offset;
H
Hongze Cheng 已提交
341

H
Hongze Cheng 已提交
342
  taosArrayClear(aDelIdx);
H
more  
Hongze Cheng 已提交
343

H
Hongze Cheng 已提交
344
  // seek
H
Hongze Cheng 已提交
345
  if (taosLSeekFile(pReader->pReadH, offset, SEEK_SET) < 0) {
H
Hongze Cheng 已提交
346 347 348 349
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
350
  // alloc
H
Hongze Cheng 已提交
351
  code = tRealloc(&pReader->aBuf[0], size);
H
Hongze Cheng 已提交
352
  if (code) goto _err;
H
Hongze Cheng 已提交
353

H
Hongze Cheng 已提交
354
  // read
H
Hongze Cheng 已提交
355
  n = taosReadFile(pReader->pReadH, pReader->aBuf[0], size);
H
Hongze Cheng 已提交
356
  if (n < 0) {
H
Hongze Cheng 已提交
357
    code = TAOS_SYSTEM_ERROR(errno);
H
Hongze Cheng 已提交
358
    goto _err;
H
Hongze Cheng 已提交
359 360 361
  } else if (n < size) {
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
H
Hongze Cheng 已提交
362 363 364
  }

  // check
H
Hongze Cheng 已提交
365
  if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) {
H
Hongze Cheng 已提交
366 367 368 369 370
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

  // decode
H
Hongze Cheng 已提交
371
  n = 0;
H
Hongze Cheng 已提交
372
  uint32_t delimiter;
H
Hongze Cheng 已提交
373
  n += tGetU32(pReader->aBuf[0] + n, &delimiter);
H
Hongze Cheng 已提交
374
  ASSERT(delimiter == TSDB_FILE_DLMT);
H
Hongze Cheng 已提交
375 376

  while (n < size - sizeof(TSCKSUM)) {
H
Hongze Cheng 已提交
377 378
    SDelIdx delIdx;

H
Hongze Cheng 已提交
379
    n += tGetDelIdx(pReader->aBuf[0] + n, &delIdx);
H
Hongze Cheng 已提交
380

H
Hongze Cheng 已提交
381
    if (taosArrayPush(aDelIdx, &delIdx) == NULL) {
H
Hongze Cheng 已提交
382 383 384 385 386 387
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
    }
  }

  ASSERT(n == size - sizeof(TSCKSUM));
H
Hongze Cheng 已提交
388 389 390 391

  return code;

_err:
S
Shengliang Guan 已提交
392
  tsdbError("vgId:%d, read del idx failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
393
  return code;
H
Hongze Cheng 已提交
394 395 396 397 398 399
}

// SDataFReader ====================================================
struct SDataFReader {
  STsdb     *pTsdb;
  SDFileSet *pSet;
H
Hongze Cheng 已提交
400 401 402 403
  TdFilePtr  pHeadFD;
  TdFilePtr  pDataFD;
  TdFilePtr  pLastFD;
  TdFilePtr  pSmaFD;
H
Hongze Cheng 已提交
404

H
Hongze Cheng 已提交
405
  uint8_t *aBuf[3];
H
Hongze Cheng 已提交
406 407 408
};

int32_t tsdbDataFReaderOpen(SDataFReader **ppReader, STsdb *pTsdb, SDFileSet *pSet) {
H
Hongze Cheng 已提交
409 410 411 412 413 414 415 416 417 418 419 420 421 422 423
  int32_t       code = 0;
  SDataFReader *pReader;
  char          fname[TSDB_FILENAME_LEN];

  // alloc
  pReader = (SDataFReader *)taosMemoryCalloc(1, sizeof(*pReader));
  if (pReader == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
  pReader->pTsdb = pTsdb;
  pReader->pSet = pSet;

  // open impl
  // head
H
Hongze Cheng 已提交
424
  tsdbHeadFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pHeadF, fname);
H
Hongze Cheng 已提交
425 426 427 428 429 430 431
  pReader->pHeadFD = taosOpenFile(fname, TD_FILE_READ);
  if (pReader->pHeadFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  // data
H
Hongze Cheng 已提交
432
  tsdbDataFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pDataF, fname);
H
Hongze Cheng 已提交
433 434 435 436 437 438 439
  pReader->pDataFD = taosOpenFile(fname, TD_FILE_READ);
  if (pReader->pDataFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  // last
H
Hongze Cheng 已提交
440
  tsdbLastFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pLastF, fname);
H
Hongze Cheng 已提交
441 442 443 444 445 446 447
  pReader->pLastFD = taosOpenFile(fname, TD_FILE_READ);
  if (pReader->pLastFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  // sma
H
Hongze Cheng 已提交
448
  tsdbSmaFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pSmaF, fname);
H
Hongze Cheng 已提交
449 450 451 452 453 454 455 456 457 458
  pReader->pSmaFD = taosOpenFile(fname, TD_FILE_READ);
  if (pReader->pSmaFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  *ppReader = pReader;
  return code;

_err:
S
Shengliang Guan 已提交
459
  tsdbError("vgId:%d, tsdb data file reader open failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
460
  *ppReader = NULL;
H
Hongze Cheng 已提交
461 462 463
  return code;
}

H
Hongze Cheng 已提交
464
int32_t tsdbDataFReaderClose(SDataFReader **ppReader) {
H
Hongze Cheng 已提交
465
  int32_t code = 0;
H
Hongze Cheng 已提交
466
  if (*ppReader == NULL) goto _exit;
H
Hongze Cheng 已提交
467

H
Hongze Cheng 已提交
468
  if (taosCloseFile(&(*ppReader)->pHeadFD) < 0) {
H
Hongze Cheng 已提交
469 470 471 472
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
473
  if (taosCloseFile(&(*ppReader)->pDataFD) < 0) {
H
Hongze Cheng 已提交
474 475 476 477
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
478
  if (taosCloseFile(&(*ppReader)->pLastFD) < 0) {
H
Hongze Cheng 已提交
479 480 481 482
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
483
  if (taosCloseFile(&(*ppReader)->pSmaFD) < 0) {
H
Hongze Cheng 已提交
484 485 486 487
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
488 489 490 491
  for (int32_t iBuf = 0; iBuf < sizeof((*ppReader)->aBuf) / sizeof(uint8_t *); iBuf++) {
    tFree((*ppReader)->aBuf[iBuf]);
  }

H
Hongze Cheng 已提交
492
  taosMemoryFree(*ppReader);
H
Hongze Cheng 已提交
493 494

_exit:
H
Hongze Cheng 已提交
495
  *ppReader = NULL;
H
Hongze Cheng 已提交
496 497 498
  return code;

_err:
S
Shengliang Guan 已提交
499
  tsdbError("vgId:%d, data file reader close failed since %s", TD_VID((*ppReader)->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
500 501 502
  return code;
}

H
Hongze Cheng 已提交
503 504 505
int32_t tsdbReadBlockIdx(SDataFReader *pReader, SArray *aBlockIdx) {
  int32_t  code = 0;
  int64_t  offset = pReader->pSet->pHeadF->offset;
H
Hongze Cheng 已提交
506
  int64_t  size = pReader->pSet->pHeadF->size - offset;
H
Hongze Cheng 已提交
507 508
  int64_t  n;
  uint32_t delimiter;
H
Hongze Cheng 已提交
509

H
Hongze Cheng 已提交
510 511 512 513
  taosArrayClear(aBlockIdx);
  if (size == 0) {
    goto _exit;
  }
H
Hongze Cheng 已提交
514 515

  // alloc
H
Hongze Cheng 已提交
516
  code = tRealloc(&pReader->aBuf[0], size);
H
Hongze Cheng 已提交
517 518 519 520 521 522 523 524 525
  if (code) goto _err;

  // seek
  if (taosLSeekFile(pReader->pHeadFD, offset, SEEK_SET) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  // read
H
Hongze Cheng 已提交
526
  n = taosReadFile(pReader->pHeadFD, pReader->aBuf[0], size);
H
Hongze Cheng 已提交
527 528 529 530 531 532 533 534 535
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  } else if (n < size) {
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

  // check
H
Hongze Cheng 已提交
536
  if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) {
H
Hongze Cheng 已提交
537 538 539 540 541
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

  // decode
H
Hongze Cheng 已提交
542
  n = 0;
H
Hongze Cheng 已提交
543
  n = tGetU32(pReader->aBuf[0] + n, &delimiter);
H
Hongze Cheng 已提交
544
  ASSERT(delimiter == TSDB_FILE_DLMT);
H
Hongze Cheng 已提交
545 546

  while (n < size - sizeof(TSCKSUM)) {
H
Hongze Cheng 已提交
547
    SBlockIdx blockIdx;
H
Hongze Cheng 已提交
548
    n += tGetBlockIdx(pReader->aBuf[0] + n, &blockIdx);
H
Hongze Cheng 已提交
549 550 551 552 553 554 555

    if (taosArrayPush(aBlockIdx, &blockIdx) == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
    }
  }

H
Hongze Cheng 已提交
556 557
  ASSERT(n + sizeof(TSCKSUM) == size);

H
Hongze Cheng 已提交
558
_exit:
H
Hongze Cheng 已提交
559 560 561
  return code;

_err:
S
Shengliang Guan 已提交
562
  tsdbError("vgId:%d, read block idx failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
563 564 565
  return code;
}

H
Hongze Cheng 已提交
566
int32_t tsdbReadBlockL(SDataFReader *pReader, SArray *aBlockL) {
H
Hongze Cheng 已提交
567
  int32_t  code = 0;
H
Hongze Cheng 已提交
568 569
  int64_t  offset = pReader->pSet->pLastF->offset;
  int64_t  size = pReader->pSet->pLastF->size - offset;
H
Hongze Cheng 已提交
570 571 572
  int64_t  n;
  uint32_t delimiter;

H
Hongze Cheng 已提交
573 574 575 576
  taosArrayClear(aBlockL);
  if (size == 0) {
    goto _exit;
  }
H
Hongze Cheng 已提交
577 578

  // alloc
H
Hongze Cheng 已提交
579
  code = tRealloc(&pReader->aBuf[0], size);
H
Hongze Cheng 已提交
580 581 582
  if (code) goto _err;

  // seek
H
Hongze Cheng 已提交
583
  if (taosLSeekFile(pReader->pLastFD, offset, SEEK_SET) < 0) {
H
Hongze Cheng 已提交
584 585 586 587 588
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  // read
H
Hongze Cheng 已提交
589
  n = taosReadFile(pReader->pLastFD, pReader->aBuf[0], size);
H
Hongze Cheng 已提交
590 591 592 593 594 595 596 597 598
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  } else if (n < size) {
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

  // check
H
Hongze Cheng 已提交
599
  if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) {
H
Hongze Cheng 已提交
600 601 602 603 604 605
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

  // decode
  n = 0;
H
Hongze Cheng 已提交
606
  n = tGetU32(pReader->aBuf[0] + n, &delimiter);
H
Hongze Cheng 已提交
607 608 609
  ASSERT(delimiter == TSDB_FILE_DLMT);

  while (n < size - sizeof(TSCKSUM)) {
H
Hongze Cheng 已提交
610
    SBlockL blockl;
H
Hongze Cheng 已提交
611
    n += tGetBlockL(pReader->aBuf[0] + n, &blockl);
H
Hongze Cheng 已提交
612 613 614 615 616 617 618 619 620

    if (taosArrayPush(aBlockL, &blockl) == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
    }
  }

  ASSERT(n + sizeof(TSCKSUM) == size);

H
Hongze Cheng 已提交
621
_exit:
H
Hongze Cheng 已提交
622 623 624 625 626 627 628
  return code;

_err:
  tsdbError("vgId:%d read blockl failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
  return code;
}

H
Hongze Cheng 已提交
629 630 631 632 633 634
int32_t tsdbReadBlock(SDataFReader *pReader, SBlockIdx *pBlockIdx, SMapData *mBlock) {
  int32_t code = 0;
  int64_t offset = pBlockIdx->offset;
  int64_t size = pBlockIdx->size;
  int64_t n;
  int64_t tn;
H
Hongze Cheng 已提交
635

H
Hongze Cheng 已提交
636
  // alloc
H
Hongze Cheng 已提交
637
  code = tRealloc(&pReader->aBuf[0], size);
H
Hongze Cheng 已提交
638 639 640 641 642 643 644 645 646
  if (code) goto _err;

  // seek
  if (taosLSeekFile(pReader->pHeadFD, offset, SEEK_SET) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  // read
H
Hongze Cheng 已提交
647
  n = taosReadFile(pReader->pHeadFD, pReader->aBuf[0], size);
H
Hongze Cheng 已提交
648 649 650 651 652 653 654 655 656
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  } else if (n < size) {
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

  // check
H
Hongze Cheng 已提交
657
  if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) {
H
Hongze Cheng 已提交
658 659 660 661 662
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

  // decode
H
Hongze Cheng 已提交
663 664 665
  n = 0;

  uint32_t delimiter;
H
Hongze Cheng 已提交
666
  n += tGetU32(pReader->aBuf[0] + n, &delimiter);
H
Hongze Cheng 已提交
667
  ASSERT(delimiter == TSDB_FILE_DLMT);
H
Hongze Cheng 已提交
668

H
Hongze Cheng 已提交
669
  tn = tGetMapData(pReader->aBuf[0] + n, mBlock);
H
Hongze Cheng 已提交
670 671 672 673 674
  if (tn < 0) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
  n += tn;
H
Hongze Cheng 已提交
675 676 677 678 679
  ASSERT(n + sizeof(TSCKSUM) == size);

  return code;

_err:
S
Shengliang Guan 已提交
680
  tsdbError("vgId:%d, read block failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
681 682 683
  return code;
}

H
Hongze Cheng 已提交
684 685 686 687 688 689 690 691 692 693
int32_t tsdbReadBlockSma(SDataFReader *pReader, SBlock *pBlock, SArray *aColumnDataAgg) {
  int32_t   code = 0;
  SSmaInfo *pSmaInfo = &pBlock->smaInfo;

  ASSERT(pSmaInfo->size > 0);

  taosArrayClear(aColumnDataAgg);

  // alloc
  int32_t size = pSmaInfo->size + sizeof(TSCKSUM);
H
Hongze Cheng 已提交
694
  code = tRealloc(&pReader->aBuf[0], size);
H
Hongze Cheng 已提交
695 696 697
  if (code) goto _err;

  // read
H
Hongze Cheng 已提交
698
  int64_t n = taosReadFile(pReader->pSmaFD, pReader->aBuf[0], size);
H
Hongze Cheng 已提交
699 700 701 702 703 704 705 706 707
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  } else if (n < size) {
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

  // check
H
Hongze Cheng 已提交
708
  if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) {
H
Hongze Cheng 已提交
709 710 711 712 713 714 715 716 717
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

  // decode
  n = 0;
  while (n < pSmaInfo->size) {
    SColumnDataAgg sma;

H
Hongze Cheng 已提交
718
    n += tGetColumnDataAgg(pReader->aBuf[0] + n, &sma);
H
Hongze Cheng 已提交
719 720 721 722 723 724 725 726 727 728 729 730 731
    if (taosArrayPush(aColumnDataAgg, &sma) == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
    }
  }

  return code;

_err:
  tsdbError("vgId:%d tsdb read block sma failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
  return code;
}

H
Hongze Cheng 已提交
732 733
static int32_t tsdbReadBlockDataImpl(SDataFReader *pReader, SBlockInfo *pBlkInfo, int8_t fromLast,
                                     SBlockData *pBlockData) {
H
Hongze Cheng 已提交
734 735
  int32_t code = 0;

H
Hongze Cheng 已提交
736
  tBlockDataClear(pBlockData);
H
Hongze Cheng 已提交
737 738 739

  TdFilePtr pFD = fromLast ? pReader->pLastFD : pReader->pDataFD;

H
Hongze Cheng 已提交
740
  // uid + version + tskey
H
Hongze Cheng 已提交
741
  code = tsdbReadAndCheck(pFD, pBlkInfo->offset, &pReader->aBuf[0], pBlkInfo->szKey, 1);
H
Hongze Cheng 已提交
742 743
  if (code) goto _err;
  SDiskDataHdr hdr;
H
Hongze Cheng 已提交
744
  uint8_t     *p = pReader->aBuf[0] + tGetDiskDataHdr(pReader->aBuf[0], &hdr);
H
Hongze Cheng 已提交
745

H
Hongze Cheng 已提交
746
  ASSERT(hdr.delimiter == TSDB_FILE_DLMT);
H
Hongze Cheng 已提交
747 748
  ASSERT(pBlockData->suid == hdr.suid);
  ASSERT(pBlockData->uid == hdr.uid);
H
Hongze Cheng 已提交
749

H
Hongze Cheng 已提交
750 751
  pBlockData->nRow = hdr.nRow;

H
Hongze Cheng 已提交
752
  // uid
H
Hongze Cheng 已提交
753 754 755
  if (hdr.uid == 0) {
    ASSERT(hdr.szUid);
    code = tsdbDecmprData(p, hdr.szUid, TSDB_DATA_TYPE_BIGINT, hdr.cmprAlg, (uint8_t **)&pBlockData->aUid,
H
Hongze Cheng 已提交
756
                          sizeof(int64_t) * hdr.nRow, &pReader->aBuf[1]);
H
Hongze Cheng 已提交
757 758
    if (code) goto _err;
  } else {
H
Hongze Cheng 已提交
759
    ASSERT(!hdr.szUid);
H
Hongze Cheng 已提交
760 761 762
  }
  p += hdr.szUid;

H
Hongze Cheng 已提交
763
  // version
H
Hongze Cheng 已提交
764
  code = tsdbDecmprData(p, hdr.szVer, TSDB_DATA_TYPE_BIGINT, hdr.cmprAlg, (uint8_t **)&pBlockData->aVersion,
H
Hongze Cheng 已提交
765
                        sizeof(int64_t) * hdr.nRow, &pReader->aBuf[1]);
H
Hongze Cheng 已提交
766 767 768
  if (code) goto _err;
  p += hdr.szVer;

H
Hongze Cheng 已提交
769 770
  // TSKEY
  code = tsdbDecmprData(p, hdr.szKey, TSDB_DATA_TYPE_TIMESTAMP, hdr.cmprAlg, (uint8_t **)&pBlockData->aTSKEY,
H
Hongze Cheng 已提交
771
                        sizeof(TSKEY) * hdr.nRow, &pReader->aBuf[1]);
H
Hongze Cheng 已提交
772 773 774
  if (code) goto _err;
  p += hdr.szKey;

H
Hongze Cheng 已提交
775
  ASSERT(p - pReader->aBuf[0] == pBlkInfo->szKey - sizeof(TSCKSUM));
H
Hongze Cheng 已提交
776 777

  // read and decode columns
H
Hongze Cheng 已提交
778 779
  if (taosArrayGetSize(pBlockData->aIdx) == 0) goto _exit;

H
Hongze Cheng 已提交
780
  if (hdr.szBlkCol > 0) {
H
Hongze Cheng 已提交
781
    int64_t offset = pBlkInfo->offset + pBlkInfo->szKey;
H
Hongze Cheng 已提交
782
    code = tsdbReadAndCheck(pFD, offset, &pReader->aBuf[0], hdr.szBlkCol + sizeof(TSCKSUM), 1);
H
Hongze Cheng 已提交
783
    if (code) goto _err;
H
Hongze Cheng 已提交
784
  }
H
Hongze Cheng 已提交
785

H
Hongze Cheng 已提交
786 787 788
  SBlockCol  blockCol = {.cid = 0};
  SBlockCol *pBlockCol = &blockCol;
  int32_t    n = 0;
H
Hongze Cheng 已提交
789

H
Hongze Cheng 已提交
790 791
  for (int32_t iColData = 0; iColData < taosArrayGetSize(pBlockData->aIdx); iColData++) {
    SColData *pColData = tBlockDataGetColDataByIdx(pBlockData, iColData);
H
Hongze Cheng 已提交
792

H
Hongze Cheng 已提交
793 794
    while (pBlockCol && pBlockCol->cid < pColData->cid) {
      if (n < hdr.szBlkCol) {
H
Hongze Cheng 已提交
795
        n += tGetBlockCol(pReader->aBuf[0] + n, pBlockCol);
H
Hongze Cheng 已提交
796 797 798 799 800
      } else {
        ASSERT(n == hdr.szBlkCol);
        pBlockCol = NULL;
      }
    }
H
Hongze Cheng 已提交
801

H
Hongze Cheng 已提交
802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817
    if (pBlockCol == NULL || pBlockCol->cid > pColData->cid) {
      // add a lot of NONE
      for (int32_t iRow = 0; iRow < hdr.nRow; iRow++) {
        code = tColDataAppendValue(pColData, &COL_VAL_NONE(pBlockCol->cid, pBlockCol->type));
        if (code) goto _err;
      }
    } else {
      ASSERT(pBlockCol->type == pColData->type);
      ASSERT(pBlockCol->flag && pBlockCol->flag != HAS_NONE);

      if (pBlockCol->flag == HAS_NULL) {
        // add a lot of NULL
        for (int32_t iRow = 0; iRow < hdr.nRow; iRow++) {
          code = tColDataAppendValue(pColData, &COL_VAL_NULL(pBlockCol->cid, pBlockCol->type));
          if (code) goto _err;
        }
H
Hongze Cheng 已提交
818
      } else {
H
Hongze Cheng 已提交
819 820 821
        // decode from binary
        int64_t offset = pBlkInfo->offset + pBlkInfo->szKey + hdr.szBlkCol + sizeof(TSCKSUM) + pBlockCol->offset;
        int32_t size = pBlockCol->szBitmap + pBlockCol->szOffset + pBlockCol->szValue + sizeof(TSCKSUM);
H
Hongze Cheng 已提交
822

H
Hongze Cheng 已提交
823
        code = tsdbReadAndCheck(pFD, offset, &pReader->aBuf[1], size, 0);
H
Hongze Cheng 已提交
824 825
        if (code) goto _err;

H
Hongze Cheng 已提交
826
        code = tsdbDecmprColData(pReader->aBuf[1], pBlockCol, hdr.cmprAlg, hdr.nRow, pColData, &pReader->aBuf[2]);
H
Hongze Cheng 已提交
827 828 829 830 831
        if (code) goto _err;
      }
    }
  }

H
Hongze Cheng 已提交
832
_exit:
H
Hongze Cheng 已提交
833 834 835 836 837 838 839
  return code;

_err:
  tsdbError("vgId:%d tsdb read block data impl failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
  return code;
}

H
Hongze Cheng 已提交
840
int32_t tsdbReadDataBlock(SDataFReader *pReader, SBlock *pBlock, SBlockData *pBlockData) {
H
Hongze Cheng 已提交
841 842
  int32_t code = 0;

H
Hongze Cheng 已提交
843
  code = tsdbReadBlockDataImpl(pReader, &pBlock->aSubBlock[0], 0, pBlockData);
H
Hongze Cheng 已提交
844 845
  if (code) goto _err;

H
Hongze Cheng 已提交
846 847 848
  if (pBlock->nSubBlock > 1) {
    SBlockData bData1;
    SBlockData bData2;
H
Hongze Cheng 已提交
849

H
Hongze Cheng 已提交
850 851
    // create
    code = tBlockDataCreate(&bData1);
H
Hongze Cheng 已提交
852
    if (code) goto _err;
H
Hongze Cheng 已提交
853
    code = tBlockDataCreate(&bData2);
H
Hongze Cheng 已提交
854 855
    if (code) goto _err;

H
Hongze Cheng 已提交
856 857 858
    // init
    tBlockDataInitEx(&bData1, pBlockData);
    tBlockDataInitEx(&bData2, pBlockData);
H
Hongze Cheng 已提交
859 860

    for (int32_t iSubBlock = 1; iSubBlock < pBlock->nSubBlock; iSubBlock++) {
H
Hongze Cheng 已提交
861
      code = tsdbReadBlockDataImpl(pReader, &pBlock->aSubBlock[iSubBlock], 0, &bData1);
H
Hongze Cheng 已提交
862
      if (code) {
H
Hongze Cheng 已提交
863 864
        tBlockDataDestroy(&bData1, 1);
        tBlockDataDestroy(&bData2, 1);
H
Hongze Cheng 已提交
865 866 867
        goto _err;
      }

H
Hongze Cheng 已提交
868
      code = tBlockDataCopy(pBlockData, &bData2);
H
Hongze Cheng 已提交
869
      if (code) {
H
Hongze Cheng 已提交
870 871
        tBlockDataDestroy(&bData1, 1);
        tBlockDataDestroy(&bData2, 1);
H
Hongze Cheng 已提交
872 873
        goto _err;
      }
H
Hongze Cheng 已提交
874

H
Hongze Cheng 已提交
875
      code = tBlockDataMerge(&bData1, &bData2, pBlockData);
H
Hongze Cheng 已提交
876
      if (code) {
H
Hongze Cheng 已提交
877 878
        tBlockDataDestroy(&bData1, 1);
        tBlockDataDestroy(&bData2, 1);
H
Hongze Cheng 已提交
879 880 881 882
        goto _err;
      }
    }

H
Hongze Cheng 已提交
883 884
    tBlockDataDestroy(&bData1, 1);
    tBlockDataDestroy(&bData2, 1);
H
Hongze Cheng 已提交
885 886 887 888 889
  }

  return code;

_err:
H
Hongze Cheng 已提交
890
  tsdbError("vgId:%d tsdb read data block failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
891 892 893
  return code;
}

H
Hongze Cheng 已提交
894
int32_t tsdbReadLastBlock(SDataFReader *pReader, SBlockL *pBlockL, SBlockData *pBlockData) {
H
Hongze Cheng 已提交
895 896
  int32_t code = 0;

H
Hongze Cheng 已提交
897
  code = tsdbReadBlockDataImpl(pReader, &pBlockL->bInfo, 1, pBlockData);
H
Hongze Cheng 已提交
898 899 900 901 902 903
  if (code) goto _err;

  return code;

_err:
  tsdbError("vgId:%d tsdb read last block failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
904 905
  return code;
}
H
Hongze Cheng 已提交
906 907 908

// SDataFWriter ====================================================
int32_t tsdbDataFWriterOpen(SDataFWriter **ppWriter, STsdb *pTsdb, SDFileSet *pSet) {
H
Hongze Cheng 已提交
909 910 911 912 913 914 915 916 917 918 919 920 921
  int32_t       code = 0;
  int32_t       flag;
  int64_t       n;
  SDataFWriter *pWriter = NULL;
  char          fname[TSDB_FILENAME_LEN];
  char          hdr[TSDB_FHDR_SIZE] = {0};

  // alloc
  pWriter = taosMemoryCalloc(1, sizeof(*pWriter));
  if (pWriter == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
H
Hongze Cheng 已提交
922
  if (code) goto _err;
H
Hongze Cheng 已提交
923
  pWriter->pTsdb = pTsdb;
H
Hongze Cheng 已提交
924 925 926 927 928 929 930 931 932 933
  pWriter->wSet = (SDFileSet){.diskId = pSet->diskId,
                              .fid = pSet->fid,
                              .pHeadF = &pWriter->fHead,
                              .pDataF = &pWriter->fData,
                              .pLastF = &pWriter->fLast,
                              .pSmaF = &pWriter->fSma};
  pWriter->fHead = *pSet->pHeadF;
  pWriter->fData = *pSet->pDataF;
  pWriter->fLast = *pSet->pLastF;
  pWriter->fSma = *pSet->pSmaF;
H
Hongze Cheng 已提交
934 935 936

  // head
  flag = TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
H
Hongze Cheng 已提交
937
  tsdbHeadFileName(pTsdb, pWriter->wSet.diskId, pWriter->wSet.fid, &pWriter->fHead, fname);
H
Hongze Cheng 已提交
938 939 940 941 942 943 944 945 946 947 948 949 950 951
  pWriter->pHeadFD = taosOpenFile(fname, flag);
  if (pWriter->pHeadFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  n = taosWriteFile(pWriter->pHeadFD, hdr, TSDB_FHDR_SIZE);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  ASSERT(n == TSDB_FHDR_SIZE);

H
Hongze Cheng 已提交
952
  pWriter->fHead.size += TSDB_FHDR_SIZE;
H
Hongze Cheng 已提交
953 954

  // data
H
Hongze Cheng 已提交
955
  if (pWriter->fData.size == 0) {
H
Hongze Cheng 已提交
956 957 958 959
    flag = TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
  } else {
    flag = TD_FILE_WRITE;
  }
H
Hongze Cheng 已提交
960
  tsdbDataFileName(pTsdb, pWriter->wSet.diskId, pWriter->wSet.fid, &pWriter->fData, fname);
H
Hongze Cheng 已提交
961 962 963 964 965
  pWriter->pDataFD = taosOpenFile(fname, flag);
  if (pWriter->pDataFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
H
Hongze Cheng 已提交
966
  if (pWriter->fData.size == 0) {
H
Hongze Cheng 已提交
967 968 969 970 971 972
    n = taosWriteFile(pWriter->pDataFD, hdr, TSDB_FHDR_SIZE);
    if (n < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

H
Hongze Cheng 已提交
973
    pWriter->fData.size += TSDB_FHDR_SIZE;
H
Hongze Cheng 已提交
974 975 976 977 978 979 980
  } else {
    n = taosLSeekFile(pWriter->pDataFD, 0, SEEK_END);
    if (n < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

H
Hongze Cheng 已提交
981
    ASSERT(n == pWriter->fData.size);
H
Hongze Cheng 已提交
982 983 984
  }

  // last
H
Hongze Cheng 已提交
985
  if (pWriter->fLast.size == 0) {
H
Hongze Cheng 已提交
986 987 988 989
    flag = TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
  } else {
    flag = TD_FILE_WRITE;
  }
H
Hongze Cheng 已提交
990
  tsdbLastFileName(pTsdb, pWriter->wSet.diskId, pWriter->wSet.fid, &pWriter->fLast, fname);
H
Hongze Cheng 已提交
991 992 993 994 995
  pWriter->pLastFD = taosOpenFile(fname, flag);
  if (pWriter->pLastFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
H
Hongze Cheng 已提交
996
  if (pWriter->fLast.size == 0) {
H
Hongze Cheng 已提交
997 998 999 1000 1001 1002
    n = taosWriteFile(pWriter->pLastFD, hdr, TSDB_FHDR_SIZE);
    if (n < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

H
Hongze Cheng 已提交
1003
    pWriter->fLast.size += TSDB_FHDR_SIZE;
H
Hongze Cheng 已提交
1004 1005 1006 1007 1008 1009 1010
  } else {
    n = taosLSeekFile(pWriter->pLastFD, 0, SEEK_END);
    if (n < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

H
Hongze Cheng 已提交
1011
    ASSERT(n == pWriter->fLast.size);
H
Hongze Cheng 已提交
1012 1013 1014
  }

  // sma
H
Hongze Cheng 已提交
1015
  if (pWriter->fSma.size == 0) {
H
Hongze Cheng 已提交
1016 1017 1018 1019
    flag = TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
  } else {
    flag = TD_FILE_WRITE;
  }
H
Hongze Cheng 已提交
1020
  tsdbSmaFileName(pTsdb, pWriter->wSet.diskId, pWriter->wSet.fid, &pWriter->fSma, fname);
H
Hongze Cheng 已提交
1021 1022 1023 1024 1025
  pWriter->pSmaFD = taosOpenFile(fname, flag);
  if (pWriter->pSmaFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
H
Hongze Cheng 已提交
1026
  if (pWriter->fSma.size == 0) {
H
Hongze Cheng 已提交
1027 1028 1029 1030 1031 1032
    n = taosWriteFile(pWriter->pSmaFD, hdr, TSDB_FHDR_SIZE);
    if (n < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

H
Hongze Cheng 已提交
1033
    pWriter->fSma.size += TSDB_FHDR_SIZE;
H
Hongze Cheng 已提交
1034 1035 1036 1037 1038 1039 1040
  } else {
    n = taosLSeekFile(pWriter->pSmaFD, 0, SEEK_END);
    if (n < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

H
Hongze Cheng 已提交
1041
    ASSERT(n == pWriter->fSma.size);
H
Hongze Cheng 已提交
1042 1043 1044 1045 1046 1047
  }

  *ppWriter = pWriter;
  return code;

_err:
S
Shengliang Guan 已提交
1048
  tsdbError("vgId:%d, tsdb data file writer open failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1049
  *ppWriter = NULL;
H
Hongze Cheng 已提交
1050 1051 1052
  return code;
}

H
Hongze Cheng 已提交
1053
int32_t tsdbDataFWriterClose(SDataFWriter **ppWriter, int8_t sync) {
H
Hongze Cheng 已提交
1054
  int32_t code = 0;
H
Hongze Cheng 已提交
1055
  STsdb  *pTsdb = NULL;
H
Hongze Cheng 已提交
1056

H
Hongze Cheng 已提交
1057
  if (*ppWriter == NULL) goto _exit;
H
Hongze Cheng 已提交
1058 1059 1060

  pTsdb = (*ppWriter)->pTsdb;
  if (sync) {
H
Hongze Cheng 已提交
1061
    if (taosFsyncFile((*ppWriter)->pHeadFD) < 0) {
H
Hongze Cheng 已提交
1062 1063 1064 1065
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

H
Hongze Cheng 已提交
1066
    if (taosFsyncFile((*ppWriter)->pDataFD) < 0) {
H
Hongze Cheng 已提交
1067 1068 1069 1070
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

H
Hongze Cheng 已提交
1071
    if (taosFsyncFile((*ppWriter)->pLastFD) < 0) {
H
Hongze Cheng 已提交
1072 1073 1074 1075
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

H
Hongze Cheng 已提交
1076
    if (taosFsyncFile((*ppWriter)->pSmaFD) < 0) {
H
Hongze Cheng 已提交
1077 1078 1079 1080 1081
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }
  }

H
Hongze Cheng 已提交
1082
  if (taosCloseFile(&(*ppWriter)->pHeadFD) < 0) {
H
Hongze Cheng 已提交
1083 1084 1085 1086
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
1087
  if (taosCloseFile(&(*ppWriter)->pDataFD) < 0) {
H
Hongze Cheng 已提交
1088 1089 1090 1091
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
1092
  if (taosCloseFile(&(*ppWriter)->pLastFD) < 0) {
H
Hongze Cheng 已提交
1093 1094 1095 1096
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
1097
  if (taosCloseFile(&(*ppWriter)->pSmaFD) < 0) {
H
Hongze Cheng 已提交
1098 1099 1100 1101
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
1102 1103 1104
  for (int32_t iBuf = 0; iBuf < sizeof((*ppWriter)->aBuf) / sizeof(uint8_t *); iBuf++) {
    tFree((*ppWriter)->aBuf[iBuf]);
  }
H
Hongze Cheng 已提交
1105
  taosMemoryFree(*ppWriter);
H
Hongze Cheng 已提交
1106
_exit:
H
Hongze Cheng 已提交
1107
  *ppWriter = NULL;
H
Hongze Cheng 已提交
1108 1109 1110
  return code;

_err:
S
Shengliang Guan 已提交
1111
  tsdbError("vgId:%d, data file writer close failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1112 1113 1114
  return code;
}

H
Hongze Cheng 已提交
1115
int32_t tsdbUpdateDFileSetHeader(SDataFWriter *pWriter) {
H
Hongze Cheng 已提交
1116
  int32_t code = 0;
H
Hongze Cheng 已提交
1117 1118
  int64_t n;
  char    hdr[TSDB_FHDR_SIZE];
H
Hongze Cheng 已提交
1119 1120

  // head ==============
H
Hongze Cheng 已提交
1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135
  memset(hdr, 0, TSDB_FHDR_SIZE);
  tPutHeadFile(hdr, &pWriter->fHead);
  taosCalcChecksumAppend(0, hdr, TSDB_FHDR_SIZE);

  n = taosLSeekFile(pWriter->pHeadFD, 0, SEEK_SET);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  n = taosWriteFile(pWriter->pHeadFD, hdr, TSDB_FHDR_SIZE);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
H
Hongze Cheng 已提交
1136 1137

  // data ==============
H
Hongze Cheng 已提交
1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152
  memset(hdr, 0, TSDB_FHDR_SIZE);
  tPutDataFile(hdr, &pWriter->fData);
  taosCalcChecksumAppend(0, hdr, TSDB_FHDR_SIZE);

  n = taosLSeekFile(pWriter->pDataFD, 0, SEEK_SET);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  n = taosWriteFile(pWriter->pDataFD, hdr, TSDB_FHDR_SIZE);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
H
Hongze Cheng 已提交
1153 1154

  // last ==============
H
Hongze Cheng 已提交
1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169
  memset(hdr, 0, TSDB_FHDR_SIZE);
  tPutLastFile(hdr, &pWriter->fLast);
  taosCalcChecksumAppend(0, hdr, TSDB_FHDR_SIZE);

  n = taosLSeekFile(pWriter->pLastFD, 0, SEEK_SET);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  n = taosWriteFile(pWriter->pLastFD, hdr, TSDB_FHDR_SIZE);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
H
Hongze Cheng 已提交
1170 1171

  // sma ==============
H
Hongze Cheng 已提交
1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186
  memset(hdr, 0, TSDB_FHDR_SIZE);
  tPutSmaFile(hdr, &pWriter->fSma);
  taosCalcChecksumAppend(0, hdr, TSDB_FHDR_SIZE);

  n = taosLSeekFile(pWriter->pSmaFD, 0, SEEK_SET);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  n = taosWriteFile(pWriter->pSmaFD, hdr, TSDB_FHDR_SIZE);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
H
Hongze Cheng 已提交
1187 1188 1189 1190

  return code;

_err:
S
Shengliang Guan 已提交
1191
  tsdbError("vgId:%d, update DFileSet header failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1192 1193 1194
  return code;
}

H
Hongze Cheng 已提交
1195
int32_t tsdbWriteBlockIdx(SDataFWriter *pWriter, SArray *aBlockIdx) {
H
Hongze Cheng 已提交
1196
  int32_t    code = 0;
H
Hongze Cheng 已提交
1197
  SHeadFile *pHeadFile = &pWriter->fHead;
H
Hongze Cheng 已提交
1198
  int64_t    size = 0;
H
Hongze Cheng 已提交
1199 1200
  int64_t    n;

H
Hongze Cheng 已提交
1201 1202 1203 1204 1205
  // check
  if (taosArrayGetSize(aBlockIdx) == 0) {
    pHeadFile->offset = pHeadFile->size;
    goto _exit;
  }
H
Hongze Cheng 已提交
1206 1207

  // prepare
H
Hongze Cheng 已提交
1208
  size = sizeof(uint32_t);
H
Hongze Cheng 已提交
1209 1210 1211 1212
  for (int32_t iBlockIdx = 0; iBlockIdx < taosArrayGetSize(aBlockIdx); iBlockIdx++) {
    size += tPutBlockIdx(NULL, taosArrayGet(aBlockIdx, iBlockIdx));
  }
  size += sizeof(TSCKSUM);
H
Hongze Cheng 已提交
1213 1214

  // alloc
H
Hongze Cheng 已提交
1215
  code = tRealloc(&pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
1216 1217 1218
  if (code) goto _err;

  // build
H
Hongze Cheng 已提交
1219
  n = 0;
H
Hongze Cheng 已提交
1220
  n = tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT);
H
Hongze Cheng 已提交
1221
  for (int32_t iBlockIdx = 0; iBlockIdx < taosArrayGetSize(aBlockIdx); iBlockIdx++) {
H
Hongze Cheng 已提交
1222
    n += tPutBlockIdx(pWriter->aBuf[0] + n, taosArrayGet(aBlockIdx, iBlockIdx));
H
Hongze Cheng 已提交
1223
  }
H
Hongze Cheng 已提交
1224
  taosCalcChecksumAppend(0, pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
1225 1226 1227 1228

  ASSERT(n + sizeof(TSCKSUM) == size);

  // write
H
Hongze Cheng 已提交
1229
  n = taosWriteFile(pWriter->pHeadFD, pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
1230 1231 1232 1233 1234
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
1235 1236 1237
  // update
  pHeadFile->offset = pHeadFile->size;
  pHeadFile->size += size;
H
Hongze Cheng 已提交
1238

H
Hongze Cheng 已提交
1239
_exit:
H
Hongze Cheng 已提交
1240 1241
  tsdbTrace("vgId:%d write block idx, offset:%" PRId64 " size:%" PRId64 " nBlockIdx:%d", TD_VID(pWriter->pTsdb->pVnode),
            pHeadFile->offset, size, taosArrayGetSize(aBlockIdx));
H
Hongze Cheng 已提交
1242 1243 1244
  return code;

_err:
S
Shengliang Guan 已提交
1245
  tsdbError("vgId:%d, write block idx failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1246 1247 1248
  return code;
}

H
Hongze Cheng 已提交
1249 1250 1251 1252 1253
int32_t tsdbWriteBlock(SDataFWriter *pWriter, SMapData *mBlock, SBlockIdx *pBlockIdx) {
  int32_t    code = 0;
  SHeadFile *pHeadFile = &pWriter->fHead;
  int64_t    size;
  int64_t    n;
H
Hongze Cheng 已提交
1254

H
Hongze Cheng 已提交
1255
  ASSERT(mBlock->nItem > 0);
H
Hongze Cheng 已提交
1256 1257

  // alloc
H
Hongze Cheng 已提交
1258
  size = sizeof(uint32_t) + tPutMapData(NULL, mBlock) + sizeof(TSCKSUM);
H
Hongze Cheng 已提交
1259
  code = tRealloc(&pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
1260 1261 1262 1263
  if (code) goto _err;

  // build
  n = 0;
H
Hongze Cheng 已提交
1264 1265 1266
  n += tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT);
  n += tPutMapData(pWriter->aBuf[0] + n, mBlock);
  taosCalcChecksumAppend(0, pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
1267 1268 1269 1270

  ASSERT(n + sizeof(TSCKSUM) == size);

  // write
H
Hongze Cheng 已提交
1271
  n = taosWriteFile(pWriter->pHeadFD, pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
1272 1273 1274 1275 1276
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
1277 1278
  // update
  pBlockIdx->offset = pHeadFile->size;
H
Hongze Cheng 已提交
1279
  pBlockIdx->size = size;
H
Hongze Cheng 已提交
1280
  pHeadFile->size += size;
H
Hongze Cheng 已提交
1281

H
Hongze Cheng 已提交
1282 1283 1284 1285
  tsdbTrace("vgId:%d, write block, file ID:%d commit ID:%d suid:%" PRId64 " uid:%" PRId64 " offset:%" PRId64
            " size:%" PRId64 " nItem:%d",
            TD_VID(pWriter->pTsdb->pVnode), pWriter->wSet.fid, pHeadFile->commitID, pBlockIdx->suid, pBlockIdx->uid,
            pBlockIdx->offset, pBlockIdx->size, mBlock->nItem);
H
Hongze Cheng 已提交
1286 1287 1288
  return code;

_err:
S
Shengliang Guan 已提交
1289
  tsdbError("vgId:%d, write block failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1290 1291 1292
  return code;
}

H
Hongze Cheng 已提交
1293
int32_t tsdbWriteBlockL(SDataFWriter *pWriter, SArray *aBlockL) {
H
Hongze Cheng 已提交
1294
  int32_t    code = 0;
H
Hongze Cheng 已提交
1295
  SLastFile *pLastFile = &pWriter->fLast;
H
Hongze Cheng 已提交
1296 1297 1298
  int64_t    size;
  int64_t    n;

H
Hongze Cheng 已提交
1299 1300
  // check
  if (taosArrayGetSize(aBlockL) == 0) {
H
Hongze Cheng 已提交
1301
    pLastFile->offset = pLastFile->size;
H
Hongze Cheng 已提交
1302 1303 1304
    goto _exit;
  }

H
Hongze Cheng 已提交
1305
  // size
H
Hongze Cheng 已提交
1306
  size = sizeof(uint32_t);  // TSDB_FILE_DLMT
H
Hongze Cheng 已提交
1307 1308 1309 1310 1311 1312
  for (int32_t iBlockL = 0; iBlockL < taosArrayGetSize(aBlockL); iBlockL++) {
    size += tPutBlockL(NULL, taosArrayGet(aBlockL, iBlockL));
  }
  size += sizeof(TSCKSUM);

  // alloc
H
Hongze Cheng 已提交
1313
  code = tRealloc(&pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
1314 1315 1316 1317
  if (code) goto _err;

  // encode
  n = 0;
H
Hongze Cheng 已提交
1318
  n += tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT);
H
Hongze Cheng 已提交
1319
  for (int32_t iBlockL = 0; iBlockL < taosArrayGetSize(aBlockL); iBlockL++) {
H
Hongze Cheng 已提交
1320
    n += tPutBlockL(pWriter->aBuf[0] + n, taosArrayGet(aBlockL, iBlockL));
H
Hongze Cheng 已提交
1321
  }
H
Hongze Cheng 已提交
1322
  taosCalcChecksumAppend(0, pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
1323 1324 1325 1326

  ASSERT(n + sizeof(TSCKSUM) == size);

  // write
H
Hongze Cheng 已提交
1327
  n = taosWriteFile(pWriter->pLastFD, pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
1328 1329 1330 1331 1332 1333
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  // update
H
Hongze Cheng 已提交
1334 1335
  pLastFile->offset = pLastFile->size;
  pLastFile->size += size;
H
Hongze Cheng 已提交
1336

H
Hongze Cheng 已提交
1337
_exit:
H
Hongze Cheng 已提交
1338
  tsdbTrace("vgId:%d tsdb write blockl, loffset:%" PRId64 " size:%" PRId64, TD_VID(pWriter->pTsdb->pVnode),
H
Hongze Cheng 已提交
1339
            pLastFile->offset, size);
H
Hongze Cheng 已提交
1340 1341 1342 1343 1344 1345 1346
  return code;

_err:
  tsdbError("vgId:%d tsdb write blockl failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
  return code;
}

H
Hongze Cheng 已提交
1347
static void tsdbUpdateBlockInfo(SBlockData *pBlockData, SBlock *pBlock) {
H
Hongze Cheng 已提交
1348
  for (int32_t iRow = 0; iRow < pBlockData->nRow; iRow++) {
H
Hongze Cheng 已提交
1349 1350
    TSDBKEY key = {.ts = pBlockData->aTSKEY[iRow], .version = pBlockData->aVersion[iRow]};

H
Hongze Cheng 已提交
1351
    if (iRow == 0) {
H
Hongze Cheng 已提交
1352 1353 1354 1355 1356 1357 1358
      if (tsdbKeyCmprFn(&pBlock->minKey, &key) > 0) {
        pBlock->minKey = key;
      }
    } else {
      if (pBlockData->aTSKEY[iRow] == pBlockData->aTSKEY[iRow - 1]) {
        pBlock->hasDup = 1;
      }
H
Hongze Cheng 已提交
1359 1360
    }

H
Hongze Cheng 已提交
1361 1362
    if (iRow == pBlockData->nRow - 1 && tsdbKeyCmprFn(&pBlock->maxKey, &key) < 0) {
      pBlock->maxKey = key;
H
Hongze Cheng 已提交
1363 1364
    }

H
Hongze Cheng 已提交
1365 1366
    pBlock->minVer = TMIN(pBlock->minVer, key.version);
    pBlock->maxVer = TMAX(pBlock->maxVer, key.version);
H
Hongze Cheng 已提交
1367 1368
  }
  pBlock->nRow += pBlockData->nRow;
H
Hongze Cheng 已提交
1369
}
H
Hongze Cheng 已提交
1370

H
Hongze Cheng 已提交
1371 1372
static int32_t tsdbWriteBlockSma(SDataFWriter *pWriter, SBlockData *pBlockData, SSmaInfo *pSmaInfo) {
  int32_t code = 0;
H
Hongze Cheng 已提交
1373

H
Hongze Cheng 已提交
1374 1375 1376 1377
  pSmaInfo->offset = 0;
  pSmaInfo->size = 0;

  // encode
H
Hongze Cheng 已提交
1378
  for (int32_t iColData = 0; iColData < taosArrayGetSize(pBlockData->aIdx); iColData++) {
H
Hongze Cheng 已提交
1379 1380 1381
    SColData *pColData = tBlockDataGetColDataByIdx(pBlockData, iColData);

    if ((!pColData->smaOn) || IS_VAR_DATA_TYPE(pColData->type)) continue;
H
Hongze Cheng 已提交
1382

H
Hongze Cheng 已提交
1383 1384
    SColumnDataAgg sma;
    tsdbCalcColDataSMA(pColData, &sma);
H
Hongze Cheng 已提交
1385

H
Hongze Cheng 已提交
1386
    code = tRealloc(&pWriter->aBuf[0], pSmaInfo->size + tPutColumnDataAgg(NULL, &sma));
H
Hongze Cheng 已提交
1387
    if (code) goto _err;
H
Hongze Cheng 已提交
1388
    pSmaInfo->size += tPutColumnDataAgg(pWriter->aBuf[0] + pSmaInfo->size, &sma);
H
Hongze Cheng 已提交
1389
  }
H
Hongze Cheng 已提交
1390

H
Hongze Cheng 已提交
1391 1392 1393
  // write
  if (pSmaInfo->size) {
    int32_t size = pSmaInfo->size + sizeof(TSCKSUM);
H
Hongze Cheng 已提交
1394

H
Hongze Cheng 已提交
1395
    code = tRealloc(&pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
1396
    if (code) goto _err;
H
Hongze Cheng 已提交
1397

H
Hongze Cheng 已提交
1398
    taosCalcChecksumAppend(0, pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
1399

H
Hongze Cheng 已提交
1400
    int64_t n = taosWriteFile(pWriter->pSmaFD, pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
1401 1402 1403 1404 1405 1406 1407
    if (n < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

    pSmaInfo->offset = pWriter->fSma.size;
    pWriter->fSma.size += size;
H
Hongze Cheng 已提交
1408
  }
H
Hongze Cheng 已提交
1409

H
Hongze Cheng 已提交
1410
  return code;
H
Hongze Cheng 已提交
1411

H
Hongze Cheng 已提交
1412
_err:
H
Hongze Cheng 已提交
1413
  tsdbError("vgId:%d tsdb write block sma failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1414 1415
  return code;
}
H
Hongze Cheng 已提交
1416

H
Hongze Cheng 已提交
1417
int32_t tsdbWriteBlockData(SDataFWriter *pWriter, SBlockData *pBlockData, SBlockInfo *pBlkInfo, SSmaInfo *pSmaInfo,
H
Hongze Cheng 已提交
1418
                           int8_t cmprAlg, int8_t toLast) {
H
Hongze Cheng 已提交
1419
  int32_t code = 0;
H
Hongze Cheng 已提交
1420 1421

  ASSERT(pBlockData->nRow > 0);
H
Hongze Cheng 已提交
1422

H
Hongze Cheng 已提交
1423 1424 1425 1426
  pBlkInfo->offset = toLast ? pWriter->fLast.size : pWriter->fData.size;
  pBlkInfo->szBlock = 0;
  pBlkInfo->szKey = 0;

H
Hongze Cheng 已提交
1427 1428
  int32_t aBufN[4] = {0};
  code = tCmprBlockData(pBlockData, cmprAlg, NULL, NULL, pWriter->aBuf, aBufN);
H
Hongze Cheng 已提交
1429 1430
  if (code) goto _err;

H
Hongze Cheng 已提交
1431 1432 1433
  // write =================
  TdFilePtr pFD = toLast ? pWriter->pLastFD : pWriter->pDataFD;

H
Hongze Cheng 已提交
1434 1435
  pBlkInfo->szKey = aBufN[3] + aBufN[2];
  pBlkInfo->szBlock = aBufN[0] + aBufN[1] + aBufN[2] + aBufN[3];
H
Hongze Cheng 已提交
1436

H
Hongze Cheng 已提交
1437
  int64_t n = taosWriteFile(pFD, pWriter->aBuf[3], aBufN[3]);
H
Hongze Cheng 已提交
1438 1439 1440 1441 1442
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
1443
  n = taosWriteFile(pFD, pWriter->aBuf[2], aBufN[2]);
H
Hongze Cheng 已提交
1444 1445 1446 1447 1448
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
1449 1450
  if (aBufN[1]) {
    n = taosWriteFile(pFD, pWriter->aBuf[1], aBufN[1]);
H
Hongze Cheng 已提交
1451 1452 1453 1454 1455
    if (n < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }
  }
H
Hongze Cheng 已提交
1456

H
Hongze Cheng 已提交
1457 1458
  if (aBufN[0]) {
    n = taosWriteFile(pFD, pWriter->aBuf[0], aBufN[0]);
H
Hongze Cheng 已提交
1459 1460 1461 1462 1463
    if (n < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }
  }
H
Hongze Cheng 已提交
1464 1465 1466 1467 1468 1469 1470

  // update info
  if (toLast) {
    pWriter->fLast.size += pBlkInfo->szBlock;
  } else {
    pWriter->fData.size += pBlkInfo->szBlock;
  }
H
Hongze Cheng 已提交
1471

H
Hongze Cheng 已提交
1472 1473 1474 1475 1476
  // ================= SMA ====================
  if (pSmaInfo) {
    code = tsdbWriteBlockSma(pWriter, pBlockData, pSmaInfo);
    if (code) goto _err;
  }
H
Hongze Cheng 已提交
1477

H
Hongze Cheng 已提交
1478
_exit:
H
Hongze Cheng 已提交
1479 1480 1481
  tsdbTrace("vgId:%d tsdb write block data, suid:%" PRId64 " uid:%" PRId64 " nRow:%d, offset:%" PRId64 " size:%d",
            TD_VID(pWriter->pTsdb->pVnode), pBlockData->suid, pBlockData->uid, pBlockData->nRow, pBlkInfo->offset,
            pBlkInfo->szBlock);
H
Hongze Cheng 已提交
1482 1483
  return code;

H
Hongze Cheng 已提交
1484
_err:
H
Hongze Cheng 已提交
1485
  tsdbError("vgId:%d tsdb write block data failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1486
  return code;
H
Hongze Cheng 已提交
1487
}
H
Hongze Cheng 已提交
1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498

int32_t tsdbDFileSetCopy(STsdb *pTsdb, SDFileSet *pSetFrom, SDFileSet *pSetTo) {
  int32_t   code = 0;
  int64_t   n;
  int64_t   size;
  TdFilePtr pOutFD = NULL;  // TODO
  TdFilePtr PInFD = NULL;   // TODO
  char      fNameFrom[TSDB_FILENAME_LEN];
  char      fNameTo[TSDB_FILENAME_LEN];

  // head
H
Hongze Cheng 已提交
1499 1500
  tsdbHeadFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->pHeadF, fNameFrom);
  tsdbHeadFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->pHeadF, fNameTo);
H
Hongze Cheng 已提交
1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513

  pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
  if (pOutFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
  if (PInFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
1514
  n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pHeadF->size);
H
Hongze Cheng 已提交
1515 1516 1517 1518 1519 1520 1521 1522
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
  taosCloseFile(&pOutFD);
  taosCloseFile(&PInFD);

  // data
H
Hongze Cheng 已提交
1523 1524
  tsdbDataFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->pDataF, fNameFrom);
  tsdbDataFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->pDataF, fNameTo);
H
Hongze Cheng 已提交
1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537

  pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
  if (pOutFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
  if (PInFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
1538
  n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pDataF->size);
H
Hongze Cheng 已提交
1539 1540 1541 1542 1543 1544 1545 1546
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
  taosCloseFile(&pOutFD);
  taosCloseFile(&PInFD);

  // last
H
Hongze Cheng 已提交
1547 1548 1549
  tsdbLastFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->pLastF, fNameFrom);
  tsdbLastFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->pLastF, fNameTo);

H
Hongze Cheng 已提交
1550 1551 1552 1553 1554 1555 1556 1557 1558 1559 1560 1561
  pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
  if (pOutFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
  if (PInFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
1562
  n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pLastF->size);
H
Hongze Cheng 已提交
1563 1564 1565 1566 1567 1568 1569 1570
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
  taosCloseFile(&pOutFD);
  taosCloseFile(&PInFD);

  // sma
H
Hongze Cheng 已提交
1571 1572
  tsdbSmaFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->pSmaF, fNameFrom);
  tsdbSmaFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->pSmaF, fNameTo);
H
Hongze Cheng 已提交
1573 1574 1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585

  pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
  if (pOutFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
  if (PInFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
1586
  n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pSmaF->size);
H
Hongze Cheng 已提交
1587 1588 1589 1590 1591 1592 1593 1594 1595 1596
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
  taosCloseFile(&pOutFD);
  taosCloseFile(&PInFD);

  return code;

_err:
S
Shengliang Guan 已提交
1597
  tsdbError("vgId:%d, tsdb DFileSet copy failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1598
  return code;
H
Hongze Cheng 已提交
1599
}