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

#define TSDB_DEFAULT_PAGE_SIZE 4096

H
Hongze Cheng 已提交
20
// =============== PAGE-WISE FILE ===============
H
Hongze Cheng 已提交
21 22 23 24 25
#define PAGE_CONTENT_SIZE(SIZE)   ((SIZE) - sizeof(TSCKSUM))
#define PAGE_OFFSET(PGNO, SIZE)   (((PGNO)-1) * (SIZE))
#define OFFSET_PGNO(OFFSET, SIZE) ((OFFSET) / (SIZE) + 1)

static int32_t tsdbOpenFile(const char *path, int32_t szPage, int32_t flag, STsdbFD **ppFD) {
H
Hongze Cheng 已提交
26 27
  int32_t  code = 0;
  STsdbFD *pFD;
H
refact  
Hongze Cheng 已提交
28

H
Hongze Cheng 已提交
29 30
  *ppFD = NULL;

H
Hongze Cheng 已提交
31
  pFD = (STsdbFD *)taosMemoryCalloc(1, sizeof(*pFD) + strlen(path) + 1);
H
Hongze Cheng 已提交
32 33 34 35
  if (pFD == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }
H
Hongze Cheng 已提交
36

H
Hongze Cheng 已提交
37 38 39 40 41
  pFD->path = (char *)&pFD[1];
  strcpy(pFD->path, path);
  pFD->szPage = szPage;
  pFD->flag = flag;
  pFD->pFD = taosOpenFile(path, flag);
H
Hongze Cheng 已提交
42
  if (pFD->pFD == NULL) {
H
Hongze Cheng 已提交
43
    code = TAOS_SYSTEM_ERROR(errno);
H
Hongze Cheng 已提交
44
    goto _exit;
H
Hongze Cheng 已提交
45
  }
H
Hongze Cheng 已提交
46
  pFD->szPage = szPage;
H
Hongze Cheng 已提交
47 48
  pFD->pgno = 0;
  pFD->nBuf = 0;
H
Hongze Cheng 已提交
49
  pFD->pBuf = taosMemoryMalloc(szPage);
H
Hongze Cheng 已提交
50
  if (pFD->pBuf == NULL) {
H
Hongze Cheng 已提交
51
    code = TSDB_CODE_OUT_OF_MEMORY;
H
Hongze Cheng 已提交
52
    taosMemoryFree(pFD);
H
Hongze Cheng 已提交
53
    goto _exit;
H
Hongze Cheng 已提交
54
  }
H
Hongze Cheng 已提交
55
  *ppFD = pFD;
H
Hongze Cheng 已提交
56 57 58

_exit:
  return code;
H
Hongze Cheng 已提交
59
}
H
Hongze Cheng 已提交
60

H
Hongze Cheng 已提交
61 62
static void tsdbCloseFile(STsdbFD **ppFD) {
  STsdbFD *pFD = *ppFD;
H
Hongze Cheng 已提交
63 64
  taosMemoryFree(pFD->pBuf);
  taosCloseFile(&pFD->pFD);
H
Hongze Cheng 已提交
65 66
  taosMemoryFree(pFD);
  *ppFD = NULL;
H
Hongze Cheng 已提交
67 68
}

H
Hongze Cheng 已提交
69
static int32_t tsdbFsyncFile(STsdbFD *pFD) {
H
Hongze Cheng 已提交
70
  int32_t code = 0;
H
Hongze Cheng 已提交
71

H
Hongze Cheng 已提交
72 73 74
  if (taosFsyncFile(pFD->pFD) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _exit;
H
Hongze Cheng 已提交
75 76
  }

H
Hongze Cheng 已提交
77
_exit:
H
Hongze Cheng 已提交
78 79 80
  return code;
}

H
Hongze Cheng 已提交
81
static int32_t tsdbWriteFile(STsdbFD *pFD, uint8_t *pBuf, int32_t nBuf, int64_t *offset) {
H
Hongze Cheng 已提交
82
  int32_t code = 0;
H
Hongze Cheng 已提交
83

H
Hongze Cheng 已提交
84 85 86 87
  int32_t n = 0;
  while (n < nBuf) {
    int32_t remain = pFD->szPage - pFD->nBuf - sizeof(TSCKSUM);
    int32_t size = TMIN(remain, nBuf - n);
H
Hongze Cheng 已提交
88

H
Hongze Cheng 已提交
89 90 91
    memcpy(pFD->pBuf + pFD->nBuf, pBuf + n, size);
    n += size;
    pFD->nBuf += size;
H
Hongze Cheng 已提交
92

H
Hongze Cheng 已提交
93 94
    if (pFD->nBuf + sizeof(TSCKSUM) == pFD->szPage) {
      taosCalcChecksumAppend(0, pFD->pBuf, pFD->szPage);
H
Hongze Cheng 已提交
95

H
Hongze Cheng 已提交
96 97 98 99 100
      int64_t n = taosWriteFile(pFD->pFD, pFD->pBuf, pFD->szPage);
      if (n < 0) {
        code = TAOS_SYSTEM_ERROR(errno);
        goto _exit;
      }
H
Hongze Cheng 已提交
101

H
Hongze Cheng 已提交
102
      pFD->nBuf = 0;
H
Hongze Cheng 已提交
103
    }
H
Hongze Cheng 已提交
104
  }
H
Hongze Cheng 已提交
105 106

_exit:
H
Hongze Cheng 已提交
107 108 109
  return code;
}

H
Hongze Cheng 已提交
110
static int32_t tsdbReadFilePage(STsdbFD *pFD, int64_t pgno) {
H
Hongze Cheng 已提交
111
  int32_t code = 0;
H
more  
Hongze Cheng 已提交
112

H
Hongze Cheng 已提交
113 114 115
  // seek
  int64_t offset = PAGE_OFFSET(pgno, pFD->szPage);
  int64_t n = taosLSeekFile(pFD->pFD, offset, SEEK_SET);
H
Hongze Cheng 已提交
116
  if (n < 0) {
H
Hongze Cheng 已提交
117
    code = TAOS_SYSTEM_ERROR(errno);
H
Hongze Cheng 已提交
118
    goto _exit;
H
Hongze Cheng 已提交
119 120
  }

H
Hongze Cheng 已提交
121
  // read
H
Hongze Cheng 已提交
122
  n = taosReadFile(pFD->pFD, pFD->pBuf, pFD->szPage);
H
Hongze Cheng 已提交
123
  if (n < 0) {
H
Hongze Cheng 已提交
124
    code = TAOS_SYSTEM_ERROR(errno);
H
Hongze Cheng 已提交
125 126
    goto _exit;
  } else if (n < pFD->szPage) {
H
Hongze Cheng 已提交
127
    code = TSDB_CODE_FILE_CORRUPTED;
H
Hongze Cheng 已提交
128
    goto _exit;
H
Hongze Cheng 已提交
129 130
  }

H
Hongze Cheng 已提交
131
  // check
H
Hongze Cheng 已提交
132
  if (!taosCheckChecksumWhole(pFD->pBuf, pFD->szPage)) {
H
Hongze Cheng 已提交
133
    code = TSDB_CODE_FILE_CORRUPTED;
H
Hongze Cheng 已提交
134
    goto _exit;
H
Hongze Cheng 已提交
135 136
  }

H
Hongze Cheng 已提交
137
  pFD->pgno = pgno;
H
Hongze Cheng 已提交
138

H
Hongze Cheng 已提交
139 140 141
_exit:
  return code;
}
H
Hongze Cheng 已提交
142

H
Hongze Cheng 已提交
143
static int32_t tsdbReadFile(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t count) {
H
Hongze Cheng 已提交
144
  int32_t code = 0;
H
Hongze Cheng 已提交
145 146 147
  int64_t n;
  int64_t pgno = OFFSET_PGNO(offset, pFD->szPage);
  int32_t szPgCont = PAGE_CONTENT_SIZE(pFD->szPage);
H
Hongze Cheng 已提交
148

H
Hongze Cheng 已提交
149
  ASSERT(pgno);
H
Hongze Cheng 已提交
150 151
  if (pFD->pgno == pgno) {
    int64_t bOff = offset % pFD->szPage;
H
Hongze Cheng 已提交
152 153 154 155 156
    int64_t nRead = TMIN(szPgCont - bOff, count);

    ASSERT(bOff < szPgCont);

    memcpy(pBuf, pFD->pBuf + bOff, nRead);
H
Hongze Cheng 已提交
157
    n = nRead;
H
Hongze Cheng 已提交
158
    pgno++;
H
Hongze Cheng 已提交
159 160
  }

H
Hongze Cheng 已提交
161 162 163
  while (n < count) {
    code = tsdbReadFilePage(pFD, pgno);
    if (code) goto _exit;
H
Hongze Cheng 已提交
164

H
Hongze Cheng 已提交
165
    int64_t nRead = TMIN(szPgCont, count - n);
H
Hongze Cheng 已提交
166
    memcpy(pBuf + n, pFD->pBuf, nRead);
H
Hongze Cheng 已提交
167

H
Hongze Cheng 已提交
168
    n += nRead;
H
Hongze Cheng 已提交
169
    pgno++;
H
Hongze Cheng 已提交
170 171 172
  }

_exit:
H
Hongze Cheng 已提交
173
  return code;
H
Hongze Cheng 已提交
174 175
}

H
Hongze Cheng 已提交
176 177 178 179 180 181
static int32_t tsdbLSeekFile(STsdbFD *pFD, int64_t offset) {
  int32_t code = 0;
  ASSERT(0);
  return code;
}

H
Hongze Cheng 已提交
182 183
// SDataFWriter ====================================================
int32_t tsdbDataFWriterOpen(SDataFWriter **ppWriter, STsdb *pTsdb, SDFileSet *pSet) {
H
Hongze Cheng 已提交
184
  int32_t       code = 0;
H
Hongze Cheng 已提交
185 186
  int32_t       flag;
  int64_t       n;
H
Hongze Cheng 已提交
187
  int32_t       szPage = TSDB_DEFAULT_PAGE_SIZE;
H
Hongze Cheng 已提交
188
  SDataFWriter *pWriter = NULL;
H
Hongze Cheng 已提交
189
  char          fname[TSDB_FILENAME_LEN];
H
Hongze Cheng 已提交
190
  char          hdr[TSDB_FHDR_SIZE] = {0};
H
Hongze Cheng 已提交
191 192

  // alloc
H
Hongze Cheng 已提交
193 194
  pWriter = taosMemoryCalloc(1, sizeof(*pWriter));
  if (pWriter == NULL) {
H
Hongze Cheng 已提交
195 196 197
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
H
Hongze Cheng 已提交
198
  pWriter->pTsdb = pTsdb;
H
Hongze Cheng 已提交
199 200 201 202 203 204
  pWriter->wSet = (SDFileSet){.diskId = pSet->diskId,
                              .fid = pSet->fid,
                              .pHeadF = &pWriter->fHead,
                              .pDataF = &pWriter->fData,
                              .pSmaF = &pWriter->fSma,
                              .nSstF = pSet->nSstF};
H
Hongze Cheng 已提交
205 206 207 208 209 210 211
  pWriter->fHead = *pSet->pHeadF;
  pWriter->fData = *pSet->pDataF;
  pWriter->fSma = *pSet->pSmaF;
  for (int8_t iSst = 0; iSst < pSet->nSstF; iSst++) {
    pWriter->wSet.aSstF[iSst] = &pWriter->fSst[iSst];
    pWriter->fSst[iSst] = *pSet->aSstF[iSst];
  }
H
Hongze Cheng 已提交
212 213

  // head
H
Hongze Cheng 已提交
214
  flag = TD_FILE_READ | TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
H
Hongze Cheng 已提交
215
  tsdbHeadFileName(pTsdb, pWriter->wSet.diskId, pWriter->wSet.fid, &pWriter->fHead, fname);
H
Hongze Cheng 已提交
216 217
  code = tsdbOpenFile(fname, szPage, flag, &pWriter->pHeadFD);
  if (code) goto _err;
H
Hongze Cheng 已提交
218

H
Hongze Cheng 已提交
219 220
  code = tsdbWriteFile(pWriter->pHeadFD, hdr, TSDB_FHDR_SIZE, NULL);
  if (code) goto _err;
H
Hongze Cheng 已提交
221

H
Hongze Cheng 已提交
222 223 224 225
  ASSERT(n == TSDB_FHDR_SIZE);

  pWriter->fHead.size += TSDB_FHDR_SIZE;

H
Hongze Cheng 已提交
226
  // data
H
Hongze Cheng 已提交
227
  if (pWriter->fData.size == 0) {
H
Hongze Cheng 已提交
228
    flag = TD_FILE_READ | TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
H
Hongze Cheng 已提交
229
  } else {
H
Hongze Cheng 已提交
230
    flag = TD_FILE_READ | TD_FILE_WRITE;
H
Hongze Cheng 已提交
231 232
  }
  tsdbDataFileName(pTsdb, pWriter->wSet.diskId, pWriter->wSet.fid, &pWriter->fData, fname);
H
Hongze Cheng 已提交
233 234
  code = tsdbOpenFile(fname, szPage, flag, &pWriter->pDataFD);
  if (code) goto _err;
H
Hongze Cheng 已提交
235
  if (pWriter->fData.size == 0) {
H
Hongze Cheng 已提交
236 237
    code = tsdbWriteFile(pWriter->pDataFD, hdr, TSDB_FHDR_SIZE, NULL);
    if (code) goto _err;
H
Hongze Cheng 已提交
238 239
    pWriter->fData.size += TSDB_FHDR_SIZE;
  } else {
H
Hongze Cheng 已提交
240 241
    // code = tsdbLSeekFile(pWriter->pDataFD, 0, SEEK_END);
    // if (code) goto _err;
H
Hongze Cheng 已提交
242
  }
H
Hongze Cheng 已提交
243 244

  // sma
H
Hongze Cheng 已提交
245
  if (pWriter->fSma.size == 0) {
H
Hongze Cheng 已提交
246
    flag = TD_FILE_READ | TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
H
Hongze Cheng 已提交
247
  } else {
H
Hongze Cheng 已提交
248
    flag = TD_FILE_READ | TD_FILE_WRITE;
H
Hongze Cheng 已提交
249 250
  }
  tsdbSmaFileName(pTsdb, pWriter->wSet.diskId, pWriter->wSet.fid, &pWriter->fSma, fname);
H
Hongze Cheng 已提交
251 252
  code = tsdbOpenFile(fname, szPage, flag, &pWriter->pSmaFD);
  if (code) goto _err;
H
Hongze Cheng 已提交
253
  if (pWriter->fSma.size == 0) {
H
Hongze Cheng 已提交
254 255
    code = tsdbWriteFile(pWriter->pSmaFD, hdr, TSDB_FHDR_SIZE, NULL);
    if (code) goto _err;
H
Hongze Cheng 已提交
256

H
Hongze Cheng 已提交
257 258
    pWriter->fSma.size += TSDB_FHDR_SIZE;
  } else {
H
Hongze Cheng 已提交
259 260
    code = tsdbLSeekFile(pWriter->pSmaFD, 0);
    if (code) goto _err;
H
Hongze Cheng 已提交
261 262
  }

H
Hongze Cheng 已提交
263 264
  // sst
  ASSERT(pWriter->fSst[pSet->nSstF - 1].size == 0);
H
Hongze Cheng 已提交
265
  flag = TD_FILE_READ | TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
H
Hongze Cheng 已提交
266
  tsdbSstFileName(pTsdb, pWriter->wSet.diskId, pWriter->wSet.fid, &pWriter->fSst[pSet->nSstF - 1], fname);
H
Hongze Cheng 已提交
267 268 269 270
  code = tsdbOpenFile(fname, szPage, flag, &pWriter->pLastFD);
  if (code) goto _err;
  code = tsdbWriteFile(pWriter->pLastFD, hdr, TSDB_FHDR_SIZE, NULL);
  if (code) goto _err;
H
Hongze Cheng 已提交
271 272 273
  pWriter->fSst[pWriter->wSet.nSstF - 1].size += TSDB_FHDR_SIZE;

  *ppWriter = pWriter;
H
Hongze Cheng 已提交
274 275 276
  return code;

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

H
Hongze Cheng 已提交
282
int32_t tsdbDataFWriterClose(SDataFWriter **ppWriter, int8_t sync) {
H
Hongze Cheng 已提交
283
  int32_t code = 0;
H
Hongze Cheng 已提交
284
  STsdb  *pTsdb = NULL;
H
Hongze Cheng 已提交
285

H
Hongze Cheng 已提交
286 287 288 289
  if (*ppWriter == NULL) goto _exit;

  pTsdb = (*ppWriter)->pTsdb;
  if (sync) {
H
Hongze Cheng 已提交
290
    if (tsdbFsyncFile((*ppWriter)->pHeadFD) < 0) {
H
Hongze Cheng 已提交
291 292 293 294
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

H
Hongze Cheng 已提交
295
    if (tsdbFsyncFile((*ppWriter)->pDataFD) < 0) {
H
Hongze Cheng 已提交
296 297 298 299
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

H
Hongze Cheng 已提交
300
    if (tsdbFsyncFile((*ppWriter)->pSmaFD) < 0) {
H
Hongze Cheng 已提交
301 302 303 304
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }

H
Hongze Cheng 已提交
305
    if (tsdbFsyncFile((*ppWriter)->pLastFD) < 0) {
H
Hongze Cheng 已提交
306 307 308 309 310
      code = TAOS_SYSTEM_ERROR(errno);
      goto _err;
    }
  }

H
Hongze Cheng 已提交
311 312 313 314
  tsdbCloseFile(&(*ppWriter)->pHeadFD);
  tsdbCloseFile(&(*ppWriter)->pDataFD);
  tsdbCloseFile(&(*ppWriter)->pSmaFD);
  tsdbCloseFile(&(*ppWriter)->pLastFD);
H
Hongze Cheng 已提交
315

H
Hongze Cheng 已提交
316 317
  for (int32_t iBuf = 0; iBuf < sizeof((*ppWriter)->aBuf) / sizeof(uint8_t *); iBuf++) {
    tFree((*ppWriter)->aBuf[iBuf]);
H
Hongze Cheng 已提交
318
  }
H
Hongze Cheng 已提交
319
  taosMemoryFree(*ppWriter);
H
Hongze Cheng 已提交
320
_exit:
H
Hongze Cheng 已提交
321
  *ppWriter = NULL;
H
Hongze Cheng 已提交
322 323 324
  return code;

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

H
Hongze Cheng 已提交
329 330 331 332
int32_t tsdbUpdateDFileSetHeader(SDataFWriter *pWriter) {
  int32_t code = 0;
  int64_t n;
  char    hdr[TSDB_FHDR_SIZE];
H
Hongze Cheng 已提交
333

H
Hongze Cheng 已提交
334 335 336
  // head ==============
  memset(hdr, 0, TSDB_FHDR_SIZE);
  tPutHeadFile(hdr, &pWriter->fHead);
H
Hongze Cheng 已提交
337

H
Hongze Cheng 已提交
338 339
  code = tsdbLSeekFile(pWriter->pHeadFD, 0);
  if (code) goto _err;
H
Hongze Cheng 已提交
340

H
Hongze Cheng 已提交
341 342
  code = tsdbWriteFile(pWriter->pHeadFD, hdr, TSDB_FHDR_SIZE, NULL);
  if (code) goto _err;
H
Hongze Cheng 已提交
343 344 345 346 347

  // data ==============
  memset(hdr, 0, TSDB_FHDR_SIZE);
  tPutDataFile(hdr, &pWriter->fData);

H
Hongze Cheng 已提交
348 349
  code = tsdbLSeekFile(pWriter->pDataFD, 0);
  if (code) goto _err;
H
Hongze Cheng 已提交
350

H
Hongze Cheng 已提交
351 352
  code = tsdbWriteFile(pWriter->pDataFD, hdr, TSDB_FHDR_SIZE, NULL);
  if (code) goto _err;
H
Hongze Cheng 已提交
353

H
Hongze Cheng 已提交
354 355 356
  // sma ==============
  memset(hdr, 0, TSDB_FHDR_SIZE);
  tPutSmaFile(hdr, &pWriter->fSma);
H
Hongze Cheng 已提交
357

H
Hongze Cheng 已提交
358 359
  code = tsdbLSeekFile(pWriter->pSmaFD, 0);
  if (code) goto _err;
H
Hongze Cheng 已提交
360

H
Hongze Cheng 已提交
361 362
  code = tsdbWriteFile(pWriter->pSmaFD, hdr, TSDB_FHDR_SIZE, NULL);
  if (code) goto _err;
H
Hongze Cheng 已提交
363

H
Hongze Cheng 已提交
364 365 366 367
  // sst ==============
  memset(hdr, 0, TSDB_FHDR_SIZE);
  tPutSstFile(hdr, &pWriter->fSst[pWriter->wSet.nSstF - 1]);

H
Hongze Cheng 已提交
368 369
  code = tsdbLSeekFile(pWriter->pLastFD, 0);
  if (code) goto _err;
H
Hongze Cheng 已提交
370

H
Hongze Cheng 已提交
371 372
  code = tsdbWriteFile(pWriter->pLastFD, hdr, TSDB_FHDR_SIZE, NULL);
  if (code) goto _err;
H
Hongze Cheng 已提交
373 374 375 376

  return code;

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

H
Hongze Cheng 已提交
381 382 383 384 385
int32_t tsdbWriteBlockIdx(SDataFWriter *pWriter, SArray *aBlockIdx) {
  int32_t    code = 0;
  SHeadFile *pHeadFile = &pWriter->fHead;
  int64_t    size = 0;
  int64_t    n;
H
Hongze Cheng 已提交
386

H
Hongze Cheng 已提交
387 388 389
  // check
  if (taosArrayGetSize(aBlockIdx) == 0) {
    pHeadFile->offset = pHeadFile->size;
H
Hongze Cheng 已提交
390 391
    goto _exit;
  }
H
Hongze Cheng 已提交
392

H
Hongze Cheng 已提交
393 394 395 396 397
  // prepare
  for (int32_t iBlockIdx = 0; iBlockIdx < taosArrayGetSize(aBlockIdx); iBlockIdx++) {
    size += tPutBlockIdx(NULL, taosArrayGet(aBlockIdx, iBlockIdx));
  }

H
Hongze Cheng 已提交
398
  // alloc
H
Hongze Cheng 已提交
399
  code = tRealloc(&pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
400 401
  if (code) goto _err;

H
Hongze Cheng 已提交
402 403 404 405
  // build
  n = 0;
  for (int32_t iBlockIdx = 0; iBlockIdx < taosArrayGetSize(aBlockIdx); iBlockIdx++) {
    n += tPutBlockIdx(pWriter->aBuf[0] + n, taosArrayGet(aBlockIdx, iBlockIdx));
H
Hongze Cheng 已提交
406
  }
H
Hongze Cheng 已提交
407
  ASSERT(n == size);
H
Hongze Cheng 已提交
408 409

  // write
H
Hongze Cheng 已提交
410 411
  code = tsdbWriteFile(pWriter->pHeadFD, pWriter->aBuf[0], size, NULL);
  if (code) goto _err;
H
Hongze Cheng 已提交
412

H
Hongze Cheng 已提交
413 414 415
  // update
  pHeadFile->offset = pHeadFile->size;
  pHeadFile->size += size;
H
Hongze Cheng 已提交
416

H
Hongze Cheng 已提交
417
_exit:
H
Hongze Cheng 已提交
418 419
  tsdbTrace("vgId:%d write block idx, offset:%" PRId64 " size:%" PRId64 " nBlockIdx:%d", TD_VID(pWriter->pTsdb->pVnode),
            pHeadFile->offset, size, taosArrayGetSize(aBlockIdx));
H
Hongze Cheng 已提交
420 421 422
  return code;

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

H
Hongze Cheng 已提交
427 428 429 430 431 432 433
int32_t tsdbWriteBlock(SDataFWriter *pWriter, SMapData *mBlock, SBlockIdx *pBlockIdx) {
  int32_t    code = 0;
  SHeadFile *pHeadFile = &pWriter->fHead;
  int64_t    size;
  int64_t    n;

  ASSERT(mBlock->nItem > 0);
H
Hongze Cheng 已提交
434

H
Hongze Cheng 已提交
435
  // alloc
H
Hongze Cheng 已提交
436
  size = tPutMapData(NULL, mBlock);
H
Hongze Cheng 已提交
437
  code = tRealloc(&pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
438 439
  if (code) goto _err;

H
Hongze Cheng 已提交
440
  // build
H
Hongze Cheng 已提交
441
  n = tPutMapData(pWriter->aBuf[0] + n, mBlock);
H
Hongze Cheng 已提交
442 443

  // write
H
Hongze Cheng 已提交
444 445
  code = tsdbWriteFile(pWriter->pHeadFD, pWriter->aBuf[0], size, NULL);
  if (code) goto _err;
H
Hongze Cheng 已提交
446

H
Hongze Cheng 已提交
447 448 449 450
  // update
  pBlockIdx->offset = pHeadFile->size;
  pBlockIdx->size = size;
  pHeadFile->size += size;
H
Hongze Cheng 已提交
451

H
Hongze Cheng 已提交
452 453 454 455
  tsdbTrace("vgId:%d, write block, file ID:%d commit ID:%d suid:%" PRId64 " uid:%" PRId64 " offset:%" PRId64
            " size:%" PRId64 " nItem:%d",
            TD_VID(pWriter->pTsdb->pVnode), pWriter->wSet.fid, pHeadFile->commitID, pBlockIdx->suid, pBlockIdx->uid,
            pBlockIdx->offset, pBlockIdx->size, mBlock->nItem);
H
Hongze Cheng 已提交
456 457 458
  return code;

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

H
Hongze Cheng 已提交
463
int32_t tsdbWriteSstBlk(SDataFWriter *pWriter, SArray *aSstBlk) {
H
Hongze Cheng 已提交
464
  int32_t   code = 0;
H
Hongze Cheng 已提交
465
  SSstFile *pSstFile = &pWriter->fSst[pWriter->wSet.nSstF - 1];
H
Hongze Cheng 已提交
466
  int64_t   size = 0;
H
Hongze Cheng 已提交
467
  int64_t   n;
H
Hongze Cheng 已提交
468

H
Hongze Cheng 已提交
469 470 471 472 473
  // check
  if (taosArrayGetSize(aSstBlk) == 0) {
    pSstFile->offset = pSstFile->size;
    goto _exit;
  }
H
Hongze Cheng 已提交
474

H
Hongze Cheng 已提交
475 476 477 478
  // size
  for (int32_t iBlockL = 0; iBlockL < taosArrayGetSize(aSstBlk); iBlockL++) {
    size += tPutSstBlk(NULL, taosArrayGet(aSstBlk, iBlockL));
  }
H
Hongze Cheng 已提交
479 480

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

H
Hongze Cheng 已提交
484 485 486 487
  // encode
  n = 0;
  for (int32_t iBlockL = 0; iBlockL < taosArrayGetSize(aSstBlk); iBlockL++) {
    n += tPutSstBlk(pWriter->aBuf[0] + n, taosArrayGet(aSstBlk, iBlockL));
H
Hongze Cheng 已提交
488
  }
H
Hongze Cheng 已提交
489 490

  // write
H
Hongze Cheng 已提交
491 492
  code = tsdbWriteFile(pWriter->pLastFD, pWriter->aBuf[0], size, NULL);
  if (code) goto _err;
H
Hongze Cheng 已提交
493

H
Hongze Cheng 已提交
494 495 496
  // update
  pSstFile->offset = pSstFile->size;
  pSstFile->size += size;
H
Hongze Cheng 已提交
497

H
Hongze Cheng 已提交
498 499 500
_exit:
  tsdbTrace("vgId:%d tsdb write blockl, loffset:%" PRId64 " size:%" PRId64, TD_VID(pWriter->pTsdb->pVnode),
            pSstFile->offset, size);
H
Hongze Cheng 已提交
501 502 503
  return code;

_err:
H
Hongze Cheng 已提交
504
  tsdbError("vgId:%d tsdb write blockl failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
505 506 507
  return code;
}

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

H
Hongze Cheng 已提交
511 512
  pSmaInfo->offset = 0;
  pSmaInfo->size = 0;
H
Hongze Cheng 已提交
513

H
Hongze Cheng 已提交
514 515 516
  // encode
  for (int32_t iColData = 0; iColData < taosArrayGetSize(pBlockData->aIdx); iColData++) {
    SColData *pColData = tBlockDataGetColDataByIdx(pBlockData, iColData);
H
Hongze Cheng 已提交
517

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

H
Hongze Cheng 已提交
520 521
    SColumnDataAgg sma;
    tsdbCalcColDataSMA(pColData, &sma);
H
Hongze Cheng 已提交
522

H
Hongze Cheng 已提交
523 524 525 526
    code = tRealloc(&pWriter->aBuf[0], pSmaInfo->size + tPutColumnDataAgg(NULL, &sma));
    if (code) goto _err;
    pSmaInfo->size += tPutColumnDataAgg(pWriter->aBuf[0] + pSmaInfo->size, &sma);
  }
H
Hongze Cheng 已提交
527

H
Hongze Cheng 已提交
528 529
  // write
  if (pSmaInfo->size) {
H
Hongze Cheng 已提交
530
    code = tRealloc(&pWriter->aBuf[0], pSmaInfo->size);
H
Hongze Cheng 已提交
531
    if (code) goto _err;
H
Hongze Cheng 已提交
532

H
Hongze Cheng 已提交
533 534
    code = tsdbWriteFile(pWriter->pSmaFD, pWriter->aBuf[0], pSmaInfo->size, NULL);
    if (code) goto _err;
H
Hongze Cheng 已提交
535

H
Hongze Cheng 已提交
536
    pSmaInfo->offset = pWriter->fSma.size;
H
Hongze Cheng 已提交
537
    // pWriter->fSma.size += size;
H
Hongze Cheng 已提交
538 539 540 541 542
  }

  return code;

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

H
Hongze Cheng 已提交
547 548
int32_t tsdbWriteBlockData(SDataFWriter *pWriter, SBlockData *pBlockData, SBlockInfo *pBlkInfo, SSmaInfo *pSmaInfo,
                           int8_t cmprAlg, int8_t toLast) {
H
Hongze Cheng 已提交
549 550
  int32_t code = 0;

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

H
Hongze Cheng 已提交
553 554 555
  pBlkInfo->offset = toLast ? pWriter->fSst[pWriter->wSet.nSstF - 1].size : pWriter->fData.size;
  pBlkInfo->szBlock = 0;
  pBlkInfo->szKey = 0;
H
Hongze Cheng 已提交
556

H
Hongze Cheng 已提交
557 558 559
  int32_t aBufN[4] = {0};
  code = tCmprBlockData(pBlockData, cmprAlg, NULL, NULL, pWriter->aBuf, aBufN);
  if (code) goto _err;
H
Hongze Cheng 已提交
560

H
Hongze Cheng 已提交
561
  // write =================
H
Hongze Cheng 已提交
562
  STsdbFD *pFD = toLast ? pWriter->pLastFD : pWriter->pDataFD;
H
Hongze Cheng 已提交
563

H
Hongze Cheng 已提交
564 565
  pBlkInfo->szKey = aBufN[3] + aBufN[2];
  pBlkInfo->szBlock = aBufN[0] + aBufN[1] + aBufN[2] + aBufN[3];
H
Hongze Cheng 已提交
566

H
Hongze Cheng 已提交
567 568
  code = tsdbWriteFile(pFD, pWriter->aBuf[3], aBufN[3], NULL);
  if (code) goto _err;
H
Hongze Cheng 已提交
569

H
Hongze Cheng 已提交
570 571
  code = tsdbWriteFile(pFD, pWriter->aBuf[2], aBufN[2], NULL);
  if (code) goto _err;
H
Hongze Cheng 已提交
572

H
Hongze Cheng 已提交
573
  if (aBufN[1]) {
H
Hongze Cheng 已提交
574 575
    code = tsdbWriteFile(pFD, pWriter->aBuf[1], aBufN[1], NULL);
    if (code) goto _err;
H
Hongze Cheng 已提交
576
  }
H
Hongze Cheng 已提交
577

H
Hongze Cheng 已提交
578
  if (aBufN[0]) {
H
Hongze Cheng 已提交
579 580
    code = tsdbWriteFile(pFD, pWriter->aBuf[0], aBufN[0], NULL);
    if (code) goto _err;
H
Hongze Cheng 已提交
581
  }
H
Hongze Cheng 已提交
582

H
Hongze Cheng 已提交
583 584 585 586 587 588
  // update info
  if (toLast) {
    pWriter->fSst[pWriter->wSet.nSstF - 1].size += pBlkInfo->szBlock;
  } else {
    pWriter->fData.size += pBlkInfo->szBlock;
  }
H
Hongze Cheng 已提交
589

H
Hongze Cheng 已提交
590 591 592 593 594
  // ================= SMA ====================
  if (pSmaInfo) {
    code = tsdbWriteBlockSma(pWriter, pBlockData, pSmaInfo);
    if (code) goto _err;
  }
H
Hongze Cheng 已提交
595

H
Hongze Cheng 已提交
596 597 598 599
_exit:
  tsdbTrace("vgId:%d tsdb write block data, suid:%" PRId64 " uid:%" PRId64 " nRow:%d, offset:%" PRId64 " size:%d",
            TD_VID(pWriter->pTsdb->pVnode), pBlockData->suid, pBlockData->uid, pBlockData->nRow, pBlkInfo->offset,
            pBlkInfo->szBlock);
H
Hongze Cheng 已提交
600 601 602
  return code;

_err:
H
Hongze Cheng 已提交
603
  tsdbError("vgId:%d tsdb write block data failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
604 605
  return code;
}
H
Hongze Cheng 已提交
606

H
Hongze Cheng 已提交
607 608 609 610 611 612 613 614
int32_t tsdbDFileSetCopy(STsdb *pTsdb, SDFileSet *pSetFrom, SDFileSet *pSetTo) {
  int32_t   code = 0;
  int64_t   n;
  int64_t   size;
  TdFilePtr pOutFD = NULL;  // TODO
  TdFilePtr PInFD = NULL;   // TODO
  char      fNameFrom[TSDB_FILENAME_LEN];
  char      fNameTo[TSDB_FILENAME_LEN];
H
Hongze Cheng 已提交
615

H
Hongze Cheng 已提交
616 617 618
  // head
  tsdbHeadFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->pHeadF, fNameFrom);
  tsdbHeadFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->pHeadF, fNameTo);
H
Hongze Cheng 已提交
619

H
Hongze Cheng 已提交
620 621 622
  pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
  if (pOutFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
H
Hongze Cheng 已提交
623 624 625
    goto _err;
  }

H
Hongze Cheng 已提交
626 627
  PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
  if (PInFD == NULL) {
H
Hongze Cheng 已提交
628 629 630 631
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
632
  n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pHeadF->size);
H
Hongze Cheng 已提交
633 634 635 636
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
H
Hongze Cheng 已提交
637 638
  taosCloseFile(&pOutFD);
  taosCloseFile(&PInFD);
H
Hongze Cheng 已提交
639 640

  // data
H
Hongze Cheng 已提交
641 642
  tsdbDataFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->pDataF, fNameFrom);
  tsdbDataFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->pDataF, fNameTo);
H
Hongze Cheng 已提交
643

H
Hongze Cheng 已提交
644 645
  pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
  if (pOutFD == NULL) {
H
Hongze Cheng 已提交
646 647 648 649
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
650 651
  PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
  if (PInFD == NULL) {
H
Hongze Cheng 已提交
652 653 654
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
H
Hongze Cheng 已提交
655 656

  n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pDataF->size);
H
Hongze Cheng 已提交
657 658 659 660
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
H
Hongze Cheng 已提交
661 662
  taosCloseFile(&pOutFD);
  taosCloseFile(&PInFD);
H
Hongze Cheng 已提交
663

H
Hongze Cheng 已提交
664 665 666
  // sst
  tsdbSstFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->aSstF[0], fNameFrom);
  tsdbSstFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->aSstF[0], fNameTo);
H
Hongze Cheng 已提交
667

H
Hongze Cheng 已提交
668 669 670 671
  pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
  if (pOutFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
H
Hongze Cheng 已提交
672 673
  }

H
Hongze Cheng 已提交
674 675
  PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
  if (PInFD == NULL) {
H
Hongze Cheng 已提交
676 677 678 679
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
680 681
  n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->aSstF[0]->size);
  if (n < 0) {
H
Hongze Cheng 已提交
682 683 684
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
H
Hongze Cheng 已提交
685 686
  taosCloseFile(&pOutFD);
  taosCloseFile(&PInFD);
H
Hongze Cheng 已提交
687

H
Hongze Cheng 已提交
688 689 690 691 692 693
  // sma
  tsdbSmaFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->pSmaF, fNameFrom);
  tsdbSmaFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->pSmaF, fNameTo);

  pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
  if (pOutFD == NULL) {
H
Hongze Cheng 已提交
694 695 696 697
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
698 699
  PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
  if (PInFD == NULL) {
H
Hongze Cheng 已提交
700 701 702 703
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
704 705 706 707
  n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pSmaF->size);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
H
Hongze Cheng 已提交
708
  }
H
Hongze Cheng 已提交
709 710 711
  taosCloseFile(&pOutFD);
  taosCloseFile(&PInFD);

H
Hongze Cheng 已提交
712 713 714
  return code;

_err:
H
Hongze Cheng 已提交
715
  tsdbError("vgId:%d, tsdb DFileSet copy failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
716 717 718
  return code;
}

H
Hongze Cheng 已提交
719 720 721 722
// SDataFReader ====================================================
int32_t tsdbDataFReaderOpen(SDataFReader **ppReader, STsdb *pTsdb, SDFileSet *pSet) {
  int32_t       code = 0;
  SDataFReader *pReader;
H
Hongze Cheng 已提交
723
  int32_t       szPage = TSDB_DEFAULT_PAGE_SIZE;
H
Hongze Cheng 已提交
724
  char          fname[TSDB_FILENAME_LEN];
H
Hongze Cheng 已提交
725

H
Hongze Cheng 已提交
726 727 728 729
  // alloc
  pReader = (SDataFReader *)taosMemoryCalloc(1, sizeof(*pReader));
  if (pReader == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
H
Hongze Cheng 已提交
730 731
    goto _err;
  }
H
Hongze Cheng 已提交
732 733
  pReader->pTsdb = pTsdb;
  pReader->pSet = pSet;
H
Hongze Cheng 已提交
734

H
Hongze Cheng 已提交
735 736
  // head
  tsdbHeadFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pHeadF, fname);
H
Hongze Cheng 已提交
737 738
  code = tsdbOpenFile(fname, szPage, TD_FILE_READ, &pReader->pHeadFD);
  if (code) goto _err;
H
Hongze Cheng 已提交
739

H
Hongze Cheng 已提交
740 741
  // data
  tsdbDataFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pDataF, fname);
H
Hongze Cheng 已提交
742 743
  code = tsdbOpenFile(fname, szPage, TD_FILE_READ, &pReader->pDataFD);
  if (code) goto _err;
H
Hongze Cheng 已提交
744

H
Hongze Cheng 已提交
745 746
  // sma
  tsdbSmaFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pSmaF, fname);
H
Hongze Cheng 已提交
747 748
  code = tsdbOpenFile(fname, szPage, TD_FILE_READ, &pReader->pSmaFD);
  if (code) goto _err;
H
Hongze Cheng 已提交
749

H
Hongze Cheng 已提交
750 751 752
  // sst
  for (int32_t iSst = 0; iSst < pSet->nSstF; iSst++) {
    tsdbSstFileName(pTsdb, pSet->diskId, pSet->fid, pSet->aSstF[iSst], fname);
H
Hongze Cheng 已提交
753
    code = tsdbOpenFile(fname, szPage, TD_FILE_READ, &pReader->aSstFD[iSst]);
H
Hongze Cheng 已提交
754
    if (code) goto _err;
H
Hongze Cheng 已提交
755
  }
H
Hongze Cheng 已提交
756

H
Hongze Cheng 已提交
757 758 759 760 761 762 763 764 765 766 767
  *ppReader = pReader;
  return code;

_err:
  tsdbError("vgId:%d, tsdb data file reader open failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
  *ppReader = NULL;
  return code;
}

int32_t tsdbDataFReaderClose(SDataFReader **ppReader) {
  int32_t code = 0;
H
Hongze Cheng 已提交
768
  if (*ppReader == NULL) return code;
H
Hongze Cheng 已提交
769 770

  // head
H
Hongze Cheng 已提交
771
  tsdbCloseFile(&(*ppReader)->pHeadFD);
H
Hongze Cheng 已提交
772

H
Hongze Cheng 已提交
773
  // data
H
Hongze Cheng 已提交
774
  tsdbCloseFile(&(*ppReader)->pDataFD);
H
Hongze Cheng 已提交
775

H
Hongze Cheng 已提交
776
  // sma
H
Hongze Cheng 已提交
777
  tsdbCloseFile(&(*ppReader)->pSmaFD);
H
Hongze Cheng 已提交
778

H
Hongze Cheng 已提交
779 780
  // sst
  for (int32_t iSst = 0; iSst < (*ppReader)->pSet->nSstF; iSst++) {
H
Hongze Cheng 已提交
781
    tsdbCloseFile(&(*ppReader)->aSstFD[iSst]);
H
Hongze Cheng 已提交
782 783 784 785
  }

  for (int32_t iBuf = 0; iBuf < sizeof((*ppReader)->aBuf) / sizeof(uint8_t *); iBuf++) {
    tFree((*ppReader)->aBuf[iBuf]);
H
Hongze Cheng 已提交
786
  }
H
Hongze Cheng 已提交
787 788
  taosMemoryFree(*ppReader);
  *ppReader = NULL;
H
Hongze Cheng 已提交
789 790 791
  return code;

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

H
Hongze Cheng 已提交
796
int32_t tsdbReadBlockIdx(SDataFReader *pReader, SArray *aBlockIdx) {
H
Hongze Cheng 已提交
797 798 799
  int32_t code = 0;
  int64_t offset = pReader->pSet->pHeadF->offset;
  int64_t size = pReader->pSet->pHeadF->size - offset;  // todo
H
Hongze Cheng 已提交
800

H
Hongze Cheng 已提交
801
  taosArrayClear(aBlockIdx);
H
Hongze Cheng 已提交
802
  if (size == 0) return code;
H
Hongze Cheng 已提交
803 804

  // alloc
H
Hongze Cheng 已提交
805
  code = tRealloc(&pReader->aBuf[0], size);
H
Hongze Cheng 已提交
806 807
  if (code) goto _err;

H
Hongze Cheng 已提交
808
  // read
H
Hongze Cheng 已提交
809 810
  code = tsdbReadFile(pReader->pHeadFD, offset, pReader->aBuf[0], size);
  if (code) goto _err;
H
Hongze Cheng 已提交
811 812

  // decode
H
Hongze Cheng 已提交
813 814
  uint32_t delimiter;
  int64_t  n = tGetU32(pReader->aBuf[0], &delimiter);
H
Hongze Cheng 已提交
815 816
  ASSERT(delimiter == TSDB_FILE_DLMT);

H
Hongze Cheng 已提交
817
  while (n < size) {
H
Hongze Cheng 已提交
818 819 820 821 822 823 824
    SBlockIdx blockIdx;
    n += tGetBlockIdx(pReader->aBuf[0] + n, &blockIdx);

    if (taosArrayPush(aBlockIdx, &blockIdx) == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
    }
H
Hongze Cheng 已提交
825
  }
H
Hongze Cheng 已提交
826
  ASSERT(n == size);
H
Hongze Cheng 已提交
827 828 829 830

  return code;

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

H
Hongze Cheng 已提交
835
int32_t tsdbReadSstBlk(SDataFReader *pReader, int32_t iSst, SArray *aSstBlk) {
H
Hongze Cheng 已提交
836 837 838
  int32_t code = 0;
  int64_t offset = pReader->pSet->aSstF[iSst]->offset;
  int64_t size = pReader->pSet->aSstF[iSst]->size - offset;  // todo
H
Hongze Cheng 已提交
839

H
Hongze Cheng 已提交
840
  taosArrayClear(aSstBlk);
H
Hongze Cheng 已提交
841
  if (size == 0) return code;
H
Hongze Cheng 已提交
842 843

  // alloc
H
Hongze Cheng 已提交
844
  code = tRealloc(&pReader->aBuf[0], size);
H
Hongze Cheng 已提交
845 846
  if (code) goto _err;

H
Hongze Cheng 已提交
847
  // read
H
Hongze Cheng 已提交
848 849
  code = tsdbReadFile(pReader->aSstFD[iSst], offset, pReader->aBuf[0], size);
  if (code) goto _err;
H
Hongze Cheng 已提交
850

H
Hongze Cheng 已提交
851
  // decode
H
Hongze Cheng 已提交
852 853
  uint32_t delimiter;
  int64_t  n = tGetU32(pReader->aBuf[0], &delimiter);
H
Hongze Cheng 已提交
854 855
  ASSERT(delimiter == TSDB_FILE_DLMT);

H
Hongze Cheng 已提交
856 857 858
  while (n < size) {
    SSstBlk sstBlk;
    n += tGetSstBlk(pReader->aBuf[0] + n, &sstBlk);
H
Hongze Cheng 已提交
859

H
Hongze Cheng 已提交
860
    if (taosArrayPush(aSstBlk, &sstBlk) == NULL) {
H
Hongze Cheng 已提交
861 862 863 864
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
    }
  }
H
Hongze Cheng 已提交
865
  ASSERT(n == size);
H
Hongze Cheng 已提交
866

H
Hongze Cheng 已提交
867 868 869
  return code;

_err:
H
Hongze Cheng 已提交
870
  tsdbError("vgId:%d read sst blk failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
871 872 873
  return code;
}

H
Hongze Cheng 已提交
874 875 876 877
int32_t tsdbReadBlock(SDataFReader *pReader, SBlockIdx *pBlockIdx, SMapData *mBlock) {
  int32_t code = 0;
  int64_t offset = pBlockIdx->offset;
  int64_t size = pBlockIdx->size;
H
Hongze Cheng 已提交
878 879

  // alloc
H
Hongze Cheng 已提交
880
  code = tRealloc(&pReader->aBuf[0], size);
H
Hongze Cheng 已提交
881 882
  if (code) goto _err;

H
Hongze Cheng 已提交
883
  // read
H
Hongze Cheng 已提交
884 885
  code = tsdbReadFile(pReader->pHeadFD, offset, pReader->aBuf[0], size);
  if (code) goto _err;
H
Hongze Cheng 已提交
886

H
Hongze Cheng 已提交
887 888
  // decode
  uint32_t delimiter;
H
Hongze Cheng 已提交
889
  int64_t  n = tGetU32(pReader->aBuf[0], &delimiter);
H
Hongze Cheng 已提交
890
  ASSERT(delimiter == TSDB_FILE_DLMT);
H
Hongze Cheng 已提交
891

H
Hongze Cheng 已提交
892
  int64_t tn = tGetMapData(pReader->aBuf[0] + n, mBlock);
H
Hongze Cheng 已提交
893 894 895 896
  if (tn < 0) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
H
Hongze Cheng 已提交
897
  ASSERT(n + tn == size);
H
Hongze Cheng 已提交
898

H
Hongze Cheng 已提交
899
  return code;
H
Hongze Cheng 已提交
900

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

H
Hongze Cheng 已提交
906 907 908
int32_t tsdbReadBlockSma(SDataFReader *pReader, SDataBlk *pDataBlk, SArray *aColumnDataAgg) {
  int32_t   code = 0;
  SSmaInfo *pSmaInfo = &pDataBlk->smaInfo;
H
Hongze Cheng 已提交
909

H
Hongze Cheng 已提交
910
  ASSERT(pSmaInfo->size > 0);
H
Hongze Cheng 已提交
911

H
Hongze Cheng 已提交
912
  taosArrayClear(aColumnDataAgg);
H
Hongze Cheng 已提交
913

H
Hongze Cheng 已提交
914
  // alloc
H
Hongze Cheng 已提交
915
  int32_t size = pSmaInfo->size;
H
Hongze Cheng 已提交
916 917
  code = tRealloc(&pReader->aBuf[0], size);
  if (code) goto _err;
H
Hongze Cheng 已提交
918

H
Hongze Cheng 已提交
919
  // read
H
Hongze Cheng 已提交
920 921
  code = tsdbReadFile(pReader->pSmaFD, pSmaInfo->offset, pReader->aBuf[0], size);
  if (code) goto _err;
H
Hongze Cheng 已提交
922

H
Hongze Cheng 已提交
923
  // decode
H
Hongze Cheng 已提交
924
  int32_t n = 0;
H
Hongze Cheng 已提交
925 926 927
  while (n < pSmaInfo->size) {
    SColumnDataAgg sma;
    n += tGetColumnDataAgg(pReader->aBuf[0] + n, &sma);
H
Hongze Cheng 已提交
928

H
Hongze Cheng 已提交
929 930
    if (taosArrayPush(aColumnDataAgg, &sma) == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
H
Hongze Cheng 已提交
931 932
      goto _err;
    }
H
Hongze Cheng 已提交
933
  }
H
Hongze Cheng 已提交
934
  ASSERT(n == pSmaInfo->size);
H
Hongze Cheng 已提交
935
  return code;
H
Hongze Cheng 已提交
936

H
Hongze Cheng 已提交
937
_err:
H
Hongze Cheng 已提交
938
  tsdbError("vgId:%d tsdb read block sma failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
939 940
  return code;
}
H
Hongze Cheng 已提交
941

H
Hongze Cheng 已提交
942 943
static int32_t tsdbReadBlockDataImpl(SDataFReader *pReader, SBlockInfo *pBlkInfo, int8_t fromLast,
                                     SBlockData *pBlockData) {
H
Hongze Cheng 已提交
944
  int32_t code = 0;
H
Hongze Cheng 已提交
945

H
Hongze Cheng 已提交
946 947
  tBlockDataClear(pBlockData);

H
Hongze Cheng 已提交
948
  STsdbFD *pFD = fromLast ? pReader->aSstFD[0] : pReader->pDataFD;  // (todo)
H
Hongze Cheng 已提交
949 950

  // todo: realloc pReader->aBuf[0]
H
Hongze Cheng 已提交
951 952

  // uid + version + tskey
H
Hongze Cheng 已提交
953
  tsdbReadFile(pFD, pBlkInfo->offset, pReader->aBuf[0], pBlkInfo->szKey);  // todo
H
Hongze Cheng 已提交
954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972
  SDiskDataHdr hdr;
  uint8_t     *p = pReader->aBuf[0] + tGetDiskDataHdr(pReader->aBuf[0], &hdr);

  ASSERT(hdr.delimiter == TSDB_FILE_DLMT);
  ASSERT(pBlockData->suid == hdr.suid);
  ASSERT(pBlockData->uid == hdr.uid);

  pBlockData->nRow = hdr.nRow;

  // uid
  if (hdr.uid == 0) {
    ASSERT(hdr.szUid);
    code = tsdbDecmprData(p, hdr.szUid, TSDB_DATA_TYPE_BIGINT, hdr.cmprAlg, (uint8_t **)&pBlockData->aUid,
                          sizeof(int64_t) * hdr.nRow, &pReader->aBuf[1]);
    if (code) goto _err;
  } else {
    ASSERT(!hdr.szUid);
  }
  p += hdr.szUid;
H
Hongze Cheng 已提交
973

H
Hongze Cheng 已提交
974 975 976 977 978
  // version
  code = tsdbDecmprData(p, hdr.szVer, TSDB_DATA_TYPE_BIGINT, hdr.cmprAlg, (uint8_t **)&pBlockData->aVersion,
                        sizeof(int64_t) * hdr.nRow, &pReader->aBuf[1]);
  if (code) goto _err;
  p += hdr.szVer;
H
Hongze Cheng 已提交
979

H
Hongze Cheng 已提交
980 981 982
  // TSKEY
  code = tsdbDecmprData(p, hdr.szKey, TSDB_DATA_TYPE_TIMESTAMP, hdr.cmprAlg, (uint8_t **)&pBlockData->aTSKEY,
                        sizeof(TSKEY) * hdr.nRow, &pReader->aBuf[1]);
H
Hongze Cheng 已提交
983
  if (code) goto _err;
H
Hongze Cheng 已提交
984
  p += hdr.szKey;
H
Hongze Cheng 已提交
985

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

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

H
Hongze Cheng 已提交
991 992
  if (hdr.szBlkCol > 0) {
    int64_t offset = pBlkInfo->offset + pBlkInfo->szKey;
H
Hongze Cheng 已提交
993
    tsdbReadFile(pFD, offset, pReader->aBuf[0], hdr.szBlkCol + sizeof(TSCKSUM));
H
Hongze Cheng 已提交
994 995
  }

H
Hongze Cheng 已提交
996 997 998
  SBlockCol  blockCol = {.cid = 0};
  SBlockCol *pBlockCol = &blockCol;
  int32_t    n = 0;
H
Hongze Cheng 已提交
999

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

H
Hongze Cheng 已提交
1003 1004 1005 1006 1007 1008 1009
    while (pBlockCol && pBlockCol->cid < pColData->cid) {
      if (n < hdr.szBlkCol) {
        n += tGetBlockCol(pReader->aBuf[0] + n, pBlockCol);
      } else {
        ASSERT(n == hdr.szBlkCol);
        pBlockCol = NULL;
      }
H
Hongze Cheng 已提交
1010
    }
H
Hongze Cheng 已提交
1011

H
Hongze Cheng 已提交
1012 1013 1014 1015 1016 1017 1018 1019 1020
    if (pBlockCol == NULL || pBlockCol->cid > pColData->cid) {
      // add a lot of NONE
      for (int32_t iRow = 0; iRow < hdr.nRow; iRow++) {
        code = tColDataAppendValue(pColData, &COL_VAL_NONE(pColData->cid, pColData->type));
        if (code) goto _err;
      }
    } else {
      ASSERT(pBlockCol->type == pColData->type);
      ASSERT(pBlockCol->flag && pBlockCol->flag != HAS_NONE);
H
Hongze Cheng 已提交
1021

H
Hongze Cheng 已提交
1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032
      if (pBlockCol->flag == HAS_NULL) {
        // add a lot of NULL
        for (int32_t iRow = 0; iRow < hdr.nRow; iRow++) {
          code = tColDataAppendValue(pColData, &COL_VAL_NULL(pBlockCol->cid, pBlockCol->type));
          if (code) goto _err;
        }
      } else {
        // decode from binary
        int64_t offset = pBlkInfo->offset + pBlkInfo->szKey + hdr.szBlkCol + sizeof(TSCKSUM) + pBlockCol->offset;
        int32_t size = pBlockCol->szBitmap + pBlockCol->szOffset + pBlockCol->szValue + sizeof(TSCKSUM);

H
Hongze Cheng 已提交
1033
        tsdbReadFile(pFD, offset, pReader->aBuf[1], size);
H
Hongze Cheng 已提交
1034 1035 1036 1037 1038

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

H
Hongze Cheng 已提交
1041
_exit:
H
Hongze Cheng 已提交
1042 1043
  return code;

H
Hongze Cheng 已提交
1044
_err:
H
Hongze Cheng 已提交
1045
  tsdbError("vgId:%d tsdb read block data impl failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1046
  return code;
H
Hongze Cheng 已提交
1047
}
H
Hongze Cheng 已提交
1048

H
Hongze Cheng 已提交
1049 1050
int32_t tsdbReadDataBlock(SDataFReader *pReader, SDataBlk *pDataBlk, SBlockData *pBlockData) {
  int32_t code = 0;
H
Hongze Cheng 已提交
1051

H
Hongze Cheng 已提交
1052 1053
  code = tsdbReadBlockDataImpl(pReader, &pDataBlk->aSubBlock[0], 0, pBlockData);
  if (code) goto _err;
H
Hongze Cheng 已提交
1054

H
Hongze Cheng 已提交
1055 1056 1057
  if (pDataBlk->nSubBlock > 1) {
    SBlockData bData1;
    SBlockData bData2;
H
Hongze Cheng 已提交
1058

H
Hongze Cheng 已提交
1059 1060 1061 1062 1063
    // create
    code = tBlockDataCreate(&bData1);
    if (code) goto _err;
    code = tBlockDataCreate(&bData2);
    if (code) goto _err;
H
Hongze Cheng 已提交
1064

H
Hongze Cheng 已提交
1065 1066 1067
    // init
    tBlockDataInitEx(&bData1, pBlockData);
    tBlockDataInitEx(&bData2, pBlockData);
H
Hongze Cheng 已提交
1068

H
Hongze Cheng 已提交
1069 1070 1071 1072 1073 1074 1075
    for (int32_t iSubBlock = 1; iSubBlock < pDataBlk->nSubBlock; iSubBlock++) {
      code = tsdbReadBlockDataImpl(pReader, &pDataBlk->aSubBlock[iSubBlock], 0, &bData1);
      if (code) {
        tBlockDataDestroy(&bData1, 1);
        tBlockDataDestroy(&bData2, 1);
        goto _err;
      }
H
Hongze Cheng 已提交
1076

H
Hongze Cheng 已提交
1077 1078 1079 1080 1081 1082
      code = tBlockDataCopy(pBlockData, &bData2);
      if (code) {
        tBlockDataDestroy(&bData1, 1);
        tBlockDataDestroy(&bData2, 1);
        goto _err;
      }
H
Hongze Cheng 已提交
1083

H
Hongze Cheng 已提交
1084 1085 1086 1087 1088 1089 1090
      code = tBlockDataMerge(&bData1, &bData2, pBlockData);
      if (code) {
        tBlockDataDestroy(&bData1, 1);
        tBlockDataDestroy(&bData2, 1);
        goto _err;
      }
    }
H
Hongze Cheng 已提交
1091

H
Hongze Cheng 已提交
1092 1093
    tBlockDataDestroy(&bData1, 1);
    tBlockDataDestroy(&bData2, 1);
H
Hongze Cheng 已提交
1094 1095
  }

H
Hongze Cheng 已提交
1096
  return code;
H
Hongze Cheng 已提交
1097

H
Hongze Cheng 已提交
1098 1099 1100 1101
_err:
  tsdbError("vgId:%d tsdb read data block failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
  return code;
}
H
Hongze Cheng 已提交
1102

H
Hongze Cheng 已提交
1103 1104
int32_t tsdbReadSstBlock(SDataFReader *pReader, int32_t iSst, SSstBlk *pSstBlk, SBlockData *pBlockData) {
  int32_t code = 0;
H
Hongze Cheng 已提交
1105

H
Hongze Cheng 已提交
1106 1107
  code = tsdbReadBlockDataImpl(pReader, &pSstBlk->bInfo, 1, pBlockData);
  if (code) goto _err;
H
Hongze Cheng 已提交
1108

H
Hongze Cheng 已提交
1109
  return code;
H
Hongze Cheng 已提交
1110

H
Hongze Cheng 已提交
1111 1112 1113 1114
_err:
  tsdbError("vgId:%d tsdb read last block failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
  return code;
}
H
Hongze Cheng 已提交
1115

H
Hongze Cheng 已提交
1116 1117
int32_t tsdbReadSstBlockEx(SDataFReader *pReader, int32_t iSst, SSstBlk *pSstBlk, SBlockData *pBlockData) {
  int32_t code = 0;
H
Hongze Cheng 已提交
1118

H
Hongze Cheng 已提交
1119
  // read
H
Hongze Cheng 已提交
1120
  tsdbReadFile(pReader->aSstFD[iSst], pSstBlk->bInfo.offset, pReader->aBuf[0], pSstBlk->bInfo.szBlock);
H
Hongze Cheng 已提交
1121

H
Hongze Cheng 已提交
1122 1123 1124
  // decmpr
  code = tDecmprBlockData(pReader->aBuf[0], pSstBlk->bInfo.szBlock, pBlockData, &pReader->aBuf[1]);
  if (code) goto _exit;
H
Hongze Cheng 已提交
1125

H
Hongze Cheng 已提交
1126
_exit:
H
Hongze Cheng 已提交
1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 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 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 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
  return code;
}

// SDelFWriter ====================================================
int32_t tsdbDelFWriterOpen(SDelFWriter **ppWriter, SDelFile *pFile, STsdb *pTsdb) {
  int32_t      code = 0;
  char         fname[TSDB_FILENAME_LEN];
  char         hdr[TSDB_FHDR_SIZE] = {0};
  SDelFWriter *pDelFWriter;
  int64_t      n;

  // alloc
  pDelFWriter = (SDelFWriter *)taosMemoryCalloc(1, sizeof(*pDelFWriter));
  if (pDelFWriter == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
  pDelFWriter->pTsdb = pTsdb;
  pDelFWriter->fDel = *pFile;

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

  // update header
  n = taosWriteFile(pDelFWriter->pWriteH, &hdr, TSDB_FHDR_SIZE);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  pDelFWriter->fDel.size = TSDB_FHDR_SIZE;
  pDelFWriter->fDel.offset = 0;

  *ppWriter = pDelFWriter;
  return code;

_err:
  tsdbError("vgId:%d, failed to open del file writer since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
  *ppWriter = NULL;
  return code;
}

int32_t tsdbDelFWriterClose(SDelFWriter **ppWriter, int8_t sync) {
  int32_t      code = 0;
  SDelFWriter *pWriter = *ppWriter;
  STsdb       *pTsdb = pWriter->pTsdb;

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

  for (int32_t iBuf = 0; iBuf < sizeof(pWriter->aBuf) / sizeof(uint8_t *); iBuf++) {
    tFree(pWriter->aBuf[iBuf]);
  }
  taosMemoryFree(pWriter);

  *ppWriter = NULL;
  return code;

_err:
  tsdbError("vgId:%d, failed to close del file writer since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
  return code;
}

int32_t tsdbWriteDelData(SDelFWriter *pWriter, SArray *aDelData, SDelIdx *pDelIdx) {
  int32_t code = 0;
  int64_t size;
  int64_t n;

  // prepare
  size = sizeof(uint32_t);
  for (int32_t iDelData = 0; iDelData < taosArrayGetSize(aDelData); iDelData++) {
    size += tPutDelData(NULL, taosArrayGet(aDelData, iDelData));
  }
  size += sizeof(TSCKSUM);

  // alloc
  code = tRealloc(&pWriter->aBuf[0], size);
  if (code) goto _err;

  // build
  n = 0;
  n += tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT);
  for (int32_t iDelData = 0; iDelData < taosArrayGetSize(aDelData); iDelData++) {
    n += tPutDelData(pWriter->aBuf[0] + n, taosArrayGet(aDelData, iDelData));
  }
  taosCalcChecksumAppend(0, pWriter->aBuf[0], size);

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

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

  ASSERT(n == size);

  // update
  pDelIdx->offset = pWriter->fDel.size;
  pDelIdx->size = size;
  pWriter->fDel.size += size;

  return code;

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

int32_t tsdbWriteDelIdx(SDelFWriter *pWriter, SArray *aDelIdx) {
  int32_t  code = 0;
  int64_t  size;
  int64_t  n;
  SDelIdx *pDelIdx;

  // prepare
  size = sizeof(uint32_t);
  for (int32_t iDelIdx = 0; iDelIdx < taosArrayGetSize(aDelIdx); iDelIdx++) {
    size += tPutDelIdx(NULL, taosArrayGet(aDelIdx, iDelIdx));
  }
  size += sizeof(TSCKSUM);

  // alloc
  code = tRealloc(&pWriter->aBuf[0], size);
  if (code) goto _err;

  // build
  n = 0;
  n += tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT);
  for (int32_t iDelIdx = 0; iDelIdx < taosArrayGetSize(aDelIdx); iDelIdx++) {
    n += tPutDelIdx(pWriter->aBuf[0] + n, taosArrayGet(aDelIdx, iDelIdx));
  }
  taosCalcChecksumAppend(0, pWriter->aBuf[0], size);

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

  // write
  n = taosWriteFile(pWriter->pWriteH, pWriter->aBuf[0], size);
H
Hongze Cheng 已提交
1279 1280 1281 1282 1283
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
1284 1285 1286
  // update
  pWriter->fDel.offset = pWriter->fDel.size;
  pWriter->fDel.size += size;
H
Hongze Cheng 已提交
1287

H
Hongze Cheng 已提交
1288
  return code;
H
Hongze Cheng 已提交
1289

H
Hongze Cheng 已提交
1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307
_err:
  tsdbError("vgId:%d, write del idx failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
  return code;
}

int32_t tsdbUpdateDelFileHdr(SDelFWriter *pWriter) {
  int32_t code = 0;
  char    hdr[TSDB_FHDR_SIZE];
  int64_t size = TSDB_FHDR_SIZE;
  int64_t n;

  // build
  memset(hdr, 0, size);
  tPutDelFile(hdr, &pWriter->fDel);
  taosCalcChecksumAppend(0, hdr, size);

  // seek
  if (taosLSeekFile(pWriter->pWriteH, 0, SEEK_SET) < 0) {
H
Hongze Cheng 已提交
1308 1309 1310 1311
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

H
Hongze Cheng 已提交
1312 1313
  // write
  n = taosWriteFile(pWriter->pWriteH, hdr, size);
H
Hongze Cheng 已提交
1314 1315 1316 1317 1318 1319 1320 1321
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  return code;

_err:
H
Hongze Cheng 已提交
1322
  tsdbError("vgId:%d, update del file hdr failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1323
  return code;
H
Hongze Cheng 已提交
1324
}
H
Hongze Cheng 已提交
1325 1326 1327 1328 1329
// SDelFReader ====================================================
struct SDelFReader {
  STsdb    *pTsdb;
  SDelFile  fDel;
  TdFilePtr pReadH;
H
Hongze Cheng 已提交
1330

H
Hongze Cheng 已提交
1331 1332
  uint8_t *aBuf[1];
};
H
Hongze Cheng 已提交
1333

H
Hongze Cheng 已提交
1334 1335 1336 1337 1338
int32_t tsdbDelFReaderOpen(SDelFReader **ppReader, SDelFile *pFile, STsdb *pTsdb) {
  int32_t      code = 0;
  char         fname[TSDB_FILENAME_LEN];
  SDelFReader *pDelFReader;
  int64_t      n;
H
Hongze Cheng 已提交
1339

H
Hongze Cheng 已提交
1340 1341 1342
  // alloc
  pDelFReader = (SDelFReader *)taosMemoryCalloc(1, sizeof(*pDelFReader));
  if (pDelFReader == NULL) {
H
Hongze Cheng 已提交
1343
    code = TSDB_CODE_OUT_OF_MEMORY;
H
Hongze Cheng 已提交
1344 1345 1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356
    goto _err;
  }

  // open impl
  pDelFReader->pTsdb = pTsdb;
  pDelFReader->fDel = *pFile;

  tsdbDelFileName(pTsdb, pFile, fname);
  pDelFReader->pReadH = taosOpenFile(fname, TD_FILE_READ);
  if (pDelFReader->pReadH == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    taosMemoryFree(pDelFReader);
    goto _err;
H
Hongze Cheng 已提交
1357 1358 1359
  }

_exit:
H
Hongze Cheng 已提交
1360
  *ppReader = pDelFReader;
H
Hongze Cheng 已提交
1361 1362
  return code;

H
Hongze Cheng 已提交
1363 1364 1365 1366
_err:
  tsdbError("vgId:%d, del file reader open failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
  *ppReader = NULL;
  return code;
H
Hongze Cheng 已提交
1367 1368
}

H
Hongze Cheng 已提交
1369 1370 1371
int32_t tsdbDelFReaderClose(SDelFReader **ppReader) {
  int32_t      code = 0;
  SDelFReader *pReader = *ppReader;
H
Hongze Cheng 已提交
1372

H
Hongze Cheng 已提交
1373 1374 1375 1376 1377 1378 1379 1380 1381
  if (pReader) {
    if (taosCloseFile(&pReader->pReadH) < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      goto _exit;
    }
    for (int32_t iBuf = 0; iBuf < sizeof(pReader->aBuf) / sizeof(uint8_t *); iBuf++) {
      tFree(pReader->aBuf[iBuf]);
    }
    taosMemoryFree(pReader);
H
Hongze Cheng 已提交
1382
  }
H
Hongze Cheng 已提交
1383
  *ppReader = NULL;
H
Hongze Cheng 已提交
1384 1385 1386 1387 1388

_exit:
  return code;
}

H
Hongze Cheng 已提交
1389
int32_t tsdbReadDelData(SDelFReader *pReader, SDelIdx *pDelIdx, SArray *aDelData) {
H
Hongze Cheng 已提交
1390
  int32_t code = 0;
H
Hongze Cheng 已提交
1391 1392 1393
  int64_t offset = pDelIdx->offset;
  int64_t size = pDelIdx->size;
  int64_t n;
H
Hongze Cheng 已提交
1394

H
Hongze Cheng 已提交
1395
  taosArrayClear(aDelData);
H
Hongze Cheng 已提交
1396

H
Hongze Cheng 已提交
1397 1398 1399 1400 1401
  // seek
  if (taosLSeekFile(pReader->pReadH, offset, SEEK_SET) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
H
Hongze Cheng 已提交
1402

H
Hongze Cheng 已提交
1403 1404 1405
  // alloc
  code = tRealloc(&pReader->aBuf[0], size);
  if (code) goto _err;
H
Hongze Cheng 已提交
1406

H
Hongze Cheng 已提交
1407 1408 1409 1410 1411 1412 1413 1414 1415
  // read
  n = taosReadFile(pReader->pReadH, pReader->aBuf[0], 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 已提交
1416

H
Hongze Cheng 已提交
1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434
  // check
  if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) {
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _err;
  }

  // // decode
  n = 0;

  uint32_t delimiter;
  n += tGetU32(pReader->aBuf[0] + n, &delimiter);
  while (n < size - sizeof(TSCKSUM)) {
    SDelData delData;
    n += tGetDelData(pReader->aBuf[0] + n, &delData);

    if (taosArrayPush(aDelData, &delData) == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
H
Hongze Cheng 已提交
1435 1436 1437
    }
  }

H
Hongze Cheng 已提交
1438 1439 1440 1441 1442 1443
  ASSERT(n == size - sizeof(TSCKSUM));

  return code;

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

H
Hongze Cheng 已提交
1447
int32_t tsdbReadDelIdx(SDelFReader *pReader, SArray *aDelIdx) {
H
Hongze Cheng 已提交
1448
  int32_t code = 0;
H
Hongze Cheng 已提交
1449 1450 1451
  int32_t n;
  int64_t offset = pReader->fDel.offset;
  int64_t size = pReader->fDel.size - offset;
H
Hongze Cheng 已提交
1452

H
Hongze Cheng 已提交
1453 1454 1455 1456
  taosArrayClear(aDelIdx);

  // seek
  if (taosLSeekFile(pReader->pReadH, offset, SEEK_SET) < 0) {
H
Hongze Cheng 已提交
1457
    code = TAOS_SYSTEM_ERROR(errno);
H
Hongze Cheng 已提交
1458
    goto _err;
H
Hongze Cheng 已提交
1459 1460
  }

H
Hongze Cheng 已提交
1461 1462 1463 1464 1465 1466
  // alloc
  code = tRealloc(&pReader->aBuf[0], size);
  if (code) goto _err;

  // read
  n = taosReadFile(pReader->pReadH, pReader->aBuf[0], size);
H
Hongze Cheng 已提交
1467 1468
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
H
Hongze Cheng 已提交
1469 1470
    goto _err;
  } else if (n < size) {
H
Hongze Cheng 已提交
1471
    code = TSDB_CODE_FILE_CORRUPTED;
H
Hongze Cheng 已提交
1472
    goto _err;
H
Hongze Cheng 已提交
1473 1474
  }

H
Hongze Cheng 已提交
1475 1476
  // check
  if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) {
H
Hongze Cheng 已提交
1477
    code = TSDB_CODE_FILE_CORRUPTED;
H
Hongze Cheng 已提交
1478
    goto _err;
H
Hongze Cheng 已提交
1479 1480
  }

H
Hongze Cheng 已提交
1481 1482 1483 1484 1485
  // decode
  n = 0;
  uint32_t delimiter;
  n += tGetU32(pReader->aBuf[0] + n, &delimiter);
  ASSERT(delimiter == TSDB_FILE_DLMT);
H
Hongze Cheng 已提交
1486

H
Hongze Cheng 已提交
1487 1488
  while (n < size - sizeof(TSCKSUM)) {
    SDelIdx delIdx;
H
Hongze Cheng 已提交
1489

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

H
Hongze Cheng 已提交
1492 1493 1494 1495
    if (taosArrayPush(aDelIdx, &delIdx) == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
    }
H
Hongze Cheng 已提交
1496 1497
  }

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

H
Hongze Cheng 已提交
1500
  return code;
H
Hongze Cheng 已提交
1501

H
Hongze Cheng 已提交
1502 1503
_err:
  tsdbError("vgId:%d, read del idx failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
H
Hongze Cheng 已提交
1504
  return code;
H
Hongze Cheng 已提交
1505
}
H
Hongze Cheng 已提交
1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537 1538 1539

static int32_t tsdbReadAndCheck(TdFilePtr pFD, int64_t offset, uint8_t **ppOut, int32_t size, int8_t toCheck) {
  int32_t code = 0;

  // alloc
  code = tRealloc(ppOut, size);
  if (code) goto _exit;

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

  // read
  n = taosReadFile(pFD, *ppOut, size);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _exit;
  } else if (n < size) {
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _exit;
  }

  // check
  if (toCheck && !taosCheckChecksumWhole(*ppOut, size)) {
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _exit;
  }

_exit:
  return code;
}