tsdbReaderWriter.c 49.9 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 19
#define TSDB_FILE_DLMT ((uint32_t)0xF00AFA0F)

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

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

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

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

H
Hongze Cheng 已提交
51 52
  pDelFWriter->fDel.size = TSDB_FHDR_SIZE;
  pDelFWriter->fDel.size = 0;
H
more  
Hongze Cheng 已提交
53

H
Hongze Cheng 已提交
54
  *ppWriter = pDelFWriter;
H
Hongze Cheng 已提交
55 56 57 58
  return code;

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

H
Hongze Cheng 已提交
63
int32_t tsdbDelFWriterClose(SDelFWriter *pWriter, int8_t sync) {
H
Hongze Cheng 已提交
64
  int32_t code = 0;
H
Hongze Cheng 已提交
65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81

  // 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;
  }

  return code;

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

H
Hongze Cheng 已提交
85 86 87 88 89 90 91 92
int32_t tsdbWriteDelData(SDelFWriter *pWriter, SArray *aDelData, uint8_t **ppBuf, SDelIdx *pDelIdx) {
  int32_t       code = 0;
  uint8_t      *pBuf = NULL;
  int64_t       size;
  int64_t       n;
  SBlockDataHdr hdr = {.delimiter = TSDB_FILE_DLMT, .suid = pDelIdx->suid, .uid = pDelIdx->uid};

  if (!ppBuf) ppBuf = &pBuf;
H
Hongze Cheng 已提交
93

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

  // alloc
  code = tsdbRealloc(ppBuf, size);
  if (code) goto _err;

  // build
H
Hongze Cheng 已提交
106 107 108 109
  n = 0;
  *(SBlockDataHdr *)(*ppBuf) = hdr;
  n += sizeof(hdr);
  for (int32_t iDelData = 0; iDelData < taosArrayGetSize(aDelData); iDelData++) {
H
Hongze Cheng 已提交
110
    n += tPutDelData(*ppBuf + n, taosArrayGet(aDelData, iDelData));
H
Hongze Cheng 已提交
111
  }
H
Hongze Cheng 已提交
112 113 114 115 116 117 118 119 120 121 122 123 124 125
  taosCalcChecksumAppend(0, *ppBuf, size);

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

  // write
  n = taosWriteFile(pWriter->pWriteH, *ppBuf, size);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  ASSERT(n == size);

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

  tsdbFree(pBuf);
  return code;

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

H
Hongze Cheng 已提交
139
int32_t tsdbWriteDelIdx(SDelFWriter *pWriter, SArray *aDelIdx, uint8_t **ppBuf) {
H
Hongze Cheng 已提交
140
  int32_t  code = 0;
H
Hongze Cheng 已提交
141 142
  int64_t  size;
  int64_t  n;
H
Hongze Cheng 已提交
143
  uint8_t *pBuf = NULL;
H
Hongze Cheng 已提交
144 145 146
  SDelIdx *pDelIdx;

  if (!ppBuf) ppBuf = &pBuf;
H
Hongze Cheng 已提交
147

H
Hongze Cheng 已提交
148
  // prepare
H
Hongze Cheng 已提交
149
  size = 0;
H
Hongze Cheng 已提交
150
  size += tPutU32(NULL, TSDB_FILE_DLMT);
H
Hongze Cheng 已提交
151 152 153 154
  for (int32_t iDelIdx = 0; iDelIdx < taosArrayGetSize(aDelIdx); iDelIdx++) {
    size += tPutDelIdx(NULL, taosArrayGet(aDelIdx, iDelIdx));
  }
  size += sizeof(TSCKSUM);
H
Hongze Cheng 已提交
155 156 157

  // alloc
  code = tsdbRealloc(ppBuf, size);
H
Hongze Cheng 已提交
158
  if (code) goto _err;
H
Hongze Cheng 已提交
159

H
Hongze Cheng 已提交
160
  // build
H
Hongze Cheng 已提交
161
  n = 0;
H
Hongze Cheng 已提交
162
  n += tPutU32(*ppBuf + n, TSDB_FILE_DLMT);
H
Hongze Cheng 已提交
163 164 165
  for (int32_t iDelIdx = 0; iDelIdx < taosArrayGetSize(aDelIdx); iDelIdx++) {
    n += tPutDelIdx(*ppBuf + n, taosArrayGet(aDelIdx, iDelIdx));
  }
H
Hongze Cheng 已提交
166 167
  taosCalcChecksumAppend(0, *ppBuf, size);

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

H
Hongze Cheng 已提交
170
  // write
H
Hongze Cheng 已提交
171 172
  n = taosWriteFile(pWriter->pWriteH, *ppBuf, size);
  if (n < 0) {
H
Hongze Cheng 已提交
173 174 175 176
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
177
  // update
H
Hongze Cheng 已提交
178 179
  pWriter->fDel.offset = pWriter->fDel.size;
  pWriter->fDel.size += size;
H
Hongze Cheng 已提交
180

H
Hongze Cheng 已提交
181 182 183 184
  tsdbFree(pBuf);
  return code;

_err:
H
more  
Hongze Cheng 已提交
185
  tsdbError("vgId:%d write del idx failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
186 187 188 189
  tsdbFree(pBuf);
  return code;
}

H
Hongze Cheng 已提交
190 191 192 193 194
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 已提交
195 196

  // build
H
Hongze Cheng 已提交
197 198 199
  memset(hdr, 0, size);
  tPutDelFile(hdr, &pWriter->fDel);
  taosCalcChecksumAppend(0, hdr, size);
H
Hongze Cheng 已提交
200 201 202 203 204 205 206 207

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

  // write
H
Hongze Cheng 已提交
208 209
  n = taosWriteFile(pWriter->pWriteH, hdr, size);
  if (n < 0) {
H
Hongze Cheng 已提交
210 211 212 213
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
214 215 216
  return code;

_err:
H
Hongze Cheng 已提交
217
  tsdbError("vgId:%d update del file hdr failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
218 219 220 221 222
  return code;
}

// SDelFReader ====================================================
struct SDelFReader {
H
Hongze Cheng 已提交
223
  STsdb    *pTsdb;
H
Hongze Cheng 已提交
224
  SDelFile  fDel;
H
Hongze Cheng 已提交
225
  TdFilePtr pReadH;
H
refact  
Hongze Cheng 已提交
226
};
H
Hongze Cheng 已提交
227

H
Hongze Cheng 已提交
228 229
int32_t tsdbDelFReaderOpen(SDelFReader **ppReader, SDelFile *pFile, STsdb *pTsdb, uint8_t **ppBuf) {
  int32_t      code = 0;
H
Hongze Cheng 已提交
230
  char         fname[TSDB_FILENAME_LEN];
H
Hongze Cheng 已提交
231
  SDelFReader *pDelFReader;
H
Hongze Cheng 已提交
232
  int64_t      n;
H
Hongze Cheng 已提交
233 234 235 236 237 238 239 240 241 242

  // 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 已提交
243 244 245
  pDelFReader->fDel = *pFile;

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

H
Hongze Cheng 已提交
253
#if 0
H
Hongze Cheng 已提交
254 255 256 257 258 259 260
  // load and check hdr if buffer is given
  if (ppBuf) {
    code = tsdbRealloc(ppBuf, TSDB_FHDR_SIZE);
    if (code) {
      goto _err;
    }

H
Hongze Cheng 已提交
261 262 263 264 265
    n = taosReadFile(pDelFReader->pReadH, *ppBuf, TSDB_FHDR_SIZE);
    if (n < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    } else if (n < TSDB_FHDR_SIZE) {
H
Hongze Cheng 已提交
266 267 268 269 270 271 272 273 274 275 276
      code = TSDB_CODE_FILE_CORRUPTED;
      goto _err;
    }

    if (!taosCheckChecksumWhole(*ppBuf, TSDB_FHDR_SIZE)) {
      code = TSDB_CODE_FILE_CORRUPTED;
      goto _err;
    }

    // TODO: check the content
  }
H
Hongze Cheng 已提交
277
#endif
H
Hongze Cheng 已提交
278 279 280 281 282 283

_exit:
  *ppReader = pDelFReader;
  return code;

_err:
H
Hongze Cheng 已提交
284
  tsdbError("vgId:%d del file reader open failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
285
  *ppReader = NULL;
H
Hongze Cheng 已提交
286 287 288 289 290
  return code;
}

int32_t tsdbDelFReaderClose(SDelFReader *pReader) {
  int32_t code = 0;
H
Hongze Cheng 已提交
291 292

  if (pReader) {
H
Hongze Cheng 已提交
293 294 295 296
    if (taosCloseFile(&pReader->pReadH) < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _exit;
    }
H
Hongze Cheng 已提交
297 298 299
    taosMemoryFree(pReader);
  }

H
Hongze Cheng 已提交
300
_exit:
H
Hongze Cheng 已提交
301 302 303
  return code;
}

H
Hongze Cheng 已提交
304 305 306 307 308 309 310 311 312 313
int32_t tsdbReadDelData(SDelFReader *pReader, SDelIdx *pDelIdx, SArray *aDelData, uint8_t **ppBuf) {
  int32_t        code = 0;
  int64_t        offset = pDelIdx->offset;
  int64_t        size = pDelIdx->size;
  int64_t        n;
  uint8_t       *pBuf = NULL;
  SBlockDataHdr *pHdr;
  SDelData      *pDelData = &(SDelData){0};

  if (!ppBuf) ppBuf = &pBuf;
H
Hongze Cheng 已提交
314 315

  // seek
H
Hongze Cheng 已提交
316
  if (taosLSeekFile(pReader->pReadH, offset, SEEK_SET) < 0) {
H
Hongze Cheng 已提交
317 318 319 320 321
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  // alloc
H
Hongze Cheng 已提交
322
  code = tsdbRealloc(ppBuf, size);
H
Hongze Cheng 已提交
323 324 325
  if (code) goto _err;

  // read
H
Hongze Cheng 已提交
326
  n = taosReadFile(pReader->pReadH, *ppBuf, size);
H
Hongze Cheng 已提交
327 328 329
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
H
Hongze Cheng 已提交
330
  } else if (n < size) {
H
Hongze Cheng 已提交
331 332
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
H
Hongze Cheng 已提交
333 334 335
  }

  // check
H
Hongze Cheng 已提交
336
  if (!taosCheckChecksumWhole(*ppBuf, size)) {
H
Hongze Cheng 已提交
337 338 339 340
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

H
Hongze Cheng 已提交
341 342
  // // decode
  n = 0;
H
Hongze Cheng 已提交
343 344 345 346 347 348 349 350
  pHdr = (SBlockDataHdr *)(*ppBuf + n);
  ASSERT(pHdr->delimiter == TSDB_FILE_DLMT);
  ASSERT(pHdr->suid == pDelIdx->suid);
  ASSERT(pHdr->uid == pDelIdx->uid);
  n += sizeof(*pHdr);
  while (n < size - sizeof(TSCKSUM)) {
    n += tGetDelData(*ppBuf + n, pDelData);
  }
H
Hongze Cheng 已提交
351

H
Hongze Cheng 已提交
352 353 354
  ASSERT(n == size - sizeof(TSCKSUM));

  tsdbFree(pBuf);
H
Hongze Cheng 已提交
355 356 357 358
  return code;

_err:
  tsdbError("vgId:%d read del data failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
359
  tsdbFree(pBuf);
H
Hongze Cheng 已提交
360 361 362
  return code;
}

H
Hongze Cheng 已提交
363
int32_t tsdbReadDelIdx(SDelFReader *pReader, SArray *aDelIdx, uint8_t **ppBuf) {
H
Hongze Cheng 已提交
364 365
  int32_t  code = 0;
  int32_t  n;
H
Hongze Cheng 已提交
366 367
  int64_t  offset = pReader->fDel.offset;
  int64_t  size = pReader->fDel.size - offset;
H
Hongze Cheng 已提交
368
  uint32_t delimiter;
H
Hongze Cheng 已提交
369
  uint8_t *pBuf = NULL;
370
  SDelIdx *pDelIdx = &(SDelIdx){0};
H
Hongze Cheng 已提交
371

H
Hongze Cheng 已提交
372
  if (!ppBuf) ppBuf = &pBuf;
H
more  
Hongze Cheng 已提交
373

H
Hongze Cheng 已提交
374
  // seek
H
Hongze Cheng 已提交
375
  if (taosLSeekFile(pReader->pReadH, offset, SEEK_SET) < 0) {
H
Hongze Cheng 已提交
376 377 378 379
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
380 381
  // alloc
  code = tsdbRealloc(ppBuf, size);
H
Hongze Cheng 已提交
382
  if (code) goto _err;
H
Hongze Cheng 已提交
383

H
Hongze Cheng 已提交
384
  // read
H
Hongze Cheng 已提交
385 386
  n = taosReadFile(pReader->pReadH, *ppBuf, size);
  if (n < 0) {
H
Hongze Cheng 已提交
387
    code = TAOS_SYSTEM_ERROR(errno);
H
Hongze Cheng 已提交
388
    goto _err;
H
Hongze Cheng 已提交
389 390 391
  } else if (n < size) {
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
H
Hongze Cheng 已提交
392 393 394 395 396 397 398 399 400
  }

  // check
  if (!taosCheckChecksumWhole(*ppBuf, size)) {
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

  // decode
H
Hongze Cheng 已提交
401 402 403
  n = 0;
  n += tGetU32(*ppBuf + n, &delimiter);
  ASSERT(delimiter == TSDB_FILE_DLMT);
H
Hongze Cheng 已提交
404 405 406 407 408 409 410 411 412 413 414 415

  taosArrayClear(aDelIdx);
  while (n < size - sizeof(TSCKSUM)) {
    n += tGetDelIdx(*ppBuf + n, pDelIdx);

    if (taosArrayPush(aDelIdx, pDelIdx) == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
    }
  }

  ASSERT(n == size - sizeof(TSCKSUM));
H
Hongze Cheng 已提交
416 417 418 419

  return code;

_err:
H
Hongze Cheng 已提交
420
  tsdbError("vgId:%d read del idx failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
421
  return code;
H
Hongze Cheng 已提交
422 423 424 425 426 427
}

// SDataFReader ====================================================
struct SDataFReader {
  STsdb     *pTsdb;
  SDFileSet *pSet;
H
Hongze Cheng 已提交
428 429 430 431
  TdFilePtr  pHeadFD;
  TdFilePtr  pDataFD;
  TdFilePtr  pLastFD;
  TdFilePtr  pSmaFD;
H
Hongze Cheng 已提交
432 433 434
};

int32_t tsdbDataFReaderOpen(SDataFReader **ppReader, STsdb *pTsdb, SDFileSet *pSet) {
H
Hongze Cheng 已提交
435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486
  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
  tsdbDataFileName(pTsdb, pSet, TSDB_HEAD_FILE, fname);
  pReader->pHeadFD = taosOpenFile(fname, TD_FILE_READ);
  if (pReader->pHeadFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  // data
  tsdbDataFileName(pTsdb, pSet, TSDB_DATA_FILE, fname);
  pReader->pDataFD = taosOpenFile(fname, TD_FILE_READ);
  if (pReader->pDataFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  // last
  tsdbDataFileName(pTsdb, pSet, TSDB_LAST_FILE, fname);
  pReader->pLastFD = taosOpenFile(fname, TD_FILE_READ);
  if (pReader->pLastFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  // sma
  tsdbDataFileName(pTsdb, pSet, TSDB_SMA_FILE, fname);
  pReader->pSmaFD = taosOpenFile(fname, TD_FILE_READ);
  if (pReader->pSmaFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  *ppReader = pReader;
  return code;

_err:
  tsdbError("vgId:%d tsdb data file reader open failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
  *ppReader = NULL;
H
Hongze Cheng 已提交
487 488 489
  return code;
}

H
Hongze Cheng 已提交
490
int32_t tsdbDataFReaderClose(SDataFReader **ppReader) {
H
Hongze Cheng 已提交
491
  int32_t code = 0;
H
Hongze Cheng 已提交
492
  if (*ppReader == NULL) goto _exit;
H
Hongze Cheng 已提交
493

H
Hongze Cheng 已提交
494
  if (taosCloseFile(&(*ppReader)->pHeadFD) < 0) {
H
Hongze Cheng 已提交
495 496 497 498
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
499
  if (taosCloseFile(&(*ppReader)->pDataFD) < 0) {
H
Hongze Cheng 已提交
500 501 502 503
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
504
  if (taosCloseFile(&(*ppReader)->pLastFD) < 0) {
H
Hongze Cheng 已提交
505 506 507 508
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
509
  if (taosCloseFile(&(*ppReader)->pSmaFD) < 0) {
H
Hongze Cheng 已提交
510 511 512 513
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
514
  taosMemoryFree(*ppReader);
H
Hongze Cheng 已提交
515 516

_exit:
H
Hongze Cheng 已提交
517
  *ppReader = NULL;
H
Hongze Cheng 已提交
518 519 520
  return code;

_err:
H
Hongze Cheng 已提交
521
  tsdbError("vgId:%d data file reader close failed since %s", TD_VID((*ppReader)->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
522 523 524
  return code;
}

H
Hongze Cheng 已提交
525 526 527 528 529 530 531 532 533 534
int32_t tsdbReadBlockIdx(SDataFReader *pReader, SArray *aBlockIdx, uint8_t **ppBuf) {
  int32_t   code = 0;
  int64_t   offset = pReader->pSet->fHead.offset;
  int64_t   size = pReader->pSet->fHead.size - offset;
  uint8_t  *pBuf = NULL;
  int64_t   n;
  uint32_t  delimiter;
  SBlockIdx blockIdx;

  if (!ppBuf) ppBuf = &pBuf;
H
Hongze Cheng 已提交
535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562

  // alloc
  code = tsdbRealloc(ppBuf, size);
  if (code) goto _err;

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

  // read
  n = taosReadFile(pReader->pHeadFD, *ppBuf, size);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  } else if (n < size) {
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

  // check
  if (!taosCheckChecksumWhole(*ppBuf, size)) {
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

  // decode
H
Hongze Cheng 已提交
563
  n = 0;
H
Hongze Cheng 已提交
564
  n = tGetU32(*ppBuf + n, &delimiter);
H
Hongze Cheng 已提交
565
  ASSERT(delimiter == TSDB_FILE_DLMT);
H
Hongze Cheng 已提交
566 567 568 569 570 571 572 573 574 575 576

  taosArrayClear(aBlockIdx);
  while (n < size - sizeof(TSCKSUM)) {
    n += tGetBlockIdx(*ppBuf + n, &blockIdx);

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

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

H
Hongze Cheng 已提交
579
  tsdbFree(pBuf);
H
Hongze Cheng 已提交
580 581 582 583
  return code;

_err:
  tsdbError("vgId:%d read block idx failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
584
  tsdbFree(pBuf);
H
Hongze Cheng 已提交
585 586 587
  return code;
}

H
Hongze Cheng 已提交
588
int32_t tsdbReadBlock(SDataFReader *pReader, SBlockIdx *pBlockIdx, SMapData *mBlock, uint8_t **ppBuf) {
H
Hongze Cheng 已提交
589 590 591
  int32_t       code = 0;
  int64_t       offset = pBlockIdx->offset;
  int64_t       size = pBlockIdx->size;
H
Hongze Cheng 已提交
592
  uint8_t      *pBuf = NULL;
H
Hongze Cheng 已提交
593
  int64_t       n;
H
Hongze Cheng 已提交
594
  int64_t       tn;
H
Hongze Cheng 已提交
595
  SBlockDataHdr hdr;
H
Hongze Cheng 已提交
596

H
Hongze Cheng 已提交
597 598
  if (!ppBuf) ppBuf = &pBuf;

H
Hongze Cheng 已提交
599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625
  // alloc
  code = tsdbRealloc(ppBuf, size);
  if (code) goto _err;

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

  // read
  n = taosReadFile(pReader->pHeadFD, *ppBuf, size);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  } else if (n < size) {
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

  // check
  if (!taosCheckChecksumWhole(*ppBuf, size)) {
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

  // decode
H
Hongze Cheng 已提交
626 627 628 629 630 631
  hdr = *(SBlockDataHdr *)(*ppBuf);
  ASSERT(hdr.delimiter == TSDB_FILE_DLMT);
  ASSERT(hdr.suid == pBlockIdx->suid);
  ASSERT(hdr.uid == pBlockIdx->uid);

  n = sizeof(hdr);
H
Hongze Cheng 已提交
632 633 634 635 636 637
  tn = tGetMapData(*ppBuf + n, mBlock);
  if (tn < 0) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
  n += tn;
H
Hongze Cheng 已提交
638 639
  ASSERT(n + sizeof(TSCKSUM) == size);

H
Hongze Cheng 已提交
640
  tsdbFree(pBuf);
H
Hongze Cheng 已提交
641 642 643 644
  return code;

_err:
  tsdbError("vgId:%d read block failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
645
  tsdbFree(pBuf);
H
Hongze Cheng 已提交
646 647 648
  return code;
}

H
Hongze Cheng 已提交
649
static int32_t tsdbReadBlockDataKey(SBlockData *pBlockData, SSubBlock *pSubBlock, uint8_t *pBuf, uint8_t **ppBuf) {
H
Hongze Cheng 已提交
650
  int32_t code = 0;
H
refact  
Hongze Cheng 已提交
651
  int64_t size = pSubBlock->szVersion + pSubBlock->szTSKEY + sizeof(TSCKSUM);
H
Hongze Cheng 已提交
652 653 654 655 656 657 658 659 660 661 662 663 664
  int64_t n;

  if (!taosCheckChecksumWhole(pBuf, size)) {
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

  code = tsdbRealloc((uint8_t **)&pBlockData->aVersion, sizeof(int64_t) * pSubBlock->nRow);
  if (code) goto _err;
  code = tsdbRealloc((uint8_t **)&pBlockData->aTSKEY, sizeof(TSKEY) * pSubBlock->nRow);
  if (code) goto _err;

  if (pSubBlock->cmprAlg == NO_COMPRESSION) {
H
refact  
Hongze Cheng 已提交
665 666
    ASSERT(pSubBlock->szVersion == sizeof(int64_t) * pSubBlock->nRow);
    ASSERT(pSubBlock->szTSKEY == sizeof(TSKEY) * pSubBlock->nRow);
H
Hongze Cheng 已提交
667 668

    // VERSION
H
refact  
Hongze Cheng 已提交
669
    memcpy(pBlockData->aVersion, pBuf, pSubBlock->szVersion);
H
Hongze Cheng 已提交
670 671

    // TSKEY
H
refact  
Hongze Cheng 已提交
672
    memcpy(pBlockData->aTSKEY, pBuf + pSubBlock->szVersion, pSubBlock->szTSKEY);
H
Hongze Cheng 已提交
673 674 675 676 677 678 679 680
  } else {
    size = sizeof(int64_t) * pSubBlock->nRow + COMP_OVERFLOW_BYTES;
    if (pSubBlock->cmprAlg == TWO_STAGE_COMP) {
      code = tsdbRealloc(ppBuf, size);
      if (code) goto _err;
    }

    // VERSION
H
refact  
Hongze Cheng 已提交
681
    n = tsDecompressBigint(pBuf, pSubBlock->szVersion, pSubBlock->nRow, (char *)pBlockData->aVersion,
H
Hongze Cheng 已提交
682 683 684 685 686 687 688
                           sizeof(int64_t) * pSubBlock->nRow, pSubBlock->cmprAlg, *ppBuf, size);
    if (n < 0) {
      code = TSDB_CODE_COMPRESS_ERROR;
      goto _err;
    }

    // TSKEY
H
Hongze Cheng 已提交
689 690 691
    n = tsDecompressTimestamp(pBuf + pSubBlock->szVersion, pSubBlock->szTSKEY, pSubBlock->nRow,
                              (char *)pBlockData->aTSKEY, sizeof(TSKEY) * pSubBlock->nRow, pSubBlock->cmprAlg, *ppBuf,
                              size);
H
Hongze Cheng 已提交
692 693 694 695 696 697 698 699 700 701 702 703
    if (n < 0) {
      code = TSDB_CODE_COMPRESS_ERROR;
      goto _err;
    }
  }

  return code;

_err:
  return code;
}

H
Hongze Cheng 已提交
704 705
static int32_t tsdbReadColDataImpl(SSubBlock *pSubBlock, SBlockCol *pBlockCol, SColData *pColData, uint8_t *pBuf,
                                   uint8_t **ppBuf) {
H
Hongze Cheng 已提交
706 707 708 709
  int32_t code = 0;
  int64_t size;
  int64_t n;

H
Hongze Cheng 已提交
710
  if (!taosCheckChecksumWhole(pBuf, pBlockCol->szBitmap + pBlockCol->szOffset + pBlockCol->szValue + sizeof(TSCKSUM))) {
H
Hongze Cheng 已提交
711 712 713 714 715 716 717
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

  pColData->nVal = pSubBlock->nRow;
  pColData->flag = pBlockCol->flag;

H
Hongze Cheng 已提交
718
  // BITMAP
H
Hongze Cheng 已提交
719
  if (pBlockCol->flag != HAS_VALUE) {
H
Hongze Cheng 已提交
720 721 722
    ASSERT(pBlockCol->szBitmap);

    size = BIT2_SIZE(pColData->nVal);
H
Hongze Cheng 已提交
723 724 725
    code = tsdbRealloc(&pColData->pBitMap, size);
    if (code) goto _err;

H
Hongze Cheng 已提交
726 727
    code = tsdbRealloc(ppBuf, size + COMP_OVERFLOW_BYTES);
    if (code) goto _err;
H
Hongze Cheng 已提交
728

H
Hongze Cheng 已提交
729 730 731 732 733 734 735 736
    n = tsDecompressTinyint(pBuf, pBlockCol->szBitmap, size, pColData->pBitMap, size, TWO_STAGE_COMP, *ppBuf,
                            size + COMP_OVERFLOW_BYTES);
    if (n <= 0) {
      code = TSDB_CODE_COMPRESS_ERROR;
      goto _err;
    }

    ASSERT(n == size);
H
Hongze Cheng 已提交
737
  } else {
H
Hongze Cheng 已提交
738
    ASSERT(pBlockCol->szBitmap == 0);
H
Hongze Cheng 已提交
739
  }
H
Hongze Cheng 已提交
740 741 742 743 744
  pBuf = pBuf + pBlockCol->szBitmap;

  // OFFSET
  if (IS_VAR_DATA_TYPE(pColData->type)) {
    ASSERT(pBlockCol->szOffset);
H
Hongze Cheng 已提交
745

H
Hongze Cheng 已提交
746 747 748 749 750 751 752 753 754 755 756 757 758 759 760
    size = sizeof(int32_t) * pColData->nVal;
    code = tsdbRealloc((uint8_t **)&pColData->aOffset, size);
    if (code) goto _err;

    code = tsdbRealloc(ppBuf, size + COMP_OVERFLOW_BYTES);
    if (code) goto _err;

    n = tsDecompressInt(pBuf, pBlockCol->szOffset, pColData->nVal, (char *)pColData->aOffset, size, TWO_STAGE_COMP,
                        *ppBuf, size + COMP_OVERFLOW_BYTES);
    if (n <= 0) {
      code = TSDB_CODE_COMPRESS_ERROR;
      goto _err;
    }

    ASSERT(n == size);
H
Hongze Cheng 已提交
761
  } else {
H
Hongze Cheng 已提交
762
    ASSERT(pBlockCol->szOffset == 0);
H
Hongze Cheng 已提交
763
  }
H
Hongze Cheng 已提交
764 765 766 767 768
  pBuf = pBuf + pBlockCol->szOffset;

  // VALUE
  pColData->nData = pBlockCol->szOrigin;

H
Hongze Cheng 已提交
769 770 771 772 773 774 775
  code = tsdbRealloc(&pColData->pData, pColData->nData);
  if (code) goto _err;

  if (pSubBlock->cmprAlg == NO_COMPRESSION) {
    memcpy(pColData->pData, pBuf, pColData->nData);
  } else {
    if (pSubBlock->cmprAlg == TWO_STAGE_COMP) {
H
Hongze Cheng 已提交
776
      code = tsdbRealloc(ppBuf, pColData->nData + COMP_OVERFLOW_BYTES);
H
Hongze Cheng 已提交
777 778 779
      if (code) goto _err;
    }

H
Hongze Cheng 已提交
780 781 782
    n = tDataTypes[pBlockCol->type].decompFunc(pBuf, pBlockCol->szValue, pSubBlock->nRow, pColData->pData,
                                               pColData->nData, pSubBlock->cmprAlg, *ppBuf,
                                               pColData->nData + COMP_OVERFLOW_BYTES);
H
Hongze Cheng 已提交
783 784 785 786 787 788 789 790 791 792 793 794 795 796
    if (n < 0) {
      code = TSDB_CODE_COMPRESS_ERROR;
      goto _err;
    }

    ASSERT(n == pColData->nData);
  }

  return code;

_err:
  return code;
}

H
Hongze Cheng 已提交
797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828
static int32_t tsdbReadBlockCol(SSubBlock *pSubBlock, uint8_t *p, SArray *aBlockCol) {
  int32_t    code = 0;
  int32_t    n = 0;
  SBlockCol  blockCol;
  SBlockCol *pBlockCol = &blockCol;

  if (!taosCheckChecksumWhole(p, pSubBlock->szBlockCol + sizeof(TSCKSUM))) {
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

  n += sizeof(SBlockDataHdr);
  while (n < pSubBlock->szBlockCol) {
    n += tGetBlockCol(p + n, pBlockCol);

    if (taosArrayPush(aBlockCol, pBlockCol) == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
    }
  }

  ASSERT(n == pSubBlock->szBlockCol);

  return code;

_err:
  return code;
}

static int32_t tsdbReadSubColData(SDataFReader *pReader, SBlockIdx *pBlockIdx, SBlock *pBlock, int32_t iSubBlock,
                                  int16_t *aColId, int32_t nCol, SBlockData *pBlockData, uint8_t **ppBuf1,
                                  uint8_t **ppBuf2) {
H
Hongze Cheng 已提交
829 830
  TdFilePtr  pFD = pBlock->last ? pReader->pLastFD : pReader->pDataFD;
  SSubBlock *pSubBlock = &pBlock->aSubBlock[iSubBlock];
H
Hongze Cheng 已提交
831
  SArray    *aBlockCol = NULL;
H
Hongze Cheng 已提交
832 833 834 835
  int32_t    code = 0;
  int64_t    offset;
  int64_t    size;
  int64_t    n;
H
Hongze Cheng 已提交
836

H
Hongze Cheng 已提交
837 838
  tBlockDataReset(pBlockData);
  pBlockData->nRow = pSubBlock->nRow;
H
Hongze Cheng 已提交
839

H
Hongze Cheng 已提交
840 841 842 843 844 845 846 847 848
  // TSDBKEY and SBlockCol
  if (nCol == 1) {
    offset = pSubBlock->offset + pSubBlock->szBlockCol + sizeof(TSCKSUM);
    size = pSubBlock->szVersion + pSubBlock->szTSKEY + sizeof(TSCKSUM);
  } else {
    offset = pSubBlock->offset;
    size = pSubBlock->szBlockCol + sizeof(TSCKSUM) + pSubBlock->szVersion + pSubBlock->szTSKEY + sizeof(TSCKSUM);
  }

H
Hongze Cheng 已提交
849 850
  code = tsdbRealloc(ppBuf1, size);
  if (code) goto _err;
H
Hongze Cheng 已提交
851

H
Hongze Cheng 已提交
852 853 854 855 856
  n = taosLSeekFile(pFD, offset, SEEK_SET);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
H
Hongze Cheng 已提交
857

H
Hongze Cheng 已提交
858 859 860 861 862 863 864 865
  n = taosReadFile(pFD, *ppBuf1, size);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  } else if (n < size) {
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }
H
Hongze Cheng 已提交
866

H
Hongze Cheng 已提交
867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884
  if (nCol == 1) {
    code = tsdbReadBlockDataKey(pBlockData, pSubBlock, *ppBuf1, ppBuf2);
    if (code) goto _err;

    goto _exit;
  } else {
    aBlockCol = taosArrayInit(0, sizeof(SBlockCol));
    if (aBlockCol == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
    }

    code = tsdbReadBlockCol(pSubBlock, *ppBuf1, aBlockCol);
    if (code) goto _err;

    code = tsdbReadBlockDataKey(pBlockData, pSubBlock, *ppBuf1 + pSubBlock->szBlockCol + sizeof(TSCKSUM), ppBuf2);
    if (code) goto _err;
  }
H
Hongze Cheng 已提交
885

H
Hongze Cheng 已提交
886
  for (int32_t iCol = 1; iCol < nCol; iCol++) {
H
Hongze Cheng 已提交
887 888 889 890 891 892 893
    void *p = taosArraySearch(aBlockCol, &(SBlockCol){.cid = aColId[iCol]}, tBlockColCmprFn, TD_EQ);

    if (p) {
      SBlockCol *pBlockCol = (SBlockCol *)p;
      SColData  *pColData;

      ASSERT(pBlockCol->flag && pBlockCol->flag != HAS_NONE);
H
Hongze Cheng 已提交
894

H
Hongze Cheng 已提交
895 896
      code = tBlockDataAddColData(pBlockData, taosArrayGetSize(pBlockData->aColDataP), &pColData);
      if (code) goto _err;
H
Hongze Cheng 已提交
897

H
Hongze Cheng 已提交
898
      tColDataInit(pColData, pBlockCol->cid, pBlockCol->type, pBlockCol->smaOn);
H
Hongze Cheng 已提交
899 900 901 902 903 904
      if (pBlockCol->flag == HAS_NULL) {
        for (int32_t iRow = 0; iRow < pSubBlock->nRow; iRow++) {
          code = tColDataAppendValue(pColData, &COL_VAL_NULL(pBlockCol->cid, pBlockCol->type));
          if (code) goto _err;
        }
      } else {
H
Hongze Cheng 已提交
905 906 907
        offset = pSubBlock->offset + pSubBlock->szBlockCol + sizeof(TSCKSUM) + pSubBlock->szVersion +
                 pSubBlock->szTSKEY + sizeof(TSCKSUM) + pBlockCol->offset;
        size = pBlockCol->szBitmap + pBlockCol->szOffset + pBlockCol->szValue + sizeof(TSCKSUM);
H
Hongze Cheng 已提交
908

H
Hongze Cheng 已提交
909
        code = tsdbRealloc(ppBuf1, size);
H
Hongze Cheng 已提交
910 911
        if (code) goto _err;

H
Hongze Cheng 已提交
912 913 914 915 916 917
        // seek
        n = taosLSeekFile(pFD, offset, SEEK_SET);
        if (n < 0) {
          code = TAOS_SYSTEM_ERROR(errno);
          goto _err;
        }
H
Hongze Cheng 已提交
918

H
Hongze Cheng 已提交
919 920 921 922 923 924 925 926
        // read
        n = taosReadFile(pFD, *ppBuf1, size);
        if (n < 0) {
          code = TAOS_SYSTEM_ERROR(errno);
          goto _err;
        } else if (n < size) {
          code = TSDB_CODE_FILE_CORRUPTED;
          goto _err;
H
Hongze Cheng 已提交
927
        }
H
Hongze Cheng 已提交
928

H
Hongze Cheng 已提交
929
        code = tsdbReadColDataImpl(pSubBlock, pBlockCol, pColData, *ppBuf1, ppBuf2);
H
Hongze Cheng 已提交
930
        if (code) goto _err;
H
Hongze Cheng 已提交
931 932 933
      }
    }
  }
H
Hongze Cheng 已提交
934 935 936

_exit:
  taosArrayDestroy(aBlockCol);
H
Hongze Cheng 已提交
937 938 939
  return code;

_err:
H
Hongze Cheng 已提交
940
  taosArrayDestroy(aBlockCol);
H
Hongze Cheng 已提交
941 942 943 944 945 946 947 948 949
  return code;
}

int32_t tsdbReadColData(SDataFReader *pReader, SBlockIdx *pBlockIdx, SBlock *pBlock, int16_t *aColId, int32_t nCol,
                        SBlockData *pBlockData, uint8_t **ppBuf1, uint8_t **ppBuf2) {
  int32_t  code = 0;
  uint8_t *pBuf1 = NULL;
  uint8_t *pBuf2 = NULL;

950
  ASSERT(aColId[0] == PRIMARYKEY_TIMESTAMP_COL_ID);
H
Hongze Cheng 已提交
951 952 953 954

  if (!ppBuf1) ppBuf1 = &pBuf1;
  if (!ppBuf2) ppBuf2 = &pBuf2;

H
Hongze Cheng 已提交
955
  code = tsdbReadSubColData(pReader, pBlockIdx, pBlock, 0, aColId, nCol, pBlockData, ppBuf1, ppBuf2);
H
Hongze Cheng 已提交
956 957 958 959 960 961 962
  if (code) goto _err;

  if (pBlock->nSubBlock > 1) {
    SBlockData *pBlockData1 = &(SBlockData){0};
    SBlockData *pBlockData2 = &(SBlockData){0};

    for (int32_t iSubBlock = 1; iSubBlock < pBlock->nSubBlock; iSubBlock++) {
H
Hongze Cheng 已提交
963
      code = tsdbReadSubColData(pReader, pBlockIdx, pBlock, iSubBlock, aColId, nCol, pBlockData1, ppBuf1, ppBuf2);
H
Hongze Cheng 已提交
964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983
      if (code) goto _err;

      code = tBlockDataCopy(pBlockData, pBlockData2);
      if (code) {
        tBlockDataClear(pBlockData1);
        tBlockDataClear(pBlockData2);
        goto _err;
      }

      code = tBlockDataMerge(pBlockData1, pBlockData2, pBlockData);
      if (code) {
        tBlockDataClear(pBlockData1);
        tBlockDataClear(pBlockData2);
        goto _err;
      }
    }

    tBlockDataClear(pBlockData1);
    tBlockDataClear(pBlockData2);
  }
H
Hongze Cheng 已提交
984 985 986 987 988 989 990 991 992

  tsdbFree(pBuf1);
  tsdbFree(pBuf2);
  return code;

_err:
  tsdbError("vgId:%d tsdb read col data failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
  tsdbFree(pBuf1);
  tsdbFree(pBuf2);
H
Hongze Cheng 已提交
993 994 995
  return code;
}

H
Hongze Cheng 已提交
996 997
static int32_t tsdbReadSubBlockData(SDataFReader *pReader, SBlockIdx *pBlockIdx, SBlock *pBlock, int32_t iSubBlock,
                                    SBlockData *pBlockData, uint8_t **ppBuf1, uint8_t **ppBuf2) {
H
Hongze Cheng 已提交
998
  int32_t    code = 0;
H
Hongze Cheng 已提交
999 1000 1001
  uint8_t   *p;
  int64_t    size;
  int64_t    n;
H
Hongze Cheng 已提交
1002
  TdFilePtr  pFD = pBlock->last ? pReader->pLastFD : pReader->pDataFD;
H
Hongze Cheng 已提交
1003
  SSubBlock *pSubBlock = &pBlock->aSubBlock[iSubBlock];
H
Hongze Cheng 已提交
1004
  SArray    *aBlockCol = NULL;
H
Hongze Cheng 已提交
1005 1006

  tBlockDataReset(pBlockData);
H
Hongze Cheng 已提交
1007

H
Hongze Cheng 已提交
1008
  // realloc
H
refact  
Hongze Cheng 已提交
1009
  code = tsdbRealloc(ppBuf1, pSubBlock->szBlock);
H
Hongze Cheng 已提交
1010
  if (code) goto _err;
H
Hongze Cheng 已提交
1011

H
Hongze Cheng 已提交
1012 1013 1014 1015 1016 1017
  // seek
  n = taosLSeekFile(pFD, pSubBlock->offset, SEEK_SET);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
H
Hongze Cheng 已提交
1018

H
Hongze Cheng 已提交
1019
  // read
H
refact  
Hongze Cheng 已提交
1020
  n = taosReadFile(pFD, *ppBuf1, pSubBlock->szBlock);
H
Hongze Cheng 已提交
1021 1022 1023
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
H
refact  
Hongze Cheng 已提交
1024
  } else if (n < pSubBlock->szBlock) {
H
Hongze Cheng 已提交
1025 1026 1027
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }
H
Hongze Cheng 已提交
1028

H
Hongze Cheng 已提交
1029 1030
  pBlockData->nRow = pSubBlock->nRow;

H
Hongze Cheng 已提交
1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043
  // TSDBKEY
  p = *ppBuf1 + pSubBlock->szBlockCol + sizeof(TSCKSUM);
  code = tsdbReadBlockDataKey(pBlockData, pSubBlock, p, ppBuf2);
  if (code) goto _err;

  // COLUMNS
  aBlockCol = taosArrayInit(0, sizeof(SBlockCol));
  if (aBlockCol == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }

  code = tsdbReadBlockCol(pSubBlock, *ppBuf1, aBlockCol);
H
Hongze Cheng 已提交
1044
  if (code) goto _err;
H
Hongze Cheng 已提交
1045

H
Hongze Cheng 已提交
1046 1047 1048
  for (int32_t iBlockCol = 0; iBlockCol < taosArrayGetSize(aBlockCol); iBlockCol++) {
    SColData  *pColData;
    SBlockCol *pBlockCol = (SBlockCol *)taosArrayGet(aBlockCol, iBlockCol);
H
Hongze Cheng 已提交
1049

H
Hongze Cheng 已提交
1050
    ASSERT(pBlockCol->flag && pBlockCol->flag != HAS_NONE);
H
Hongze Cheng 已提交
1051

H
Hongze Cheng 已提交
1052 1053
    code = tBlockDataAddColData(pBlockData, iBlockCol, &pColData);
    if (code) goto _err;
H
Hongze Cheng 已提交
1054

H
Hongze Cheng 已提交
1055
    tColDataInit(pColData, pBlockCol->cid, pBlockCol->type, pBlockCol->smaOn);
H
Hongze Cheng 已提交
1056 1057 1058 1059
    if (pBlockCol->flag == HAS_NULL) {
      for (int32_t iRow = 0; iRow < pSubBlock->nRow; iRow++) {
        code = tColDataAppendValue(pColData, &COL_VAL_NULL(pBlockCol->cid, pBlockCol->type));
        if (code) goto _err;
H
Hongze Cheng 已提交
1060
      }
H
Hongze Cheng 已提交
1061
    } else {
H
Hongze Cheng 已提交
1062 1063 1064
      p = *ppBuf1 + pSubBlock->szBlockCol + sizeof(TSCKSUM) + pSubBlock->szVersion + pSubBlock->szTSKEY +
          sizeof(TSCKSUM) + pBlockCol->offset;
      code = tsdbReadColDataImpl(pSubBlock, pBlockCol, pColData, p, ppBuf2);
H
Hongze Cheng 已提交
1065
      if (code) goto _err;
H
Hongze Cheng 已提交
1066
    }
H
Hongze Cheng 已提交
1067 1068
  }

H
Hongze Cheng 已提交
1069
  taosArrayDestroy(aBlockCol);
H
Hongze Cheng 已提交
1070 1071 1072 1073
  return code;

_err:
  tsdbError("vgId:%d tsdb read sub block data failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1074
  taosArrayDestroy(aBlockCol);
H
Hongze Cheng 已提交
1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094
  return code;
}

int32_t tsdbReadBlockData(SDataFReader *pReader, SBlockIdx *pBlockIdx, SBlock *pBlock, SBlockData *pBlockData,
                          uint8_t **ppBuf1, uint8_t **ppBuf2) {
  int32_t   code = 0;
  TdFilePtr pFD = pBlock->last ? pReader->pLastFD : pReader->pDataFD;
  uint8_t  *pBuf1 = NULL;
  uint8_t  *pBuf2 = NULL;
  int32_t   iSubBlock;

  if (!ppBuf1) ppBuf1 = &pBuf1;
  if (!ppBuf2) ppBuf2 = &pBuf2;

  // read the first sub-block
  iSubBlock = 0;
  code = tsdbReadSubBlockData(pReader, pBlockIdx, pBlock, iSubBlock, pBlockData, ppBuf1, ppBuf2);
  if (code) goto _err;

  // read remain block data and do merg
H
Hongze Cheng 已提交
1095 1096 1097 1098 1099
  if (pBlock->nSubBlock > 1) {
    SBlockData *pBlockData1 = &(SBlockData){0};
    SBlockData *pBlockData2 = &(SBlockData){0};

    for (iSubBlock = 1; iSubBlock < pBlock->nSubBlock; iSubBlock++) {
H
Hongze Cheng 已提交
1100
      code = tsdbReadSubBlockData(pReader, pBlockIdx, pBlock, iSubBlock, pBlockData1, ppBuf1, ppBuf2);
H
Hongze Cheng 已提交
1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124
      if (code) {
        tBlockDataClear(pBlockData1);
        tBlockDataClear(pBlockData2);
        goto _err;
      }

      code = tBlockDataCopy(pBlockData, pBlockData2);
      if (code) {
        tBlockDataClear(pBlockData1);
        tBlockDataClear(pBlockData2);
        goto _err;
      }

      // merge two block data
      code = tBlockDataMerge(pBlockData1, pBlockData2, pBlockData);
      if (code) {
        tBlockDataClear(pBlockData1);
        tBlockDataClear(pBlockData2);
        goto _err;
      }
    }

    tBlockDataClear(pBlockData1);
    tBlockDataClear(pBlockData2);
H
Hongze Cheng 已提交
1125 1126
  }

H
Hongze Cheng 已提交
1127 1128 1129 1130
  ASSERT(pBlock->nRow == pBlockData->nRow);
  ASSERT(tsdbKeyCmprFn(&pBlock->minKey, &TSDBROW_KEY(&tBlockDataFirstRow(pBlockData))) == 0);
  ASSERT(tsdbKeyCmprFn(&pBlock->maxKey, &TSDBROW_KEY(&tBlockDataLastRow(pBlockData))) == 0);

H
Hongze Cheng 已提交
1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141
  if (pBuf1) tsdbFree(pBuf1);
  if (pBuf2) tsdbFree(pBuf2);
  return code;

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

H
Hongze Cheng 已提交
1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190
int32_t tsdbReadBlockSma(SDataFReader *pReader, SBlock *pBlock, SArray *aColumnDataAgg, uint8_t **ppBuf) {
  int32_t   code = 0;
  TdFilePtr pFD = pReader->pSmaFD;
  int64_t   offset = pBlock->aSubBlock[0].offset;
  int64_t   size = pBlock->aSubBlock[0].nSma * sizeof(SColumnDataAgg) + sizeof(TSCKSUM);
  uint8_t  *pBuf = NULL;
  int64_t   n;

  ASSERT(tBlockHasSma(pBlock));

  if (!ppBuf) ppBuf = &pBuf;
  code = tsdbRealloc(ppBuf, size);
  if (code) goto _err;

  // lseek
  n = taosLSeekFile(pFD, offset, SEEK_SET);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  // read
  n = taosReadFile(pFD, *ppBuf, size);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  // check
  if (!taosCheckChecksumWhole(NULL, size)) {
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

  // decode
  taosArrayClear(aColumnDataAgg);
  for (int32_t iSma = 0; iSma < pBlock->aSubBlock[0].nSma; iSma++) {
    if (taosArrayPush(aColumnDataAgg, &((SColumnDataAgg *)(*ppBuf))[iSma]) == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
    }
  }

  tsdbFree(pBuf);
  return code;

_err:
  tsdbError("vgId:%d read block sma failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
  tsdbFree(pBuf);
H
Hongze Cheng 已提交
1191 1192 1193 1194 1195
  return code;
}

// SDataFWriter ====================================================
struct SDataFWriter {
H
Hongze Cheng 已提交
1196 1197 1198 1199 1200 1201
  STsdb    *pTsdb;
  SDFileSet wSet;
  TdFilePtr pHeadFD;
  TdFilePtr pDataFD;
  TdFilePtr pLastFD;
  TdFilePtr pSmaFD;
H
Hongze Cheng 已提交
1202 1203
};

H
Hongze Cheng 已提交
1204 1205
SDFileSet *tsdbDataFWriterGetWSet(SDataFWriter *pWriter) { return &pWriter->wSet; }

H
Hongze Cheng 已提交
1206
int32_t tsdbDataFWriterOpen(SDataFWriter **ppWriter, STsdb *pTsdb, SDFileSet *pSet) {
H
Hongze Cheng 已提交
1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220
  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;
  }
  pWriter->pTsdb = pTsdb;
H
Hongze Cheng 已提交
1221 1222
  pWriter->wSet = *pSet;
  pSet = &pWriter->wSet;
H
Hongze Cheng 已提交
1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338

  // head
  flag = TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
  tsdbDataFileName(pTsdb, pSet, TSDB_HEAD_FILE, fname);
  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);

  pSet->fHead.size += TSDB_FHDR_SIZE;

  // data
  if (pSet->fData.size == 0) {
    flag = TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
  } else {
    flag = TD_FILE_WRITE;
  }
  tsdbDataFileName(pTsdb, pSet, TSDB_DATA_FILE, fname);
  pWriter->pDataFD = taosOpenFile(fname, flag);
  if (pWriter->pDataFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
  if (pSet->fData.size == 0) {
    n = taosWriteFile(pWriter->pDataFD, hdr, TSDB_FHDR_SIZE);
    if (n < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

    pSet->fData.size += TSDB_FHDR_SIZE;
  } else {
    n = taosLSeekFile(pWriter->pDataFD, 0, SEEK_END);
    if (n < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

    ASSERT(n == pSet->fData.size);
  }

  // last
  if (pSet->fLast.size == 0) {
    flag = TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
  } else {
    flag = TD_FILE_WRITE;
  }
  tsdbDataFileName(pTsdb, pSet, TSDB_LAST_FILE, fname);
  pWriter->pLastFD = taosOpenFile(fname, flag);
  if (pWriter->pLastFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
  if (pSet->fLast.size == 0) {
    n = taosWriteFile(pWriter->pLastFD, hdr, TSDB_FHDR_SIZE);
    if (n < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

    pSet->fLast.size += TSDB_FHDR_SIZE;
  } else {
    n = taosLSeekFile(pWriter->pLastFD, 0, SEEK_END);
    if (n < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

    ASSERT(n == pSet->fLast.size);
  }

  // sma
  if (pSet->fSma.size == 0) {
    flag = TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
  } else {
    flag = TD_FILE_WRITE;
  }
  tsdbDataFileName(pTsdb, pSet, TSDB_SMA_FILE, fname);
  pWriter->pSmaFD = taosOpenFile(fname, flag);
  if (pWriter->pSmaFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
  if (pSet->fSma.size == 0) {
    n = taosWriteFile(pWriter->pSmaFD, hdr, TSDB_FHDR_SIZE);
    if (n < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

    pSet->fSma.size += TSDB_FHDR_SIZE;
  } else {
    n = taosLSeekFile(pWriter->pSmaFD, 0, SEEK_END);
    if (n < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

    ASSERT(n == pSet->fSma.size);
  }

  *ppWriter = pWriter;
  return code;

_err:
  tsdbError("vgId:%d tsdb data file writer open failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
  *ppWriter = NULL;
H
Hongze Cheng 已提交
1339 1340 1341
  return code;
}

H
Hongze Cheng 已提交
1342
int32_t tsdbDataFWriterClose(SDataFWriter **ppWriter, int8_t sync) {
H
Hongze Cheng 已提交
1343
  int32_t code = 0;
H
Hongze Cheng 已提交
1344
  STsdb  *pTsdb = (*ppWriter)->pTsdb;
H
Hongze Cheng 已提交
1345

H
Hongze Cheng 已提交
1346 1347
  if (*ppWriter == NULL) goto _exit;

H
Hongze Cheng 已提交
1348
  if (sync) {
H
Hongze Cheng 已提交
1349
    if (taosFsyncFile((*ppWriter)->pHeadFD) < 0) {
H
Hongze Cheng 已提交
1350 1351 1352 1353
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

H
Hongze Cheng 已提交
1354
    if (taosFsyncFile((*ppWriter)->pDataFD) < 0) {
H
Hongze Cheng 已提交
1355 1356 1357 1358
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

H
Hongze Cheng 已提交
1359
    if (taosFsyncFile((*ppWriter)->pLastFD) < 0) {
H
Hongze Cheng 已提交
1360 1361 1362 1363
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

H
Hongze Cheng 已提交
1364
    if (taosFsyncFile((*ppWriter)->pSmaFD) < 0) {
H
Hongze Cheng 已提交
1365 1366 1367 1368 1369
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }
  }

H
Hongze Cheng 已提交
1370
  if (taosCloseFile(&(*ppWriter)->pHeadFD) < 0) {
H
Hongze Cheng 已提交
1371 1372 1373 1374
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
1375
  if (taosCloseFile(&(*ppWriter)->pDataFD) < 0) {
H
Hongze Cheng 已提交
1376 1377 1378 1379
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
1380
  if (taosCloseFile(&(*ppWriter)->pLastFD) < 0) {
H
Hongze Cheng 已提交
1381 1382 1383 1384
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
1385
  if (taosCloseFile(&(*ppWriter)->pSmaFD) < 0) {
H
Hongze Cheng 已提交
1386 1387 1388 1389
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
1390
  taosMemoryFree(*ppWriter);
H
Hongze Cheng 已提交
1391
_exit:
H
Hongze Cheng 已提交
1392
  *ppWriter = NULL;
H
Hongze Cheng 已提交
1393 1394 1395
  return code;

_err:
H
Hongze Cheng 已提交
1396
  tsdbError("vgId:%d data file writer close failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1397 1398 1399
  return code;
}

H
Hongze Cheng 已提交
1400
int32_t tsdbUpdateDFileSetHeader(SDataFWriter *pWriter) {
H
Hongze Cheng 已提交
1401
  int32_t code = 0;
H
Hongze Cheng 已提交
1402 1403

  // head ==============
H
Hongze Cheng 已提交
1404 1405
  code = tsdbUpdateDFileHdr(pWriter->pHeadFD, &pWriter->wSet, TSDB_HEAD_FILE);
  if (code) goto _err;
H
Hongze Cheng 已提交
1406 1407

  // data ==============
H
Hongze Cheng 已提交
1408 1409
  code = tsdbUpdateDFileHdr(pWriter->pHeadFD, &pWriter->wSet, TSDB_DATA_FILE);
  if (code) goto _err;
H
Hongze Cheng 已提交
1410 1411

  // last ==============
H
Hongze Cheng 已提交
1412 1413
  code = tsdbUpdateDFileHdr(pWriter->pHeadFD, &pWriter->wSet, TSDB_LAST_FILE);
  if (code) goto _err;
H
Hongze Cheng 已提交
1414 1415

  // sma ==============
H
Hongze Cheng 已提交
1416 1417
  code = tsdbUpdateDFileHdr(pWriter->pHeadFD, &pWriter->wSet, TSDB_SMA_FILE);
  if (code) goto _err;
H
Hongze Cheng 已提交
1418 1419 1420 1421 1422

  return code;

_err:
  tsdbError("vgId:%d update DFileSet header failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1423 1424 1425
  return code;
}

H
Hongze Cheng 已提交
1426
int32_t tsdbWriteBlockIdx(SDataFWriter *pWriter, SArray *aBlockIdx, uint8_t **ppBuf) {
H
Hongze Cheng 已提交
1427
  int32_t    code = 0;
H
Hongze Cheng 已提交
1428
  SHeadFile *pHeadFile = &pWriter->wSet.fHead;
H
Hongze Cheng 已提交
1429
  uint8_t   *pBuf = NULL;
H
Hongze Cheng 已提交
1430 1431 1432 1433
  int64_t    size;
  int64_t    n;

  if (!ppBuf) ppBuf = &pBuf;
H
Hongze Cheng 已提交
1434 1435

  // prepare
H
Hongze Cheng 已提交
1436 1437 1438 1439 1440
  size = tPutU32(NULL, TSDB_FILE_DLMT);
  for (int32_t iBlockIdx = 0; iBlockIdx < taosArrayGetSize(aBlockIdx); iBlockIdx++) {
    size += tPutBlockIdx(NULL, taosArrayGet(aBlockIdx, iBlockIdx));
  }
  size += sizeof(TSCKSUM);
H
Hongze Cheng 已提交
1441 1442 1443 1444 1445 1446

  // alloc
  code = tsdbRealloc(ppBuf, size);
  if (code) goto _err;

  // build
H
Hongze Cheng 已提交
1447
  n = 0;
H
Hongze Cheng 已提交
1448 1449 1450 1451
  n = tPutU32(*ppBuf + n, TSDB_FILE_DLMT);
  for (int32_t iBlockIdx = 0; iBlockIdx < taosArrayGetSize(aBlockIdx); iBlockIdx++) {
    n += tPutBlockIdx(*ppBuf + n, taosArrayGet(aBlockIdx, iBlockIdx));
  }
H
Hongze Cheng 已提交
1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462
  taosCalcChecksumAppend(0, *ppBuf, size);

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

  // write
  n = taosWriteFile(pWriter->pHeadFD, *ppBuf, size);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
1463 1464 1465
  // update
  pHeadFile->offset = pHeadFile->size;
  pHeadFile->size += size;
H
Hongze Cheng 已提交
1466 1467 1468 1469 1470 1471 1472

  tsdbFree(pBuf);
  return code;

_err:
  tsdbError("vgId:%d write block idx failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
  tsdbFree(pBuf);
H
Hongze Cheng 已提交
1473 1474 1475
  return code;
}

H
Hongze Cheng 已提交
1476
int32_t tsdbWriteBlock(SDataFWriter *pWriter, SMapData *mBlock, uint8_t **ppBuf, SBlockIdx *pBlockIdx) {
H
Hongze Cheng 已提交
1477 1478 1479 1480 1481 1482
  int32_t       code = 0;
  SHeadFile    *pHeadFile = &pWriter->wSet.fHead;
  SBlockDataHdr hdr = {.delimiter = TSDB_FILE_DLMT, .suid = pBlockIdx->suid, .uid = pBlockIdx->uid};
  uint8_t      *pBuf = NULL;
  int64_t       size;
  int64_t       n;
H
Hongze Cheng 已提交
1483

H
Hongze Cheng 已提交
1484
  ASSERT(mBlock->nItem > 0);
H
Hongze Cheng 已提交
1485 1486

  // prepare
H
Hongze Cheng 已提交
1487
  size = sizeof(SBlockDataHdr) + tPutMapData(NULL, mBlock) + sizeof(TSCKSUM);
H
Hongze Cheng 已提交
1488 1489 1490 1491 1492 1493 1494 1495

  // alloc
  if (!ppBuf) ppBuf = &pBuf;
  code = tsdbRealloc(ppBuf, size);
  if (code) goto _err;

  // build
  n = 0;
H
Hongze Cheng 已提交
1496 1497
  *(SBlockDataHdr *)(*ppBuf) = hdr;
  n += sizeof(hdr);
H
Hongze Cheng 已提交
1498
  n += tPutMapData(*ppBuf + n, mBlock);
H
Hongze Cheng 已提交
1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509
  taosCalcChecksumAppend(0, *ppBuf, size);

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

  // write
  n = taosWriteFile(pWriter->pHeadFD, *ppBuf, size);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
1510 1511
  // update
  pBlockIdx->offset = pHeadFile->size;
H
Hongze Cheng 已提交
1512
  pBlockIdx->size = size;
H
Hongze Cheng 已提交
1513
  pHeadFile->size += size;
H
Hongze Cheng 已提交
1514

H
Hongze Cheng 已提交
1515
  tsdbFree(pBuf);
H
Hongze Cheng 已提交
1516 1517 1518 1519 1520
  tsdbTrace("vgId:%d write block, offset:%" PRId64 " size:%" PRId64, TD_VID(pWriter->pTsdb->pVnode), pBlockIdx->offset,
            pBlockIdx->size);
  return code;

_err:
H
Hongze Cheng 已提交
1521
  tsdbFree(pBuf);
H
Hongze Cheng 已提交
1522
  tsdbError("vgId:%d write block failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1523 1524 1525
  return code;
}

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

H
Hongze Cheng 已提交
1530
    if (iRow == 0) {
H
Hongze Cheng 已提交
1531 1532 1533 1534 1535 1536 1537
      if (tsdbKeyCmprFn(&pBlock->minKey, &key) > 0) {
        pBlock->minKey = key;
      }
    } else {
      if (pBlockData->aTSKEY[iRow] == pBlockData->aTSKEY[iRow - 1]) {
        pBlock->hasDup = 1;
      }
H
Hongze Cheng 已提交
1538 1539
    }

H
Hongze Cheng 已提交
1540 1541
    if (iRow == pBlockData->nRow - 1 && tsdbKeyCmprFn(&pBlock->maxKey, &key) < 0) {
      pBlock->maxKey = key;
H
Hongze Cheng 已提交
1542 1543 1544 1545 1546 1547
    }

    pBlock->minVersion = TMIN(pBlock->minVersion, key.version);
    pBlock->maxVersion = TMAX(pBlock->maxVersion, key.version);
  }
  pBlock->nRow += pBlockData->nRow;
H
Hongze Cheng 已提交
1548
}
H
Hongze Cheng 已提交
1549

H
Hongze Cheng 已提交
1550 1551 1552 1553 1554 1555 1556 1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567
static int32_t tsdbWriteBlockDataKey(SSubBlock *pSubBlock, SBlockData *pBlockData, uint8_t **ppBuf1, int64_t *nDataP,
                                     uint8_t **ppBuf2) {
  int32_t code = 0;
  int64_t size;
  int64_t tsize;

  if (pSubBlock->cmprAlg == NO_COMPRESSION) {
    pSubBlock->szVersion = sizeof(int64_t) * pSubBlock->nRow;
    pSubBlock->szTSKEY = sizeof(TSKEY) * pSubBlock->nRow;

    code = tsdbRealloc(ppBuf1, *nDataP + pSubBlock->szVersion + pSubBlock->szTSKEY + sizeof(TSCKSUM));
    if (code) goto _err;

    // VERSION
    memcpy(*ppBuf1 + *nDataP, pBlockData->aVersion, pSubBlock->szVersion);

    // TSKEY
    memcpy(*ppBuf1 + *nDataP + pSubBlock->szVersion, pBlockData->aTSKEY, pSubBlock->szTSKEY);
H
Hongze Cheng 已提交
1568
  } else {
H
Hongze Cheng 已提交
1569
    size = (sizeof(int64_t) + sizeof(TSKEY)) * pSubBlock->nRow + COMP_OVERFLOW_BYTES * 2;
H
Hongze Cheng 已提交
1570

H
Hongze Cheng 已提交
1571 1572
    code = tsdbRealloc(ppBuf1, *nDataP + size + sizeof(TSCKSUM));
    if (code) goto _err;
H
Hongze Cheng 已提交
1573

H
Hongze Cheng 已提交
1574 1575 1576 1577 1578
    tsize = sizeof(int64_t) * pSubBlock->nRow + COMP_OVERFLOW_BYTES;
    if (pSubBlock->cmprAlg == TWO_STAGE_COMP) {
      code = tsdbRealloc(ppBuf2, tsize);
      if (code) goto _err;
    }
H
Hongze Cheng 已提交
1579

H
Hongze Cheng 已提交
1580 1581 1582 1583 1584 1585
    // VERSION
    pSubBlock->szVersion =
        tsCompressBigint((char *)pBlockData->aVersion, sizeof(int64_t) * pBlockData->nRow, pBlockData->nRow,
                         *ppBuf1 + *nDataP, size, pSubBlock->cmprAlg, *ppBuf2, tsize);
    if (pSubBlock->szVersion <= 0) {
      code = TSDB_CODE_COMPRESS_ERROR;
H
Hongze Cheng 已提交
1586 1587 1588
      goto _err;
    }

H
Hongze Cheng 已提交
1589
    // TSKEY
H
Hongze Cheng 已提交
1590 1591 1592 1593 1594
    pSubBlock->szTSKEY = tsCompressTimestamp((char *)pBlockData->aTSKEY, sizeof(TSKEY) * pBlockData->nRow,
                                             pBlockData->nRow, *ppBuf1 + *nDataP + pSubBlock->szVersion,
                                             size - pSubBlock->szVersion, pSubBlock->cmprAlg, *ppBuf2, tsize);
    if (pSubBlock->szTSKEY <= 0) {
      code = TSDB_CODE_COMPRESS_ERROR;
H
Hongze Cheng 已提交
1595 1596 1597
      goto _err;
    }

H
Hongze Cheng 已提交
1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632
    ASSERT(pSubBlock->szVersion + pSubBlock->szTSKEY <= size);
  }

  // checksum
  size = pSubBlock->szVersion + pSubBlock->szTSKEY + sizeof(TSCKSUM);
  taosCalcChecksumAppend(0, *ppBuf1 + *nDataP, size);

  *nDataP += size;
  return code;

_err:
  return code;
}

static int32_t tsdbWriteColData(SColData *pColData, SBlockCol *pBlockCol, SSubBlock *pSubBlock, uint8_t **ppBuf1,
                                int64_t *nDataP, uint8_t **ppBuf2) {
  int32_t code = 0;
  int64_t size;
  int64_t n = 0;

  // BITMAP
  if (pColData->flag != HAS_VALUE) {
    size = BIT2_SIZE(pColData->nVal) + COMP_OVERFLOW_BYTES;

    code = tsdbRealloc(ppBuf1, *nDataP + n + size);
    if (code) goto _err;

    code = tsdbRealloc(ppBuf2, size);
    if (code) goto _err;

    pBlockCol->szBitmap =
        tsCompressTinyint((char *)pColData->pBitMap, BIT2_SIZE(pColData->nVal), BIT2_SIZE(pColData->nVal),
                          *ppBuf1 + *nDataP + n, size, TWO_STAGE_COMP, *ppBuf2, size);
    if (pBlockCol->szBitmap <= 0) {
      code = TSDB_CODE_COMPRESS_ERROR;
H
Hongze Cheng 已提交
1633 1634 1635
      goto _err;
    }
  } else {
H
Hongze Cheng 已提交
1636 1637 1638
    pBlockCol->szBitmap = 0;
  }
  n += pBlockCol->szBitmap;
H
Hongze Cheng 已提交
1639

H
Hongze Cheng 已提交
1640 1641 1642
  // OFFSET
  if (IS_VAR_DATA_TYPE(pColData->type)) {
    size = sizeof(int32_t) * pColData->nVal + COMP_OVERFLOW_BYTES;
H
Hongze Cheng 已提交
1643

H
Hongze Cheng 已提交
1644
    code = tsdbRealloc(ppBuf1, *nDataP + n + size);
H
Hongze Cheng 已提交
1645 1646
    if (code) goto _err;

H
Hongze Cheng 已提交
1647 1648
    code = tsdbRealloc(ppBuf2, size);
    if (code) goto _err;
H
Hongze Cheng 已提交
1649

H
Hongze Cheng 已提交
1650 1651 1652
    pBlockCol->szOffset = tsCompressInt((char *)pColData->aOffset, sizeof(int32_t) * pColData->nVal, pColData->nVal,
                                        *ppBuf1 + *nDataP + n, size, TWO_STAGE_COMP, *ppBuf2, size);
    if (pBlockCol->szOffset <= 0) {
H
Hongze Cheng 已提交
1653 1654 1655
      code = TSDB_CODE_COMPRESS_ERROR;
      goto _err;
    }
H
Hongze Cheng 已提交
1656 1657 1658 1659
  } else {
    pBlockCol->szOffset = 0;
  }
  n += pBlockCol->szOffset;
H
Hongze Cheng 已提交
1660

H
Hongze Cheng 已提交
1661 1662 1663 1664 1665 1666 1667 1668 1669 1670 1671 1672 1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683
  // VALUE
  if (pSubBlock->cmprAlg == NO_COMPRESSION) {
    pBlockCol->szValue = pColData->nData;

    code = tsdbRealloc(ppBuf1, *nDataP + n + pBlockCol->szValue + sizeof(TSCKSUM));
    if (code) goto _err;

    memcpy(*ppBuf1 + *nDataP + n, pColData->pData, pBlockCol->szValue);
  } else {
    size = pColData->nData + COMP_OVERFLOW_BYTES;

    code = tsdbRealloc(ppBuf1, *nDataP + n + size + sizeof(TSCKSUM));
    if (code) goto _err;

    if (pSubBlock->cmprAlg == TWO_STAGE_COMP) {
      code = tsdbRealloc(ppBuf2, size);
      if (code) goto _err;
    }

    pBlockCol->szValue =
        tDataTypes[pColData->type].compFunc((char *)pColData->pData, pColData->nData, pColData->nVal,
                                            *ppBuf1 + *nDataP + n, size, pSubBlock->cmprAlg, *ppBuf2, size);
    if (pBlockCol->szValue <= 0) {
H
Hongze Cheng 已提交
1684 1685 1686
      code = TSDB_CODE_COMPRESS_ERROR;
      goto _err;
    }
H
Hongze Cheng 已提交
1687 1688 1689
  }
  n += pBlockCol->szValue;
  pBlockCol->szOrigin = pColData->nData;
H
Hongze Cheng 已提交
1690

H
Hongze Cheng 已提交
1691 1692 1693
  // checksum
  n += sizeof(TSCKSUM);
  taosCalcChecksumAppend(0, *ppBuf1 + *nDataP, n);
H
Hongze Cheng 已提交
1694

H
Hongze Cheng 已提交
1695 1696 1697 1698 1699 1700 1701 1702 1703 1704 1705 1706 1707 1708 1709 1710 1711 1712 1713 1714 1715
  *nDataP += n;

  return code;

_err:
  return code;
}

static int32_t tsdbWriteBlockDataImpl(TdFilePtr pFD, SSubBlock *pSubBlock, SBlockDataHdr hdr, SArray *aBlockCol,
                                      uint8_t *pData, int64_t nData, uint8_t **ppBuf) {
  int32_t code = 0;
  int32_t nBlockCol = taosArrayGetSize(aBlockCol);
  int64_t size;
  int64_t n;

  // HDR + SArray<SBlockCol>
  pSubBlock->szBlockCol = sizeof(hdr);
  for (int32_t iBlockCol = 0; iBlockCol < nBlockCol; iBlockCol++) {
    pSubBlock->szBlockCol += tPutBlockCol(NULL, taosArrayGet(aBlockCol, iBlockCol));
  }

H
Hongze Cheng 已提交
1716
  code = tsdbRealloc(ppBuf, pSubBlock->szBlockCol + sizeof(TSCKSUM));
H
Hongze Cheng 已提交
1717 1718 1719 1720 1721 1722 1723 1724 1725 1726 1727 1728 1729 1730 1731 1732 1733 1734 1735 1736 1737 1738 1739 1740 1741 1742 1743 1744 1745 1746 1747
  if (code) goto _err;

  n = 0;
  memcpy(*ppBuf, &hdr, sizeof(hdr));
  n += sizeof(hdr);
  for (int32_t iBlockCol = 0; iBlockCol < nBlockCol; iBlockCol++) {
    n += tPutBlockCol(*ppBuf + n, taosArrayGet(aBlockCol, iBlockCol));
  }
  taosCalcChecksumAppend(0, *ppBuf, pSubBlock->szBlockCol + sizeof(TSCKSUM));

  ASSERT(n == pSubBlock->szBlockCol);

  n = taosWriteFile(pFD, *ppBuf, pSubBlock->szBlockCol + sizeof(TSCKSUM));
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  // SBlockData
  n = taosWriteFile(pFD, pData, nData);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  return code;

_err:
  return code;
}

H
Hongze Cheng 已提交
1748
static int32_t tsdbWriteBlockSma(TdFilePtr pFD, SBlockData *pBlockData, SSubBlock *pSubBlock, uint8_t **ppBuf) {
H
Hongze Cheng 已提交
1749 1750 1751
  int32_t   code = 0;
  int64_t   n;
  SColData *pColData;
H
Hongze Cheng 已提交
1752

H
Hongze Cheng 已提交
1753 1754 1755 1756
  // prepare
  pSubBlock->nSma = 0;
  for (int32_t iColData = 0; iColData < taosArrayGetSize(pBlockData->aColDataP); iColData++) {
    pColData = (SColData *)taosArrayGetP(pBlockData->aColDataP, iColData);
H
Hongze Cheng 已提交
1757

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

H
Hongze Cheng 已提交
1760 1761 1762
    pSubBlock->nSma++;
  }
  if (pSubBlock->nSma == 0) goto _exit;
H
Hongze Cheng 已提交
1763

H
Hongze Cheng 已提交
1764 1765 1766 1767 1768 1769
  // calc
  code = tsdbRealloc(ppBuf, sizeof(SColumnDataAgg) * pSubBlock->nSma + sizeof(TSCKSUM));
  if (code) goto _err;
  n = 0;
  for (int32_t iColData = 0; iColData < taosArrayGetSize(pBlockData->aColDataP); iColData++) {
    pColData = (SColData *)taosArrayGetP(pBlockData->aColDataP, iColData);
H
Hongze Cheng 已提交
1770

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

H
Hongze Cheng 已提交
1773 1774 1775 1776
    tsdbCalcColDataSMA(pColData, &((SColumnDataAgg *)(*ppBuf))[n]);
    n++;
  }
  taosCalcChecksumAppend(0, *ppBuf, sizeof(SColumnDataAgg) * pSubBlock->nSma + sizeof(TSCKSUM));
H
Hongze Cheng 已提交
1777

H
Hongze Cheng 已提交
1778 1779 1780 1781 1782 1783
  // write
  n = taosWriteFile(pFD, *ppBuf, sizeof(SColumnDataAgg) * pSubBlock->nSma + sizeof(TSCKSUM));
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
H
Hongze Cheng 已提交
1784

H
Hongze Cheng 已提交
1785 1786
_exit:
  return code;
H
Hongze Cheng 已提交
1787

H
Hongze Cheng 已提交
1788 1789 1790
_err:
  return code;
}
H
Hongze Cheng 已提交
1791

H
Hongze Cheng 已提交
1792 1793 1794 1795 1796 1797 1798 1799 1800 1801 1802 1803 1804 1805
int32_t tsdbWriteBlockData(SDataFWriter *pWriter, SBlockData *pBlockData, uint8_t **ppBuf1, uint8_t **ppBuf2,
                           SBlockIdx *pBlockIdx, SBlock *pBlock, int8_t cmprAlg) {
  int32_t       code = 0;
  SSubBlock    *pSubBlock = &pBlock->aSubBlock[pBlock->nSubBlock++];
  SBlockCol     blockCol;
  SBlockCol    *pBlockCol = &blockCol;
  int64_t       n;
  TdFilePtr     pFileFD = pBlock->last ? pWriter->pLastFD : pWriter->pDataFD;
  SBlockDataHdr hdr = {.delimiter = TSDB_FILE_DLMT, .suid = pBlockIdx->suid, .uid = pBlockIdx->uid};
  uint8_t      *p;
  int64_t       nData;
  uint8_t      *pBuf1 = NULL;
  uint8_t      *pBuf2 = NULL;
  SArray       *aBlockCol = NULL;
H
Hongze Cheng 已提交
1806

H
Hongze Cheng 已提交
1807 1808
  if (!ppBuf1) ppBuf1 = &pBuf1;
  if (!ppBuf2) ppBuf2 = &pBuf2;
H
Hongze Cheng 已提交
1809

H
Hongze Cheng 已提交
1810
  tsdbUpdateBlockInfo(pBlockData, pBlock);
H
Hongze Cheng 已提交
1811

H
Hongze Cheng 已提交
1812 1813 1814 1815 1816 1817 1818 1819 1820 1821 1822 1823 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837 1838 1839 1840 1841
  pSubBlock->nRow = pBlockData->nRow;
  pSubBlock->cmprAlg = cmprAlg;
  if (pBlock->last) {
    pSubBlock->offset = pWriter->wSet.fLast.size;
  } else {
    pSubBlock->offset = pWriter->wSet.fData.size;
  }

  // ======================= BLOCK DATA =======================
  // TSDBKEY
  nData = 0;
  code = tsdbWriteBlockDataKey(pSubBlock, pBlockData, ppBuf1, &nData, ppBuf2);
  if (code) goto _err;

  // COLUMNS
  aBlockCol = taosArrayInit(taosArrayGetSize(pBlockData->aColDataP), sizeof(SBlockCol));
  if (aBlockCol == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
  int32_t offset = 0;
  for (int32_t iCol = 0; iCol < taosArrayGetSize(pBlockData->aColDataP); iCol++) {
    SColData *pColData = (SColData *)taosArrayGetP(pBlockData->aColDataP, iCol);

    ASSERT(pColData->flag);

    if (pColData->flag == HAS_NONE) continue;

    pBlockCol->cid = pColData->cid;
    pBlockCol->type = pColData->type;
H
Hongze Cheng 已提交
1842
    pBlockCol->smaOn = pColData->smaOn;
H
Hongze Cheng 已提交
1843 1844 1845 1846 1847 1848 1849 1850
    pBlockCol->flag = pColData->flag;

    if (pColData->flag != HAS_NULL) {
      code = tsdbWriteColData(pColData, pBlockCol, pSubBlock, ppBuf1, &nData, ppBuf2);
      if (code) goto _err;

      pBlockCol->offset = offset;
      offset = offset + pBlockCol->szBitmap + pBlockCol->szOffset + pBlockCol->szValue + sizeof(TSCKSUM);
H
Hongze Cheng 已提交
1851 1852
    }

H
Hongze Cheng 已提交
1853 1854 1855 1856
    if (taosArrayPush(aBlockCol, pBlockCol) == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
    }
H
Hongze Cheng 已提交
1857 1858
  }

H
Hongze Cheng 已提交
1859 1860 1861 1862 1863
  // write
  code = tsdbWriteBlockDataImpl(pFileFD, pSubBlock, hdr, aBlockCol, *ppBuf1, nData, ppBuf2);
  if (code) goto _err;

  pSubBlock->szBlock = pSubBlock->szBlockCol + sizeof(TSCKSUM) + nData;
H
Hongze Cheng 已提交
1864
  if (pBlock->last) {
H
refact  
Hongze Cheng 已提交
1865
    pWriter->wSet.fLast.size += pSubBlock->szBlock;
H
Hongze Cheng 已提交
1866
  } else {
H
refact  
Hongze Cheng 已提交
1867
    pWriter->wSet.fData.size += pSubBlock->szBlock;
H
Hongze Cheng 已提交
1868 1869
  }

H
Hongze Cheng 已提交
1870 1871 1872 1873 1874 1875
  // ======================= BLOCK SMA =======================
  pSubBlock->sOffset = 0;
  pSubBlock->nSma = 0;

  if (pBlock->nSubBlock > 1 || pBlock->last || pBlock->hasDup) goto _exit;

H
Hongze Cheng 已提交
1876
  code = tsdbWriteBlockSma(pWriter->pSmaFD, pBlockData, pSubBlock, ppBuf1);
H
Hongze Cheng 已提交
1877 1878 1879 1880 1881 1882 1883 1884
  if (code) goto _err;

  if (pSubBlock->nSma > 0) {
    pSubBlock->sOffset = pWriter->wSet.fSma.size;
    pWriter->wSet.fSma.size += (sizeof(SColumnDataAgg) * pSubBlock->nSma + sizeof(TSCKSUM));
  }

_exit:
H
Hongze Cheng 已提交
1885 1886
  tsdbFree(pBuf1);
  tsdbFree(pBuf2);
H
Hongze Cheng 已提交
1887
  taosArrayDestroy(aBlockCol);
H
Hongze Cheng 已提交
1888 1889 1890
  return code;

_err:
H
Hongze Cheng 已提交
1891
  tsdbError("vgId:%d write block data failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1892 1893
  tsdbFree(pBuf1);
  tsdbFree(pBuf2);
H
Hongze Cheng 已提交
1894
  taosArrayDestroy(aBlockCol);
H
Hongze Cheng 已提交
1895 1896
  return code;
}