tsdbRead.c 165.5 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;

38
typedef struct SBlockIndex {
39 40
  int32_t     ordinalIndex;
  int64_t     inFileOffset;
H
Haojun Liao 已提交
41
  STimeWindow window;  // todo replace it with overlap flag.
42 43
} SBlockIndex;

H
Haojun Liao 已提交
44
typedef struct STableBlockScanInfo {
dengyihao's avatar
dengyihao 已提交
45 46
  uint64_t  uid;
  TSKEY     lastKey;
47
  TSKEY     lastKeyInStt;       // last accessed key in stt
H
Hongze Cheng 已提交
48
  SMapData  mapData;            // block info (compressed)
49
  SArray*   pBlockList;         // block data index list, SArray<SBlockIndex>
H
Hongze Cheng 已提交
50 51 52 53 54 55
  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 已提交
56 57 58
} STableBlockScanInfo;

typedef struct SBlockOrderWrapper {
dengyihao's avatar
dengyihao 已提交
59
  int64_t uid;
60
  int64_t offset;
H
Haojun Liao 已提交
61
} SBlockOrderWrapper;
H
Hongze Cheng 已提交
62 63

typedef struct SBlockOrderSupporter {
64 65 66 67
  SBlockOrderWrapper** pDataBlockInfo;
  int32_t*             indexPerTable;
  int32_t*             numOfBlocksPerTable;
  int32_t              numOfTables;
H
Hongze Cheng 已提交
68 69 70
} SBlockOrderSupporter;

typedef struct SIOCostSummary {
71 72 73
  int64_t numOfBlocks;
  double  blockLoadTime;
  double  buildmemBlock;
74
  int64_t headFileLoad;
75
  double  headFileLoadTime;
76
  int64_t smaDataLoad;
77
  double  smaLoadTime;
78 79
  int64_t lastBlockLoad;
  double  lastBlockLoadTime;
H
Haojun Liao 已提交
80 81
  int64_t composedBlocks;
  double  buildComposedBlockTime;
H
Haojun Liao 已提交
82
  double  createScanInfoList;
X
Xiaoyu Wang 已提交
83 84 85
  //  double  getTbFromMemTime;
  //  double  getTbFromIMemTime;
  double initDelSkylineIterTime;
H
Hongze Cheng 已提交
86 87 88
} SIOCostSummary;

typedef struct SBlockLoadSuppInfo {
89 90 91 92 93 94 95
  SArray*        pColAgg;
  SColumnDataAgg tsColAgg;
  int16_t*       colId;
  int16_t*       slotId;
  int32_t        numOfCols;
  char**         buildBuf;  // build string tmp buffer, todo remove it later after all string format being updated.
  bool           smaValid;  // the sma on all queried columns are activated
H
Hongze Cheng 已提交
96 97
} SBlockLoadSuppInfo;

98
typedef struct SLastBlockReader {
H
Hongze Cheng 已提交
99 100 101 102 103
  STimeWindow        window;
  SVersionRange      verRange;
  int32_t            order;
  uint64_t           uid;
  SMergeTree         mergeTree;
104
  SSttBlockLoadInfo* pInfo;
105 106
} SLastBlockReader;

107
typedef struct SFilesetIter {
H
Hongze Cheng 已提交
108 109 110
  int32_t           numOfFiles;  // number of total files
  int32_t           index;       // current accessed index in the list
  SArray*           pFileList;   // data file list
111
  int32_t           order;
H
Hongze Cheng 已提交
112
  SLastBlockReader* pLastBlockReader;  // last file block reader
113
} SFilesetIter;
H
Haojun Liao 已提交
114 115

typedef struct SFileDataBlockInfo {
116
  // index position in STableBlockScanInfo in order to check whether neighbor block overlaps with it
dengyihao's avatar
dengyihao 已提交
117
  uint64_t uid;
118
  int32_t  tbBlockIdx;
H
Haojun Liao 已提交
119 120 121
} SFileDataBlockInfo;

typedef struct SDataBlockIter {
122
  int32_t   numOfBlocks;
123
  int32_t   index;
H
Hongze Cheng 已提交
124
  SArray*   blockList;  // SArray<SFileDataBlockInfo>
125
  int32_t   order;
H
Hongze Cheng 已提交
126
  SDataBlk  block;  // current SDataBlk data
127
  SHashObj* pTableMap;
H
Haojun Liao 已提交
128 129 130
} SDataBlockIter;

typedef struct SFileBlockDumpInfo {
dengyihao's avatar
dengyihao 已提交
131 132 133 134
  int32_t totalRows;
  int32_t rowIndex;
  int64_t lastKey;
  bool    allDumped;
H
Haojun Liao 已提交
135 136
} SFileBlockDumpInfo;

137
typedef struct STableUidList {
138 139
  uint64_t* tableUidList;  // access table uid list in uid ascending order list
  int32_t   currentIndex;  // index in table uid list
140
} STableUidList;
141

H
Haojun Liao 已提交
142
typedef struct SReaderStatus {
H
Hongze Cheng 已提交
143 144 145
  bool                  loadFromFile;       // check file stage
  bool                  composedDataBlock;  // the returned data block is a composed block or not
  SHashObj*             pTableMap;          // SHash<STableBlockScanInfo>
146
  STableBlockScanInfo** pTableIter;         // table iterator used in building in-memory buffer data blocks.
147
  STableUidList         uidList;            // check tables in uid order, to avoid the repeatly load of blocks in STT.
H
Hongze Cheng 已提交
148 149 150 151 152
  SFileBlockDumpInfo    fBlockDumpInfo;
  SDFileSet*            pCurrentFileset;  // current opened file set
  SBlockData            fileBlockData;
  SFilesetIter          fileIter;
  SDataBlockIter        blockIter;
H
Haojun Liao 已提交
153 154
} SReaderStatus;

155
typedef struct SBlockInfoBuf {
H
Hongze Cheng 已提交
156 157 158
  int32_t currentIndex;
  SArray* pData;
  int32_t numPerBucket;
D
dapan1121 已提交
159
  int32_t numOfTables;
160 161
} SBlockInfoBuf;

H
Hongze Cheng 已提交
162
struct STsdbReader {
H
Haojun Liao 已提交
163
  STsdb*             pTsdb;
164 165 166
  SVersionRange      verRange;
  TdThreadMutex      readerMutex;
  bool               suspended;
H
Haojun Liao 已提交
167 168
  uint64_t           suid;
  int16_t            order;
H
Haojun Liao 已提交
169
  bool               freeBlock;
H
Haojun Liao 已提交
170 171 172 173
  STimeWindow        window;  // the primary query time window that applies to all queries
  SSDataBlock*       pResBlock;
  int32_t            capacity;
  SReaderStatus      status;
174 175
  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 已提交
176
  SBlockLoadSuppInfo suppInfo;
H
Hongze Cheng 已提交
177
  STsdbReadSnap*     pReadSnap;
178
  SIOCostSummary     cost;
179 180 181 182 183
  STSchema*          pSchema;      // the newest version schema
  STSchema*          pMemSchema;   // the previous schema for in-memory data, to avoid load schema too many times
  SDataFReader*      pFileReader;  // the file reader
  SDelFReader*       pDelFReader;  // the del file reader
  SArray*            pDelIdx;      // del file block index;
184 185 186
  SBlockInfoBuf      blockInfoBuf;
  int32_t            step;
  STsdbReader*       innerReader[2];
H
Hongze Cheng 已提交
187
};
H
Hongze Cheng 已提交
188

H
Haojun Liao 已提交
189
static SFileDataBlockInfo* getCurrentBlockInfo(SDataBlockIter* pBlockIter);
190 191
static int      buildDataBlockFromBufImpl(STableBlockScanInfo* pBlockScanInfo, int64_t endKey, int32_t capacity,
                                          STsdbReader* pReader);
192
static TSDBROW* getValidMemRow(SIterInfo* pIter, const SArray* pDelList, STsdbReader* pReader);
193 194
static int32_t  doMergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pScanInfo, STsdbReader* pReader,
                                        SRowMerger* pMerger);
H
Hongze Cheng 已提交
195
static int32_t  doMergeRowsInLastBlock(SLastBlockReader* pLastBlockReader, STableBlockScanInfo* pScanInfo, int64_t ts,
196
                                       SRowMerger* pMerger, SVersionRange* pVerRange, const char* id);
197
static int32_t  doMergeRowsInBuf(SIterInfo* pIter, uint64_t uid, int64_t ts, SArray* pDelList, SRowMerger* pMerger,
dengyihao's avatar
dengyihao 已提交
198
                                 STsdbReader* pReader);
H
Hongze Cheng 已提交
199
static int32_t  doAppendRowFromTSRow(SSDataBlock* pBlock, STsdbReader* pReader, SRow* pTSRow,
H
Hongze Cheng 已提交
200
                                     STableBlockScanInfo* pInfo);
201
static int32_t  doAppendRowFromFileBlock(SSDataBlock* pResBlock, STsdbReader* pReader, SBlockData* pBlockData,
H
Hongze Cheng 已提交
202
                                         int32_t rowIndex);
203
static void     setComposedBlockFlag(STsdbReader* pReader, bool composed);
H
Hongze Cheng 已提交
204 205
static bool     hasBeenDropped(const SArray* pDelList, int32_t* index, TSDBKEY* pKey, int32_t order,
                               SVersionRange* pVerRange);
206

H
Hongze Cheng 已提交
207
static int32_t doMergeMemTableMultiRows(TSDBROW* pRow, uint64_t uid, SIterInfo* pIter, SArray* pDelList,
208
                                        TSDBROW* pTSRow, STsdbReader* pReader, bool* freeTSRow);
H
Hongze Cheng 已提交
209
static int32_t doMergeMemIMemRows(TSDBROW* pRow, TSDBROW* piRow, STableBlockScanInfo* pBlockScanInfo,
H
Hongze Cheng 已提交
210
                                  STsdbReader* pReader, SRow** pTSRow);
211 212
static int32_t mergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pBlockScanInfo, int64_t key,
                                     STsdbReader* pReader);
213

dengyihao's avatar
dengyihao 已提交
214 215 216 217
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);
218
static SVersionRange getQueryVerRange(SVnode* pVnode, SQueryTableDataCond* pCond, int8_t level);
H
Hongze Cheng 已提交
219 220 221
static int64_t       getCurrentKeyInLastBlock(SLastBlockReader* pLastBlockReader);
static bool          hasDataInLastBlock(SLastBlockReader* pLastBlockReader);
static int32_t       doBuildDataBlock(STsdbReader* pReader);
C
Cary Xu 已提交
222
static TSDBKEY       getCurrentKeyInBuf(STableBlockScanInfo* pScanInfo, STsdbReader* pReader);
223
static bool          hasDataInFileBlock(const SBlockData* pBlockData, const SFileBlockDumpInfo* pDumpInfo);
224
static void          initBlockDumpInfo(STsdbReader* pReader, SDataBlockIter* pBlockIter);
225
static int32_t       getInitialDelIndex(const SArray* pDelSkyline, int32_t order);
C
Cary Xu 已提交
226

H
Haojun Liao 已提交
227 228
static STableBlockScanInfo* getTableBlockScanInfo(SHashObj* pTableMap, uint64_t uid, const char* id);

229 230
static FORCE_INLINE STSchema* getLatestTableSchema(STsdbReader* pReader, uint64_t uid);

C
Cary Xu 已提交
231
static bool outOfTimeWindow(int64_t ts, STimeWindow* pWindow) { return (ts > pWindow->ekey) || (ts < pWindow->skey); }
H
Haojun Liao 已提交
232

233 234
static int32_t setColumnIdSlotList(SBlockLoadSuppInfo* pSupInfo, SColumnInfo* pCols, const int32_t* pSlotIdList,
                                   int32_t numOfCols) {
235
  pSupInfo->smaValid = true;
236
  pSupInfo->numOfCols = numOfCols;
237
  pSupInfo->colId = taosMemoryMalloc(numOfCols * (sizeof(int16_t) * 2 + POINTER_BYTES));
H
Haojun Liao 已提交
238 239
  if (pSupInfo->colId == NULL) {
    taosMemoryFree(pSupInfo->colId);
H
Haojun Liao 已提交
240 241
    return TSDB_CODE_OUT_OF_MEMORY;
  }
H
Hongze Cheng 已提交
242

H
Haojun Liao 已提交
243
  pSupInfo->slotId = (int16_t*)((char*)pSupInfo->colId + (sizeof(int16_t) * numOfCols));
244
  pSupInfo->buildBuf = (char**)((char*)pSupInfo->slotId + (sizeof(int16_t) * numOfCols));
H
Haojun Liao 已提交
245
  for (int32_t i = 0; i < numOfCols; ++i) {
H
Haojun Liao 已提交
246 247
    pSupInfo->colId[i] = pCols[i].colId;
    pSupInfo->slotId[i] = pSlotIdList[i];
248

H
Haojun Liao 已提交
249 250
    if (IS_VAR_DATA_TYPE(pCols[i].type)) {
      pSupInfo->buildBuf[i] = taosMemoryMalloc(pCols[i].bytes);
H
Haojun Liao 已提交
251 252
    } else {
      pSupInfo->buildBuf[i] = NULL;
253
    }
H
Haojun Liao 已提交
254
  }
H
Hongze Cheng 已提交
255

H
Haojun Liao 已提交
256 257
  return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
258

H
Haojun Liao 已提交
259
static int32_t updateBlockSMAInfo(STSchema* pSchema, SBlockLoadSuppInfo* pSupInfo) {
260 261
  int32_t i = 0, j = 0;

H
Hongze Cheng 已提交
262
  while (i < pSchema->numOfCols && j < pSupInfo->numOfCols) {
263
    STColumn* pTCol = &pSchema->columns[i];
H
Haojun Liao 已提交
264
    if (pTCol->colId == pSupInfo->colId[j]) {
265 266
      if (!IS_BSMA_ON(pTCol)) {
        pSupInfo->smaValid = false;
H
Haojun Liao 已提交
267
        return TSDB_CODE_SUCCESS;
268 269 270 271
      }

      i += 1;
      j += 1;
H
Haojun Liao 已提交
272
    } else if (pTCol->colId < pSupInfo->colId[j]) {
273 274 275
      // do nothing
      i += 1;
    } else {
H
Haojun Liao 已提交
276
      return TSDB_CODE_INVALID_PARA;
277 278
    }
  }
H
Haojun Liao 已提交
279 280

  return TSDB_CODE_SUCCESS;
281 282
}

283
static int32_t initBlockScanInfoBuf(SBlockInfoBuf* pBuf, int32_t numOfTables) {
H
Hongze Cheng 已提交
284
  int32_t num = numOfTables / pBuf->numPerBucket;
285 286 287 288 289
  int32_t remainder = numOfTables % pBuf->numPerBucket;
  if (pBuf->pData == NULL) {
    pBuf->pData = taosArrayInit(num + 1, POINTER_BYTES);
  }

H
Hongze Cheng 已提交
290
  for (int32_t i = 0; i < num; ++i) {
291 292 293 294
    char* p = taosMemoryCalloc(pBuf->numPerBucket, sizeof(STableBlockScanInfo));
    if (p == NULL) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }
295

296 297 298 299 300 301 302
    taosArrayPush(pBuf->pData, &p);
  }

  if (remainder > 0) {
    char* p = taosMemoryCalloc(remainder, sizeof(STableBlockScanInfo));
    if (p == NULL) {
      return TSDB_CODE_OUT_OF_MEMORY;
303
    }
304
    taosArrayPush(pBuf->pData, &p);
H
Haojun Liao 已提交
305
  }
H
Hongze Cheng 已提交
306

D
dapan1121 已提交
307 308 309 310 311 312 313 314 315 316 317
  pBuf->numOfTables = numOfTables;

  return TSDB_CODE_SUCCESS;
}

static int32_t ensureBlockScanInfoBuf(SBlockInfoBuf* pBuf, int32_t numOfTables) {
  if (numOfTables <= pBuf->numOfTables) {
    return TSDB_CODE_SUCCESS;
  }

  if (pBuf->numOfTables > 0) {
D
dapan1121 已提交
318 319
    STableBlockScanInfo **p = (STableBlockScanInfo**)taosArrayPop(pBuf->pData);
    taosMemoryFree(*p);
D
dapan1121 已提交
320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347
    pBuf->numOfTables /= pBuf->numPerBucket;
  }
  
  int32_t num = (numOfTables - pBuf->numOfTables) / pBuf->numPerBucket;
  int32_t remainder = (numOfTables - pBuf->numOfTables) % pBuf->numPerBucket;
  if (pBuf->pData == NULL) {
    pBuf->pData = taosArrayInit(num + 1, POINTER_BYTES);
  }

  for (int32_t i = 0; i < num; ++i) {
    char* p = taosMemoryCalloc(pBuf->numPerBucket, sizeof(STableBlockScanInfo));
    if (p == NULL) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }

    taosArrayPush(pBuf->pData, &p);
  }

  if (remainder > 0) {
    char* p = taosMemoryCalloc(remainder, sizeof(STableBlockScanInfo));
    if (p == NULL) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }
    taosArrayPush(pBuf->pData, &p);
  }

  pBuf->numOfTables = numOfTables;

H
Haojun Liao 已提交
348 349
  return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
350

351 352
static void clearBlockScanInfoBuf(SBlockInfoBuf* pBuf) {
  size_t num = taosArrayGetSize(pBuf->pData);
H
Hongze Cheng 已提交
353
  for (int32_t i = 0; i < num; ++i) {
354 355 356 357 358 359 360 361 362
    char** p = taosArrayGet(pBuf->pData, i);
    taosMemoryFree(*p);
  }

  taosArrayDestroy(pBuf->pData);
}

static void* getPosInBlockInfoBuf(SBlockInfoBuf* pBuf, int32_t index) {
  int32_t bucketIndex = index / pBuf->numPerBucket;
H
Hongze Cheng 已提交
363
  char**  pBucket = taosArrayGet(pBuf->pData, bucketIndex);
364 365 366
  return (*pBucket) + (index % pBuf->numPerBucket) * sizeof(STableBlockScanInfo);
}

H
Haojun Liao 已提交
367 368 369 370 371 372 373 374 375 376
static int32_t uidComparFunc(const void* p1, const void* p2) {
  uint64_t pu1 = *(uint64_t*)p1;
  uint64_t pu2 = *(uint64_t*)p2;
  if (pu1 == pu2) {
    return 0;
  } else {
    return (pu1 < pu2) ? -1 : 1;
  }
}

377
// NOTE: speedup the whole processing by preparing the buffer for STableBlockScanInfo in batch model
H
Hongze Cheng 已提交
378
static SHashObj* createDataBlockScanInfo(STsdbReader* pTsdbReader, SBlockInfoBuf* pBuf, const STableKeyInfo* idList,
X
Xiaoyu Wang 已提交
379
                                         STableUidList* pUidList, int32_t numOfTables) {
H
Haojun Liao 已提交
380
  // allocate buffer in order to load data blocks from file
381
  // todo use simple hash instead, optimize the memory consumption
382 383 384
  SHashObj* pTableMap =
      taosHashInit(numOfTables, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
  if (pTableMap == NULL) {
H
Haojun Liao 已提交
385 386 387
    return NULL;
  }

H
Haojun Liao 已提交
388
  int64_t st = taosGetTimestampUs();
H
Haojun Liao 已提交
389
  initBlockScanInfoBuf(pBuf, numOfTables);
H
Haojun Liao 已提交
390

H
Haojun Liao 已提交
391 392
  pUidList->tableUidList = taosMemoryMalloc(numOfTables * sizeof(uint64_t));
  if (pUidList->tableUidList == NULL) {
H
Haojun Liao 已提交
393 394
    return NULL;
  }
H
Haojun Liao 已提交
395
  pUidList->currentIndex = 0;
H
Haojun Liao 已提交
396

397
  for (int32_t j = 0; j < numOfTables; ++j) {
H
Haojun Liao 已提交
398
    STableBlockScanInfo* pScanInfo = getPosInBlockInfoBuf(pBuf, j);
H
Haojun Liao 已提交
399

400
    pScanInfo->uid = idList[j].uid;
H
Haojun Liao 已提交
401
    pUidList->tableUidList[j] = idList[j].uid;
H
Haojun Liao 已提交
402

403
    if (ASCENDING_TRAVERSE(pTsdbReader->order)) {
H
Haojun Liao 已提交
404
      int64_t skey = pTsdbReader->window.skey;
405
      pScanInfo->lastKey = (skey > INT64_MIN) ? (skey - 1) : skey;
H
Haojun Liao 已提交
406
      pScanInfo->lastKeyInStt = skey;
wmmhello's avatar
wmmhello 已提交
407
    } else {
H
Haojun Liao 已提交
408
      int64_t ekey = pTsdbReader->window.ekey;
409
      pScanInfo->lastKey = (ekey < INT64_MAX) ? (ekey + 1) : ekey;
H
Haojun Liao 已提交
410
      pScanInfo->lastKeyInStt = ekey;
H
Haojun Liao 已提交
411
    }
wmmhello's avatar
wmmhello 已提交
412

413
    taosHashPut(pTableMap, &pScanInfo->uid, sizeof(uint64_t), &pScanInfo, POINTER_BYTES);
H
Hongze Cheng 已提交
414 415
    tsdbTrace("%p check table uid:%" PRId64 " from lastKey:%" PRId64 " %s", pTsdbReader, pScanInfo->uid,
              pScanInfo->lastKey, pTsdbReader->idStr);
H
Haojun Liao 已提交
416 417
  }

H
Haojun Liao 已提交
418
  taosSort(pUidList->tableUidList, numOfTables, sizeof(uint64_t), uidComparFunc);
H
Haojun Liao 已提交
419

H
Haojun Liao 已提交
420 421 422 423
  pTsdbReader->cost.createScanInfoList = (taosGetTimestampUs() - st) / 1000.0;
  tsdbDebug("%p create %d tables scan-info, size:%.2f Kb, elapsed time:%.2f ms, %s", pTsdbReader, numOfTables,
            (sizeof(STableBlockScanInfo) * numOfTables) / 1024.0, pTsdbReader->cost.createScanInfoList,
            pTsdbReader->idStr);
424

425
  return pTableMap;
H
Hongze Cheng 已提交
426
}
H
Hongze Cheng 已提交
427

428 429
static void resetAllDataBlockScanInfo(SHashObj* pTableMap, int64_t ts) {
  STableBlockScanInfo** p = NULL;
dengyihao's avatar
dengyihao 已提交
430
  while ((p = taosHashIterate(pTableMap, p)) != NULL) {
H
Hongze Cheng 已提交
431
    STableBlockScanInfo* pInfo = *(STableBlockScanInfo**)p;
432 433

    pInfo->iterInit = false;
H
Haojun Liao 已提交
434
    pInfo->iter.hasVal = false;
435
    pInfo->iiter.hasVal = false;
H
Haojun Liao 已提交
436

437 438
    if (pInfo->iter.iter != NULL) {
      pInfo->iter.iter = tsdbTbDataIterDestroy(pInfo->iter.iter);
439 440
    }

H
Haojun Liao 已提交
441 442
    if (pInfo->iiter.iter != NULL) {
      pInfo->iiter.iter = tsdbTbDataIterDestroy(pInfo->iiter.iter);
443 444
    }

445 446
    pInfo->delSkyline = taosArrayDestroy(pInfo->delSkyline);
    pInfo->lastKey = ts;
447 448 449
  }
}

450 451
static void clearBlockScanInfo(STableBlockScanInfo* p) {
  p->iterInit = false;
H
Haojun Liao 已提交
452 453

  p->iter.hasVal = false;
454
  p->iiter.hasVal = false;
455

456 457 458
  if (p->iter.iter != NULL) {
    p->iter.iter = tsdbTbDataIterDestroy(p->iter.iter);
  }
459

460 461 462
  if (p->iiter.iter != NULL) {
    p->iiter.iter = tsdbTbDataIterDestroy(p->iiter.iter);
  }
463

464 465 466 467
  p->delSkyline = taosArrayDestroy(p->delSkyline);
  p->pBlockList = taosArrayDestroy(p->pBlockList);
  tMapDataClear(&p->mapData);
}
468

H
Haojun Liao 已提交
469
static void destroyAllBlockScanInfo(SHashObj* pTableMap) {
470
  void* p = NULL;
471
  while ((p = taosHashIterate(pTableMap, p)) != NULL) {
472
    clearBlockScanInfo(*(STableBlockScanInfo**)p);
473 474 475 476 477
  }

  taosHashCleanup(pTableMap);
}

478
static bool isEmptyQueryTimeWindow(STimeWindow* pWindow) { return pWindow->skey > pWindow->ekey; }
H
Hongze Cheng 已提交
479

480 481 482
// 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 已提交
483
  STsdbKeepCfg* pCfg = &pTsdb->keepCfg;
H
Hongze Cheng 已提交
484

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

dengyihao's avatar
dengyihao 已提交
488
  STimeWindow win = *pWindow;
489 490 491 492 493 494
  if (win.skey < earilyTs) {
    win.skey = earilyTs;
  }

  return win;
}
H
Hongze Cheng 已提交
495

H
Haojun Liao 已提交
496
// init file iterator
497
static int32_t initFilesetIterator(SFilesetIter* pIter, SArray* aDFileSet, STsdbReader* pReader) {
H
Hongze Cheng 已提交
498
  size_t numOfFileset = taosArrayGetSize(aDFileSet);
499

500 501
  pIter->index = ASCENDING_TRAVERSE(pReader->order) ? -1 : numOfFileset;
  pIter->order = pReader->order;
H
Hongze Cheng 已提交
502
  pIter->pFileList = aDFileSet;
503
  pIter->numOfFiles = numOfFileset;
H
Haojun Liao 已提交
504

505 506 507 508
  if (pIter->pLastBlockReader == NULL) {
    pIter->pLastBlockReader = taosMemoryCalloc(1, sizeof(struct SLastBlockReader));
    if (pIter->pLastBlockReader == NULL) {
      int32_t code = TSDB_CODE_OUT_OF_MEMORY;
509
      tsdbError("failed to prepare the last block iterator, since:%s %s", tstrerror(code), pReader->idStr);
510 511
      return code;
    }
512 513
  }

514 515 516 517 518 519 520 521
  SLastBlockReader* pLReader = pIter->pLastBlockReader;
  pLReader->order = pReader->order;
  pLReader->window = pReader->window;
  pLReader->verRange = pReader->verRange;

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

522
  if (pLReader->pInfo == NULL) {
523
    // here we ignore the first column, which is always be the primary timestamp column
524 525 526
    SBlockLoadSuppInfo* pInfo = &pReader->suppInfo;

    int32_t numOfStt = pReader->pTsdb->pVnode->config.sttTrigger;
X
Xiaoyu Wang 已提交
527
    pLReader->pInfo = tCreateLastBlockLoadInfo(pReader->pSchema, &pInfo->colId[1], pInfo->numOfCols - 1, numOfStt);
H
Haojun Liao 已提交
528 529 530 531
    if (pLReader->pInfo == NULL) {
      tsdbDebug("init fileset iterator failed, code:%s %s", tstrerror(terrno), pReader->idStr);
      return terrno;
    }
532 533
  }

534
  tsdbDebug("init fileset iterator, total files:%d %s", pIter->numOfFiles, pReader->idStr);
H
Haojun Liao 已提交
535 536 537
  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
538
static int32_t filesetIteratorNext(SFilesetIter* pIter, STsdbReader* pReader, bool* hasNext) {
539 540
  bool    asc = ASCENDING_TRAVERSE(pIter->order);
  int32_t step = asc ? 1 : -1;
541
  pIter->index += step;
D
dapan1121 已提交
542
  int32_t code = 0;
543 544

  if ((asc && pIter->index >= pIter->numOfFiles) || ((!asc) && pIter->index < 0)) {
D
dapan1121 已提交
545 546
    *hasNext = false;
    return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
547 548
  }

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

552 553
  pIter->pLastBlockReader->uid = 0;
  tMergeTreeClose(&pIter->pLastBlockReader->mergeTree);
554
  resetLastBlockLoadInfo(pIter->pLastBlockReader->pInfo);
555

H
Haojun Liao 已提交
556 557
  // check file the time range of coverage
  STimeWindow win = {0};
H
Hongze Cheng 已提交
558

559
  while (1) {
H
Haojun Liao 已提交
560 561 562
    if (pReader->pFileReader != NULL) {
      tsdbDataFReaderClose(&pReader->pFileReader);
    }
563

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

D
dapan1121 已提交
566
    code = tsdbDataFReaderOpen(&pReader->pFileReader, pReader->pTsdb, pReader->status.pCurrentFileset);
567 568 569
    if (code != TSDB_CODE_SUCCESS) {
      goto _err;
    }
H
Haojun Liao 已提交
570

571 572
    pReader->cost.headFileLoad += 1;

573 574 575 576 577 578 579
    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);
D
dapan1121 已提交
580 581
      *hasNext = false;
      return TSDB_CODE_SUCCESS;
582 583 584 585
    }

    if ((asc && (win.ekey < pReader->window.skey)) || ((!asc) && (win.skey > pReader->window.ekey))) {
      pIter->index += step;
586
      if ((asc && pIter->index >= pIter->numOfFiles) || ((!asc) && pIter->index < 0)) {
D
dapan1121 已提交
587 588
        *hasNext = false;
        return TSDB_CODE_SUCCESS;
589
      }
590 591
      continue;
    }
C
Cary Xu 已提交
592

593
    tsdbDebug("%p file found fid:%d for qrange:%" PRId64 "-%" PRId64 ", %s", pReader, fid, pReader->window.skey,
594
              pReader->window.ekey, pReader->idStr);
D
dapan1121 已提交
595 596
    *hasNext = true;
    return TSDB_CODE_SUCCESS;
597
  }
598

599
_err:
D
dapan1121 已提交
600 601
  *hasNext = false;
  return code;
H
Haojun Liao 已提交
602 603
}

604
static void resetDataBlockIterator(SDataBlockIter* pIter, int32_t order) {
605 606
  pIter->order = order;
  pIter->index = -1;
607
  pIter->numOfBlocks = 0;
608 609 610 611 612 613 614
  if (pIter->blockList == NULL) {
    pIter->blockList = taosArrayInit(4, sizeof(SFileDataBlockInfo));
  } else {
    taosArrayClear(pIter->blockList);
  }
}

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

H
Haojun Liao 已提交
617
static void initReaderStatus(SReaderStatus* pStatus) {
dengyihao's avatar
dengyihao 已提交
618 619
  pStatus->pTableIter = NULL;
  pStatus->loadFromFile = true;
H
Haojun Liao 已提交
620 621
}

622 623 624 625 626 627 628 629
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) {
H
Haojun Liao 已提交
630
    SColumnInfoData colInfo = {0};
631 632 633 634 635 636 637 638 639 640 641 642 643
    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;
}

644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698
static int32_t tsdbInitReaderLock(STsdbReader* pReader) {
  int32_t code = -1;
  qTrace("tsdb/read: %p, pre-init read mutex: %p, code: %d", pReader, &pReader->readerMutex, code);

  code = taosThreadMutexInit(&pReader->readerMutex, NULL);

  qTrace("tsdb/read: %p, post-init read mutex: %p, code: %d", pReader, &pReader->readerMutex, code);

  return code;
}

static int32_t tsdbUninitReaderLock(STsdbReader* pReader) {
  int32_t code = -1;
  qTrace("tsdb/read: %p, pre-uninit read mutex: %p, code: %d", pReader, &pReader->readerMutex, code);

  code = taosThreadMutexDestroy(&pReader->readerMutex);

  qTrace("tsdb/read: %p, post-uninit read mutex: %p, code: %d", pReader, &pReader->readerMutex, code);

  return code;
}

static int32_t tsdbAcquireReader(STsdbReader* pReader) {
  int32_t code = -1;
  qTrace("tsdb/read: %p, pre-take read mutex: %p, code: %d", pReader, &pReader->readerMutex, code);

  code = taosThreadMutexLock(&pReader->readerMutex);

  qTrace("tsdb/read: %p, post-take read mutex: %p, code: %d", pReader, &pReader->readerMutex, code);

  return code;
}

static int32_t tsdbTryAcquireReader(STsdbReader* pReader) {
  int32_t code = -1;
  qTrace("tsdb/read: %p, pre-trytake read mutex: %p, code: %d", pReader, &pReader->readerMutex, code);

  code = taosThreadMutexTryLock(&pReader->readerMutex);

  qTrace("tsdb/read: %p, post-trytake read mutex: %p, code: %d", pReader, &pReader->readerMutex, code);

  return code;
}

static int32_t tsdbReleaseReader(STsdbReader* pReader) {
  int32_t code = -1;
  qTrace("tsdb/read: %p, pre-untake read mutex: %p, code: %d", pReader, &pReader->readerMutex, code);

  code = taosThreadMutexUnlock(&pReader->readerMutex);

  qTrace("tsdb/read: %p, post-untake read mutex: %p, code: %d", pReader, &pReader->readerMutex, code);

  return code;
}

699 700 701 702 703 704
void tsdbReleaseDataBlock(STsdbReader* pReader) {
  SReaderStatus* pStatus = &pReader->status;
  if (!pStatus->composedDataBlock) {
    tsdbReleaseReader(pReader);
  }
}
705

706
static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, STsdbReader** ppReader, int32_t capacity,
H
Haojun Liao 已提交
707
                                SSDataBlock* pResBlock, const char* idstr) {
H
Haojun Liao 已提交
708
  int32_t      code = 0;
709
  int8_t       level = 0;
H
Haojun Liao 已提交
710
  STsdbReader* pReader = (STsdbReader*)taosMemoryCalloc(1, sizeof(*pReader));
H
Hongze Cheng 已提交
711 712
  if (pReader == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
H
Haojun Liao 已提交
713
    goto _end;
H
Hongze Cheng 已提交
714 715
  }

C
Cary Xu 已提交
716
  if (VND_IS_TSMA(pVnode)) {
H
Haojun Liao 已提交
717
    tsdbDebug("vgId:%d, tsma is selected to query, %s", TD_VID(pVnode), idstr);
C
Cary Xu 已提交
718 719
  }

H
Haojun Liao 已提交
720
  initReaderStatus(&pReader->status);
721

L
Liu Jicong 已提交
722
  pReader->pTsdb = getTsdbByRetentions(pVnode, pCond->twindows.skey, pVnode->config.tsdbCfg.retentions, idstr, &level);
dengyihao's avatar
dengyihao 已提交
723 724
  pReader->suid = pCond->suid;
  pReader->order = pCond->order;
725
  pReader->capacity = capacity;
H
Haojun Liao 已提交
726
  pReader->pResBlock = pResBlock;
727
  pReader->idStr = (idstr != NULL) ? taosStrdup(idstr) : NULL;
dengyihao's avatar
dengyihao 已提交
728
  pReader->verRange = getQueryVerRange(pVnode, pCond, level);
729
  pReader->type = pCond->type;
730
  pReader->window = updateQueryTimeWindow(pReader->pTsdb, &pCond->twindows);
H
Hongze Cheng 已提交
731
  pReader->blockInfoBuf.numPerBucket = 1000;  // 1000 tables per bucket
H
Hongze Cheng 已提交
732

H
Haojun Liao 已提交
733 734 735 736 737 738 739 740
  if (pReader->pResBlock == NULL) {
    pReader->freeBlock = true;
    pReader->pResBlock = createResBlock(pCond, pReader->capacity);
    if (pReader->pResBlock == NULL) {
      code = terrno;
      goto _end;
    }
  }
741

H
Haojun Liao 已提交
742 743 744 745 746
  if (pCond->numOfCols <= 0) {
    tsdbError("vgId:%d, invalid column number %d in query cond, %s", TD_VID(pVnode), pCond->numOfCols, idstr);
    code = TSDB_CODE_INVALID_PARA;
    goto _end;
  }
H
Hongze Cheng 已提交
747

748 749
  // allocate buffer in order to load data blocks from file
  SBlockLoadSuppInfo* pSup = &pReader->suppInfo;
750
  pSup->pColAgg = taosArrayInit(pCond->numOfCols, sizeof(SColumnDataAgg));
H
Haojun Liao 已提交
751
  if (pSup->pColAgg == NULL) {
752 753 754
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _end;
  }
H
Haojun Liao 已提交
755

756 757
  pSup->tsColAgg.colId = PRIMARYKEY_TIMESTAMP_COL_ID;

H
Hongze Cheng 已提交
758
  code = tBlockDataCreate(&pReader->status.fileBlockData);
H
Haojun Liao 已提交
759 760 761 762 763
  if (code != TSDB_CODE_SUCCESS) {
    terrno = code;
    goto _end;
  }

764
  setColumnIdSlotList(pSup, pCond->colList, pCond->pSlotList, pCond->numOfCols);
765

766
  tsdbInitReaderLock(pReader);
767

H
Hongze Cheng 已提交
768 769
  *ppReader = pReader;
  return code;
H
Hongze Cheng 已提交
770

H
Haojun Liao 已提交
771 772
_end:
  tsdbReaderClose(pReader);
H
Hongze Cheng 已提交
773 774 775
  *ppReader = NULL;
  return code;
}
H
Hongze Cheng 已提交
776

H
Haojun Liao 已提交
777
static int32_t doLoadBlockIndex(STsdbReader* pReader, SDataFReader* pFileReader, SArray* pIndexList) {
X
Xiaoyu Wang 已提交
778
  int64_t    st = taosGetTimestampUs();
779 780 781
  LRUHandle* handle = NULL;
  int32_t    code = tsdbCacheGetBlockIdx(pFileReader->pTsdb->biCache, pFileReader, &handle);
  if (code != TSDB_CODE_SUCCESS || handle == NULL) {
782
    goto _end;
H
Haojun Liao 已提交
783
  }
H
Hongze Cheng 已提交
784

H
Haojun Liao 已提交
785 786
  int32_t numOfTables = taosHashGetSize(pReader->status.pTableMap);

787 788
  SArray* aBlockIdx = (SArray*)taosLRUCacheValue(pFileReader->pTsdb->biCache, handle);
  size_t  num = taosArrayGetSize(aBlockIdx);
789
  if (num == 0) {
790
    tsdbBICacheRelease(pFileReader->pTsdb->biCache, handle);
H
Haojun Liao 已提交
791 792
    return TSDB_CODE_SUCCESS;
  }
H
Hongze Cheng 已提交
793

H
Haojun Liao 已提交
794
  // todo binary search to the start position
795 796
  int64_t et1 = taosGetTimestampUs();

X
Xiaoyu Wang 已提交
797
  SBlockIdx*     pBlockIdx = NULL;
798
  STableUidList* pList = &pReader->status.uidList;
H
Haojun Liao 已提交
799

H
Haojun Liao 已提交
800
  int32_t i = 0, j = 0;
X
Xiaoyu Wang 已提交
801
  while (i < num && j < numOfTables) {
H
Haojun Liao 已提交
802
    pBlockIdx = (SBlockIdx*)taosArrayGet(aBlockIdx, i);
H
Hongze Cheng 已提交
803
    if (pBlockIdx->suid != pReader->suid) {
H
Haojun Liao 已提交
804
      i += 1;
H
Haojun Liao 已提交
805 806 807
      continue;
    }

H
Haojun Liao 已提交
808 809
    if (pBlockIdx->uid < pList->tableUidList[j]) {
      i += 1;
H
Haojun Liao 已提交
810 811 812
      continue;
    }

H
Haojun Liao 已提交
813
    if (pBlockIdx->uid > pList->tableUidList[j]) {
H
Haojun Liao 已提交
814
      j += 1;
H
Haojun Liao 已提交
815
      continue;
H
Haojun Liao 已提交
816 817
    }

H
Haojun Liao 已提交
818
    if (pBlockIdx->uid == pList->tableUidList[j]) {
H
Haojun Liao 已提交
819
      // this block belongs to a table that is not queried.
H
Haojun Liao 已提交
820 821 822
      STableBlockScanInfo* pScanInfo = getTableBlockScanInfo(pReader->status.pTableMap, pBlockIdx->uid, pReader->idStr);
      if (pScanInfo == NULL) {
        return terrno;
H
Haojun Liao 已提交
823 824 825 826 827 828 829
      }

      if (pScanInfo->pBlockList == NULL) {
        pScanInfo->pBlockList = taosArrayInit(4, sizeof(SBlockIndex));
      }

      taosArrayPush(pIndexList, pBlockIdx);
H
Haojun Liao 已提交
830

H
Haojun Liao 已提交
831
      i += 1;
H
Haojun Liao 已提交
832
      j += 1;
833
    }
H
Haojun Liao 已提交
834
  }
H
Hongze Cheng 已提交
835

836
  int64_t et2 = taosGetTimestampUs();
H
Haojun Liao 已提交
837 838 839
  tsdbDebug("load block index for %d/%d tables completed, elapsed time:%.2f ms, set blockIdx:%.2f ms, size:%.2f Kb %s",
            numOfTables, (int32_t)num, (et1 - st) / 1000.0, (et2 - et1) / 1000.0, num * sizeof(SBlockIdx) / 1024.0,
            pReader->idStr);
840 841 842

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

843
_end:
844
  tsdbBICacheRelease(pFileReader->pTsdb->biCache, handle);
H
Haojun Liao 已提交
845 846
  return code;
}
H
Hongze Cheng 已提交
847

848
static void cleanupTableScanInfo(SHashObj* pTableMap) {
849
  STableBlockScanInfo** px = NULL;
dengyihao's avatar
dengyihao 已提交
850
  while (1) {
851
    px = taosHashIterate(pTableMap, px);
852 853 854 855
    if (px == NULL) {
      break;
    }

856
    // reset the index in last block when handing a new file
857 858
    tMapDataClear(&(*px)->mapData);
    taosArrayClear((*px)->pBlockList);
859
  }
860 861
}

862
static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, SBlockNumber* pBlockNum) {
863 864 865 866 867 868
  int32_t numOfQTable = 0;
  size_t  sizeInDisk = 0;
  size_t  numOfTables = taosArrayGetSize(pIndexList);

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

dengyihao's avatar
dengyihao 已提交
870
  for (int32_t i = 0; i < numOfTables; ++i) {
X
Xiaoyu Wang 已提交
871
    SBlockIdx*           pBlockIdx = taosArrayGet(pIndexList, i);
H
Haojun Liao 已提交
872 873 874 875
    STableBlockScanInfo* pScanInfo = getTableBlockScanInfo(pReader->status.pTableMap, pBlockIdx->uid, pReader->idStr);
    if (pScanInfo == NULL) {
      return terrno;
    }
H
Hongze Cheng 已提交
876

877
    tMapDataReset(&pScanInfo->mapData);
H
Hongze Cheng 已提交
878
    tsdbReadDataBlk(pReader->pFileReader, pBlockIdx, &pScanInfo->mapData);
H
Haojun Liao 已提交
879
    taosArrayEnsureCap(pScanInfo->pBlockList, pScanInfo->mapData.nItem);
880

881
    sizeInDisk += pScanInfo->mapData.nData;
882 883 884 885 886 887 888 889 890 891 892 893 894

    int32_t     step = ASCENDING_TRAVERSE(pReader->order) ? 1 : -1;
    STimeWindow w = pReader->window;
    if (ASCENDING_TRAVERSE(pReader->order)) {
      w.skey = pScanInfo->lastKey + step;
    } else {
      w.ekey = pScanInfo->lastKey + step;
    }

    if (isEmptyQueryTimeWindow(&w)) {
      continue;
    }

H
Haojun Liao 已提交
895
    SDataBlk block = {0};
896
    for (int32_t j = 0; j < pScanInfo->mapData.nItem; ++j) {
H
Haojun Liao 已提交
897
      tGetDataBlk(pScanInfo->mapData.pData + pScanInfo->mapData.aOffset[j], &block);
H
Hongze Cheng 已提交
898

899
      // 1. time range check
900 901
      // if (block.minKey.ts > pReader->window.ekey || block.maxKey.ts < pReader->window.skey) {
      if (block.minKey.ts > w.ekey || block.maxKey.ts < w.skey) {
H
Haojun Liao 已提交
902 903
        continue;
      }
H
Hongze Cheng 已提交
904

905
      // 2. version range check
H
Hongze Cheng 已提交
906
      if (block.minVer > pReader->verRange.maxVer || block.maxVer < pReader->verRange.minVer) {
907 908
        continue;
      }
909

910
      SBlockIndex bIndex = {.ordinalIndex = j, .inFileOffset = block.aSubBlock->offset};
911
      bIndex.window = (STimeWindow){.skey = block.minKey.ts, .ekey = block.maxKey.ts};
912

H
Haojun Liao 已提交
913 914
      void* p1 = taosArrayPush(pScanInfo->pBlockList, &bIndex);
      if (p1 == NULL) {
915
        tMapDataClear(&pScanInfo->mapData);
H
Haojun Liao 已提交
916 917
        return TSDB_CODE_OUT_OF_MEMORY;
      }
918

919
      pBlockNum->numOfBlocks += 1;
H
Haojun Liao 已提交
920
    }
H
Hongze Cheng 已提交
921

H
Haojun Liao 已提交
922
    if ((pScanInfo->pBlockList != NULL )&& (taosArrayGetSize(pScanInfo->pBlockList) > 0)) {
923 924 925 926
      numOfQTable += 1;
    }
  }

H
Hongze Cheng 已提交
927
  pBlockNum->numOfLastFiles = pReader->pFileReader->pSet->nSttF;
928
  int32_t total = pBlockNum->numOfLastFiles + pBlockNum->numOfBlocks;
929

930
  double el = (taosGetTimestampUs() - st) / 1000.0;
H
Hongze Cheng 已提交
931
  tsdbDebug(
932
      "load block of %ld tables completed, blocks:%d in %d tables, last-files:%d, block-info-size:%.2f Kb, elapsed "
933
      "time:%.2f ms %s",
934
      numOfTables, pBlockNum->numOfBlocks, numOfQTable, pBlockNum->numOfLastFiles, sizeInDisk / 1000.0, el,
H
Hongze Cheng 已提交
935
      pReader->idStr);
936

937
  pReader->cost.numOfBlocks += total;
938
  pReader->cost.headFileLoadTime += el;
939

H
Haojun Liao 已提交
940 941
  return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
942

943
static void setBlockAllDumped(SFileBlockDumpInfo* pDumpInfo, int64_t maxKey, int32_t order) {
944
  int32_t step = ASCENDING_TRAVERSE(order) ? 1 : -1;
945
  pDumpInfo->allDumped = true;
946
  pDumpInfo->lastKey = maxKey + step;
H
Haojun Liao 已提交
947 948
}

949 950
static void doCopyColVal(SColumnInfoData* pColInfoData, int32_t rowIndex, int32_t colIndex, SColVal* pColVal,
                         SBlockLoadSuppInfo* pSup) {
H
Haojun Liao 已提交
951
  if (IS_VAR_DATA_TYPE(pColVal->type)) {
H
Hongze Cheng 已提交
952
    if (!COL_VAL_IS_VALUE(pColVal)) {
953
      colDataSetNULL(pColInfoData, rowIndex);
H
Haojun Liao 已提交
954 955
    } else {
      varDataSetLen(pSup->buildBuf[colIndex], pColVal->value.nData);
H
Haojun Liao 已提交
956
      ASSERT(pColVal->value.nData <= pColInfoData->info.bytes);
957 958 959 960
      if (pColVal->value.nData > 0) {  // pData may be null, if nData is 0
        memcpy(varDataVal(pSup->buildBuf[colIndex]), pColVal->value.pData, pColVal->value.nData);
      }

961
      colDataSetVal(pColInfoData, rowIndex, pSup->buildBuf[colIndex], false);
H
Haojun Liao 已提交
962 963
    }
  } else {
964
    colDataSetVal(pColInfoData, rowIndex, (const char*)&pColVal->value, !COL_VAL_IS_VALUE(pColVal));
H
Haojun Liao 已提交
965
  }
H
Haojun Liao 已提交
966 967
}

968
static SFileDataBlockInfo* getCurrentBlockInfo(SDataBlockIter* pBlockIter) {
H
Haojun Liao 已提交
969 970 971
  size_t num = taosArrayGetSize(pBlockIter->blockList);
  if (num == 0) {
    ASSERT(pBlockIter->numOfBlocks == num);
972 973
    return NULL;
  }
974 975 976

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

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

C
Cary Xu 已提交
981 982 983 984 985 986
static int doBinarySearchKey(TSKEY* keyList, int num, int pos, TSKEY key, int order) {
  // start end position
  int s, e;
  s = pos;

  // check
H
Haojun Liao 已提交
987
  ASSERT(pos >= 0 && pos < num && num > 0);
C
Cary Xu 已提交
988 989
  if (order == TSDB_ORDER_ASC) {
    // find the first position which is smaller than the key
H
Hongze Cheng 已提交
990 991
    e = num - 1;
    if (key < keyList[pos]) return -1;
C
Cary Xu 已提交
992 993
    while (1) {
      // check can return
H
Hongze Cheng 已提交
994 995 996
      if (key >= keyList[e]) return e;
      if (key <= keyList[s]) return s;
      if (e - s <= 1) return s;
C
Cary Xu 已提交
997 998

      // change start or end position
H
Hongze Cheng 已提交
999
      int mid = s + (e - s + 1) / 2;
C
Cary Xu 已提交
1000 1001
      if (keyList[mid] > key)
        e = mid;
H
Hongze Cheng 已提交
1002
      else if (keyList[mid] < key)
C
Cary Xu 已提交
1003 1004 1005 1006
        s = mid;
      else
        return mid;
    }
H
Hongze Cheng 已提交
1007
  } else {  // DESC
C
Cary Xu 已提交
1008
    // find the first position which is bigger than the key
H
Hongze Cheng 已提交
1009 1010
    e = 0;
    if (key > keyList[pos]) return -1;
C
Cary Xu 已提交
1011 1012
    while (1) {
      // check can return
H
Hongze Cheng 已提交
1013 1014 1015
      if (key <= keyList[e]) return e;
      if (key >= keyList[s]) return s;
      if (s - e <= 1) return s;
C
Cary Xu 已提交
1016 1017

      // change start or end position
H
Hongze Cheng 已提交
1018
      int mid = s - (s - e + 1) / 2;
C
Cary Xu 已提交
1019 1020
      if (keyList[mid] < key)
        e = mid;
H
Hongze Cheng 已提交
1021
      else if (keyList[mid] > key)
C
Cary Xu 已提交
1022 1023 1024 1025 1026 1027 1028
        s = mid;
      else
        return mid;
    }
  }
}

H
Haojun Liao 已提交
1029
static int32_t getEndPosInDataBlock(STsdbReader* pReader, SBlockData* pBlockData, SDataBlk* pBlock, int32_t pos) {
C
Cary Xu 已提交
1030 1031
  // NOTE: reverse the order to find the end position in data block
  int32_t endPos = -1;
H
Hongze Cheng 已提交
1032
  bool    asc = ASCENDING_TRAVERSE(pReader->order);
C
Cary Xu 已提交
1033 1034 1035 1036 1037 1038

  if (asc && pReader->window.ekey >= pBlock->maxKey.ts) {
    endPos = pBlock->nRow - 1;
  } else if (!asc && pReader->window.skey <= pBlock->minKey.ts) {
    endPos = 0;
  } else {
C
Cary Xu 已提交
1039 1040
    int64_t key = asc ? pReader->window.ekey : pReader->window.skey;
    endPos = doBinarySearchKey(pBlockData->aTSKEY, pBlock->nRow, pos, key, pReader->order);
C
Cary Xu 已提交
1041 1042 1043 1044 1045
  }

  return endPos;
}

H
Haojun Liao 已提交
1046
static void copyPrimaryTsCol(const SBlockData* pBlockData, SFileBlockDumpInfo* pDumpInfo, SColumnInfoData* pColData,
H
Haojun Liao 已提交
1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065
                             int32_t dumpedRows, bool asc) {
  if (asc) {
    memcpy(pColData->pData, &pBlockData->aTSKEY[pDumpInfo->rowIndex], dumpedRows * sizeof(int64_t));
  } else {
    int32_t startIndex = pDumpInfo->rowIndex - dumpedRows + 1;
    memcpy(pColData->pData, &pBlockData->aTSKEY[startIndex], dumpedRows * sizeof(int64_t));

    // todo: opt perf by extract the loop
    // reverse the array list
    int32_t  mid = dumpedRows >> 1u;
    int64_t* pts = (int64_t*)pColData->pData;
    for (int32_t j = 0; j < mid; ++j) {
      int64_t t = pts[j];
      pts[j] = pts[dumpedRows - j - 1];
      pts[dumpedRows - j - 1] = t;
    }
  }
}

H
Haojun Liao 已提交
1066 1067
// a faster version of copy procedure.
static void copyNumericCols(const SColData* pData, SFileBlockDumpInfo* pDumpInfo, SColumnInfoData* pColData,
H
Hongze Cheng 已提交
1068
                            int32_t dumpedRows, bool asc) {
H
Haojun Liao 已提交
1069 1070 1071 1072 1073 1074 1075 1076
  uint8_t* p = NULL;
  if (asc) {
    p = pData->pData + tDataTypes[pData->type].bytes * pDumpInfo->rowIndex;
  } else {
    int32_t startIndex = pDumpInfo->rowIndex - dumpedRows + 1;
    p = pData->pData + tDataTypes[pData->type].bytes * startIndex;
  }

H
Hongze Cheng 已提交
1077
  int32_t step = asc ? 1 : -1;
H
Haojun Liao 已提交
1078

H
Haojun Liao 已提交
1079
  // make sure it is aligned to 8bit, the allocated memory address is aligned to 256bit
1080
  //  ASSERT((((uint64_t)pColData->pData) & (0x8 - 1)) == 0);
H
Haojun Liao 已提交
1081 1082 1083 1084 1085 1086

  // 1. copy data in a batch model
  memcpy(pColData->pData, p, dumpedRows * tDataTypes[pData->type].bytes);

  // 2. reverse the array list in case of descending order scan data block
  if (!asc) {
H
Hongze Cheng 已提交
1087
    switch (pColData->info.type) {
H
Haojun Liao 已提交
1088 1089 1090
      case TSDB_DATA_TYPE_TIMESTAMP:
      case TSDB_DATA_TYPE_DOUBLE:
      case TSDB_DATA_TYPE_BIGINT:
H
Hongze Cheng 已提交
1091
      case TSDB_DATA_TYPE_UBIGINT: {
H
Haojun Liao 已提交
1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104
        int32_t  mid = dumpedRows >> 1u;
        int64_t* pts = (int64_t*)pColData->pData;
        for (int32_t j = 0; j < mid; ++j) {
          int64_t t = pts[j];
          pts[j] = pts[dumpedRows - j - 1];
          pts[dumpedRows - j - 1] = t;
        }
        break;
      }

      case TSDB_DATA_TYPE_BOOL:
      case TSDB_DATA_TYPE_TINYINT:
      case TSDB_DATA_TYPE_UTINYINT: {
H
Hongze Cheng 已提交
1105
        int32_t mid = dumpedRows >> 1u;
H
Haojun Liao 已提交
1106 1107
        int8_t* pts = (int8_t*)pColData->pData;
        for (int32_t j = 0; j < mid; ++j) {
H
Haojun Liao 已提交
1108
          int8_t t = pts[j];
H
Haojun Liao 已提交
1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132
          pts[j] = pts[dumpedRows - j - 1];
          pts[dumpedRows - j - 1] = t;
        }
        break;
      }

      case TSDB_DATA_TYPE_SMALLINT:
      case TSDB_DATA_TYPE_USMALLINT: {
        int32_t  mid = dumpedRows >> 1u;
        int16_t* pts = (int16_t*)pColData->pData;
        for (int32_t j = 0; j < mid; ++j) {
          int64_t t = pts[j];
          pts[j] = pts[dumpedRows - j - 1];
          pts[dumpedRows - j - 1] = t;
        }
        break;
      }

      case TSDB_DATA_TYPE_FLOAT:
      case TSDB_DATA_TYPE_INT:
      case TSDB_DATA_TYPE_UINT: {
        int32_t  mid = dumpedRows >> 1u;
        int32_t* pts = (int32_t*)pColData->pData;
        for (int32_t j = 0; j < mid; ++j) {
H
Haojun Liao 已提交
1133
          int32_t t = pts[j];
H
Haojun Liao 已提交
1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155
          pts[j] = pts[dumpedRows - j - 1];
          pts[dumpedRows - j - 1] = t;
        }
        break;
      }
    }
  }

  // 3. if the  null value exists, check items one-by-one
  if (pData->flag != HAS_VALUE) {
    int32_t rowIndex = 0;

    for (int32_t j = pDumpInfo->rowIndex; rowIndex < dumpedRows; j += step, rowIndex++) {
      uint8_t v = tColDataGetBitValue(pData, j);
      if (v == 0 || v == 1) {
        colDataSetNull_f(pColData->nullbitmap, rowIndex);
        pColData->hasNull = true;
      }
    }
  }
}

1156
static int32_t copyBlockDataToSDataBlock(STsdbReader* pReader) {
H
Haojun Liao 已提交
1157 1158 1159 1160
  SReaderStatus*      pStatus = &pReader->status;
  SDataBlockIter*     pBlockIter = &pStatus->blockIter;
  SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo;
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
H
Hongze Cheng 已提交
1161

1162
  SBlockData*         pBlockData = &pStatus->fileBlockData;
C
Cary Xu 已提交
1163
  SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter);
H
Hongze Cheng 已提交
1164
  SDataBlk*           pBlock = getCurrentBlock(pBlockIter);
H
Haojun Liao 已提交
1165
  SSDataBlock*        pResBlock = pReader->pResBlock;
H
Haojun Liao 已提交
1166
  int32_t             numOfOutputCols = pSupInfo->numOfCols;
H
Haojun Liao 已提交
1167

H
Haojun Liao 已提交
1168
  SColVal cv = {0};
1169
  int64_t st = taosGetTimestampUs();
1170 1171
  bool    asc = ASCENDING_TRAVERSE(pReader->order);
  int32_t step = asc ? 1 : -1;
1172

1173 1174
  // no data exists, return directly.
  if (pBlockData->nRow == 0 || pBlockData->aTSKEY == 0) {
X
Xiaoyu Wang 已提交
1175 1176
    tsdbWarn("%p no need to copy since no data in blockData, table uid:%" PRIu64 " has been dropped, %s", pReader,
             pBlockInfo->uid, pReader->idStr);
1177 1178 1179 1180
    pResBlock->info.rows = 0;
    return 0;
  }

1181 1182
  if ((pDumpInfo->rowIndex == 0 && asc) || (pDumpInfo->rowIndex == pBlock->nRow - 1 && (!asc))) {
    if (asc && pReader->window.skey <= pBlock->minKey.ts) {
1183 1184 1185
      // pDumpInfo->rowIndex = 0;
    } else if (!asc && pReader->window.ekey >= pBlock->maxKey.ts) {
      // pDumpInfo->rowIndex = pBlock->nRow - 1;
H
Haojun Liao 已提交
1186
    } else {  // find the appropriate the start position in current block, and set it to be the current rowIndex
1187
      int32_t pos = asc ? pBlock->nRow - 1 : 0;
C
Cary Xu 已提交
1188 1189 1190
      int32_t order = asc ? TSDB_ORDER_DESC : TSDB_ORDER_ASC;
      int64_t key = asc ? pReader->window.skey : pReader->window.ekey;
      pDumpInfo->rowIndex = doBinarySearchKey(pBlockData->aTSKEY, pBlock->nRow, pos, key, order);
H
Haojun Liao 已提交
1191 1192 1193 1194 1195 1196 1197 1198 1199

      if (pDumpInfo->rowIndex < 0) {
        tsdbError(
            "%p failed to locate the start position in current block, global index:%d, table index:%d, brange:%" PRId64
            "-%" PRId64 ", minVer:%" PRId64 ", maxVer:%" PRId64 " %s",
            pReader, pBlockIter->index, pBlockInfo->tbBlockIdx, pBlock->minKey.ts, pBlock->maxKey.ts, pBlock->minVer,
            pBlock->maxVer, pReader->idStr);
        return TSDB_CODE_INVALID_PARA;
      }
1200
    }
C
Cary Xu 已提交
1201 1202 1203 1204 1205 1206 1207 1208 1209 1210
  }

  // time window check
  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 已提交
1211 1212 1213
  int32_t dumpedRows = asc ? (endIndex - pDumpInfo->rowIndex) : (pDumpInfo->rowIndex - endIndex);
  if (dumpedRows > pReader->capacity) {  // output buffer check
    dumpedRows = pReader->capacity;
1214 1215
  }

H
Haojun Liao 已提交
1216
  int32_t i = 0;
C
Cary Xu 已提交
1217 1218
  int32_t rowIndex = 0;

H
Haojun Liao 已提交
1219 1220
  SColumnInfoData* pColData = taosArrayGet(pResBlock->pDataBlock, pSupInfo->slotId[i]);
  if (pSupInfo->colId[i] == PRIMARYKEY_TIMESTAMP_COL_ID) {
H
Haojun Liao 已提交
1221
    copyPrimaryTsCol(pBlockData, pDumpInfo, pColData, dumpedRows, asc);
1222 1223 1224
    i += 1;
  }

1225
  int32_t colIndex = 0;
H
Hongze Cheng 已提交
1226
  int32_t num = pBlockData->nColData;
1227
  while (i < numOfOutputCols && colIndex < num) {
1228 1229
    rowIndex = 0;

H
Hongze Cheng 已提交
1230
    SColData* pData = tBlockDataGetColDataByIdx(pBlockData, colIndex);
H
Haojun Liao 已提交
1231
    if (pData->cid < pSupInfo->colId[i]) {
1232
      colIndex += 1;
H
Haojun Liao 已提交
1233 1234
    } else if (pData->cid == pSupInfo->colId[i]) {
      pColData = taosArrayGet(pResBlock->pDataBlock, pSupInfo->slotId[i]);
H
Haojun Liao 已提交
1235

H
Hongze Cheng 已提交
1236
      if (pData->flag == HAS_NONE || pData->flag == HAS_NULL || pData->flag == (HAS_NULL | HAS_NONE)) {
1237
        colDataSetNNULL(pColData, 0, dumpedRows);
C
Cary Xu 已提交
1238
      } else {
H
Haojun Liao 已提交
1239
        if (IS_MATHABLE_TYPE(pColData->info.type)) {
H
Haojun Liao 已提交
1240 1241
          copyNumericCols(pData, pDumpInfo, pColData, dumpedRows, asc);
        } else {  // varchar/nchar type
H
Haojun Liao 已提交
1242
          for (int32_t j = pDumpInfo->rowIndex; rowIndex < dumpedRows; j += step) {
C
Cary Xu 已提交
1243 1244 1245 1246
            tColDataGetValue(pData, j, &cv);
            doCopyColVal(pColData, rowIndex++, i, &cv, pSupInfo);
          }
        }
H
Haojun Liao 已提交
1247
      }
C
Cary Xu 已提交
1248

1249
      colIndex += 1;
1250
      i += 1;
1251
    } else {  // the specified column does not exist in file block, fill with null data
H
Haojun Liao 已提交
1252
      pColData = taosArrayGet(pResBlock->pDataBlock, pSupInfo->slotId[i]);
1253
      colDataSetNNULL(pColData, 0, dumpedRows);
1254
      i += 1;
H
Haojun Liao 已提交
1255
    }
1256 1257
  }

1258
  // fill the mis-matched columns with null value
1259
  while (i < numOfOutputCols) {
H
Haojun Liao 已提交
1260
    pColData = taosArrayGet(pResBlock->pDataBlock, pSupInfo->slotId[i]);
1261
    colDataSetNNULL(pColData, 0, dumpedRows);
1262
    i += 1;
H
Haojun Liao 已提交
1263
  }
H
Haojun Liao 已提交
1264

1265
  pResBlock->info.dataLoad = 1;
H
Haojun Liao 已提交
1266 1267
  pResBlock->info.rows = dumpedRows;
  pDumpInfo->rowIndex += step * dumpedRows;
1268

1269
  // check if current block are all handled
C
Cary Xu 已提交
1270 1271
  if (pDumpInfo->rowIndex >= 0 && pDumpInfo->rowIndex < pBlock->nRow) {
    int64_t ts = pBlockData->aTSKEY[pDumpInfo->rowIndex];
1272 1273 1274
    if (outOfTimeWindow(ts, &pReader->window)) {  // the remain data has out of query time window, ignore current block
      setBlockAllDumped(pDumpInfo, ts, pReader->order);
    }
C
Cary Xu 已提交
1275
  } else {
1276 1277
    int64_t ts = asc ? pBlock->maxKey.ts : pBlock->minKey.ts;
    setBlockAllDumped(pDumpInfo, ts, pReader->order);
C
Cary Xu 已提交
1278
  }
H
Haojun Liao 已提交
1279

1280
  double elapsedTime = (taosGetTimestampUs() - st) / 1000.0;
H
Haojun Liao 已提交
1281
  pReader->cost.blockLoadTime += elapsedTime;
H
Haojun Liao 已提交
1282

1283
  int32_t unDumpedRows = asc ? pBlock->nRow - pDumpInfo->rowIndex : pDumpInfo->rowIndex + 1;
H
Haojun Liao 已提交
1284
  tsdbDebug("%p copy file block to sdatablock, global index:%d, table index:%d, brange:%" PRId64 "-%" PRId64
1285
            ", rows:%d, remain:%d, minVer:%" PRId64 ", maxVer:%" PRId64 ", uid:%" PRIu64 " elapsed time:%.2f ms, %s",
H
Haojun Liao 已提交
1286
            pReader, pBlockIter->index, pBlockInfo->tbBlockIdx, pBlock->minKey.ts, pBlock->maxKey.ts, dumpedRows,
H
Haojun Liao 已提交
1287
            unDumpedRows, pBlock->minVer, pBlock->maxVer, pBlockInfo->uid, elapsedTime, pReader->idStr);
1288 1289 1290 1291

  return TSDB_CODE_SUCCESS;
}

1292 1293
static int32_t doLoadFileBlockData(STsdbReader* pReader, SDataBlockIter* pBlockIter, SBlockData* pBlockData,
                                   uint64_t uid) {
1294
  int32_t code = 0;
1295 1296
  int64_t st = taosGetTimestampUs();

1297
  tBlockDataReset(pBlockData);
1298 1299
  STSchema* pSchema = getLatestTableSchema(pReader, uid);
  if (pSchema == NULL) {
X
Xiaoyu Wang 已提交
1300
    tsdbDebug("%p table uid:%" PRIu64 " has been dropped, no data existed, %s", pReader, uid, pReader->idStr);
1301 1302 1303 1304
    return code;
  }

  SBlockLoadSuppInfo* pSup = &pReader->suppInfo;
X
Xiaoyu Wang 已提交
1305
  TABLEID             tid = {.suid = pReader->suid, .uid = uid};
1306
  code = tBlockDataInit(pBlockData, &tid, pSchema, &pSup->colId[1], pSup->numOfCols - 1);
1307 1308 1309 1310
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

1311
  SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter);
1312
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
1313

H
Hongze Cheng 已提交
1314
  SDataBlk* pBlock = getCurrentBlock(pBlockIter);
1315
  code = tsdbReadDataBlock(pReader->pFileReader, pBlock, pBlockData);
1316 1317 1318
  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",
1319
              pReader, pBlockIter->index, pBlockInfo->tbBlockIdx, pBlock->minKey.ts, pBlock->maxKey.ts, pBlock->nRow,
1320 1321 1322
              tstrerror(code), pReader->idStr);
    return code;
  }
1323

1324
  double elapsedTime = (taosGetTimestampUs() - st) / 1000.0;
1325

1326 1327 1328 1329
  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);
1330 1331 1332

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

H
Haojun Liao 已提交
1334
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1335
}
H
Hongze Cheng 已提交
1336

H
Haojun Liao 已提交
1337 1338 1339
static void cleanupBlockOrderSupporter(SBlockOrderSupporter* pSup) {
  taosMemoryFreeClear(pSup->numOfBlocksPerTable);
  taosMemoryFreeClear(pSup->indexPerTable);
H
Hongze Cheng 已提交
1340

H
Haojun Liao 已提交
1341 1342 1343 1344
  for (int32_t i = 0; i < pSup->numOfTables; ++i) {
    SBlockOrderWrapper* pBlockInfo = pSup->pDataBlockInfo[i];
    taosMemoryFreeClear(pBlockInfo);
  }
H
Hongze Cheng 已提交
1345

H
Haojun Liao 已提交
1346 1347
  taosMemoryFreeClear(pSup->pDataBlockInfo);
}
H
Hongze Cheng 已提交
1348

H
Haojun Liao 已提交
1349 1350
static int32_t initBlockOrderSupporter(SBlockOrderSupporter* pSup, int32_t numOfTables) {
  pSup->numOfBlocksPerTable = taosMemoryCalloc(1, sizeof(int32_t) * numOfTables);
1351 1352
  pSup->indexPerTable = taosMemoryCalloc(1, sizeof(int32_t) * numOfTables);
  pSup->pDataBlockInfo = taosMemoryCalloc(1, POINTER_BYTES * numOfTables);
H
Hongze Cheng 已提交
1353

H
Haojun Liao 已提交
1354 1355 1356 1357
  if (pSup->numOfBlocksPerTable == NULL || pSup->indexPerTable == NULL || pSup->pDataBlockInfo == NULL) {
    cleanupBlockOrderSupporter(pSup);
    return TSDB_CODE_OUT_OF_MEMORY;
  }
H
Hongze Cheng 已提交
1358

H
Haojun Liao 已提交
1359 1360
  return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
1361

H
Haojun Liao 已提交
1362
static int32_t fileDataBlockOrderCompar(const void* pLeft, const void* pRight, void* param) {
1363
  int32_t leftIndex = *(int32_t*)pLeft;
H
Haojun Liao 已提交
1364
  int32_t rightIndex = *(int32_t*)pRight;
H
Hongze Cheng 已提交
1365

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

H
Haojun Liao 已提交
1368 1369
  int32_t leftTableBlockIndex = pSupporter->indexPerTable[leftIndex];
  int32_t rightTableBlockIndex = pSupporter->indexPerTable[rightIndex];
H
Hongze Cheng 已提交
1370

H
Haojun Liao 已提交
1371 1372 1373 1374 1375 1376 1377
  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 已提交
1378

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

1382 1383 1384
  return pLeftBlock->offset > pRightBlock->offset ? 1 : -1;
}

H
Haojun Liao 已提交
1385
static int32_t doSetCurrentBlock(SDataBlockIter* pBlockIter, const char* idStr) {
1386 1387
  SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter);
  if (pBlockInfo != NULL) {
H
Haojun Liao 已提交
1388
    STableBlockScanInfo* pScanInfo = getTableBlockScanInfo(pBlockIter->pTableMap, pBlockInfo->uid, idStr);
H
Haojun Liao 已提交
1389
    if (pScanInfo == NULL) {
H
Haojun Liao 已提交
1390
      return terrno;
H
Haojun Liao 已提交
1391 1392
    }

H
Haojun Liao 已提交
1393 1394
    SBlockIndex* pIndex = taosArrayGet(pScanInfo->pBlockList, pBlockInfo->tbBlockIdx);
    tMapDataGetItemByIdx(&pScanInfo->mapData, pIndex->ordinalIndex, &pBlockIter->block, tGetDataBlk);
1395
  }
1396 1397 1398 1399 1400 1401

#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 已提交
1402
}
H
Hongze Cheng 已提交
1403

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

1407
  SBlockOrderSupporter sup = {0};
1408
  pBlockIter->numOfBlocks = numOfBlocks;
1409
  taosArrayClear(pBlockIter->blockList);
1410
  pBlockIter->pTableMap = pReader->status.pTableMap;
1411

1412 1413
  // 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 已提交
1414

1415
  int64_t st = taosGetTimestampUs();
1416
  int32_t code = initBlockOrderSupporter(&sup, numOfTables);
1417 1418 1419
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }
H
Haojun Liao 已提交
1420

1421 1422 1423 1424 1425 1426 1427
  int32_t cnt = 0;
  void*   ptr = NULL;
  while (1) {
    ptr = taosHashIterate(pReader->status.pTableMap, ptr);
    if (ptr == NULL) {
      break;
    }
H
Haojun Liao 已提交
1428

1429
    STableBlockScanInfo* pTableScanInfo = *(STableBlockScanInfo**)ptr;
1430 1431 1432
    if (pTableScanInfo->pBlockList == NULL || taosArrayGetSize(pTableScanInfo->pBlockList) == 0) {
      continue;
    }
H
Haojun Liao 已提交
1433

1434 1435
    size_t num = taosArrayGetSize(pTableScanInfo->pBlockList);
    sup.numOfBlocksPerTable[sup.numOfTables] = num;
H
Haojun Liao 已提交
1436

1437 1438 1439
    char* buf = taosMemoryMalloc(sizeof(SBlockOrderWrapper) * num);
    if (buf == NULL) {
      cleanupBlockOrderSupporter(&sup);
S
Shengliang Guan 已提交
1440
      return TSDB_CODE_OUT_OF_MEMORY;
1441
    }
H
Haojun Liao 已提交
1442

1443
    sup.pDataBlockInfo[sup.numOfTables] = (SBlockOrderWrapper*)buf;
1444

1445 1446 1447
    for (int32_t k = 0; k < num; ++k) {
      SBlockIndex* pIndex = taosArrayGet(pTableScanInfo->pBlockList, k);
      sup.pDataBlockInfo[sup.numOfTables][k] =
1448
          (SBlockOrderWrapper){.uid = pTableScanInfo->uid, .offset = pIndex->inFileOffset};
1449 1450 1451 1452 1453
      cnt++;
    }

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

H
Haojun Liao 已提交
1455 1456 1457 1458
  if (numOfBlocks != cnt && sup.numOfTables != numOfTables) {
    cleanupBlockOrderSupporter(&sup);
    return TSDB_CODE_INVALID_PARA;
  }
H
Haojun Liao 已提交
1459

1460
  // since there is only one table qualified, blocks are not sorted
1461 1462
  if (sup.numOfTables == 1) {
    for (int32_t i = 0; i < numOfBlocks; ++i) {
1463 1464
      SFileDataBlockInfo blockInfo = {.uid = sup.pDataBlockInfo[0][i].uid, .tbBlockIdx = i};
      taosArrayPush(pBlockIter->blockList, &blockInfo);
1465
    }
1466

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

1471
    pBlockIter->index = asc ? 0 : (numOfBlocks - 1);
H
Haojun Liao 已提交
1472
    cleanupBlockOrderSupporter(&sup);
H
Haojun Liao 已提交
1473
    doSetCurrentBlock(pBlockIter, pReader->idStr);
1474
    return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1475
  }
H
Haojun Liao 已提交
1476

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

1480
  SMultiwayMergeTreeInfo* pTree = NULL;
H
Haojun Liao 已提交
1481 1482

  uint8_t ret = tMergeTreeCreate(&pTree, sup.numOfTables, &sup, fileDataBlockOrderCompar);
1483 1484
  if (ret != TSDB_CODE_SUCCESS) {
    cleanupBlockOrderSupporter(&sup);
S
Shengliang Guan 已提交
1485
    return TSDB_CODE_OUT_OF_MEMORY;
H
Haojun Liao 已提交
1486
  }
H
Haojun Liao 已提交
1487

1488 1489 1490 1491
  int32_t numOfTotal = 0;
  while (numOfTotal < cnt) {
    int32_t pos = tMergeTreeGetChosenIndex(pTree);
    int32_t index = sup.indexPerTable[pos]++;
H
Haojun Liao 已提交
1492

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

1496 1497 1498 1499
    // 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 已提交
1500

1501 1502
    numOfTotal += 1;
    tMergeTreeAdjust(pTree, tMergeTreeGetAdjustIndex(pTree));
H
Haojun Liao 已提交
1503
  }
H
Haojun Liao 已提交
1504

1505
  int64_t et = taosGetTimestampUs();
H
Hongze Cheng 已提交
1506 1507
  tsdbDebug("%p %d data blocks access order completed, elapsed time:%.2f ms %s", pReader, numOfBlocks,
            (et - st) / 1000.0, pReader->idStr);
1508 1509
  cleanupBlockOrderSupporter(&sup);
  taosMemoryFree(pTree);
H
Haojun Liao 已提交
1510

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

1514
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1515
}
H
Hongze Cheng 已提交
1516

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

1520
  int32_t step = asc ? 1 : -1;
1521
  if ((pBlockIter->index >= pBlockIter->numOfBlocks - 1 && asc) || (pBlockIter->index <= 0 && (!asc))) {
1522 1523 1524
    return false;
  }

1525
  pBlockIter->index += step;
H
Haojun Liao 已提交
1526
  doSetCurrentBlock(pBlockIter, idStr);
1527

1528 1529 1530
  return true;
}

1531 1532 1533
/**
 * This is an two rectangles overlap cases.
 */
H
Hongze Cheng 已提交
1534
static int32_t dataBlockPartiallyRequired(STimeWindow* pWindow, SVersionRange* pVerRange, SDataBlk* pBlock) {
1535 1536
  return (pWindow->ekey < pBlock->maxKey.ts && pWindow->ekey >= pBlock->minKey.ts) ||
         (pWindow->skey > pBlock->minKey.ts && pWindow->skey <= pBlock->maxKey.ts) ||
H
Hongze Cheng 已提交
1537 1538
         (pVerRange->minVer > pBlock->minVer && pVerRange->minVer <= pBlock->maxVer) ||
         (pVerRange->maxVer < pBlock->maxVer && pVerRange->maxVer >= pBlock->minVer);
H
Haojun Liao 已提交
1539
}
H
Hongze Cheng 已提交
1540

1541
static bool getNeighborBlockOfSameTable(SFileDataBlockInfo* pBlockInfo, STableBlockScanInfo* pTableBlockScanInfo,
1542
                                        int32_t* nextIndex, int32_t order, SBlockIndex* pBlockIndex) {
1543
  bool asc = ASCENDING_TRAVERSE(order);
H
Haojun Liao 已提交
1544
  if (asc && pBlockInfo->tbBlockIdx >= taosArrayGetSize(pTableBlockScanInfo->pBlockList) - 1) {
1545
    return false;
1546 1547
  }

H
Haojun Liao 已提交
1548
  if (!asc && pBlockInfo->tbBlockIdx == 0) {
1549
    return false;
1550 1551
  }

1552
  int32_t step = asc ? 1 : -1;
H
Haojun Liao 已提交
1553
  *nextIndex = pBlockInfo->tbBlockIdx + step;
1554 1555
  *pBlockIndex = *(SBlockIndex*)taosArrayGet(pTableBlockScanInfo->pBlockList, *nextIndex);
  //  tMapDataGetItemByIdx(&pTableBlockScanInfo->mapData, pIndex->ordinalIndex, pBlock, tGetDataBlk);
1556
  return true;
1557 1558 1559
}

static int32_t findFileBlockInfoIndex(SDataBlockIter* pBlockIter, SFileDataBlockInfo* pFBlockInfo) {
1560
  int32_t step = ASCENDING_TRAVERSE(pBlockIter->order) ? 1 : -1;
1561 1562
  int32_t index = pBlockIter->index;

1563
  while (index < pBlockIter->numOfBlocks && index >= 0) {
1564 1565 1566 1567 1568 1569 1570 1571 1572 1573 1574
    SFileDataBlockInfo* pFBlock = taosArrayGet(pBlockIter->blockList, index);
    if (pFBlock->uid == pFBlockInfo->uid && pFBlock->tbBlockIdx == pFBlockInfo->tbBlockIdx) {
      return index;
    }

    index += step;
  }

  return -1;
}

1575
static int32_t setFileBlockActiveInBlockIter(SDataBlockIter* pBlockIter, int32_t index, int32_t step) {
1576
  if (index < 0 || index >= pBlockIter->numOfBlocks) {
1577 1578 1579 1580
    return -1;
  }

  SFileDataBlockInfo fblock = *(SFileDataBlockInfo*)taosArrayGet(pBlockIter->blockList, index);
1581 1582 1583 1584 1585
  pBlockIter->index += step;

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

1587 1588 1589
    SFileDataBlockInfo* pBlockInfo = taosArrayGet(pBlockIter->blockList, pBlockIter->index);
    ASSERT(pBlockInfo->uid == fblock.uid && pBlockInfo->tbBlockIdx == fblock.tbBlockIdx);
  }
1590

H
Haojun Liao 已提交
1591
  doSetCurrentBlock(pBlockIter, "");
1592 1593 1594
  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
1595
// todo: this attribute could be acquired during extractin the global ordered block list.
1596
static bool overlapWithNeighborBlock(SDataBlk* pBlock, SBlockIndex* pNeighborBlockIndex, int32_t order) {
1597 1598
  // it is the last block in current file, no chance to overlap with neighbor blocks.
  if (ASCENDING_TRAVERSE(order)) {
1599
    return pBlock->maxKey.ts == pNeighborBlockIndex->window.skey;
1600
  } else {
1601
    return pBlock->minKey.ts == pNeighborBlockIndex->window.ekey;
1602
  }
H
Haojun Liao 已提交
1603
}
H
Hongze Cheng 已提交
1604

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

1608
  return (ascScan && (key.ts != TSKEY_INITIAL_VAL && key.ts <= pBlock->minKey.ts)) ||
1609
         (!ascScan && (key.ts != TSKEY_INITIAL_VAL && key.ts >= pBlock->maxKey.ts));
H
Haojun Liao 已提交
1610
}
H
Hongze Cheng 已提交
1611

H
Hongze Cheng 已提交
1612
static bool keyOverlapFileBlock(TSDBKEY key, SDataBlk* pBlock, SVersionRange* pVerRange) {
H
Hongze Cheng 已提交
1613 1614
  return (key.ts >= pBlock->minKey.ts && key.ts <= pBlock->maxKey.ts) && (pBlock->maxVer >= pVerRange->minVer) &&
         (pBlock->minVer <= pVerRange->maxVer);
H
Haojun Liao 已提交
1615 1616
}

H
Hongze Cheng 已提交
1617 1618
static bool doCheckforDatablockOverlap(STableBlockScanInfo* pBlockScanInfo, const SDataBlk* pBlock,
                                       int32_t startIndex) {
1619 1620
  size_t num = taosArrayGetSize(pBlockScanInfo->delSkyline);

1621
  for (int32_t i = startIndex; i < num; i += 1) {
1622 1623
    TSDBKEY* p = taosArrayGet(pBlockScanInfo->delSkyline, i);
    if (p->ts >= pBlock->minKey.ts && p->ts <= pBlock->maxKey.ts) {
H
Hongze Cheng 已提交
1624
      if (p->version >= pBlock->minVer) {
1625 1626 1627
        return true;
      }
    } else if (p->ts < pBlock->minKey.ts) {  // p->ts < pBlock->minKey.ts
H
Hongze Cheng 已提交
1628
      if (p->version >= pBlock->minVer) {
1629 1630
        if (i < num - 1) {
          TSDBKEY* pnext = taosArrayGet(pBlockScanInfo->delSkyline, i + 1);
H
Hongze Cheng 已提交
1631 1632
          if (pnext->ts >= pBlock->minKey.ts) {
            return true;
1633 1634 1635 1636 1637 1638 1639 1640 1641 1642 1643 1644 1645
          }
        } else {  // it must be the last point
          ASSERT(p->version == 0);
        }
      }
    } else {  // (p->ts > pBlock->maxKey.ts) {
      return false;
    }
  }

  return false;
}

H
Hongze Cheng 已提交
1646
static bool overlapWithDelSkyline(STableBlockScanInfo* pBlockScanInfo, const SDataBlk* pBlock, int32_t order) {
1647 1648 1649 1650
  if (pBlockScanInfo->delSkyline == NULL) {
    return false;
  }

1651
  // ts is not overlap
1652
  TSDBKEY* pFirst = taosArrayGet(pBlockScanInfo->delSkyline, 0);
L
Liu Jicong 已提交
1653
  TSDBKEY* pLast = taosArrayGetLast(pBlockScanInfo->delSkyline);
1654 1655 1656 1657 1658
  if (pBlock->minKey.ts > pLast->ts || pBlock->maxKey.ts < pFirst->ts) {
    return false;
  }

  // version is not overlap
1659
  if (ASCENDING_TRAVERSE(order)) {
1660
    return doCheckforDatablockOverlap(pBlockScanInfo, pBlock, pBlockScanInfo->fileDelIndex);
1661 1662
  } else {
    int32_t index = pBlockScanInfo->fileDelIndex;
1663
    while (1) {
1664 1665 1666 1667 1668
      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;
1669 1670 1671
      }
    }

1672
    return doCheckforDatablockOverlap(pBlockScanInfo, pBlock, index);
1673
  }
1674 1675
}

C
Cary Xu 已提交
1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 1686 1687 1688
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) {
1689 1690
  int32_t     neighborIndex = 0;
  SBlockIndex bIndex = {0};
1691

1692
  bool hasNeighbor = getNeighborBlockOfSameTable(pBlockInfo, pScanInfo, &neighborIndex, pReader->order, &bIndex);
1693

1694
  // overlap with neighbor
1695
  if (hasNeighbor) {
1696
    pInfo->overlapWithNeighborBlock = overlapWithNeighborBlock(pBlock, &bIndex, pReader->order);
1697 1698
  }

1699
  // has duplicated ts of different version in this block
C
Cary Xu 已提交
1700 1701
  pInfo->hasDupTs = (pBlock->nSubBlock == 1) ? pBlock->hasDup : true;
  pInfo->overlapWithDelInfo = overlapWithDelSkyline(pScanInfo, pBlock, pReader->order);
1702

1703 1704 1705
  if (hasDataInLastBlock(pLastBlockReader)) {
    int64_t tsLast = getCurrentKeyInLastBlock(pLastBlockReader);
    pInfo->overlapWithLastBlock = !(pBlock->maxKey.ts < tsLast || pBlock->minKey.ts > tsLast);
1706 1707
  }

C
Cary Xu 已提交
1708 1709 1710 1711 1712 1713 1714 1715 1716 1717 1718 1719 1720 1721 1722
  pInfo->moreThanCapcity = pBlock->nRow > pReader->capacity;
  pInfo->partiallyRequired = dataBlockPartiallyRequired(&pReader->window, &pReader->verRange, pBlock);
  pInfo->overlapWithKeyInBuf = keyOverlapFileBlock(keyInBuf, pBlock, &pReader->verRange);
}

// 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);
1723

C
Cary Xu 已提交
1724 1725 1726
  bool loadDataBlock =
      (info.overlapWithNeighborBlock || info.hasDupTs || info.partiallyRequired || info.overlapWithKeyInBuf ||
       info.moreThanCapcity || info.overlapWithDelInfo || info.overlapWithLastBlock);
1727 1728 1729 1730

  // log the reason why load the datablock for profile
  if (loadDataBlock) {
    tsdbDebug("%p uid:%" PRIu64
X
Xiaoyu Wang 已提交
1731
              " need to load the datablock, overlapneighbor:%d, hasDup:%d, partiallyRequired:%d, "
1732
              "overlapWithKey:%d, greaterThanBuf:%d, overlapWithDel:%d, overlapWithlastBlock:%d, %s",
C
Cary Xu 已提交
1733 1734 1735
              pReader, pBlockInfo->uid, info.overlapWithNeighborBlock, info.hasDupTs, info.partiallyRequired,
              info.overlapWithKeyInBuf, info.moreThanCapcity, info.overlapWithDelInfo, info.overlapWithLastBlock,
              pReader->idStr);
1736 1737 1738
  }

  return loadDataBlock;
H
Haojun Liao 已提交
1739 1740
}

C
Cary Xu 已提交
1741 1742 1743 1744 1745 1746 1747 1748 1749
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;
}

1750
static int32_t buildDataBlockFromBuf(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo, int64_t endKey) {
1751
  if (!(pBlockScanInfo->iiter.hasVal || pBlockScanInfo->iter.hasVal)) {
1752 1753
    return TSDB_CODE_SUCCESS;
  }
H
Haojun Liao 已提交
1754

1755 1756 1757
  SSDataBlock* pBlock = pReader->pResBlock;

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

H
Haojun Liao 已提交
1760
  blockDataUpdateTsWindow(pBlock, pReader->suppInfo.slotId[0]);
H
Haojun Liao 已提交
1761
  pBlock->info.id.uid = pBlockScanInfo->uid;
1762

1763
  setComposedBlockFlag(pReader, true);
1764

1765
  double elapsedTime = (taosGetTimestampUs() - st) / 1000.0;
S
Shengliang Guan 已提交
1766
  tsdbDebug("%p build data block from cache completed, elapsed time:%.2f ms, numOfRows:%d, brange:%" PRId64
X
Xiaoyu Wang 已提交
1767
            " - %" PRId64 ", uid:%" PRIu64 ",  %s",
1768
            pReader, elapsedTime, pBlock->info.rows, pBlock->info.window.skey, pBlock->info.window.ekey,
1769
            pBlockScanInfo->uid, pReader->idStr);
1770 1771

  pReader->cost.buildmemBlock += elapsedTime;
H
Haojun Liao 已提交
1772 1773 1774
  return code;
}

1775 1776
static bool tryCopyDistinctRowFromFileBlock(STsdbReader* pReader, SBlockData* pBlockData, int64_t key,
                                            SFileBlockDumpInfo* pDumpInfo) {
1777 1778 1779
  // opt version
  // 1. it is not a border point
  // 2. the direct next point is not an duplicated timestamp
1780 1781
  bool asc = (pReader->order == TSDB_ORDER_ASC);
  if ((pDumpInfo->rowIndex < pDumpInfo->totalRows - 1 && asc) || (pDumpInfo->rowIndex > 0 && (!asc))) {
1782
    int32_t step = pReader->order == TSDB_ORDER_ASC ? 1 : -1;
1783 1784

    int64_t nextKey = pBlockData->aTSKEY[pDumpInfo->rowIndex + step];
1785
    if (nextKey != key) {  // merge is not needed
1786
      doAppendRowFromFileBlock(pReader->pResBlock, pReader, pBlockData, pDumpInfo->rowIndex);
1787 1788 1789 1790 1791 1792 1793 1794
      pDumpInfo->rowIndex += step;
      return true;
    }
  }

  return false;
}

1795
static bool nextRowFromLastBlocks(SLastBlockReader* pLastBlockReader, STableBlockScanInfo* pScanInfo,
1796
                                  SVersionRange* pVerRange) {
X
Xiaoyu Wang 已提交
1797
  int32_t step = ASCENDING_TRAVERSE(pLastBlockReader->order) ? 1 : -1;
H
Haojun Liao 已提交
1798

1799 1800
  while (1) {
    bool hasVal = tMergeTreeNext(&pLastBlockReader->mergeTree);
1801 1802
    if (!hasVal) { // the next value will be the accessed key in stt
      pScanInfo->lastKeyInStt += step;
1803 1804 1805 1806 1807
      return false;
    }

    TSDBROW row = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
    TSDBKEY k = TSDBROW_KEY(&row);
1808 1809 1810
    pScanInfo->lastKeyInStt = k.ts;

    if (!hasBeenDropped(pScanInfo->delSkyline, &pScanInfo->lastBlockDelIndex, &k, pLastBlockReader->order, pVerRange)) {
H
Haojun Liao 已提交
1811 1812
      // the qualifed ts may equal to k.ts, only a greater version one.
      // here we need to fallback one step.
1813 1814 1815 1816 1817 1818 1819
      return true;
    }
  }
}

static bool tryCopyDistinctRowFromSttBlock(TSDBROW* fRow, SLastBlockReader* pLastBlockReader,
                                           STableBlockScanInfo* pScanInfo, int64_t ts, STsdbReader* pReader) {
1820
  bool hasVal = nextRowFromLastBlocks(pLastBlockReader, pScanInfo, &pReader->verRange);
1821 1822 1823 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834
  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;
}

1835 1836 1837 1838 1839 1840
static FORCE_INLINE STSchema* getLatestTableSchema(STsdbReader* pReader, uint64_t uid) {
  if (pReader->pSchema != NULL) {
    return pReader->pSchema;
  }

  pReader->pSchema = metaGetTbTSchema(pReader->pTsdb->pVnode->pMeta, uid, -1, 1);
X
Xiaoyu Wang 已提交
1841
  if (pReader->pSchema == NULL) {
1842 1843 1844
    tsdbError("failed to get table schema, uid:%" PRIu64 ", it may have been dropped, ver:-1, %s", uid, pReader->idStr);
  }

X
Xiaoyu Wang 已提交
1845
  return pReader->pSchema;
1846 1847
}

H
Haojun Liao 已提交
1848 1849 1850
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) {
1851
    pReader->pSchema = metaGetTbTSchema(pReader->pTsdb->pVnode->pMeta, uid, -1, 1);
H
Haojun Liao 已提交
1852 1853
  }

1854
  if (pReader->pSchema && sversion == pReader->pSchema->version) {
H
Haojun Liao 已提交
1855 1856 1857 1858 1859 1860
    return pReader->pSchema;
  }

  if (pReader->pMemSchema == NULL) {
    int32_t code =
        metaGetTbTSchemaEx(pReader->pTsdb->pVnode->pMeta, pReader->suid, uid, sversion, &pReader->pMemSchema);
H
Haojun Liao 已提交
1861 1862 1863 1864 1865 1866
    if (code != TSDB_CODE_SUCCESS) {
      terrno = code;
      return NULL;
    } else {
      return pReader->pMemSchema;
    }
H
Haojun Liao 已提交
1867 1868 1869 1870 1871 1872
  }

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

H
Haojun Liao 已提交
1873
  taosMemoryFreeClear(pReader->pMemSchema);
H
Haojun Liao 已提交
1874
  int32_t code = metaGetTbTSchemaEx(pReader->pTsdb->pVnode->pMeta, pReader->suid, uid, sversion, &pReader->pMemSchema);
H
Haojun Liao 已提交
1875
  if (code != TSDB_CODE_SUCCESS || pReader->pMemSchema == NULL) {
H
Haojun Liao 已提交
1876 1877
    terrno = code;
    return NULL;
H
Haojun Liao 已提交
1878 1879
  } else {
    return pReader->pMemSchema;
H
Haojun Liao 已提交
1880
  }
H
Haojun Liao 已提交
1881 1882
}

1883
static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo, TSDBROW* pRow,
1884 1885
                                     SIterInfo* pIter, int64_t key, SLastBlockReader* pLastBlockReader) {
  SRowMerger          merge = {0};
H
Hongze Cheng 已提交
1886
  SRow*               pTSRow = NULL;
1887 1888 1889
  SBlockData*         pBlockData = &pReader->status.fileBlockData;
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;

1890
  int64_t tsLast = INT64_MIN;
1891
  if (hasDataInLastBlock(pLastBlockReader)) {
1892 1893
    tsLast = getCurrentKeyInLastBlock(pLastBlockReader);
  }
1894

H
Hongze Cheng 已提交
1895 1896
  TSDBKEY k = TSDBROW_KEY(pRow);
  TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
1897

1898 1899
  int64_t minKey = 0;
  if (pReader->order == TSDB_ORDER_ASC) {
H
Hongze Cheng 已提交
1900
    minKey = INT64_MAX;  // chosen the minimum value
1901
    if (minKey > tsLast && hasDataInLastBlock(pLastBlockReader)) {
1902 1903
      minKey = tsLast;
    }
1904

1905 1906 1907
    if (minKey > k.ts) {
      minKey = k.ts;
    }
1908

1909
    if (minKey > key && hasDataInFileBlock(pBlockData, pDumpInfo)) {
1910 1911 1912 1913
      minKey = key;
    }
  } else {
    minKey = INT64_MIN;
1914
    if (minKey < tsLast && hasDataInLastBlock(pLastBlockReader)) {
1915 1916 1917 1918 1919 1920 1921
      minKey = tsLast;
    }

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

1922
    if (minKey < key && hasDataInFileBlock(pBlockData, pDumpInfo)) {
1923 1924
      minKey = key;
    }
1925 1926 1927 1928
  }

  bool init = false;

1929
  // ASC: file block ---> last block -----> imem -----> mem
H
Hongze Cheng 已提交
1930
  // DESC: mem -----> imem -----> last block -----> file block
1931 1932
  if (pReader->order == TSDB_ORDER_ASC) {
    if (minKey == key) {
1933
      init = true;
H
Hongze Cheng 已提交
1934
      int32_t code = tsdbRowMergerInit(&merge, &fRow, pReader->pSchema);
H
Haojun Liao 已提交
1935 1936 1937
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
1938
      doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
1939 1940
    }

1941
    if (minKey == tsLast) {
1942
      TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
H
Haojun Liao 已提交
1943
      if (init) {
H
Hongze Cheng 已提交
1944
        tsdbRowMerge(&merge, &fRow1);
H
Haojun Liao 已提交
1945
      } else {
1946
        init = true;
H
Hongze Cheng 已提交
1947
        int32_t code = tsdbRowMergerInit(&merge, &fRow1, pReader->pSchema);
H
Haojun Liao 已提交
1948 1949 1950
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
1951
      }
1952
      doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, &merge, &pReader->verRange, pReader->idStr);
1953
    }
1954

1955
    if (minKey == k.ts) {
K
kailixu 已提交
1956 1957 1958 1959
      STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(pRow), pReader, pBlockScanInfo->uid);
      if (pSchema == NULL) {
        return terrno;
      }
H
Haojun Liao 已提交
1960
      if (init) {
X
Xiaoyu Wang 已提交
1961
        tsdbRowMergerAdd(&merge, pRow, pSchema);
H
Haojun Liao 已提交
1962
      } else {
1963
        init = true;
X
Xiaoyu Wang 已提交
1964
        int32_t code = tsdbRowMergerInit(&merge, pRow, pSchema);
H
Haojun Liao 已提交
1965 1966 1967 1968 1969 1970 1971
        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;
1972 1973 1974 1975 1976
      }
    }
  } else {
    if (minKey == k.ts) {
      init = true;
1977
      STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(pRow), pReader, pBlockScanInfo->uid);
H
Hongze Cheng 已提交
1978
      int32_t   code = tsdbRowMergerInit(&merge, pRow, pSchema);
H
Haojun Liao 已提交
1979 1980 1981 1982 1983
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }

      code = doMergeRowsInBuf(pIter, pBlockScanInfo->uid, k.ts, pBlockScanInfo->delSkyline, &merge, pReader);
1984
      if (code != TSDB_CODE_SUCCESS || merge.pTSchema == NULL) {
H
Haojun Liao 已提交
1985 1986
        return code;
      }
1987 1988
    }

1989
    if (minKey == tsLast) {
1990
      TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
H
Haojun Liao 已提交
1991
      if (init) {
H
Hongze Cheng 已提交
1992
        tsdbRowMerge(&merge, &fRow1);
H
Haojun Liao 已提交
1993
      } else {
1994
        init = true;
H
Hongze Cheng 已提交
1995
        int32_t code = tsdbRowMergerInit(&merge, &fRow1, pReader->pSchema);
H
Haojun Liao 已提交
1996 1997 1998
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
1999
      }
2000
      doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, &merge, &pReader->verRange, pReader->idStr);
2001 2002 2003
    }

    if (minKey == key) {
H
Haojun Liao 已提交
2004
      if (init) {
H
Hongze Cheng 已提交
2005
        tsdbRowMerge(&merge, &fRow);
H
Haojun Liao 已提交
2006
      } else {
2007
        init = true;
H
Hongze Cheng 已提交
2008
        int32_t code = tsdbRowMergerInit(&merge, &fRow, pReader->pSchema);
H
Haojun Liao 已提交
2009 2010 2011
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
2012 2013 2014
      }
      doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
    }
2015 2016
  }

H
Hongze Cheng 已提交
2017
  int32_t code = tsdbRowMergerGetRow(&merge, &pTSRow);
2018 2019 2020 2021
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

2022
  doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo);
2023 2024

  taosMemoryFree(pTSRow);
H
Hongze Cheng 已提交
2025
  tsdbRowMergerClear(&merge);
2026 2027 2028
  return TSDB_CODE_SUCCESS;
}

2029 2030 2031
static int32_t doMergeFileBlockAndLastBlock(SLastBlockReader* pLastBlockReader, STsdbReader* pReader,
                                            STableBlockScanInfo* pBlockScanInfo, SBlockData* pBlockData,
                                            bool mergeBlockData) {
2032
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
H
Hongze Cheng 已提交
2033
  int64_t             tsLastBlock = getCurrentKeyInLastBlock(pLastBlockReader);
2034

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

2040 2041 2042
  // only last block exists
  if ((!mergeBlockData) || (tsLastBlock != pBlockData->aTSKEY[pDumpInfo->rowIndex])) {
    if (tryCopyDistinctRowFromSttBlock(&fRow, pLastBlockReader, pBlockScanInfo, tsLastBlock, pReader)) {
2043
      pBlockScanInfo->lastKey = tsLastBlock;
2044 2045
      return TSDB_CODE_SUCCESS;
    } else {
H
Hongze Cheng 已提交
2046
      int32_t code = tsdbRowMergerInit(&merge, &fRow, pReader->pSchema);
H
Haojun Liao 已提交
2047 2048 2049
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
2050

2051
      TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
H
Hongze Cheng 已提交
2052
      tsdbRowMerge(&merge, &fRow1);
2053
      doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLastBlock, &merge, &pReader->verRange, pReader->idStr);
2054

H
Hongze Cheng 已提交
2055
      code = tsdbRowMergerGetRow(&merge, &pTSRow);
2056 2057 2058
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
2059

2060
      doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo);
2061 2062

      taosMemoryFree(pTSRow);
H
Hongze Cheng 已提交
2063
      tsdbRowMergerClear(&merge);
2064 2065
    }
  } else {  // not merge block data
H
Hongze Cheng 已提交
2066
    int32_t code = tsdbRowMergerInit(&merge, &fRow, pReader->pSchema);
H
Haojun Liao 已提交
2067 2068 2069 2070
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

2071
    doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLastBlock, &merge, &pReader->verRange, pReader->idStr);
2072 2073

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

H
Hongze Cheng 已提交
2078
    code = tsdbRowMergerGetRow(&merge, &pTSRow);
2079 2080 2081 2082
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

2083
    doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo);
2084 2085

    taosMemoryFree(pTSRow);
H
Hongze Cheng 已提交
2086
    tsdbRowMergerClear(&merge);
2087
  }
2088 2089 2090 2091

  return TSDB_CODE_SUCCESS;
}

2092 2093
static int32_t mergeFileBlockAndLastBlock(STsdbReader* pReader, SLastBlockReader* pLastBlockReader, int64_t key,
                                          STableBlockScanInfo* pBlockScanInfo, SBlockData* pBlockData) {
2094 2095
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;

2096
  if (hasDataInFileBlock(pBlockData, pDumpInfo)) {
2097
    // no last block available, only data block exists
2098
    if (!hasDataInLastBlock(pLastBlockReader)) {
2099 2100 2101 2102 2103 2104 2105 2106 2107
      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)) {
2108
      if (key < ts) {  // imem, mem are all empty, file blocks (data blocks and last block) exist
2109 2110
        return mergeRowsInFileBlocks(pBlockData, pBlockScanInfo, key, pReader);
      } else if (key == ts) {
H
Hongze Cheng 已提交
2111
        SRow*      pTSRow = NULL;
2112
        SRowMerger merge = {0};
2113

H
Hongze Cheng 已提交
2114
        int32_t code = tsdbRowMergerInit(&merge, &fRow, pReader->pSchema);
H
Haojun Liao 已提交
2115 2116 2117 2118
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }

2119
        doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
2120 2121

        TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
H
Hongze Cheng 已提交
2122
        tsdbRowMerge(&merge, &fRow1);
2123

2124
        doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, ts, &merge, &pReader->verRange, pReader->idStr);
2125

H
Hongze Cheng 已提交
2126
        code = tsdbRowMergerGetRow(&merge, &pTSRow);
2127 2128 2129 2130
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }

2131
        doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo);
2132

2133
        taosMemoryFree(pTSRow);
H
Hongze Cheng 已提交
2134
        tsdbRowMergerClear(&merge);
2135
        return code;
2136
      } else {
2137
        return TSDB_CODE_SUCCESS;
2138
      }
2139
    } else {  // desc order
2140
      return doMergeFileBlockAndLastBlock(pLastBlockReader, pReader, pBlockScanInfo, pBlockData, true);
2141
    }
2142
  } else {  // only last block exists
2143
    return doMergeFileBlockAndLastBlock(pLastBlockReader, pReader, pBlockScanInfo, NULL, false);
H
Haojun Liao 已提交
2144
  }
2145 2146
}

2147 2148
static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo, SBlockData* pBlockData,
                                     SLastBlockReader* pLastBlockReader) {
H
Haojun Liao 已提交
2149
  SRowMerger          merge = {0};
H
Hongze Cheng 已提交
2150
  SRow*               pTSRow = NULL;
H
Haojun Liao 已提交
2151
  int32_t             code = TSDB_CODE_SUCCESS;
2152 2153 2154
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
  SArray*             pDelList = pBlockScanInfo->delSkyline;

2155 2156
  TSDBROW* pRow = getValidMemRow(&pBlockScanInfo->iter, pDelList, pReader);
  TSDBROW* piRow = getValidMemRow(&pBlockScanInfo->iiter, pDelList, pReader);
2157

2158
  int64_t tsLast = INT64_MIN;
2159 2160 2161
  if (hasDataInLastBlock(pLastBlockReader)) {
    tsLast = getCurrentKeyInLastBlock(pLastBlockReader);
  }
2162

H
Hongze Cheng 已提交
2163
  int64_t key = hasDataInFileBlock(pBlockData, pDumpInfo) ? pBlockData->aTSKEY[pDumpInfo->rowIndex] : INT64_MIN;
2164 2165 2166 2167

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

2168
  int64_t minKey = 0;
2169 2170 2171 2172 2173
  if (ASCENDING_TRAVERSE(pReader->order)) {
    minKey = INT64_MAX;  // let's find the minimum
    if (minKey > k.ts) {
      minKey = k.ts;
    }
2174

2175 2176 2177
    if (minKey > ik.ts) {
      minKey = ik.ts;
    }
2178

2179
    if (minKey > key && hasDataInFileBlock(pBlockData, pDumpInfo)) {
2180 2181
      minKey = key;
    }
2182

2183
    if (minKey > tsLast && hasDataInLastBlock(pLastBlockReader)) {
2184 2185 2186
      minKey = tsLast;
    }
  } else {
H
Hongze Cheng 已提交
2187
    minKey = INT64_MIN;  // let find the maximum ts value
2188 2189 2190 2191 2192 2193 2194 2195
    if (minKey < k.ts) {
      minKey = k.ts;
    }

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

2196
    if (minKey < key && hasDataInFileBlock(pBlockData, pDumpInfo)) {
2197 2198 2199
      minKey = key;
    }

2200
    if (minKey < tsLast && hasDataInLastBlock(pLastBlockReader)) {
2201 2202
      minKey = tsLast;
    }
2203 2204 2205 2206
  }

  bool init = false;

2207 2208 2209 2210
  // ASC: file block -----> last block -----> imem -----> mem
  // DESC: mem -----> imem -----> last block -----> file block
  if (ASCENDING_TRAVERSE(pReader->order)) {
    if (minKey == key) {
2211
      init = true;
2212
      TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
H
Hongze Cheng 已提交
2213
      code = tsdbRowMergerInit(&merge, &fRow, pReader->pSchema);
H
Haojun Liao 已提交
2214 2215 2216 2217
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }

2218
      doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
2219 2220
    }

2221
    if (minKey == tsLast) {
2222
      TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
H
Haojun Liao 已提交
2223
      if (init) {
H
Hongze Cheng 已提交
2224
        tsdbRowMerge(&merge, &fRow1);
H
Haojun Liao 已提交
2225
      } else {
2226
        init = true;
H
Hongze Cheng 已提交
2227
        code = tsdbRowMergerInit(&merge, &fRow1, pReader->pSchema);
H
Haojun Liao 已提交
2228 2229 2230
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
2231
      }
H
Haojun Liao 已提交
2232

2233
      doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, &merge, &pReader->verRange, pReader->idStr);
2234 2235 2236
    }

    if (minKey == ik.ts) {
H
Haojun Liao 已提交
2237
      if (init) {
H
Hongze Cheng 已提交
2238
        tsdbRowMerge(&merge, piRow);
H
Haojun Liao 已提交
2239
      } else {
2240 2241
        init = true;
        STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(piRow), pReader, pBlockScanInfo->uid);
H
Haojun Liao 已提交
2242 2243 2244 2245
        if (pSchema == NULL) {
          return code;
        }

H
Hongze Cheng 已提交
2246
        code = tsdbRowMergerInit(&merge, piRow, pSchema);
H
Haojun Liao 已提交
2247 2248 2249
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
2250
      }
H
Haojun Liao 已提交
2251

2252 2253
      code = doMergeRowsInBuf(&pBlockScanInfo->iiter, pBlockScanInfo->uid, ik.ts, pBlockScanInfo->delSkyline, &merge,
                              pReader);
H
Haojun Liao 已提交
2254 2255
      if (code != TSDB_CODE_SUCCESS) {
        return code;
2256
      }
2257 2258
    }

2259
    if (minKey == k.ts) {
H
Haojun Liao 已提交
2260
      if (init) {
2261 2262 2263 2264
        if (merge.pTSchema == NULL) {
          return code;
        }

H
Hongze Cheng 已提交
2265
        tsdbRowMerge(&merge, pRow);
H
Haojun Liao 已提交
2266
      } else {
2267
        STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(pRow), pReader, pBlockScanInfo->uid);
H
Hongze Cheng 已提交
2268
        code = tsdbRowMergerInit(&merge, pRow, pSchema);
H
Haojun Liao 已提交
2269 2270 2271
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
2272
      }
H
Haojun Liao 已提交
2273 2274 2275 2276
      code = doMergeRowsInBuf(&pBlockScanInfo->iter, pBlockScanInfo->uid, k.ts, pBlockScanInfo->delSkyline, &merge,
                              pReader);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
2277 2278 2279 2280 2281
      }
    }
  } else {
    if (minKey == k.ts) {
      init = true;
2282
      STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(pRow), pReader, pBlockScanInfo->uid);
H
Hongze Cheng 已提交
2283
      code = tsdbRowMergerInit(&merge, pRow, pSchema);
H
Haojun Liao 已提交
2284 2285 2286 2287
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }

H
Haojun Liao 已提交
2288 2289 2290 2291 2292
      code = doMergeRowsInBuf(&pBlockScanInfo->iter, pBlockScanInfo->uid, k.ts, pBlockScanInfo->delSkyline, &merge,
                              pReader);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
2293 2294 2295
    }

    if (minKey == ik.ts) {
H
Haojun Liao 已提交
2296
      if (init) {
H
Hongze Cheng 已提交
2297
        tsdbRowMerge(&merge, piRow);
H
Haojun Liao 已提交
2298
      } else {
2299 2300
        init = true;
        STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(piRow), pReader, pBlockScanInfo->uid);
H
Hongze Cheng 已提交
2301
        code = tsdbRowMergerInit(&merge, piRow, pSchema);
H
Haojun Liao 已提交
2302 2303 2304
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
2305
      }
H
Haojun Liao 已提交
2306 2307 2308 2309
      code = doMergeRowsInBuf(&pBlockScanInfo->iiter, pBlockScanInfo->uid, ik.ts, pBlockScanInfo->delSkyline, &merge,
                              pReader);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
2310 2311 2312 2313
      }
    }

    if (minKey == tsLast) {
2314
      TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
H
Haojun Liao 已提交
2315
      if (init) {
H
Hongze Cheng 已提交
2316
        tsdbRowMerge(&merge, &fRow1);
H
Haojun Liao 已提交
2317
      } else {
2318
        init = true;
H
Hongze Cheng 已提交
2319
        code = tsdbRowMergerInit(&merge, &fRow1, pReader->pSchema);
H
Haojun Liao 已提交
2320 2321 2322
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
2323
      }
2324
      doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, &merge, &pReader->verRange, pReader->idStr);
2325 2326 2327
    }

    if (minKey == key) {
H
Haojun Liao 已提交
2328
      TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
2329
      if (!init) {
H
Hongze Cheng 已提交
2330
        code = tsdbRowMergerInit(&merge, &fRow, pReader->pSchema);
H
Haojun Liao 已提交
2331 2332 2333
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
H
Haojun Liao 已提交
2334
      } else {
2335 2336 2337
        if (merge.pTSchema == NULL) {
          return code;
        }
H
Hongze Cheng 已提交
2338
        tsdbRowMerge(&merge, &fRow);
2339 2340
      }
      doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
2341 2342 2343
    }
  }

2344
  if (merge.pTSchema == NULL) {
2345 2346 2347
    return code;
  }

H
Hongze Cheng 已提交
2348
  code = tsdbRowMergerGetRow(&merge, &pTSRow);
2349 2350 2351 2352
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

2353
  doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo);
2354 2355

  taosMemoryFree(pTSRow);
H
Hongze Cheng 已提交
2356
  tsdbRowMergerClear(&merge);
2357
  return code;
2358 2359
}

2360 2361 2362 2363 2364 2365 2366 2367 2368
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)) {
2369 2370
    // startKey = (TSDBKEY){.ts = pReader->window.skey, .version = pReader->verRange.minVer};
    startKey = (TSDBKEY){.ts = pBlockScanInfo->lastKey + 1, .version = pReader->verRange.minVer};
2371
  } else {
2372 2373
    // startKey = (TSDBKEY){.ts = pReader->window.ekey, .version = pReader->verRange.maxVer};
    startKey = (TSDBKEY){.ts = pBlockScanInfo->lastKey - 1, .version = pReader->verRange.maxVer};
2374 2375 2376
  }

  int32_t backward = (!ASCENDING_TRAVERSE(pReader->order));
D
dapan1121 已提交
2377
  int64_t st = 0;
2378 2379 2380 2381 2382 2383 2384 2385 2386

  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);

H
Haojun Liao 已提交
2387
        tsdbDebug("%p uid:%" PRIu64 ", check data in mem from skey:%" PRId64 ", order:%d, ts range in buf:%" PRId64
H
Hongze Cheng 已提交
2388
                  "-%" PRId64 " %s",
2389 2390
                  pReader, pBlockScanInfo->uid, startKey.ts, pReader->order, d->minKey, d->maxKey, pReader->idStr);
      } else {
H
Haojun Liao 已提交
2391
        tsdbError("%p uid:%" PRIu64 ", failed to create iterator for imem, code:%s, %s", pReader, pBlockScanInfo->uid,
2392 2393 2394 2395 2396
                  tstrerror(code), pReader->idStr);
        return code;
      }
    }
  } else {
H
Haojun Liao 已提交
2397
    tsdbDebug("%p uid:%" PRIu64 ", no data in mem, %s", pReader, pBlockScanInfo->uid, pReader->idStr);
2398 2399 2400 2401 2402 2403 2404 2405 2406 2407
  }

  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);

H
Haojun Liao 已提交
2408
        tsdbDebug("%p uid:%" PRIu64 ", check data in imem from skey:%" PRId64 ", order:%d, ts range in buf:%" PRId64
H
Hongze Cheng 已提交
2409
                  "-%" PRId64 " %s",
2410 2411
                  pReader, pBlockScanInfo->uid, startKey.ts, pReader->order, di->minKey, di->maxKey, pReader->idStr);
      } else {
H
Haojun Liao 已提交
2412
        tsdbError("%p uid:%" PRIu64 ", failed to create iterator for mem, code:%s, %s", pReader, pBlockScanInfo->uid,
2413 2414 2415 2416 2417
                  tstrerror(code), pReader->idStr);
        return code;
      }
    }
  } else {
H
Haojun Liao 已提交
2418
    tsdbDebug("%p uid:%" PRIu64 ", no data in imem, %s", pReader, pBlockScanInfo->uid, pReader->idStr);
2419 2420
  }

2421
  st = taosGetTimestampUs();
2422
  initDelSkylineIterator(pBlockScanInfo, pReader, d, di);
2423
  pReader->cost.initDelSkylineIterTime += (taosGetTimestampUs() - st) / 1000.0;
2424 2425 2426 2427 2428

  pBlockScanInfo->iterInit = true;
  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
2429 2430
static bool isValidFileBlockRow(SBlockData* pBlockData, SFileBlockDumpInfo* pDumpInfo,
                                STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader) {
2431 2432 2433 2434 2435 2436 2437 2438
  // 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;
    }
  }

2439 2440 2441 2442 2443 2444 2445 2446 2447 2448 2449
  // 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;
  }

2450
  TSDBKEY k = {.ts = ts, .version = ver};
2451 2452
  if (hasBeenDropped(pBlockScanInfo->delSkyline, &pBlockScanInfo->fileDelIndex, &k, pReader->order,
                     &pReader->verRange)) {
2453 2454 2455
    return false;
  }

2456 2457 2458
  return true;
}

2459
static bool initLastBlockReader(SLastBlockReader* pLBlockReader, STableBlockScanInfo* pScanInfo, STsdbReader* pReader) {
2460
  // the last block reader has been initialized for this table.
2461
  if (pLBlockReader->uid == pScanInfo->uid) {
2462
    return hasDataInLastBlock(pLBlockReader);
2463 2464
  }

2465 2466
  if (pLBlockReader->uid != 0) {
    tMergeTreeClose(&pLBlockReader->mergeTree);
2467 2468
  }

2469 2470
  initMemDataIterator(pScanInfo, pReader);
  pLBlockReader->uid = pScanInfo->uid;
2471

H
Hongze Cheng 已提交
2472
  int32_t     step = ASCENDING_TRAVERSE(pLBlockReader->order) ? 1 : -1;
2473 2474
  STimeWindow w = pLBlockReader->window;
  if (ASCENDING_TRAVERSE(pLBlockReader->order)) {
2475
    w.skey = pScanInfo->lastKeyInStt;
2476
  } else {
2477
    w.ekey = pScanInfo->lastKeyInStt;
2478 2479
  }

X
Xiaoyu Wang 已提交
2480 2481
  tsdbDebug("init last block reader, window:%" PRId64 "-%" PRId64 ", uid:%" PRIu64 ", %s", w.skey, w.ekey,
            pScanInfo->uid, pReader->idStr);
2482 2483 2484
  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);
2485 2486 2487 2488
  if (code != TSDB_CODE_SUCCESS) {
    return false;
  }

2489
  return nextRowFromLastBlocks(pLBlockReader, pScanInfo, &pReader->verRange);
2490 2491
}

2492
static int64_t getCurrentKeyInLastBlock(SLastBlockReader* pLastBlockReader) {
2493
  TSDBROW row = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
2494
  return TSDBROW_TS(&row);
2495 2496
}

H
Hongze Cheng 已提交
2497
static bool hasDataInLastBlock(SLastBlockReader* pLastBlockReader) { return pLastBlockReader->mergeTree.pIter != NULL; }
2498

2499
bool hasDataInFileBlock(const SBlockData* pBlockData, const SFileBlockDumpInfo* pDumpInfo) {
H
Haojun Liao 已提交
2500
  if ((pBlockData->nRow > 0) && (pBlockData->nRow != pDumpInfo->totalRows)) {
2501
    return false;  // this is an invalid result.
2502
  }
2503
  return pBlockData->nRow > 0 && (!pDumpInfo->allDumped);
2504
}
2505

2506 2507
int32_t mergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pBlockScanInfo, int64_t key,
                              STsdbReader* pReader) {
2508 2509
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
  if (tryCopyDistinctRowFromFileBlock(pReader, pBlockData, key, pDumpInfo)) {
2510
    pBlockScanInfo->lastKey = key;
2511 2512
    return TSDB_CODE_SUCCESS;
  } else {
C
Cary Xu 已提交
2513 2514
    TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);

H
Hongze Cheng 已提交
2515
    SRow*      pTSRow = NULL;
2516 2517
    SRowMerger merge = {0};

H
Hongze Cheng 已提交
2518
    int32_t code = tsdbRowMergerInit(&merge, &fRow, pReader->pSchema);
H
Haojun Liao 已提交
2519 2520 2521 2522
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

2523
    doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
H
Hongze Cheng 已提交
2524
    code = tsdbRowMergerGetRow(&merge, &pTSRow);
2525 2526 2527 2528
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

2529
    doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo);
2530 2531

    taosMemoryFree(pTSRow);
H
Hongze Cheng 已提交
2532
    tsdbRowMergerClear(&merge);
2533 2534 2535 2536
    return TSDB_CODE_SUCCESS;
  }
}

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

2541
  TSDBROW *pRow = NULL, *piRow = NULL;
2542
  int64_t key = (pBlockData->nRow > 0 && (!pDumpInfo->allDumped)) ? pBlockData->aTSKEY[pDumpInfo->rowIndex] : INT64_MIN;
2543 2544 2545
  if (pBlockScanInfo->iter.hasVal) {
    pRow = getValidMemRow(&pBlockScanInfo->iter, pBlockScanInfo->delSkyline, pReader);
  }
C
Cary Xu 已提交
2546

2547 2548 2549
  if (pBlockScanInfo->iiter.hasVal) {
    piRow = getValidMemRow(&pBlockScanInfo->iiter, pBlockScanInfo->delSkyline, pReader);
  }
C
Cary Xu 已提交
2550

2551 2552 2553 2554
  // two levels of mem-table does contain the valid rows
  if (pRow != NULL && piRow != NULL) {
    return doMergeMultiLevelRows(pReader, pBlockScanInfo, pBlockData, pLastBlockReader);
  }
2555

2556 2557 2558 2559
  // imem + file + last block
  if (pBlockScanInfo->iiter.hasVal) {
    return doMergeBufAndFileRows(pReader, pBlockScanInfo, piRow, &pBlockScanInfo->iiter, key, pLastBlockReader);
  }
2560

2561 2562 2563
  // mem + file + last block
  if (pBlockScanInfo->iter.hasVal) {
    return doMergeBufAndFileRows(pReader, pBlockScanInfo, pRow, &pBlockScanInfo->iter, key, pLastBlockReader);
2564
  }
2565 2566 2567

  // files data blocks + last block
  return mergeFileBlockAndLastBlock(pReader, pLastBlockReader, key, pBlockScanInfo, pBlockData);
2568 2569
}

H
Haojun Liao 已提交
2570 2571 2572 2573 2574 2575 2576 2577 2578 2579 2580 2581 2582 2583 2584 2585 2586 2587 2588 2589 2590 2591 2592 2593 2594 2595 2596 2597 2598 2599 2600 2601 2602 2603 2604 2605 2606 2607 2608 2609
static int32_t loadNeighborIfOverlap(SFileDataBlockInfo* pBlockInfo, STableBlockScanInfo* pBlockScanInfo,
                                     STsdbReader* pReader, bool* loadNeighbor) {
  int32_t     code = TSDB_CODE_SUCCESS;
  int32_t     step = ASCENDING_TRAVERSE(pReader->order) ? 1 : -1;
  int32_t     nextIndex = -1;
  SBlockIndex nxtBIndex = {0};

  *loadNeighbor = false;
  SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter);

  bool hasNeighbor = getNeighborBlockOfSameTable(pBlockInfo, pBlockScanInfo, &nextIndex, pReader->order, &nxtBIndex);
  if (!hasNeighbor) {  // do nothing
    return code;
  }

  if (overlapWithNeighborBlock(pBlock, &nxtBIndex, pReader->order)) {  // load next block
    SReaderStatus*  pStatus = &pReader->status;
    SDataBlockIter* pBlockIter = &pStatus->blockIter;

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

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

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

    // 4. check the data values
    initBlockDumpInfo(pReader, pBlockIter);
    *loadNeighbor = true;
  }

  return code;
}

2610 2611 2612 2613 2614 2615 2616 2617 2618 2619 2620 2621 2622
static void updateComposedBlockInfo(STsdbReader* pReader, double el, STableBlockScanInfo* pBlockScanInfo) {
  SSDataBlock* pResBlock = pReader->pResBlock;

  pResBlock->info.id.uid = (pBlockScanInfo != NULL) ? pBlockScanInfo->uid : 0;
  pResBlock->info.dataLoad = 1;
  blockDataUpdateTsWindow(pResBlock, pReader->suppInfo.slotId[0]);

  setComposedBlockFlag(pReader, true);

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

2623
static int32_t buildComposedDataBlock(STsdbReader* pReader) {
H
Haojun Liao 已提交
2624 2625
  int32_t code = TSDB_CODE_SUCCESS;

2626 2627
  SSDataBlock* pResBlock = pReader->pResBlock;

H
Hongze Cheng 已提交
2628
  SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(&pReader->status.blockIter);
C
Cary Xu 已提交
2629 2630
  SLastBlockReader*   pLastBlockReader = pReader->status.fileIter.pLastBlockReader;

2631
  bool    asc = ASCENDING_TRAVERSE(pReader->order);
C
Cary Xu 已提交
2632
  int64_t st = taosGetTimestampUs();
2633
  int32_t step = asc ? 1 : -1;
2634
  double  el = 0;
2635 2636 2637

  STableBlockScanInfo* pBlockScanInfo = NULL;
  if (pBlockInfo != NULL) {
H
Haojun Liao 已提交
2638 2639
    pBlockScanInfo = getTableBlockScanInfo(pReader->status.pTableMap, pBlockInfo->uid, pReader->idStr);
    if (pBlockScanInfo == NULL) {
H
Haojun Liao 已提交
2640 2641 2642
      goto _end;
    }

C
Cary Xu 已提交
2643
    SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter);
H
Hongze Cheng 已提交
2644
    TSDBKEY   keyInBuf = getCurrentKeyInBuf(pBlockScanInfo, pReader);
C
Cary Xu 已提交
2645 2646

    // it is a clean block, load it directly
H
Hongze Cheng 已提交
2647
    if (isCleanFileDataBlock(pReader, pBlockInfo, pBlock, pBlockScanInfo, keyInBuf, pLastBlockReader) &&
2648
        pBlock->nRow <= pReader->capacity) {
2649
      if (asc || ((!asc) && (!hasDataInLastBlock(pLastBlockReader)))) {
2650
        copyBlockDataToSDataBlock(pReader);
2651 2652

        // record the last key value
H
Hongze Cheng 已提交
2653
        pBlockScanInfo->lastKey = asc ? pBlock->maxKey.ts : pBlock->minKey.ts;
H
Haojun Liao 已提交
2654 2655
        goto _end;
      }
C
Cary Xu 已提交
2656 2657
    }
  } else {  // file blocks not exist
2658
    pBlockScanInfo = *pReader->status.pTableIter;
2659 2660
  }

2661
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
2662
  SBlockData*         pBlockData = &pReader->status.fileBlockData;
2663

2664
  while (1) {
2665
    bool hasBlockData = false;
2666
    {
2667 2668
      while (pBlockData->nRow > 0 &&
             pBlockData->uid == pBlockScanInfo->uid) {  // find the first qualified row in data block
2669 2670 2671 2672 2673
        if (isValidFileBlockRow(pBlockData, pDumpInfo, pBlockScanInfo, pReader)) {
          hasBlockData = true;
          break;
        }

2674 2675
        pDumpInfo->rowIndex += step;

H
Hongze Cheng 已提交
2676
        SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter);
2677
        if (pDumpInfo->rowIndex >= pBlock->nRow || pDumpInfo->rowIndex < 0) {
H
Haojun Liao 已提交
2678
          pBlockInfo = getCurrentBlockInfo(&pReader->status.blockIter);  // NOTE: get the new block info
H
Haojun Liao 已提交
2679

H
Haojun Liao 已提交
2680 2681 2682 2683 2684
          // continue check for the next file block if the last ts in the current block
          // is overlapped with the next neighbor block
          bool loadNeighbor = false;
          code = loadNeighborIfOverlap(pBlockInfo, pBlockScanInfo, pReader, &loadNeighbor);
          if ((!loadNeighbor) || (code != 0)) {
2685 2686
            setBlockAllDumped(pDumpInfo, pBlock->maxKey.ts, pReader->order);
            break;
2687
          }
2688 2689
        }
      }
2690
    }
2691

2692
    // no data in last block and block, no need to proceed.
2693
    if (hasBlockData == false) {
2694
      break;
2695 2696
    }

2697
    buildComposedDataBlockImpl(pReader, pBlockScanInfo, pBlockData, pLastBlockReader);
2698

2699
    // currently loaded file data block is consumed
2700
    if ((pBlockData->nRow > 0) && (pDumpInfo->rowIndex >= pBlockData->nRow || pDumpInfo->rowIndex < 0)) {
H
Hongze Cheng 已提交
2701
      SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter);
2702
      setBlockAllDumped(pDumpInfo, pBlock->maxKey.ts, pReader->order);
2703 2704 2705 2706 2707
      break;
    }

    if (pResBlock->info.rows >= pReader->capacity) {
      break;
2708 2709 2710
    }
  }

H
Hongze Cheng 已提交
2711
_end:
2712 2713
  el = (taosGetTimestampUs() - st) / 1000.0;
  updateComposedBlockInfo(pReader, el, pBlockScanInfo);
2714

2715 2716 2717
  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 已提交
2718
              pReader, pResBlock->info.id.uid, pResBlock->info.window.skey, pResBlock->info.window.ekey,
H
Haojun Liao 已提交
2719
              pResBlock->info.rows, el, pReader->idStr);
2720
  }
2721

H
Haojun Liao 已提交
2722
  return code;
2723 2724 2725 2726
}

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

2727 2728 2729 2730 2731 2732 2733 2734
int32_t getInitialDelIndex(const SArray* pDelSkyline, int32_t order) {
  if (pDelSkyline == NULL) {
    return 0;
  }

  return ASCENDING_TRAVERSE(order) ? 0 : taosArrayGetSize(pDelSkyline) - 1;
}

dengyihao's avatar
dengyihao 已提交
2735 2736
int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader, STbData* pMemTbData,
                               STbData* piMemTbData) {
2737 2738 2739
  if (pBlockScanInfo->delSkyline != NULL) {
    return TSDB_CODE_SUCCESS;
  }
2740

2741
  int32_t code = 0;
2742 2743
  SArray* pDelData = taosArrayInit(4, sizeof(SDelData));

H
Hongze Cheng 已提交
2744
  SDelFile* pDelFile = pReader->pReadSnap->fs.pDelFile;
2745
  if (pDelFile && taosArrayGetSize(pReader->pDelIdx) > 0) {
2746
    SDelIdx  idx = {.suid = pReader->suid, .uid = pBlockScanInfo->uid};
2747
    SDelIdx* pIdx = taosArraySearch(pReader->pDelIdx, &idx, tCmprDelIdx, TD_EQ);
2748

H
Haojun Liao 已提交
2749
    if (pIdx != NULL) {
2750
      code = tsdbReadDelData(pReader->pDelFReader, pIdx, pDelData);
2751 2752 2753
    }
    if (code != TSDB_CODE_SUCCESS) {
      goto _err;
2754
    }
2755
  }
2756

2757 2758 2759 2760 2761 2762 2763
  SDelData* p = NULL;
  if (pMemTbData != NULL) {
    p = pMemTbData->pHead;
    while (p) {
      taosArrayPush(pDelData, p);
      p = p->pNext;
    }
2764 2765
  }

2766 2767 2768 2769 2770 2771 2772 2773 2774 2775 2776 2777 2778 2779
  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);
2780 2781 2782 2783 2784 2785 2786
  int32_t index = getInitialDelIndex(pBlockScanInfo->delSkyline, pReader->order);

  pBlockScanInfo->iter.index = index;
  pBlockScanInfo->iiter.index = index;
  pBlockScanInfo->fileDelIndex = index;
  pBlockScanInfo->lastBlockDelIndex = index;

2787 2788
  return code;

2789 2790 2791
_err:
  taosArrayDestroy(pDelData);
  return code;
2792 2793
}

C
Cary Xu 已提交
2794
TSDBKEY getCurrentKeyInBuf(STableBlockScanInfo* pScanInfo, STsdbReader* pReader) {
2795
  bool asc = ASCENDING_TRAVERSE(pReader->order);
X
Xiaoyu Wang 已提交
2796
  //  TSKEY initialVal = asc? TSKEY_MIN:TSKEY_MAX;
2797

X
Xiaoyu Wang 已提交
2798
  TSDBKEY key = {.ts = TSKEY_INITIAL_VAL}, ikey = {.ts = TSKEY_INITIAL_VAL};
2799

X
Xiaoyu Wang 已提交
2800
  bool     hasKey = false, hasIKey = false;
2801
  TSDBROW* pRow = getValidMemRow(&pScanInfo->iter, pScanInfo->delSkyline, pReader);
2802
  if (pRow != NULL) {
2803
    hasKey = true;
2804 2805 2806
    key = TSDBROW_KEY(pRow);
  }

2807 2808 2809 2810
  TSDBROW* pIRow = getValidMemRow(&pScanInfo->iiter, pScanInfo->delSkyline, pReader);
  if (pIRow != NULL) {
    hasIKey = true;
    ikey = TSDBROW_KEY(pIRow);
2811 2812
  }

2813
  if (hasKey) {
X
Xiaoyu Wang 已提交
2814
    if (hasIKey) {  // has data in mem & imem
2815 2816
      if (asc) {
        return key.ts <= ikey.ts ? key : ikey;
X
Xiaoyu Wang 已提交
2817 2818
      } else {
        return key.ts <= ikey.ts ? ikey : key;
2819 2820 2821
      }
    } else {  // no data in imem
      return key;
2822
    }
2823 2824 2825 2826
  } else {
    // no data in mem & imem, return the initial value
    // only imem has data, return ikey
    return ikey;
2827 2828 2829
  }
}

2830
static int32_t moveToNextFile(STsdbReader* pReader, SBlockNumber* pBlockNum) {
H
Haojun Liao 已提交
2831
  SReaderStatus* pStatus = &pReader->status;
2832
  pBlockNum->numOfBlocks = 0;
2833
  pBlockNum->numOfLastFiles = 0;
2834

2835
  size_t  numOfTables = taosHashGetSize(pReader->status.pTableMap);
2836
  SArray* pIndexList = taosArrayInit(numOfTables, sizeof(SBlockIdx));
H
Haojun Liao 已提交
2837 2838

  while (1) {
dengyihao's avatar
dengyihao 已提交
2839
    bool    hasNext = false;
D
dapan1121 已提交
2840 2841 2842 2843 2844
    int32_t code = filesetIteratorNext(&pStatus->fileIter, pReader, &hasNext);
    if (code) {
      taosArrayDestroy(pIndexList);
      return code;
    }
dengyihao's avatar
dengyihao 已提交
2845

2846
    if (!hasNext) {  // no data files on disk
H
Haojun Liao 已提交
2847 2848 2849
      break;
    }

H
Haojun Liao 已提交
2850
    taosArrayClear(pIndexList);
D
dapan1121 已提交
2851
    code = doLoadBlockIndex(pReader, pReader->pFileReader, pIndexList);
H
Haojun Liao 已提交
2852
    if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2853
      taosArrayDestroy(pIndexList);
H
Haojun Liao 已提交
2854 2855 2856
      return code;
    }

H
Hongze Cheng 已提交
2857
    if (taosArrayGetSize(pIndexList) > 0 || pReader->pFileReader->pSet->nSttF > 0) {
2858
      code = doLoadFileBlock(pReader, pIndexList, pBlockNum);
H
Haojun Liao 已提交
2859
      if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2860
        taosArrayDestroy(pIndexList);
H
Haojun Liao 已提交
2861 2862 2863
        return code;
      }

2864
      if (pBlockNum->numOfBlocks + pBlockNum->numOfLastFiles > 0) {
H
Haojun Liao 已提交
2865 2866 2867
        break;
      }
    }
2868

H
Haojun Liao 已提交
2869 2870 2871
    // no blocks in current file, try next files
  }

H
Haojun Liao 已提交
2872
  taosArrayDestroy(pIndexList);
2873

H
Haojun Liao 已提交
2874 2875 2876 2877 2878 2879 2880
  if (pReader->pReadSnap != NULL) {
    SDelFile* pDelFile = pReader->pReadSnap->fs.pDelFile;
    if (pReader->pDelFReader == NULL && pDelFile != NULL) {
      int32_t code = tsdbDelFReaderOpen(&pReader->pDelFReader, pDelFile, pReader->pTsdb);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
2881

H
Haojun Liao 已提交
2882 2883 2884 2885 2886 2887 2888 2889 2890 2891 2892
      pReader->pDelIdx = taosArrayInit(4, sizeof(SDelIdx));
      if (pReader->pDelIdx == NULL) {
        code = TSDB_CODE_OUT_OF_MEMORY;
        return code;
      }

      code = tsdbReadDelIdx(pReader->pDelFReader, pReader->pDelIdx);
      if (code != TSDB_CODE_SUCCESS) {
        taosArrayDestroy(pReader->pDelIdx);
        return code;
      }
2893 2894 2895
    }
  }

H
Haojun Liao 已提交
2896 2897 2898
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
2899
static void resetTableListIndex(SReaderStatus* pStatus) {
2900
  STableUidList* pList = &pStatus->uidList;
2901

H
Haojun Liao 已提交
2902 2903 2904
  pList->currentIndex = 0;
  uint64_t uid = pList->tableUidList[0];
  pStatus->pTableIter = taosHashGet(pStatus->pTableMap, &uid, sizeof(uid));
2905 2906
}

2907
static bool moveToNextTable(STableUidList* pOrderedCheckInfo, SReaderStatus* pStatus) {
2908 2909 2910 2911 2912 2913 2914 2915
  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));
2916
  return (pStatus->pTableIter != NULL);
2917 2918
}

2919
static int32_t doLoadLastBlockSequentially(STsdbReader* pReader) {
2920
  SReaderStatus*    pStatus = &pReader->status;
2921
  SLastBlockReader* pLastBlockReader = pStatus->fileIter.pLastBlockReader;
2922
  STableUidList*    pUidList = &pStatus->uidList;
2923

H
Haojun Liao 已提交
2924 2925
  if (taosHashGetSize(pStatus->pTableMap) == 0) {
    return TSDB_CODE_SUCCESS;
2926
  }
2927

2928 2929
  SSDataBlock* pResBlock = pReader->pResBlock;

2930
  while (1) {
2931
    // load the last data block of current table
H
Hongze Cheng 已提交
2932
    STableBlockScanInfo* pScanInfo = *(STableBlockScanInfo**)pStatus->pTableIter;
2933 2934

    bool hasVal = initLastBlockReader(pLastBlockReader, pScanInfo, pReader);
2935
    if (!hasVal) {
2936
      bool hasNexTable = moveToNextTable(pUidList, pStatus);
2937
      if (!hasNexTable) {
2938 2939
        return TSDB_CODE_SUCCESS;
      }
2940

2941
      continue;
2942 2943
    }

2944 2945 2946 2947 2948 2949 2950 2951 2952 2953 2954 2955 2956
    int64_t st = taosGetTimestampUs();
    while (1) {
      bool hasBlockLData = hasDataInLastBlock(pLastBlockReader);

      // no data in last block and block, no need to proceed.
      if (hasBlockLData == false) {
        break;
      }

      buildComposedDataBlockImpl(pReader, pScanInfo, &pReader->status.fileBlockData, pLastBlockReader);
      if (pResBlock->info.rows >= pReader->capacity) {
        break;
      }
2957 2958
    }

2959 2960 2961 2962 2963 2964 2965 2966
    double el = (taosGetTimestampUs() - st) / 1000.0;
    updateComposedBlockInfo(pReader, el, pScanInfo);

    if (pResBlock->info.rows > 0) {
      tsdbDebug("%p uid:%" PRIu64 ", composed data block created, brange:%" PRIu64 "-%" PRIu64
                " rows:%d, elapsed time:%.2f ms %s",
                pReader, pResBlock->info.id.uid, pResBlock->info.window.skey, pResBlock->info.window.ekey,
                pResBlock->info.rows, el, pReader->idStr);
2967 2968
      return TSDB_CODE_SUCCESS;
    }
2969

2970
    // current table is exhausted, let's try next table
2971
    bool hasNexTable = moveToNextTable(pUidList, pStatus);
2972
    if (!hasNexTable) {
2973 2974
      return TSDB_CODE_SUCCESS;
    }
2975 2976 2977
  }
}

2978
static int32_t doBuildDataBlock(STsdbReader* pReader) {
H
Hongze Cheng 已提交
2979 2980
  int32_t   code = TSDB_CODE_SUCCESS;
  SDataBlk* pBlock = NULL;
2981 2982 2983

  SReaderStatus*       pStatus = &pReader->status;
  SDataBlockIter*      pBlockIter = &pStatus->blockIter;
2984 2985 2986
  STableBlockScanInfo* pScanInfo = NULL;
  SFileDataBlockInfo*  pBlockInfo = getCurrentBlockInfo(pBlockIter);
  SLastBlockReader*    pLastBlockReader = pReader->status.fileIter.pLastBlockReader;
2987

H
Haojun Liao 已提交
2988
  pScanInfo = getTableBlockScanInfo(pReader->status.pTableMap, pBlockInfo->uid, pReader->idStr);
H
Haojun Liao 已提交
2989
  if (pScanInfo == NULL) {
H
Haojun Liao 已提交
2990
    return terrno;
H
Haojun Liao 已提交
2991 2992
  }

2993
  pBlock = getCurrentBlock(pBlockIter);
2994

2995
  initLastBlockReader(pLastBlockReader, pScanInfo, pReader);
C
Cary Xu 已提交
2996
  TSDBKEY keyInBuf = getCurrentKeyInBuf(pScanInfo, pReader);
2997

2998
  if (fileBlockShouldLoad(pReader, pBlockInfo, pBlock, pScanInfo, keyInBuf, pLastBlockReader)) {
2999
    code = doLoadFileBlockData(pReader, pBlockIter, &pStatus->fileBlockData, pScanInfo->uid);
3000 3001
    if (code != TSDB_CODE_SUCCESS) {
      return code;
3002 3003 3004
    }

    // build composed data block
3005
    code = buildComposedDataBlock(pReader);
C
Cary Xu 已提交
3006
  } else if (bufferDataInFileBlockGap(pReader->order, keyInBuf, pBlock)) {
3007
    // data in memory that are earlier than current file block
3008
    // rows in buffer should be less than the file block in asc, greater than file block in desc
3009
    int64_t endKey = (ASCENDING_TRAVERSE(pReader->order)) ? pBlock->minKey.ts : pBlock->maxKey.ts;
3010
    code = buildDataBlockFromBuf(pReader, pScanInfo, endKey);
3011 3012 3013 3014
  } else {
    if (hasDataInLastBlock(pLastBlockReader) && !ASCENDING_TRAVERSE(pReader->order)) {
      // only return the rows in last block
      int64_t tsLast = getCurrentKeyInLastBlock(pLastBlockReader);
H
Hongze Cheng 已提交
3015
      ASSERT(tsLast >= pBlock->maxKey.ts);
3016

3017 3018 3019 3020
      SBlockData* pBData = &pReader->status.fileBlockData;
      tBlockDataReset(pBData);

      SSDataBlock* pResBlock = pReader->pResBlock;
3021
      tsdbDebug("load data in last block firstly, due to desc scan data, %s", pReader->idStr);
3022 3023 3024 3025 3026 3027 3028 3029 3030 3031 3032 3033 3034 3035 3036 3037 3038 3039 3040 3041 3042 3043

      int64_t st = taosGetTimestampUs();

      while (1) {
        bool hasBlockLData = hasDataInLastBlock(pLastBlockReader);

        // no data in last block and block, no need to proceed.
        if (hasBlockLData == false) {
          break;
        }

        buildComposedDataBlockImpl(pReader, pScanInfo, &pReader->status.fileBlockData, pLastBlockReader);
        if (pResBlock->info.rows >= pReader->capacity) {
          break;
        }
      }

      double el = (taosGetTimestampUs() - st) / 1000.0;
      updateComposedBlockInfo(pReader, el, pScanInfo);

      if (pResBlock->info.rows > 0) {
        tsdbDebug("%p uid:%" PRIu64 ", composed data block created, brange:%" PRIu64 "-%" PRIu64
X
Xiaoyu Wang 已提交
3044
                  " rows:%d, elapsed time:%.2f ms %s",
3045 3046 3047
                  pReader, pResBlock->info.id.uid, pResBlock->info.window.skey, pResBlock->info.window.ekey,
                  pResBlock->info.rows, el, pReader->idStr);
      }
H
Hongze Cheng 已提交
3048
    } else {  // whole block is required, return it directly
3049 3050
      SDataBlockInfo* pInfo = &pReader->pResBlock->info;
      pInfo->rows = pBlock->nRow;
H
Haojun Liao 已提交
3051
      pInfo->id.uid = pScanInfo->uid;
3052
      pInfo->dataLoad = 0;
3053 3054 3055
      pInfo->window = (STimeWindow){.skey = pBlock->minKey.ts, .ekey = pBlock->maxKey.ts};
      setComposedBlockFlag(pReader, false);
      setBlockAllDumped(&pStatus->fBlockDumpInfo, pBlock->maxKey.ts, pReader->order);
3056

3057
      // update the last key for the corresponding table
H
Hongze Cheng 已提交
3058
      pScanInfo->lastKey = ASCENDING_TRAVERSE(pReader->order) ? pInfo->window.ekey : pInfo->window.skey;
X
Xiaoyu Wang 已提交
3059 3060
      tsdbDebug("%p uid:%" PRIu64
                " clean file block retrieved from file, global index:%d, "
H
Haojun Liao 已提交
3061 3062 3063
                "table index:%d, rows:%d, brange:%" PRId64 "-%" PRId64 ", %s",
                pReader, pScanInfo->uid, pBlockIter->index, pBlockInfo->tbBlockIdx, pBlock->nRow, pBlock->minKey.ts,
                pBlock->maxKey.ts, pReader->idStr);
3064
    }
3065 3066 3067 3068 3069
  }

  return code;
}

H
Haojun Liao 已提交
3070
static int32_t buildBlockFromBufferSequentially(STsdbReader* pReader) {
3071
  SReaderStatus* pStatus = &pReader->status;
3072
  STableUidList* pUidList = &pStatus->uidList;
3073

3074
  while (1) {
X
Xiaoyu Wang 已提交
3075 3076 3077 3078 3079 3080
    //    if (pStatus->pTableIter == NULL) {
    //      pStatus->pTableIter = taosHashIterate(pStatus->pTableMap, NULL);
    //      if (pStatus->pTableIter == NULL) {
    //        return TSDB_CODE_SUCCESS;
    //      }
    //    }
3081

3082 3083
    STableBlockScanInfo** pBlockScanInfo = pStatus->pTableIter;
    initMemDataIterator(*pBlockScanInfo, pReader);
3084

3085
    int64_t endKey = (ASCENDING_TRAVERSE(pReader->order)) ? INT64_MAX : INT64_MIN;
3086
    int32_t code = buildDataBlockFromBuf(pReader, *pBlockScanInfo, endKey);
H
Haojun Liao 已提交
3087 3088 3089 3090
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

3091
    if (pReader->pResBlock->info.rows > 0) {
H
Haojun Liao 已提交
3092
      return TSDB_CODE_SUCCESS;
3093 3094
    }

3095 3096 3097
    // current table is exhausted, let's try next table
    bool hasNexTable = moveToNextTable(pUidList, pStatus);
    if (!hasNexTable) {
H
Haojun Liao 已提交
3098
      return TSDB_CODE_SUCCESS;
3099 3100 3101 3102
    }
  }
}

3103
// set the correct start position in case of the first/last file block, according to the query time window
3104
static void initBlockDumpInfo(STsdbReader* pReader, SDataBlockIter* pBlockIter) {
3105 3106 3107 3108 3109 3110 3111 3112
  int64_t             lastKey = ASCENDING_TRAVERSE(pReader->order) ? INT64_MIN : INT64_MAX;
  SDataBlk*           pBlock = getCurrentBlock(pBlockIter);
  SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter);
  if (pBlockInfo) {
    STableBlockScanInfo* pScanInfo = taosHashGet(pBlockIter->pTableMap, &pBlockInfo->uid, sizeof(pBlockInfo->uid));
    if (pScanInfo) {
      lastKey = pScanInfo->lastKey;
    }
3113
  }
3114 3115 3116
  SReaderStatus* pStatus = &pReader->status;

  SFileBlockDumpInfo* pDumpInfo = &pStatus->fBlockDumpInfo;
3117 3118 3119

  pDumpInfo->totalRows = pBlock->nRow;
  pDumpInfo->allDumped = false;
3120
  pDumpInfo->rowIndex = ASCENDING_TRAVERSE(pReader->order) ? 0 : pBlock->nRow - 1;
3121
  pDumpInfo->lastKey = lastKey;
3122 3123
}

3124
static int32_t initForFirstBlockInFile(STsdbReader* pReader, SDataBlockIter* pBlockIter) {
3125
  SBlockNumber num = {0};
X
Xiaoyu Wang 已提交
3126
  int32_t      code = moveToNextFile(pReader, &num);
3127 3128 3129 3130 3131
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

  // all data files are consumed, try data in buffer
3132
  if (num.numOfBlocks + num.numOfLastFiles == 0) {
3133 3134 3135 3136 3137
    pReader->status.loadFromFile = false;
    return code;
  }

  // initialize the block iterator for a new fileset
3138 3139
  if (num.numOfBlocks > 0) {
    code = initBlockIterator(pReader, pBlockIter, num.numOfBlocks);
H
Hongze Cheng 已提交
3140
  } else {  // no block data, only last block exists
3141
    tBlockDataReset(&pReader->status.fileBlockData);
3142
    resetDataBlockIterator(pBlockIter, pReader->order);
H
Haojun Liao 已提交
3143
    resetTableListIndex(&pReader->status);
3144
  }
3145 3146

  // set the correct start position according to the query time window
3147
  initBlockDumpInfo(pReader, pBlockIter);
3148 3149 3150
  return code;
}

3151
static bool fileBlockPartiallyRead(SFileBlockDumpInfo* pDumpInfo, bool asc) {
3152 3153
  return (!pDumpInfo->allDumped) &&
         ((pDumpInfo->rowIndex > 0 && asc) || (pDumpInfo->rowIndex < (pDumpInfo->totalRows - 1) && (!asc)));
3154 3155
}

3156
static int32_t buildBlockFromFiles(STsdbReader* pReader) {
H
Haojun Liao 已提交
3157
  int32_t code = TSDB_CODE_SUCCESS;
3158 3159
  bool    asc = ASCENDING_TRAVERSE(pReader->order);

3160 3161
  SDataBlockIter* pBlockIter = &pReader->status.blockIter;

3162
  if (pBlockIter->numOfBlocks == 0) {
H
Hongze Cheng 已提交
3163
  _begin:
3164 3165 3166 3167 3168
    code = doLoadLastBlockSequentially(pReader);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

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

3173
    // all data blocks are checked in this last block file, now let's try the next file
3174 3175 3176 3177 3178 3179 3180 3181
    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;
      }

3182
      // this file does not have data files, let's start check the last block file if exists
3183
      if (pBlockIter->numOfBlocks == 0) {
H
Haojun Liao 已提交
3184
        resetTableListIndex(&pReader->status);
3185 3186 3187 3188 3189 3190 3191 3192 3193 3194 3195 3196 3197 3198
        goto _begin;
      }
    }

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

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

3199
  while (1) {
3200 3201
    SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;

3202
    if (fileBlockPartiallyRead(pDumpInfo, asc)) {  // file data block is partially loaded
3203
      code = buildComposedDataBlock(pReader);
3204 3205 3206 3207
    } else {
      // current block are exhausted, try the next file block
      if (pDumpInfo->allDumped) {
        // try next data block in current file
H
Haojun Liao 已提交
3208
        bool hasNext = blockIteratorNext(&pReader->status.blockIter, pReader->idStr);
3209 3210
        if (hasNext) {  // check for the next block in the block accessed order list
          initBlockDumpInfo(pReader, pBlockIter);
3211
        } else {
H
Haojun Liao 已提交
3212
          if (pReader->status.pCurrentFileset->nSttF > 0) {
3213
            // data blocks in current file are exhausted, let's try the next file now
H
Haojun Liao 已提交
3214 3215 3216 3217 3218 3219
            SBlockData* pBlockData = &pReader->status.fileBlockData;
            if (pBlockData->uid != 0) {
              tBlockDataClear(pBlockData);
            }

            tBlockDataReset(pBlockData);
3220
            resetDataBlockIterator(pBlockIter, pReader->order);
H
Haojun Liao 已提交
3221
            resetTableListIndex(&pReader->status);
3222 3223 3224
            goto _begin;
          } else {
            code = initForFirstBlockInFile(pReader, pBlockIter);
3225

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

3231 3232
            // this file does not have blocks, let's start check the last block file
            if (pBlockIter->numOfBlocks == 0) {
H
Haojun Liao 已提交
3233
              resetTableListIndex(&pReader->status);
3234 3235
              goto _begin;
            }
3236
          }
3237
        }
H
Haojun Liao 已提交
3238
      }
3239 3240

      code = doBuildDataBlock(pReader);
3241 3242
    }

3243 3244 3245 3246 3247 3248 3249 3250
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

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

3253 3254
static STsdb* getTsdbByRetentions(SVnode* pVnode, TSKEY winSKey, SRetention* retentions, const char* idStr,
                                  int8_t* pLevel) {
3255
  if (VND_IS_RSMA(pVnode)) {
3256
    int8_t  level = 0;
3257 3258
    int8_t  precision = pVnode->config.tsdbCfg.precision;
    int64_t now = taosGetTimestamp(precision);
H
Haojun Liao 已提交
3259 3260 3261
    int64_t offset = tsQueryRsmaTolerance * ((precision == TSDB_TIME_PRECISION_MILLI)   ? 1L
                                             : (precision == TSDB_TIME_PRECISION_MICRO) ? 1000L
                                                                                        : 1000000L);
3262

3263
    for (int8_t i = 0; i < TSDB_RETENTION_MAX; ++i) {
3264 3265 3266 3267 3268 3269 3270
      SRetention* pRetention = retentions + level;
      if (pRetention->keep <= 0) {
        if (level > 0) {
          --level;
        }
        break;
      }
3271
      if ((now - pRetention->keep) <= (winSKey + offset)) {
3272 3273 3274 3275 3276
        break;
      }
      ++level;
    }

3277
    const char* str = (idStr != NULL) ? idStr : "";
3278 3279

    if (level == TSDB_RETENTION_L0) {
3280
      *pLevel = TSDB_RETENTION_L0;
C
Cary Xu 已提交
3281
      tsdbDebug("vgId:%d, rsma level %d is selected to query %s", TD_VID(pVnode), TSDB_RETENTION_L0, str);
3282 3283
      return VND_RSMA0(pVnode);
    } else if (level == TSDB_RETENTION_L1) {
3284
      *pLevel = TSDB_RETENTION_L1;
C
Cary Xu 已提交
3285
      tsdbDebug("vgId:%d, rsma level %d is selected to query %s", TD_VID(pVnode), TSDB_RETENTION_L1, str);
3286 3287
      return VND_RSMA1(pVnode);
    } else {
3288
      *pLevel = TSDB_RETENTION_L2;
C
Cary Xu 已提交
3289
      tsdbDebug("vgId:%d, rsma level %d is selected to query %s", TD_VID(pVnode), TSDB_RETENTION_L2, str);
3290 3291 3292 3293 3294 3295 3296
      return VND_RSMA2(pVnode);
    }
  }

  return VND_TSDB(pVnode);
}

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

  int64_t endVer = 0;
L
Liu Jicong 已提交
3301 3302
  if (pCond->endVersion ==
      -1) {  // user not specified end version, set current maximum version of vnode as the endVersion
H
Haojun Liao 已提交
3303 3304
    endVer = pVnode->state.applied;
  } else {
L
Liu Jicong 已提交
3305
    endVer = (pCond->endVersion > pVnode->state.applied) ? pVnode->state.applied : pCond->endVersion;
3306 3307
  }

H
Haojun Liao 已提交
3308
  return (SVersionRange){.minVer = startVer, .maxVer = endVer};
3309 3310
}

3311
bool hasBeenDropped(const SArray* pDelList, int32_t* index, TSDBKEY* pKey, int32_t order, SVersionRange* pVerRange) {
3312 3313 3314
  if (pDelList == NULL) {
    return false;
  }
H
Haojun Liao 已提交
3315

L
Liu Jicong 已提交
3316 3317 3318
  size_t  num = taosArrayGetSize(pDelList);
  bool    asc = ASCENDING_TRAVERSE(order);
  int32_t step = asc ? 1 : -1;
3319

3320 3321 3322 3323 3324 3325
  if (asc) {
    if (*index >= num - 1) {
      TSDBKEY* last = taosArrayGetLast(pDelList);
      ASSERT(pKey->ts >= last->ts);

      if (pKey->ts > last->ts) {
3326
        return false;
3327 3328
      } else if (pKey->ts == last->ts) {
        TSDBKEY* prev = taosArrayGet(pDelList, num - 2);
3329 3330
        return (prev->version >= pKey->version && prev->version <= pVerRange->maxVer &&
                prev->version >= pVerRange->minVer);
3331 3332
      }
    } else {
3333 3334 3335 3336 3337 3338 3339
      TSDBKEY* pCurrent = taosArrayGet(pDelList, *index);
      TSDBKEY* pNext = taosArrayGet(pDelList, (*index) + 1);

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

3340 3341
      if (pCurrent->ts <= pKey->ts && pNext->ts >= pKey->ts && pCurrent->version >= pKey->version &&
          pVerRange->maxVer >= pCurrent->version) {
3342 3343 3344 3345 3346 3347 3348 3349 3350 3351 3352 3353 3354 3355 3356
        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;
          }

3357 3358
          if (pCurrent->ts <= pKey->ts && pNext->ts >= pKey->ts && pCurrent->version >= pKey->version &&
              pVerRange->maxVer >= pCurrent->version) {
3359 3360 3361 3362 3363 3364
            return true;
          }
        }
      }

      return false;
3365 3366
    }
  } else {
3367 3368
    if (*index <= 0) {
      TSDBKEY* pFirst = taosArrayGet(pDelList, 0);
3369

3370 3371 3372 3373 3374 3375 3376
      if (pKey->ts < pFirst->ts) {
        return false;
      } else if (pKey->ts == pFirst->ts) {
        return pFirst->version >= pKey->version;
      } else {
        ASSERT(0);
      }
3377
    } else {
3378 3379 3380 3381 3382 3383 3384 3385 3386 3387 3388 3389 3390 3391 3392 3393 3394 3395 3396 3397 3398 3399 3400 3401 3402 3403 3404
      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;
          }
        }
3405 3406 3407 3408 3409
      }

      return false;
    }
  }
3410 3411

  return false;
3412 3413
}

3414
TSDBROW* getValidMemRow(SIterInfo* pIter, const SArray* pDelList, STsdbReader* pReader) {
3415
  if (!pIter->hasVal) {
H
Haojun Liao 已提交
3416 3417
    return NULL;
  }
H
Hongze Cheng 已提交
3418

3419
  TSDBROW* pRow = tsdbTbDataIterGet(pIter->iter);
H
Hongze Cheng 已提交
3420 3421
  TSDBKEY  key = TSDBROW_KEY(pRow);

3422
  if (outOfTimeWindow(key.ts, &pReader->window)) {
3423
    pIter->hasVal = false;
H
Haojun Liao 已提交
3424 3425
    return NULL;
  }
H
Hongze Cheng 已提交
3426

3427
  // it is a valid data version
dengyihao's avatar
dengyihao 已提交
3428
  if ((key.version <= pReader->verRange.maxVer && key.version >= pReader->verRange.minVer) &&
3429
      (!hasBeenDropped(pDelList, &pIter->index, &key, pReader->order, &pReader->verRange))) {
H
Haojun Liao 已提交
3430 3431
    return pRow;
  }
H
Hongze Cheng 已提交
3432

3433
  while (1) {
3434 3435
    pIter->hasVal = tsdbTbDataIterNext(pIter->iter);
    if (!pIter->hasVal) {
H
Haojun Liao 已提交
3436 3437
      return NULL;
    }
H
Hongze Cheng 已提交
3438

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

H
Haojun Liao 已提交
3441
    key = TSDBROW_KEY(pRow);
3442
    if (outOfTimeWindow(key.ts, &pReader->window)) {
3443
      pIter->hasVal = false;
H
Haojun Liao 已提交
3444 3445
      return NULL;
    }
H
Hongze Cheng 已提交
3446

dengyihao's avatar
dengyihao 已提交
3447
    if (key.version <= pReader->verRange.maxVer && key.version >= pReader->verRange.minVer &&
3448
        (!hasBeenDropped(pDelList, &pIter->index, &key, pReader->order, &pReader->verRange))) {
H
Haojun Liao 已提交
3449 3450 3451 3452
      return pRow;
    }
  }
}
H
Hongze Cheng 已提交
3453

3454 3455
int32_t doMergeRowsInBuf(SIterInfo* pIter, uint64_t uid, int64_t ts, SArray* pDelList, SRowMerger* pMerger,
                         STsdbReader* pReader) {
H
Haojun Liao 已提交
3456
  while (1) {
3457 3458
    pIter->hasVal = tsdbTbDataIterNext(pIter->iter);
    if (!pIter->hasVal) {
H
Haojun Liao 已提交
3459 3460
      break;
    }
H
Hongze Cheng 已提交
3461

3462
    // data exists but not valid
3463
    TSDBROW* pRow = getValidMemRow(pIter, pDelList, pReader);
3464 3465 3466 3467 3468
    if (pRow == NULL) {
      break;
    }

    // ts is not identical, quit
H
Haojun Liao 已提交
3469
    TSDBKEY k = TSDBROW_KEY(pRow);
3470
    if (k.ts != ts) {
H
Haojun Liao 已提交
3471 3472 3473
      break;
    }

3474 3475 3476 3477 3478
    if (pRow->type == TSDBROW_ROW_FMT) {
      STSchema* pTSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(pRow), pReader, uid);
      if (pTSchema == NULL) {
        return terrno;
      }
H
Haojun Liao 已提交
3479

3480 3481 3482 3483
      tsdbRowMergerAdd(pMerger, pRow, pTSchema);
    } else {  // column format
      tsdbRowMerge(pMerger, pRow);
    }
H
Haojun Liao 已提交
3484 3485 3486 3487 3488
  }

  return TSDB_CODE_SUCCESS;
}

3489
static int32_t doMergeRowsInFileBlockImpl(SBlockData* pBlockData, int32_t rowIndex, int64_t key, SRowMerger* pMerger,
3490
                                          SVersionRange* pVerRange, int32_t step) {
3491
  while (rowIndex < pBlockData->nRow && rowIndex >= 0 && pBlockData->aTSKEY[rowIndex] == key) {
3492
    if (pBlockData->aVersion[rowIndex] > pVerRange->maxVer || pBlockData->aVersion[rowIndex] < pVerRange->minVer) {
3493
      rowIndex += step;
3494 3495 3496 3497
      continue;
    }

    TSDBROW fRow = tsdbRowFromBlockData(pBlockData, rowIndex);
H
Hongze Cheng 已提交
3498
    tsdbRowMerge(pMerger, &fRow);
3499 3500 3501 3502 3503 3504 3505 3506 3507 3508 3509
    rowIndex += step;
  }

  return rowIndex;
}

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

H
Hongze Cheng 已提交
3510
static int32_t checkForNeighborFileBlock(STsdbReader* pReader, STableBlockScanInfo* pScanInfo, SDataBlk* pBlock,
3511 3512
                                         SFileDataBlockInfo* pFBlock, SRowMerger* pMerger, int64_t key,
                                         CHECK_FILEBLOCK_STATE* state) {
3513
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
3514
  SBlockData*         pBlockData = &pReader->status.fileBlockData;
3515
  bool    asc = ASCENDING_TRAVERSE(pReader->order);
3516

3517
  *state = CHECK_FILEBLOCK_QUIT;
3518
  int32_t step = ASCENDING_TRAVERSE(pReader->order) ? 1 : -1;
3519

3520
  bool    loadNeighbor = true;
H
Haojun Liao 已提交
3521
  int32_t code = loadNeighborIfOverlap(pFBlock, pScanInfo, pReader, &loadNeighbor);
3522

H
Haojun Liao 已提交
3523
  if (loadNeighbor && (code == TSDB_CODE_SUCCESS)) {
3524 3525
    pDumpInfo->rowIndex =
        doMergeRowsInFileBlockImpl(pBlockData, pDumpInfo->rowIndex, key, pMerger, &pReader->verRange, step);
3526
    if ((pDumpInfo->rowIndex >= pDumpInfo->totalRows && asc) || (pDumpInfo->rowIndex < 0 && !asc)) {
3527 3528 3529 3530
      *state = CHECK_FILEBLOCK_CONT;
    }
  }

H
Haojun Liao 已提交
3531
  return code;
3532 3533
}

3534 3535
int32_t doMergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pScanInfo, STsdbReader* pReader,
                                SRowMerger* pMerger) {
3536 3537
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;

3538
  bool    asc = ASCENDING_TRAVERSE(pReader->order);
3539
  int64_t key = pBlockData->aTSKEY[pDumpInfo->rowIndex];
3540
  int32_t step = asc ? 1 : -1;
3541

3542
  pDumpInfo->rowIndex += step;
3543
  if ((pDumpInfo->rowIndex <= pBlockData->nRow - 1 && asc) || (pDumpInfo->rowIndex >= 0 && !asc)) {
3544 3545 3546
    pDumpInfo->rowIndex =
        doMergeRowsInFileBlockImpl(pBlockData, pDumpInfo->rowIndex, key, pMerger, &pReader->verRange, step);
  }
3547

3548 3549 3550 3551
  // 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;
3552

3553
      SFileDataBlockInfo* pFileBlockInfo = getCurrentBlockInfo(&pReader->status.blockIter);
H
Hongze Cheng 已提交
3554
      SDataBlk*           pCurrentBlock = getCurrentBlock(&pReader->status.blockIter);
H
Haojun Liao 已提交
3555 3556 3557 3558 3559
      if (pFileBlockInfo == NULL) {
        st = CHECK_FILEBLOCK_QUIT;
        break;
      }

3560 3561 3562
      checkForNeighborFileBlock(pReader, pScanInfo, pCurrentBlock, pFileBlockInfo, pMerger, key, &st);
      if (st == CHECK_FILEBLOCK_QUIT) {
        break;
3563
      }
3564
    }
H
Haojun Liao 已提交
3565
  }
3566

H
Haojun Liao 已提交
3567 3568 3569
  return TSDB_CODE_SUCCESS;
}

H
Hongze Cheng 已提交
3570
int32_t doMergeRowsInLastBlock(SLastBlockReader* pLastBlockReader, STableBlockScanInfo* pScanInfo, int64_t ts,
3571
                               SRowMerger* pMerger, SVersionRange* pVerRange, const char* idStr) {
3572
  while (nextRowFromLastBlocks(pLastBlockReader, pScanInfo, pVerRange)) {
3573 3574
    int64_t next1 = getCurrentKeyInLastBlock(pLastBlockReader);
    if (next1 == ts) {
3575
      TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
H
Hongze Cheng 已提交
3576
      tsdbRowMerge(pMerger, &fRow1);
3577
    } else {
3578 3579 3580
      tsdbTrace("uid:%" PRIu64 " last del index:%d, del range:%d, lastKeyInStt:%" PRId64 ", %s", pScanInfo->uid,
                pScanInfo->lastBlockDelIndex, (int32_t)taosArrayGetSize(pScanInfo->delSkyline), pScanInfo->lastKeyInStt,
                idStr);
3581 3582 3583 3584 3585 3586 3587
      break;
    }
  }

  return TSDB_CODE_SUCCESS;
}

3588
int32_t doMergeMemTableMultiRows(TSDBROW* pRow, uint64_t uid, SIterInfo* pIter, SArray* pDelList, TSDBROW* pResRow,
3589
                                 STsdbReader* pReader, bool* freeTSRow) {
H
Haojun Liao 已提交
3590
  TSDBROW* pNextRow = NULL;
3591
  TSDBROW  current = *pRow;
3592

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

3596
    if (!pIter->hasVal) {
3597
      *pResRow = *pRow;
3598
      *freeTSRow = false;
3599
      return TSDB_CODE_SUCCESS;
3600
    } else {  // has next point in mem/imem
3601
      pNextRow = getValidMemRow(pIter, pDelList, pReader);
3602
      if (pNextRow == NULL) {
H
Haojun Liao 已提交
3603
        *pResRow = current;
3604
        *freeTSRow = false;
3605
        return TSDB_CODE_SUCCESS;
3606 3607
      }

H
Hongze Cheng 已提交
3608
      if (TSDBROW_TS(&current) != TSDBROW_TS(pNextRow)) {
H
Haojun Liao 已提交
3609
        *pResRow = current;
3610
        *freeTSRow = false;
3611
        return TSDB_CODE_SUCCESS;
3612
      }
3613
    }
3614 3615
  }

3616
  SRowMerger merge = {0};
H
Haojun Liao 已提交
3617
  terrno = 0;
3618
  int32_t code = 0;
H
Haojun Liao 已提交
3619

3620 3621 3622 3623 3624 3625 3626
  // start to merge duplicated rows
  if (current.type == TSDBROW_ROW_FMT) {
    // get the correct schema for data in memory
    STSchema* pTSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(&current), pReader, uid);
    if (pTSchema == NULL) {
      return terrno;
    }
H
Haojun Liao 已提交
3627

3628 3629 3630
    if (pReader->pSchema == NULL) {
      pReader->pSchema = pTSchema;
    }
H
Haojun Liao 已提交
3631

3632
    code = tsdbRowMergerInit2(&merge, pReader->pSchema, &current, pTSchema);
3633 3634 3635
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
H
Haojun Liao 已提交
3636

3637 3638 3639 3640
    STSchema* pTSchema1 = doGetSchemaForTSRow(TSDBROW_SVERSION(pNextRow), pReader, uid);
    if (pTSchema1 == NULL) {
      return terrno;
    }
H
Haojun Liao 已提交
3641

3642
    tsdbRowMergerAdd(&merge, pNextRow, pTSchema1);
3643 3644
  } else {  // let's merge rows in file block
    code = tsdbRowMergerInit(&merge, &current, pReader->pSchema);
3645 3646 3647
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
H
Haojun Liao 已提交
3648

3649 3650
    tsdbRowMerge(&merge, pNextRow);
  }
H
Haojun Liao 已提交
3651

wmmhello's avatar
wmmhello 已提交
3652
  code = doMergeRowsInBuf(pIter, uid, TSDBROW_TS(&current), pDelList, &merge, pReader);
H
Haojun Liao 已提交
3653 3654 3655 3656
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

3657
  code = tsdbRowMergerGetRow(&merge, &pResRow->pTSRow);
3658 3659 3660
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }
M
Minglei Jin 已提交
3661

wmmhello's avatar
wmmhello 已提交
3662
  pResRow->type = TSDBROW_ROW_FMT;
3663
  tsdbRowMergerClear(&merge);
3664
  *freeTSRow = true;
3665

3666
  return TSDB_CODE_SUCCESS;
3667 3668
}

3669
int32_t doMergeMemIMemRows(TSDBROW* pRow, TSDBROW* piRow, STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader,
H
Hongze Cheng 已提交
3670
                           SRow** pTSRow) {
H
Haojun Liao 已提交
3671 3672
  SRowMerger merge = {0};

3673 3674 3675
  TSDBKEY k = TSDBROW_KEY(pRow);
  TSDBKEY ik = TSDBROW_KEY(piRow);

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

H
Hongze Cheng 已提交
3679
    int32_t code = tsdbRowMergerInit(&merge, piRow, pSchema);
H
Haojun Liao 已提交
3680 3681 3682 3683 3684 3685 3686 3687 3688
    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;
    }
3689

H
Hongze Cheng 已提交
3690
    tsdbRowMerge(&merge, pRow);
H
Haojun Liao 已提交
3691 3692 3693 3694 3695 3696
    code =
        doMergeRowsInBuf(&pBlockScanInfo->iter, pBlockScanInfo->uid, k.ts, pBlockScanInfo->delSkyline, &merge, pReader);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

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

H
Hongze Cheng 已提交
3700
    int32_t code = tsdbRowMergerInit(&merge, pRow, pSchema);
3701
    if (code != TSDB_CODE_SUCCESS || merge.pTSchema == NULL) {
H
Haojun Liao 已提交
3702 3703 3704 3705 3706 3707 3708 3709
      return code;
    }

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

H
Hongze Cheng 已提交
3711
    tsdbRowMerge(&merge, piRow);
H
Haojun Liao 已提交
3712 3713 3714 3715 3716
    code = doMergeRowsInBuf(&pBlockScanInfo->iiter, pBlockScanInfo->uid, ik.ts, pBlockScanInfo->delSkyline, &merge,
                            pReader);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
3717
  }
3718

H
Haojun Liao 已提交
3719 3720
  int32_t code = tsdbRowMergerGetRow(&merge, pTSRow);
  tsdbRowMergerClear(&merge);
3721
  return code;
3722 3723
}

3724
int32_t tsdbGetNextRowInMem(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader, TSDBROW* pResRow, int64_t endKey,
3725
                            bool* freeTSRow) {
3726 3727
  TSDBROW* pRow = getValidMemRow(&pBlockScanInfo->iter, pBlockScanInfo->delSkyline, pReader);
  TSDBROW* piRow = getValidMemRow(&pBlockScanInfo->iiter, pBlockScanInfo->delSkyline, pReader);
dengyihao's avatar
dengyihao 已提交
3728
  SArray*  pDelList = pBlockScanInfo->delSkyline;
3729
  uint64_t uid = pBlockScanInfo->uid;
H
Haojun Liao 已提交
3730

3731 3732
  // todo refactor
  bool asc = ASCENDING_TRAVERSE(pReader->order);
3733
  if (pBlockScanInfo->iter.hasVal) {
3734 3735 3736 3737 3738 3739
    TSDBKEY k = TSDBROW_KEY(pRow);
    if ((k.ts >= endKey && asc) || (k.ts <= endKey && !asc)) {
      pRow = NULL;
    }
  }

3740
  if (pBlockScanInfo->iiter.hasVal) {
3741 3742 3743 3744 3745 3746
    TSDBKEY k = TSDBROW_KEY(piRow);
    if ((k.ts >= endKey && asc) || (k.ts <= endKey && !asc)) {
      piRow = NULL;
    }
  }

3747
  if (pBlockScanInfo->iter.hasVal && pBlockScanInfo->iiter.hasVal && pRow != NULL && piRow != NULL) {
3748
    TSDBKEY k = TSDBROW_KEY(pRow);
3749
    TSDBKEY ik = TSDBROW_KEY(piRow);
H
Haojun Liao 已提交
3750

3751
    int32_t code = TSDB_CODE_SUCCESS;
3752 3753
    if (ik.ts != k.ts) {
      if (((ik.ts < k.ts) && asc) || ((ik.ts > k.ts) && (!asc))) {  // ik.ts < k.ts
3754
        code = doMergeMemTableMultiRows(piRow, uid, &pBlockScanInfo->iiter, pDelList, pResRow, pReader, freeTSRow);
3755
      } else if (((k.ts < ik.ts) && asc) || ((k.ts > ik.ts) && (!asc))) {
3756
        code = doMergeMemTableMultiRows(pRow, uid, &pBlockScanInfo->iter, pDelList, pResRow, pReader, freeTSRow);
3757
      }
3758
    } else {  // ik.ts == k.ts
3759
      *freeTSRow = true;
3760 3761
      pResRow->type = TSDBROW_ROW_FMT;
      code = doMergeMemIMemRows(pRow, piRow, pBlockScanInfo, pReader, &pResRow->pTSRow);
3762 3763 3764
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
H
Haojun Liao 已提交
3765
    }
3766

3767
    return code;
H
Haojun Liao 已提交
3768 3769
  }

3770
  if (pBlockScanInfo->iter.hasVal && pRow != NULL) {
3771
    return doMergeMemTableMultiRows(pRow, pBlockScanInfo->uid, &pBlockScanInfo->iter, pDelList, pResRow, pReader,
H
Hongze Cheng 已提交
3772
                                    freeTSRow);
H
Haojun Liao 已提交
3773 3774
  }

3775
  if (pBlockScanInfo->iiter.hasVal && piRow != NULL) {
3776
    return doMergeMemTableMultiRows(piRow, uid, &pBlockScanInfo->iiter, pDelList, pResRow, pReader, freeTSRow);
H
Haojun Liao 已提交
3777 3778 3779 3780 3781
  }

  return TSDB_CODE_SUCCESS;
}

H
Hongze Cheng 已提交
3782
int32_t doAppendRowFromTSRow(SSDataBlock* pBlock, STsdbReader* pReader, SRow* pTSRow, STableBlockScanInfo* pScanInfo) {
H
Haojun Liao 已提交
3783
  int32_t outputRowIndex = pBlock->info.rows;
3784
  int64_t uid = pScanInfo->uid;
3785 3786 3787

  int32_t numOfCols = (int32_t)taosArrayGetSize(pBlock->pDataBlock);

3788
  SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo;
3789
  STSchema*           pSchema = doGetSchemaForTSRow(pTSRow->sver, pReader, uid);
3790

3791
  SColVal colVal = {0};
3792
  int32_t i = 0, j = 0;
H
Haojun Liao 已提交
3793

3794
  if (pSupInfo->colId[i] == PRIMARYKEY_TIMESTAMP_COL_ID) {
H
Haojun Liao 已提交
3795
    SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, pSupInfo->slotId[i]);
H
Haojun Liao 已提交
3796
    ((int64_t*)pColData->pData)[outputRowIndex] = pTSRow->ts;
3797 3798 3799
    i += 1;
  }

H
Haojun Liao 已提交
3800
  while (i < pSupInfo->numOfCols && j < pSchema->numOfCols) {
H
Haojun Liao 已提交
3801
    col_id_t colId = pSupInfo->colId[i];
3802 3803

    if (colId == pSchema->columns[j].colId) {
H
Haojun Liao 已提交
3804
      SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pSupInfo->slotId[i]);
H
Haojun Liao 已提交
3805

H
Hongze Cheng 已提交
3806
      tRowGet(pTSRow, pSchema, j, &colVal);
H
Haojun Liao 已提交
3807
      doCopyColVal(pColInfoData, outputRowIndex, i, &colVal, pSupInfo);
3808 3809 3810
      i += 1;
      j += 1;
    } else if (colId < pSchema->columns[j].colId) {
H
Haojun Liao 已提交
3811
      SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pSupInfo->slotId[i]);
H
Haojun Liao 已提交
3812

3813
      colDataSetNULL(pColInfoData, outputRowIndex);
3814 3815 3816
      i += 1;
    } else if (colId > pSchema->columns[j].colId) {
      j += 1;
3817
    }
3818 3819
  }

3820
  // set null value since current column does not exist in the "pSchema"
H
Haojun Liao 已提交
3821
  while (i < pSupInfo->numOfCols) {
H
Haojun Liao 已提交
3822
    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pSupInfo->slotId[i]);
3823
    colDataSetNULL(pColInfoData, outputRowIndex);
3824 3825 3826
    i += 1;
  }

3827
  pBlock->info.dataLoad = 1;
3828
  pBlock->info.rows += 1;
3829
  pScanInfo->lastKey = pTSRow->ts;
3830 3831 3832
  return TSDB_CODE_SUCCESS;
}

H
Hongze Cheng 已提交
3833 3834
int32_t doAppendRowFromFileBlock(SSDataBlock* pResBlock, STsdbReader* pReader, SBlockData* pBlockData,
                                 int32_t rowIndex) {
3835 3836 3837 3838
  int32_t i = 0, j = 0;
  int32_t outputRowIndex = pResBlock->info.rows;

  SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo;
3839
  if (pReader->suppInfo.colId[i] == PRIMARYKEY_TIMESTAMP_COL_ID) {
H
Haojun Liao 已提交
3840
    SColumnInfoData* pColData = taosArrayGet(pResBlock->pDataBlock, pSupInfo->slotId[i]);
H
Haojun Liao 已提交
3841
    ((int64_t*)pColData->pData)[outputRowIndex] = pBlockData->aTSKEY[rowIndex];
3842
    i += 1;
3843 3844 3845
  }

  SColVal cv = {0};
H
Hongze Cheng 已提交
3846
  int32_t numOfInputCols = pBlockData->nColData;
H
Haojun Liao 已提交
3847
  int32_t numOfOutputCols = pSupInfo->numOfCols;
3848

3849
  while (i < numOfOutputCols && j < numOfInputCols) {
H
Haojun Liao 已提交
3850
    SColData* pData = tBlockDataGetColDataByIdx(pBlockData, j);
H
Haojun Liao 已提交
3851
    if (pData->cid < pSupInfo->colId[i]) {
3852 3853 3854 3855
      j += 1;
      continue;
    }

H
Haojun Liao 已提交
3856 3857
    SColumnInfoData* pCol = TARRAY_GET_ELEM(pResBlock->pDataBlock, pSupInfo->slotId[i]);
    if (pData->cid == pSupInfo->colId[i]) {
3858 3859
      tColDataGetValue(pData, rowIndex, &cv);
      doCopyColVal(pCol, outputRowIndex, i, &cv, pSupInfo);
3860
      j += 1;
H
Haojun Liao 已提交
3861 3862
    } else if (pData->cid > pCol->info.colId) {
      // the specified column does not exist in file block, fill with null data
3863
      colDataSetNULL(pCol, outputRowIndex);
3864 3865 3866 3867 3868 3869
    }

    i += 1;
  }

  while (i < numOfOutputCols) {
H
Haojun Liao 已提交
3870
    SColumnInfoData* pCol = taosArrayGet(pResBlock->pDataBlock, pSupInfo->slotId[i]);
3871
    colDataSetNULL(pCol, outputRowIndex);
3872 3873 3874
    i += 1;
  }

3875
  pResBlock->info.dataLoad = 1;
3876 3877 3878 3879
  pResBlock->info.rows += 1;
  return TSDB_CODE_SUCCESS;
}

3880 3881
int32_t buildDataBlockFromBufImpl(STableBlockScanInfo* pBlockScanInfo, int64_t endKey, int32_t capacity,
                                  STsdbReader* pReader) {
H
Haojun Liao 已提交
3882 3883 3884
  SSDataBlock* pBlock = pReader->pResBlock;

  do {
3885
    //    SRow* pTSRow = NULL;
3886
    TSDBROW row = {.type = -1};
3887
    bool    freeTSRow = false;
3888 3889
    tsdbGetNextRowInMem(pBlockScanInfo, pReader, &row, endKey, &freeTSRow);
    if (row.type == -1) {
3890
      break;
H
Haojun Liao 已提交
3891 3892
    }

3893 3894
    if (row.type == TSDBROW_ROW_FMT) {
      doAppendRowFromTSRow(pBlock, pReader, row.pTSRow, pBlockScanInfo);
3895

3896 3897 3898 3899 3900
      if (freeTSRow) {
        taosMemoryFree(row.pTSRow);
      }
    } else {
      doAppendRowFromFileBlock(pBlock, pReader, row.pBlockData, row.iRow);
3901
    }
H
Haojun Liao 已提交
3902 3903

    // no data in buffer, return immediately
3904
    if (!(pBlockScanInfo->iter.hasVal || pBlockScanInfo->iiter.hasVal)) {
H
Haojun Liao 已提交
3905 3906 3907
      break;
    }

3908
    if (pBlock->info.rows >= capacity) {
H
Haojun Liao 已提交
3909 3910 3911 3912 3913 3914
      break;
    }
  } while (1);

  return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
3915

3916 3917
// TODO refactor: with createDataBlockScanInfo
int32_t tsdbSetTableList(STsdbReader* pReader, const void* pTableList, int32_t num) {
3918
  int32_t size = taosHashGetSize(pReader->status.pTableMap);
3919

3920
  STableBlockScanInfo** p = NULL;
3921
  while ((p = taosHashIterate(pReader->status.pTableMap, p)) != NULL) {
3922
    clearBlockScanInfo(*p);
3923 3924
  }

D
dapan1121 已提交
3925 3926 3927 3928 3929
  if (size < num) {
    int32_t code = ensureBlockScanInfoBuf(&pReader->blockInfoBuf, num);
    if (code) {
      return code;
    }
D
dapan1121 已提交
3930
    pReader->status.uidList.tableUidList = (uint64_t*)taosMemoryRealloc(pReader->status.uidList.tableUidList, sizeof(uint64_t) * num);
D
dapan1121 已提交
3931
  }
3932

3933
  taosHashClear(pReader->status.pTableMap);
3934
  STableUidList* pUidList = &pReader->status.uidList;
H
Haojun Liao 已提交
3935
  pUidList->currentIndex = 0;
3936

3937 3938
  STableKeyInfo* pList = (STableKeyInfo*)pTableList;
  for (int32_t i = 0; i < num; ++i) {
3939 3940
    STableBlockScanInfo* pInfo = getPosInBlockInfoBuf(&pReader->blockInfoBuf, i);
    pInfo->uid = pList[i].uid;
H
Haojun Liao 已提交
3941 3942
    pUidList->tableUidList[i] = pList[i].uid;

3943 3944 3945 3946 3947 3948 3949 3950 3951 3952 3953
    // todo extract method
    if (ASCENDING_TRAVERSE(pReader->order)) {
      int64_t skey = pReader->window.skey;
      pInfo->lastKey = (skey > INT64_MIN) ? (skey - 1) : skey;
      pInfo->lastKeyInStt = skey;
    } else {
      int64_t ekey = pReader->window.ekey;
      pInfo->lastKey = (ekey < INT64_MAX) ? (ekey + 1) : ekey;
      pInfo->lastKeyInStt = ekey;
    }

3954
    taosHashPut(pReader->status.pTableMap, &pInfo->uid, sizeof(uint64_t), &pInfo, POINTER_BYTES);
3955 3956
  }

H
Hongze Cheng 已提交
3957 3958 3959
  return TDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
3960 3961 3962 3963 3964 3965
void* tsdbGetIdx(SMeta* pMeta) {
  if (pMeta == NULL) {
    return NULL;
  }
  return metaGetIdx(pMeta);
}
dengyihao's avatar
dengyihao 已提交
3966

dengyihao's avatar
dengyihao 已提交
3967 3968 3969 3970 3971 3972
void* tsdbGetIvtIdx(SMeta* pMeta) {
  if (pMeta == NULL) {
    return NULL;
  }
  return metaGetIvtIdx(pMeta);
}
L
Liu Jicong 已提交
3973

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

3976
static int32_t doOpenReaderImpl(STsdbReader* pReader) {
3977 3978
  SReaderStatus*  pStatus = &pReader->status;
  SDataBlockIter* pBlockIter = &pStatus->blockIter;
3979

3980 3981
  initFilesetIterator(&pStatus->fileIter, pReader->pReadSnap->fs.aDFileSet, pReader);
  resetDataBlockIterator(&pStatus->blockIter, pReader->order);
3982

3983 3984 3985
  int32_t code = TSDB_CODE_SUCCESS;
  if (pStatus->fileIter.numOfFiles == 0) {
    pStatus->loadFromFile = false;
3986
  } else {
3987
    code = initForFirstBlockInFile(pReader, pBlockIter);
3988
  }
3989 3990 3991

  if (!pStatus->loadFromFile) {
    resetTableListIndex(pStatus);
3992
  }
3993 3994

  return code;
3995 3996
}

H
refact  
Hongze Cheng 已提交
3997
// ====================================== EXPOSED APIs ======================================
3998
int32_t tsdbReaderOpen(SVnode* pVnode, SQueryTableDataCond* pCond, void* pTableList, int32_t numOfTables,
H
Haojun Liao 已提交
3999
                       SSDataBlock* pResBlock, STsdbReader** ppReader, const char* idstr) {
4000 4001 4002 4003 4004 4005
  STimeWindow window = pCond->twindows;
  if (pCond->type == TIMEWINDOW_RANGE_EXTERNAL) {
    pCond->twindows.skey += 1;
    pCond->twindows.ekey -= 1;
  }

4006 4007 4008
  int32_t capacity = pVnode->config.tsdbCfg.maxRows;
  if (pResBlock != NULL) {
    blockDataEnsureCapacity(pResBlock, capacity);
H
Haojun Liao 已提交
4009 4010 4011
  }

  int32_t code = tsdbReaderCreate(pVnode, pCond, ppReader, capacity, pResBlock, idstr);
4012
  if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
4013 4014
    goto _err;
  }
H
Hongze Cheng 已提交
4015

4016
  // check for query time window
H
Haojun Liao 已提交
4017
  STsdbReader* pReader = *ppReader;
4018
  if (isEmptyQueryTimeWindow(&pReader->window) && pCond->type == TIMEWINDOW_RANGE_CONTAINED) {
H
Haojun Liao 已提交
4019 4020 4021
    tsdbDebug("%p query window not overlaps with the data set, no result returned, %s", pReader, pReader->idStr);
    return TSDB_CODE_SUCCESS;
  }
H
Hongze Cheng 已提交
4022

4023 4024
  if (pCond->type == TIMEWINDOW_RANGE_EXTERNAL) {
    // update the SQueryTableDataCond to create inner reader
4025
    int32_t order = pCond->order;
4026
    if (order == TSDB_ORDER_ASC) {
4027
      pCond->twindows.ekey = window.skey;
4028 4029 4030
      pCond->twindows.skey = INT64_MIN;
      pCond->order = TSDB_ORDER_DESC;
    } else {
4031
      pCond->twindows.skey = window.ekey;
4032 4033 4034 4035
      pCond->twindows.ekey = INT64_MAX;
      pCond->order = TSDB_ORDER_ASC;
    }

4036
    // here we only need one more row, so the capacity is set to be ONE.
H
Haojun Liao 已提交
4037
    code = tsdbReaderCreate(pVnode, pCond, &pReader->innerReader[0], 1, pResBlock, idstr);
4038 4039 4040 4041 4042
    if (code != TSDB_CODE_SUCCESS) {
      goto _err;
    }

    if (order == TSDB_ORDER_ASC) {
4043
      pCond->twindows.skey = window.ekey;
4044
      pCond->twindows.ekey = INT64_MAX;
4045
    } else {
4046
      pCond->twindows.skey = INT64_MIN;
4047
      pCond->twindows.ekey = window.ekey;
4048
    }
4049 4050
    pCond->order = order;

H
Haojun Liao 已提交
4051
    code = tsdbReaderCreate(pVnode, pCond, &pReader->innerReader[1], 1, pResBlock, idstr);
4052 4053 4054 4055 4056
    if (code != TSDB_CODE_SUCCESS) {
      goto _err;
    }
  }

H
Haojun Liao 已提交
4057
  // NOTE: the endVersion in pCond is the data version not schema version, so pCond->endVersion is not correct here.
4058 4059
  //  no valid error code set in metaGetTbTSchema, so let's set the error code here.
  //  we should proceed in case of tmq processing.
4060
  if (pCond->suid != 0) {
4061
    pReader->pSchema = metaGetTbTSchema(pReader->pTsdb->pVnode->pMeta, pReader->suid, -1, 1);
H
Haojun Liao 已提交
4062
    if (pReader->pSchema == NULL) {
H
Haojun Liao 已提交
4063
      tsdbError("failed to get table schema, suid:%" PRIu64 ", ver:-1, %s", pReader->suid, pReader->idStr);
H
Haojun Liao 已提交
4064
    }
4065 4066
  } else if (numOfTables > 0) {
    STableKeyInfo* pKey = pTableList;
4067
    pReader->pSchema = metaGetTbTSchema(pReader->pTsdb->pVnode->pMeta, pKey->uid, -1, 1);
H
Haojun Liao 已提交
4068
    if (pReader->pSchema == NULL) {
H
Haojun Liao 已提交
4069
      tsdbError("failed to get table schema, uid:%" PRIu64 ", ver:-1, %s", pKey->uid, pReader->idStr);
H
Haojun Liao 已提交
4070
    }
4071 4072
  }

4073
  if (pReader->pSchema != NULL) {
H
Haojun Liao 已提交
4074 4075 4076 4077
    code = updateBlockSMAInfo(pReader->pSchema, &pReader->suppInfo);
    if (code != TSDB_CODE_SUCCESS) {
      goto _err;
    }
4078
  }
4079

4080
  STsdbReader* p = (pReader->innerReader[0] != NULL) ? pReader->innerReader[0] : pReader;
X
Xiaoyu Wang 已提交
4081 4082
  pReader->status.pTableMap =
      createDataBlockScanInfo(p, &pReader->blockInfoBuf, pTableList, &pReader->status.uidList, numOfTables);
H
Haojun Liao 已提交
4083 4084
  if (pReader->status.pTableMap == NULL) {
    *ppReader = NULL;
S
Shengliang Guan 已提交
4085
    code = TSDB_CODE_OUT_OF_MEMORY;
H
Haojun Liao 已提交
4086 4087
    goto _err;
  }
H
Hongze Cheng 已提交
4088

4089
  pReader->suspended = true;
4090

4091
  tsdbDebug("%p total numOfTable:%d in this query %s", pReader, numOfTables, pReader->idStr);
H
Hongze Cheng 已提交
4092
  return code;
H
Hongze Cheng 已提交
4093 4094

_err:
H
Haojun Liao 已提交
4095
  tsdbError("failed to create data reader, code:%s %s", tstrerror(code), idstr);
4096
  tsdbReaderClose(pReader);
X
Xiaoyu Wang 已提交
4097
  *ppReader = NULL;  // reset the pointer value.
H
Hongze Cheng 已提交
4098
  return code;
H
refact  
Hongze Cheng 已提交
4099 4100 4101
}

void tsdbReaderClose(STsdbReader* pReader) {
4102 4103
  if (pReader == NULL) {
    return;
4104
  }
H
refact  
Hongze Cheng 已提交
4105

4106
  tsdbAcquireReader(pReader);
4107
  {
H
Haojun Liao 已提交
4108
    if (pReader->innerReader[0] != NULL || pReader->innerReader[1] != NULL) {
4109
      STsdbReader* p = pReader->innerReader[0];
4110

4111
      p->status.pTableMap = NULL;
H
Haojun Liao 已提交
4112
      p->status.uidList.tableUidList = NULL;
4113 4114 4115 4116 4117 4118 4119
      p->pReadSnap = NULL;
      p->pSchema = NULL;
      p->pMemSchema = NULL;

      p = pReader->innerReader[1];

      p->status.pTableMap = NULL;
H
Haojun Liao 已提交
4120
      p->status.uidList.tableUidList = NULL;
4121 4122 4123
      p->pReadSnap = NULL;
      p->pSchema = NULL;
      p->pMemSchema = NULL;
4124 4125 4126 4127 4128 4129

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

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

4132
  taosArrayDestroy(pSupInfo->pColAgg);
H
Haojun Liao 已提交
4133
  for (int32_t i = 0; i < pSupInfo->numOfCols; ++i) {
4134 4135 4136 4137
    if (pSupInfo->buildBuf[i] != NULL) {
      taosMemoryFreeClear(pSupInfo->buildBuf[i]);
    }
  }
4138

H
Haojun Liao 已提交
4139 4140 4141
  if (pReader->freeBlock) {
    pReader->pResBlock = blockDataDestroy(pReader->pResBlock);
  }
4142

H
Haojun Liao 已提交
4143
  taosMemoryFree(pSupInfo->colId);
H
Hongze Cheng 已提交
4144
  tBlockDataDestroy(&pReader->status.fileBlockData);
4145
  cleanupDataBlockIterator(&pReader->status.blockIter);
4146 4147

  size_t numOfTables = taosHashGetSize(pReader->status.pTableMap);
H
Haojun Liao 已提交
4148 4149 4150 4151
  if (pReader->status.pTableMap != NULL) {
    destroyAllBlockScanInfo(pReader->status.pTableMap);
    clearBlockScanInfoBuf(&pReader->blockInfoBuf);
  }
4152

H
Haojun Liao 已提交
4153 4154 4155
  if (pReader->pFileReader != NULL) {
    tsdbDataFReaderClose(&pReader->pFileReader);
  }
H
refact  
Hongze Cheng 已提交
4156

4157 4158 4159 4160 4161 4162 4163 4164 4165
  if (pReader->pDelFReader != NULL) {
    tsdbDelFReaderClose(&pReader->pDelFReader);
  }

  if (pReader->pDelIdx != NULL) {
    taosArrayDestroy(pReader->pDelIdx);
    pReader->pDelIdx = NULL;
  }

4166
  qTrace("tsdb/reader-close: %p, untake snapshot", pReader);
4167
  tsdbUntakeReadSnap(pReader, pReader->pReadSnap, true);
4168
  pReader->pReadSnap = NULL;
4169

4170 4171
  tsdbReleaseReader(pReader);

4172
  tsdbUninitReaderLock(pReader);
4173

4174
  taosMemoryFree(pReader->status.uidList.tableUidList);
H
Haojun Liao 已提交
4175
  SIOCostSummary* pCost = &pReader->cost;
4176

H
Haojun Liao 已提交
4177 4178
  SFilesetIter* pFilesetIter = &pReader->status.fileIter;
  if (pFilesetIter->pLastBlockReader != NULL) {
H
Haojun Liao 已提交
4179 4180
    SLastBlockReader* pLReader = pFilesetIter->pLastBlockReader;
    tMergeTreeClose(&pLReader->mergeTree);
H
Haojun Liao 已提交
4181

H
Haojun Liao 已提交
4182
    getLastBlockLoadInfo(pLReader->pInfo, &pCost->lastBlockLoad, &pCost->lastBlockLoadTime);
H
refact  
Hongze Cheng 已提交
4183

H
Haojun Liao 已提交
4184 4185 4186
    pLReader->pInfo = destroyLastBlockLoadInfo(pLReader->pInfo);
    taosMemoryFree(pLReader);
  }
H
refact  
Hongze Cheng 已提交
4187

4188 4189 4190 4191 4192
  tsdbDebug(
      "%p :io-cost summary: head-file:%" PRIu64 ", head-file time:%.2f ms, SMA:%" PRId64
      " SMA-time:%.2f ms, fileBlocks:%" PRId64
      ", fileBlocks-load-time:%.2f ms, "
      "build in-memory-block-time:%.2f ms, lastBlocks:%" PRId64 ", lastBlocks-time:%.2f ms, composed-blocks:%" PRId64
X
Xiaoyu Wang 已提交
4193 4194
      ", composed-blocks-time:%.2fms, STableBlockScanInfo size:%.2f Kb, createTime:%.2f ms,initDelSkylineIterTime:%.2f "
      "ms, %s",
4195 4196 4197
      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, pCost->createScanInfoList,
H
Haojun Liao 已提交
4198
      pCost->initDelSkylineIterTime, pReader->idStr);
H
refact  
Hongze Cheng 已提交
4199

4200 4201
  taosMemoryFree(pReader->idStr);
  taosMemoryFree(pReader->pSchema);
4202

4203 4204 4205
  if (pReader->pMemSchema != pReader->pSchema) {
    taosMemoryFree(pReader->pMemSchema);
  }
4206

4207
  taosMemoryFreeClear(pReader);
H
refact  
Hongze Cheng 已提交
4208 4209
}

4210 4211 4212 4213 4214 4215 4216 4217 4218 4219
int32_t tsdbReaderSuspend(STsdbReader* pReader) {
  int32_t code = 0;

  // save reader's base state & reset top state to be reconstructed from base state
  SReaderStatus*       pStatus = &pReader->status;
  STableBlockScanInfo* pBlockScanInfo = NULL;

  if (pStatus->loadFromFile) {
    SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(&pReader->status.blockIter);
    if (pBlockInfo != NULL) {
4220 4221
      pBlockScanInfo =
          *(STableBlockScanInfo**)taosHashGet(pStatus->pTableMap, &pBlockInfo->uid, sizeof(pBlockInfo->uid));
4222 4223 4224 4225 4226 4227 4228
      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 _err;
      }
    } else {
4229
      pBlockScanInfo = *pStatus->pTableIter;
4230 4231 4232 4233 4234
    }

    tsdbDataFReaderClose(&pReader->pFileReader);

    // resetDataBlockScanInfo excluding lastKey
4235
    STableBlockScanInfo** p = NULL;
4236 4237

    while ((p = taosHashIterate(pStatus->pTableMap, p)) != NULL) {
4238 4239 4240 4241 4242 4243 4244 4245
      STableBlockScanInfo* pInfo = *(STableBlockScanInfo**)p;

      pInfo->iterInit = false;
      pInfo->iter.hasVal = false;
      pInfo->iiter.hasVal = false;

      if (pInfo->iter.iter != NULL) {
        pInfo->iter.iter = tsdbTbDataIterDestroy(pInfo->iter.iter);
4246 4247
      }

4248 4249 4250 4251 4252 4253
      if (pInfo->iiter.iter != NULL) {
        pInfo->iiter.iter = tsdbTbDataIterDestroy(pInfo->iiter.iter);
      }

      pInfo->delSkyline = taosArrayDestroy(pInfo->delSkyline);
      // pInfo->lastKey = ts;
4254 4255
    }
  } else {
4256 4257 4258 4259 4260 4261 4262 4263 4264 4265 4266 4267 4268 4269 4270 4271 4272 4273 4274 4275 4276 4277
    // resetDataBlockScanInfo excluding lastKey
    STableBlockScanInfo** p = NULL;

    while ((p = taosHashIterate(pStatus->pTableMap, p)) != NULL) {
      STableBlockScanInfo* pInfo = *(STableBlockScanInfo**)p;

      pInfo->iterInit = false;
      pInfo->iter.hasVal = false;
      pInfo->iiter.hasVal = false;

      if (pInfo->iter.iter != NULL) {
        pInfo->iter.iter = tsdbTbDataIterDestroy(pInfo->iter.iter);
      }

      if (pInfo->iiter.iter != NULL) {
        pInfo->iiter.iter = tsdbTbDataIterDestroy(pInfo->iiter.iter);
      }

      pInfo->delSkyline = taosArrayDestroy(pInfo->delSkyline);
      // pInfo->lastKey = ts;
    }

4278
    pBlockScanInfo = pStatus->pTableIter == NULL ? NULL : *pStatus->pTableIter;
4279 4280 4281 4282 4283 4284 4285
    if (pBlockScanInfo) {
      // save lastKey to restore memory iterator
      STimeWindow w = pReader->pResBlock->info.window;
      pBlockScanInfo->lastKey = ASCENDING_TRAVERSE(pReader->order) ? w.ekey : w.skey;

      // reset current current table's data block scan info,
      pBlockScanInfo->iterInit = false;
4286 4287
      pBlockScanInfo->iter.hasVal = false;
      pBlockScanInfo->iiter.hasVal = false;
4288 4289 4290 4291 4292 4293 4294 4295 4296 4297 4298 4299 4300 4301 4302
      if (pBlockScanInfo->iter.iter != NULL) {
        pBlockScanInfo->iter.iter = tsdbTbDataIterDestroy(pBlockScanInfo->iter.iter);
      }

      if (pBlockScanInfo->iiter.iter != NULL) {
        pBlockScanInfo->iiter.iter = tsdbTbDataIterDestroy(pBlockScanInfo->iiter.iter);
      }

      pBlockScanInfo->pBlockList = taosArrayDestroy(pBlockScanInfo->pBlockList);
      tMapDataClear(&pBlockScanInfo->mapData);
      // TODO: keep skyline for reuse
      pBlockScanInfo->delSkyline = taosArrayDestroy(pBlockScanInfo->delSkyline);
    }
  }

4303
  tsdbUntakeReadSnap(pReader, pReader->pReadSnap, false);
4304
  pReader->pReadSnap = NULL;
4305 4306 4307

  pReader->suspended = true;

4308 4309
  tsdbDebug("reader: %p suspended uid %" PRIu64 " in this query %s", pReader, pBlockScanInfo ? pBlockScanInfo->uid : 0,
            pReader->idStr);
4310 4311 4312 4313 4314 4315 4316 4317 4318 4319 4320
  return code;

_err:
  tsdbError("failed to suspend data reader, code:%s %s", tstrerror(code), pReader->idStr);
  return code;
}

static int32_t tsdbSetQueryReseek(void* pQHandle) {
  int32_t      code = 0;
  STsdbReader* pReader = pQHandle;

4321
  code = tsdbTryAcquireReader(pReader);
4322 4323
  if (code == 0) {
    if (pReader->suspended) {
4324
      tsdbReleaseReader(pReader);
4325 4326 4327 4328
      return code;
    }

    tsdbReaderSuspend(pReader);
4329

4330
    tsdbReleaseReader(pReader);
4331

4332
    return code;
4333 4334 4335
  } else if (code == EBUSY) {
    return TSDB_CODE_VND_QUERY_BUSY;
  } else {
4336 4337
    terrno = TAOS_SYSTEM_ERROR(code);
    return TSDB_CODE_FAILED;
4338 4339 4340 4341 4342 4343
  }
}

int32_t tsdbReaderResume(STsdbReader* pReader) {
  int32_t code = 0;

4344
  STableBlockScanInfo** pBlockScanInfo = pReader->status.pTableIter;
4345 4346 4347

  //  restore reader's state
  //  task snapshot
4348
  int32_t numOfTables = taosHashGetSize(pReader->status.pTableMap);
4349
  if (numOfTables > 0) {
4350
    qTrace("tsdb/reader: %p, take snapshot", pReader);
4351 4352 4353 4354 4355 4356 4357 4358 4359 4360 4361 4362 4363 4364 4365 4366 4367
    code = tsdbTakeReadSnap(pReader, tsdbSetQueryReseek, &pReader->pReadSnap);
    if (code != TSDB_CODE_SUCCESS) {
      goto _err;
    }

    if (pReader->type == TIMEWINDOW_RANGE_CONTAINED) {
      code = doOpenReaderImpl(pReader);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
    } else {
      STsdbReader* pPrevReader = pReader->innerReader[0];
      STsdbReader* pNextReader = pReader->innerReader[1];

      // we need only one row
      pPrevReader->capacity = 1;
      pPrevReader->status.pTableMap = pReader->status.pTableMap;
X
Xiaoyu Wang 已提交
4368
      pPrevReader->status.uidList = pReader->status.uidList;
4369 4370 4371 4372 4373 4374
      pPrevReader->pSchema = pReader->pSchema;
      pPrevReader->pMemSchema = pReader->pMemSchema;
      pPrevReader->pReadSnap = pReader->pReadSnap;

      pNextReader->capacity = 1;
      pNextReader->status.pTableMap = pReader->status.pTableMap;
X
Xiaoyu Wang 已提交
4375
      pNextReader->status.uidList = pReader->status.uidList;
4376 4377 4378 4379 4380 4381 4382 4383 4384 4385 4386 4387 4388
      pNextReader->pSchema = pReader->pSchema;
      pNextReader->pMemSchema = pReader->pMemSchema;
      pNextReader->pReadSnap = pReader->pReadSnap;

      code = doOpenReaderImpl(pPrevReader);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
    }
  }

  pReader->suspended = false;

4389 4390
  tsdbDebug("reader: %p resumed uid %" PRIu64 ", numOfTable:%" PRId32 ", in this query %s", pReader,
            pBlockScanInfo ? (*pBlockScanInfo)->uid : 0, numOfTables, pReader->idStr);
4391 4392 4393 4394 4395 4396 4397
  return code;

_err:
  tsdbError("failed to resume data reader, code:%s %s", tstrerror(code), pReader->idStr);
  return code;
}

4398
static bool doTsdbNextDataBlock(STsdbReader* pReader) {
H
Haojun Liao 已提交
4399
  // cleanup the data that belongs to the previous data block
4400 4401
  SSDataBlock* pBlock = pReader->pResBlock;
  blockDataCleanup(pBlock);
H
Hongze Cheng 已提交
4402

4403
  SReaderStatus* pStatus = &pReader->status;
4404
  if (taosHashGetSize(pStatus->pTableMap) == 0) {
4405 4406
    return false;
  }
H
Haojun Liao 已提交
4407

4408 4409 4410 4411 4412
  if (pStatus->loadFromFile) {
    int32_t code = buildBlockFromFiles(pReader);
    if (code != TSDB_CODE_SUCCESS) {
      return false;
    }
4413

4414 4415 4416
    if (pBlock->info.rows > 0) {
      return true;
    } else {
4417
      resetTableListIndex(&pReader->status);
H
Haojun Liao 已提交
4418
      buildBlockFromBufferSequentially(pReader);
4419
      return pBlock->info.rows > 0;
H
Haojun Liao 已提交
4420
    }
4421 4422 4423
  } else {  // no data in files, let's try the buffer
    buildBlockFromBufferSequentially(pReader);
    return pBlock->info.rows > 0;
H
Haojun Liao 已提交
4424
  }
H
refact  
Hongze Cheng 已提交
4425 4426
}

4427
bool tsdbNextDataBlock(STsdbReader* pReader) {
4428
  if (isEmptyQueryTimeWindow(&pReader->window) || pReader->step == EXTERNAL_ROWS_NEXT) {
4429 4430 4431
    return false;
  }

4432 4433
  SReaderStatus* pStatus = &pReader->status;

4434 4435 4436
  int32_t code = tsdbAcquireReader(pReader);
  qTrace("tsdb/read: %p, take read mutex, code: %d", pReader, code);

4437 4438 4439 4440
  if (pReader->suspended) {
    tsdbReaderResume(pReader);
  }

4441
  if (pReader->innerReader[0] != NULL && pReader->step == 0) {
4442
    bool ret = doTsdbNextDataBlock(pReader->innerReader[0]);
4443
    pReader->step = EXTERNAL_ROWS_PREV;
4444
    if (ret) {
4445
      pStatus = &pReader->innerReader[0]->status;
4446
      if (pStatus->composedDataBlock) {
4447
        qTrace("tsdb/read: %p, unlock read mutex", pReader);
4448
        tsdbReleaseReader(pReader);
4449 4450
      }

4451 4452
      return ret;
    }
4453
  }
4454

4455
  if (pReader->step == EXTERNAL_ROWS_PREV) {
4456 4457
    // prepare for the main scan
    int32_t code = doOpenReaderImpl(pReader);
4458
    resetAllDataBlockScanInfo(pReader->status.pTableMap, pReader->innerReader[0]->window.ekey);
4459 4460 4461 4462 4463

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

4464
    pReader->step = EXTERNAL_ROWS_MAIN;
4465 4466 4467 4468
  }

  bool ret = doTsdbNextDataBlock(pReader);
  if (ret) {
4469
    if (pStatus->composedDataBlock) {
4470
      qTrace("tsdb/read: %p, unlock read mutex", pReader);
4471
      tsdbReleaseReader(pReader);
4472 4473
    }

4474 4475 4476
    return ret;
  }

4477
  if (pReader->step == EXTERNAL_ROWS_MAIN && pReader->innerReader[1] != NULL) {
4478 4479
    // prepare for the next row scan
    int32_t code = doOpenReaderImpl(pReader->innerReader[1]);
4480
    resetAllDataBlockScanInfo(pReader->innerReader[1]->status.pTableMap, pReader->window.ekey);
4481 4482 4483 4484
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

4485
    ret = doTsdbNextDataBlock(pReader->innerReader[1]);
4486
    pReader->step = EXTERNAL_ROWS_NEXT;
4487
    if (ret) {
4488
      pStatus = &pReader->innerReader[1]->status;
4489
      if (pStatus->composedDataBlock) {
4490
        qTrace("tsdb/read: %p, unlock read mutex", pReader);
4491
        tsdbReleaseReader(pReader);
4492 4493
      }

4494
      return ret;
4495 4496 4497
    }
  }

4498
  qTrace("tsdb/read: %p, unlock read mutex", pReader);
4499
  tsdbReleaseReader(pReader);
4500

4501 4502 4503
  return false;
}

4504
static void doFillNullColSMA(SBlockLoadSuppInfo* pSup, int32_t numOfRows, int32_t numOfCols, SColumnDataAgg* pTsAgg) {
4505 4506
  // do fill all null column value SMA info
  int32_t i = 0, j = 0;
4507
  int32_t size = (int32_t)taosArrayGetSize(pSup->pColAgg);
4508
  taosArrayInsert(pSup->pColAgg, 0, pTsAgg);
4509 4510 4511 4512 4513 4514 4515 4516 4517 4518 4519

  while (j < numOfCols && i < size) {
    SColumnDataAgg* pAgg = taosArrayGet(pSup->pColAgg, i);
    if (pAgg->colId == pSup->colId[j]) {
      i += 1;
      j += 1;
    } else if (pAgg->colId < pSup->colId[j]) {
      i += 1;
    } else if (pSup->colId[j] < pAgg->colId) {
      if (pSup->colId[j] != PRIMARYKEY_TIMESTAMP_COL_ID) {
        SColumnDataAgg nullColAgg = {.colId = pSup->colId[j], .numOfNull = numOfRows};
4520
        taosArrayInsert(pSup->pColAgg, i, &nullColAgg);
4521 4522 4523 4524 4525 4526
      }
      j += 1;
    }
  }
}

4527
int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SSDataBlock* pDataBlock, bool* allHave) {
H
Haojun Liao 已提交
4528 4529
  SColumnDataAgg*** pBlockSMA = &pDataBlock->pBlockAgg;

H
Hongze Cheng 已提交
4530
  int32_t code = 0;
4531
  *allHave = false;
H
Haojun Liao 已提交
4532
  *pBlockSMA = NULL;
H
Hongze Cheng 已提交
4533

4534
  if (pReader->type == TIMEWINDOW_RANGE_EXTERNAL) {
4535 4536 4537
    return TSDB_CODE_SUCCESS;
  }

4538
  // there is no statistics data for composed block
4539
  if (pReader->status.composedDataBlock || (!pReader->suppInfo.smaValid)) {
4540 4541
    return TSDB_CODE_SUCCESS;
  }
H
Hongze Cheng 已提交
4542

4543
  SFileDataBlockInfo* pFBlock = getCurrentBlockInfo(&pReader->status.blockIter);
4544 4545
  SBlockLoadSuppInfo* pSup = &pReader->suppInfo;

H
Haojun Liao 已提交
4546 4547 4548
  if (pReader->pResBlock->info.id.uid != pFBlock->uid) {
    return TSDB_CODE_SUCCESS;
  }
4549 4550

  SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter);
H
Hongze Cheng 已提交
4551
  if (tDataBlkHasSma(pBlock)) {
H
Hongze Cheng 已提交
4552
    code = tsdbReadBlockSma(pReader->pFileReader, pBlock, pSup->pColAgg);
4553
    if (code != TSDB_CODE_SUCCESS) {
4554 4555
      tsdbDebug("vgId:%d, failed to load block SMA for uid %" PRIu64 ", code:%s, %s", 0, pFBlock->uid, tstrerror(code),
                pReader->idStr);
4556 4557
      return code;
    }
4558
  } else {
H
Haojun Liao 已提交
4559
    *pBlockSMA = NULL;
4560
    return TSDB_CODE_SUCCESS;
4561
  }
H
Hongze Cheng 已提交
4562

4563
  *allHave = true;
H
Hongze Cheng 已提交
4564

4565 4566
  // always load the first primary timestamp column data
  SColumnDataAgg* pTsAgg = &pSup->tsColAgg;
4567

4568 4569
  pTsAgg->numOfNull = 0;
  pTsAgg->colId = PRIMARYKEY_TIMESTAMP_COL_ID;
4570 4571 4572 4573
  pTsAgg->min = pReader->pResBlock->info.window.skey;
  pTsAgg->max = pReader->pResBlock->info.window.ekey;

  // update the number of NULL data rows
4574
  size_t numOfCols = pSup->numOfCols;
4575

4576
  // ensure capacity
H
Haojun Liao 已提交
4577 4578 4579
  if (pDataBlock->pDataBlock) {
    size_t colsNum = taosArrayGetSize(pDataBlock->pDataBlock);
    taosArrayEnsureCap(pSup->pColAgg, colsNum);
4580 4581
  }

4582 4583 4584
  SSDataBlock* pResBlock = pReader->pResBlock;
  if (pResBlock->pBlockAgg == NULL) {
    size_t num = taosArrayGetSize(pResBlock->pDataBlock);
H
Haojun Liao 已提交
4585
    pResBlock->pBlockAgg = taosMemoryCalloc(num, POINTER_BYTES);
4586
  }
4587

4588
  // do fill all null column value SMA info
4589
  doFillNullColSMA(pSup, pBlock->nRow, numOfCols, pTsAgg);
H
Haojun Liao 已提交
4590
  size_t size = taosArrayGetSize(pSup->pColAgg);
4591

H
Haojun Liao 已提交
4592
  int32_t i = 0, j = 0;
4593
  while (j < numOfCols && i < size) {
4594
    SColumnDataAgg* pAgg = taosArrayGet(pSup->pColAgg, i);
H
Haojun Liao 已提交
4595 4596
    if (pAgg->colId == pSup->colId[j]) {
      pResBlock->pBlockAgg[pSup->slotId[j]] = pAgg;
4597 4598
      i += 1;
      j += 1;
H
Haojun Liao 已提交
4599
    } else if (pAgg->colId < pSup->colId[j]) {
4600
      i += 1;
H
Haojun Liao 已提交
4601
    } else if (pSup->colId[j] < pAgg->colId) {
H
Haojun Liao 已提交
4602
      // ASSERT(pSup->colId[j] == PRIMARYKEY_TIMESTAMP_COL_ID);
4603
      pResBlock->pBlockAgg[pSup->slotId[j]] = &pSup->tsColAgg;
4604 4605 4606 4607
      j += 1;
    }
  }

H
Haojun Liao 已提交
4608
  *pBlockSMA = pResBlock->pBlockAgg;
4609
  pReader->cost.smaDataLoad += 1;
4610

4611
  tsdbDebug("vgId:%d, succeed to load block SMA for uid %" PRIu64 ", %s", 0, pFBlock->uid, pReader->idStr);
H
Hongze Cheng 已提交
4612
  return code;
H
Hongze Cheng 已提交
4613 4614
}

H
Haojun Liao 已提交
4615 4616 4617 4618 4619 4620 4621 4622 4623 4624 4625 4626
STableBlockScanInfo* getTableBlockScanInfo(SHashObj* pTableMap, uint64_t uid, const char* id) {
  STableBlockScanInfo** p = taosHashGet(pTableMap, &uid, sizeof(uid));
  if (p == NULL || *p == NULL) {
    terrno = TSDB_CODE_INVALID_PARA;
    int32_t size = taosHashGetSize(pTableMap);
    tsdbError("failed to locate the uid:%" PRIu64 " in query table uid list, total tables:%d, %s", uid, size, id);
    return NULL;
  }

  return *p;
}

H
Haojun Liao 已提交
4627
static SSDataBlock* doRetrieveDataBlock(STsdbReader* pReader) {
4628
  SReaderStatus*       pStatus = &pReader->status;
H
Haojun Liao 已提交
4629
  SFileDataBlockInfo*  pBlockInfo = getCurrentBlockInfo(&pStatus->blockIter);
H
Haojun Liao 已提交
4630
  STableBlockScanInfo* pBlockScanInfo = getTableBlockScanInfo(pStatus->pTableMap, pBlockInfo->uid, pReader->idStr);
H
Haojun Liao 已提交
4631
  if (pBlockScanInfo == NULL) {
4632
    return NULL;
4633 4634
  }

4635
  int32_t code = doLoadFileBlockData(pReader, &pStatus->blockIter, &pStatus->fileBlockData, pBlockScanInfo->uid);
4636
  if (code != TSDB_CODE_SUCCESS) {
H
Hongze Cheng 已提交
4637
    tBlockDataDestroy(&pStatus->fileBlockData);
4638 4639
    terrno = code;
    return NULL;
4640
  }
4641

4642
  copyBlockDataToSDataBlock(pReader);
H
Haojun Liao 已提交
4643
  return pReader->pResBlock;
H
Hongze Cheng 已提交
4644 4645
}

H
Haojun Liao 已提交
4646
SSDataBlock* tsdbRetrieveDataBlock(STsdbReader* pReader, SArray* pIdList) {
4647
  STsdbReader* pTReader = pReader;
4648 4649
  if (pReader->type == TIMEWINDOW_RANGE_EXTERNAL) {
    if (pReader->step == EXTERNAL_ROWS_PREV) {
4650
      pTReader = pReader->innerReader[0];
4651
    } else if (pReader->step == EXTERNAL_ROWS_NEXT) {
4652
      pTReader = pReader->innerReader[1];
4653 4654 4655
    }
  }

4656 4657 4658 4659 4660 4661 4662
  SReaderStatus* pStatus = &pTReader->status;
  if (pStatus->composedDataBlock) {
    return pTReader->pResBlock;
  }

  SSDataBlock* ret = doRetrieveDataBlock(pTReader);

4663
  qTrace("tsdb/read-retrieve: %p, unlock read mutex", pReader);
4664
  tsdbReleaseReader(pReader);
4665 4666

  return ret;
4667 4668
}

H
Haojun Liao 已提交
4669
int32_t tsdbReaderReset(STsdbReader* pReader, SQueryTableDataCond* pCond) {
4670
  qTrace("tsdb/reader-reset: %p, take read mutex", pReader);
4671
  tsdbAcquireReader(pReader);
L
Liu Jicong 已提交
4672 4673 4674 4675 4676

  if (pReader->suspended) {
    tsdbReaderResume(pReader);
  }

H
Haojun Liao 已提交
4677
  if (isEmptyQueryTimeWindow(&pReader->window) || pReader->pReadSnap == NULL) {
L
Liu Jicong 已提交
4678
    tsdbDebug("tsdb reader reset return %p", pReader->pReadSnap);
4679

4680
    tsdbReleaseReader(pReader);
4681

4682 4683
    return TSDB_CODE_SUCCESS;
  }
H
Hongze Cheng 已提交
4684

H
Haojun Liao 已提交
4685 4686 4687
  SReaderStatus* pStatus = &pReader->status;

  SDataBlockIter* pBlockIter = &pStatus->blockIter;
4688

L
Liu Jicong 已提交
4689
  pReader->order = pCond->order;
4690
  pReader->type = TIMEWINDOW_RANGE_CONTAINED;
H
Haojun Liao 已提交
4691 4692
  pStatus->loadFromFile = true;
  pStatus->pTableIter = NULL;
H
Haojun Liao 已提交
4693
  pReader->window = updateQueryTimeWindow(pReader->pTsdb, &pCond->twindows);
H
Hongze Cheng 已提交
4694

4695
  // allocate buffer in order to load data blocks from file
4696
  memset(&pReader->suppInfo.tsColAgg, 0, sizeof(SColumnDataAgg));
4697

4698
  pReader->suppInfo.tsColAgg.colId = PRIMARYKEY_TIMESTAMP_COL_ID;
4699
  tsdbDataFReaderClose(&pReader->pFileReader);
4700

H
Haojun Liao 已提交
4701
  int32_t numOfTables = taosHashGetSize(pStatus->pTableMap);
L
Liu Jicong 已提交
4702

H
Haojun Liao 已提交
4703
  initFilesetIterator(&pStatus->fileIter, pReader->pReadSnap->fs.aDFileSet, pReader);
4704
  resetDataBlockIterator(pBlockIter, pReader->order);
H
Haojun Liao 已提交
4705
  resetTableListIndex(&pReader->status);
H
Haojun Liao 已提交
4706

H
Hongze Cheng 已提交
4707
  int64_t ts = ASCENDING_TRAVERSE(pReader->order) ? pReader->window.skey - 1 : pReader->window.ekey + 1;
H
Haojun Liao 已提交
4708
  resetAllDataBlockScanInfo(pStatus->pTableMap, ts);
4709

4710
  int32_t code = 0;
4711

4712
  // no data in files, let's try buffer in memory
H
Haojun Liao 已提交
4713 4714
  if (pStatus->fileIter.numOfFiles == 0) {
    pStatus->loadFromFile = false;
4715
    resetTableListIndex(pStatus);
4716 4717 4718
  } else {
    code = initForFirstBlockInFile(pReader, pBlockIter);
    if (code != TSDB_CODE_SUCCESS) {
4719 4720
      tsdbError("%p reset reader failed, numOfTables:%d, query range:%" PRId64 " - %" PRId64 " in query %s", pReader,
                numOfTables, pReader->window.skey, pReader->window.ekey, pReader->idStr);
4721

4722
      tsdbReleaseReader(pReader);
4723

4724 4725 4726
      return code;
    }
  }
H
Hongze Cheng 已提交
4727

H
Hongze Cheng 已提交
4728 4729 4730 4731
  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);
4732

4733
  tsdbReleaseReader(pReader);
4734

4735
  return code;
H
Hongze Cheng 已提交
4736
}
H
Hongze Cheng 已提交
4737

4738 4739 4740 4741 4742 4743
static int32_t getBucketIndex(int32_t startRow, int32_t bucketRange, int32_t numOfRows, int32_t numOfBucket) {
  int32_t bucketIndex = (numOfRows - startRow) / bucketRange;
  if (bucketIndex == numOfBucket) {
    bucketIndex -= 1;
  }
  return bucketIndex;
4744
}
H
Hongze Cheng 已提交
4745

4746 4747 4748 4749
int32_t tsdbGetFileBlocksDistInfo(STsdbReader* pReader, STableBlockDistInfo* pTableBlockInfo) {
  int32_t code = TSDB_CODE_SUCCESS;
  pTableBlockInfo->totalSize = 0;
  pTableBlockInfo->totalRows = 0;
4750
  pTableBlockInfo->numOfVgroups = 1;
H
Hongze Cheng 已提交
4751

4752
  const int32_t numOfBucket = 20.0;
dengyihao's avatar
dengyihao 已提交
4753

4754
  // find the start data block in file
dengyihao's avatar
dengyihao 已提交
4755 4756 4757 4758
  tsdbAcquireReader(pReader);
  if (pReader->suspended) {
    tsdbReaderResume(pReader);
  }
4759
  SReaderStatus* pStatus = &pReader->status;
H
Hongze Cheng 已提交
4760

4761 4762 4763
  STsdbCfg* pc = &pReader->pTsdb->pVnode->config.tsdbCfg;
  pTableBlockInfo->defMinRows = pc->minRows;
  pTableBlockInfo->defMaxRows = pc->maxRows;
H
Hongze Cheng 已提交
4764

4765
  int32_t bucketRange = ceil((pc->maxRows - pc->minRows) / numOfBucket);
H
Hongze Cheng 已提交
4766

4767
  pTableBlockInfo->numOfFiles += 1;
H
Hongze Cheng 已提交
4768

4769 4770
  int32_t numOfTables = (int32_t)taosHashGetSize(pStatus->pTableMap);
  int     defaultRows = 4096;
H
Hongze Cheng 已提交
4771

4772 4773
  SDataBlockIter* pBlockIter = &pStatus->blockIter;
  pTableBlockInfo->numOfFiles += pStatus->fileIter.numOfFiles;
H
Haojun Liao 已提交
4774

4775 4776
  if (pBlockIter->numOfBlocks > 0) {
    pTableBlockInfo->numOfBlocks += pBlockIter->numOfBlocks;
H
Haojun Liao 已提交
4777
  }
H
Hongze Cheng 已提交
4778

4779
  pTableBlockInfo->numOfTables = numOfTables;
4780
  bool hasNext = (pBlockIter->numOfBlocks > 0);
H
Hongze Cheng 已提交
4781

4782 4783
  while (true) {
    if (hasNext) {
H
Hongze Cheng 已提交
4784
      SDataBlk* pBlock = getCurrentBlock(pBlockIter);
H
Hongze Cheng 已提交
4785

4786 4787
      int32_t numOfRows = pBlock->nRow;
      pTableBlockInfo->totalRows += numOfRows;
H
Hongze Cheng 已提交
4788

4789 4790 4791
      if (numOfRows > pTableBlockInfo->maxRows) {
        pTableBlockInfo->maxRows = numOfRows;
      }
H
refact  
Hongze Cheng 已提交
4792

4793 4794 4795
      if (numOfRows < pTableBlockInfo->minRows) {
        pTableBlockInfo->minRows = numOfRows;
      }
H
refact  
Hongze Cheng 已提交
4796

4797 4798 4799
      if (numOfRows < defaultRows) {
        pTableBlockInfo->numOfSmallBlocks += 1;
      }
H
refact  
Hongze Cheng 已提交
4800

4801
      pTableBlockInfo->totalSize += pBlock->aSubBlock[0].szBlock;
4802

4803
      int32_t bucketIndex = getBucketIndex(pTableBlockInfo->defMinRows, bucketRange, numOfRows, numOfBucket);
4804
      pTableBlockInfo->blockRowsHisto[bucketIndex]++;
4805

H
Haojun Liao 已提交
4806
      hasNext = blockIteratorNext(&pStatus->blockIter, pReader->idStr);
4807 4808
    } else {
      code = initForFirstBlockInFile(pReader, pBlockIter);
H
Haojun Liao 已提交
4809
      if ((code != TSDB_CODE_SUCCESS) || (pStatus->loadFromFile == false)) {
4810 4811
        break;
      }
H
refact  
Hongze Cheng 已提交
4812

4813 4814
      pTableBlockInfo->numOfBlocks += pBlockIter->numOfBlocks;
      hasNext = (pBlockIter->numOfBlocks > 0);
4815
    }
H
refact  
Hongze Cheng 已提交
4816

H
Hongze Cheng 已提交
4817 4818
    //    tsdbDebug("%p %d blocks found in file for %d table(s), fid:%d, %s", pReader, numOfBlocks, numOfTables,
    //              pReader->pFileGroup->fid, pReader->idStr);
4819
  }
dengyihao's avatar
dengyihao 已提交
4820
  tsdbReleaseReader(pReader);
H
refact  
Hongze Cheng 已提交
4821 4822
  return code;
}
H
Hongze Cheng 已提交
4823

H
refact  
Hongze Cheng 已提交
4824
int64_t tsdbGetNumOfRowsInMemTable(STsdbReader* pReader) {
4825
  int64_t rows = 0;
H
Hongze Cheng 已提交
4826

4827
  SReaderStatus* pStatus = &pReader->status;
4828
  tsdbAcquireReader(pReader);
4829 4830 4831 4832
  if (pReader->suspended) {
    tsdbReaderResume(pReader);
  }

4833
  pStatus->pTableIter = taosHashIterate(pStatus->pTableMap, NULL);
H
Hongze Cheng 已提交
4834

4835
  while (pStatus->pTableIter != NULL) {
4836
    STableBlockScanInfo* pBlockScanInfo = *(STableBlockScanInfo**)pStatus->pTableIter;
4837 4838

    STbData* d = NULL;
4839
    if (pReader->pReadSnap->pMem != NULL) {
H
Hongze Cheng 已提交
4840
      d = tsdbGetTbDataFromMemTable(pReader->pReadSnap->pMem, pReader->suid, pBlockScanInfo->uid);
4841 4842 4843 4844 4845 4846
      if (d != NULL) {
        rows += tsdbGetNRowsInTbData(d);
      }
    }

    STbData* di = NULL;
4847
    if (pReader->pReadSnap->pIMem != NULL) {
H
Hongze Cheng 已提交
4848
      di = tsdbGetTbDataFromMemTable(pReader->pReadSnap->pIMem, pReader->suid, pBlockScanInfo->uid);
4849 4850 4851 4852 4853 4854 4855 4856
      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 已提交
4857

4858
  tsdbReleaseReader(pReader);
4859

H
refact  
Hongze Cheng 已提交
4860
  return rows;
H
Hongze Cheng 已提交
4861
}
D
dapan1121 已提交
4862

L
Liu Jicong 已提交
4863
int32_t tsdbGetTableSchema(SVnode* pVnode, int64_t uid, STSchema** pSchema, int64_t* suid) {
D
dapan1121 已提交
4864 4865 4866 4867
  int32_t sversion = 1;

  SMetaReader mr = {0};
  metaReaderInit(&mr, pVnode->pMeta, 0);
H
Haojun Liao 已提交
4868
  int32_t code = metaGetTableEntryByUidCache(&mr, uid);
D
dapan1121 已提交
4869 4870 4871 4872 4873 4874 4875
  if (code != TSDB_CODE_SUCCESS) {
    terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
    metaReaderClear(&mr);
    return terrno;
  }

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

D
dapan1121 已提交
4877
  if (mr.me.type == TSDB_CHILD_TABLE) {
D
dapan1121 已提交
4878
    tDecoderClear(&mr.coder);
D
dapan1121 已提交
4879
    *suid = mr.me.ctbEntry.suid;
H
Haojun Liao 已提交
4880
    code = metaGetTableEntryByUidCache(&mr, *suid);
D
dapan1121 已提交
4881 4882 4883 4884 4885 4886
    if (code != TSDB_CODE_SUCCESS) {
      terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
      metaReaderClear(&mr);
      return terrno;
    }
    sversion = mr.me.stbEntry.schemaRow.version;
H
Haojun Liao 已提交
4887
  } else if (mr.me.type == TSDB_NORMAL_TABLE) {
D
dapan1121 已提交
4888
    sversion = mr.me.ntbEntry.schemaRow.version;
H
Haojun Liao 已提交
4889 4890 4891 4892
  } else {
    terrno = TSDB_CODE_INVALID_PARA;
    metaReaderClear(&mr);
    return terrno;
D
dapan1121 已提交
4893 4894 4895
  }

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

D
dapan1121 已提交
4898 4899
  return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
4900

H
Hongze Cheng 已提交
4901
int32_t tsdbTakeReadSnap(STsdbReader* pReader, _query_reseek_func_t reseek, STsdbReadSnap** ppSnap) {
H
Hongze Cheng 已提交
4902 4903 4904
  int32_t        code = 0;
  STsdb*         pTsdb = pReader->pTsdb;
  SVersionRange* pRange = &pReader->verRange;
H
Hongze Cheng 已提交
4905 4906

  // alloc
H
Hongze Cheng 已提交
4907 4908
  STsdbReadSnap* pSnap = (STsdbReadSnap*)taosMemoryCalloc(1, sizeof(*pSnap));
  if (pSnap == NULL) {
H
Hongze Cheng 已提交
4909 4910 4911 4912 4913
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

  // lock
H
Hongze Cheng 已提交
4914
  taosThreadRwlockRdlock(&pTsdb->rwLock);
H
Hongze Cheng 已提交
4915 4916

  // take snapshot
H
Hongze Cheng 已提交
4917
  if (pTsdb->mem && (pRange->minVer <= pTsdb->mem->maxVer && pRange->maxVer >= pTsdb->mem->minVer)) {
H
Hongze Cheng 已提交
4918 4919 4920 4921 4922 4923 4924 4925 4926 4927 4928
    pSnap->pMem = pTsdb->mem;
    pSnap->pNode = taosMemoryMalloc(sizeof(*pSnap->pNode));
    if (pSnap->pNode == NULL) {
      taosThreadRwlockUnlock(&pTsdb->rwLock);
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _exit;
    }
    pSnap->pNode->pQHandle = pReader;
    pSnap->pNode->reseek = reseek;

    tsdbRefMemTable(pTsdb->mem, pSnap->pNode);
H
Hongze Cheng 已提交
4929 4930
  }

H
Hongze Cheng 已提交
4931
  if (pTsdb->imem && (pRange->minVer <= pTsdb->imem->maxVer && pRange->maxVer >= pTsdb->imem->minVer)) {
H
Hongze Cheng 已提交
4932 4933 4934 4935 4936 4937 4938 4939 4940 4941 4942
    pSnap->pIMem = pTsdb->imem;
    pSnap->pINode = taosMemoryMalloc(sizeof(*pSnap->pINode));
    if (pSnap->pINode == NULL) {
      taosThreadRwlockUnlock(&pTsdb->rwLock);
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _exit;
    }
    pSnap->pINode->pQHandle = pReader;
    pSnap->pINode->reseek = reseek;

    tsdbRefMemTable(pTsdb->imem, pSnap->pINode);
H
Hongze Cheng 已提交
4943 4944
  }

H
Hongze Cheng 已提交
4945
  // fs
H
Hongze Cheng 已提交
4946
  code = tsdbFSRef(pTsdb, &pSnap->fs);
H
Hongze Cheng 已提交
4947 4948 4949 4950
  if (code) {
    taosThreadRwlockUnlock(&pTsdb->rwLock);
    goto _exit;
  }
H
Hongze Cheng 已提交
4951 4952

  // unlock
H
Hongze Cheng 已提交
4953
  taosThreadRwlockUnlock(&pTsdb->rwLock);
H
Hongze Cheng 已提交
4954

4955
  tsdbTrace("vgId:%d, take read snapshot", TD_VID(pTsdb->pVnode));
H
Hongze Cheng 已提交
4956

H
Hongze Cheng 已提交
4957
_exit:
H
Hongze Cheng 已提交
4958 4959 4960 4961 4962 4963 4964 4965 4966 4967
  if (code) {
    *ppSnap = NULL;
    if (pSnap) {
      if (pSnap->pNode) taosMemoryFree(pSnap->pNode);
      if (pSnap->pINode) taosMemoryFree(pSnap->pINode);
      taosMemoryFree(pSnap);
    }
  } else {
    *ppSnap = pSnap;
  }
H
Hongze Cheng 已提交
4968 4969 4970
  return code;
}

4971
void tsdbUntakeReadSnap(STsdbReader* pReader, STsdbReadSnap* pSnap, bool proactive) {
H
Hongze Cheng 已提交
4972 4973
  STsdb* pTsdb = pReader->pTsdb;

H
Hongze Cheng 已提交
4974 4975
  if (pSnap) {
    if (pSnap->pMem) {
4976
      tsdbUnrefMemTable(pSnap->pMem, pSnap->pNode, proactive);
H
Hongze Cheng 已提交
4977 4978 4979
    }

    if (pSnap->pIMem) {
4980
      tsdbUnrefMemTable(pSnap->pIMem, pSnap->pINode, proactive);
H
Hongze Cheng 已提交
4981 4982
    }

H
Hongze Cheng 已提交
4983
    tsdbFSUnref(pTsdb, &pSnap->fs);
H
Hongze Cheng 已提交
4984 4985
    if (pSnap->pNode) taosMemoryFree(pSnap->pNode);
    if (pSnap->pINode) taosMemoryFree(pSnap->pINode);
H
Hongze Cheng 已提交
4986
    taosMemoryFree(pSnap);
H
Hongze Cheng 已提交
4987
  }
4988
  tsdbTrace("vgId:%d, untake read snapshot", TD_VID(pTsdb->pVnode));
H
Hongze Cheng 已提交
4989
}