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

H
Haojun Liao 已提交
16
#include "osDef.h"
H
Hongze Cheng 已提交
17
#include "tsdb.h"
18

H
Hongze Cheng 已提交
19
#define ASCENDING_TRAVERSE(o) (o == TSDB_ORDER_ASC)
H
Hongze Cheng 已提交
20

21 22 23 24 25 26
typedef enum {
  EXTERNAL_ROWS_PREV = 0x1,
  EXTERNAL_ROWS_MAIN = 0x2,
  EXTERNAL_ROWS_NEXT = 0x3,
} EContentData;

27
typedef struct {
dengyihao's avatar
dengyihao 已提交
28
  STbDataIter* iter;
29 30 31 32
  int32_t      index;
  bool         hasVal;
} SIterInfo;

33 34
typedef struct {
  int32_t numOfBlocks;
35
  int32_t numOfLastFiles;
36 37
} SBlockNumber;

H
Haojun Liao 已提交
38
typedef struct STableBlockScanInfo {
dengyihao's avatar
dengyihao 已提交
39 40
  uint64_t  uid;
  TSKEY     lastKey;
H
Hongze Cheng 已提交
41 42 43 44 45 46 47 48
  SMapData  mapData;            // block info (compressed)
  SArray*   pBlockList;         // block data index list
  SIterInfo iter;               // mem buffer skip list iterator
  SIterInfo iiter;              // imem buffer skip list iterator
  SArray*   delSkyline;         // delete info for this table
  int32_t   fileDelIndex;       // file block delete index
  int32_t   lastBlockDelIndex;  // delete index for last block
  bool      iterInit;           // whether to initialize the in-memory skip list iterator or not
H
Haojun Liao 已提交
49 50 51
} STableBlockScanInfo;

typedef struct SBlockOrderWrapper {
dengyihao's avatar
dengyihao 已提交
52
  int64_t uid;
53
  int64_t offset;
H
Haojun Liao 已提交
54
} SBlockOrderWrapper;
H
Hongze Cheng 已提交
55 56

typedef struct SBlockOrderSupporter {
57 58 59 60
  SBlockOrderWrapper** pDataBlockInfo;
  int32_t*             indexPerTable;
  int32_t*             numOfBlocksPerTable;
  int32_t              numOfTables;
H
Hongze Cheng 已提交
61 62 63
} SBlockOrderSupporter;

typedef struct SIOCostSummary {
64 65 66
  int64_t numOfBlocks;
  double  blockLoadTime;
  double  buildmemBlock;
67
  int64_t headFileLoad;
68
  double  headFileLoadTime;
69
  int64_t smaDataLoad;
70
  double  smaLoadTime;
71 72
  int64_t lastBlockLoad;
  double  lastBlockLoadTime;
H
Haojun Liao 已提交
73 74
  int64_t composedBlocks;
  double  buildComposedBlockTime;
H
Hongze Cheng 已提交
75 76 77
} SIOCostSummary;

typedef struct SBlockLoadSuppInfo {
78
  SArray*          pColAgg;
79
  SColumnDataAgg   tsColAgg;
C
Cary Xu 已提交
80
  SColumnDataAgg** plist;
81
  int16_t*         colIds;  // column ids for loading file block data
82
  int32_t          numOfCols;
83
  char**           buildBuf;  // build string tmp buffer, todo remove it later after all string format being updated.
H
Hongze Cheng 已提交
84 85
} SBlockLoadSuppInfo;

86
typedef struct SLastBlockReader {
H
Hongze Cheng 已提交
87 88 89 90 91
  STimeWindow        window;
  SVersionRange      verRange;
  int32_t            order;
  uint64_t           uid;
  SMergeTree         mergeTree;
92
  SSttBlockLoadInfo* pInfo;
93 94
} SLastBlockReader;

95
typedef struct SFilesetIter {
H
Hongze Cheng 已提交
96 97 98
  int32_t           numOfFiles;  // number of total files
  int32_t           index;       // current accessed index in the list
  SArray*           pFileList;   // data file list
99
  int32_t           order;
H
Hongze Cheng 已提交
100
  SLastBlockReader* pLastBlockReader;  // last file block reader
101
} SFilesetIter;
H
Haojun Liao 已提交
102 103

typedef struct SFileDataBlockInfo {
104
  // index position in STableBlockScanInfo in order to check whether neighbor block overlaps with it
dengyihao's avatar
dengyihao 已提交
105
  uint64_t uid;
106
  int32_t  tbBlockIdx;
H
Haojun Liao 已提交
107 108 109
} SFileDataBlockInfo;

typedef struct SDataBlockIter {
110
  int32_t   numOfBlocks;
111
  int32_t   index;
H
Hongze Cheng 已提交
112
  SArray*   blockList;  // SArray<SFileDataBlockInfo>
113
  int32_t   order;
H
Haojun Liao 已提交
114
  SDataBlk  block;      // current SDataBlk data
115
  SHashObj* pTableMap;
H
Haojun Liao 已提交
116 117 118
} SDataBlockIter;

typedef struct SFileBlockDumpInfo {
dengyihao's avatar
dengyihao 已提交
119 120 121 122
  int32_t totalRows;
  int32_t rowIndex;
  int64_t lastKey;
  bool    allDumped;
H
Haojun Liao 已提交
123 124
} SFileBlockDumpInfo;

125
typedef struct SUidOrderCheckInfo {
126 127
  uint64_t* tableUidList;  // access table uid list in uid ascending order list
  int32_t   currentIndex;  // index in table uid list
128 129
} SUidOrderCheckInfo;

H
Haojun Liao 已提交
130
typedef struct SReaderStatus {
131
  bool                 loadFromFile;       // check file stage
132
  bool                 composedDataBlock;  // the returned data block is a composed block or not
133 134
  SHashObj*            pTableMap;          // SHash<STableBlockScanInfo>
  STableBlockScanInfo* pTableIter;         // table iterator used in building in-memory buffer data blocks.
135
  SUidOrderCheckInfo   uidCheckInfo;       // check all table in uid order
136
  SFileBlockDumpInfo   fBlockDumpInfo;
137 138 139 140
  SDFileSet*           pCurrentFileset;  // current opened file set
  SBlockData           fileBlockData;
  SFilesetIter         fileIter;
  SDataBlockIter       blockIter;
H
Haojun Liao 已提交
141 142
} SReaderStatus;

H
Hongze Cheng 已提交
143
struct STsdbReader {
H
Haojun Liao 已提交
144 145 146 147 148 149 150
  STsdb*             pTsdb;
  uint64_t           suid;
  int16_t            order;
  STimeWindow        window;  // the primary query time window that applies to all queries
  SSDataBlock*       pResBlock;
  int32_t            capacity;
  SReaderStatus      status;
151 152
  char*              idStr;  // query info handle, for debug purpose
  int32_t            type;   // query type: 1. retrieve all data blocks, 2. retrieve direct prev|next rows
H
Hongze Cheng 已提交
153
  SBlockLoadSuppInfo suppInfo;
H
Hongze Cheng 已提交
154
  STsdbReadSnap*     pReadSnap;
155
  SIOCostSummary     cost;
156 157
  STSchema*          pSchema;     // the newest version schema
  STSchema*          pMemSchema;  // the previous schema for in-memory data, to avoid load schema too many times
158 159
  SDataFReader*      pFileReader;
  SVersionRange      verRange;
160

161 162
  int32_t      step;
  STsdbReader* innerReader[2];
H
Hongze Cheng 已提交
163
};
H
Hongze Cheng 已提交
164

H
Haojun Liao 已提交
165
static SFileDataBlockInfo* getCurrentBlockInfo(SDataBlockIter* pBlockIter);
166 167
static int      buildDataBlockFromBufImpl(STableBlockScanInfo* pBlockScanInfo, int64_t endKey, int32_t capacity,
                                          STsdbReader* pReader);
168
static TSDBROW* getValidMemRow(SIterInfo* pIter, const SArray* pDelList, STsdbReader* pReader);
169 170
static int32_t  doMergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pScanInfo, STsdbReader* pReader,
                                        SRowMerger* pMerger);
H
Hongze Cheng 已提交
171 172
static int32_t  doMergeRowsInLastBlock(SLastBlockReader* pLastBlockReader, STableBlockScanInfo* pScanInfo, int64_t ts,
                                       SRowMerger* pMerger);
173
static int32_t  doMergeRowsInBuf(SIterInfo* pIter, uint64_t uid, int64_t ts, SArray* pDelList, SRowMerger* pMerger,
dengyihao's avatar
dengyihao 已提交
174
                                 STsdbReader* pReader);
H
Haojun Liao 已提交
175
static int32_t  doAppendRowFromTSRow(SSDataBlock* pBlock, STsdbReader* pReader, STSRow* pTSRow, uint64_t uid);
176
static int32_t  doAppendRowFromFileBlock(SSDataBlock* pResBlock, STsdbReader* pReader, SBlockData* pBlockData,
H
Hongze Cheng 已提交
177
                                         int32_t rowIndex);
178
static void     setComposedBlockFlag(STsdbReader* pReader, bool composed);
179
static bool     hasBeenDropped(const SArray* pDelList, int32_t* index, TSDBKEY* pKey, int32_t order);
180

H
Hongze Cheng 已提交
181 182 183 184
static int32_t doMergeMemTableMultiRows(TSDBROW* pRow, uint64_t uid, SIterInfo* pIter, SArray* pDelList,
                                        STSRow** pTSRow, STsdbReader* pReader, bool* freeTSRow);
static int32_t doMergeMemIMemRows(TSDBROW* pRow, TSDBROW* piRow, STableBlockScanInfo* pBlockScanInfo,
                                  STsdbReader* pReader, STSRow** pTSRow);
185 186
static int32_t mergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pBlockScanInfo, int64_t key,
                                     STsdbReader* pReader);
187

dengyihao's avatar
dengyihao 已提交
188 189 190 191
static int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader, STbData* pMemTbData,
                                      STbData* piMemTbData);
static STsdb*  getTsdbByRetentions(SVnode* pVnode, TSKEY winSKey, SRetention* retentions, const char* idstr,
                                   int8_t* pLevel);
192
static SVersionRange getQueryVerRange(SVnode* pVnode, SQueryTableDataCond* pCond, int8_t level);
H
Hongze Cheng 已提交
193 194 195
static int64_t       getCurrentKeyInLastBlock(SLastBlockReader* pLastBlockReader);
static bool          hasDataInLastBlock(SLastBlockReader* pLastBlockReader);
static int32_t       doBuildDataBlock(STsdbReader* pReader);
H
Haojun Liao 已提交
196
static TSDBKEY       getCurrentKeyInBuf(STableBlockScanInfo* pScanInfo, STsdbReader* pReader);
197 198
static bool          hasDataInFileBlock(const SBlockData* pBlockData, const SFileBlockDumpInfo* pDumpInfo);
static bool          hasDataInLastBlock(SLastBlockReader* pLastBlockReader);
H
Haojun Liao 已提交
199

200 201
static bool outOfTimeWindow(int64_t ts, STimeWindow* pWindow) { return (ts > pWindow->ekey) || (ts < pWindow->skey); }

202 203 204
static int32_t setColumnIdSlotList(STsdbReader* pReader, SSDataBlock* pBlock) {
  SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo;

205
  size_t numOfCols = blockDataGetNumOfCols(pBlock);
206

207
  pSupInfo->numOfCols = numOfCols;
208
  pSupInfo->colIds = taosMemoryMalloc(numOfCols * sizeof(int16_t));
209
  pSupInfo->buildBuf = taosMemoryCalloc(numOfCols, POINTER_BYTES);
210 211 212
  if (pSupInfo->buildBuf == NULL || pSupInfo->colIds == NULL) {
    taosMemoryFree(pSupInfo->colIds);
    taosMemoryFree(pSupInfo->buildBuf);
H
Haojun Liao 已提交
213 214
    return TSDB_CODE_OUT_OF_MEMORY;
  }
H
Hongze Cheng 已提交
215

H
Haojun Liao 已提交
216 217
  for (int32_t i = 0; i < numOfCols; ++i) {
    SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, i);
218
    pSupInfo->colIds[i] = pCol->info.colId;
219 220 221 222

    if (IS_VAR_DATA_TYPE(pCol->info.type)) {
      pSupInfo->buildBuf[i] = taosMemoryMalloc(pCol->info.bytes);
    }
H
Haojun Liao 已提交
223
  }
H
Hongze Cheng 已提交
224

H
Haojun Liao 已提交
225 226
  return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
227

228
static SHashObj* createDataBlockScanInfo(STsdbReader* pTsdbReader, const STableKeyInfo* idList, int32_t numOfTables) {
H
Haojun Liao 已提交
229
  // allocate buffer in order to load data blocks from file
230
  // todo use simple hash instead, optimize the memory consumption
231 232 233
  SHashObj* pTableMap =
      taosHashInit(numOfTables, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
  if (pTableMap == NULL) {
H
Haojun Liao 已提交
234 235 236
    return NULL;
  }

237
  for (int32_t j = 0; j < numOfTables; ++j) {
238
    STableBlockScanInfo info = {.lastKey = 0, .uid = idList[j].uid};
239
    if (ASCENDING_TRAVERSE(pTsdbReader->order)) {
H
Haojun Liao 已提交
240
      int64_t skey = pTsdbReader->window.skey;
H
Hongze Cheng 已提交
241
      info.lastKey = (skey > INT64_MIN) ? (skey - 1) : skey;
wmmhello's avatar
wmmhello 已提交
242
    } else {
H
Haojun Liao 已提交
243
      int64_t ekey = pTsdbReader->window.ekey;
H
Hongze Cheng 已提交
244
      info.lastKey = (ekey < INT64_MAX) ? (ekey + 1) : ekey;
H
Haojun Liao 已提交
245
    }
wmmhello's avatar
wmmhello 已提交
246

247 248 249
    taosHashPut(pTableMap, &info.uid, sizeof(uint64_t), &info, sizeof(info));
    tsdbDebug("%p check table uid:%" PRId64 " from lastKey:%" PRId64 " %s", pTsdbReader, info.uid, info.lastKey,
              pTsdbReader->idStr);
H
Haojun Liao 已提交
250 251
  }

252 253
  tsdbDebug("%p create %d tables scan-info, size:%.2f Kb, %s", pTsdbReader, numOfTables,
            (sizeof(STableBlockScanInfo) * numOfTables) / 1024.0, pTsdbReader->idStr);
254

255
  return pTableMap;
H
Hongze Cheng 已提交
256
}
H
Hongze Cheng 已提交
257

H
Haojun Liao 已提交
258
static void resetDataBlockScanInfo(SHashObj* pTableMap, int64_t ts) {
259 260
  STableBlockScanInfo* p = NULL;

dengyihao's avatar
dengyihao 已提交
261
  while ((p = taosHashIterate(pTableMap, p)) != NULL) {
262 263
    p->iterInit = false;
    p->iiter.hasVal = false;
dengyihao's avatar
dengyihao 已提交
264
    if (p->iter.iter != NULL) {
265
      p->iter.iter = tsdbTbDataIterDestroy(p->iter.iter);
266 267
    }

268
    p->delSkyline = taosArrayDestroy(p->delSkyline);
H
Haojun Liao 已提交
269
    p->lastKey = ts;
270 271 272
  }
}

273 274 275 276 277 278 279 280
static void destroyBlockScanInfo(SHashObj* pTableMap) {
  STableBlockScanInfo* p = NULL;

  while ((p = taosHashIterate(pTableMap, p)) != NULL) {
    p->iterInit = false;
    p->iiter.hasVal = false;

    if (p->iter.iter != NULL) {
281
      p->iter.iter = tsdbTbDataIterDestroy(p->iter.iter);
282 283 284
    }

    if (p->iiter.iter != NULL) {
285
      p->iiter.iter = tsdbTbDataIterDestroy(p->iiter.iter);
286 287
    }

288 289
    p->delSkyline = taosArrayDestroy(p->delSkyline);
    p->pBlockList = taosArrayDestroy(p->pBlockList);
290
    tMapDataClear(&p->mapData);
291 292 293 294 295
  }

  taosHashCleanup(pTableMap);
}

296
static bool isEmptyQueryTimeWindow(STimeWindow* pWindow) {
297 298
  ASSERT(pWindow != NULL);
  return pWindow->skey > pWindow->ekey;
H
Haojun Liao 已提交
299
}
H
Hongze Cheng 已提交
300

301 302 303
// Update the query time window according to the data time to live(TTL) information, in order to avoid to return
// the expired data to client, even it is queried already.
static STimeWindow updateQueryTimeWindow(STsdb* pTsdb, STimeWindow* pWindow) {
dengyihao's avatar
dengyihao 已提交
304
  STsdbKeepCfg* pCfg = &pTsdb->keepCfg;
H
Hongze Cheng 已提交
305

306
  int64_t now = taosGetTimestamp(pCfg->precision);
dengyihao's avatar
dengyihao 已提交
307
  int64_t earilyTs = now - (tsTickPerMin[pCfg->precision] * pCfg->keep2) + 1;  // needs to add one tick
308

dengyihao's avatar
dengyihao 已提交
309
  STimeWindow win = *pWindow;
310 311 312 313 314 315
  if (win.skey < earilyTs) {
    win.skey = earilyTs;
  }

  return win;
}
H
Hongze Cheng 已提交
316

H
Haojun Liao 已提交
317
static void limitOutputBufferSize(const SQueryTableDataCond* pCond, int32_t* capacity) {
H
Haojun Liao 已提交
318 319 320 321 322 323
  int32_t rowLen = 0;
  for (int32_t i = 0; i < pCond->numOfCols; ++i) {
    rowLen += pCond->colList[i].bytes;
  }

  // make sure the output SSDataBlock size be less than 2MB.
H
Haojun Liao 已提交
324 325 326
  const int32_t TWOMB = 2 * 1024 * 1024;
  if ((*capacity) * rowLen > TWOMB) {
    (*capacity) = TWOMB / rowLen;
H
Haojun Liao 已提交
327 328 329 330
  }
}

// init file iterator
331
static int32_t initFilesetIterator(SFilesetIter* pIter, SArray* aDFileSet, STsdbReader* pReader) {
H
Hongze Cheng 已提交
332
  size_t numOfFileset = taosArrayGetSize(aDFileSet);
333

334 335
  pIter->index = ASCENDING_TRAVERSE(pReader->order) ? -1 : numOfFileset;
  pIter->order = pReader->order;
H
Hongze Cheng 已提交
336
  pIter->pFileList = aDFileSet;
337
  pIter->numOfFiles = numOfFileset;
H
Haojun Liao 已提交
338

339 340 341 342
  if (pIter->pLastBlockReader == NULL) {
    pIter->pLastBlockReader = taosMemoryCalloc(1, sizeof(struct SLastBlockReader));
    if (pIter->pLastBlockReader == NULL) {
      int32_t code = TSDB_CODE_OUT_OF_MEMORY;
343
      tsdbError("failed to prepare the last block iterator, since:%s %s", tstrerror(code), pReader->idStr);
344 345
      return code;
    }
346 347
  }

348 349 350 351 352 353 354 355
  SLastBlockReader* pLReader = pIter->pLastBlockReader;
  pLReader->order = pReader->order;
  pLReader->window = pReader->window;
  pLReader->verRange = pReader->verRange;

  pLReader->uid = 0;
  tMergeTreeClose(&pLReader->mergeTree);

356
  if (pLReader->pInfo == NULL) {
357
    // here we ignore the first column, which is always be the primary timestamp column
358 359
    pLReader->pInfo =
        tCreateLastBlockLoadInfo(pReader->pSchema, &pReader->suppInfo.colIds[1], pReader->suppInfo.numOfCols - 1);
H
Haojun Liao 已提交
360 361 362 363
    if (pLReader->pInfo == NULL) {
      tsdbDebug("init fileset iterator failed, code:%s %s", tstrerror(terrno), pReader->idStr);
      return terrno;
    }
364 365
  }

366
  tsdbDebug("init fileset iterator, total files:%d %s", pIter->numOfFiles, pReader->idStr);
H
Haojun Liao 已提交
367 368 369
  return TSDB_CODE_SUCCESS;
}

370
static bool filesetIteratorNext(SFilesetIter* pIter, STsdbReader* pReader) {
371 372
  bool    asc = ASCENDING_TRAVERSE(pIter->order);
  int32_t step = asc ? 1 : -1;
373 374 375
  pIter->index += step;

  if ((asc && pIter->index >= pIter->numOfFiles) || ((!asc) && pIter->index < 0)) {
H
Haojun Liao 已提交
376 377 378
    return false;
  }

H
Haojun Liao 已提交
379 380 381
  SIOCostSummary* pSum = &pReader->cost;
  getLastBlockLoadInfo(pIter->pLastBlockReader->pInfo, &pSum->lastBlockLoad, &pReader->cost.lastBlockLoadTime);

382 383
  pIter->pLastBlockReader->uid = 0;
  tMergeTreeClose(&pIter->pLastBlockReader->mergeTree);
384
  resetLastBlockLoadInfo(pIter->pLastBlockReader->pInfo);
385

H
Haojun Liao 已提交
386 387
  // check file the time range of coverage
  STimeWindow win = {0};
H
Hongze Cheng 已提交
388

389
  while (1) {
H
Haojun Liao 已提交
390 391 392
    if (pReader->pFileReader != NULL) {
      tsdbDataFReaderClose(&pReader->pFileReader);
    }
393

394
    pReader->status.pCurrentFileset = (SDFileSet*)taosArrayGet(pIter->pFileList, pIter->index);
H
Haojun Liao 已提交
395

396 397 398 399
    int32_t code = tsdbDataFReaderOpen(&pReader->pFileReader, pReader->pTsdb, pReader->status.pCurrentFileset);
    if (code != TSDB_CODE_SUCCESS) {
      goto _err;
    }
H
Haojun Liao 已提交
400

401 402
    pReader->cost.headFileLoad += 1;

403 404 405 406 407 408 409 410 411 412 413 414
    int32_t fid = pReader->status.pCurrentFileset->fid;
    tsdbFidKeyRange(fid, pReader->pTsdb->keepCfg.days, pReader->pTsdb->keepCfg.precision, &win.skey, &win.ekey);

    // current file are no longer overlapped with query time window, ignore remain files
    if ((asc && win.skey > pReader->window.ekey) || (!asc && win.ekey < pReader->window.skey)) {
      tsdbDebug("%p remain files are not qualified for qrange:%" PRId64 "-%" PRId64 ", ignore, %s", pReader,
                pReader->window.skey, pReader->window.ekey, pReader->idStr);
      return false;
    }

    if ((asc && (win.ekey < pReader->window.skey)) || ((!asc) && (win.skey > pReader->window.ekey))) {
      pIter->index += step;
415 416 417
      if ((asc && pIter->index >= pIter->numOfFiles) || ((!asc) && pIter->index < 0)) {
        return false;
      }
418 419
      continue;
    }
C
Cary Xu 已提交
420

421
    tsdbDebug("%p file found fid:%d for qrange:%" PRId64 "-%" PRId64 ", %s", pReader, fid, pReader->window.skey,
422
              pReader->window.ekey, pReader->idStr);
423 424
    return true;
  }
425

426
_err:
H
Haojun Liao 已提交
427 428 429
  return false;
}

430
static void resetDataBlockIterator(SDataBlockIter* pIter, int32_t order) {
431 432
  pIter->order = order;
  pIter->index = -1;
433
  pIter->numOfBlocks = 0;
434 435 436 437 438 439 440
  if (pIter->blockList == NULL) {
    pIter->blockList = taosArrayInit(4, sizeof(SFileDataBlockInfo));
  } else {
    taosArrayClear(pIter->blockList);
  }
}

L
Liu Jicong 已提交
441
static void cleanupDataBlockIterator(SDataBlockIter* pIter) { taosArrayDestroy(pIter->blockList); }
H
Haojun Liao 已提交
442

H
Haojun Liao 已提交
443
static void initReaderStatus(SReaderStatus* pStatus) {
dengyihao's avatar
dengyihao 已提交
444 445
  pStatus->pTableIter = NULL;
  pStatus->loadFromFile = true;
H
Haojun Liao 已提交
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
static SSDataBlock* createResBlock(SQueryTableDataCond* pCond, int32_t capacity) {
  SSDataBlock* pResBlock = createDataBlock();
  if (pResBlock == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

  for (int32_t i = 0; i < pCond->numOfCols; ++i) {
    SColumnInfoData colInfo = {{0}, 0};
    colInfo.info = pCond->colList[i];
    blockDataAppendColInfo(pResBlock, &colInfo);
  }

  int32_t code = blockDataEnsureCapacity(pResBlock, capacity);
  if (code != TSDB_CODE_SUCCESS) {
    terrno = code;
    taosMemoryFree(pResBlock);
    return NULL;
  }

  return pResBlock;
}

471 472
static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, STsdbReader** ppReader, int32_t capacity,
                                const char* idstr) {
H
Haojun Liao 已提交
473
  int32_t      code = 0;
474
  int8_t       level = 0;
H
Haojun Liao 已提交
475
  STsdbReader* pReader = (STsdbReader*)taosMemoryCalloc(1, sizeof(*pReader));
H
Hongze Cheng 已提交
476 477
  if (pReader == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
H
Haojun Liao 已提交
478
    goto _end;
H
Hongze Cheng 已提交
479 480
  }

C
Cary Xu 已提交
481 482 483 484
  if (VND_IS_TSMA(pVnode)) {
    tsdbDebug("vgId:%d, tsma is selected to query", TD_VID(pVnode));
  }

H
Haojun Liao 已提交
485
  initReaderStatus(&pReader->status);
486

L
Liu Jicong 已提交
487
  pReader->pTsdb = getTsdbByRetentions(pVnode, pCond->twindows.skey, pVnode->config.tsdbCfg.retentions, idstr, &level);
dengyihao's avatar
dengyihao 已提交
488 489
  pReader->suid = pCond->suid;
  pReader->order = pCond->order;
490
  pReader->capacity = capacity;
dengyihao's avatar
dengyihao 已提交
491 492
  pReader->idStr = (idstr != NULL) ? strdup(idstr) : NULL;
  pReader->verRange = getQueryVerRange(pVnode, pCond, level);
493
  pReader->type = pCond->type;
494
  pReader->window = updateQueryTimeWindow(pReader->pTsdb, &pCond->twindows);
495

496
  ASSERT(pCond->numOfCols > 0);
H
Hongze Cheng 已提交
497

498
  limitOutputBufferSize(pCond, &pReader->capacity);
499

500 501
  // allocate buffer in order to load data blocks from file
  SBlockLoadSuppInfo* pSup = &pReader->suppInfo;
502
  pSup->pColAgg = taosArrayInit(4, sizeof(SColumnDataAgg));
503
  pSup->plist = taosMemoryCalloc(pCond->numOfCols, POINTER_BYTES);
504
  if (pSup->pColAgg == NULL || pSup->plist == NULL) {
505 506 507
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _end;
  }
H
Haojun Liao 已提交
508

509 510
  pSup->tsColAgg.colId = PRIMARYKEY_TIMESTAMP_COL_ID;

H
Hongze Cheng 已提交
511
  code = tBlockDataCreate(&pReader->status.fileBlockData);
H
Haojun Liao 已提交
512 513 514 515 516
  if (code != TSDB_CODE_SUCCESS) {
    terrno = code;
    goto _end;
  }

517 518 519 520
  pReader->pResBlock = createResBlock(pCond, pReader->capacity);
  if (pReader->pResBlock == NULL) {
    code = terrno;
    goto _end;
H
Hongze Cheng 已提交
521
  }
H
Hongze Cheng 已提交
522

523 524
  setColumnIdSlotList(pReader, pReader->pResBlock);

H
Hongze Cheng 已提交
525 526
  *ppReader = pReader;
  return code;
H
Hongze Cheng 已提交
527

H
Haojun Liao 已提交
528 529
_end:
  tsdbReaderClose(pReader);
H
Hongze Cheng 已提交
530 531 532
  *ppReader = NULL;
  return code;
}
H
Hongze Cheng 已提交
533

H
Haojun Liao 已提交
534
static int32_t doLoadBlockIndex(STsdbReader* pReader, SDataFReader* pFileReader, SArray* pIndexList) {
535
  SArray* aBlockIdx = taosArrayInit(8, sizeof(SBlockIdx));
H
Hongze Cheng 已提交
536

537
  int64_t st = taosGetTimestampUs();
H
Hongze Cheng 已提交
538
  int32_t code = tsdbReadBlockIdx(pFileReader, aBlockIdx);
H
Haojun Liao 已提交
539
  if (code != TSDB_CODE_SUCCESS) {
540
    goto _end;
H
Haojun Liao 已提交
541
  }
H
Hongze Cheng 已提交
542

543 544
  size_t num = taosArrayGetSize(aBlockIdx);
  if (num == 0) {
H
Haojun Liao 已提交
545
    taosArrayDestroy(aBlockIdx);
H
Haojun Liao 已提交
546 547
    return TSDB_CODE_SUCCESS;
  }
H
Hongze Cheng 已提交
548

549 550 551 552
  int64_t et1 = taosGetTimestampUs();

  SBlockIdx* pBlockIdx = NULL;
  for (int32_t i = 0; i < num; ++i) {
553
    pBlockIdx = (SBlockIdx*)taosArrayGet(aBlockIdx, i);
H
Haojun Liao 已提交
554

555
    // uid check
H
Hongze Cheng 已提交
556
    if (pBlockIdx->suid != pReader->suid) {
H
Haojun Liao 已提交
557 558 559 560
      continue;
    }

    // this block belongs to a table that is not queried.
H
Hongze Cheng 已提交
561
    void* p = taosHashGet(pReader->status.pTableMap, &pBlockIdx->uid, sizeof(uint64_t));
H
Haojun Liao 已提交
562 563 564 565 566 567
    if (p == NULL) {
      continue;
    }

    STableBlockScanInfo* pScanInfo = p;
    if (pScanInfo->pBlockList == NULL) {
568
      pScanInfo->pBlockList = taosArrayInit(4, sizeof(int32_t));
H
Haojun Liao 已提交
569 570
    }

H
Hongze Cheng 已提交
571
    taosArrayPush(pIndexList, pBlockIdx);
H
Haojun Liao 已提交
572
  }
H
Hongze Cheng 已提交
573

574
  int64_t et2 = taosGetTimestampUs();
575
  tsdbDebug("load block index for %d tables completed, elapsed time:%.2f ms, set blockIdx:%.2f ms, size:%.2f Kb %s",
576
            (int32_t)num, (et1 - st) / 1000.0, (et2 - et1) / 1000.0, num * sizeof(SBlockIdx) / 1024.0, pReader->idStr);
577 578 579

  pReader->cost.headFileLoadTime += (et1 - st) / 1000.0;

580
_end:
H
Hongze Cheng 已提交
581
  taosArrayDestroy(aBlockIdx);
H
Haojun Liao 已提交
582 583
  return code;
}
H
Hongze Cheng 已提交
584

585
static void cleanupTableScanInfo(SHashObj* pTableMap) {
586
  STableBlockScanInfo* px = NULL;
dengyihao's avatar
dengyihao 已提交
587
  while (1) {
588
    px = taosHashIterate(pTableMap, px);
589 590 591 592
    if (px == NULL) {
      break;
    }

593
    // reset the index in last block when handing a new file
594
    tMapDataClear(&px->mapData);
595 596
    taosArrayClear(px->pBlockList);
  }
597 598
}

599
static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, SBlockNumber* pBlockNum) {
600 601 602 603 604 605
  int32_t numOfQTable = 0;
  size_t  sizeInDisk = 0;
  size_t  numOfTables = taosArrayGetSize(pIndexList);

  int64_t st = taosGetTimestampUs();
  cleanupTableScanInfo(pReader->status.pTableMap);
606

dengyihao's avatar
dengyihao 已提交
607
  for (int32_t i = 0; i < numOfTables; ++i) {
H
Haojun Liao 已提交
608
    SBlockIdx* pBlockIdx = taosArrayGet(pIndexList, i);
H
Hongze Cheng 已提交
609

610
    STableBlockScanInfo* pScanInfo = taosHashGet(pReader->status.pTableMap, &pBlockIdx->uid, sizeof(int64_t));
H
Hongze Cheng 已提交
611

612
    tMapDataReset(&pScanInfo->mapData);
H
Hongze Cheng 已提交
613
    tsdbReadDataBlk(pReader->pFileReader, pBlockIdx, &pScanInfo->mapData);
614

615
    sizeInDisk += pScanInfo->mapData.nData;
616
    for (int32_t j = 0; j < pScanInfo->mapData.nItem; ++j) {
H
Hongze Cheng 已提交
617 618
      SDataBlk block = {0};
      tMapDataGetItemByIdx(&pScanInfo->mapData, j, &block, tGetDataBlk);
H
Hongze Cheng 已提交
619

620
      // 1. time range check
621
      if (block.minKey.ts > pReader->window.ekey || block.maxKey.ts < pReader->window.skey) {
H
Haojun Liao 已提交
622 623
        continue;
      }
H
Hongze Cheng 已提交
624

625
      // 2. version range check
H
Hongze Cheng 已提交
626
      if (block.minVer > pReader->verRange.maxVer || block.maxVer < pReader->verRange.minVer) {
627 628
        continue;
      }
629

630
      void* p = taosArrayPush(pScanInfo->pBlockList, &j);
H
Haojun Liao 已提交
631
      if (p == NULL) {
632
        tMapDataClear(&pScanInfo->mapData);
H
Haojun Liao 已提交
633 634
        return TSDB_CODE_OUT_OF_MEMORY;
      }
635

636
      pBlockNum->numOfBlocks += 1;
H
Haojun Liao 已提交
637
    }
H
Hongze Cheng 已提交
638

H
Haojun Liao 已提交
639
    if (pScanInfo->pBlockList != NULL && taosArrayGetSize(pScanInfo->pBlockList) > 0) {
640 641 642 643
      numOfQTable += 1;
    }
  }

H
Hongze Cheng 已提交
644
  pBlockNum->numOfLastFiles = pReader->pFileReader->pSet->nSttF;
645
  int32_t total = pBlockNum->numOfLastFiles + pBlockNum->numOfBlocks;
646

647
  double el = (taosGetTimestampUs() - st) / 1000.0;
648 649 650 651 652
  tsdbDebug(
      "load block of %ld tables completed, blocks:%d in %d tables, last-files:%d, block-info-size:%.2f Kb, elapsed "
      "time:%.2f ms %s",
      numOfTables, pBlockNum->numOfBlocks, numOfQTable, pBlockNum->numOfLastFiles, sizeInDisk / 1000.0, el,
      pReader->idStr);
653

654
  pReader->cost.numOfBlocks += total;
655
  pReader->cost.headFileLoadTime += el;
656

H
Haojun Liao 已提交
657 658
  return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
659

660
static void setBlockAllDumped(SFileBlockDumpInfo* pDumpInfo, int64_t maxKey, int32_t order) {
661
  int32_t step = ASCENDING_TRAVERSE(order) ? 1 : -1;
662
  pDumpInfo->allDumped = true;
663
  pDumpInfo->lastKey = maxKey + step;
H
Haojun Liao 已提交
664 665
}

666 667
static void doCopyColVal(SColumnInfoData* pColInfoData, int32_t rowIndex, int32_t colIndex, SColVal* pColVal,
                         SBlockLoadSuppInfo* pSup) {
H
Haojun Liao 已提交
668
  if (IS_VAR_DATA_TYPE(pColVal->type)) {
H
Hongze Cheng 已提交
669
    if (!COL_VAL_IS_VALUE(pColVal)) {
H
Haojun Liao 已提交
670 671 672
      colDataAppendNULL(pColInfoData, rowIndex);
    } else {
      varDataSetLen(pSup->buildBuf[colIndex], pColVal->value.nData);
H
Haojun Liao 已提交
673
      ASSERT(pColVal->value.nData <= pColInfoData->info.bytes);
H
Haojun Liao 已提交
674 675 676 677
      memcpy(varDataVal(pSup->buildBuf[colIndex]), pColVal->value.pData, pColVal->value.nData);
      colDataAppend(pColInfoData, rowIndex, pSup->buildBuf[colIndex], false);
    }
  } else {
H
Hongze Cheng 已提交
678
    colDataAppend(pColInfoData, rowIndex, (const char*)&pColVal->value, !COL_VAL_IS_VALUE(pColVal));
H
Haojun Liao 已提交
679
  }
H
Haojun Liao 已提交
680 681
}

682
static SFileDataBlockInfo* getCurrentBlockInfo(SDataBlockIter* pBlockIter) {
683 684
  if (taosArrayGetSize(pBlockIter->blockList) == 0) {
    ASSERT(pBlockIter->numOfBlocks == taosArrayGetSize(pBlockIter->blockList));
685 686
    return NULL;
  }
687 688 689

  SFileDataBlockInfo* pBlockInfo = taosArrayGet(pBlockIter->blockList, pBlockIter->index);
  return pBlockInfo;
690 691
}

H
Hongze Cheng 已提交
692
static SDataBlk* getCurrentBlock(SDataBlockIter* pBlockIter) { return &pBlockIter->block; }
693

H
Haojun Liao 已提交
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 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766
int32_t binarySearchForTs(char* pValue, int num, TSKEY key, int order) {
  int32_t midPos = -1;
  int32_t numOfRows;

  ASSERT(order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC);

  TSKEY*  keyList = (TSKEY*)pValue;
  int32_t firstPos = 0;
  int32_t lastPos = num - 1;

  if (order == TSDB_ORDER_DESC) {
    // find the first position which is smaller than the key
    while (1) {
      if (key >= keyList[firstPos]) return firstPos;
      if (key == keyList[lastPos]) return lastPos;

      if (key < keyList[lastPos]) {
        lastPos += 1;
        if (lastPos >= num) {
          return -1;
        } else {
          return lastPos;
        }
      }

      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1) + firstPos;

      if (key < keyList[midPos]) {
        firstPos = midPos + 1;
      } else if (key > keyList[midPos]) {
        lastPos = midPos - 1;
      } else {
        break;
      }
    }

  } else {
    // find the first position which is bigger than the key
    while (1) {
      if (key <= keyList[firstPos]) return firstPos;
      if (key == keyList[lastPos]) return lastPos;

      if (key > keyList[lastPos]) {
        lastPos = lastPos + 1;
        if (lastPos >= num)
          return -1;
        else
          return lastPos;
      }

      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1u) + firstPos;

      if (key < keyList[midPos]) {
        lastPos = midPos - 1;
      } else if (key > keyList[midPos]) {
        firstPos = midPos + 1;
      } else {
        break;
      }
    }
  }

  return midPos;
}

static int doBinarySearchKey(TSKEY* keyList, int num, int pos, TSKEY key, int order) {
  // start end position
  int s, e;
  s = pos;

  // check
H
Hongze Cheng 已提交
767
  assert(pos >= 0 && pos < num);
H
Haojun Liao 已提交
768 769 770 771
  assert(num > 0);

  if (order == TSDB_ORDER_ASC) {
    // find the first position which is smaller than the key
H
Hongze Cheng 已提交
772 773
    e = num - 1;
    if (key < keyList[pos]) return -1;
H
Haojun Liao 已提交
774 775
    while (1) {
      // check can return
H
Hongze Cheng 已提交
776 777 778
      if (key >= keyList[e]) return e;
      if (key <= keyList[s]) return s;
      if (e - s <= 1) return s;
H
Haojun Liao 已提交
779 780

      // change start or end position
H
Hongze Cheng 已提交
781
      int mid = s + (e - s + 1) / 2;
H
Haojun Liao 已提交
782 783
      if (keyList[mid] > key)
        e = mid;
H
Hongze Cheng 已提交
784
      else if (keyList[mid] < key)
H
Haojun Liao 已提交
785 786 787 788
        s = mid;
      else
        return mid;
    }
H
Hongze Cheng 已提交
789
  } else {  // DESC
H
Haojun Liao 已提交
790
    // find the first position which is bigger than the key
H
Hongze Cheng 已提交
791 792
    e = 0;
    if (key > keyList[pos]) return -1;
H
Haojun Liao 已提交
793 794
    while (1) {
      // check can return
H
Hongze Cheng 已提交
795 796 797
      if (key <= keyList[e]) return e;
      if (key >= keyList[s]) return s;
      if (s - e <= 1) return s;
H
Haojun Liao 已提交
798 799

      // change start or end position
H
Hongze Cheng 已提交
800
      int mid = s - (s - e + 1) / 2;
H
Haojun Liao 已提交
801 802
      if (keyList[mid] < key)
        e = mid;
H
Hongze Cheng 已提交
803
      else if (keyList[mid] > key)
H
Haojun Liao 已提交
804 805 806 807 808 809 810 811 812 813
        s = mid;
      else
        return mid;
    }
  }
}

int32_t getEndPosInDataBlock(STsdbReader* pReader, SBlockData* pBlockData, SDataBlk* pBlock, int32_t pos) {
  // NOTE: reverse the order to find the end position in data block
  int32_t endPos = -1;
H
Hongze Cheng 已提交
814
  bool    asc = ASCENDING_TRAVERSE(pReader->order);
H
Haojun Liao 已提交
815 816 817 818 819 820 821 822 823 824 825 826

  if (asc && pReader->window.ekey >= pBlock->maxKey.ts) {
    endPos = pBlock->nRow - 1;
  } else if (!asc && pReader->window.skey <= pBlock->minKey.ts) {
    endPos = 0;
  } else {
    endPos = doBinarySearchKey(pBlockData->aTSKEY, pBlock->nRow, pos, pReader->window.ekey, pReader->order);
  }

  return endPos;
}

827
static int32_t copyBlockDataToSDataBlock(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo) {
828
  SReaderStatus*  pStatus = &pReader->status;
829
  SDataBlockIter* pBlockIter = &pStatus->blockIter;
H
Hongze Cheng 已提交
830

831
  SBlockData*         pBlockData = &pStatus->fileBlockData;
H
Haojun Liao 已提交
832
  SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter);
H
Hongze Cheng 已提交
833
  SDataBlk*           pBlock = getCurrentBlock(pBlockIter);
H
Haojun Liao 已提交
834
  SSDataBlock*        pResBlock = pReader->pResBlock;
835
  int32_t             numOfOutputCols = blockDataGetNumOfCols(pResBlock);
H
Haojun Liao 已提交
836

H
Haojun Liao 已提交
837
  SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo;
838
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
H
Haojun Liao 已提交
839

H
Haojun Liao 已提交
840
  SColVal cv = {0};
841
  int64_t st = taosGetTimestampUs();
842 843
  bool    asc = ASCENDING_TRAVERSE(pReader->order);
  int32_t step = asc ? 1 : -1;
844

845 846
  if ((pDumpInfo->rowIndex == 0 && asc) || (pDumpInfo->rowIndex == pBlock->nRow - 1 && (!asc))) {
    if (asc && pReader->window.skey <= pBlock->minKey.ts) {
847 848 849
      // pDumpInfo->rowIndex = 0;
    } else if (!asc && pReader->window.ekey >= pBlock->maxKey.ts) {
      // pDumpInfo->rowIndex = pBlock->nRow - 1;
850 851 852 853 854
    } else {
      int32_t pos = asc ? pBlock->nRow - 1 : 0;
      int32_t order = (pReader->order == TSDB_ORDER_ASC) ? TSDB_ORDER_DESC : TSDB_ORDER_ASC;
      pDumpInfo->rowIndex = doBinarySearchKey(pBlockData->aTSKEY, pBlock->nRow, pos, pReader->window.skey, order);
    }
H
Haojun Liao 已提交
855 856 857
  }

  // time window check
858 859 860 861 862 863 864
  int32_t endIndex = getEndPosInDataBlock(pReader, pBlockData, pBlock, pDumpInfo->rowIndex);
  if (endIndex == -1) {
    setBlockAllDumped(pDumpInfo, pReader->window.ekey, pReader->order);
    return TSDB_CODE_SUCCESS;
  }

  endIndex += step;
H
Haojun Liao 已提交
865
  int32_t remain = asc ? (endIndex - pDumpInfo->rowIndex) : (pDumpInfo->rowIndex - endIndex);
H
Hongze Cheng 已提交
866
  if (remain > pReader->capacity) {  // output buffer check
867 868 869
    remain = pReader->capacity;
  }

H
Haojun Liao 已提交
870 871
  int32_t rowIndex = 0;

H
Hongze Cheng 已提交
872
  int32_t          i = 0;
873 874
  SColumnInfoData* pColData = taosArrayGet(pResBlock->pDataBlock, i);
  if (pColData->info.colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
H
Haojun Liao 已提交
875
    if (asc) {
H
Haojun Liao 已提交
876
      memcpy(pColData->pData, &pBlockData->aTSKEY[pDumpInfo->rowIndex], remain * sizeof(int64_t));
H
Haojun Liao 已提交
877 878 879 880
    } else {
      for (int32_t j = pDumpInfo->rowIndex; rowIndex < remain; j += step) {
        colDataAppendInt64(pColData, rowIndex++, &pBlockData->aTSKEY[j]);
      }
881
    }
H
Haojun Liao 已提交
882

883 884 885
    i += 1;
  }

886 887 888
  int32_t colIndex = 0;
  int32_t num = taosArrayGetSize(pBlockData->aIdx);
  while (i < numOfOutputCols && colIndex < num) {
889 890 891
    rowIndex = 0;
    pColData = taosArrayGet(pResBlock->pDataBlock, i);

H
Hongze Cheng 已提交
892
    SColData* pData = tBlockDataGetColDataByIdx(pBlockData, colIndex);
893 894 895
    if (pData->cid < pColData->info.colId) {
      colIndex += 1;
    } else if (pData->cid == pColData->info.colId) {
H
Hongze Cheng 已提交
896
      if (pData->flag == HAS_NONE || pData->flag == HAS_NULL || pData->flag == (HAS_NULL | HAS_NONE)) {
H
Haojun Liao 已提交
897 898 899 900
        colDataAppendNNULL(pColData, 0, remain);
      } else {
        if (IS_NUMERIC_TYPE(pColData->info.type) && asc) {
          uint8_t* p = pData->pData + tDataTypes[pData->type].bytes * pDumpInfo->rowIndex;
H
Haojun Liao 已提交
901
          memcpy(pColData->pData, p, remain * tDataTypes[pData->type].bytes);
H
Haojun Liao 已提交
902 903 904 905

          // null value exists, check one-by-one
          if (pData->flag != HAS_VALUE) {
            for (int32_t j = pDumpInfo->rowIndex; rowIndex < remain; j += step, rowIndex++) {
H
Haojun Liao 已提交
906
              uint8_t v = tColDataGetBitValue(pData, j);
H
Haojun Liao 已提交
907 908 909 910 911 912 913 914 915 916 917
              if (v == 0 || v == 1) {
                colDataSetNull_f(pColData->nullbitmap, rowIndex);
              }
            }
          }
        } else {
          for (int32_t j = pDumpInfo->rowIndex; rowIndex < remain; j += step) {
            tColDataGetValue(pData, j, &cv);
            doCopyColVal(pColData, rowIndex++, i, &cv, pSupInfo);
          }
        }
H
Haojun Liao 已提交
918
      }
H
Haojun Liao 已提交
919

920
      colIndex += 1;
921
      i += 1;
922 923
    } else {  // the specified column does not exist in file block, fill with null data
      colDataAppendNNULL(pColData, 0, remain);
924
      i += 1;
H
Haojun Liao 已提交
925
    }
926 927
  }

928
  // fill the mis-matched columns with null value
929
  while (i < numOfOutputCols) {
930 931 932
    pColData = taosArrayGet(pResBlock->pDataBlock, i);
    colDataAppendNNULL(pColData, 0, remain);
    i += 1;
H
Haojun Liao 已提交
933
  }
H
Haojun Liao 已提交
934

935
  pResBlock->info.rows = remain;
936
  pDumpInfo->rowIndex += step * remain;
937

938
  // check if current block are all handled
939
  if (pDumpInfo->rowIndex >= 0 && pDumpInfo->rowIndex < pBlock->nRow) {
940 941 942 943
    int64_t ts = pBlockData->aTSKEY[pDumpInfo->rowIndex];
    if (outOfTimeWindow(ts, &pReader->window)) {  // the remain data has out of query time window, ignore current block
      setBlockAllDumped(pDumpInfo, ts, pReader->order);
    }
944
  } else {
945 946
    int64_t ts = asc ? pBlock->maxKey.ts : pBlock->minKey.ts;
    setBlockAllDumped(pDumpInfo, ts, pReader->order);
H
Haojun Liao 已提交
947
  }
H
Haojun Liao 已提交
948

949
  double elapsedTime = (taosGetTimestampUs() - st) / 1000.0;
H
Haojun Liao 已提交
950
  pReader->cost.blockLoadTime += elapsedTime;
H
Haojun Liao 已提交
951

952
  int32_t unDumpedRows = asc ? pBlock->nRow - pDumpInfo->rowIndex : pDumpInfo->rowIndex + 1;
H
Haojun Liao 已提交
953
  tsdbDebug("%p copy file block to sdatablock, global index:%d, table index:%d, brange:%" PRId64 "-%" PRId64
954
            ", rows:%d, remain:%d, minVer:%" PRId64 ", maxVer:%" PRId64 ", elapsed time:%.2f ms, %s",
H
Hongze Cheng 已提交
955 956
            pReader, pBlockIter->index, pBlockInfo->tbBlockIdx, pBlock->minKey.ts, pBlock->maxKey.ts, remain,
            unDumpedRows, pBlock->minVer, pBlock->maxVer, elapsedTime, pReader->idStr);
957 958 959 960

  return TSDB_CODE_SUCCESS;
}

961 962
static int32_t doLoadFileBlockData(STsdbReader* pReader, SDataBlockIter* pBlockIter, SBlockData* pBlockData,
                                   uint64_t uid) {
963 964
  int64_t st = taosGetTimestampUs();

965 966
  tBlockDataReset(pBlockData);
  TABLEID tid = {.suid = pReader->suid, .uid = uid};
967 968
  int32_t code =
      tBlockDataInit(pBlockData, &tid, pReader->pSchema, &pReader->suppInfo.colIds[1], pReader->suppInfo.numOfCols - 1);
969 970 971 972
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

973
  SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter);
974
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
975
  ASSERT(pBlockInfo != NULL);
976

H
Hongze Cheng 已提交
977
  SDataBlk* pBlock = getCurrentBlock(pBlockIter);
978
  code = tsdbReadDataBlock(pReader->pFileReader, pBlock, pBlockData);
979 980 981
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("%p error occurs in loading file block, global index:%d, table index:%d, brange:%" PRId64 "-%" PRId64
              ", rows:%d, code:%s %s",
982
              pReader, pBlockIter->index, pBlockInfo->tbBlockIdx, pBlock->minKey.ts, pBlock->maxKey.ts, pBlock->nRow,
983 984 985
              tstrerror(code), pReader->idStr);
    return code;
  }
986

987
  double elapsedTime = (taosGetTimestampUs() - st) / 1000.0;
988

989 990 991 992
  tsdbDebug("%p load file block into buffer, global index:%d, index in table block list:%d, brange:%" PRId64 "-%" PRId64
            ", rows:%d, minVer:%" PRId64 ", maxVer:%" PRId64 ", elapsed time:%.2f ms, %s",
            pReader, pBlockIter->index, pBlockInfo->tbBlockIdx, pBlock->minKey.ts, pBlock->maxKey.ts, pBlock->nRow,
            pBlock->minVer, pBlock->maxVer, elapsedTime, pReader->idStr);
993 994 995

  pReader->cost.blockLoadTime += elapsedTime;
  pDumpInfo->allDumped = false;
996

H
Haojun Liao 已提交
997
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
998
}
H
Hongze Cheng 已提交
999

H
Haojun Liao 已提交
1000 1001 1002
static void cleanupBlockOrderSupporter(SBlockOrderSupporter* pSup) {
  taosMemoryFreeClear(pSup->numOfBlocksPerTable);
  taosMemoryFreeClear(pSup->indexPerTable);
H
Hongze Cheng 已提交
1003

H
Haojun Liao 已提交
1004 1005 1006 1007
  for (int32_t i = 0; i < pSup->numOfTables; ++i) {
    SBlockOrderWrapper* pBlockInfo = pSup->pDataBlockInfo[i];
    taosMemoryFreeClear(pBlockInfo);
  }
H
Hongze Cheng 已提交
1008

H
Haojun Liao 已提交
1009 1010
  taosMemoryFreeClear(pSup->pDataBlockInfo);
}
H
Hongze Cheng 已提交
1011

H
Haojun Liao 已提交
1012 1013
static int32_t initBlockOrderSupporter(SBlockOrderSupporter* pSup, int32_t numOfTables) {
  ASSERT(numOfTables >= 1);
H
Hongze Cheng 已提交
1014

H
Haojun Liao 已提交
1015
  pSup->numOfBlocksPerTable = taosMemoryCalloc(1, sizeof(int32_t) * numOfTables);
1016 1017
  pSup->indexPerTable = taosMemoryCalloc(1, sizeof(int32_t) * numOfTables);
  pSup->pDataBlockInfo = taosMemoryCalloc(1, POINTER_BYTES * numOfTables);
H
Hongze Cheng 已提交
1018

H
Haojun Liao 已提交
1019 1020 1021 1022
  if (pSup->numOfBlocksPerTable == NULL || pSup->indexPerTable == NULL || pSup->pDataBlockInfo == NULL) {
    cleanupBlockOrderSupporter(pSup);
    return TSDB_CODE_OUT_OF_MEMORY;
  }
H
Hongze Cheng 已提交
1023

H
Haojun Liao 已提交
1024 1025
  return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
1026

H
Haojun Liao 已提交
1027
static int32_t fileDataBlockOrderCompar(const void* pLeft, const void* pRight, void* param) {
1028
  int32_t leftIndex = *(int32_t*)pLeft;
H
Haojun Liao 已提交
1029
  int32_t rightIndex = *(int32_t*)pRight;
H
Hongze Cheng 已提交
1030

H
Haojun Liao 已提交
1031
  SBlockOrderSupporter* pSupporter = (SBlockOrderSupporter*)param;
H
Hongze Cheng 已提交
1032

H
Haojun Liao 已提交
1033 1034
  int32_t leftTableBlockIndex = pSupporter->indexPerTable[leftIndex];
  int32_t rightTableBlockIndex = pSupporter->indexPerTable[rightIndex];
H
Hongze Cheng 已提交
1035

H
Haojun Liao 已提交
1036 1037 1038 1039 1040 1041 1042
  if (leftTableBlockIndex > pSupporter->numOfBlocksPerTable[leftIndex]) {
    /* left block is empty */
    return 1;
  } else if (rightTableBlockIndex > pSupporter->numOfBlocksPerTable[rightIndex]) {
    /* right block is empty */
    return -1;
  }
H
Hongze Cheng 已提交
1043

1044
  SBlockOrderWrapper* pLeftBlock = &pSupporter->pDataBlockInfo[leftIndex][leftTableBlockIndex];
H
Haojun Liao 已提交
1045
  SBlockOrderWrapper* pRightBlock = &pSupporter->pDataBlockInfo[rightIndex][rightTableBlockIndex];
H
Hongze Cheng 已提交
1046

1047 1048 1049
  return pLeftBlock->offset > pRightBlock->offset ? 1 : -1;
}

H
Haojun Liao 已提交
1050
static int32_t doSetCurrentBlock(SDataBlockIter* pBlockIter, const char* idStr) {
1051 1052 1053
  SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter);
  if (pBlockInfo != NULL) {
    STableBlockScanInfo* pScanInfo = taosHashGet(pBlockIter->pTableMap, &pBlockInfo->uid, sizeof(pBlockInfo->uid));
H
Haojun Liao 已提交
1054 1055 1056 1057 1058 1059
    if (pScanInfo == NULL) {
      tsdbError("failed to locate the uid:%"PRIu64" in query table uid list, %s", pBlockInfo->uid, idStr);
      return TSDB_CODE_INVALID_PARA;
    }

    int32_t* mapDataIndex = taosArrayGet(pScanInfo->pBlockList, pBlockInfo->tbBlockIdx);
H
Hongze Cheng 已提交
1060
    tMapDataGetItemByIdx(&pScanInfo->mapData, *mapDataIndex, &pBlockIter->block, tGetDataBlk);
1061
  }
1062 1063 1064 1065 1066 1067

#if 0
  qDebug("check file block, table uid:%"PRIu64" index:%d offset:%"PRId64", ", pScanInfo->uid, *mapDataIndex, pBlockIter->block.aSubBlock[0].offset);
#endif

  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1068
}
H
Hongze Cheng 已提交
1069

1070
static int32_t initBlockIterator(STsdbReader* pReader, SDataBlockIter* pBlockIter, int32_t numOfBlocks) {
1071
  bool asc = ASCENDING_TRAVERSE(pReader->order);
H
Haojun Liao 已提交
1072

1073
  pBlockIter->numOfBlocks = numOfBlocks;
1074
  taosArrayClear(pBlockIter->blockList);
1075
  pBlockIter->pTableMap = pReader->status.pTableMap;
1076

1077 1078
  // access data blocks according to the offset of each block in asc/desc order.
  int32_t numOfTables = (int32_t)taosHashGetSize(pReader->status.pTableMap);
H
Haojun Liao 已提交
1079

1080
  int64_t st = taosGetTimestampUs();
H
Haojun Liao 已提交
1081

1082
  SBlockOrderSupporter sup = {0};
1083
  int32_t              code = initBlockOrderSupporter(&sup, numOfTables);
1084 1085 1086
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }
H
Haojun Liao 已提交
1087

1088 1089 1090 1091 1092 1093 1094
  int32_t cnt = 0;
  void*   ptr = NULL;
  while (1) {
    ptr = taosHashIterate(pReader->status.pTableMap, ptr);
    if (ptr == NULL) {
      break;
    }
H
Haojun Liao 已提交
1095

1096 1097 1098 1099
    STableBlockScanInfo* pTableScanInfo = (STableBlockScanInfo*)ptr;
    if (pTableScanInfo->pBlockList == NULL || taosArrayGetSize(pTableScanInfo->pBlockList) == 0) {
      continue;
    }
H
Haojun Liao 已提交
1100

1101 1102
    size_t num = taosArrayGetSize(pTableScanInfo->pBlockList);
    sup.numOfBlocksPerTable[sup.numOfTables] = num;
H
Haojun Liao 已提交
1103

1104 1105 1106 1107 1108
    char* buf = taosMemoryMalloc(sizeof(SBlockOrderWrapper) * num);
    if (buf == NULL) {
      cleanupBlockOrderSupporter(&sup);
      return TSDB_CODE_TDB_OUT_OF_MEMORY;
    }
H
Haojun Liao 已提交
1109

1110
    sup.pDataBlockInfo[sup.numOfTables] = (SBlockOrderWrapper*)buf;
H
Hongze Cheng 已提交
1111
    SDataBlk block = {0};
1112 1113
    for (int32_t k = 0; k < num; ++k) {
      SBlockOrderWrapper wrapper = {0};
1114 1115

      int32_t* mapDataIndex = taosArrayGet(pTableScanInfo->pBlockList, k);
H
Hongze Cheng 已提交
1116
      tMapDataGetItemByIdx(&pTableScanInfo->mapData, *mapDataIndex, &block, tGetDataBlk);
1117

1118
      wrapper.uid = pTableScanInfo->uid;
1119
      wrapper.offset = block.aSubBlock[0].offset;
H
Haojun Liao 已提交
1120

1121 1122 1123 1124 1125 1126
      sup.pDataBlockInfo[sup.numOfTables][k] = wrapper;
      cnt++;
    }

    sup.numOfTables += 1;
  }
H
Haojun Liao 已提交
1127

1128
  ASSERT(numOfBlocks == cnt);
H
Haojun Liao 已提交
1129

1130
  // since there is only one table qualified, blocks are not sorted
1131 1132
  if (sup.numOfTables == 1) {
    for (int32_t i = 0; i < numOfBlocks; ++i) {
1133 1134
      SFileDataBlockInfo blockInfo = {.uid = sup.pDataBlockInfo[0][i].uid, .tbBlockIdx = i};
      taosArrayPush(pBlockIter->blockList, &blockInfo);
1135
    }
1136

1137
    int64_t et = taosGetTimestampUs();
1138
    tsdbDebug("%p create blocks info struct completed for one table, %d blocks not sorted, elapsed time:%.2f ms %s",
1139
              pReader, numOfBlocks, (et - st) / 1000.0, pReader->idStr);
H
Haojun Liao 已提交
1140

1141
    pBlockIter->index = asc ? 0 : (numOfBlocks - 1);
H
Haojun Liao 已提交
1142
    cleanupBlockOrderSupporter(&sup);
H
Haojun Liao 已提交
1143
    doSetCurrentBlock(pBlockIter, pReader->idStr);
1144
    return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1145
  }
H
Haojun Liao 已提交
1146

1147 1148
  tsdbDebug("%p create data blocks info struct completed, %d blocks in %d tables %s", pReader, cnt, sup.numOfTables,
            pReader->idStr);
1149

1150
  ASSERT(cnt <= numOfBlocks && sup.numOfTables <= numOfTables);
H
Haojun Liao 已提交
1151

1152 1153 1154 1155 1156
  SMultiwayMergeTreeInfo* pTree = NULL;
  uint8_t                 ret = tMergeTreeCreate(&pTree, sup.numOfTables, &sup, fileDataBlockOrderCompar);
  if (ret != TSDB_CODE_SUCCESS) {
    cleanupBlockOrderSupporter(&sup);
    return TSDB_CODE_TDB_OUT_OF_MEMORY;
H
Haojun Liao 已提交
1157
  }
H
Haojun Liao 已提交
1158

1159 1160 1161 1162
  int32_t numOfTotal = 0;
  while (numOfTotal < cnt) {
    int32_t pos = tMergeTreeGetChosenIndex(pTree);
    int32_t index = sup.indexPerTable[pos]++;
H
Haojun Liao 已提交
1163

1164 1165
    SFileDataBlockInfo blockInfo = {.uid = sup.pDataBlockInfo[pos][index].uid, .tbBlockIdx = index};
    taosArrayPush(pBlockIter->blockList, &blockInfo);
H
Haojun Liao 已提交
1166

1167 1168 1169 1170
    // set data block index overflow, in order to disable the offset comparator
    if (sup.indexPerTable[pos] >= sup.numOfBlocksPerTable[pos]) {
      sup.indexPerTable[pos] = sup.numOfBlocksPerTable[pos] + 1;
    }
H
Haojun Liao 已提交
1171

1172 1173
    numOfTotal += 1;
    tMergeTreeAdjust(pTree, tMergeTreeGetAdjustIndex(pTree));
H
Haojun Liao 已提交
1174
  }
H
Haojun Liao 已提交
1175

1176
  int64_t et = taosGetTimestampUs();
H
Hongze Cheng 已提交
1177 1178
  tsdbDebug("%p %d data blocks access order completed, elapsed time:%.2f ms %s", pReader, numOfBlocks,
            (et - st) / 1000.0, pReader->idStr);
1179 1180
  cleanupBlockOrderSupporter(&sup);
  taosMemoryFree(pTree);
H
Haojun Liao 已提交
1181

1182
  pBlockIter->index = asc ? 0 : (numOfBlocks - 1);
H
Haojun Liao 已提交
1183
  doSetCurrentBlock(pBlockIter, pReader->idStr);
1184

1185
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1186
}
H
Hongze Cheng 已提交
1187

H
Haojun Liao 已提交
1188
static bool blockIteratorNext(SDataBlockIter* pBlockIter, const char* idStr) {
1189 1190
  bool asc = ASCENDING_TRAVERSE(pBlockIter->order);

1191
  int32_t step = asc ? 1 : -1;
1192
  if ((pBlockIter->index >= pBlockIter->numOfBlocks - 1 && asc) || (pBlockIter->index <= 0 && (!asc))) {
1193 1194 1195
    return false;
  }

1196
  pBlockIter->index += step;
H
Haojun Liao 已提交
1197
  doSetCurrentBlock(pBlockIter, idStr);
1198

1199 1200 1201
  return true;
}

1202 1203 1204
/**
 * This is an two rectangles overlap cases.
 */
H
Hongze Cheng 已提交
1205
static int32_t dataBlockPartiallyRequired(STimeWindow* pWindow, SVersionRange* pVerRange, SDataBlk* pBlock) {
1206 1207
  return (pWindow->ekey < pBlock->maxKey.ts && pWindow->ekey >= pBlock->minKey.ts) ||
         (pWindow->skey > pBlock->minKey.ts && pWindow->skey <= pBlock->maxKey.ts) ||
H
Hongze Cheng 已提交
1208 1209
         (pVerRange->minVer > pBlock->minVer && pVerRange->minVer <= pBlock->maxVer) ||
         (pVerRange->maxVer < pBlock->maxVer && pVerRange->maxVer >= pBlock->minVer);
H
Haojun Liao 已提交
1210
}
H
Hongze Cheng 已提交
1211

H
Haojun Liao 已提交
1212
static SDataBlk* getNeighborBlockOfSameTable(SFileDataBlockInfo* pBlockInfo, STableBlockScanInfo* pTableBlockScanInfo,
H
Hongze Cheng 已提交
1213
                                             int32_t* nextIndex, int32_t order) {
1214
  bool asc = ASCENDING_TRAVERSE(order);
H
Haojun Liao 已提交
1215
  if (asc && pBlockInfo->tbBlockIdx >= taosArrayGetSize(pTableBlockScanInfo->pBlockList) - 1) {
1216
    return NULL;
1217 1218
  }

H
Haojun Liao 已提交
1219
  if (!asc && pBlockInfo->tbBlockIdx == 0) {
1220 1221 1222
    return NULL;
  }

1223
  int32_t step = asc ? 1 : -1;
H
Haojun Liao 已提交
1224
  *nextIndex = pBlockInfo->tbBlockIdx + step;
1225

H
Hongze Cheng 已提交
1226 1227
  SDataBlk* pBlock = taosMemoryCalloc(1, sizeof(SDataBlk));
  int32_t*  indexInMapdata = taosArrayGet(pTableBlockScanInfo->pBlockList, *nextIndex);
1228

H
Hongze Cheng 已提交
1229
  tMapDataGetItemByIdx(&pTableBlockScanInfo->mapData, *indexInMapdata, pBlock, tGetDataBlk);
1230
  return pBlock;
1231 1232 1233 1234 1235
}

static int32_t findFileBlockInfoIndex(SDataBlockIter* pBlockIter, SFileDataBlockInfo* pFBlockInfo) {
  ASSERT(pBlockIter != NULL && pFBlockInfo != NULL);

1236
  int32_t step = ASCENDING_TRAVERSE(pBlockIter->order) ? 1 : -1;
1237 1238
  int32_t index = pBlockIter->index;

1239
  while (index < pBlockIter->numOfBlocks && index >= 0) {
1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251
    SFileDataBlockInfo* pFBlock = taosArrayGet(pBlockIter->blockList, index);
    if (pFBlock->uid == pFBlockInfo->uid && pFBlock->tbBlockIdx == pFBlockInfo->tbBlockIdx) {
      return index;
    }

    index += step;
  }

  ASSERT(0);
  return -1;
}

1252
static int32_t setFileBlockActiveInBlockIter(SDataBlockIter* pBlockIter, int32_t index, int32_t step) {
1253
  if (index < 0 || index >= pBlockIter->numOfBlocks) {
1254 1255 1256 1257
    return -1;
  }

  SFileDataBlockInfo fblock = *(SFileDataBlockInfo*)taosArrayGet(pBlockIter->blockList, index);
1258 1259 1260 1261 1262
  pBlockIter->index += step;

  if (index != pBlockIter->index) {
    taosArrayRemove(pBlockIter->blockList, index);
    taosArrayInsert(pBlockIter->blockList, pBlockIter->index, &fblock);
1263

1264 1265 1266
    SFileDataBlockInfo* pBlockInfo = taosArrayGet(pBlockIter->blockList, pBlockIter->index);
    ASSERT(pBlockInfo->uid == fblock.uid && pBlockInfo->tbBlockIdx == fblock.tbBlockIdx);
  }
1267

H
Haojun Liao 已提交
1268
  doSetCurrentBlock(pBlockIter, "");
1269 1270 1271
  return TSDB_CODE_SUCCESS;
}

H
Hongze Cheng 已提交
1272
static bool overlapWithNeighborBlock(SDataBlk* pBlock, SDataBlk* pNeighbor, int32_t order) {
1273 1274 1275 1276 1277 1278
  // it is the last block in current file, no chance to overlap with neighbor blocks.
  if (ASCENDING_TRAVERSE(order)) {
    return pBlock->maxKey.ts == pNeighbor->minKey.ts;
  } else {
    return pBlock->minKey.ts == pNeighbor->maxKey.ts;
  }
H
Haojun Liao 已提交
1279
}
H
Hongze Cheng 已提交
1280

H
Hongze Cheng 已提交
1281
static bool bufferDataInFileBlockGap(int32_t order, TSDBKEY key, SDataBlk* pBlock) {
H
Haojun Liao 已提交
1282
  bool ascScan = ASCENDING_TRAVERSE(order);
H
Hongze Cheng 已提交
1283

1284
  return (ascScan && (key.ts != TSKEY_INITIAL_VAL && key.ts <= pBlock->minKey.ts)) ||
1285
         (!ascScan && (key.ts != TSKEY_INITIAL_VAL && key.ts >= pBlock->maxKey.ts));
H
Haojun Liao 已提交
1286
}
H
Hongze Cheng 已提交
1287

H
Hongze Cheng 已提交
1288
static bool keyOverlapFileBlock(TSDBKEY key, SDataBlk* pBlock, SVersionRange* pVerRange) {
H
Hongze Cheng 已提交
1289 1290
  return (key.ts >= pBlock->minKey.ts && key.ts <= pBlock->maxKey.ts) && (pBlock->maxVer >= pVerRange->minVer) &&
         (pBlock->minVer <= pVerRange->maxVer);
H
Haojun Liao 已提交
1291 1292
}

H
Hongze Cheng 已提交
1293
static bool doCheckforDatablockOverlap(STableBlockScanInfo* pBlockScanInfo, const SDataBlk* pBlock) {
1294 1295 1296 1297 1298
  size_t num = taosArrayGetSize(pBlockScanInfo->delSkyline);

  for (int32_t i = pBlockScanInfo->fileDelIndex; i < num; i += 1) {
    TSDBKEY* p = taosArrayGet(pBlockScanInfo->delSkyline, i);
    if (p->ts >= pBlock->minKey.ts && p->ts <= pBlock->maxKey.ts) {
H
Hongze Cheng 已提交
1299
      if (p->version >= pBlock->minVer) {
1300 1301 1302
        return true;
      }
    } else if (p->ts < pBlock->minKey.ts) {  // p->ts < pBlock->minKey.ts
H
Hongze Cheng 已提交
1303
      if (p->version >= pBlock->minVer) {
1304 1305
        if (i < num - 1) {
          TSDBKEY* pnext = taosArrayGet(pBlockScanInfo->delSkyline, i + 1);
H
Hongze Cheng 已提交
1306 1307
          if (pnext->ts >= pBlock->minKey.ts) {
            return true;
1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320
          }
        } else {  // it must be the last point
          ASSERT(p->version == 0);
        }
      }
    } else {  // (p->ts > pBlock->maxKey.ts) {
      return false;
    }
  }

  return false;
}

H
Hongze Cheng 已提交
1321
static bool overlapWithDelSkyline(STableBlockScanInfo* pBlockScanInfo, const SDataBlk* pBlock, int32_t order) {
1322 1323 1324 1325
  if (pBlockScanInfo->delSkyline == NULL) {
    return false;
  }

1326
  // ts is not overlap
1327
  TSDBKEY* pFirst = taosArrayGet(pBlockScanInfo->delSkyline, 0);
L
Liu Jicong 已提交
1328
  TSDBKEY* pLast = taosArrayGetLast(pBlockScanInfo->delSkyline);
1329 1330 1331 1332 1333
  if (pBlock->minKey.ts > pLast->ts || pBlock->maxKey.ts < pFirst->ts) {
    return false;
  }

  // version is not overlap
1334 1335 1336 1337
  if (ASCENDING_TRAVERSE(order)) {
    return doCheckforDatablockOverlap(pBlockScanInfo, pBlock);
  } else {
    int32_t index = pBlockScanInfo->fileDelIndex;
1338
    while (1) {
1339 1340 1341 1342 1343
      TSDBKEY* p = taosArrayGet(pBlockScanInfo->delSkyline, index);
      if (p->ts > pBlock->minKey.ts && index > 0) {
        index -= 1;
      } else {  // find the first point that is smaller than the minKey.ts of dataBlock.
        break;
1344 1345 1346
      }
    }

1347 1348
    return doCheckforDatablockOverlap(pBlockScanInfo, pBlock);
  }
1349 1350
}

H
Haojun Liao 已提交
1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363
typedef struct {
  bool overlapWithNeighborBlock;
  bool hasDupTs;
  bool overlapWithDelInfo;
  bool overlapWithLastBlock;
  bool overlapWithKeyInBuf;
  bool partiallyRequired;
  bool moreThanCapcity;
} SDataBlockToLoadInfo;

static void getBlockToLoadInfo(SDataBlockToLoadInfo* pInfo, SFileDataBlockInfo* pBlockInfo, SDataBlk* pBlock,
                               STableBlockScanInfo* pScanInfo, TSDBKEY keyInBuf, SLastBlockReader* pLastBlockReader,
                               STsdbReader* pReader) {
H
Hongze Cheng 已提交
1364
  int32_t   neighborIndex = 0;
H
Haojun Liao 已提交
1365
  SDataBlk* pNeighbor = getNeighborBlockOfSameTable(pBlockInfo, pScanInfo, &neighborIndex, pReader->order);
1366

1367
  // overlap with neighbor
1368
  if (pNeighbor) {
H
Haojun Liao 已提交
1369
    pInfo->overlapWithNeighborBlock = overlapWithNeighborBlock(pBlock, pNeighbor, pReader->order);
1370
    taosMemoryFree(pNeighbor);
1371 1372
  }

1373
  // has duplicated ts of different version in this block
H
Haojun Liao 已提交
1374 1375
  pInfo->hasDupTs = (pBlock->nSubBlock == 1) ? pBlock->hasDup : true;
  pInfo->overlapWithDelInfo = overlapWithDelSkyline(pScanInfo, pBlock, pReader->order);
1376

H
Haojun Liao 已提交
1377 1378 1379
  if (hasDataInLastBlock(pLastBlockReader)) {
    int64_t tsLast = getCurrentKeyInLastBlock(pLastBlockReader);
    pInfo->overlapWithLastBlock = !(pBlock->maxKey.ts < tsLast || pBlock->minKey.ts > tsLast);
1380 1381
  }

H
Haojun Liao 已提交
1382 1383 1384 1385
  pInfo->moreThanCapcity = pBlock->nRow > pReader->capacity;
  pInfo->partiallyRequired = dataBlockPartiallyRequired(&pReader->window, &pReader->verRange, pBlock);
  pInfo->overlapWithKeyInBuf = keyOverlapFileBlock(keyInBuf, pBlock, &pReader->verRange);
}
1386

H
Haojun Liao 已提交
1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400
// 1. the version of all rows should be less than the endVersion
// 2. current block should not overlap with next neighbor block
// 3. current timestamp should not be overlap with each other
// 4. output buffer should be large enough to hold all rows in current block
// 5. delete info should not overlap with current block data
// 6. current block should not contain the duplicated ts
static bool fileBlockShouldLoad(STsdbReader* pReader, SFileDataBlockInfo* pBlockInfo, SDataBlk* pBlock,
                                STableBlockScanInfo* pScanInfo, TSDBKEY keyInBuf, SLastBlockReader* pLastBlockReader) {
  SDataBlockToLoadInfo info = {0};
  getBlockToLoadInfo(&info, pBlockInfo, pBlock, pScanInfo, keyInBuf, pLastBlockReader, pReader);

  bool loadDataBlock =
      (info.overlapWithNeighborBlock || info.hasDupTs || info.partiallyRequired || info.overlapWithKeyInBuf ||
       info.moreThanCapcity || info.overlapWithDelInfo || info.overlapWithLastBlock);
1401 1402 1403 1404

  // log the reason why load the datablock for profile
  if (loadDataBlock) {
    tsdbDebug("%p uid:%" PRIu64
1405
              " need to load the datablock, overlapwithneighborblock:%d, hasDup:%d, partiallyRequired:%d, "
1406
              "overlapWithKey:%d, greaterThanBuf:%d, overlapWithDel:%d, overlapWithlastBlock:%d, %s",
H
Haojun Liao 已提交
1407 1408 1409
              pReader, pBlockInfo->uid, info.overlapWithNeighborBlock, info.hasDupTs, info.partiallyRequired,
              info.overlapWithKeyInBuf, info.moreThanCapcity, info.overlapWithDelInfo, info.overlapWithLastBlock,
              pReader->idStr);
1410 1411 1412
  }

  return loadDataBlock;
H
Haojun Liao 已提交
1413 1414
}

H
Haojun Liao 已提交
1415 1416 1417 1418 1419 1420 1421 1422 1423
static bool isCleanFileDataBlock(STsdbReader* pReader, SFileDataBlockInfo* pBlockInfo, SDataBlk* pBlock,
                                 STableBlockScanInfo* pScanInfo, TSDBKEY keyInBuf, SLastBlockReader* pLastBlockReader) {
  SDataBlockToLoadInfo info = {0};
  getBlockToLoadInfo(&info, pBlockInfo, pBlock, pScanInfo, keyInBuf, pLastBlockReader, pReader);
  bool isCleanFileBlock = !(info.overlapWithNeighborBlock || info.hasDupTs || info.overlapWithKeyInBuf ||
                            info.overlapWithDelInfo || info.overlapWithLastBlock);
  return isCleanFileBlock;
}

1424
static int32_t buildDataBlockFromBuf(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo, int64_t endKey) {
1425
  if (!(pBlockScanInfo->iiter.hasVal || pBlockScanInfo->iter.hasVal)) {
1426 1427
    return TSDB_CODE_SUCCESS;
  }
H
Haojun Liao 已提交
1428

1429 1430 1431
  SSDataBlock* pBlock = pReader->pResBlock;

  int64_t st = taosGetTimestampUs();
1432
  int32_t code = buildDataBlockFromBufImpl(pBlockScanInfo, endKey, pReader->capacity, pReader);
H
Haojun Liao 已提交
1433

1434
  blockDataUpdateTsWindow(pBlock, 0);
1435
  pBlock->info.uid = pBlockScanInfo->uid;
1436

1437
  setComposedBlockFlag(pReader, true);
1438

1439
  double elapsedTime = (taosGetTimestampUs() - st) / 1000.0;
S
Shengliang Guan 已提交
1440
  tsdbDebug("%p build data block from cache completed, elapsed time:%.2f ms, numOfRows:%d, brange:%" PRId64
1441 1442 1443
            " - %" PRId64 " %s",
            pReader, elapsedTime, pBlock->info.rows, pBlock->info.window.skey, pBlock->info.window.ekey,
            pReader->idStr);
1444 1445

  pReader->cost.buildmemBlock += elapsedTime;
H
Haojun Liao 已提交
1446 1447 1448
  return code;
}

1449 1450
static bool tryCopyDistinctRowFromFileBlock(STsdbReader* pReader, SBlockData* pBlockData, int64_t key,
                                            SFileBlockDumpInfo* pDumpInfo) {
1451 1452 1453 1454 1455
  // opt version
  // 1. it is not a border point
  // 2. the direct next point is not an duplicated timestamp
  if ((pDumpInfo->rowIndex < pDumpInfo->totalRows - 1 && pReader->order == TSDB_ORDER_ASC) ||
      (pDumpInfo->rowIndex > 0 && pReader->order == TSDB_ORDER_DESC)) {
1456
    int32_t step = pReader->order == TSDB_ORDER_ASC ? 1 : -1;
1457 1458

    int64_t nextKey = pBlockData->aTSKEY[pDumpInfo->rowIndex + step];
1459
    if (nextKey != key) {  // merge is not needed
1460
      doAppendRowFromFileBlock(pReader->pResBlock, pReader, pBlockData, pDumpInfo->rowIndex);
1461 1462 1463 1464 1465 1466 1467 1468
      pDumpInfo->rowIndex += step;
      return true;
    }
  }

  return false;
}

1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500
static bool nextRowFromLastBlocks(SLastBlockReader* pLastBlockReader, STableBlockScanInfo* pBlockScanInfo) {
  while (1) {
    bool hasVal = tMergeTreeNext(&pLastBlockReader->mergeTree);
    if (!hasVal) {
      return false;
    }

    TSDBROW row = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
    TSDBKEY k = TSDBROW_KEY(&row);
    if (!hasBeenDropped(pBlockScanInfo->delSkyline, &pBlockScanInfo->lastBlockDelIndex, &k, pLastBlockReader->order)) {
      return true;
    }
  }
}

static bool tryCopyDistinctRowFromSttBlock(TSDBROW* fRow, SLastBlockReader* pLastBlockReader,
                                           STableBlockScanInfo* pScanInfo, int64_t ts, STsdbReader* pReader) {
  bool hasVal = nextRowFromLastBlocks(pLastBlockReader, pScanInfo);
  if (hasVal) {
    int64_t next1 = getCurrentKeyInLastBlock(pLastBlockReader);
    if (next1 != ts) {
      doAppendRowFromFileBlock(pReader->pResBlock, pReader, fRow->pBlockData, fRow->iRow);
      return true;
    }
  } else {
    doAppendRowFromFileBlock(pReader->pResBlock, pReader, fRow->pBlockData, fRow->iRow);
    return true;
  }

  return false;
}

H
Haojun Liao 已提交
1501 1502 1503
static FORCE_INLINE STSchema* doGetSchemaForTSRow(int32_t sversion, STsdbReader* pReader, uint64_t uid) {
  // always set the newest schema version in pReader->pSchema
  if (pReader->pSchema == NULL) {
1504
    pReader->pSchema = metaGetTbTSchema(pReader->pTsdb->pVnode->pMeta, uid, -1, 1);
H
Haojun Liao 已提交
1505 1506
  }

1507
  if (pReader->pSchema && sversion == pReader->pSchema->version) {
H
Haojun Liao 已提交
1508 1509 1510 1511 1512 1513
    return pReader->pSchema;
  }

  if (pReader->pMemSchema == NULL) {
    int32_t code =
        metaGetTbTSchemaEx(pReader->pTsdb->pVnode->pMeta, pReader->suid, uid, sversion, &pReader->pMemSchema);
H
Haojun Liao 已提交
1514 1515 1516 1517 1518 1519
    if (code != TSDB_CODE_SUCCESS) {
      terrno = code;
      return NULL;
    } else {
      return pReader->pMemSchema;
    }
H
Haojun Liao 已提交
1520 1521 1522 1523 1524 1525
  }

  if (pReader->pMemSchema->version == sversion) {
    return pReader->pMemSchema;
  }

H
Haojun Liao 已提交
1526
  taosMemoryFreeClear(pReader->pMemSchema);
H
Haojun Liao 已提交
1527
  int32_t code = metaGetTbTSchemaEx(pReader->pTsdb->pVnode->pMeta, pReader->suid, uid, sversion, &pReader->pMemSchema);
H
Haojun Liao 已提交
1528
  if (code != TSDB_CODE_SUCCESS || pReader->pMemSchema == NULL) {
H
Haojun Liao 已提交
1529 1530
    terrno = code;
    return NULL;
H
Haojun Liao 已提交
1531 1532
  } else {
    return pReader->pMemSchema;
H
Haojun Liao 已提交
1533
  }
H
Haojun Liao 已提交
1534 1535
}

1536
static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo, TSDBROW* pRow,
1537 1538 1539 1540 1541 1542
                                     SIterInfo* pIter, int64_t key, SLastBlockReader* pLastBlockReader) {
  SRowMerger          merge = {0};
  STSRow*             pTSRow = NULL;
  SBlockData*         pBlockData = &pReader->status.fileBlockData;
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;

1543
  int64_t tsLast = INT64_MIN;
1544
  if (hasDataInLastBlock(pLastBlockReader)) {
1545 1546
    tsLast = getCurrentKeyInLastBlock(pLastBlockReader);
  }
1547

H
Hongze Cheng 已提交
1548 1549
  TSDBKEY k = TSDBROW_KEY(pRow);
  TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
1550

1551 1552
  int64_t minKey = 0;
  if (pReader->order == TSDB_ORDER_ASC) {
H
Hongze Cheng 已提交
1553
    minKey = INT64_MAX;  // chosen the minimum value
1554
    if (minKey > tsLast && hasDataInLastBlock(pLastBlockReader)) {
1555 1556
      minKey = tsLast;
    }
1557

1558 1559 1560
    if (minKey > k.ts) {
      minKey = k.ts;
    }
1561

1562
    if (minKey > key && hasDataInFileBlock(pBlockData, pDumpInfo)) {
1563 1564 1565 1566
      minKey = key;
    }
  } else {
    minKey = INT64_MIN;
1567
    if (minKey < tsLast && hasDataInLastBlock(pLastBlockReader)) {
1568 1569 1570 1571 1572 1573 1574
      minKey = tsLast;
    }

    if (minKey < k.ts) {
      minKey = k.ts;
    }

1575
    if (minKey < key && hasDataInFileBlock(pBlockData, pDumpInfo)) {
1576 1577
      minKey = key;
    }
1578 1579 1580 1581
  }

  bool init = false;

1582
  // ASC: file block ---> last block -----> imem -----> mem
H
Hongze Cheng 已提交
1583
  // DESC: mem -----> imem -----> last block -----> file block
1584 1585
  if (pReader->order == TSDB_ORDER_ASC) {
    if (minKey == key) {
1586
      init = true;
H
Haojun Liao 已提交
1587 1588 1589 1590
      int32_t code = tRowMergerInit(&merge, &fRow, pReader->pSchema);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
1591
      doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
1592 1593
    }

1594
    if (minKey == tsLast) {
1595
      TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
H
Haojun Liao 已提交
1596 1597 1598
      if (init) {
        tRowMerge(&merge, &fRow1);
      } else {
1599
        init = true;
H
Haojun Liao 已提交
1600 1601 1602 1603
        int32_t code = tRowMergerInit(&merge, &fRow1, pReader->pSchema);
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
1604
      }
H
Haojun Liao 已提交
1605
      doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, &merge);
1606
    }
1607

1608
    if (minKey == k.ts) {
H
Haojun Liao 已提交
1609 1610 1611
      if (init) {
        tRowMerge(&merge, pRow);
      } else {
1612 1613
        init = true;
        STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(pRow), pReader, pBlockScanInfo->uid);
H
Haojun Liao 已提交
1614 1615 1616 1617 1618 1619 1620 1621
        int32_t   code = tRowMergerInit(&merge, pRow, pSchema);
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
      }
      int32_t code = doMergeRowsInBuf(pIter, pBlockScanInfo->uid, k.ts, pBlockScanInfo->delSkyline, &merge, pReader);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
1622 1623 1624 1625 1626
      }
    }
  } else {
    if (minKey == k.ts) {
      init = true;
1627
      STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(pRow), pReader, pBlockScanInfo->uid);
H
Haojun Liao 已提交
1628 1629 1630 1631 1632 1633
      int32_t   code = tRowMergerInit(&merge, pRow, pSchema);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }

      code = doMergeRowsInBuf(pIter, pBlockScanInfo->uid, k.ts, pBlockScanInfo->delSkyline, &merge, pReader);
1634
      if (code != TSDB_CODE_SUCCESS || merge.pTSchema == NULL) {
H
Haojun Liao 已提交
1635 1636
        return code;
      }
1637 1638
    }

1639
    if (minKey == tsLast) {
1640
      TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
H
Haojun Liao 已提交
1641 1642 1643
      if (init) {
        tRowMerge(&merge, &fRow1);
      } else {
1644
        init = true;
H
Haojun Liao 已提交
1645 1646 1647 1648
        int32_t code = tRowMergerInit(&merge, &fRow1, pReader->pSchema);
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
1649
      }
H
Haojun Liao 已提交
1650
      doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, &merge);
1651 1652 1653
    }

    if (minKey == key) {
H
Haojun Liao 已提交
1654 1655 1656
      if (init) {
        tRowMerge(&merge, &fRow);
      } else {
1657
        init = true;
H
Haojun Liao 已提交
1658 1659 1660 1661
        int32_t code = tRowMergerInit(&merge, &fRow, pReader->pSchema);
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
1662 1663 1664
      }
      doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
    }
1665 1666
  }

1667 1668 1669 1670 1671
  int32_t code = tRowMergerGetRow(&merge, &pTSRow);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

1672 1673 1674 1675 1676 1677 1678
  doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo->uid);

  taosMemoryFree(pTSRow);
  tRowMergerClear(&merge);
  return TSDB_CODE_SUCCESS;
}

1679 1680 1681
static int32_t doMergeFileBlockAndLastBlock(SLastBlockReader* pLastBlockReader, STsdbReader* pReader,
                                            STableBlockScanInfo* pBlockScanInfo, SBlockData* pBlockData,
                                            bool mergeBlockData) {
1682
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
1683
  int64_t             tsLastBlock = getCurrentKeyInLastBlock(pLastBlockReader);
1684 1685 1686

  STSRow*    pTSRow = NULL;
  SRowMerger merge = {0};
1687
  TSDBROW    fRow = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
H
Hongze Cheng 已提交
1688
  tsdbTrace("fRow ptr:%p, %d, uid:%" PRIu64 ", %s", fRow.pBlockData, fRow.iRow, pLastBlockReader->uid, pReader->idStr);
1689

1690 1691 1692 1693 1694
  // only last block exists
  if ((!mergeBlockData) || (tsLastBlock != pBlockData->aTSKEY[pDumpInfo->rowIndex])) {
    if (tryCopyDistinctRowFromSttBlock(&fRow, pLastBlockReader, pBlockScanInfo, tsLastBlock, pReader)) {
      return TSDB_CODE_SUCCESS;
    } else {
H
Haojun Liao 已提交
1695 1696 1697 1698
      int32_t code = tRowMergerInit(&merge, &fRow, pReader->pSchema);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
1699 1700 1701

      TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
      tRowMerge(&merge, &fRow1);
1702
      doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLastBlock, &merge);
1703

H
Haojun Liao 已提交
1704
      code = tRowMergerGetRow(&merge, &pTSRow);
1705 1706 1707
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
1708

1709 1710 1711 1712 1713 1714
      doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo->uid);

      taosMemoryFree(pTSRow);
      tRowMergerClear(&merge);
    }
  } else {  // not merge block data
H
Haojun Liao 已提交
1715 1716 1717 1718 1719
    int32_t code = tRowMergerInit(&merge, &fRow, pReader->pSchema);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

1720
    doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLastBlock, &merge);
H
Haojun Liao 已提交
1721
    ASSERT(mergeBlockData);
1722 1723

    // merge with block data if ts == key
H
Haojun Liao 已提交
1724
    if (tsLastBlock == pBlockData->aTSKEY[pDumpInfo->rowIndex]) {
1725 1726 1727
      doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
    }

H
Haojun Liao 已提交
1728
    code = tRowMergerGetRow(&merge, &pTSRow);
1729 1730 1731 1732 1733 1734 1735 1736 1737
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

    doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo->uid);

    taosMemoryFree(pTSRow);
    tRowMergerClear(&merge);
  }
1738 1739 1740 1741

  return TSDB_CODE_SUCCESS;
}

1742 1743
static int32_t mergeFileBlockAndLastBlock(STsdbReader* pReader, SLastBlockReader* pLastBlockReader, int64_t key,
                                          STableBlockScanInfo* pBlockScanInfo, SBlockData* pBlockData) {
1744 1745
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;

1746
  if (hasDataInFileBlock(pBlockData, pDumpInfo)) {
1747
    // no last block available, only data block exists
1748
    if (!hasDataInLastBlock(pLastBlockReader)) {
1749 1750 1751 1752 1753 1754 1755 1756 1757
      return mergeRowsInFileBlocks(pBlockData, pBlockScanInfo, key, pReader);
    }

    // row in last file block
    TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
    int64_t ts = getCurrentKeyInLastBlock(pLastBlockReader);
    ASSERT(ts >= key);

    if (ASCENDING_TRAVERSE(pReader->order)) {
1758
      if (key < ts) {  // imem, mem are all empty, file blocks (data blocks and last block) exist
1759 1760 1761 1762
        return mergeRowsInFileBlocks(pBlockData, pBlockScanInfo, key, pReader);
      } else if (key == ts) {
        STSRow*    pTSRow = NULL;
        SRowMerger merge = {0};
1763

H
Haojun Liao 已提交
1764 1765 1766 1767 1768
        int32_t code = tRowMergerInit(&merge, &fRow, pReader->pSchema);
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }

1769
        doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
1770 1771 1772 1773

        TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
        tRowMerge(&merge, &fRow1);

1774
        doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, ts, &merge);
1775

H
Haojun Liao 已提交
1776
        code = tRowMergerGetRow(&merge, &pTSRow);
1777 1778 1779 1780
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }

1781
        doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo->uid);
1782

1783 1784
        taosMemoryFree(pTSRow);
        tRowMergerClear(&merge);
1785
        return code;
1786
      } else {
1787 1788
        ASSERT(0);
        return TSDB_CODE_SUCCESS;
1789
      }
1790
    } else {  // desc order
1791
      return doMergeFileBlockAndLastBlock(pLastBlockReader, pReader, pBlockScanInfo, pBlockData, true);
1792
    }
1793
  } else {  // only last block exists
1794
    return doMergeFileBlockAndLastBlock(pLastBlockReader, pReader, pBlockScanInfo, NULL, false);
H
Haojun Liao 已提交
1795
  }
1796 1797
}

1798 1799
static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo, SBlockData* pBlockData,
                                     SLastBlockReader* pLastBlockReader) {
H
Haojun Liao 已提交
1800 1801 1802
  SRowMerger          merge = {0};
  STSRow*             pTSRow = NULL;
  int32_t             code = TSDB_CODE_SUCCESS;
1803 1804 1805
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
  SArray*             pDelList = pBlockScanInfo->delSkyline;

1806 1807
  TSDBROW* pRow = getValidMemRow(&pBlockScanInfo->iter, pDelList, pReader);
  TSDBROW* piRow = getValidMemRow(&pBlockScanInfo->iiter, pDelList, pReader);
1808 1809
  ASSERT(pRow != NULL && piRow != NULL);

1810
  int64_t tsLast = INT64_MIN;
1811 1812 1813
  if (hasDataInLastBlock(pLastBlockReader)) {
    tsLast = getCurrentKeyInLastBlock(pLastBlockReader);
  }
1814

H
Hongze Cheng 已提交
1815
  int64_t key = hasDataInFileBlock(pBlockData, pDumpInfo) ? pBlockData->aTSKEY[pDumpInfo->rowIndex] : INT64_MIN;
1816 1817 1818 1819

  TSDBKEY k = TSDBROW_KEY(pRow);
  TSDBKEY ik = TSDBROW_KEY(piRow);

1820
  int64_t minKey = 0;
1821 1822 1823 1824 1825
  if (ASCENDING_TRAVERSE(pReader->order)) {
    minKey = INT64_MAX;  // let's find the minimum
    if (minKey > k.ts) {
      minKey = k.ts;
    }
1826

1827 1828 1829
    if (minKey > ik.ts) {
      minKey = ik.ts;
    }
1830

1831
    if (minKey > key && hasDataInFileBlock(pBlockData, pDumpInfo)) {
1832 1833
      minKey = key;
    }
1834

1835
    if (minKey > tsLast && hasDataInLastBlock(pLastBlockReader)) {
1836 1837 1838
      minKey = tsLast;
    }
  } else {
H
Hongze Cheng 已提交
1839
    minKey = INT64_MIN;  // let find the maximum ts value
1840 1841 1842 1843 1844 1845 1846 1847
    if (minKey < k.ts) {
      minKey = k.ts;
    }

    if (minKey < ik.ts) {
      minKey = ik.ts;
    }

1848
    if (minKey < key && hasDataInFileBlock(pBlockData, pDumpInfo)) {
1849 1850 1851
      minKey = key;
    }

1852
    if (minKey < tsLast && hasDataInLastBlock(pLastBlockReader)) {
1853 1854
      minKey = tsLast;
    }
1855 1856 1857 1858
  }

  bool init = false;

1859 1860 1861 1862
  // ASC: file block -----> last block -----> imem -----> mem
  // DESC: mem -----> imem -----> last block -----> file block
  if (ASCENDING_TRAVERSE(pReader->order)) {
    if (minKey == key) {
1863
      init = true;
1864
      TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
H
Haojun Liao 已提交
1865 1866 1867 1868 1869
      int32_t code = tRowMergerInit(&merge, &fRow, pReader->pSchema);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }

1870
      doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
1871 1872
    }

1873
    if (minKey == tsLast) {
1874
      TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
H
Haojun Liao 已提交
1875 1876 1877
      if (init) {
        tRowMerge(&merge, &fRow1);
      } else {
1878
        init = true;
H
Haojun Liao 已提交
1879 1880 1881 1882
        int32_t code = tRowMergerInit(&merge, &fRow1, pReader->pSchema);
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
1883
      }
H
Haojun Liao 已提交
1884

H
Haojun Liao 已提交
1885
      doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, &merge);
1886 1887 1888
    }

    if (minKey == ik.ts) {
H
Haojun Liao 已提交
1889 1890 1891
      if (init) {
        tRowMerge(&merge, piRow);
      } else {
1892 1893
        init = true;
        STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(piRow), pReader, pBlockScanInfo->uid);
H
Haojun Liao 已提交
1894 1895 1896 1897
        if (pSchema == NULL) {
          return code;
        }

H
Haojun Liao 已提交
1898
        code = tRowMergerInit(&merge, piRow, pSchema);
H
Haojun Liao 已提交
1899 1900 1901
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
1902
      }
H
Haojun Liao 已提交
1903 1904 1905 1906 1907 1908

      code = doMergeRowsInBuf(&pBlockScanInfo->iiter, pBlockScanInfo->uid, ik.ts, pBlockScanInfo->delSkyline,
                                      &merge, pReader);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
1909 1910
    }

1911
    if (minKey == k.ts) {
H
Haojun Liao 已提交
1912
      if (init) {
1913 1914 1915 1916
        if (merge.pTSchema == NULL) {
          return code;
        }

H
Haojun Liao 已提交
1917 1918
        tRowMerge(&merge, pRow);
      } else {
1919
        STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(pRow), pReader, pBlockScanInfo->uid);
H
Haojun Liao 已提交
1920
        code = tRowMergerInit(&merge, pRow, pSchema);
H
Haojun Liao 已提交
1921 1922 1923
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
1924
      }
H
Haojun Liao 已提交
1925 1926 1927 1928 1929
      code = doMergeRowsInBuf(&pBlockScanInfo->iter, pBlockScanInfo->uid, k.ts, pBlockScanInfo->delSkyline, &merge,
                              pReader);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
1930 1931 1932 1933
    }
  } else {
    if (minKey == k.ts) {
      init = true;
1934
      STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(pRow), pReader, pBlockScanInfo->uid);
H
Haojun Liao 已提交
1935
      code = tRowMergerInit(&merge, pRow, pSchema);
H
Haojun Liao 已提交
1936 1937 1938 1939
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }

H
Haojun Liao 已提交
1940 1941 1942 1943 1944
      code = doMergeRowsInBuf(&pBlockScanInfo->iter, pBlockScanInfo->uid, k.ts, pBlockScanInfo->delSkyline, &merge,
                              pReader);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
1945 1946 1947
    }

    if (minKey == ik.ts) {
H
Haojun Liao 已提交
1948 1949 1950
      if (init) {
        tRowMerge(&merge, piRow);
      } else {
1951 1952
        init = true;
        STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(piRow), pReader, pBlockScanInfo->uid);
H
Haojun Liao 已提交
1953
        code = tRowMergerInit(&merge, piRow, pSchema);
H
Haojun Liao 已提交
1954 1955 1956
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
1957
      }
H
Haojun Liao 已提交
1958 1959 1960 1961 1962
      code = doMergeRowsInBuf(&pBlockScanInfo->iiter, pBlockScanInfo->uid, ik.ts, pBlockScanInfo->delSkyline, &merge,
                              pReader);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
1963 1964 1965
    }

    if (minKey == tsLast) {
1966
      TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
H
Haojun Liao 已提交
1967 1968 1969
      if (init) {
        tRowMerge(&merge, &fRow1);
      } else {
1970
        init = true;
1971
        code = tRowMergerInit(&merge, &fRow1, pReader->pSchema);
H
Haojun Liao 已提交
1972 1973 1974
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
1975
      }
H
Haojun Liao 已提交
1976
      doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, &merge);
1977 1978 1979
    }

    if (minKey == key) {
H
Haojun Liao 已提交
1980
      TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
1981
      if (!init) {
1982
        code = tRowMergerInit(&merge, &fRow, pReader->pSchema);
H
Haojun Liao 已提交
1983 1984 1985
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
H
Haojun Liao 已提交
1986
      } else {
1987 1988 1989
        if (merge.pTSchema == NULL) {
          return code;
        }
H
Haojun Liao 已提交
1990
        tRowMerge(&merge, &fRow);
1991 1992
      }
      doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
1993 1994 1995
    }
  }

1996 1997 1998 1999
  if (merge.pTSchema == NULL)  {
    return code;
  }

H
Haojun Liao 已提交
2000
  code = tRowMergerGetRow(&merge, &pTSRow);
2001 2002 2003 2004
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

2005 2006 2007 2008
  doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo->uid);

  taosMemoryFree(pTSRow);
  tRowMergerClear(&merge);
2009
  return code;
2010 2011
}

2012 2013 2014 2015 2016 2017 2018 2019 2020 2021 2022 2023 2024 2025 2026 2027 2028 2029 2030 2031 2032 2033 2034 2035 2036
static int32_t initMemDataIterator(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader) {
  if (pBlockScanInfo->iterInit) {
    return TSDB_CODE_SUCCESS;
  }

  int32_t code = TSDB_CODE_SUCCESS;

  TSDBKEY startKey = {0};
  if (ASCENDING_TRAVERSE(pReader->order)) {
    startKey = (TSDBKEY){.ts = pReader->window.skey, .version = pReader->verRange.minVer};
  } else {
    startKey = (TSDBKEY){.ts = pReader->window.ekey, .version = pReader->verRange.maxVer};
  }

  int32_t backward = (!ASCENDING_TRAVERSE(pReader->order));

  STbData* d = NULL;
  if (pReader->pReadSnap->pMem != NULL) {
    d = tsdbGetTbDataFromMemTable(pReader->pReadSnap->pMem, pReader->suid, pBlockScanInfo->uid);
    if (d != NULL) {
      code = tsdbTbDataIterCreate(d, &startKey, backward, &pBlockScanInfo->iter.iter);
      if (code == TSDB_CODE_SUCCESS) {
        pBlockScanInfo->iter.hasVal = (tsdbTbDataIterGet(pBlockScanInfo->iter.iter) != NULL);

        tsdbDebug("%p uid:%" PRId64 ", check data in mem from skey:%" PRId64 ", order:%d, ts range in buf:%" PRId64
H
Hongze Cheng 已提交
2037
                  "-%" PRId64 " %s",
2038 2039 2040 2041 2042 2043 2044 2045 2046 2047 2048 2049 2050 2051 2052 2053 2054 2055 2056 2057
                  pReader, pBlockScanInfo->uid, startKey.ts, pReader->order, d->minKey, d->maxKey, pReader->idStr);
      } else {
        tsdbError("%p uid:%" PRId64 ", failed to create iterator for imem, code:%s, %s", pReader, pBlockScanInfo->uid,
                  tstrerror(code), pReader->idStr);
        return code;
      }
    }
  } else {
    tsdbDebug("%p uid:%" PRId64 ", no data in mem, %s", pReader, pBlockScanInfo->uid, pReader->idStr);
  }

  STbData* di = NULL;
  if (pReader->pReadSnap->pIMem != NULL) {
    di = tsdbGetTbDataFromMemTable(pReader->pReadSnap->pIMem, pReader->suid, pBlockScanInfo->uid);
    if (di != NULL) {
      code = tsdbTbDataIterCreate(di, &startKey, backward, &pBlockScanInfo->iiter.iter);
      if (code == TSDB_CODE_SUCCESS) {
        pBlockScanInfo->iiter.hasVal = (tsdbTbDataIterGet(pBlockScanInfo->iiter.iter) != NULL);

        tsdbDebug("%p uid:%" PRId64 ", check data in imem from skey:%" PRId64 ", order:%d, ts range in buf:%" PRId64
H
Hongze Cheng 已提交
2058
                  "-%" PRId64 " %s",
2059 2060 2061 2062 2063 2064 2065 2066 2067 2068 2069 2070 2071 2072 2073 2074 2075
                  pReader, pBlockScanInfo->uid, startKey.ts, pReader->order, di->minKey, di->maxKey, pReader->idStr);
      } else {
        tsdbError("%p uid:%" PRId64 ", failed to create iterator for mem, code:%s, %s", pReader, pBlockScanInfo->uid,
                  tstrerror(code), pReader->idStr);
        return code;
      }
    }
  } else {
    tsdbDebug("%p uid:%" PRId64 ", no data in imem, %s", pReader, pBlockScanInfo->uid, pReader->idStr);
  }

  initDelSkylineIterator(pBlockScanInfo, pReader, d, di);

  pBlockScanInfo->iterInit = true;
  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
2076 2077
static bool isValidFileBlockRow(SBlockData* pBlockData, SFileBlockDumpInfo* pDumpInfo,
                                STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader) {
2078 2079 2080 2081 2082 2083 2084 2085
  // it is an multi-table data block
  if (pBlockData->aUid != NULL) {
    uint64_t uid = pBlockData->aUid[pDumpInfo->rowIndex];
    if (uid != pBlockScanInfo->uid) {  // move to next row
      return false;
    }
  }

2086 2087 2088 2089 2090 2091 2092 2093 2094 2095 2096
  // check for version and time range
  int64_t ver = pBlockData->aVersion[pDumpInfo->rowIndex];
  if (ver > pReader->verRange.maxVer || ver < pReader->verRange.minVer) {
    return false;
  }

  int64_t ts = pBlockData->aTSKEY[pDumpInfo->rowIndex];
  if (ts > pReader->window.ekey || ts < pReader->window.skey) {
    return false;
  }

2097
  TSDBKEY k = {.ts = ts, .version = ver};
2098
  if (hasBeenDropped(pBlockScanInfo->delSkyline, &pBlockScanInfo->fileDelIndex, &k, pReader->order)) {
2099 2100 2101
    return false;
  }

2102 2103 2104
  return true;
}

2105
static bool initLastBlockReader(SLastBlockReader* pLBlockReader, STableBlockScanInfo* pScanInfo, STsdbReader* pReader) {
2106
  // the last block reader has been initialized for this table.
2107
  if (pLBlockReader->uid == pScanInfo->uid) {
2108 2109 2110
    return true;
  }

2111 2112
  if (pLBlockReader->uid != 0) {
    tMergeTreeClose(&pLBlockReader->mergeTree);
2113 2114
  }

2115 2116
  initMemDataIterator(pScanInfo, pReader);
  pLBlockReader->uid = pScanInfo->uid;
2117

H
Hongze Cheng 已提交
2118
  int32_t     step = ASCENDING_TRAVERSE(pLBlockReader->order) ? 1 : -1;
2119 2120 2121
  STimeWindow w = pLBlockReader->window;
  if (ASCENDING_TRAVERSE(pLBlockReader->order)) {
    w.skey = pScanInfo->lastKey + step;
2122
  } else {
2123
    w.ekey = pScanInfo->lastKey + step;
2124 2125
  }

2126 2127 2128
  int32_t code = tMergeTreeOpen(&pLBlockReader->mergeTree, (pLBlockReader->order == TSDB_ORDER_DESC),
                                pReader->pFileReader, pReader->suid, pScanInfo->uid, &w, &pLBlockReader->verRange,
                                pLBlockReader->pInfo, false, pReader->idStr);
2129 2130 2131 2132
  if (code != TSDB_CODE_SUCCESS) {
    return false;
  }

2133
  return nextRowFromLastBlocks(pLBlockReader, pScanInfo);
2134 2135
}

2136
static int64_t getCurrentKeyInLastBlock(SLastBlockReader* pLastBlockReader) {
2137
  TSDBROW row = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
2138
  return TSDBROW_TS(&row);
2139 2140
}

H
Hongze Cheng 已提交
2141
static bool hasDataInLastBlock(SLastBlockReader* pLastBlockReader) { return pLastBlockReader->mergeTree.pIter != NULL; }
2142 2143 2144 2145

bool hasDataInFileBlock(const SBlockData* pBlockData, const SFileBlockDumpInfo* pDumpInfo) {
  if (pBlockData->nRow > 0) {
    ASSERT(pBlockData->nRow == pDumpInfo->totalRows);
2146 2147
  }

2148
  return pBlockData->nRow > 0 && (!pDumpInfo->allDumped);
2149
}
2150

2151 2152
int32_t mergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pBlockScanInfo, int64_t key,
                              STsdbReader* pReader) {
2153 2154 2155 2156
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
  if (tryCopyDistinctRowFromFileBlock(pReader, pBlockData, key, pDumpInfo)) {
    return TSDB_CODE_SUCCESS;
  } else {
2157 2158
    TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);

2159 2160 2161
    STSRow*    pTSRow = NULL;
    SRowMerger merge = {0};

H
Haojun Liao 已提交
2162 2163 2164 2165 2166
    int32_t code = tRowMergerInit(&merge, &fRow, pReader->pSchema);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

2167
    doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
H
Haojun Liao 已提交
2168
    code = tRowMergerGetRow(&merge, &pTSRow);
2169 2170 2171 2172
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

2173 2174 2175 2176 2177 2178 2179 2180
    doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo->uid);

    taosMemoryFree(pTSRow);
    tRowMergerClear(&merge);
    return TSDB_CODE_SUCCESS;
  }
}

H
Haojun Liao 已提交
2181 2182
static int32_t buildComposedDataBlockImpl(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo,
                                          SBlockData* pBlockData, SLastBlockReader* pLastBlockReader) {
2183 2184
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;

2185
  int64_t key = (pBlockData->nRow > 0 && (!pDumpInfo->allDumped)) ? pBlockData->aTSKEY[pDumpInfo->rowIndex] : INT64_MIN;
2186
  if (pBlockScanInfo->iter.hasVal && pBlockScanInfo->iiter.hasVal) {
2187
    return doMergeMultiLevelRows(pReader, pBlockScanInfo, pBlockData, pLastBlockReader);
2188
  } else {
2189 2190 2191 2192 2193 2194 2195 2196 2197
    TSDBROW *pRow = NULL, *piRow = NULL;
    if (pBlockScanInfo->iter.hasVal) {
      pRow = getValidMemRow(&pBlockScanInfo->iter, pBlockScanInfo->delSkyline, pReader);
    }

    if (pBlockScanInfo->iiter.hasVal) {
      piRow = getValidMemRow(&pBlockScanInfo->iiter, pBlockScanInfo->delSkyline, pReader);
    }

2198
    // imem + file + last block
2199
    if (pBlockScanInfo->iiter.hasVal) {
2200
      return doMergeBufAndFileRows(pReader, pBlockScanInfo, piRow, &pBlockScanInfo->iiter, key, pLastBlockReader);
2201 2202
    }

2203
    // mem + file + last block
2204
    if (pBlockScanInfo->iter.hasVal) {
2205
      return doMergeBufAndFileRows(pReader, pBlockScanInfo, pRow, &pBlockScanInfo->iter, key, pLastBlockReader);
H
Haojun Liao 已提交
2206
    }
2207

2208 2209
    // files data blocks + last block
    return mergeFileBlockAndLastBlock(pReader, pLastBlockReader, key, pBlockScanInfo, pBlockData);
2210 2211 2212
  }
}

2213
static int32_t buildComposedDataBlock(STsdbReader* pReader) {
H
Haojun Liao 已提交
2214 2215
  int32_t code = TSDB_CODE_SUCCESS;

2216 2217
  SSDataBlock* pResBlock = pReader->pResBlock;

H
Hongze Cheng 已提交
2218
  SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(&pReader->status.blockIter);
H
Haojun Liao 已提交
2219 2220 2221
  SLastBlockReader*   pLastBlockReader = pReader->status.fileIter.pLastBlockReader;

  int64_t st = taosGetTimestampUs();
2222 2223 2224 2225

  STableBlockScanInfo* pBlockScanInfo = NULL;
  if (pBlockInfo != NULL) {
    pBlockScanInfo = taosHashGet(pReader->status.pTableMap, &pBlockInfo->uid, sizeof(pBlockInfo->uid));
H
Haojun Liao 已提交
2226 2227 2228 2229 2230 2231 2232
    if (pBlockScanInfo == NULL) {
      code = TSDB_CODE_INVALID_PARA;
      tsdbError("failed to locate the uid:%"PRIu64" in query table uid list, total tables:%d, %s",
          pBlockInfo->uid, taosHashGetSize(pReader->status.pTableMap), pReader->idStr);
      goto _end;
    }

H
Haojun Liao 已提交
2233
    SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter);
H
Hongze Cheng 已提交
2234
    TSDBKEY   keyInBuf = getCurrentKeyInBuf(pBlockScanInfo, pReader);
H
Haojun Liao 已提交
2235 2236 2237

    // it is a clean block, load it directly
    if (isCleanFileDataBlock(pReader, pBlockInfo, pBlock, pBlockScanInfo, keyInBuf, pLastBlockReader)) {
H
Hongze Cheng 已提交
2238 2239
      if (pReader->order == TSDB_ORDER_ASC ||
          (pReader->order == TSDB_ORDER_DESC && (!hasDataInLastBlock(pLastBlockReader)))) {
H
Haojun Liao 已提交
2240 2241 2242
        copyBlockDataToSDataBlock(pReader, pBlockScanInfo);
        goto _end;
      }
H
Haojun Liao 已提交
2243 2244
    }
  } else {  // file blocks not exist
2245 2246 2247
    pBlockScanInfo = pReader->status.pTableIter;
  }

2248
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
2249 2250
  SBlockData*         pBlockData = &pReader->status.fileBlockData;
  int32_t             step = ASCENDING_TRAVERSE(pReader->order) ? 1 : -1;
2251

2252
  while (1) {
2253
    bool hasBlockData = false;
2254
    {
H
Haojun Liao 已提交
2255
      while (pBlockData->nRow > 0) {  // find the first qualified row in data block
2256 2257 2258 2259 2260
        if (isValidFileBlockRow(pBlockData, pDumpInfo, pBlockScanInfo, pReader)) {
          hasBlockData = true;
          break;
        }

2261 2262
        pDumpInfo->rowIndex += step;

H
Hongze Cheng 已提交
2263
        SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter);
2264
        if (pDumpInfo->rowIndex >= pBlock->nRow || pDumpInfo->rowIndex < 0) {
2265
          setBlockAllDumped(pDumpInfo, pBlock->maxKey.ts, pReader->order);
2266 2267 2268
          break;
        }
      }
2269
    }
2270

2271
    bool hasBlockLData = hasDataInLastBlock(pLastBlockReader);
2272

2273 2274 2275
    // no data in last block and block, no need to proceed.
    if ((hasBlockData == false) && (hasBlockLData == false)) {
      break;
2276 2277
    }

2278
    buildComposedDataBlockImpl(pReader, pBlockScanInfo, pBlockData, pLastBlockReader);
2279

2280
    // currently loaded file data block is consumed
2281
    if ((pBlockData->nRow > 0) && (pDumpInfo->rowIndex >= pBlockData->nRow || pDumpInfo->rowIndex < 0)) {
H
Hongze Cheng 已提交
2282
      SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter);
2283
      setBlockAllDumped(pDumpInfo, pBlock->maxKey.ts, pReader->order);
2284 2285 2286 2287 2288
      break;
    }

    if (pResBlock->info.rows >= pReader->capacity) {
      break;
2289 2290 2291
    }
  }

H
Hongze Cheng 已提交
2292
_end:
H
Haojun Liao 已提交
2293
  pResBlock->info.uid = (pBlockScanInfo != NULL)? pBlockScanInfo->uid:0;
2294 2295
  blockDataUpdateTsWindow(pResBlock, 0);

2296
  setComposedBlockFlag(pReader, true);
H
Hongze Cheng 已提交
2297
  double el = (taosGetTimestampUs() - st) / 1000.0;
H
Haojun Liao 已提交
2298 2299 2300

  pReader->cost.composedBlocks += 1;
  pReader->cost.buildComposedBlockTime += el;
2301

2302 2303 2304
  if (pResBlock->info.rows > 0) {
    tsdbDebug("%p uid:%" PRIu64 ", composed data block created, brange:%" PRIu64 "-%" PRIu64
              " rows:%d, elapsed time:%.2f ms %s",
H
Haojun Liao 已提交
2305
              pReader, pResBlock->info.uid, pResBlock->info.window.skey, pResBlock->info.window.ekey,
H
Haojun Liao 已提交
2306
              pResBlock->info.rows, el, pReader->idStr);
2307
  }
2308

H
Haojun Liao 已提交
2309
  return code;
2310 2311 2312 2313
}

void setComposedBlockFlag(STsdbReader* pReader, bool composed) { pReader->status.composedDataBlock = composed; }

dengyihao's avatar
dengyihao 已提交
2314 2315
int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader, STbData* pMemTbData,
                               STbData* piMemTbData) {
2316 2317 2318
  if (pBlockScanInfo->delSkyline != NULL) {
    return TSDB_CODE_SUCCESS;
  }
2319

2320 2321 2322
  int32_t code = 0;
  STsdb*  pTsdb = pReader->pTsdb;

2323 2324
  SArray* pDelData = taosArrayInit(4, sizeof(SDelData));

H
Hongze Cheng 已提交
2325
  SDelFile* pDelFile = pReader->pReadSnap->fs.pDelFile;
2326 2327
  if (pDelFile) {
    SDelFReader* pDelFReader = NULL;
H
more  
Hongze Cheng 已提交
2328
    code = tsdbDelFReaderOpen(&pDelFReader, pDelFile, pTsdb);
2329
    if (code != TSDB_CODE_SUCCESS) {
2330 2331 2332 2333 2334
      goto _err;
    }

    SArray* aDelIdx = taosArrayInit(4, sizeof(SDelIdx));
    if (aDelIdx == NULL) {
2335
      tsdbDelFReaderClose(&pDelFReader);
2336 2337 2338
      goto _err;
    }

H
Hongze Cheng 已提交
2339
    code = tsdbReadDelIdx(pDelFReader, aDelIdx);
2340 2341 2342
    if (code != TSDB_CODE_SUCCESS) {
      taosArrayDestroy(aDelIdx);
      tsdbDelFReaderClose(&pDelFReader);
2343 2344
      goto _err;
    }
2345

2346 2347 2348
    SDelIdx  idx = {.suid = pReader->suid, .uid = pBlockScanInfo->uid};
    SDelIdx* pIdx = taosArraySearch(aDelIdx, &idx, tCmprDelIdx, TD_EQ);

H
Haojun Liao 已提交
2349
    if (pIdx != NULL) {
H
Hongze Cheng 已提交
2350
      code = tsdbReadDelData(pDelFReader, pIdx, pDelData);
2351 2352 2353 2354 2355 2356 2357
    }

    taosArrayDestroy(aDelIdx);
    tsdbDelFReaderClose(&pDelFReader);

    if (code != TSDB_CODE_SUCCESS) {
      goto _err;
2358
    }
2359
  }
2360

2361 2362 2363 2364 2365 2366 2367
  SDelData* p = NULL;
  if (pMemTbData != NULL) {
    p = pMemTbData->pHead;
    while (p) {
      taosArrayPush(pDelData, p);
      p = p->pNext;
    }
2368 2369
  }

2370 2371 2372 2373 2374 2375 2376 2377 2378 2379 2380 2381 2382 2383
  if (piMemTbData != NULL) {
    p = piMemTbData->pHead;
    while (p) {
      taosArrayPush(pDelData, p);
      p = p->pNext;
    }
  }

  if (taosArrayGetSize(pDelData) > 0) {
    pBlockScanInfo->delSkyline = taosArrayInit(4, sizeof(TSDBKEY));
    code = tsdbBuildDeleteSkyline(pDelData, 0, (int32_t)(taosArrayGetSize(pDelData) - 1), pBlockScanInfo->delSkyline);
  }

  taosArrayDestroy(pDelData);
dengyihao's avatar
dengyihao 已提交
2384 2385
  pBlockScanInfo->iter.index =
      ASCENDING_TRAVERSE(pReader->order) ? 0 : taosArrayGetSize(pBlockScanInfo->delSkyline) - 1;
2386 2387
  pBlockScanInfo->iiter.index = pBlockScanInfo->iter.index;
  pBlockScanInfo->fileDelIndex = pBlockScanInfo->iter.index;
2388
  pBlockScanInfo->lastBlockDelIndex = pBlockScanInfo->iter.index;
2389 2390
  return code;

2391 2392 2393
_err:
  taosArrayDestroy(pDelData);
  return code;
2394 2395
}

H
Haojun Liao 已提交
2396
TSDBKEY getCurrentKeyInBuf(STableBlockScanInfo* pScanInfo, STsdbReader* pReader) {
H
Hongze Cheng 已提交
2397
  TSDBKEY  key = {.ts = TSKEY_INITIAL_VAL};
2398
  TSDBROW* pRow = getValidMemRow(&pScanInfo->iter, pScanInfo->delSkyline, pReader);
2399
  if (pRow != NULL) {
2400 2401 2402
    key = TSDBROW_KEY(pRow);
  }

2403
  pRow = getValidMemRow(&pScanInfo->iiter, pScanInfo->delSkyline, pReader);
2404
  if (pRow != NULL) {
2405 2406 2407 2408 2409 2410 2411 2412 2413
    TSDBKEY k = TSDBROW_KEY(pRow);
    if (key.ts > k.ts) {
      key = k;
    }
  }

  return key;
}

2414
static int32_t moveToNextFile(STsdbReader* pReader, SBlockNumber* pBlockNum) {
H
Haojun Liao 已提交
2415
  SReaderStatus* pStatus = &pReader->status;
2416
  pBlockNum->numOfBlocks = 0;
2417
  pBlockNum->numOfLastFiles = 0;
2418

2419
  size_t  numOfTables = taosHashGetSize(pReader->status.pTableMap);
2420
  SArray* pIndexList = taosArrayInit(numOfTables, sizeof(SBlockIdx));
H
Haojun Liao 已提交
2421 2422

  while (1) {
2423
    bool hasNext = filesetIteratorNext(&pStatus->fileIter, pReader);
2424
    if (!hasNext) {  // no data files on disk
H
Haojun Liao 已提交
2425 2426 2427
      break;
    }

H
Haojun Liao 已提交
2428
    taosArrayClear(pIndexList);
H
Haojun Liao 已提交
2429 2430
    int32_t code = doLoadBlockIndex(pReader, pReader->pFileReader, pIndexList);
    if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2431
      taosArrayDestroy(pIndexList);
H
Haojun Liao 已提交
2432 2433 2434
      return code;
    }

H
Hongze Cheng 已提交
2435
    if (taosArrayGetSize(pIndexList) > 0 || pReader->pFileReader->pSet->nSttF > 0) {
2436
      code = doLoadFileBlock(pReader, pIndexList, pBlockNum);
H
Haojun Liao 已提交
2437
      if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2438
        taosArrayDestroy(pIndexList);
H
Haojun Liao 已提交
2439 2440 2441
        return code;
      }

2442
      if (pBlockNum->numOfBlocks + pBlockNum->numOfLastFiles > 0) {
H
Haojun Liao 已提交
2443 2444 2445
        break;
      }
    }
2446

H
Haojun Liao 已提交
2447 2448 2449
    // no blocks in current file, try next files
  }

H
Haojun Liao 已提交
2450
  taosArrayDestroy(pIndexList);
H
Haojun Liao 已提交
2451 2452 2453
  return TSDB_CODE_SUCCESS;
}

2454
static int32_t uidComparFunc(const void* p1, const void* p2) {
2455 2456
  uint64_t pu1 = *(uint64_t*)p1;
  uint64_t pu2 = *(uint64_t*)p2;
2457 2458 2459
  if (pu1 == pu2) {
    return 0;
  } else {
2460
    return (pu1 < pu2) ? -1 : 1;
2461 2462
  }
}
2463

2464
static void extractOrderedTableUidList(SUidOrderCheckInfo* pOrderCheckInfo, SReaderStatus* pStatus) {
2465 2466 2467 2468
  int32_t index = 0;
  int32_t total = taosHashGetSize(pStatus->pTableMap);

  void* p = taosHashIterate(pStatus->pTableMap, NULL);
2469
  while (p != NULL) {
2470 2471 2472 2473 2474 2475 2476 2477
    STableBlockScanInfo* pScanInfo = p;
    pOrderCheckInfo->tableUidList[index++] = pScanInfo->uid;
    p = taosHashIterate(pStatus->pTableMap, p);
  }

  taosSort(pOrderCheckInfo->tableUidList, total, sizeof(uint64_t), uidComparFunc);
}

2478
static int32_t initOrderCheckInfo(SUidOrderCheckInfo* pOrderCheckInfo, SReaderStatus* pStatus) {
2479 2480 2481 2482
  int32_t total = taosHashGetSize(pStatus->pTableMap);
  if (total == 0) {
    return TSDB_CODE_SUCCESS;
  }
2483

2484
  if (pOrderCheckInfo->tableUidList == NULL) {
2485 2486 2487 2488 2489 2490
    pOrderCheckInfo->currentIndex = 0;
    pOrderCheckInfo->tableUidList = taosMemoryMalloc(total * sizeof(uint64_t));
    if (pOrderCheckInfo->tableUidList == NULL) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }

2491
    extractOrderedTableUidList(pOrderCheckInfo, pStatus);
2492 2493 2494
    uint64_t uid = pOrderCheckInfo->tableUidList[0];
    pStatus->pTableIter = taosHashGet(pStatus->pTableMap, &uid, sizeof(uid));
  } else {
2495 2496
    if (pStatus->pTableIter == NULL) {  // it is the last block of a new file
      pOrderCheckInfo->currentIndex = 0;
2497 2498
      uint64_t uid = pOrderCheckInfo->tableUidList[pOrderCheckInfo->currentIndex];
      pStatus->pTableIter = taosHashGet(pStatus->pTableMap, &uid, sizeof(uid));
2499 2500

      // the tableMap has already updated
2501
      if (pStatus->pTableIter == NULL) {
2502
        void* p = taosMemoryRealloc(pOrderCheckInfo->tableUidList, total * sizeof(uint64_t));
2503 2504 2505 2506 2507 2508 2509 2510 2511
        if (p == NULL) {
          return TSDB_CODE_OUT_OF_MEMORY;
        }

        pOrderCheckInfo->tableUidList = p;
        extractOrderedTableUidList(pOrderCheckInfo, pStatus);

        uid = pOrderCheckInfo->tableUidList[0];
        pStatus->pTableIter = taosHashGet(pStatus->pTableMap, &uid, sizeof(uid));
2512
      }
2513
    }
2514
  }
2515

2516 2517 2518
  return TSDB_CODE_SUCCESS;
}

2519
static bool moveToNextTable(SUidOrderCheckInfo* pOrderedCheckInfo, SReaderStatus* pStatus) {
2520 2521 2522 2523 2524 2525 2526 2527 2528 2529 2530 2531
  pOrderedCheckInfo->currentIndex += 1;
  if (pOrderedCheckInfo->currentIndex >= taosHashGetSize(pStatus->pTableMap)) {
    pStatus->pTableIter = NULL;
    return false;
  }

  uint64_t uid = pOrderedCheckInfo->tableUidList[pOrderedCheckInfo->currentIndex];
  pStatus->pTableIter = taosHashGet(pStatus->pTableMap, &uid, sizeof(uid));
  ASSERT(pStatus->pTableIter != NULL);
  return true;
}

2532
static int32_t doLoadLastBlockSequentially(STsdbReader* pReader) {
2533
  SReaderStatus*    pStatus = &pReader->status;
2534 2535
  SLastBlockReader* pLastBlockReader = pStatus->fileIter.pLastBlockReader;

2536 2537
  SUidOrderCheckInfo* pOrderedCheckInfo = &pStatus->uidCheckInfo;
  int32_t             code = initOrderCheckInfo(pOrderedCheckInfo, pStatus);
2538
  if (code != TSDB_CODE_SUCCESS || (taosHashGetSize(pStatus->pTableMap) == 0)) {
2539 2540
    return code;
  }
2541

2542
  while (1) {
2543 2544
    // load the last data block of current table
    STableBlockScanInfo* pScanInfo = pStatus->pTableIter;
H
Hongze Cheng 已提交
2545
    bool                 hasVal = initLastBlockReader(pLastBlockReader, pScanInfo, pReader);
2546
    if (!hasVal) {
2547 2548
      bool hasNexTable = moveToNextTable(pOrderedCheckInfo, pStatus);
      if (!hasNexTable) {
2549 2550 2551
        return TSDB_CODE_SUCCESS;
      }
      continue;
2552 2553
    }

2554 2555 2556 2557 2558 2559 2560 2561
    code = doBuildDataBlock(pReader);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

    if (pReader->pResBlock->info.rows > 0) {
      return TSDB_CODE_SUCCESS;
    }
2562

2563
    // current table is exhausted, let's try next table
2564 2565
    bool hasNexTable = moveToNextTable(pOrderedCheckInfo, pStatus);
    if (!hasNexTable) {
2566 2567
      return TSDB_CODE_SUCCESS;
    }
2568 2569 2570
  }
}

2571
static int32_t doBuildDataBlock(STsdbReader* pReader) {
H
Hongze Cheng 已提交
2572 2573
  int32_t   code = TSDB_CODE_SUCCESS;
  SDataBlk* pBlock = NULL;
2574 2575 2576

  SReaderStatus*       pStatus = &pReader->status;
  SDataBlockIter*      pBlockIter = &pStatus->blockIter;
2577 2578 2579
  STableBlockScanInfo* pScanInfo = NULL;
  SFileDataBlockInfo*  pBlockInfo = getCurrentBlockInfo(pBlockIter);
  SLastBlockReader*    pLastBlockReader = pReader->status.fileIter.pLastBlockReader;
2580

2581
  if (pBlockInfo != NULL) {
2582 2583 2584 2585 2586
    pScanInfo = taosHashGet(pReader->status.pTableMap, &pBlockInfo->uid, sizeof(pBlockInfo->uid));
  } else {
    pScanInfo = pReader->status.pTableIter;
  }

H
Haojun Liao 已提交
2587
  if (pScanInfo == NULL) {
H
Haojun Liao 已提交
2588
    tsdbError("failed to get table scan-info, %s", pReader->idStr);
H
Haojun Liao 已提交
2589 2590 2591 2592
    code = TSDB_CODE_INVALID_PARA;
    return code;
  }

2593
  if (pBlockInfo != NULL) {
2594
    pBlock = getCurrentBlock(pBlockIter);
2595 2596
  }

2597
  initLastBlockReader(pLastBlockReader, pScanInfo, pReader);
H
Haojun Liao 已提交
2598
  TSDBKEY keyInBuf = getCurrentKeyInBuf(pScanInfo, pReader);
2599

2600 2601 2602
  if (pBlockInfo == NULL) {  // build data block from last data file
    ASSERT(pBlockIter->numOfBlocks == 0);
    code = buildComposedDataBlock(pReader);
H
Haojun Liao 已提交
2603
  } else if (fileBlockShouldLoad(pReader, pBlockInfo, pBlock, pScanInfo, keyInBuf, pLastBlockReader)) {
2604
    code = doLoadFileBlockData(pReader, pBlockIter, &pStatus->fileBlockData, pScanInfo->uid);
2605 2606
    if (code != TSDB_CODE_SUCCESS) {
      return code;
2607 2608 2609
    }

    // build composed data block
2610
    code = buildComposedDataBlock(pReader);
H
Haojun Liao 已提交
2611
  } else if (bufferDataInFileBlockGap(pReader->order, keyInBuf, pBlock)) {
2612
    // data in memory that are earlier than current file block
H
Haojun Liao 已提交
2613
    // rows in buffer should be less than the file block in asc, greater than file block in desc
2614
    int64_t endKey = (ASCENDING_TRAVERSE(pReader->order)) ? pBlock->minKey.ts : pBlock->maxKey.ts;
2615
    code = buildDataBlockFromBuf(pReader, pScanInfo, endKey);
2616 2617 2618 2619
  } else {
    if (hasDataInLastBlock(pLastBlockReader) && !ASCENDING_TRAVERSE(pReader->order)) {
      // only return the rows in last block
      int64_t tsLast = getCurrentKeyInLastBlock(pLastBlockReader);
H
Hongze Cheng 已提交
2620
      ASSERT(tsLast >= pBlock->maxKey.ts);
2621 2622
      tBlockDataReset(&pReader->status.fileBlockData);

2623
      tsdbDebug("load data in last block firstly, due to desc scan data, %s", pReader->idStr);
2624
      code = buildComposedDataBlock(pReader);
H
Hongze Cheng 已提交
2625
    } else {  // whole block is required, return it directly
2626 2627 2628 2629 2630 2631 2632
      SDataBlockInfo* pInfo = &pReader->pResBlock->info;
      pInfo->rows = pBlock->nRow;
      pInfo->uid = pScanInfo->uid;
      pInfo->window = (STimeWindow){.skey = pBlock->minKey.ts, .ekey = pBlock->maxKey.ts};
      setComposedBlockFlag(pReader, false);
      setBlockAllDumped(&pStatus->fBlockDumpInfo, pBlock->maxKey.ts, pReader->order);
    }
2633 2634 2635 2636 2637
  }

  return code;
}

H
Haojun Liao 已提交
2638
static int32_t buildBlockFromBufferSequentially(STsdbReader* pReader) {
2639 2640
  SReaderStatus* pStatus = &pReader->status;

2641
  while (1) {
2642 2643 2644
    if (pStatus->pTableIter == NULL) {
      pStatus->pTableIter = taosHashIterate(pStatus->pTableMap, NULL);
      if (pStatus->pTableIter == NULL) {
H
Haojun Liao 已提交
2645
        return TSDB_CODE_SUCCESS;
2646 2647 2648 2649
      }
    }

    STableBlockScanInfo* pBlockScanInfo = pStatus->pTableIter;
2650
    initMemDataIterator(pBlockScanInfo, pReader);
2651

2652
    int64_t endKey = (ASCENDING_TRAVERSE(pReader->order)) ? INT64_MAX : INT64_MIN;
2653
    int32_t code = buildDataBlockFromBuf(pReader, pBlockScanInfo, endKey);
H
Haojun Liao 已提交
2654 2655 2656 2657
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

2658
    if (pReader->pResBlock->info.rows > 0) {
H
Haojun Liao 已提交
2659
      return TSDB_CODE_SUCCESS;
2660 2661 2662 2663 2664
    }

    // current table is exhausted, let's try the next table
    pStatus->pTableIter = taosHashIterate(pStatus->pTableMap, pStatus->pTableIter);
    if (pStatus->pTableIter == NULL) {
H
Haojun Liao 已提交
2665
      return TSDB_CODE_SUCCESS;
2666 2667 2668 2669
    }
  }
}

2670
// set the correct start position in case of the first/last file block, according to the query time window
2671
static void initBlockDumpInfo(STsdbReader* pReader, SDataBlockIter* pBlockIter) {
H
Hongze Cheng 已提交
2672
  SDataBlk* pBlock = getCurrentBlock(pBlockIter);
2673

2674 2675 2676
  SReaderStatus* pStatus = &pReader->status;

  SFileBlockDumpInfo* pDumpInfo = &pStatus->fBlockDumpInfo;
2677 2678 2679

  pDumpInfo->totalRows = pBlock->nRow;
  pDumpInfo->allDumped = false;
2680
  pDumpInfo->rowIndex = ASCENDING_TRAVERSE(pReader->order) ? 0 : pBlock->nRow - 1;
2681 2682
}

2683
static int32_t initForFirstBlockInFile(STsdbReader* pReader, SDataBlockIter* pBlockIter) {
2684 2685
  SBlockNumber num = {0};

2686
  int32_t code = moveToNextFile(pReader, &num);
2687 2688 2689 2690 2691
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

  // all data files are consumed, try data in buffer
2692
  if (num.numOfBlocks + num.numOfLastFiles == 0) {
2693 2694 2695 2696 2697
    pReader->status.loadFromFile = false;
    return code;
  }

  // initialize the block iterator for a new fileset
2698 2699
  if (num.numOfBlocks > 0) {
    code = initBlockIterator(pReader, pBlockIter, num.numOfBlocks);
H
Hongze Cheng 已提交
2700
  } else {  // no block data, only last block exists
2701
    tBlockDataReset(&pReader->status.fileBlockData);
2702
    resetDataBlockIterator(pBlockIter, pReader->order);
2703
  }
2704 2705

  // set the correct start position according to the query time window
2706
  initBlockDumpInfo(pReader, pBlockIter);
2707 2708 2709
  return code;
}

2710
static bool fileBlockPartiallyRead(SFileBlockDumpInfo* pDumpInfo, bool asc) {
2711 2712
  return (!pDumpInfo->allDumped) &&
         ((pDumpInfo->rowIndex > 0 && asc) || (pDumpInfo->rowIndex < (pDumpInfo->totalRows - 1) && (!asc)));
2713 2714
}

2715
static int32_t buildBlockFromFiles(STsdbReader* pReader) {
H
Haojun Liao 已提交
2716
  int32_t code = TSDB_CODE_SUCCESS;
2717 2718
  bool    asc = ASCENDING_TRAVERSE(pReader->order);

2719 2720
  SDataBlockIter* pBlockIter = &pReader->status.blockIter;

2721
  if (pBlockIter->numOfBlocks == 0) {
H
Hongze Cheng 已提交
2722
  _begin:
2723 2724 2725 2726 2727
    code = doLoadLastBlockSequentially(pReader);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

2728 2729 2730 2731
    if (pReader->pResBlock->info.rows > 0) {
      return TSDB_CODE_SUCCESS;
    }

2732
    // all data blocks are checked in this last block file, now let's try the next file
2733 2734 2735 2736 2737 2738 2739 2740
    if (pReader->status.pTableIter == NULL) {
      code = initForFirstBlockInFile(pReader, pBlockIter);

      // error happens or all the data files are completely checked
      if ((code != TSDB_CODE_SUCCESS) || (pReader->status.loadFromFile == false)) {
        return code;
      }

2741
      // this file does not have data files, let's start check the last block file if exists
2742 2743 2744 2745 2746 2747 2748 2749 2750 2751 2752 2753 2754 2755 2756
      if (pBlockIter->numOfBlocks == 0) {
        goto _begin;
      }
    }

    code = doBuildDataBlock(pReader);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

    if (pReader->pResBlock->info.rows > 0) {
      return TSDB_CODE_SUCCESS;
    }
  }

2757
  while (1) {
2758 2759
    SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;

2760
    if (fileBlockPartiallyRead(pDumpInfo, asc)) {  // file data block is partially loaded
2761
      code = buildComposedDataBlock(pReader);
2762 2763 2764 2765
    } else {
      // current block are exhausted, try the next file block
      if (pDumpInfo->allDumped) {
        // try next data block in current file
H
Haojun Liao 已提交
2766
        bool hasNext = blockIteratorNext(&pReader->status.blockIter, pReader->idStr);
2767 2768
        if (hasNext) {  // check for the next block in the block accessed order list
          initBlockDumpInfo(pReader, pBlockIter);
2769
        } else {
H
Haojun Liao 已提交
2770
          if (pReader->status.pCurrentFileset->nSttF > 0) {
2771 2772 2773 2774 2775 2776
            // data blocks in current file are exhausted, let's try the next file now
            tBlockDataReset(&pReader->status.fileBlockData);
            resetDataBlockIterator(pBlockIter, pReader->order);
            goto _begin;
          } else {
            code = initForFirstBlockInFile(pReader, pBlockIter);
2777

2778 2779 2780 2781
            // error happens or all the data files are completely checked
            if ((code != TSDB_CODE_SUCCESS) || (pReader->status.loadFromFile == false)) {
              return code;
            }
2782

2783 2784 2785 2786
            // this file does not have blocks, let's start check the last block file
            if (pBlockIter->numOfBlocks == 0) {
              goto _begin;
            }
2787
          }
2788
        }
H
Haojun Liao 已提交
2789
      }
2790 2791

      code = doBuildDataBlock(pReader);
2792 2793
    }

2794 2795 2796 2797 2798 2799 2800 2801
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

    if (pReader->pResBlock->info.rows > 0) {
      return TSDB_CODE_SUCCESS;
    }
  }
2802
}
H
refact  
Hongze Cheng 已提交
2803

2804 2805
static STsdb* getTsdbByRetentions(SVnode* pVnode, TSKEY winSKey, SRetention* retentions, const char* idStr,
                                  int8_t* pLevel) {
2806
  if (VND_IS_RSMA(pVnode)) {
2807
    int8_t  level = 0;
2808 2809
    int8_t  precision = pVnode->config.tsdbCfg.precision;
    int64_t now = taosGetTimestamp(precision);
H
Haojun Liao 已提交
2810 2811 2812
    int64_t offset = tsQueryRsmaTolerance * ((precision == TSDB_TIME_PRECISION_MILLI)   ? 1L
                                             : (precision == TSDB_TIME_PRECISION_MICRO) ? 1000L
                                                                                        : 1000000L);
2813

2814
    for (int8_t i = 0; i < TSDB_RETENTION_MAX; ++i) {
2815 2816 2817 2818 2819 2820 2821
      SRetention* pRetention = retentions + level;
      if (pRetention->keep <= 0) {
        if (level > 0) {
          --level;
        }
        break;
      }
2822
      if ((now - pRetention->keep) <= (winSKey + offset)) {
2823 2824 2825 2826 2827
        break;
      }
      ++level;
    }

2828
    const char* str = (idStr != NULL) ? idStr : "";
2829 2830

    if (level == TSDB_RETENTION_L0) {
2831
      *pLevel = TSDB_RETENTION_L0;
C
Cary Xu 已提交
2832
      tsdbDebug("vgId:%d, rsma level %d is selected to query %s", TD_VID(pVnode), TSDB_RETENTION_L0, str);
2833 2834
      return VND_RSMA0(pVnode);
    } else if (level == TSDB_RETENTION_L1) {
2835
      *pLevel = TSDB_RETENTION_L1;
C
Cary Xu 已提交
2836
      tsdbDebug("vgId:%d, rsma level %d is selected to query %s", TD_VID(pVnode), TSDB_RETENTION_L1, str);
2837 2838
      return VND_RSMA1(pVnode);
    } else {
2839
      *pLevel = TSDB_RETENTION_L2;
C
Cary Xu 已提交
2840
      tsdbDebug("vgId:%d, rsma level %d is selected to query %s", TD_VID(pVnode), TSDB_RETENTION_L2, str);
2841 2842 2843 2844 2845 2846 2847
      return VND_RSMA2(pVnode);
    }
  }

  return VND_TSDB(pVnode);
}

H
Haojun Liao 已提交
2848
SVersionRange getQueryVerRange(SVnode* pVnode, SQueryTableDataCond* pCond, int8_t level) {
L
Liu Jicong 已提交
2849
  int64_t startVer = (pCond->startVersion == -1) ? 0 : pCond->startVersion;
H
Haojun Liao 已提交
2850 2851

  int64_t endVer = 0;
L
Liu Jicong 已提交
2852 2853
  if (pCond->endVersion ==
      -1) {  // user not specified end version, set current maximum version of vnode as the endVersion
H
Haojun Liao 已提交
2854 2855
    endVer = pVnode->state.applied;
  } else {
L
Liu Jicong 已提交
2856
    endVer = (pCond->endVersion > pVnode->state.applied) ? pVnode->state.applied : pCond->endVersion;
2857 2858
  }

H
Haojun Liao 已提交
2859
  return (SVersionRange){.minVer = startVer, .maxVer = endVer};
2860 2861
}

2862
bool hasBeenDropped(const SArray* pDelList, int32_t* index, TSDBKEY* pKey, int32_t order) {
2863 2864 2865 2866
  ASSERT(pKey != NULL);
  if (pDelList == NULL) {
    return false;
  }
L
Liu Jicong 已提交
2867 2868 2869
  size_t  num = taosArrayGetSize(pDelList);
  bool    asc = ASCENDING_TRAVERSE(order);
  int32_t step = asc ? 1 : -1;
2870

2871 2872 2873 2874 2875 2876
  if (asc) {
    if (*index >= num - 1) {
      TSDBKEY* last = taosArrayGetLast(pDelList);
      ASSERT(pKey->ts >= last->ts);

      if (pKey->ts > last->ts) {
2877
        return false;
2878 2879 2880
      } else if (pKey->ts == last->ts) {
        TSDBKEY* prev = taosArrayGet(pDelList, num - 2);
        return (prev->version >= pKey->version);
2881 2882
      }
    } else {
2883 2884 2885 2886 2887 2888 2889 2890 2891 2892 2893 2894 2895 2896 2897 2898 2899 2900 2901 2902 2903 2904 2905 2906 2907 2908 2909 2910 2911 2912
      TSDBKEY* pCurrent = taosArrayGet(pDelList, *index);
      TSDBKEY* pNext = taosArrayGet(pDelList, (*index) + 1);

      if (pKey->ts < pCurrent->ts) {
        return false;
      }

      if (pCurrent->ts <= pKey->ts && pNext->ts >= pKey->ts && pCurrent->version >= pKey->version) {
        return true;
      }

      while (pNext->ts <= pKey->ts && (*index) < num - 1) {
        (*index) += 1;

        if ((*index) < num - 1) {
          pCurrent = taosArrayGet(pDelList, *index);
          pNext = taosArrayGet(pDelList, (*index) + 1);

          // it is not a consecutive deletion range, ignore it
          if (pCurrent->version == 0 && pNext->version > 0) {
            continue;
          }

          if (pCurrent->ts <= pKey->ts && pNext->ts >= pKey->ts && pCurrent->version >= pKey->version) {
            return true;
          }
        }
      }

      return false;
2913 2914
    }
  } else {
2915 2916
    if (*index <= 0) {
      TSDBKEY* pFirst = taosArrayGet(pDelList, 0);
2917

2918 2919 2920 2921 2922 2923 2924
      if (pKey->ts < pFirst->ts) {
        return false;
      } else if (pKey->ts == pFirst->ts) {
        return pFirst->version >= pKey->version;
      } else {
        ASSERT(0);
      }
2925
    } else {
2926 2927 2928 2929 2930 2931 2932 2933 2934 2935 2936 2937 2938 2939 2940 2941 2942 2943 2944 2945 2946 2947 2948 2949 2950 2951 2952
      TSDBKEY* pCurrent = taosArrayGet(pDelList, *index);
      TSDBKEY* pPrev = taosArrayGet(pDelList, (*index) - 1);

      if (pKey->ts > pCurrent->ts) {
        return false;
      }

      if (pPrev->ts <= pKey->ts && pCurrent->ts >= pKey->ts && pPrev->version >= pKey->version) {
        return true;
      }

      while (pPrev->ts >= pKey->ts && (*index) > 1) {
        (*index) += step;

        if ((*index) >= 1) {
          pCurrent = taosArrayGet(pDelList, *index);
          pPrev = taosArrayGet(pDelList, (*index) - 1);

          // it is not a consecutive deletion range, ignore it
          if (pCurrent->version > 0 && pPrev->version == 0) {
            continue;
          }

          if (pPrev->ts <= pKey->ts && pCurrent->ts >= pKey->ts && pPrev->version >= pKey->version) {
            return true;
          }
        }
2953 2954 2955 2956 2957
      }

      return false;
    }
  }
2958 2959

  return false;
2960 2961
}

2962
TSDBROW* getValidMemRow(SIterInfo* pIter, const SArray* pDelList, STsdbReader* pReader) {
2963
  if (!pIter->hasVal) {
H
Haojun Liao 已提交
2964 2965
    return NULL;
  }
H
Hongze Cheng 已提交
2966

2967
  TSDBROW* pRow = tsdbTbDataIterGet(pIter->iter);
2968
  TSDBKEY  key = {.ts = pRow->pTSRow->ts, .version = pRow->version};
2969
  if (outOfTimeWindow(key.ts, &pReader->window)) {
2970
    pIter->hasVal = false;
H
Haojun Liao 已提交
2971 2972
    return NULL;
  }
H
Hongze Cheng 已提交
2973

2974
  // it is a valid data version
dengyihao's avatar
dengyihao 已提交
2975
  if ((key.version <= pReader->verRange.maxVer && key.version >= pReader->verRange.minVer) &&
2976
      (!hasBeenDropped(pDelList, &pIter->index, &key, pReader->order))) {
H
Haojun Liao 已提交
2977 2978
    return pRow;
  }
H
Hongze Cheng 已提交
2979

2980
  while (1) {
2981 2982
    pIter->hasVal = tsdbTbDataIterNext(pIter->iter);
    if (!pIter->hasVal) {
H
Haojun Liao 已提交
2983 2984
      return NULL;
    }
H
Hongze Cheng 已提交
2985

2986
    pRow = tsdbTbDataIterGet(pIter->iter);
H
Hongze Cheng 已提交
2987

H
Haojun Liao 已提交
2988
    key = TSDBROW_KEY(pRow);
2989
    if (outOfTimeWindow(key.ts, &pReader->window)) {
2990
      pIter->hasVal = false;
H
Haojun Liao 已提交
2991 2992
      return NULL;
    }
H
Hongze Cheng 已提交
2993

dengyihao's avatar
dengyihao 已提交
2994
    if (key.version <= pReader->verRange.maxVer && key.version >= pReader->verRange.minVer &&
2995
        (!hasBeenDropped(pDelList, &pIter->index, &key, pReader->order))) {
H
Haojun Liao 已提交
2996 2997 2998 2999
      return pRow;
    }
  }
}
H
Hongze Cheng 已提交
3000

3001 3002
int32_t doMergeRowsInBuf(SIterInfo* pIter, uint64_t uid, int64_t ts, SArray* pDelList, SRowMerger* pMerger,
                         STsdbReader* pReader) {
H
Haojun Liao 已提交
3003
  while (1) {
3004 3005
    pIter->hasVal = tsdbTbDataIterNext(pIter->iter);
    if (!pIter->hasVal) {
H
Haojun Liao 已提交
3006 3007
      break;
    }
H
Hongze Cheng 已提交
3008

3009
    // data exists but not valid
3010
    TSDBROW* pRow = getValidMemRow(pIter, pDelList, pReader);
3011 3012 3013 3014 3015
    if (pRow == NULL) {
      break;
    }

    // ts is not identical, quit
H
Haojun Liao 已提交
3016
    TSDBKEY k = TSDBROW_KEY(pRow);
3017
    if (k.ts != ts) {
H
Haojun Liao 已提交
3018 3019 3020
      break;
    }

H
Haojun Liao 已提交
3021
    STSchema* pTSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(pRow), pReader, uid);
H
Haojun Liao 已提交
3022 3023 3024 3025
    if (pTSchema == NULL) {
      return terrno;
    }

3026
    tRowMergerAdd(pMerger, pRow, pTSchema);
H
Haojun Liao 已提交
3027 3028 3029 3030 3031
  }

  return TSDB_CODE_SUCCESS;
}

3032
static int32_t doMergeRowsInFileBlockImpl(SBlockData* pBlockData, int32_t rowIndex, int64_t key, SRowMerger* pMerger,
3033
                                          SVersionRange* pVerRange, int32_t step) {
3034 3035
  while (pBlockData->aTSKEY[rowIndex] == key && rowIndex < pBlockData->nRow && rowIndex >= 0) {
    if (pBlockData->aVersion[rowIndex] > pVerRange->maxVer || pBlockData->aVersion[rowIndex] < pVerRange->minVer) {
3036
      rowIndex += step;
3037 3038 3039 3040 3041 3042 3043 3044 3045 3046 3047 3048 3049 3050 3051 3052
      continue;
    }

    TSDBROW fRow = tsdbRowFromBlockData(pBlockData, rowIndex);
    tRowMerge(pMerger, &fRow);
    rowIndex += step;
  }

  return rowIndex;
}

typedef enum {
  CHECK_FILEBLOCK_CONT = 0x1,
  CHECK_FILEBLOCK_QUIT = 0x2,
} CHECK_FILEBLOCK_STATE;

H
Hongze Cheng 已提交
3053
static int32_t checkForNeighborFileBlock(STsdbReader* pReader, STableBlockScanInfo* pScanInfo, SDataBlk* pBlock,
3054 3055
                                         SFileDataBlockInfo* pFBlock, SRowMerger* pMerger, int64_t key,
                                         CHECK_FILEBLOCK_STATE* state) {
3056
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
3057
  SBlockData*         pBlockData = &pReader->status.fileBlockData;
3058

3059
  *state = CHECK_FILEBLOCK_QUIT;
3060
  int32_t step = ASCENDING_TRAVERSE(pReader->order) ? 1 : -1;
3061

H
Hongze Cheng 已提交
3062 3063
  int32_t   nextIndex = -1;
  SDataBlk* pNeighborBlock = getNeighborBlockOfSameTable(pFBlock, pScanInfo, &nextIndex, pReader->order);
3064
  if (pNeighborBlock == NULL) {  // do nothing
3065 3066 3067 3068
    return 0;
  }

  bool overlap = overlapWithNeighborBlock(pBlock, pNeighborBlock, pReader->order);
3069 3070
  taosMemoryFree(pNeighborBlock);

3071
  if (overlap) {  // load next block
3072
    SReaderStatus*  pStatus = &pReader->status;
3073 3074
    SDataBlockIter* pBlockIter = &pStatus->blockIter;

3075
    // 1. find the next neighbor block in the scan block list
3076
    SFileDataBlockInfo fb = {.uid = pFBlock->uid, .tbBlockIdx = nextIndex};
3077
    int32_t            neighborIndex = findFileBlockInfoIndex(pBlockIter, &fb);
3078

3079
    // 2. remove it from the scan block list
3080
    setFileBlockActiveInBlockIter(pBlockIter, neighborIndex, step);
3081

3082
    // 3. load the neighbor block, and set it to be the currently accessed file data block
3083
    int32_t code = doLoadFileBlockData(pReader, pBlockIter, &pStatus->fileBlockData, pFBlock->uid);
3084 3085 3086 3087
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

3088
    // 4. check the data values
3089 3090 3091 3092
    initBlockDumpInfo(pReader, pBlockIter);

    pDumpInfo->rowIndex =
        doMergeRowsInFileBlockImpl(pBlockData, pDumpInfo->rowIndex, key, pMerger, &pReader->verRange, step);
H
Haojun Liao 已提交
3093
    if (pDumpInfo->rowIndex >= pDumpInfo->totalRows) {
3094 3095 3096 3097 3098 3099 3100
      *state = CHECK_FILEBLOCK_CONT;
    }
  }

  return TSDB_CODE_SUCCESS;
}

3101 3102
int32_t doMergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pScanInfo, STsdbReader* pReader,
                                SRowMerger* pMerger) {
3103 3104
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;

3105
  bool    asc = ASCENDING_TRAVERSE(pReader->order);
3106
  int64_t key = pBlockData->aTSKEY[pDumpInfo->rowIndex];
3107
  int32_t step = asc ? 1 : -1;
3108

3109
  pDumpInfo->rowIndex += step;
3110
  if ((pDumpInfo->rowIndex <= pBlockData->nRow - 1 && asc) || (pDumpInfo->rowIndex >= 0 && !asc)) {
3111 3112 3113
    pDumpInfo->rowIndex =
        doMergeRowsInFileBlockImpl(pBlockData, pDumpInfo->rowIndex, key, pMerger, &pReader->verRange, step);
  }
3114

3115 3116 3117 3118
  // all rows are consumed, let's try next file block
  if ((pDumpInfo->rowIndex >= pBlockData->nRow && asc) || (pDumpInfo->rowIndex < 0 && !asc)) {
    while (1) {
      CHECK_FILEBLOCK_STATE st;
3119

3120
      SFileDataBlockInfo* pFileBlockInfo = getCurrentBlockInfo(&pReader->status.blockIter);
H
Hongze Cheng 已提交
3121
      SDataBlk*           pCurrentBlock = getCurrentBlock(&pReader->status.blockIter);
3122 3123 3124
      checkForNeighborFileBlock(pReader, pScanInfo, pCurrentBlock, pFileBlockInfo, pMerger, key, &st);
      if (st == CHECK_FILEBLOCK_QUIT) {
        break;
3125
      }
3126
    }
H
Haojun Liao 已提交
3127
  }
3128

H
Haojun Liao 已提交
3129 3130 3131
  return TSDB_CODE_SUCCESS;
}

H
Hongze Cheng 已提交
3132 3133
int32_t doMergeRowsInLastBlock(SLastBlockReader* pLastBlockReader, STableBlockScanInfo* pScanInfo, int64_t ts,
                               SRowMerger* pMerger) {
H
Haojun Liao 已提交
3134
  pScanInfo->lastKey = ts;
3135
  while (nextRowFromLastBlocks(pLastBlockReader, pScanInfo)) {
3136 3137
    int64_t next1 = getCurrentKeyInLastBlock(pLastBlockReader);
    if (next1 == ts) {
3138
      TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
3139 3140 3141 3142 3143 3144 3145 3146 3147
      tRowMerge(pMerger, &fRow1);
    } else {
      break;
    }
  }

  return TSDB_CODE_SUCCESS;
}

3148 3149
int32_t doMergeMemTableMultiRows(TSDBROW* pRow, uint64_t uid, SIterInfo* pIter, SArray* pDelList, STSRow** pTSRow,
                                 STsdbReader* pReader, bool* freeTSRow) {
H
Haojun Liao 已提交
3150
  TSDBROW* pNextRow = NULL;
3151
  TSDBROW  current = *pRow;
3152

3153 3154
  {  // if the timestamp of the next valid row has a different ts, return current row directly
    pIter->hasVal = tsdbTbDataIterNext(pIter->iter);
3155

3156 3157 3158
    if (!pIter->hasVal) {
      *pTSRow = current.pTSRow;
      *freeTSRow = false;
3159
      return TSDB_CODE_SUCCESS;
3160
    } else {  // has next point in mem/imem
3161
      pNextRow = getValidMemRow(pIter, pDelList, pReader);
3162 3163 3164
      if (pNextRow == NULL) {
        *pTSRow = current.pTSRow;
        *freeTSRow = false;
3165
        return TSDB_CODE_SUCCESS;
3166 3167
      }

H
Haojun Liao 已提交
3168
      if (current.pTSRow->ts != pNextRow->pTSRow->ts) {
3169 3170
        *pTSRow = current.pTSRow;
        *freeTSRow = false;
3171
        return TSDB_CODE_SUCCESS;
3172
      }
3173
    }
3174 3175
  }

3176 3177
  SRowMerger merge = {0};

3178
  // get the correct schema for data in memory
H
Haojun Liao 已提交
3179
  terrno = 0;
H
Haojun Liao 已提交
3180
  STSchema* pTSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(&current), pReader, uid);
H
Haojun Liao 已提交
3181 3182 3183
  if (pTSchema == NULL) {
    return terrno;
  }
H
Haojun Liao 已提交
3184

3185 3186
  if (pReader->pSchema == NULL) {
    pReader->pSchema = pTSchema;
3187
  }
H
Haojun Liao 已提交
3188

H
Haojun Liao 已提交
3189 3190 3191 3192
  int32_t code = tRowMergerInit2(&merge, pReader->pSchema, &current, pTSchema);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }
H
Haojun Liao 已提交
3193 3194

  STSchema* pTSchema1 = doGetSchemaForTSRow(TSDBROW_SVERSION(pNextRow), pReader, uid);
H
Haojun Liao 已提交
3195
  if (pTSchema1 == NULL) {
H
Haojun Liao 已提交
3196 3197 3198
    return terrno;
  }

H
Haojun Liao 已提交
3199 3200
  tRowMergerAdd(&merge, pNextRow, pTSchema1);

H
Haojun Liao 已提交
3201 3202 3203 3204 3205
  code = doMergeRowsInBuf(pIter, uid, current.pTSRow->ts, pDelList, &merge, pReader);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

H
Haojun Liao 已提交
3206
  code = tRowMergerGetRow(&merge, pTSRow);
3207 3208 3209
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }
M
Minglei Jin 已提交
3210

3211
  tRowMergerClear(&merge);
3212
  *freeTSRow = true;
3213
  return TSDB_CODE_SUCCESS;
3214 3215
}

3216
int32_t doMergeMemIMemRows(TSDBROW* pRow, TSDBROW* piRow, STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader,
H
Hongze Cheng 已提交
3217
                           STSRow** pTSRow) {
H
Haojun Liao 已提交
3218 3219
  SRowMerger merge = {0};

3220 3221 3222
  TSDBKEY k = TSDBROW_KEY(pRow);
  TSDBKEY ik = TSDBROW_KEY(piRow);

3223
  if (ASCENDING_TRAVERSE(pReader->order)) {  // ascending order imem --> mem
H
Haojun Liao 已提交
3224
    STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(pRow), pReader, pBlockScanInfo->uid);
3225

H
Haojun Liao 已提交
3226 3227 3228 3229 3230 3231 3232 3233 3234 3235
    int32_t code = tRowMergerInit(&merge, piRow, pSchema);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

    code = doMergeRowsInBuf(&pBlockScanInfo->iiter, pBlockScanInfo->uid, ik.ts, pBlockScanInfo->delSkyline, &merge,
                            pReader);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
3236

3237
    tRowMerge(&merge, pRow);
H
Haojun Liao 已提交
3238 3239 3240 3241 3242 3243
    code =
        doMergeRowsInBuf(&pBlockScanInfo->iter, pBlockScanInfo->uid, k.ts, pBlockScanInfo->delSkyline, &merge, pReader);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

3244
  } else {
H
Haojun Liao 已提交
3245
    STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(pRow), pReader, pBlockScanInfo->uid);
3246

H
Haojun Liao 已提交
3247
    int32_t code = tRowMergerInit(&merge, pRow, pSchema);
3248
    if (code != TSDB_CODE_SUCCESS || merge.pTSchema == NULL) {
H
Haojun Liao 已提交
3249 3250 3251 3252 3253 3254 3255 3256
      return code;
    }

    code =
        doMergeRowsInBuf(&pBlockScanInfo->iter, pBlockScanInfo->uid, k.ts, pBlockScanInfo->delSkyline, &merge, pReader);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
3257 3258

    tRowMerge(&merge, piRow);
H
Haojun Liao 已提交
3259 3260 3261 3262 3263
    code = doMergeRowsInBuf(&pBlockScanInfo->iiter, pBlockScanInfo->uid, ik.ts, pBlockScanInfo->delSkyline, &merge,
                            pReader);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
3264
  }
3265

3266 3267
  int32_t code = tRowMergerGetRow(&merge, pTSRow);
  return code;
3268 3269
}

3270 3271
int32_t tsdbGetNextRowInMem(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader, STSRow** pTSRow, int64_t endKey,
                            bool* freeTSRow) {
3272 3273
  TSDBROW* pRow = getValidMemRow(&pBlockScanInfo->iter, pBlockScanInfo->delSkyline, pReader);
  TSDBROW* piRow = getValidMemRow(&pBlockScanInfo->iiter, pBlockScanInfo->delSkyline, pReader);
dengyihao's avatar
dengyihao 已提交
3274
  SArray*  pDelList = pBlockScanInfo->delSkyline;
3275
  uint64_t uid = pBlockScanInfo->uid;
H
Haojun Liao 已提交
3276

3277 3278
  // todo refactor
  bool asc = ASCENDING_TRAVERSE(pReader->order);
3279
  if (pBlockScanInfo->iter.hasVal) {
3280 3281 3282 3283 3284 3285
    TSDBKEY k = TSDBROW_KEY(pRow);
    if ((k.ts >= endKey && asc) || (k.ts <= endKey && !asc)) {
      pRow = NULL;
    }
  }

3286
  if (pBlockScanInfo->iiter.hasVal) {
3287 3288 3289 3290 3291 3292
    TSDBKEY k = TSDBROW_KEY(piRow);
    if ((k.ts >= endKey && asc) || (k.ts <= endKey && !asc)) {
      piRow = NULL;
    }
  }

3293
  if (pBlockScanInfo->iter.hasVal && pBlockScanInfo->iiter.hasVal && pRow != NULL && piRow != NULL) {
3294
    TSDBKEY k = TSDBROW_KEY(pRow);
3295
    TSDBKEY ik = TSDBROW_KEY(piRow);
H
Haojun Liao 已提交
3296

3297
    int32_t code = TSDB_CODE_SUCCESS;
3298 3299
    if (ik.ts != k.ts) {
      if (((ik.ts < k.ts) && asc) || ((ik.ts > k.ts) && (!asc))) {  // ik.ts < k.ts
3300
        code = doMergeMemTableMultiRows(piRow, uid, &pBlockScanInfo->iiter, pDelList, pTSRow, pReader, freeTSRow);
3301
      } else if (((k.ts < ik.ts) && asc) || ((k.ts > ik.ts) && (!asc))) {
3302
        code = doMergeMemTableMultiRows(pRow, uid, &pBlockScanInfo->iter, pDelList, pTSRow, pReader, freeTSRow);
3303
      }
3304
    } else {  // ik.ts == k.ts
3305
      *freeTSRow = true;
3306 3307 3308 3309
      code = doMergeMemIMemRows(pRow, piRow, pBlockScanInfo, pReader, pTSRow);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
H
Haojun Liao 已提交
3310
    }
3311

3312
    return code;
H
Haojun Liao 已提交
3313 3314
  }

3315
  if (pBlockScanInfo->iter.hasVal && pRow != NULL) {
H
Hongze Cheng 已提交
3316 3317
    return doMergeMemTableMultiRows(pRow, pBlockScanInfo->uid, &pBlockScanInfo->iter, pDelList, pTSRow, pReader,
                                    freeTSRow);
H
Haojun Liao 已提交
3318 3319
  }

3320
  if (pBlockScanInfo->iiter.hasVal && piRow != NULL) {
3321
    return doMergeMemTableMultiRows(piRow, uid, &pBlockScanInfo->iiter, pDelList, pTSRow, pReader, freeTSRow);
H
Haojun Liao 已提交
3322 3323 3324 3325 3326
  }

  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
3327
int32_t doAppendRowFromTSRow(SSDataBlock* pBlock, STsdbReader* pReader, STSRow* pTSRow, uint64_t uid) {
3328 3329 3330
  int32_t numOfRows = pBlock->info.rows;
  int32_t numOfCols = (int32_t)taosArrayGetSize(pBlock->pDataBlock);

3331
  SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo;
3332
  STSchema*           pSchema = doGetSchemaForTSRow(pTSRow->sver, pReader, uid);
3333

3334
  SColVal colVal = {0};
3335
  int32_t i = 0, j = 0;
H
Haojun Liao 已提交
3336

3337 3338 3339 3340 3341 3342 3343 3344 3345 3346 3347
  SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
  if (pColInfoData->info.colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
    colDataAppend(pColInfoData, numOfRows, (const char*)&pTSRow->ts, false);
    i += 1;
  }

  while (i < numOfCols && j < pSchema->numOfCols) {
    pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
    col_id_t colId = pColInfoData->info.colId;

    if (colId == pSchema->columns[j].colId) {
H
Haojun Liao 已提交
3348
      tTSRowGetVal(pTSRow, pSchema, j, &colVal);
3349 3350 3351 3352 3353 3354 3355 3356
      doCopyColVal(pColInfoData, numOfRows, i, &colVal, pSupInfo);
      i += 1;
      j += 1;
    } else if (colId < pSchema->columns[j].colId) {
      colDataAppendNULL(pColInfoData, numOfRows);
      i += 1;
    } else if (colId > pSchema->columns[j].colId) {
      j += 1;
3357
    }
3358 3359
  }

3360
  // set null value since current column does not exist in the "pSchema"
3361
  while (i < numOfCols) {
3362 3363 3364 3365 3366
    pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
    colDataAppendNULL(pColInfoData, numOfRows);
    i += 1;
  }

3367 3368 3369 3370
  pBlock->info.rows += 1;
  return TSDB_CODE_SUCCESS;
}

H
Hongze Cheng 已提交
3371 3372
int32_t doAppendRowFromFileBlock(SSDataBlock* pResBlock, STsdbReader* pReader, SBlockData* pBlockData,
                                 int32_t rowIndex) {
3373 3374 3375 3376 3377 3378 3379 3380
  int32_t i = 0, j = 0;
  int32_t outputRowIndex = pResBlock->info.rows;

  SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo;

  SColumnInfoData* pColData = taosArrayGet(pResBlock->pDataBlock, i);
  if (pColData->info.colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
    colDataAppendInt64(pColData, outputRowIndex, &pBlockData->aTSKEY[rowIndex]);
3381
    i += 1;
3382 3383 3384
  }

  SColVal cv = {0};
3385 3386
  int32_t numOfInputCols = pBlockData->aIdx->size;
  int32_t numOfOutputCols = pResBlock->pDataBlock->size;
3387

3388
  while (i < numOfOutputCols && j < numOfInputCols) {
3389
    SColumnInfoData* pCol = TARRAY_GET_ELEM(pResBlock->pDataBlock, i);
3390
    SColData*        pData = tBlockDataGetColDataByIdx(pBlockData, j);
3391

3392 3393 3394 3395 3396
    if (pData->cid < pCol->info.colId) {
      j += 1;
      continue;
    }

3397
    if (pData->cid == pCol->info.colId) {
3398 3399
      tColDataGetValue(pData, rowIndex, &cv);
      doCopyColVal(pCol, outputRowIndex, i, &cv, pSupInfo);
3400
      j += 1;
H
Haojun Liao 已提交
3401 3402
    } else if (pData->cid > pCol->info.colId) {
      // the specified column does not exist in file block, fill with null data
3403 3404 3405 3406 3407 3408 3409 3410
      colDataAppendNULL(pCol, outputRowIndex);
    }

    i += 1;
  }

  while (i < numOfOutputCols) {
    SColumnInfoData* pCol = taosArrayGet(pResBlock->pDataBlock, i);
3411
    colDataAppendNULL(pCol, outputRowIndex);
3412 3413 3414 3415 3416 3417 3418
    i += 1;
  }

  pResBlock->info.rows += 1;
  return TSDB_CODE_SUCCESS;
}

3419 3420
int32_t buildDataBlockFromBufImpl(STableBlockScanInfo* pBlockScanInfo, int64_t endKey, int32_t capacity,
                                  STsdbReader* pReader) {
H
Haojun Liao 已提交
3421 3422 3423 3424
  SSDataBlock* pBlock = pReader->pResBlock;

  do {
    STSRow* pTSRow = NULL;
3425
    bool    freeTSRow = false;
3426
    tsdbGetNextRowInMem(pBlockScanInfo, pReader, &pTSRow, endKey, &freeTSRow);
3427 3428
    if (pTSRow == NULL) {
      break;
H
Haojun Liao 已提交
3429 3430
    }

H
Haojun Liao 已提交
3431
    doAppendRowFromTSRow(pBlock, pReader, pTSRow, pBlockScanInfo->uid);
3432 3433 3434
    if (freeTSRow) {
      taosMemoryFree(pTSRow);
    }
H
Haojun Liao 已提交
3435 3436

    // no data in buffer, return immediately
3437
    if (!(pBlockScanInfo->iter.hasVal || pBlockScanInfo->iiter.hasVal)) {
H
Haojun Liao 已提交
3438 3439 3440
      break;
    }

3441
    if (pBlock->info.rows >= capacity) {
H
Haojun Liao 已提交
3442 3443 3444 3445
      break;
    }
  } while (1);

3446
  ASSERT(pBlock->info.rows <= capacity);
H
Haojun Liao 已提交
3447 3448
  return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
3449

3450
// todo refactor, use arraylist instead
H
Hongze Cheng 已提交
3451
int32_t tsdbSetTableId(STsdbReader* pReader, int64_t uid) {
3452 3453 3454
  ASSERT(pReader != NULL);
  taosHashClear(pReader->status.pTableMap);

3455
  STableBlockScanInfo info = {.lastKey = 0, .uid = uid};
3456
  taosHashPut(pReader->status.pTableMap, &info.uid, sizeof(uint64_t), &info, sizeof(info));
H
Hongze Cheng 已提交
3457 3458 3459
  return TDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
3460 3461 3462 3463 3464 3465
void* tsdbGetIdx(SMeta* pMeta) {
  if (pMeta == NULL) {
    return NULL;
  }
  return metaGetIdx(pMeta);
}
dengyihao's avatar
dengyihao 已提交
3466

dengyihao's avatar
dengyihao 已提交
3467 3468 3469 3470 3471 3472
void* tsdbGetIvtIdx(SMeta* pMeta) {
  if (pMeta == NULL) {
    return NULL;
  }
  return metaGetIvtIdx(pMeta);
}
L
Liu Jicong 已提交
3473

H
Hongze Cheng 已提交
3474
uint64_t getReaderMaxVersion(STsdbReader* pReader) { return pReader->verRange.maxVer; }
3475

3476 3477 3478 3479 3480 3481 3482 3483 3484 3485 3486 3487 3488 3489 3490
static int32_t doOpenReaderImpl(STsdbReader* pReader) {
  SDataBlockIter* pBlockIter = &pReader->status.blockIter;

  initFilesetIterator(&pReader->status.fileIter, pReader->pReadSnap->fs.aDFileSet, pReader);
  resetDataBlockIterator(&pReader->status.blockIter, pReader->order);

  // no data in files, let's try buffer in memory
  if (pReader->status.fileIter.numOfFiles == 0) {
    pReader->status.loadFromFile = false;
    return TSDB_CODE_SUCCESS;
  } else {
    return initForFirstBlockInFile(pReader, pBlockIter);
  }
}

H
refact  
Hongze Cheng 已提交
3491
// ====================================== EXPOSED APIs ======================================
3492 3493
int32_t tsdbReaderOpen(SVnode* pVnode, SQueryTableDataCond* pCond, SArray* pTableList, STsdbReader** ppReader,
                       const char* idstr) {
3494 3495 3496 3497 3498 3499
  STimeWindow window = pCond->twindows;
  if (pCond->type == TIMEWINDOW_RANGE_EXTERNAL) {
    pCond->twindows.skey += 1;
    pCond->twindows.ekey -= 1;
  }

3500 3501
  int32_t code = tsdbReaderCreate(pVnode, pCond, ppReader, 4096, idstr);
  if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
3502 3503
    goto _err;
  }
H
Hongze Cheng 已提交
3504

3505
  // check for query time window
H
Haojun Liao 已提交
3506
  STsdbReader* pReader = *ppReader;
3507
  if (isEmptyQueryTimeWindow(&pReader->window) && pCond->type == TIMEWINDOW_RANGE_CONTAINED) {
H
Haojun Liao 已提交
3508 3509 3510
    tsdbDebug("%p query window not overlaps with the data set, no result returned, %s", pReader, pReader->idStr);
    return TSDB_CODE_SUCCESS;
  }
H
Hongze Cheng 已提交
3511

3512 3513
  if (pCond->type == TIMEWINDOW_RANGE_EXTERNAL) {
    // update the SQueryTableDataCond to create inner reader
3514
    int32_t order = pCond->order;
3515
    if (order == TSDB_ORDER_ASC) {
3516
      pCond->twindows.ekey = window.skey;
3517 3518 3519
      pCond->twindows.skey = INT64_MIN;
      pCond->order = TSDB_ORDER_DESC;
    } else {
3520
      pCond->twindows.skey = window.ekey;
3521 3522 3523 3524
      pCond->twindows.ekey = INT64_MAX;
      pCond->order = TSDB_ORDER_ASC;
    }

3525
    // here we only need one more row, so the capacity is set to be ONE.
3526 3527 3528 3529 3530 3531
    code = tsdbReaderCreate(pVnode, pCond, &pReader->innerReader[0], 1, idstr);
    if (code != TSDB_CODE_SUCCESS) {
      goto _err;
    }

    if (order == TSDB_ORDER_ASC) {
3532
      pCond->twindows.skey = window.ekey;
3533
      pCond->twindows.ekey = INT64_MAX;
3534
    } else {
3535
      pCond->twindows.skey = INT64_MIN;
3536
      pCond->twindows.ekey = window.ekey;
3537
    }
3538 3539
    pCond->order = order;

3540 3541 3542 3543 3544 3545
    code = tsdbReaderCreate(pVnode, pCond, &pReader->innerReader[1], 1, idstr);
    if (code != TSDB_CODE_SUCCESS) {
      goto _err;
    }
  }

H
Haojun Liao 已提交
3546
  // NOTE: the endVersion in pCond is the data version not schema version, so pCond->endVersion is not correct here.
3547
  if (pCond->suid != 0) {
3548
    pReader->pSchema = metaGetTbTSchema(pReader->pTsdb->pVnode->pMeta, pReader->suid, -1, 1);
H
Haojun Liao 已提交
3549
    if (pReader->pSchema == NULL) {
H
Haojun Liao 已提交
3550
      tsdbError("failed to get table schema, suid:%" PRIu64 ", ver:-1, %s", pReader->suid, pReader->idStr);
H
Haojun Liao 已提交
3551
    }
3552 3553
  } else if (taosArrayGetSize(pTableList) > 0) {
    STableKeyInfo* pKey = taosArrayGet(pTableList, 0);
3554
    pReader->pSchema = metaGetTbTSchema(pReader->pTsdb->pVnode->pMeta, pKey->uid, -1, 1);
H
Haojun Liao 已提交
3555
    if (pReader->pSchema == NULL) {
H
Haojun Liao 已提交
3556
      tsdbError("failed to get table schema, uid:%" PRIu64 ", ver:-1, %s", pKey->uid, pReader->idStr);
H
Haojun Liao 已提交
3557
    }
3558 3559
  }

H
Hongze Cheng 已提交
3560
  STsdbReader* p = pReader->innerReader[0] != NULL ? pReader->innerReader[0] : pReader;
3561

3562
  int32_t numOfTables = taosArrayGetSize(pTableList);
3563
  pReader->status.pTableMap = createDataBlockScanInfo(p, pTableList->pData, numOfTables);
H
Haojun Liao 已提交
3564 3565 3566
  if (pReader->status.pTableMap == NULL) {
    tsdbReaderClose(pReader);
    *ppReader = NULL;
H
Haojun Liao 已提交
3567

H
Haojun Liao 已提交
3568 3569 3570
    code = TSDB_CODE_TDB_OUT_OF_MEMORY;
    goto _err;
  }
H
Hongze Cheng 已提交
3571

3572
  if (numOfTables > 0) {
H
Haojun Liao 已提交
3573
    code = tsdbTakeReadSnap(pReader->pTsdb, &pReader->pReadSnap, pReader->idStr);
3574 3575 3576
    if (code != TSDB_CODE_SUCCESS) {
      goto _err;
    }
H
Hongze Cheng 已提交
3577

3578
    if (pReader->type == TIMEWINDOW_RANGE_CONTAINED) {
H
Haojun Liao 已提交
3579 3580 3581
      code = doOpenReaderImpl(pReader);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
3582
      }
3583
    } else {
H
Haojun Liao 已提交
3584 3585
      STsdbReader* pPrevReader = pReader->innerReader[0];
      STsdbReader* pNextReader = pReader->innerReader[1];
3586

H
Haojun Liao 已提交
3587 3588 3589 3590 3591 3592 3593 3594 3595 3596 3597 3598
      // we need only one row
      pPrevReader->capacity = 1;
      pPrevReader->status.pTableMap = pReader->status.pTableMap;
      pPrevReader->pSchema = pReader->pSchema;
      pPrevReader->pMemSchema = pReader->pMemSchema;
      pPrevReader->pReadSnap = pReader->pReadSnap;

      pNextReader->capacity = 1;
      pNextReader->status.pTableMap = pReader->status.pTableMap;
      pNextReader->pSchema = pReader->pSchema;
      pNextReader->pMemSchema = pReader->pMemSchema;
      pNextReader->pReadSnap = pReader->pReadSnap;
3599

H
Haojun Liao 已提交
3600
      code = doOpenReaderImpl(pPrevReader);
3601
      if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
3602
        return code;
3603 3604
      }

H
Haojun Liao 已提交
3605 3606 3607 3608
      code = doOpenReaderImpl(pNextReader);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
3609

H
Haojun Liao 已提交
3610 3611 3612
      code = doOpenReaderImpl(pReader);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
3613
      }
3614 3615 3616
    }
  }

3617
  tsdbDebug("%p total numOfTable:%d in this query %s", pReader, numOfTables, pReader->idStr);
H
Hongze Cheng 已提交
3618
  return code;
H
Hongze Cheng 已提交
3619 3620

_err:
H
Haojun Liao 已提交
3621
  tsdbError("failed to create data reader, code:%s %s", tstrerror(code), idstr);
H
Hongze Cheng 已提交
3622
  return code;
H
refact  
Hongze Cheng 已提交
3623 3624 3625
}

void tsdbReaderClose(STsdbReader* pReader) {
3626 3627
  if (pReader == NULL) {
    return;
3628
  }
H
refact  
Hongze Cheng 已提交
3629

3630 3631
  {
    if (pReader->innerReader[0] != NULL) {
3632
      STsdbReader* p = pReader->innerReader[0];
3633

3634 3635 3636 3637 3638 3639 3640 3641 3642 3643 3644
      p->status.pTableMap = NULL;
      p->pReadSnap = NULL;
      p->pSchema = NULL;
      p->pMemSchema = NULL;

      p = pReader->innerReader[1];

      p->status.pTableMap = NULL;
      p->pReadSnap = NULL;
      p->pSchema = NULL;
      p->pMemSchema = NULL;
3645 3646 3647 3648 3649 3650

      tsdbReaderClose(pReader->innerReader[0]);
      tsdbReaderClose(pReader->innerReader[1]);
    }
  }

3651
  SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo;
H
Hongze Cheng 已提交
3652

3653 3654 3655 3656
  taosMemoryFreeClear(pSupInfo->plist);
  taosMemoryFree(pSupInfo->colIds);

  taosArrayDestroy(pSupInfo->pColAgg);
L
Liu Jicong 已提交
3657
  for (int32_t i = 0; i < blockDataGetNumOfCols(pReader->pResBlock); ++i) {
3658 3659 3660 3661
    if (pSupInfo->buildBuf[i] != NULL) {
      taosMemoryFreeClear(pSupInfo->buildBuf[i]);
    }
  }
3662

3663
  taosMemoryFree(pSupInfo->buildBuf);
H
Hongze Cheng 已提交
3664
  tBlockDataDestroy(&pReader->status.fileBlockData, true);
3665 3666

  cleanupDataBlockIterator(&pReader->status.blockIter);
3667 3668

  size_t numOfTables = taosHashGetSize(pReader->status.pTableMap);
3669
  destroyBlockScanInfo(pReader->status.pTableMap);
3670
  blockDataDestroy(pReader->pResBlock);
3671

H
Haojun Liao 已提交
3672 3673 3674
  if (pReader->pFileReader != NULL) {
    tsdbDataFReaderClose(&pReader->pFileReader);
  }
H
refact  
Hongze Cheng 已提交
3675

H
Haojun Liao 已提交
3676
  tsdbUntakeReadSnap(pReader->pTsdb, pReader->pReadSnap, pReader->idStr);
3677

3678
  taosMemoryFree(pReader->status.uidCheckInfo.tableUidList);
H
Haojun Liao 已提交
3679
  SIOCostSummary* pCost = &pReader->cost;
3680

H
Haojun Liao 已提交
3681 3682
  SFilesetIter* pFilesetIter = &pReader->status.fileIter;
  if (pFilesetIter->pLastBlockReader != NULL) {
H
Haojun Liao 已提交
3683 3684
    SLastBlockReader* pLReader = pFilesetIter->pLastBlockReader;
    tMergeTreeClose(&pLReader->mergeTree);
H
Haojun Liao 已提交
3685

H
Haojun Liao 已提交
3686 3687 3688 3689 3690
    getLastBlockLoadInfo(pLReader->pInfo, &pCost->lastBlockLoad, &pCost->lastBlockLoadTime);

    pLReader->pInfo = destroyLastBlockLoadInfo(pLReader->pInfo);
    taosMemoryFree(pLReader);
  }
H
refact  
Hongze Cheng 已提交
3691

H
Haojun Liao 已提交
3692 3693
  tsdbDebug(
      "%p :io-cost summary: head-file:%" PRIu64 ", head-file time:%.2f ms, SMA:%" PRId64
H
Hongze Cheng 已提交
3694 3695
      " SMA-time:%.2f ms, fileBlocks:%" PRId64
      ", fileBlocks-load-time:%.2f ms, "
H
Haojun Liao 已提交
3696 3697 3698 3699 3700
      "build in-memory-block-time:%.2f ms, lastBlocks:%" PRId64 ", lastBlocks-time:%.2f ms, composed-blocks:%" PRId64
      ", composed-blocks-time:%.2fms, STableBlockScanInfo size:%.2f Kb %s",
      pReader, pCost->headFileLoad, pCost->headFileLoadTime, pCost->smaDataLoad, pCost->smaLoadTime, pCost->numOfBlocks,
      pCost->blockLoadTime, pCost->buildmemBlock, pCost->lastBlockLoad, pCost->lastBlockLoadTime, pCost->composedBlocks,
      pCost->buildComposedBlockTime, numOfTables * sizeof(STableBlockScanInfo) / 1000.0, pReader->idStr);
H
refact  
Hongze Cheng 已提交
3701

3702 3703
  taosMemoryFree(pReader->idStr);
  taosMemoryFree(pReader->pSchema);
3704 3705 3706
  if (pReader->pMemSchema != pReader->pSchema) {
    taosMemoryFree(pReader->pMemSchema);
  }
3707
  taosMemoryFreeClear(pReader);
H
refact  
Hongze Cheng 已提交
3708 3709
}

3710
static bool doTsdbNextDataBlock(STsdbReader* pReader) {
H
Haojun Liao 已提交
3711
  // cleanup the data that belongs to the previous data block
3712 3713
  SSDataBlock* pBlock = pReader->pResBlock;
  blockDataCleanup(pBlock);
H
Hongze Cheng 已提交
3714

3715
  SReaderStatus* pStatus = &pReader->status;
H
Haojun Liao 已提交
3716

3717 3718 3719 3720 3721
  if (pStatus->loadFromFile) {
    int32_t code = buildBlockFromFiles(pReader);
    if (code != TSDB_CODE_SUCCESS) {
      return false;
    }
3722

3723 3724 3725
    if (pBlock->info.rows > 0) {
      return true;
    } else {
H
Haojun Liao 已提交
3726
      buildBlockFromBufferSequentially(pReader);
3727
      return pBlock->info.rows > 0;
H
Haojun Liao 已提交
3728
    }
3729 3730 3731
  } else {  // no data in files, let's try the buffer
    buildBlockFromBufferSequentially(pReader);
    return pBlock->info.rows > 0;
H
Haojun Liao 已提交
3732
  }
3733

3734
  return false;
H
refact  
Hongze Cheng 已提交
3735 3736
}

3737 3738 3739 3740 3741
bool tsdbNextDataBlock(STsdbReader* pReader) {
  if (isEmptyQueryTimeWindow(&pReader->window)) {
    return false;
  }

3742
  if (pReader->innerReader[0] != NULL && pReader->step == 0) {
3743
    bool ret = doTsdbNextDataBlock(pReader->innerReader[0]);
3744
    resetDataBlockScanInfo(pReader->innerReader[0]->status.pTableMap, pReader->innerReader[0]->window.ekey);
3745 3746
    pReader->step = EXTERNAL_ROWS_PREV;

3747 3748 3749
    if (ret) {
      return ret;
    }
3750
  }
3751

3752 3753
  if (pReader->step == EXTERNAL_ROWS_PREV) {
    pReader->step = EXTERNAL_ROWS_MAIN;
3754 3755 3756 3757 3758 3759 3760
  }

  bool ret = doTsdbNextDataBlock(pReader);
  if (ret) {
    return ret;
  }

3761 3762
  if (pReader->innerReader[1] != NULL && pReader->step == EXTERNAL_ROWS_MAIN) {
    resetDataBlockScanInfo(pReader->innerReader[1]->status.pTableMap, pReader->window.ekey);
3763
    bool ret1 = doTsdbNextDataBlock(pReader->innerReader[1]);
3764
    pReader->step = EXTERNAL_ROWS_NEXT;
3765 3766 3767 3768 3769 3770 3771 3772
    if (ret1) {
      return ret1;
    }
  }

  return false;
}

3773
bool tsdbTableNextDataBlock(STsdbReader* pReader, uint64_t uid) {
H
Haojun Liao 已提交
3774
  STableBlockScanInfo* pBlockScanInfo = taosHashGet(pReader->status.pTableMap, &uid, sizeof(uid));
3775 3776 3777 3778
  if (pBlockScanInfo == NULL) { // no data block for the table of given uid
    return false;
  }

H
Haojun Liao 已提交
3779
  return true;
3780 3781
}

3782
static void setBlockInfo(STsdbReader* pReader, SDataBlockInfo* pDataBlockInfo) {
3783 3784 3785 3786
  ASSERT(pDataBlockInfo != NULL && pReader != NULL);
  pDataBlockInfo->rows = pReader->pResBlock->info.rows;
  pDataBlockInfo->uid = pReader->pResBlock->info.uid;
  pDataBlockInfo->window = pReader->pResBlock->info.window;
H
Hongze Cheng 已提交
3787 3788
}

3789 3790
void tsdbRetrieveDataBlockInfo(STsdbReader* pReader, SDataBlockInfo* pDataBlockInfo) {
  if (pReader->type == TIMEWINDOW_RANGE_EXTERNAL) {
3791
    if (pReader->step == EXTERNAL_ROWS_MAIN) {
3792
      setBlockInfo(pReader, pDataBlockInfo);
3793
    } else if (pReader->step == EXTERNAL_ROWS_PREV) {
3794 3795 3796 3797 3798 3799 3800 3801 3802
      setBlockInfo(pReader->innerReader[0], pDataBlockInfo);
    } else {
      setBlockInfo(pReader->innerReader[1], pDataBlockInfo);
    }
  } else {
    setBlockInfo(pReader, pDataBlockInfo);
  }
}

3803
int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SColumnDataAgg*** pBlockStatis, bool* allHave) {
H
Hongze Cheng 已提交
3804
  int32_t code = 0;
3805
  *allHave = false;
H
Hongze Cheng 已提交
3806

3807
  if (pReader->type == TIMEWINDOW_RANGE_EXTERNAL) {
3808 3809 3810 3811
    *pBlockStatis = NULL;
    return TSDB_CODE_SUCCESS;
  }

3812
  // there is no statistics data for composed block
3813 3814 3815 3816
  if (pReader->status.composedDataBlock) {
    *pBlockStatis = NULL;
    return TSDB_CODE_SUCCESS;
  }
H
Hongze Cheng 已提交
3817

3818
  SFileDataBlockInfo* pFBlock = getCurrentBlockInfo(&pReader->status.blockIter);
H
Hongze Cheng 已提交
3819

H
Hongze Cheng 已提交
3820 3821
  SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter);
  int64_t   stime = taosGetTimestampUs();
H
Hongze Cheng 已提交
3822

3823 3824
  SBlockLoadSuppInfo* pSup = &pReader->suppInfo;

H
Hongze Cheng 已提交
3825
  if (tDataBlkHasSma(pBlock)) {
H
Hongze Cheng 已提交
3826
    code = tsdbReadBlockSma(pReader->pFileReader, pBlock, pSup->pColAgg);
3827
    if (code != TSDB_CODE_SUCCESS) {
3828 3829
      tsdbDebug("vgId:%d, failed to load block SMA for uid %" PRIu64 ", code:%s, %s", 0, pFBlock->uid, tstrerror(code),
                pReader->idStr);
3830 3831
      return code;
    }
3832 3833 3834
  } else {
    *pBlockStatis = NULL;
    return TSDB_CODE_SUCCESS;
3835
  }
H
Hongze Cheng 已提交
3836

3837
  *allHave = true;
H
Hongze Cheng 已提交
3838

3839 3840
  // always load the first primary timestamp column data
  SColumnDataAgg* pTsAgg = &pSup->tsColAgg;
3841

3842 3843
  pTsAgg->numOfNull = 0;
  pTsAgg->colId = PRIMARYKEY_TIMESTAMP_COL_ID;
3844 3845 3846 3847 3848 3849 3850 3851 3852 3853 3854 3855 3856 3857 3858 3859
  pTsAgg->min = pReader->pResBlock->info.window.skey;
  pTsAgg->max = pReader->pResBlock->info.window.ekey;
  pSup->plist[0] = pTsAgg;

  // update the number of NULL data rows
  size_t numOfCols = blockDataGetNumOfCols(pReader->pResBlock);

  int32_t i = 0, j = 0;
  while (j < numOfCols && i < taosArrayGetSize(pSup->pColAgg)) {
    SColumnDataAgg* pAgg = taosArrayGet(pSup->pColAgg, i);
    if (pAgg->colId == pSup->colIds[j]) {
      if (IS_BSMA_ON(&(pReader->pSchema->columns[i]))) {
        pSup->plist[j] = pAgg;
      } else {
        *allHave = false;
      }
3860 3861
      i += 1;
      j += 1;
3862 3863 3864 3865 3866 3867 3868
    } else if (pAgg->colId < pSup->colIds[j]) {
      i += 1;
    } else if (pSup->colIds[j] < pAgg->colId) {
      j += 1;
    }
  }

3869
  double elapsed = (taosGetTimestampUs() - stime) / 1000.0;
3870
  pReader->cost.smaLoadTime += elapsed;
3871
  pReader->cost.smaDataLoad += 1;
3872 3873 3874

  *pBlockStatis = pSup->plist;

3875
  tsdbDebug("vgId:%d, succeed to load block SMA for uid %" PRIu64 ", elapsed time:%.2f ms, %s", 0, pFBlock->uid,
3876 3877
            elapsed, pReader->idStr);

H
Hongze Cheng 已提交
3878
  return code;
H
Hongze Cheng 已提交
3879 3880
}

3881
static SArray* doRetrieveDataBlock(STsdbReader* pReader) {
H
Haojun Liao 已提交
3882 3883 3884
  SReaderStatus* pStatus = &pReader->status;

  if (pStatus->composedDataBlock) {
3885
    return pReader->pResBlock->pDataBlock;
3886
  }
3887

H
Haojun Liao 已提交
3888 3889 3890 3891 3892 3893 3894 3895
  SFileDataBlockInfo*  pBlockInfo = getCurrentBlockInfo(&pStatus->blockIter);
  STableBlockScanInfo* pBlockScanInfo = taosHashGet(pStatus->pTableMap, &pBlockInfo->uid, sizeof(pBlockInfo->uid));
  if (pBlockScanInfo == NULL) {
    terrno = TSDB_CODE_INVALID_PARA;
    tsdbError("failed to locate the uid:%" PRIu64 " in query table uid list, total tables:%d, %s", pBlockInfo->uid,
              taosHashGetSize(pReader->status.pTableMap), pReader->idStr);
    return NULL;
  }
3896

3897
  int32_t code = doLoadFileBlockData(pReader, &pStatus->blockIter, &pStatus->fileBlockData, pBlockScanInfo->uid);
3898
  if (code != TSDB_CODE_SUCCESS) {
H
Hongze Cheng 已提交
3899
    tBlockDataDestroy(&pStatus->fileBlockData, 1);
3900 3901
    terrno = code;
    return NULL;
3902
  }
3903 3904 3905

  copyBlockDataToSDataBlock(pReader, pBlockScanInfo);
  return pReader->pResBlock->pDataBlock;
H
Hongze Cheng 已提交
3906 3907
}

3908 3909 3910 3911 3912 3913 3914 3915 3916 3917 3918 3919
SArray* tsdbRetrieveDataBlock(STsdbReader* pReader, SArray* pIdList) {
  if (pReader->type == TIMEWINDOW_RANGE_EXTERNAL) {
    if (pReader->step == EXTERNAL_ROWS_PREV) {
      return doRetrieveDataBlock(pReader->innerReader[0]);
    } else if (pReader->step == EXTERNAL_ROWS_NEXT) {
      return doRetrieveDataBlock(pReader->innerReader[1]);
    }
  }

  return doRetrieveDataBlock(pReader);
}

H
Haojun Liao 已提交
3920
int32_t tsdbReaderReset(STsdbReader* pReader, SQueryTableDataCond* pCond) {
H
Haojun Liao 已提交
3921
  if (isEmptyQueryTimeWindow(&pReader->window) || pReader->pReadSnap == NULL) {
3922 3923
    return TSDB_CODE_SUCCESS;
  }
H
Hongze Cheng 已提交
3924

L
Liu Jicong 已提交
3925
  pReader->order = pCond->order;
3926
  pReader->type = TIMEWINDOW_RANGE_CONTAINED;
3927
  pReader->status.loadFromFile = true;
dengyihao's avatar
dengyihao 已提交
3928
  pReader->status.pTableIter = NULL;
H
Haojun Liao 已提交
3929
  pReader->window = updateQueryTimeWindow(pReader->pTsdb, &pCond->twindows);
H
Hongze Cheng 已提交
3930

3931
  // allocate buffer in order to load data blocks from file
3932
  memset(&pReader->suppInfo.tsColAgg, 0, sizeof(SColumnDataAgg));
3933 3934
  memset(pReader->suppInfo.plist, 0, POINTER_BYTES);

3935
  pReader->suppInfo.tsColAgg.colId = PRIMARYKEY_TIMESTAMP_COL_ID;
3936
  tsdbDataFReaderClose(&pReader->pFileReader);
3937

3938
  int32_t numOfTables = taosHashGetSize(pReader->status.pTableMap);
L
Liu Jicong 已提交
3939

3940
  initFilesetIterator(&pReader->status.fileIter, pReader->pReadSnap->fs.aDFileSet, pReader);
3941
  resetDataBlockIterator(&pReader->status.blockIter, pReader->order);
H
Haojun Liao 已提交
3942

H
Hongze Cheng 已提交
3943
  int64_t ts = ASCENDING_TRAVERSE(pReader->order) ? pReader->window.skey - 1 : pReader->window.ekey + 1;
H
Haojun Liao 已提交
3944
  resetDataBlockScanInfo(pReader->status.pTableMap, ts);
3945

3946
  int32_t         code = 0;
3947 3948
  SDataBlockIter* pBlockIter = &pReader->status.blockIter;

3949 3950 3951 3952 3953 3954
  // no data in files, let's try buffer in memory
  if (pReader->status.fileIter.numOfFiles == 0) {
    pReader->status.loadFromFile = false;
  } else {
    code = initForFirstBlockInFile(pReader, pBlockIter);
    if (code != TSDB_CODE_SUCCESS) {
3955 3956
      tsdbError("%p reset reader failed, numOfTables:%d, query range:%" PRId64 " - %" PRId64 " in query %s", pReader,
                numOfTables, pReader->window.skey, pReader->window.ekey, pReader->idStr);
3957 3958 3959
      return code;
    }
  }
H
Hongze Cheng 已提交
3960

H
Hongze Cheng 已提交
3961 3962 3963 3964
  tsdbDebug("%p reset reader, suid:%" PRIu64 ", numOfTables:%d, skey:%" PRId64 ", query range:%" PRId64 " - %" PRId64
            " in query %s",
            pReader, pReader->suid, numOfTables, pCond->twindows.skey, pReader->window.skey, pReader->window.ekey,
            pReader->idStr);
3965

3966
  return code;
H
Hongze Cheng 已提交
3967
}
H
Hongze Cheng 已提交
3968

3969 3970 3971
static int32_t getBucketIndex(int32_t startRow, int32_t bucketRange, int32_t numOfRows) {
  return (numOfRows - startRow) / bucketRange;
}
H
Hongze Cheng 已提交
3972

3973 3974 3975 3976
int32_t tsdbGetFileBlocksDistInfo(STsdbReader* pReader, STableBlockDistInfo* pTableBlockInfo) {
  int32_t code = TSDB_CODE_SUCCESS;
  pTableBlockInfo->totalSize = 0;
  pTableBlockInfo->totalRows = 0;
H
Hongze Cheng 已提交
3977

3978 3979
  // find the start data block in file
  SReaderStatus* pStatus = &pReader->status;
H
Hongze Cheng 已提交
3980

3981 3982 3983
  STsdbCfg* pc = &pReader->pTsdb->pVnode->config.tsdbCfg;
  pTableBlockInfo->defMinRows = pc->minRows;
  pTableBlockInfo->defMaxRows = pc->maxRows;
H
Hongze Cheng 已提交
3984

3985
  int32_t bucketRange = ceil((pc->maxRows - pc->minRows) / 20.0);
H
Hongze Cheng 已提交
3986

3987
  pTableBlockInfo->numOfFiles += 1;
H
Hongze Cheng 已提交
3988

3989 3990
  int32_t numOfTables = (int32_t)taosHashGetSize(pStatus->pTableMap);
  int     defaultRows = 4096;
H
Hongze Cheng 已提交
3991

3992 3993
  SDataBlockIter* pBlockIter = &pStatus->blockIter;
  pTableBlockInfo->numOfFiles += pStatus->fileIter.numOfFiles;
H
Haojun Liao 已提交
3994

3995 3996
  if (pBlockIter->numOfBlocks > 0) {
    pTableBlockInfo->numOfBlocks += pBlockIter->numOfBlocks;
H
Haojun Liao 已提交
3997
  }
H
Hongze Cheng 已提交
3998

3999
  pTableBlockInfo->numOfTables = numOfTables;
4000
  bool hasNext = (pBlockIter->numOfBlocks > 0);
H
Hongze Cheng 已提交
4001

4002 4003
  while (true) {
    if (hasNext) {
H
Hongze Cheng 已提交
4004
      SDataBlk* pBlock = getCurrentBlock(pBlockIter);
H
Hongze Cheng 已提交
4005

4006 4007
      int32_t numOfRows = pBlock->nRow;
      pTableBlockInfo->totalRows += numOfRows;
H
Hongze Cheng 已提交
4008

4009 4010 4011
      if (numOfRows > pTableBlockInfo->maxRows) {
        pTableBlockInfo->maxRows = numOfRows;
      }
H
refact  
Hongze Cheng 已提交
4012

4013 4014 4015
      if (numOfRows < pTableBlockInfo->minRows) {
        pTableBlockInfo->minRows = numOfRows;
      }
H
refact  
Hongze Cheng 已提交
4016

4017 4018 4019
      if (numOfRows < defaultRows) {
        pTableBlockInfo->numOfSmallBlocks += 1;
      }
H
refact  
Hongze Cheng 已提交
4020

4021 4022
      int32_t bucketIndex = getBucketIndex(pTableBlockInfo->defMinRows, bucketRange, numOfRows);
      pTableBlockInfo->blockRowsHisto[bucketIndex]++;
4023

H
Haojun Liao 已提交
4024
      hasNext = blockIteratorNext(&pStatus->blockIter, pReader->idStr);
4025 4026 4027 4028 4029
    } else {
      code = initForFirstBlockInFile(pReader, pBlockIter);
      if ((code != TSDB_CODE_SUCCESS) || (pReader->status.loadFromFile == false)) {
        break;
      }
H
refact  
Hongze Cheng 已提交
4030

4031 4032
      pTableBlockInfo->numOfBlocks += pBlockIter->numOfBlocks;
      hasNext = (pBlockIter->numOfBlocks > 0);
4033
    }
H
refact  
Hongze Cheng 已提交
4034

H
Hongze Cheng 已提交
4035 4036
    //    tsdbDebug("%p %d blocks found in file for %d table(s), fid:%d, %s", pReader, numOfBlocks, numOfTables,
    //              pReader->pFileGroup->fid, pReader->idStr);
4037
  }
H
Hongze Cheng 已提交
4038

H
refact  
Hongze Cheng 已提交
4039 4040
  return code;
}
H
Hongze Cheng 已提交
4041

H
refact  
Hongze Cheng 已提交
4042
int64_t tsdbGetNumOfRowsInMemTable(STsdbReader* pReader) {
4043
  int64_t rows = 0;
H
Hongze Cheng 已提交
4044

4045 4046
  SReaderStatus* pStatus = &pReader->status;
  pStatus->pTableIter = taosHashIterate(pStatus->pTableMap, NULL);
H
Hongze Cheng 已提交
4047

4048 4049 4050 4051 4052
  while (pStatus->pTableIter != NULL) {
    STableBlockScanInfo* pBlockScanInfo = pStatus->pTableIter;

    STbData* d = NULL;
    if (pReader->pTsdb->mem != NULL) {
H
Hongze Cheng 已提交
4053
      d = tsdbGetTbDataFromMemTable(pReader->pReadSnap->pMem, pReader->suid, pBlockScanInfo->uid);
4054 4055 4056 4057 4058 4059 4060
      if (d != NULL) {
        rows += tsdbGetNRowsInTbData(d);
      }
    }

    STbData* di = NULL;
    if (pReader->pTsdb->imem != NULL) {
H
Hongze Cheng 已提交
4061
      di = tsdbGetTbDataFromMemTable(pReader->pReadSnap->pIMem, pReader->suid, pBlockScanInfo->uid);
4062 4063 4064 4065 4066 4067 4068 4069
      if (di != NULL) {
        rows += tsdbGetNRowsInTbData(di);
      }
    }

    // current table is exhausted, let's try the next table
    pStatus->pTableIter = taosHashIterate(pStatus->pTableMap, pStatus->pTableIter);
  }
H
Hongze Cheng 已提交
4070

H
refact  
Hongze Cheng 已提交
4071
  return rows;
H
Hongze Cheng 已提交
4072
}
D
dapan1121 已提交
4073

L
Liu Jicong 已提交
4074
int32_t tsdbGetTableSchema(SVnode* pVnode, int64_t uid, STSchema** pSchema, int64_t* suid) {
D
dapan1121 已提交
4075 4076 4077 4078 4079 4080 4081 4082 4083 4084 4085 4086
  int32_t sversion = 1;

  SMetaReader mr = {0};
  metaReaderInit(&mr, pVnode->pMeta, 0);
  int32_t code = metaGetTableEntryByUid(&mr, uid);
  if (code != TSDB_CODE_SUCCESS) {
    terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
    metaReaderClear(&mr);
    return terrno;
  }

  *suid = 0;
L
Liu Jicong 已提交
4087

D
dapan1121 已提交
4088
  if (mr.me.type == TSDB_CHILD_TABLE) {
D
dapan1121 已提交
4089
    tDecoderClear(&mr.coder);
D
dapan1121 已提交
4090 4091 4092 4093 4094 4095 4096 4097 4098 4099 4100 4101 4102 4103
    *suid = mr.me.ctbEntry.suid;
    code = metaGetTableEntryByUid(&mr, *suid);
    if (code != TSDB_CODE_SUCCESS) {
      terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
      metaReaderClear(&mr);
      return terrno;
    }
    sversion = mr.me.stbEntry.schemaRow.version;
  } else {
    ASSERT(mr.me.type == TSDB_NORMAL_TABLE);
    sversion = mr.me.ntbEntry.schemaRow.version;
  }

  metaReaderClear(&mr);
4104
  *pSchema = metaGetTbTSchema(pVnode->pMeta, uid, sversion, 1);
L
Liu Jicong 已提交
4105

D
dapan1121 已提交
4106 4107
  return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
4108

H
Haojun Liao 已提交
4109
int32_t tsdbTakeReadSnap(STsdb* pTsdb, STsdbReadSnap** ppSnap, const char* idStr) {
H
Hongze Cheng 已提交
4110 4111 4112 4113 4114 4115 4116 4117 4118 4119 4120 4121 4122 4123 4124 4125 4126 4127 4128 4129 4130 4131 4132 4133 4134 4135 4136 4137
  int32_t code = 0;

  // alloc
  *ppSnap = (STsdbReadSnap*)taosMemoryCalloc(1, sizeof(STsdbReadSnap));
  if (*ppSnap == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

  // lock
  code = taosThreadRwlockRdlock(&pTsdb->rwLock);
  if (code) {
    code = TAOS_SYSTEM_ERROR(code);
    goto _exit;
  }

  // take snapshot
  (*ppSnap)->pMem = pTsdb->mem;
  (*ppSnap)->pIMem = pTsdb->imem;

  if ((*ppSnap)->pMem) {
    tsdbRefMemTable((*ppSnap)->pMem);
  }

  if ((*ppSnap)->pIMem) {
    tsdbRefMemTable((*ppSnap)->pIMem);
  }

H
Hongze Cheng 已提交
4138
  // fs
H
Hongze Cheng 已提交
4139 4140 4141 4142 4143
  code = tsdbFSRef(pTsdb, &(*ppSnap)->fs);
  if (code) {
    taosThreadRwlockUnlock(&pTsdb->rwLock);
    goto _exit;
  }
H
Hongze Cheng 已提交
4144 4145 4146 4147 4148 4149 4150 4151

  // unlock
  code = taosThreadRwlockUnlock(&pTsdb->rwLock);
  if (code) {
    code = TAOS_SYSTEM_ERROR(code);
    goto _exit;
  }

H
Haojun Liao 已提交
4152
  tsdbTrace("vgId:%d, take read snapshot, %s", TD_VID(pTsdb->pVnode), idStr);
H
Hongze Cheng 已提交
4153 4154 4155 4156
_exit:
  return code;
}

H
Haojun Liao 已提交
4157
void tsdbUntakeReadSnap(STsdb* pTsdb, STsdbReadSnap* pSnap, const char* idStr) {
H
Hongze Cheng 已提交
4158 4159 4160 4161 4162 4163 4164 4165 4166
  if (pSnap) {
    if (pSnap->pMem) {
      tsdbUnrefMemTable(pSnap->pMem);
    }

    if (pSnap->pIMem) {
      tsdbUnrefMemTable(pSnap->pIMem);
    }

H
Hongze Cheng 已提交
4167
    tsdbFSUnref(pTsdb, &pSnap->fs);
H
Hongze Cheng 已提交
4168
    taosMemoryFree(pSnap);
H
Hongze Cheng 已提交
4169
  }
H
Haojun Liao 已提交
4170 4171
  tsdbTrace("vgId:%d, untake read snapshot, %s", TD_VID(pTsdb->pVnode), idStr);
}