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

H
Hongze Cheng 已提交
21 22
// SDelFWriter ====================================================
struct SDelFWriter {
H
Hongze Cheng 已提交
23
  STsdb    *pTsdb;
H
Hongze Cheng 已提交
24 25
  SDelFile *pFile;
  TdFilePtr pWriteH;
H
refact  
Hongze Cheng 已提交
26 27
};

H
Hongze Cheng 已提交
28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51
int32_t tsdbDelFWriterOpen(SDelFWriter **ppWriter, SDelFile *pFile, STsdb *pTsdb) {
  int32_t      code = 0;
  char        *fname = NULL;  // TODO
  SDelFWriter *pDelFWriter;

  pDelFWriter = (SDelFWriter *)taosMemoryCalloc(1, sizeof(*pDelFWriter));
  if (pDelFWriter == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }

  pDelFWriter->pTsdb = pTsdb;
  pDelFWriter->pFile = pFile;
  pDelFWriter->pWriteH = taosOpenFile(fname, TD_FILE_WRITE | TD_FILE_CREATE);
  if (pDelFWriter->pWriteH == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

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

H
more  
Hongze Cheng 已提交
52 53 54
  pDelFWriter->pFile->size = TSDB_FHDR_SIZE;
  pDelFWriter->pFile->offset = 0;

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 60 61
  return code;
}

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

  // 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 已提交
81 82 83
  return code;
}

H
Hongze Cheng 已提交
84
int32_t tsdbWriteDelData(SDelFWriter *pWriter, SMapData *pDelDataMap, uint8_t **ppBuf, SDelIdx *pDelIdx) {
H
Hongze Cheng 已提交
85 86
  int32_t  code = 0;
  uint8_t *pBuf = NULL;
H
Hongze Cheng 已提交
87 88
  int64_t  size = 0;
  int64_t  n = 0;
H
Hongze Cheng 已提交
89

H
Hongze Cheng 已提交
90
  // prepare
H
Hongze Cheng 已提交
91
  size += tPutU32(NULL, TSDB_FILE_DLMT);
H
Hongze Cheng 已提交
92 93
  size += tPutI64(NULL, pDelIdx->suid);
  size += tPutI64(NULL, pDelIdx->uid);
H
Hongze Cheng 已提交
94
  size = size + tPutMapData(NULL, pDelDataMap) + sizeof(TSCKSUM);
H
Hongze Cheng 已提交
95 96 97 98 99 100 101

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

  // build
H
Hongze Cheng 已提交
102
  n += tPutU32(*ppBuf + n, TSDB_FILE_DLMT);
H
Hongze Cheng 已提交
103 104
  n += tPutI64(*ppBuf + n, pDelIdx->suid);
  n += tPutI64(*ppBuf + n, pDelIdx->uid);
H
Hongze Cheng 已提交
105
  n += tPutMapData(*ppBuf + n, pDelDataMap);
H
Hongze Cheng 已提交
106 107 108 109 110 111 112 113 114 115 116 117 118 119
  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 已提交
120 121
  pDelIdx->offset = pWriter->pFile->size;
  pDelIdx->size = size;
H
Hongze Cheng 已提交
122 123 124 125 126 127 128 129 130
  pWriter->pFile->offset = pWriter->pFile->size;
  pWriter->pFile->size += size;

  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 已提交
131 132 133
  return code;
}

H
Hongze Cheng 已提交
134
int32_t tsdbWriteDelIdx(SDelFWriter *pWriter, SMapData *pDelIdxMap, uint8_t **ppBuf) {
H
Hongze Cheng 已提交
135
  int32_t  code = 0;
H
Hongze Cheng 已提交
136 137
  int64_t  size = 0;
  int64_t  n = 0;
H
Hongze Cheng 已提交
138
  uint8_t *pBuf = NULL;
H
Hongze Cheng 已提交
139

H
Hongze Cheng 已提交
140
  // prepare
H
Hongze Cheng 已提交
141 142
  size += tPutU32(NULL, TSDB_FILE_DLMT);
  size = size + tPutMapData(NULL, pDelIdxMap) + sizeof(TSCKSUM);
H
Hongze Cheng 已提交
143 144

  // alloc
H
Hongze Cheng 已提交
145
  if (!ppBuf) ppBuf = &pBuf;
H
Hongze Cheng 已提交
146
  code = tsdbRealloc(ppBuf, size);
H
Hongze Cheng 已提交
147
  if (code) goto _err;
H
Hongze Cheng 已提交
148

H
Hongze Cheng 已提交
149
  // build
H
Hongze Cheng 已提交
150 151
  n += tPutU32(*ppBuf + n, TSDB_FILE_DLMT);
  n += tPutMapData(*ppBuf + n, pDelIdxMap);
H
Hongze Cheng 已提交
152 153
  taosCalcChecksumAppend(0, *ppBuf, size);

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

H
Hongze Cheng 已提交
156
  // write
H
Hongze Cheng 已提交
157 158
  n = taosWriteFile(pWriter->pWriteH, *ppBuf, size);
  if (n < 0) {
H
Hongze Cheng 已提交
159 160 161 162
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
163 164 165
  ASSERT(n == size);

  // update
H
Hongze Cheng 已提交
166 167 168
  pWriter->pFile->offset = pWriter->pFile->size;
  pWriter->pFile->size += size;

H
Hongze Cheng 已提交
169 170 171 172
  tsdbFree(pBuf);
  return code;

_err:
H
more  
Hongze Cheng 已提交
173
  tsdbError("vgId:%d write del idx failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208
  tsdbFree(pBuf);
  return code;
}

int32_t tsdbUpdateDelFileHdr(SDelFWriter *pWriter, uint8_t **ppBuf) {
  int32_t  code = 0;
  uint8_t *pBuf = NULL;
  int64_t  size = TSDB_FHDR_SIZE;
  int64_t  n;

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

  // build
  memset(*ppBuf, 0, size);
  n = tPutDelFileHdr(*ppBuf, pWriter->pFile);
  taosCalcChecksumAppend(0, *ppBuf, size);

  ASSERT(n <= size - sizeof(TSCKSUM));

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

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

  tsdbFree(pBuf);
H
Hongze Cheng 已提交
209 210 211
  return code;

_err:
H
Hongze Cheng 已提交
212
  tsdbError("vgId:%d update del file hdr failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
213
  tsdbFree(pBuf);
H
Hongze Cheng 已提交
214 215 216 217 218
  return code;
}

// SDelFReader ====================================================
struct SDelFReader {
H
Hongze Cheng 已提交
219
  STsdb    *pTsdb;
H
Hongze Cheng 已提交
220 221
  SDelFile *pFile;
  TdFilePtr pReadH;
H
refact  
Hongze Cheng 已提交
222
};
H
Hongze Cheng 已提交
223

H
Hongze Cheng 已提交
224 225 226 227
int32_t tsdbDelFReaderOpen(SDelFReader **ppReader, SDelFile *pFile, STsdb *pTsdb, uint8_t **ppBuf) {
  int32_t      code = 0;
  char        *fname = NULL;  // todo
  SDelFReader *pDelFReader;
H
Hongze Cheng 已提交
228
  int64_t      n;
H
Hongze Cheng 已提交
229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253

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

  // open impl
  pDelFReader->pTsdb = pTsdb;
  pDelFReader->pFile = pFile;
  pDelFReader->pReadH = taosOpenFile(fname, TD_FILE_READ);
  if (pDelFReader == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    taosMemoryFree(pDelFReader);
    goto _err;
  }

  // load and check hdr if buffer is given
  if (ppBuf) {
    code = tsdbRealloc(ppBuf, TSDB_FHDR_SIZE);
    if (code) {
      goto _err;
    }

H
Hongze Cheng 已提交
254 255 256 257 258
    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 已提交
259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275
      code = TSDB_CODE_FILE_CORRUPTED;
      goto _err;
    }

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

    // TODO: check the content
  }

_exit:
  *ppReader = pDelFReader;
  return code;

_err:
H
Hongze Cheng 已提交
276
  tsdbError("vgId:%d del file reader open failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
277
  *ppReader = NULL;
H
Hongze Cheng 已提交
278 279 280 281 282
  return code;
}

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

  if (pReader) {
H
Hongze Cheng 已提交
285 286 287 288
    if (taosCloseFile(&pReader->pReadH) < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _exit;
    }
H
Hongze Cheng 已提交
289 290 291
    taosMemoryFree(pReader);
  }

H
Hongze Cheng 已提交
292
_exit:
H
Hongze Cheng 已提交
293 294 295
  return code;
}

H
Hongze Cheng 已提交
296
int32_t tsdbReadDelData(SDelFReader *pReader, SDelIdx *pDelIdx, SMapData *pDelDataMap, uint8_t **ppBuf) {
H
Hongze Cheng 已提交
297 298 299 300 301
  int32_t  code = 0;
  int64_t  n;
  uint32_t delimiter;
  tb_uid_t suid;
  tb_uid_t uid;
H
Hongze Cheng 已提交
302 303

  // seek
H
Hongze Cheng 已提交
304
  if (taosLSeekFile(pReader->pReadH, pDelIdx->offset, SEEK_SET) < 0) {
H
Hongze Cheng 已提交
305 306 307 308 309
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  // alloc
H
Hongze Cheng 已提交
310 311
  if (!ppBuf) ppBuf = &pDelDataMap->pBuf;
  code = tsdbRealloc(ppBuf, pDelIdx->size);
H
Hongze Cheng 已提交
312 313 314
  if (code) goto _err;

  // read
H
Hongze Cheng 已提交
315
  n = taosReadFile(pReader->pReadH, *ppBuf, pDelIdx->size);
H
Hongze Cheng 已提交
316 317 318
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
H
Hongze Cheng 已提交
319 320 321
  } else if (n < pDelIdx->size) {
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
H
Hongze Cheng 已提交
322 323 324
  }

  // check
H
Hongze Cheng 已提交
325
  if (!taosCheckChecksumWhole(*ppBuf, pDelIdx->size)) {
H
Hongze Cheng 已提交
326 327 328 329
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

H
Hongze Cheng 已提交
330 331 332 333
  // // decode
  n = 0;
  n += tGetU32(*ppBuf + n, &delimiter);
  ASSERT(delimiter == TSDB_FILE_DLMT);
H
Hongze Cheng 已提交
334 335 336 337
  n += tGetI64(*ppBuf + n, &suid);
  ASSERT(suid == pDelIdx->suid);
  n += tGetI64(*ppBuf + n, &uid);
  ASSERT(uid == pDelIdx->uid);
H
Hongze Cheng 已提交
338 339
  n += tGetMapData(*ppBuf + n, pDelDataMap);
  ASSERT(n + sizeof(TSCKSUM) == pDelIdx->size);
H
Hongze Cheng 已提交
340 341 342 343 344

  return code;

_err:
  tsdbError("vgId:%d read del data failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
345 346 347
  return code;
}

H
Hongze Cheng 已提交
348 349 350 351 352 353
int32_t tsdbReadDelIdx(SDelFReader *pReader, SMapData *pDelIdxMap, uint8_t **ppBuf) {
  int32_t  code = 0;
  int32_t  n;
  int64_t  offset = pReader->pFile->offset;
  int64_t  size = pReader->pFile->size - offset;
  uint32_t delimiter;
H
Hongze Cheng 已提交
354

H
more  
Hongze Cheng 已提交
355 356
  ASSERT(ppBuf && *ppBuf);

H
Hongze Cheng 已提交
357
  // seek
H
Hongze Cheng 已提交
358
  if (taosLSeekFile(pReader->pReadH, offset, SEEK_SET) < 0) {
H
Hongze Cheng 已提交
359 360 361 362
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
363
  // alloc
H
Hongze Cheng 已提交
364
  if (!ppBuf) ppBuf = &pDelIdxMap->pBuf;
H
Hongze Cheng 已提交
365
  code = tsdbRealloc(ppBuf, size);
H
Hongze Cheng 已提交
366
  if (code) goto _err;
H
Hongze Cheng 已提交
367

H
Hongze Cheng 已提交
368
  // read
H
Hongze Cheng 已提交
369 370
  n = taosReadFile(pReader->pReadH, *ppBuf, size);
  if (n < 0) {
H
Hongze Cheng 已提交
371
    code = TAOS_SYSTEM_ERROR(errno);
H
Hongze Cheng 已提交
372
    goto _err;
H
Hongze Cheng 已提交
373 374 375
  } else if (n < size) {
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
H
Hongze Cheng 已提交
376 377 378 379 380 381 382 383 384
  }

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

  // decode
H
Hongze Cheng 已提交
385 386 387 388 389
  n = 0;
  n += tGetU32(*ppBuf + n, &delimiter);
  ASSERT(delimiter == TSDB_FILE_DLMT);
  n += tGetMapData(*ppBuf + n, pDelIdxMap);
  ASSERT(n + sizeof(TSCKSUM) == size);
H
Hongze Cheng 已提交
390 391 392 393

  return code;

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

// SDataFReader ====================================================
struct SDataFReader {
  STsdb     *pTsdb;
  SDFileSet *pSet;
H
Hongze Cheng 已提交
402 403 404 405
  TdFilePtr  pHeadFD;
  TdFilePtr  pDataFD;
  TdFilePtr  pLastFD;
  TdFilePtr  pSmaFD;
H
Hongze Cheng 已提交
406 407 408 409 410 411 412 413 414 415
};

int32_t tsdbDataFReaderOpen(SDataFReader **ppReader, STsdb *pTsdb, SDFileSet *pSet) {
  int32_t code = 0;
  // TODO
  return code;
}

int32_t tsdbDataFReaderClose(SDataFReader *pReader) {
  int32_t code = 0;
H
Hongze Cheng 已提交
416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440

  if (taosCloseFile(&pReader->pHeadFD) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  if (taosCloseFile(&pReader->pDataFD) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  if (taosCloseFile(&pReader->pLastFD) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  if (taosCloseFile(&pReader->pSmaFD) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  return code;

_err:
  tsdbError("vgId:%d data file reader close failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
441 442 443 444
  return code;
}

int32_t tsdbReadBlockIdx(SDataFReader *pReader, SMapData *pMapData, uint8_t **ppBuf) {
H
Hongze Cheng 已提交
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 487 488
  int32_t  code = 0;
  int64_t  offset = -1;  // TODO
  int64_t  size = -1;    // TODO
  int64_t  n;
  uint32_t delimiter;

  // alloc
  if (!ppBuf) ppBuf = &pMapData->pBuf;
  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
  n = 0;
  n += tGetU32(*ppBuf + n, &delimiter);
  ASSERT(delimiter == TSDB_FILE_DLMT);
  n += tGetMapData(*ppBuf + n, pMapData);
  ASSERT(n + sizeof(TSCKSUM) == size);

  return code;

_err:
  tsdbError("vgId:%d read block idx failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
489 490 491 492
  return code;
}

int32_t tsdbReadBlock(SDataFReader *pReader, SBlockIdx *pBlockIdx, SMapData *pMapData, uint8_t **ppBuf) {
H
Hongze Cheng 已提交
493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542
  int32_t  code = 0;
  int64_t  offset = pBlockIdx->offset;
  int64_t  size = pBlockIdx->size;
  int64_t  n;
  uint32_t delimiter;
  tb_uid_t suid;
  tb_uid_t uid;

  // alloc
  if (!ppBuf) ppBuf = &pMapData->pBuf;
  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
  n = 0;
  n += tGetU32(*ppBuf + n, &delimiter);
  ASSERT(delimiter == TSDB_FILE_DLMT);
  n += tGetI64(*ppBuf + n, &suid);
  ASSERT(suid == pBlockIdx->suid);
  n += tGetI64(*ppBuf + n, &uid);
  ASSERT(uid == pBlockIdx->uid);
  n += tGetMapData(*ppBuf + n, pMapData);
  ASSERT(n + sizeof(TSCKSUM) == size);

  return code;

_err:
  tsdbError("vgId:%d read block failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
543 544 545
  return code;
}

H
Hongze Cheng 已提交
546 547
int32_t tsdbReadBlockData(SDataFReader *pReader, SBlockIdx *pBlockIdx, SBlock *pBlock, SBlockData *pBlockData,
                          int16_t *aColId, int32_t nCol, uint8_t **ppBuf1, uint8_t **ppBuf2) {
H
Hongze Cheng 已提交
548 549 550 551 552 553 554 555 556 557 558 559 560 561 562
  int32_t code = 0;
  // TODO
  return code;
}

int32_t tsdbReadBlockSMA(SDataFReader *pReader, SBlockSMA *pBlkSMA) {
  int32_t code = 0;
  // TODO
  return code;
}

// SDataFWriter ====================================================
struct SDataFWriter {
  STsdb     *pTsdb;
  SDFileSet *pSet;
H
Hongze Cheng 已提交
563 564 565 566
  TdFilePtr  pHeadFD;
  TdFilePtr  pDataFD;
  TdFilePtr  pLastFD;
  TdFilePtr  pSmaFD;
H
Hongze Cheng 已提交
567 568 569 570 571 572 573 574 575 576
};

int32_t tsdbDataFWriterOpen(SDataFWriter **ppWriter, STsdb *pTsdb, SDFileSet *pSet) {
  int32_t code = 0;
  // TODO
  return code;
}

int32_t tsdbDataFWriterClose(SDataFWriter *pWriter, int8_t sync) {
  int32_t code = 0;
H
Hongze Cheng 已提交
577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 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

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

    if (taosFsyncFile(pWriter->pDataFD) < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

    if (taosFsyncFile(pWriter->pLastFD) < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

    if (taosFsyncFile(pWriter->pSmaFD) < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }
  }

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

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

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

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

  return code;

_err:
  tsdbError("vgId:%d data file writer close failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
624 625 626 627 628 629 630 631 632
  return code;
}

int32_t tsdbUpdateDFileSetHeader(SDataFWriter *pWriter, uint8_t **ppBuf) {
  int32_t code = 0;
  // TODO
  return code;
}

H
Hongze Cheng 已提交
633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669
int32_t tsdbWriteBlockIdx(SDataFWriter *pWriter, SMapData *pBlockIdxMap, uint8_t **ppBuf) {
  int32_t  code = 0;
  int64_t  size = 0;
  int64_t  n = 0;
  uint8_t *pBuf = NULL;

  // prepare
  size += tPutU32(NULL, TSDB_FILE_DLMT);
  size = size + tPutMapData(NULL, pBlockIdxMap) + sizeof(TSCKSUM);

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

  // build
  n += tPutU32(*ppBuf + n, TSDB_FILE_DLMT);
  n += tPutMapData(*ppBuf, pBlockIdxMap);
  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;
  }

  // update (todo)

  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 已提交
670 671 672
  return code;
}

H
Hongze Cheng 已提交
673
int32_t tsdbWriteBlock(SDataFWriter *pWriter, SMapData *pBlockMap, uint8_t **ppBuf, SBlockIdx *pBlockIdx) {
H
Hongze Cheng 已提交
674 675 676 677
  int32_t  code = 0;
  uint8_t *pBuf = NULL;
  int64_t  size;
  int64_t  n;
H
Hongze Cheng 已提交
678 679

  ASSERT(pBlockMap->nItem > 0);
H
Hongze Cheng 已提交
680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720

  // prepare
  size = 0;
  size += tPutU32(NULL, TSDB_FILE_DLMT);
  size += tPutI64(NULL, pBlockIdx->suid);
  size += tPutI64(NULL, pBlockIdx->uid);
  size = size + tPutMapData(NULL, pBlockMap) + sizeof(TSCKSUM);

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

  // build
  n = 0;
  n += tPutU32(*ppBuf + n, TSDB_FILE_DLMT);
  n += tPutI64(*ppBuf + n, pBlockIdx->suid);
  n += tPutI64(*ppBuf + n, pBlockIdx->uid);
  n += tPutMapData(*ppBuf + n, pBlockMap);
  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;
  }

  // update (todo)
  // pBlockIdx->offset = -1;
  pBlockIdx->size = size;
  // pWriter->pSet->pHeadF.offset

  tsdbTrace("vgId:%d write block, offset:%" PRId64 " size:%" PRId64, TD_VID(pWriter->pTsdb->pVnode), pBlockIdx->offset,
            pBlockIdx->size);
  return code;

_err:
  tsdbError("vgId:%d write block failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
721 722 723
  return code;
}

H
Hongze Cheng 已提交
724 725 726 727 728 729 730
int32_t tsdbWriteBlockData(SDataFWriter *pWriter, SBlockData *pBlockData, uint8_t **ppBuf1, uint8_t **ppBuf2,
                           SBlockIdx *pBlockIdx, SBlock *pBlock) {
  int32_t       code = 0;
  SSubBlock    *pSubBlock = &pBlock->aSubBlock[pBlock->nSubBlock++];
  SBlockCol     bCol;
  int64_t       size;
  int64_t       n;
H
more  
Hongze Cheng 已提交
731
  TdFilePtr     pFileFD = pWriter->pDataFD;  // TODO
H
Hongze Cheng 已提交
732 733 734 735 736 737
  SBlockDataHdr hdr = {.delimiter = TSDB_FILE_DLMT, .suid = pBlockIdx->suid, .uid = pBlockIdx->uid};
  TSCKSUM       cksm;
  uint8_t      *p;
  int64_t       offset;

  pSubBlock->offset = 0;  // TODO: set as file offset
H
Hongze Cheng 已提交
738 739

  pSubBlock->bsize = 0;
H
Hongze Cheng 已提交
740 741 742 743 744 745 746 747

  // HDR
  n = taosWriteFile(pFileFD, &hdr, sizeof(hdr));
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
  pSubBlock->bsize += n;
H
Hongze Cheng 已提交
748 749

  // TSDBKEY
H
Hongze Cheng 已提交
750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793
  pSubBlock->ksize = 0;
  if (pBlock->cmprAlg == NO_COMPRESSION) {
    // TSKEY
    size = sizeof(TSKEY) * pBlockData->nRow;
    n = taosWriteFile(pFileFD, pBlockData->aTSKEY, size);
    if (n < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }
    pSubBlock->ksize += size;
    cksm = taosCalcChecksum(0, (uint8_t *)pBlockData->aTSKEY, size);

    // version
    size = sizeof(int64_t) * pBlockData->nRow;
    n = taosWriteFile(pFileFD, pBlockData->aVersion, size);
    if (n < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }
    pSubBlock->ksize += size;
    cksm = taosCalcChecksum(cksm, (uint8_t *)pBlockData->aVersion, size);

    // cksm
    size = sizeof(cksm);
    n = taosWriteFile(pFileFD, (uint8_t *)&cksm, size);
    if (n < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }
    pSubBlock->ksize += size;
  } else {
    ASSERT(pBlock->cmprAlg == ONE_STAGE_COMP || pBlock->cmprAlg == TWO_STAGE_COMP);

    size = (sizeof(TSKEY) + sizeof(int64_t)) * pBlockData->nRow + COMP_OVERFLOW_BYTES * 2 + sizeof(TSCKSUM);

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

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

    // TSKEY
H
more  
Hongze Cheng 已提交
794 795
    n = tsCompressTimestamp((char *)pBlockData->aTSKEY, sizeof(TSKEY) * pBlockData->nRow, pBlockData->nRow, *ppBuf1,
                            size, pBlock->cmprAlg, *ppBuf2, size);
H
Hongze Cheng 已提交
796 797 798 799 800 801 802
    if (n <= 0) {
      code = TSDB_CODE_COMPRESS_ERROR;
      goto _err;
    }
    pSubBlock->ksize += n;

    // version
H
more  
Hongze Cheng 已提交
803
    n = tsCompressBigint((char *)pBlockData->aVersion, sizeof(int64_t) * pBlockData->nRow, pBlockData->nRow,
H
Hongze Cheng 已提交
804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823
                         *ppBuf1 + pSubBlock->ksize, size - pSubBlock->ksize, pBlock->cmprAlg, *ppBuf2, size);
    if (n <= 0) {
      code = TSDB_CODE_COMPRESS_ERROR;
      goto _err;
    }
    pSubBlock->ksize += n;

    // cksm
    pSubBlock->ksize += sizeof(TSCKSUM);
    ASSERT(pSubBlock->ksize <= size);
    taosCalcChecksumAppend(0, *ppBuf1, pSubBlock->ksize);

    // write
    n = taosWriteFile(pFileFD, *ppBuf1, pSubBlock->ksize);
    if (n < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }
  }
  pSubBlock->bsize += pSubBlock->ksize;
H
Hongze Cheng 已提交
824 825

  // other columns
H
Hongze Cheng 已提交
826 827
  offset = 0;
  tMapDataClear(&pSubBlock->mBlockCol);
H
Hongze Cheng 已提交
828 829 830 831 832 833 834 835 836 837 838 839
  for (int32_t iCol = 0; iCol < pBlockData->nCol; iCol++) {
    SColData *pColData = &pBlockData->aColData[iCol];

    ASSERT(pColData->flags);

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

    bCol.cid = pColData->cid;
    bCol.type = pColData->type;
    bCol.flag = pColData->flags;

    if (pColData->flags != HAS_NULL) {
H
Hongze Cheng 已提交
840 841 842 843 844
      cksm = 0;
      bCol.offset = offset;
      bCol.size = 0;

      // bitmap
H
Hongze Cheng 已提交
845
      if (pColData->flags != HAS_VALUE) {
H
Hongze Cheng 已提交
846 847 848 849 850 851 852 853 854
        // TODO: optimize bitmap part
        n = taosWriteFile(pFileFD, pColData->pBitMap, BIT2_SIZE(pBlockData->nRow));
        if (n < 0) {
          code = TAOS_SYSTEM_ERROR(errno);
          goto _err;
        }

        cksm = taosCalcChecksum(cksm, pColData->pBitMap, n);
        bCol.size += n;
H
Hongze Cheng 已提交
855 856
      }

H
Hongze Cheng 已提交
857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906
      // data
      if (pBlock->cmprAlg == NO_COMPRESSION) {
        // data
        n = taosWriteFile(pFileFD, pColData->pData, pColData->nData);
        if (n < 0) {
          code = TAOS_SYSTEM_ERROR(errno);
          goto _err;
        }
        bCol.size += n;

        // checksum
        cksm = taosCalcChecksum(cksm, pColData->pData, pColData->nData);
        n = taosWriteFile(pFileFD, &cksm, sizeof(cksm));
        if (n < 0) {
          code = TAOS_SYSTEM_ERROR(errno);
          goto _err;
        }
        bCol.size += n;
      } else {
        size = pColData->nData + COMP_OVERFLOW_BYTES + sizeof(TSCKSUM);

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

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

        // data
        n = tDataTypes->compFunc(pColData->pData, pColData->nData, pBlockData->nRow, *ppBuf1, size, pBlock->cmprAlg,
                                 *ppBuf2, size);
        if (n <= 0) {
          code = TSDB_CODE_COMPRESS_ERROR;
          goto _err;
        }

        // cksm
        n += sizeof(TSCKSUM);
        ASSERT(n <= size);
        taosCalcChecksumAppend(cksm, *ppBuf1, n);
        bCol.size += n;

        // write
        n = taosWriteFile(pFileFD, *ppBuf1, bCol.size);
        if (n < 0) {
          code = TAOS_SYSTEM_ERROR(errno);
          goto _err;
        }
      }
H
Hongze Cheng 已提交
907

H
Hongze Cheng 已提交
908 909 910
      // state
      offset += bCol.size;
      pSubBlock->bsize += bCol.size;
H
Hongze Cheng 已提交
911 912 913 914 915 916 917 918 919
    }

    code = tMapDataPutItem(&pSubBlock->mBlockCol, &bCol, tPutBlockCol);
    if (code) goto _err;
  }

  return code;

_err:
H
Hongze Cheng 已提交
920
  tsdbError("vgId:%d write block data failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
921 922 923 924 925 926 927 928
  return code;
}

int32_t tsdbWriteBlockSMA(SDataFWriter *pWriter, SBlockSMA *pBlockSMA, int64_t *rOffset, int64_t *rSize) {
  int32_t code = 0;
  // TODO
  return code;
}