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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

H
Hongze Cheng 已提交
142
struct STsdbReader {
H
Haojun Liao 已提交
143 144 145 146 147 148 149
  STsdb*             pTsdb;
  uint64_t           suid;
  int16_t            order;
  STimeWindow        window;  // the primary query time window that applies to all queries
  SSDataBlock*       pResBlock;
  int32_t            capacity;
  SReaderStatus      status;
150 151
  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 已提交
152
  SBlockLoadSuppInfo suppInfo;
H
Hongze Cheng 已提交
153
  STsdbReadSnap*     pReadSnap;
154
  SIOCostSummary     cost;
155 156
  STSchema*          pSchema;     // the newest version schema
  STSchema*          pMemSchema;  // the previous schema for in-memory data, to avoid load schema too many times
157 158
  SDataFReader*      pFileReader;
  SVersionRange      verRange;
159

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

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

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

dengyihao's avatar
dengyihao 已提交
187 188 189 190
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);
191
static SVersionRange getQueryVerRange(SVnode* pVnode, SQueryTableDataCond* pCond, int8_t level);
H
Hongze Cheng 已提交
192 193 194
static int64_t       getCurrentKeyInLastBlock(SLastBlockReader* pLastBlockReader);
static bool          hasDataInLastBlock(SLastBlockReader* pLastBlockReader);
static int32_t       doBuildDataBlock(STsdbReader* pReader);
C
Cary Xu 已提交
195
static TSDBKEY       getCurrentKeyInBuf(STableBlockScanInfo* pScanInfo, STsdbReader* pReader);
196 197
static bool          hasDataInFileBlock(const SBlockData* pBlockData, const SFileBlockDumpInfo* pDumpInfo);
static bool          hasDataInLastBlock(SLastBlockReader* pLastBlockReader);
C
Cary Xu 已提交
198 199

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

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

204
  size_t numOfCols = blockDataGetNumOfCols(pBlock);
205

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

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

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

H
Haojun Liao 已提交
223 224
  return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
225

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

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

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

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

253
  return pTableMap;
H
Hongze Cheng 已提交
254
}
H
Hongze Cheng 已提交
255

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

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

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

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

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

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

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

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

  taosHashCleanup(pTableMap);
}

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

299 300 301
// 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 已提交
302
  STsdbKeepCfg* pCfg = &pTsdb->keepCfg;
H
Hongze Cheng 已提交
303

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

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

  return win;
}
H
Hongze Cheng 已提交
314

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

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

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

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

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

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

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

354
  if (pLReader->pInfo == NULL) {
H
Haojun Liao 已提交
355 356 357 358 359
    pLReader->pInfo = tCreateLastBlockLoadInfo();
    if (pLReader->pInfo == NULL) {
      tsdbDebug("init fileset iterator failed, code:%s %s", tstrerror(terrno), pReader->idStr);
      return terrno;
    }
360 361
  }

362
  tsdbDebug("init fileset iterator, total files:%d %s", pIter->numOfFiles, pReader->idStr);
H
Haojun Liao 已提交
363 364 365
  return TSDB_CODE_SUCCESS;
}

366
static bool filesetIteratorNext(SFilesetIter* pIter, STsdbReader* pReader) {
367 368
  bool    asc = ASCENDING_TRAVERSE(pIter->order);
  int32_t step = asc ? 1 : -1;
369 370 371
  pIter->index += step;

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

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

378 379
  pIter->pLastBlockReader->uid = 0;
  tMergeTreeClose(&pIter->pLastBlockReader->mergeTree);
380
  resetLastBlockLoadInfo(pIter->pLastBlockReader->pInfo);
381

H
Haojun Liao 已提交
382 383
  // check file the time range of coverage
  STimeWindow win = {0};
H
Hongze Cheng 已提交
384

385
  while (1) {
H
Haojun Liao 已提交
386 387 388
    if (pReader->pFileReader != NULL) {
      tsdbDataFReaderClose(&pReader->pFileReader);
    }
389

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

392 393 394 395
    int32_t code = tsdbDataFReaderOpen(&pReader->pFileReader, pReader->pTsdb, pReader->status.pCurrentFileset);
    if (code != TSDB_CODE_SUCCESS) {
      goto _err;
    }
H
Haojun Liao 已提交
396

397 398
    pReader->cost.headFileLoad += 1;

399 400 401 402 403 404 405 406 407 408 409 410
    int32_t fid = pReader->status.pCurrentFileset->fid;
    tsdbFidKeyRange(fid, pReader->pTsdb->keepCfg.days, pReader->pTsdb->keepCfg.precision, &win.skey, &win.ekey);

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

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

417
    tsdbDebug("%p file found fid:%d for qrange:%" PRId64 "-%" PRId64 ", %s", pReader, fid, pReader->window.skey,
418
              pReader->window.ekey, pReader->idStr);
419 420
    return true;
  }
421

422
_err:
H
Haojun Liao 已提交
423 424 425
  return false;
}

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

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

H
Haojun Liao 已提交
439
static void initReaderStatus(SReaderStatus* pStatus) {
dengyihao's avatar
dengyihao 已提交
440 441
  pStatus->pTableIter = NULL;
  pStatus->loadFromFile = true;
H
Haojun Liao 已提交
442 443
}

444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466
static SSDataBlock* createResBlock(SQueryTableDataCond* pCond, int32_t capacity) {
  SSDataBlock* pResBlock = createDataBlock();
  if (pResBlock == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

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

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

  return pResBlock;
}

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

C
Cary Xu 已提交
477 478 479 480
  if (VND_IS_TSMA(pVnode)) {
    tsdbDebug("vgId:%d, tsma is selected to query", TD_VID(pVnode));
  }

H
Haojun Liao 已提交
481
  initReaderStatus(&pReader->status);
482

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

492
  ASSERT(pCond->numOfCols > 0);
H
Hongze Cheng 已提交
493

494
  limitOutputBufferSize(pCond, &pReader->capacity);
495

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

505 506
  pSup->tsColAgg.colId = PRIMARYKEY_TIMESTAMP_COL_ID;

H
Hongze Cheng 已提交
507
  code = tBlockDataCreate(&pReader->status.fileBlockData);
H
Haojun Liao 已提交
508 509 510 511 512
  if (code != TSDB_CODE_SUCCESS) {
    terrno = code;
    goto _end;
  }

513 514 515 516
  pReader->pResBlock = createResBlock(pCond, pReader->capacity);
  if (pReader->pResBlock == NULL) {
    code = terrno;
    goto _end;
H
Hongze Cheng 已提交
517
  }
H
Hongze Cheng 已提交
518

519 520
  setColumnIdSlotList(pReader, pReader->pResBlock);

H
Hongze Cheng 已提交
521 522
  *ppReader = pReader;
  return code;
H
Hongze Cheng 已提交
523

H
Haojun Liao 已提交
524 525
_end:
  tsdbReaderClose(pReader);
H
Hongze Cheng 已提交
526 527 528
  *ppReader = NULL;
  return code;
}
H
Hongze Cheng 已提交
529

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

533
  int64_t st = taosGetTimestampUs();
H
Hongze Cheng 已提交
534
  int32_t code = tsdbReadBlockIdx(pFileReader, aBlockIdx);
H
Haojun Liao 已提交
535
  if (code != TSDB_CODE_SUCCESS) {
536
    goto _end;
H
Haojun Liao 已提交
537
  }
H
Hongze Cheng 已提交
538

539 540
  size_t num = taosArrayGetSize(aBlockIdx);
  if (num == 0) {
H
Haojun Liao 已提交
541
    taosArrayDestroy(aBlockIdx);
H
Haojun Liao 已提交
542 543
    return TSDB_CODE_SUCCESS;
  }
H
Hongze Cheng 已提交
544

545 546 547 548
  int64_t et1 = taosGetTimestampUs();

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

551
    // uid check
H
Hongze Cheng 已提交
552
    if (pBlockIdx->suid != pReader->suid) {
H
Haojun Liao 已提交
553 554 555 556
      continue;
    }

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

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

H
Hongze Cheng 已提交
567
    taosArrayPush(pIndexList, pBlockIdx);
H
Haojun Liao 已提交
568
  }
H
Hongze Cheng 已提交
569

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

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

576
_end:
H
Hongze Cheng 已提交
577
  taosArrayDestroy(aBlockIdx);
H
Haojun Liao 已提交
578 579
  return code;
}
H
Hongze Cheng 已提交
580

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

589
    // reset the index in last block when handing a new file
590
    tMapDataClear(&px->mapData);
591 592
    taosArrayClear(px->pBlockList);
  }
593 594
}

595
static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, SBlockNumber* pBlockNum) {
596 597 598 599 600 601
  int32_t numOfQTable = 0;
  size_t  sizeInDisk = 0;
  size_t  numOfTables = taosArrayGetSize(pIndexList);

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

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

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

608
    tMapDataReset(&pScanInfo->mapData);
H
Hongze Cheng 已提交
609
    tsdbReadDataBlk(pReader->pFileReader, pBlockIdx, &pScanInfo->mapData);
610

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

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

621
      // 2. version range check
H
Hongze Cheng 已提交
622
      if (block.minVer > pReader->verRange.maxVer || block.maxVer < pReader->verRange.minVer) {
623 624
        continue;
      }
625

626
      void* p = taosArrayPush(pScanInfo->pBlockList, &j);
H
Haojun Liao 已提交
627
      if (p == NULL) {
628
        tMapDataClear(&pScanInfo->mapData);
H
Haojun Liao 已提交
629 630
        return TSDB_CODE_OUT_OF_MEMORY;
      }
631

632
      pBlockNum->numOfBlocks += 1;
H
Haojun Liao 已提交
633
    }
H
Hongze Cheng 已提交
634

H
Haojun Liao 已提交
635
    if (pScanInfo->pBlockList != NULL && taosArrayGetSize(pScanInfo->pBlockList) > 0) {
636 637 638 639
      numOfQTable += 1;
    }
  }

H
Hongze Cheng 已提交
640
  pBlockNum->numOfLastFiles = pReader->pFileReader->pSet->nSttF;
641
  int32_t total = pBlockNum->numOfLastFiles + pBlockNum->numOfBlocks;
642

643
  double el = (taosGetTimestampUs() - st) / 1000.0;
H
Hongze Cheng 已提交
644
  tsdbDebug(
645
      "load block of %d tables completed, blocks:%d in %d tables, last-files:%d, block-info-size:%.2f Kb, elapsed "
646
      "time:%.2f ms %s",
647
      numOfTables, pBlockNum->numOfBlocks, numOfQTable, pBlockNum->numOfLastFiles, sizeInDisk / 1000.0, el,
H
Hongze Cheng 已提交
648
      pReader->idStr);
649

650
  pReader->cost.numOfBlocks += total;
651
  pReader->cost.headFileLoadTime += el;
652

H
Haojun Liao 已提交
653 654
  return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
655

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

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

678
static SFileDataBlockInfo* getCurrentBlockInfo(SDataBlockIter* pBlockIter) {
679 680
  if (taosArrayGetSize(pBlockIter->blockList) == 0) {
    ASSERT(pBlockIter->numOfBlocks == taosArrayGetSize(pBlockIter->blockList));
681 682
    return NULL;
  }
683 684 685

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

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

C
Cary Xu 已提交
690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762
int32_t binarySearchForTs(char* pValue, int num, TSKEY key, int order) {
  int32_t midPos = -1;
  int32_t numOfRows;

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

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

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

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

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

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

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

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

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

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

  return midPos;
}

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

  // check
C
Cary Xu 已提交
763
  assert(pos >=0 && pos < num);
C
Cary Xu 已提交
764 765 766 767
  assert(num > 0);

  if (order == TSDB_ORDER_ASC) {
    // find the first position which is smaller than the key
C
Cary Xu 已提交
768 769 770
    e  = num - 1;
    if (key < keyList[pos])
      return -1;
C
Cary Xu 已提交
771 772
    while (1) {
      // check can return
C
Cary Xu 已提交
773 774 775 776 777 778
      if (key >= keyList[e])
        return e;
      if (key <= keyList[s])
        return s;
      if (e - s <= 1)
        return s;
C
Cary Xu 已提交
779 780

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

      // change start or end position
C
Cary Xu 已提交
804
      int mid = s - (s - e + 1)/2;
C
Cary Xu 已提交
805 806
      if (keyList[mid] < key)
        e = mid;
C
Cary Xu 已提交
807
      else if(keyList[mid] > key)
C
Cary Xu 已提交
808 809 810 811 812 813 814 815 816 817
        s = mid;
      else
        return mid;
    }
  }
}

int32_t getEndPosInDataBlock(STsdbReader* pReader, SBlockData* pBlockData, SDataBlk* pBlock, int32_t pos) {
  // NOTE: reverse the order to find the end position in data block
  int32_t endPos = -1;
C
Cary Xu 已提交
818
  bool asc = ASCENDING_TRAVERSE(pReader->order);
C
Cary Xu 已提交
819 820 821 822 823 824 825 826 827 828 829 830

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

  return endPos;
}

831
static int32_t copyBlockDataToSDataBlock(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo) {
832
  SReaderStatus*  pStatus = &pReader->status;
833
  SDataBlockIter* pBlockIter = &pStatus->blockIter;
H
Hongze Cheng 已提交
834

835
  SBlockData*         pBlockData = &pStatus->fileBlockData;
C
Cary Xu 已提交
836
  SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter);
H
Hongze Cheng 已提交
837
  SDataBlk*           pBlock = getCurrentBlock(pBlockIter);
H
Haojun Liao 已提交
838
  SSDataBlock*        pResBlock = pReader->pResBlock;
839
  int32_t             numOfOutputCols = blockDataGetNumOfCols(pResBlock);
H
Haojun Liao 已提交
840

H
Haojun Liao 已提交
841
  SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo;
842
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
H
Haojun Liao 已提交
843

H
Haojun Liao 已提交
844
  SColVal cv = {0};
845
  int64_t st = taosGetTimestampUs();
846 847
  bool    asc = ASCENDING_TRAVERSE(pReader->order);
  int32_t step = asc ? 1 : -1;
848

C
Cary Xu 已提交
849 850 851 852
  if (asc && pReader->window.skey <= pBlock->minKey.ts) {
    pDumpInfo->rowIndex = 0;
  } else if (!asc && pReader->window.ekey >= pBlock->maxKey.ts) {
    pDumpInfo->rowIndex = pBlock->nRow - 1;
853
  } else {
C
Cary Xu 已提交
854 855
    int32_t pos = asc? pBlock->nRow-1:0;
    int32_t order = (pReader->order == TSDB_ORDER_ASC)? TSDB_ORDER_DESC:TSDB_ORDER_ASC;
C
Cary Xu 已提交
856 857 858 859 860 861 862 863 864 865 866 867
    pDumpInfo->rowIndex = doBinarySearchKey(pBlockData->aTSKEY, pBlock->nRow, pos, pReader->window.skey, order);
  }

  // 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;
  int32_t remain = asc ? (endIndex - pDumpInfo->rowIndex) : (pDumpInfo->rowIndex - endIndex);
C
Cary Xu 已提交
868
  if (remain > pReader->capacity) { // output buffer check
869 870 871
    remain = pReader->capacity;
  }

C
Cary Xu 已提交
872 873
  int32_t rowIndex = 0;

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

885 886 887
    i += 1;
  }

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

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

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

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

930
  while (i < numOfOutputCols) {
931 932 933
    pColData = taosArrayGet(pResBlock->pDataBlock, i);
    colDataAppendNNULL(pColData, 0, remain);
    i += 1;
H
Haojun Liao 已提交
934
  }
H
Haojun Liao 已提交
935

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

C
Cary Xu 已提交
939 940 941 942
  if (pDumpInfo->rowIndex >= 0 && pDumpInfo->rowIndex < pBlock->nRow) {
    int64_t ts = pBlockData->aTSKEY[pDumpInfo->rowIndex];
    setBlockAllDumped(pDumpInfo, ts, pReader->order);
  } else {
C
Cary Xu 已提交
943
    int64_t k = asc? pBlock->maxKey.ts:pBlock->minKey.ts;
C
Cary Xu 已提交
944 945
    setBlockAllDumped(pDumpInfo, k, pReader->order);
  }
H
Haojun Liao 已提交
946

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

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

  return TSDB_CODE_SUCCESS;
}

959
static int32_t doLoadFileBlockData(STsdbReader* pReader, SDataBlockIter* pBlockIter, SBlockData* pBlockData) {
960 961
  int64_t st = taosGetTimestampUs();

962
  SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter);
963
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
964
  ASSERT(pBlockInfo != NULL);
965

H
Hongze Cheng 已提交
966 967
  SDataBlk* pBlock = getCurrentBlock(pBlockIter);
  int32_t   code = tsdbReadDataBlock(pReader->pFileReader, pBlock, pBlockData);
968 969 970
  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",
971
              pReader, pBlockIter->index, pBlockInfo->tbBlockIdx, pBlock->minKey.ts, pBlock->maxKey.ts, pBlock->nRow,
972 973 974
              tstrerror(code), pReader->idStr);
    return code;
  }
975

976
  double elapsedTime = (taosGetTimestampUs() - st) / 1000.0;
977

978 979 980 981
  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);
982 983 984

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

H
Haojun Liao 已提交
986
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
987
}
H
Hongze Cheng 已提交
988

H
Haojun Liao 已提交
989 990 991
static void cleanupBlockOrderSupporter(SBlockOrderSupporter* pSup) {
  taosMemoryFreeClear(pSup->numOfBlocksPerTable);
  taosMemoryFreeClear(pSup->indexPerTable);
H
Hongze Cheng 已提交
992

H
Haojun Liao 已提交
993 994 995 996
  for (int32_t i = 0; i < pSup->numOfTables; ++i) {
    SBlockOrderWrapper* pBlockInfo = pSup->pDataBlockInfo[i];
    taosMemoryFreeClear(pBlockInfo);
  }
H
Hongze Cheng 已提交
997

H
Haojun Liao 已提交
998 999
  taosMemoryFreeClear(pSup->pDataBlockInfo);
}
H
Hongze Cheng 已提交
1000

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

H
Haojun Liao 已提交
1004
  pSup->numOfBlocksPerTable = taosMemoryCalloc(1, sizeof(int32_t) * numOfTables);
1005 1006
  pSup->indexPerTable = taosMemoryCalloc(1, sizeof(int32_t) * numOfTables);
  pSup->pDataBlockInfo = taosMemoryCalloc(1, POINTER_BYTES * numOfTables);
H
Hongze Cheng 已提交
1007

H
Haojun Liao 已提交
1008 1009 1010 1011
  if (pSup->numOfBlocksPerTable == NULL || pSup->indexPerTable == NULL || pSup->pDataBlockInfo == NULL) {
    cleanupBlockOrderSupporter(pSup);
    return TSDB_CODE_OUT_OF_MEMORY;
  }
H
Hongze Cheng 已提交
1012

H
Haojun Liao 已提交
1013 1014
  return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
1015

H
Haojun Liao 已提交
1016
static int32_t fileDataBlockOrderCompar(const void* pLeft, const void* pRight, void* param) {
1017
  int32_t leftIndex = *(int32_t*)pLeft;
H
Haojun Liao 已提交
1018
  int32_t rightIndex = *(int32_t*)pRight;
H
Hongze Cheng 已提交
1019

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

H
Haojun Liao 已提交
1022 1023
  int32_t leftTableBlockIndex = pSupporter->indexPerTable[leftIndex];
  int32_t rightTableBlockIndex = pSupporter->indexPerTable[rightIndex];
H
Hongze Cheng 已提交
1024

H
Haojun Liao 已提交
1025 1026 1027 1028 1029 1030 1031
  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 已提交
1032

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

1036 1037 1038 1039
  return pLeftBlock->offset > pRightBlock->offset ? 1 : -1;
}

static int32_t doSetCurrentBlock(SDataBlockIter* pBlockIter) {
1040 1041 1042
  SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter);
  if (pBlockInfo != NULL) {
    STableBlockScanInfo* pScanInfo = taosHashGet(pBlockIter->pTableMap, &pBlockInfo->uid, sizeof(pBlockInfo->uid));
1043
    int32_t*             mapDataIndex = taosArrayGet(pScanInfo->pBlockList, pBlockInfo->tbBlockIdx);
H
Hongze Cheng 已提交
1044
    tMapDataGetItemByIdx(&pScanInfo->mapData, *mapDataIndex, &pBlockIter->block, tGetDataBlk);
1045
  }
1046 1047 1048 1049 1050 1051

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

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

1057
  pBlockIter->numOfBlocks = numOfBlocks;
1058
  taosArrayClear(pBlockIter->blockList);
1059
  pBlockIter->pTableMap = pReader->status.pTableMap;
1060

1061 1062
  // 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 已提交
1063

1064
  int64_t st = taosGetTimestampUs();
H
Haojun Liao 已提交
1065

1066
  SBlockOrderSupporter sup = {0};
1067
  int32_t              code = initBlockOrderSupporter(&sup, numOfTables);
1068 1069 1070
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }
H
Haojun Liao 已提交
1071

1072 1073 1074 1075 1076 1077 1078
  int32_t cnt = 0;
  void*   ptr = NULL;
  while (1) {
    ptr = taosHashIterate(pReader->status.pTableMap, ptr);
    if (ptr == NULL) {
      break;
    }
H
Haojun Liao 已提交
1079

1080 1081 1082 1083
    STableBlockScanInfo* pTableScanInfo = (STableBlockScanInfo*)ptr;
    if (pTableScanInfo->pBlockList == NULL || taosArrayGetSize(pTableScanInfo->pBlockList) == 0) {
      continue;
    }
H
Haojun Liao 已提交
1084

1085 1086
    size_t num = taosArrayGetSize(pTableScanInfo->pBlockList);
    sup.numOfBlocksPerTable[sup.numOfTables] = num;
H
Haojun Liao 已提交
1087

1088 1089 1090 1091 1092
    char* buf = taosMemoryMalloc(sizeof(SBlockOrderWrapper) * num);
    if (buf == NULL) {
      cleanupBlockOrderSupporter(&sup);
      return TSDB_CODE_TDB_OUT_OF_MEMORY;
    }
H
Haojun Liao 已提交
1093

1094
    sup.pDataBlockInfo[sup.numOfTables] = (SBlockOrderWrapper*)buf;
H
Hongze Cheng 已提交
1095
    SDataBlk block = {0};
1096 1097
    for (int32_t k = 0; k < num; ++k) {
      SBlockOrderWrapper wrapper = {0};
1098 1099

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

1102
      wrapper.uid = pTableScanInfo->uid;
1103
      wrapper.offset = block.aSubBlock[0].offset;
H
Haojun Liao 已提交
1104

1105 1106 1107 1108 1109 1110
      sup.pDataBlockInfo[sup.numOfTables][k] = wrapper;
      cnt++;
    }

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

1112
  ASSERT(numOfBlocks == cnt);
H
Haojun Liao 已提交
1113

1114
  // since there is only one table qualified, blocks are not sorted
1115 1116
  if (sup.numOfTables == 1) {
    for (int32_t i = 0; i < numOfBlocks; ++i) {
1117 1118
      SFileDataBlockInfo blockInfo = {.uid = sup.pDataBlockInfo[0][i].uid, .tbBlockIdx = i};
      taosArrayPush(pBlockIter->blockList, &blockInfo);
1119
    }
1120

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

1125
    pBlockIter->index = asc ? 0 : (numOfBlocks - 1);
H
Haojun Liao 已提交
1126
    cleanupBlockOrderSupporter(&sup);
1127
    doSetCurrentBlock(pBlockIter);
1128
    return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1129
  }
H
Haojun Liao 已提交
1130

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

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

1136 1137 1138 1139 1140
  SMultiwayMergeTreeInfo* pTree = NULL;
  uint8_t                 ret = tMergeTreeCreate(&pTree, sup.numOfTables, &sup, fileDataBlockOrderCompar);
  if (ret != TSDB_CODE_SUCCESS) {
    cleanupBlockOrderSupporter(&sup);
    return TSDB_CODE_TDB_OUT_OF_MEMORY;
H
Haojun Liao 已提交
1141
  }
H
Haojun Liao 已提交
1142

1143 1144 1145 1146
  int32_t numOfTotal = 0;
  while (numOfTotal < cnt) {
    int32_t pos = tMergeTreeGetChosenIndex(pTree);
    int32_t index = sup.indexPerTable[pos]++;
H
Haojun Liao 已提交
1147

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

1151 1152 1153 1154
    // 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 已提交
1155

1156 1157
    numOfTotal += 1;
    tMergeTreeAdjust(pTree, tMergeTreeGetAdjustIndex(pTree));
H
Haojun Liao 已提交
1158
  }
H
Haojun Liao 已提交
1159

1160
  int64_t et = taosGetTimestampUs();
H
Hongze Cheng 已提交
1161 1162
  tsdbDebug("%p %d data blocks access order completed, elapsed time:%.2f ms %s", pReader, numOfBlocks,
            (et - st) / 1000.0, pReader->idStr);
1163 1164
  cleanupBlockOrderSupporter(&sup);
  taosMemoryFree(pTree);
H
Haojun Liao 已提交
1165

1166
  pBlockIter->index = asc ? 0 : (numOfBlocks - 1);
1167 1168
  doSetCurrentBlock(pBlockIter);

1169
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1170
}
H
Hongze Cheng 已提交
1171

H
Haojun Liao 已提交
1172
static bool blockIteratorNext(SDataBlockIter* pBlockIter) {
1173 1174
  bool asc = ASCENDING_TRAVERSE(pBlockIter->order);

1175
  int32_t step = asc ? 1 : -1;
1176
  if ((pBlockIter->index >= pBlockIter->numOfBlocks - 1 && asc) || (pBlockIter->index <= 0 && (!asc))) {
1177 1178 1179
    return false;
  }

1180
  pBlockIter->index += step;
1181 1182
  doSetCurrentBlock(pBlockIter);

1183 1184 1185
  return true;
}

1186 1187 1188
/**
 * This is an two rectangles overlap cases.
 */
H
Hongze Cheng 已提交
1189
static int32_t dataBlockPartiallyRequired(STimeWindow* pWindow, SVersionRange* pVerRange, SDataBlk* pBlock) {
1190 1191
  return (pWindow->ekey < pBlock->maxKey.ts && pWindow->ekey >= pBlock->minKey.ts) ||
         (pWindow->skey > pBlock->minKey.ts && pWindow->skey <= pBlock->maxKey.ts) ||
H
Hongze Cheng 已提交
1192 1193
         (pVerRange->minVer > pBlock->minVer && pVerRange->minVer <= pBlock->maxVer) ||
         (pVerRange->maxVer < pBlock->maxVer && pVerRange->maxVer >= pBlock->minVer);
H
Haojun Liao 已提交
1194
}
H
Hongze Cheng 已提交
1195

H
Hongze Cheng 已提交
1196 1197
static SDataBlk* getNeighborBlockOfSameTable(SFileDataBlockInfo* pFBlockInfo, STableBlockScanInfo* pTableBlockScanInfo,
                                             int32_t* nextIndex, int32_t order) {
1198 1199 1200
  bool asc = ASCENDING_TRAVERSE(order);
  if (asc && pFBlockInfo->tbBlockIdx >= taosArrayGetSize(pTableBlockScanInfo->pBlockList) - 1) {
    return NULL;
1201 1202
  }

1203
  if (!asc && pFBlockInfo->tbBlockIdx == 0) {
1204 1205 1206
    return NULL;
  }

1207
  int32_t step = asc ? 1 : -1;
1208
  *nextIndex = pFBlockInfo->tbBlockIdx + step;
1209

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

H
Hongze Cheng 已提交
1213
  tMapDataGetItemByIdx(&pTableBlockScanInfo->mapData, *indexInMapdata, pBlock, tGetDataBlk);
1214
  return pBlock;
1215 1216 1217 1218 1219
}

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

1220
  int32_t step = ASCENDING_TRAVERSE(pBlockIter->order) ? 1 : -1;
1221 1222
  int32_t index = pBlockIter->index;

1223
  while (index < pBlockIter->numOfBlocks && index >= 0) {
1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235
    SFileDataBlockInfo* pFBlock = taosArrayGet(pBlockIter->blockList, index);
    if (pFBlock->uid == pFBlockInfo->uid && pFBlock->tbBlockIdx == pFBlockInfo->tbBlockIdx) {
      return index;
    }

    index += step;
  }

  ASSERT(0);
  return -1;
}

1236
static int32_t setFileBlockActiveInBlockIter(SDataBlockIter* pBlockIter, int32_t index, int32_t step) {
1237
  if (index < 0 || index >= pBlockIter->numOfBlocks) {
1238 1239 1240 1241
    return -1;
  }

  SFileDataBlockInfo fblock = *(SFileDataBlockInfo*)taosArrayGet(pBlockIter->blockList, index);
1242 1243 1244 1245 1246
  pBlockIter->index += step;

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

1248 1249 1250
    SFileDataBlockInfo* pBlockInfo = taosArrayGet(pBlockIter->blockList, pBlockIter->index);
    ASSERT(pBlockInfo->uid == fblock.uid && pBlockInfo->tbBlockIdx == fblock.tbBlockIdx);
  }
1251

1252
  doSetCurrentBlock(pBlockIter);
1253 1254 1255
  return TSDB_CODE_SUCCESS;
}

H
Hongze Cheng 已提交
1256
static bool overlapWithNeighborBlock(SDataBlk* pBlock, SDataBlk* pNeighbor, int32_t order) {
1257 1258 1259 1260 1261 1262
  // it is the last block in current file, no chance to overlap with neighbor blocks.
  if (ASCENDING_TRAVERSE(order)) {
    return pBlock->maxKey.ts == pNeighbor->minKey.ts;
  } else {
    return pBlock->minKey.ts == pNeighbor->maxKey.ts;
  }
H
Haojun Liao 已提交
1263
}
H
Hongze Cheng 已提交
1264

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

1268
  return (ascScan && (key.ts != TSKEY_INITIAL_VAL && key.ts <= pBlock->minKey.ts)) ||
1269
         (!ascScan && (key.ts != TSKEY_INITIAL_VAL && key.ts >= pBlock->maxKey.ts));
H
Haojun Liao 已提交
1270
}
H
Hongze Cheng 已提交
1271

H
Hongze Cheng 已提交
1272
static bool keyOverlapFileBlock(TSDBKEY key, SDataBlk* pBlock, SVersionRange* pVerRange) {
H
Hongze Cheng 已提交
1273 1274
  return (key.ts >= pBlock->minKey.ts && key.ts <= pBlock->maxKey.ts) && (pBlock->maxVer >= pVerRange->minVer) &&
         (pBlock->minVer <= pVerRange->maxVer);
H
Haojun Liao 已提交
1275 1276
}

H
Hongze Cheng 已提交
1277
static bool doCheckforDatablockOverlap(STableBlockScanInfo* pBlockScanInfo, const SDataBlk* pBlock) {
1278 1279 1280 1281 1282
  size_t num = taosArrayGetSize(pBlockScanInfo->delSkyline);

  for (int32_t i = pBlockScanInfo->fileDelIndex; i < num; i += 1) {
    TSDBKEY* p = taosArrayGet(pBlockScanInfo->delSkyline, i);
    if (p->ts >= pBlock->minKey.ts && p->ts <= pBlock->maxKey.ts) {
H
Hongze Cheng 已提交
1283
      if (p->version >= pBlock->minVer) {
1284 1285 1286
        return true;
      }
    } else if (p->ts < pBlock->minKey.ts) {  // p->ts < pBlock->minKey.ts
H
Hongze Cheng 已提交
1287
      if (p->version >= pBlock->minVer) {
1288 1289
        if (i < num - 1) {
          TSDBKEY* pnext = taosArrayGet(pBlockScanInfo->delSkyline, i + 1);
1290 1291
          if (pnext->ts >= pBlock->minKey.ts) {
            return true;
1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304
          }
        } else {  // it must be the last point
          ASSERT(p->version == 0);
        }
      }
    } else {  // (p->ts > pBlock->maxKey.ts) {
      return false;
    }
  }

  return false;
}

H
Hongze Cheng 已提交
1305
static bool overlapWithDelSkyline(STableBlockScanInfo* pBlockScanInfo, const SDataBlk* pBlock, int32_t order) {
1306 1307 1308 1309
  if (pBlockScanInfo->delSkyline == NULL) {
    return false;
  }

1310
  // ts is not overlap
1311
  TSDBKEY* pFirst = taosArrayGet(pBlockScanInfo->delSkyline, 0);
L
Liu Jicong 已提交
1312
  TSDBKEY* pLast = taosArrayGetLast(pBlockScanInfo->delSkyline);
1313 1314 1315 1316 1317
  if (pBlock->minKey.ts > pLast->ts || pBlock->maxKey.ts < pFirst->ts) {
    return false;
  }

  // version is not overlap
1318 1319 1320 1321
  if (ASCENDING_TRAVERSE(order)) {
    return doCheckforDatablockOverlap(pBlockScanInfo, pBlock);
  } else {
    int32_t index = pBlockScanInfo->fileDelIndex;
1322
    while (1) {
1323 1324 1325 1326 1327
      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;
1328 1329 1330
      }
    }

1331 1332
    return doCheckforDatablockOverlap(pBlockScanInfo, pBlock);
  }
1333 1334
}

C
Cary Xu 已提交
1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347
typedef struct {
  bool overlapWithNeighborBlock;
  bool hasDupTs;
  bool overlapWithDelInfo;
  bool overlapWithLastBlock;
  bool overlapWithKeyInBuf;
  bool partiallyRequired;
  bool moreThanCapcity;
} SDataBlockToLoadInfo;

static void getBlockToLoadInfo(SDataBlockToLoadInfo* pInfo, SFileDataBlockInfo* pBlockInfo, SDataBlk* pBlock,
                               STableBlockScanInfo* pScanInfo, TSDBKEY keyInBuf, SLastBlockReader* pLastBlockReader,
                               STsdbReader* pReader) {
H
Hongze Cheng 已提交
1348
  int32_t   neighborIndex = 0;
C
Cary Xu 已提交
1349
  SDataBlk* pNeighbor = getNeighborBlockOfSameTable(pBlockInfo, pScanInfo, &neighborIndex, pReader->order);
1350

1351
  // overlap with neighbor
1352
  if (pNeighbor) {
C
Cary Xu 已提交
1353
    pInfo->overlapWithNeighborBlock = overlapWithNeighborBlock(pBlock, pNeighbor, pReader->order);
1354
    taosMemoryFree(pNeighbor);
1355 1356
  }

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

1361 1362 1363
  if (hasDataInLastBlock(pLastBlockReader)) {
    int64_t tsLast = getCurrentKeyInLastBlock(pLastBlockReader);
    pInfo->overlapWithLastBlock = !(pBlock->maxKey.ts < tsLast || pBlock->minKey.ts > tsLast);
1364 1365
  }

C
Cary Xu 已提交
1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378 1379 1380
  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);
1381

C
Cary Xu 已提交
1382 1383 1384
  bool loadDataBlock =
      (info.overlapWithNeighborBlock || info.hasDupTs || info.partiallyRequired || info.overlapWithKeyInBuf ||
       info.moreThanCapcity || info.overlapWithDelInfo || info.overlapWithLastBlock);
1385 1386 1387 1388

  // log the reason why load the datablock for profile
  if (loadDataBlock) {
    tsdbDebug("%p uid:%" PRIu64
1389
              " need to load the datablock, overlapwithneighborblock:%d, hasDup:%d, partiallyRequired:%d, "
1390
              "overlapWithKey:%d, greaterThanBuf:%d, overlapWithDel:%d, overlapWithlastBlock:%d, %s",
C
Cary Xu 已提交
1391 1392 1393
              pReader, pBlockInfo->uid, info.overlapWithNeighborBlock, info.hasDupTs, info.partiallyRequired,
              info.overlapWithKeyInBuf, info.moreThanCapcity, info.overlapWithDelInfo, info.overlapWithLastBlock,
              pReader->idStr);
1394 1395 1396
  }

  return loadDataBlock;
H
Haojun Liao 已提交
1397 1398
}

C
Cary Xu 已提交
1399 1400 1401 1402 1403 1404 1405 1406 1407
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;
}

1408
static int32_t buildDataBlockFromBuf(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo, int64_t endKey) {
1409
  if (!(pBlockScanInfo->iiter.hasVal || pBlockScanInfo->iter.hasVal)) {
1410 1411
    return TSDB_CODE_SUCCESS;
  }
H
Haojun Liao 已提交
1412

1413 1414 1415
  SSDataBlock* pBlock = pReader->pResBlock;

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

1418
  blockDataUpdateTsWindow(pBlock, 0);
1419
  pBlock->info.uid = pBlockScanInfo->uid;
1420

1421
  setComposedBlockFlag(pReader, true);
1422

1423
  double elapsedTime = (taosGetTimestampUs() - st) / 1000.0;
S
Shengliang Guan 已提交
1424
  tsdbDebug("%p build data block from cache completed, elapsed time:%.2f ms, numOfRows:%d, brange:%" PRId64
1425 1426 1427
            " - %" PRId64 " %s",
            pReader, elapsedTime, pBlock->info.rows, pBlock->info.window.skey, pBlock->info.window.ekey,
            pReader->idStr);
1428 1429

  pReader->cost.buildmemBlock += elapsedTime;
H
Haojun Liao 已提交
1430 1431 1432
  return code;
}

1433 1434
static bool tryCopyDistinctRowFromFileBlock(STsdbReader* pReader, SBlockData* pBlockData, int64_t key,
                                            SFileBlockDumpInfo* pDumpInfo) {
1435 1436 1437 1438 1439
  // opt version
  // 1. it is not a border point
  // 2. the direct next point is not an duplicated timestamp
  if ((pDumpInfo->rowIndex < pDumpInfo->totalRows - 1 && pReader->order == TSDB_ORDER_ASC) ||
      (pDumpInfo->rowIndex > 0 && pReader->order == TSDB_ORDER_DESC)) {
1440
    int32_t step = pReader->order == TSDB_ORDER_ASC ? 1 : -1;
1441 1442

    int64_t nextKey = pBlockData->aTSKEY[pDumpInfo->rowIndex + step];
1443
    if (nextKey != key) {  // merge is not needed
1444
      doAppendRowFromFileBlock(pReader->pResBlock, pReader, pBlockData, pDumpInfo->rowIndex);
1445 1446 1447 1448 1449 1450 1451 1452
      pDumpInfo->rowIndex += step;
      return true;
    }
  }

  return false;
}

1453 1454 1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484
static bool nextRowFromLastBlocks(SLastBlockReader* pLastBlockReader, STableBlockScanInfo* pBlockScanInfo) {
  while (1) {
    bool hasVal = tMergeTreeNext(&pLastBlockReader->mergeTree);
    if (!hasVal) {
      return false;
    }

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

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

  return false;
}

H
Haojun Liao 已提交
1485 1486 1487 1488 1489 1490
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) {
    pReader->pSchema = metaGetTbTSchema(pReader->pTsdb->pVnode->pMeta, uid, -1);
  }

1491
  if (pReader->pSchema && sversion == pReader->pSchema->version) {
H
Haojun Liao 已提交
1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509
    return pReader->pSchema;
  }

  if (pReader->pMemSchema == NULL) {
    int32_t code =
        metaGetTbTSchemaEx(pReader->pTsdb->pVnode->pMeta, pReader->suid, uid, sversion, &pReader->pMemSchema);
    return pReader->pMemSchema;
  }

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

  taosMemoryFree(pReader->pMemSchema);
  int32_t code = metaGetTbTSchemaEx(pReader->pTsdb->pVnode->pMeta, pReader->suid, uid, sversion, &pReader->pMemSchema);
  return pReader->pMemSchema;
}

1510
static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo, TSDBROW* pRow,
1511 1512 1513 1514 1515 1516
                                     SIterInfo* pIter, int64_t key, SLastBlockReader* pLastBlockReader) {
  SRowMerger          merge = {0};
  STSRow*             pTSRow = NULL;
  SBlockData*         pBlockData = &pReader->status.fileBlockData;
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;

1517
  int64_t tsLast = INT64_MIN;
1518
  if (hasDataInLastBlock(pLastBlockReader)) {
1519 1520
    tsLast = getCurrentKeyInLastBlock(pLastBlockReader);
  }
1521

H
Hongze Cheng 已提交
1522 1523
  TSDBKEY k = TSDBROW_KEY(pRow);
  TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
1524

1525 1526
  int64_t minKey = 0;
  if (pReader->order == TSDB_ORDER_ASC) {
H
Hongze Cheng 已提交
1527
    minKey = INT64_MAX;  // chosen the minimum value
1528
    if (minKey > tsLast && hasDataInLastBlock(pLastBlockReader)) {
1529 1530
      minKey = tsLast;
    }
1531

1532 1533 1534
    if (minKey > k.ts) {
      minKey = k.ts;
    }
1535

1536
    if (minKey > key && hasDataInFileBlock(pBlockData, pDumpInfo)) {
1537 1538 1539 1540
      minKey = key;
    }
  } else {
    minKey = INT64_MIN;
1541
    if (minKey < tsLast && hasDataInLastBlock(pLastBlockReader)) {
1542 1543 1544 1545 1546 1547 1548
      minKey = tsLast;
    }

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

1549
    if (minKey < key && hasDataInFileBlock(pBlockData, pDumpInfo)) {
1550 1551
      minKey = key;
    }
1552 1553 1554 1555
  }

  bool init = false;

1556
  // ASC: file block ---> last block -----> imem -----> mem
H
Hongze Cheng 已提交
1557
  // DESC: mem -----> imem -----> last block -----> file block
1558 1559
  if (pReader->order == TSDB_ORDER_ASC) {
    if (minKey == key) {
1560
      init = true;
1561 1562
      tRowMergerInit(&merge, &fRow, pReader->pSchema);
      doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
1563 1564
    }

1565
    if (minKey == tsLast) {
1566
      TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
H
Haojun Liao 已提交
1567 1568 1569
      if (init) {
        tRowMerge(&merge, &fRow1);
      } else {
1570 1571 1572
        init = true;
        tRowMergerInit(&merge, &fRow1, pReader->pSchema);
      }
H
Haojun Liao 已提交
1573
      doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, &merge);
1574
    }
1575

1576
    if (minKey == k.ts) {
H
Haojun Liao 已提交
1577 1578 1579
      if (init) {
        tRowMerge(&merge, pRow);
      } else {
1580 1581 1582 1583 1584 1585 1586 1587 1588
        init = true;
        STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(pRow), pReader, pBlockScanInfo->uid);
        tRowMergerInit(&merge, pRow, pSchema);
      }
      doMergeRowsInBuf(pIter, pBlockScanInfo->uid, k.ts, pBlockScanInfo->delSkyline, &merge, pReader);
    }
  } else {
    if (minKey == k.ts) {
      init = true;
1589 1590
      STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(pRow), pReader, pBlockScanInfo->uid);
      tRowMergerInit(&merge, pRow, pSchema);
1591
      doMergeRowsInBuf(pIter, pBlockScanInfo->uid, k.ts, pBlockScanInfo->delSkyline, &merge, pReader);
1592 1593
    }

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

    if (minKey == key) {
H
Haojun Liao 已提交
1606 1607 1608
      if (init) {
        tRowMerge(&merge, &fRow);
      } else {
1609 1610 1611 1612 1613
        init = true;
        tRowMergerInit(&merge, &fRow, pReader->pSchema);
      }
      doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
    }
1614 1615
  }

1616 1617 1618 1619 1620
  int32_t code = tRowMergerGetRow(&merge, &pTSRow);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

1621 1622 1623 1624 1625 1626 1627
  doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo->uid);

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

1628 1629 1630
static int32_t doMergeFileBlockAndLastBlock(SLastBlockReader* pLastBlockReader, STsdbReader* pReader,
                                            STableBlockScanInfo* pBlockScanInfo, SBlockData* pBlockData,
                                            bool mergeBlockData) {
1631
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
H
Hongze Cheng 已提交
1632
  int64_t             tsLastBlock = getCurrentKeyInLastBlock(pLastBlockReader);
1633 1634 1635

  STSRow*    pTSRow = NULL;
  SRowMerger merge = {0};
1636
  TSDBROW    fRow = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
1637

1638 1639 1640 1641 1642 1643
  // only last block exists
  if ((!mergeBlockData) || (tsLastBlock != pBlockData->aTSKEY[pDumpInfo->rowIndex])) {
    if (tryCopyDistinctRowFromSttBlock(&fRow, pLastBlockReader, pBlockScanInfo, tsLastBlock, pReader)) {
      return TSDB_CODE_SUCCESS;
    } else {
      tRowMergerInit(&merge, &fRow, pReader->pSchema);
1644

1645 1646
      TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
      tRowMerge(&merge, &fRow1);
1647
      doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLastBlock, &merge);
1648

1649 1650 1651 1652
      int32_t code = tRowMergerGetRow(&merge, &pTSRow);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
1653

1654 1655 1656 1657 1658 1659 1660 1661
      doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo->uid);

      taosMemoryFree(pTSRow);
      tRowMergerClear(&merge);
    }
  } else {  // not merge block data
    tRowMergerInit(&merge, &fRow, pReader->pSchema);
    doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLastBlock, &merge);
H
Haojun Liao 已提交
1662
    ASSERT(mergeBlockData);
1663 1664

    // merge with block data if ts == key
H
Haojun Liao 已提交
1665
    if (tsLastBlock == pBlockData->aTSKEY[pDumpInfo->rowIndex]) {
1666 1667 1668 1669 1670 1671 1672 1673 1674 1675 1676 1677 1678
      doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
    }

    int32_t code = tRowMergerGetRow(&merge, &pTSRow);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

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

    taosMemoryFree(pTSRow);
    tRowMergerClear(&merge);
  }
1679 1680 1681 1682

  return TSDB_CODE_SUCCESS;
}

1683 1684
static int32_t mergeFileBlockAndLastBlock(STsdbReader* pReader, SLastBlockReader* pLastBlockReader, int64_t key,
                                          STableBlockScanInfo* pBlockScanInfo, SBlockData* pBlockData) {
1685 1686
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;

1687
  if (hasDataInFileBlock(pBlockData, pDumpInfo)) {
1688
    // no last block available, only data block exists
1689
    if (!hasDataInLastBlock(pLastBlockReader)) {
1690 1691 1692 1693 1694 1695 1696 1697 1698
      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)) {
1699
      if (key < ts) {  // imem, mem are all empty, file blocks (data blocks and last block) exist
1700 1701 1702 1703
        return mergeRowsInFileBlocks(pBlockData, pBlockScanInfo, key, pReader);
      } else if (key == ts) {
        STSRow*    pTSRow = NULL;
        SRowMerger merge = {0};
1704

1705 1706
        tRowMergerInit(&merge, &fRow, pReader->pSchema);
        doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
1707 1708 1709 1710

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

1711
        doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, ts, &merge);
1712

1713 1714 1715 1716 1717
        int32_t code = tRowMergerGetRow(&merge, &pTSRow);
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }

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

1720 1721
        taosMemoryFree(pTSRow);
        tRowMergerClear(&merge);
1722
        return code;
1723
      } else {
1724 1725
        ASSERT(0);
        return TSDB_CODE_SUCCESS;
1726
      }
1727
    } else {  // desc order
1728
      return doMergeFileBlockAndLastBlock(pLastBlockReader, pReader, pBlockScanInfo, pBlockData, true);
1729
    }
1730
  } else {  // only last block exists
1731
    return doMergeFileBlockAndLastBlock(pLastBlockReader, pReader, pBlockScanInfo, NULL, false);
H
Haojun Liao 已提交
1732
  }
1733 1734
}

1735 1736
static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo, SBlockData* pBlockData,
                                     SLastBlockReader* pLastBlockReader) {
1737 1738 1739 1740 1741 1742
  SRowMerger merge = {0};
  STSRow*    pTSRow = NULL;

  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
  SArray*             pDelList = pBlockScanInfo->delSkyline;

1743 1744
  TSDBROW* pRow = getValidMemRow(&pBlockScanInfo->iter, pDelList, pReader);
  TSDBROW* piRow = getValidMemRow(&pBlockScanInfo->iiter, pDelList, pReader);
1745 1746
  ASSERT(pRow != NULL && piRow != NULL);

1747
  int64_t tsLast = INT64_MIN;
1748 1749 1750
  if (hasDataInLastBlock(pLastBlockReader)) {
    tsLast = getCurrentKeyInLastBlock(pLastBlockReader);
  }
1751

1752
  int64_t key = hasDataInFileBlock(pBlockData, pDumpInfo)? pBlockData->aTSKEY[pDumpInfo->rowIndex]:INT64_MIN;
1753 1754 1755 1756

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

1757
  int64_t minKey = 0;
1758 1759 1760 1761 1762
  if (ASCENDING_TRAVERSE(pReader->order)) {
    minKey = INT64_MAX;  // let's find the minimum
    if (minKey > k.ts) {
      minKey = k.ts;
    }
1763

1764 1765 1766
    if (minKey > ik.ts) {
      minKey = ik.ts;
    }
1767

1768
    if (minKey > key && hasDataInFileBlock(pBlockData, pDumpInfo)) {
1769 1770
      minKey = key;
    }
1771

1772
    if (minKey > tsLast && hasDataInLastBlock(pLastBlockReader)) {
1773 1774 1775
      minKey = tsLast;
    }
  } else {
H
Hongze Cheng 已提交
1776
    minKey = INT64_MIN;  // let find the maximum ts value
1777 1778 1779 1780 1781 1782 1783 1784
    if (minKey < k.ts) {
      minKey = k.ts;
    }

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

1785
    if (minKey < key && hasDataInFileBlock(pBlockData, pDumpInfo)) {
1786 1787 1788
      minKey = key;
    }

1789
    if (minKey < tsLast && hasDataInLastBlock(pLastBlockReader)) {
1790 1791
      minKey = tsLast;
    }
1792 1793 1794 1795
  }

  bool init = false;

1796 1797 1798 1799
  // ASC: file block -----> last block -----> imem -----> mem
  // DESC: mem -----> imem -----> last block -----> file block
  if (ASCENDING_TRAVERSE(pReader->order)) {
    if (minKey == key) {
1800
      init = true;
1801 1802 1803
      TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
      tRowMergerInit(&merge, &fRow, pReader->pSchema);
      doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
1804 1805
    }

1806
    if (minKey == tsLast) {
1807
      TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
H
Haojun Liao 已提交
1808 1809 1810
      if (init) {
        tRowMerge(&merge, &fRow1);
      } else {
1811 1812 1813
        init = true;
        tRowMergerInit(&merge, &fRow1, pReader->pSchema);
      }
H
Haojun Liao 已提交
1814
      doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, &merge);
1815 1816 1817
    }

    if (minKey == ik.ts) {
H
Haojun Liao 已提交
1818 1819 1820
      if (init) {
        tRowMerge(&merge, piRow);
      } else {
1821 1822 1823 1824 1825
        init = true;
        STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(piRow), pReader, pBlockScanInfo->uid);
        tRowMergerInit(&merge, piRow, pSchema);
      }
      doMergeRowsInBuf(&pBlockScanInfo->iiter, pBlockScanInfo->uid, ik.ts, pBlockScanInfo->delSkyline, &merge, pReader);
1826 1827
    }

1828
    if (minKey == k.ts) {
H
Haojun Liao 已提交
1829 1830 1831
      if (init) {
        tRowMerge(&merge, pRow);
      } else {
1832 1833 1834 1835 1836 1837 1838 1839
        STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(pRow), pReader, pBlockScanInfo->uid);
        tRowMergerInit(&merge, pRow, pSchema);
      }
      doMergeRowsInBuf(&pBlockScanInfo->iter, pBlockScanInfo->uid, k.ts, pBlockScanInfo->delSkyline, &merge, pReader);
    }
  } else {
    if (minKey == k.ts) {
      init = true;
1840 1841
      STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(pRow), pReader, pBlockScanInfo->uid);
      tRowMergerInit(&merge, pRow, pSchema);
1842 1843 1844 1845
      doMergeRowsInBuf(&pBlockScanInfo->iter, pBlockScanInfo->uid, k.ts, pBlockScanInfo->delSkyline, &merge, pReader);
    }

    if (minKey == ik.ts) {
H
Haojun Liao 已提交
1846 1847 1848
      if (init) {
        tRowMerge(&merge, piRow);
      } else {
1849 1850 1851 1852 1853 1854 1855 1856
        init = true;
        STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(piRow), pReader, pBlockScanInfo->uid);
        tRowMergerInit(&merge, piRow, pSchema);
      }
      doMergeRowsInBuf(&pBlockScanInfo->iiter, pBlockScanInfo->uid, ik.ts, pBlockScanInfo->delSkyline, &merge, pReader);
    }

    if (minKey == tsLast) {
1857
      TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
H
Haojun Liao 已提交
1858 1859 1860
      if (init) {
        tRowMerge(&merge, &fRow1);
      } else {
1861 1862 1863
        init = true;
        tRowMergerInit(&merge, &fRow1, pReader->pSchema);
      }
H
Haojun Liao 已提交
1864
      doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, &merge);
1865 1866 1867
    }

    if (minKey == key) {
H
Haojun Liao 已提交
1868
      TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
1869 1870
      if (!init) {
        tRowMergerInit(&merge, &fRow, pReader->pSchema);
H
Haojun Liao 已提交
1871 1872
      } else {
        tRowMerge(&merge, &fRow);
1873 1874
      }
      doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
1875 1876 1877
    }
  }

1878 1879 1880 1881 1882
  int32_t code = tRowMergerGetRow(&merge, &pTSRow);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

1883 1884 1885 1886
  doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo->uid);

  taosMemoryFree(pTSRow);
  tRowMergerClear(&merge);
1887
  return code;
1888 1889
}

1890 1891 1892 1893 1894 1895 1896 1897 1898 1899 1900 1901 1902 1903 1904 1905 1906 1907 1908 1909 1910 1911 1912 1913 1914
static int32_t initMemDataIterator(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader) {
  if (pBlockScanInfo->iterInit) {
    return TSDB_CODE_SUCCESS;
  }

  int32_t code = TSDB_CODE_SUCCESS;

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

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

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

        tsdbDebug("%p uid:%" PRId64 ", check data in mem from skey:%" PRId64 ", order:%d, ts range in buf:%" PRId64
H
Hongze Cheng 已提交
1915
                  "-%" PRId64 " %s",
1916 1917 1918 1919 1920 1921 1922 1923 1924 1925 1926 1927 1928 1929 1930 1931 1932 1933 1934 1935
                  pReader, pBlockScanInfo->uid, startKey.ts, pReader->order, d->minKey, d->maxKey, pReader->idStr);
      } else {
        tsdbError("%p uid:%" PRId64 ", failed to create iterator for imem, code:%s, %s", pReader, pBlockScanInfo->uid,
                  tstrerror(code), pReader->idStr);
        return code;
      }
    }
  } else {
    tsdbDebug("%p uid:%" PRId64 ", no data in mem, %s", pReader, pBlockScanInfo->uid, pReader->idStr);
  }

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

        tsdbDebug("%p uid:%" PRId64 ", check data in imem from skey:%" PRId64 ", order:%d, ts range in buf:%" PRId64
H
Hongze Cheng 已提交
1936
                  "-%" PRId64 " %s",
1937 1938 1939 1940 1941 1942 1943 1944 1945 1946 1947 1948 1949 1950 1951 1952 1953
                  pReader, pBlockScanInfo->uid, startKey.ts, pReader->order, di->minKey, di->maxKey, pReader->idStr);
      } else {
        tsdbError("%p uid:%" PRId64 ", failed to create iterator for mem, code:%s, %s", pReader, pBlockScanInfo->uid,
                  tstrerror(code), pReader->idStr);
        return code;
      }
    }
  } else {
    tsdbDebug("%p uid:%" PRId64 ", no data in imem, %s", pReader, pBlockScanInfo->uid, pReader->idStr);
  }

  initDelSkylineIterator(pBlockScanInfo, pReader, d, di);

  pBlockScanInfo->iterInit = true;
  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
1954 1955
static bool isValidFileBlockRow(SBlockData* pBlockData, SFileBlockDumpInfo* pDumpInfo,
                                STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader) {
1956 1957 1958 1959 1960 1961 1962 1963
  // 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;
    }
  }

1964 1965 1966 1967 1968 1969 1970 1971 1972 1973 1974
  // 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;
  }

1975
  TSDBKEY k = {.ts = ts, .version = ver};
1976
  if (hasBeenDropped(pBlockScanInfo->delSkyline, &pBlockScanInfo->fileDelIndex, &k, pReader->order)) {
1977 1978 1979
    return false;
  }

1980 1981 1982
  return true;
}

1983
static bool initLastBlockReader(SLastBlockReader* pLBlockReader, STableBlockScanInfo* pScanInfo, STsdbReader* pReader) {
1984
  // the last block reader has been initialized for this table.
1985
  if (pLBlockReader->uid == pScanInfo->uid) {
1986 1987 1988
    return true;
  }

1989 1990
  if (pLBlockReader->uid != 0) {
    tMergeTreeClose(&pLBlockReader->mergeTree);
1991 1992
  }

1993 1994
  initMemDataIterator(pScanInfo, pReader);
  pLBlockReader->uid = pScanInfo->uid;
1995

H
Hongze Cheng 已提交
1996
  int32_t     step = ASCENDING_TRAVERSE(pLBlockReader->order) ? 1 : -1;
1997 1998 1999
  STimeWindow w = pLBlockReader->window;
  if (ASCENDING_TRAVERSE(pLBlockReader->order)) {
    w.skey = pScanInfo->lastKey + step;
2000
  } else {
2001
    w.ekey = pScanInfo->lastKey + step;
2002 2003
  }

2004
  int32_t code =
2005
      tMergeTreeOpen(&pLBlockReader->mergeTree, (pLBlockReader->order == TSDB_ORDER_DESC), pReader->pFileReader,
H
Haojun Liao 已提交
2006
                     pReader->suid, pScanInfo->uid, &w, &pLBlockReader->verRange, pLBlockReader->pInfo, pReader->idStr);
2007 2008 2009 2010
  if (code != TSDB_CODE_SUCCESS) {
    return false;
  }

2011
  return nextRowFromLastBlocks(pLBlockReader, pScanInfo);
2012 2013
}

2014
static int64_t getCurrentKeyInLastBlock(SLastBlockReader* pLastBlockReader) {
2015
  TSDBROW row = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
2016
  return TSDBROW_TS(&row);
2017 2018
}

H
Hongze Cheng 已提交
2019
static bool hasDataInLastBlock(SLastBlockReader* pLastBlockReader) { return pLastBlockReader->mergeTree.pIter != NULL; }
2020 2021 2022 2023 2024 2025 2026
bool hasDataInFileBlock(const SBlockData* pBlockData, const SFileBlockDumpInfo* pDumpInfo) {
  if (pBlockData->nRow > 0) {
    ASSERT(pBlockData->nRow == pDumpInfo->totalRows);
  }

  return pBlockData->nRow > 0 && (!pDumpInfo->allDumped);
}
2027

2028 2029
int32_t mergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pBlockScanInfo, int64_t key,
                              STsdbReader* pReader) {
2030 2031 2032 2033
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
  if (tryCopyDistinctRowFromFileBlock(pReader, pBlockData, key, pDumpInfo)) {
    return TSDB_CODE_SUCCESS;
  } else {
C
Cary Xu 已提交
2034 2035
    TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);

2036 2037 2038 2039 2040
    STSRow*    pTSRow = NULL;
    SRowMerger merge = {0};

    tRowMergerInit(&merge, &fRow, pReader->pSchema);
    doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
2041 2042 2043 2044 2045
    int32_t code = tRowMergerGetRow(&merge, &pTSRow);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

2046 2047 2048 2049 2050 2051 2052 2053
    doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo->uid);

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

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

2058
  int64_t key = (pBlockData->nRow > 0 && (!pDumpInfo->allDumped)) ? pBlockData->aTSKEY[pDumpInfo->rowIndex] : INT64_MIN;
2059
  if (pBlockScanInfo->iter.hasVal && pBlockScanInfo->iiter.hasVal) {
2060
    return doMergeMultiLevelRows(pReader, pBlockScanInfo, pBlockData, pLastBlockReader);
2061
  } else {
C
Cary Xu 已提交
2062 2063 2064 2065 2066 2067 2068 2069 2070
    TSDBROW *pRow = NULL, *piRow = NULL;
    if (pBlockScanInfo->iter.hasVal) {
      pRow = getValidMemRow(&pBlockScanInfo->iter, pBlockScanInfo->delSkyline, pReader);
    }

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

2071
    // imem + file + last block
2072
    if (pBlockScanInfo->iiter.hasVal) {
2073
      return doMergeBufAndFileRows(pReader, pBlockScanInfo, piRow, &pBlockScanInfo->iiter, key, pLastBlockReader);
2074 2075
    }

2076
    // mem + file + last block
2077
    if (pBlockScanInfo->iter.hasVal) {
2078
      return doMergeBufAndFileRows(pReader, pBlockScanInfo, pRow, &pBlockScanInfo->iter, key, pLastBlockReader);
H
Haojun Liao 已提交
2079
    }
2080

2081 2082
    // files data blocks + last block
    return mergeFileBlockAndLastBlock(pReader, pLastBlockReader, key, pBlockScanInfo, pBlockData);
2083 2084 2085
  }
}

2086
static int32_t buildComposedDataBlock(STsdbReader* pReader) {
2087 2088
  SSDataBlock* pResBlock = pReader->pResBlock;

H
Hongze Cheng 已提交
2089
  SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(&pReader->status.blockIter);
C
Cary Xu 已提交
2090 2091 2092
  SLastBlockReader*   pLastBlockReader = pReader->status.fileIter.pLastBlockReader;

  int64_t st = taosGetTimestampUs();
2093 2094 2095 2096

  STableBlockScanInfo* pBlockScanInfo = NULL;
  if (pBlockInfo != NULL) {
    pBlockScanInfo = taosHashGet(pReader->status.pTableMap, &pBlockInfo->uid, sizeof(pBlockInfo->uid));
C
Cary Xu 已提交
2097
    SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter);
C
Cary Xu 已提交
2098
    TSDBKEY keyInBuf = getCurrentKeyInBuf(pBlockScanInfo, pReader);
C
Cary Xu 已提交
2099 2100 2101

    // it is a clean block, load it directly
    if (isCleanFileDataBlock(pReader, pBlockInfo, pBlock, pBlockScanInfo, keyInBuf, pLastBlockReader)) {
2102 2103 2104 2105
      if (pReader->order == TSDB_ORDER_ASC || (pReader->order == TSDB_ORDER_DESC && (!hasDataInLastBlock(pLastBlockReader)))) {
        copyBlockDataToSDataBlock(pReader, pBlockScanInfo);
        goto _end;
      }
C
Cary Xu 已提交
2106 2107
    }
  } else {  // file blocks not exist
2108 2109 2110
    pBlockScanInfo = pReader->status.pTableIter;
  }

2111
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
2112 2113
  SBlockData*         pBlockData = &pReader->status.fileBlockData;
  int32_t             step = ASCENDING_TRAVERSE(pReader->order) ? 1 : -1;
2114

2115
  while (1) {
2116
    bool hasBlockData = false;
2117
    {
H
Haojun Liao 已提交
2118
      while (pBlockData->nRow > 0) {  // find the first qualified row in data block
2119 2120 2121 2122 2123
        if (isValidFileBlockRow(pBlockData, pDumpInfo, pBlockScanInfo, pReader)) {
          hasBlockData = true;
          break;
        }

2124 2125
        pDumpInfo->rowIndex += step;

H
Hongze Cheng 已提交
2126
        SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter);
2127
        if (pDumpInfo->rowIndex >= pBlock->nRow || pDumpInfo->rowIndex < 0) {
2128
          setBlockAllDumped(pDumpInfo, pBlock->maxKey.ts, pReader->order);
2129 2130 2131
          break;
        }
      }
2132
    }
2133

2134
    bool hasBlockLData = hasDataInLastBlock(pLastBlockReader);
2135

2136 2137 2138
    // no data in last block and block, no need to proceed.
    if ((hasBlockData == false) && (hasBlockLData == false)) {
      break;
2139 2140
    }

2141
    buildComposedDataBlockImpl(pReader, pBlockScanInfo, pBlockData, pLastBlockReader);
2142

2143
    // currently loaded file data block is consumed
2144
    if ((pBlockData->nRow > 0) && (pDumpInfo->rowIndex >= pBlockData->nRow || pDumpInfo->rowIndex < 0)) {
H
Hongze Cheng 已提交
2145
      SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter);
2146
      setBlockAllDumped(pDumpInfo, pBlock->maxKey.ts, pReader->order);
2147 2148 2149 2150 2151
      break;
    }

    if (pResBlock->info.rows >= pReader->capacity) {
      break;
2152 2153 2154
    }
  }

C
Cary Xu 已提交
2155
  _end:
2156
  pResBlock->info.uid = pBlockScanInfo->uid;
2157 2158
  blockDataUpdateTsWindow(pResBlock, 0);

2159
  setComposedBlockFlag(pReader, true);
H
Haojun Liao 已提交
2160 2161 2162 2163
  double el = (taosGetTimestampUs() - st)/1000.0;

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

2165 2166 2167 2168
  if (pResBlock->info.rows > 0) {
    tsdbDebug("%p uid:%" PRIu64 ", composed data block created, brange:%" PRIu64 "-%" PRIu64
              " rows:%d, elapsed time:%.2f ms %s",
              pReader, pBlockScanInfo->uid, pResBlock->info.window.skey, pResBlock->info.window.ekey,
H
Haojun Liao 已提交
2169
              pResBlock->info.rows, el, pReader->idStr);
2170
  }
2171

2172 2173 2174 2175 2176
  return TSDB_CODE_SUCCESS;
}

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

dengyihao's avatar
dengyihao 已提交
2177 2178
int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader, STbData* pMemTbData,
                               STbData* piMemTbData) {
2179 2180 2181
  if (pBlockScanInfo->delSkyline != NULL) {
    return TSDB_CODE_SUCCESS;
  }
2182

2183 2184 2185
  int32_t code = 0;
  STsdb*  pTsdb = pReader->pTsdb;

2186 2187
  SArray* pDelData = taosArrayInit(4, sizeof(SDelData));

H
Hongze Cheng 已提交
2188
  SDelFile* pDelFile = pReader->pReadSnap->fs.pDelFile;
2189 2190
  if (pDelFile) {
    SDelFReader* pDelFReader = NULL;
H
more  
Hongze Cheng 已提交
2191
    code = tsdbDelFReaderOpen(&pDelFReader, pDelFile, pTsdb);
2192
    if (code != TSDB_CODE_SUCCESS) {
2193 2194 2195 2196 2197
      goto _err;
    }

    SArray* aDelIdx = taosArrayInit(4, sizeof(SDelIdx));
    if (aDelIdx == NULL) {
2198
      tsdbDelFReaderClose(&pDelFReader);
2199 2200 2201
      goto _err;
    }

H
Hongze Cheng 已提交
2202
    code = tsdbReadDelIdx(pDelFReader, aDelIdx);
2203 2204 2205
    if (code != TSDB_CODE_SUCCESS) {
      taosArrayDestroy(aDelIdx);
      tsdbDelFReaderClose(&pDelFReader);
2206 2207
      goto _err;
    }
2208

2209 2210 2211
    SDelIdx  idx = {.suid = pReader->suid, .uid = pBlockScanInfo->uid};
    SDelIdx* pIdx = taosArraySearch(aDelIdx, &idx, tCmprDelIdx, TD_EQ);

H
Haojun Liao 已提交
2212
    if (pIdx != NULL) {
H
Hongze Cheng 已提交
2213
      code = tsdbReadDelData(pDelFReader, pIdx, pDelData);
2214 2215 2216 2217 2218 2219 2220
    }

    taosArrayDestroy(aDelIdx);
    tsdbDelFReaderClose(&pDelFReader);

    if (code != TSDB_CODE_SUCCESS) {
      goto _err;
2221
    }
2222
  }
2223

2224 2225 2226 2227 2228 2229 2230
  SDelData* p = NULL;
  if (pMemTbData != NULL) {
    p = pMemTbData->pHead;
    while (p) {
      taosArrayPush(pDelData, p);
      p = p->pNext;
    }
2231 2232
  }

2233 2234 2235 2236 2237 2238 2239 2240 2241 2242 2243 2244 2245 2246
  if (piMemTbData != NULL) {
    p = piMemTbData->pHead;
    while (p) {
      taosArrayPush(pDelData, p);
      p = p->pNext;
    }
  }

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

  taosArrayDestroy(pDelData);
dengyihao's avatar
dengyihao 已提交
2247 2248
  pBlockScanInfo->iter.index =
      ASCENDING_TRAVERSE(pReader->order) ? 0 : taosArrayGetSize(pBlockScanInfo->delSkyline) - 1;
2249 2250
  pBlockScanInfo->iiter.index = pBlockScanInfo->iter.index;
  pBlockScanInfo->fileDelIndex = pBlockScanInfo->iter.index;
2251
  pBlockScanInfo->lastBlockDelIndex = pBlockScanInfo->iter.index;
2252 2253
  return code;

2254 2255 2256
_err:
  taosArrayDestroy(pDelData);
  return code;
2257 2258
}

C
Cary Xu 已提交
2259
TSDBKEY getCurrentKeyInBuf(STableBlockScanInfo* pScanInfo, STsdbReader* pReader) {
H
Hongze Cheng 已提交
2260
  TSDBKEY  key = {.ts = TSKEY_INITIAL_VAL};
2261
  TSDBROW* pRow = getValidMemRow(&pScanInfo->iter, pScanInfo->delSkyline, pReader);
2262
  if (pRow != NULL) {
2263 2264 2265
    key = TSDBROW_KEY(pRow);
  }

2266
  pRow = getValidMemRow(&pScanInfo->iiter, pScanInfo->delSkyline, pReader);
2267
  if (pRow != NULL) {
2268 2269 2270 2271 2272 2273 2274 2275 2276
    TSDBKEY k = TSDBROW_KEY(pRow);
    if (key.ts > k.ts) {
      key = k;
    }
  }

  return key;
}

2277
static int32_t moveToNextFile(STsdbReader* pReader, SBlockNumber* pBlockNum) {
H
Haojun Liao 已提交
2278
  SReaderStatus* pStatus = &pReader->status;
2279
  pBlockNum->numOfBlocks = 0;
2280
  pBlockNum->numOfLastFiles = 0;
2281

2282
  size_t  numOfTables = taosHashGetSize(pReader->status.pTableMap);
2283
  SArray* pIndexList = taosArrayInit(numOfTables, sizeof(SBlockIdx));
H
Haojun Liao 已提交
2284 2285

  while (1) {
2286
    bool hasNext = filesetIteratorNext(&pStatus->fileIter, pReader);
2287
    if (!hasNext) {  // no data files on disk
H
Haojun Liao 已提交
2288 2289 2290
      break;
    }

H
Haojun Liao 已提交
2291
    taosArrayClear(pIndexList);
H
Haojun Liao 已提交
2292 2293
    int32_t code = doLoadBlockIndex(pReader, pReader->pFileReader, pIndexList);
    if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2294
      taosArrayDestroy(pIndexList);
H
Haojun Liao 已提交
2295 2296 2297
      return code;
    }

H
Hongze Cheng 已提交
2298
    if (taosArrayGetSize(pIndexList) > 0 || pReader->pFileReader->pSet->nSttF > 0) {
2299
      code = doLoadFileBlock(pReader, pIndexList, pBlockNum);
H
Haojun Liao 已提交
2300
      if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2301
        taosArrayDestroy(pIndexList);
H
Haojun Liao 已提交
2302 2303 2304
        return code;
      }

2305
      if (pBlockNum->numOfBlocks + pBlockNum->numOfLastFiles > 0) {
H
Haojun Liao 已提交
2306 2307 2308
        break;
      }
    }
2309

H
Haojun Liao 已提交
2310 2311 2312
    // no blocks in current file, try next files
  }

H
Haojun Liao 已提交
2313
  taosArrayDestroy(pIndexList);
H
Haojun Liao 已提交
2314 2315 2316
  return TSDB_CODE_SUCCESS;
}

2317
static int32_t uidComparFunc(const void* p1, const void* p2) {
2318 2319
  uint64_t pu1 = *(uint64_t*)p1;
  uint64_t pu2 = *(uint64_t*)p2;
2320 2321 2322
  if (pu1 == pu2) {
    return 0;
  } else {
2323
    return (pu1 < pu2) ? -1 : 1;
2324 2325
  }
}
2326

2327
static void extractOrderedTableUidList(SUidOrderCheckInfo* pOrderCheckInfo, SReaderStatus* pStatus) {
2328 2329 2330 2331
  int32_t index = 0;
  int32_t total = taosHashGetSize(pStatus->pTableMap);

  void* p = taosHashIterate(pStatus->pTableMap, NULL);
2332
  while (p != NULL) {
2333 2334 2335 2336 2337 2338 2339 2340
    STableBlockScanInfo* pScanInfo = p;
    pOrderCheckInfo->tableUidList[index++] = pScanInfo->uid;
    p = taosHashIterate(pStatus->pTableMap, p);
  }

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

2341
static int32_t initOrderCheckInfo(SUidOrderCheckInfo* pOrderCheckInfo, SReaderStatus* pStatus) {
2342 2343 2344 2345
  int32_t total = taosHashGetSize(pStatus->pTableMap);
  if (total == 0) {
    return TSDB_CODE_SUCCESS;
  }
2346

2347
  if (pOrderCheckInfo->tableUidList == NULL) {
2348 2349 2350 2351 2352 2353
    pOrderCheckInfo->currentIndex = 0;
    pOrderCheckInfo->tableUidList = taosMemoryMalloc(total * sizeof(uint64_t));
    if (pOrderCheckInfo->tableUidList == NULL) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }

2354
    extractOrderedTableUidList(pOrderCheckInfo, pStatus);
2355 2356 2357
    uint64_t uid = pOrderCheckInfo->tableUidList[0];
    pStatus->pTableIter = taosHashGet(pStatus->pTableMap, &uid, sizeof(uid));
  } else {
2358 2359
    if (pStatus->pTableIter == NULL) {  // it is the last block of a new file
      pOrderCheckInfo->currentIndex = 0;
2360 2361
      uint64_t uid = pOrderCheckInfo->tableUidList[pOrderCheckInfo->currentIndex];
      pStatus->pTableIter = taosHashGet(pStatus->pTableMap, &uid, sizeof(uid));
2362 2363

      // the tableMap has already updated
2364
      if (pStatus->pTableIter == NULL) {
2365
        void* p = taosMemoryRealloc(pOrderCheckInfo->tableUidList, total * sizeof(uint64_t));
2366 2367 2368 2369 2370 2371 2372 2373 2374
        if (p == NULL) {
          return TSDB_CODE_OUT_OF_MEMORY;
        }

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

        uid = pOrderCheckInfo->tableUidList[0];
        pStatus->pTableIter = taosHashGet(pStatus->pTableMap, &uid, sizeof(uid));
2375
      }
2376
    }
2377
  }
2378

2379 2380 2381
  return TSDB_CODE_SUCCESS;
}

2382
static bool moveToNextTable(SUidOrderCheckInfo* pOrderedCheckInfo, SReaderStatus* pStatus) {
2383 2384 2385 2386 2387 2388 2389 2390 2391 2392 2393 2394
  pOrderedCheckInfo->currentIndex += 1;
  if (pOrderedCheckInfo->currentIndex >= taosHashGetSize(pStatus->pTableMap)) {
    pStatus->pTableIter = NULL;
    return false;
  }

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

2395
static int32_t doLoadLastBlockSequentially(STsdbReader* pReader) {
2396
  SReaderStatus*    pStatus = &pReader->status;
2397 2398
  SLastBlockReader* pLastBlockReader = pStatus->fileIter.pLastBlockReader;

2399 2400
  SUidOrderCheckInfo* pOrderedCheckInfo = &pStatus->uidCheckInfo;
  int32_t             code = initOrderCheckInfo(pOrderedCheckInfo, pStatus);
2401
  if (code != TSDB_CODE_SUCCESS || (taosHashGetSize(pStatus->pTableMap) == 0)) {
2402 2403
    return code;
  }
2404

2405
  while (1) {
2406 2407
    // load the last data block of current table
    STableBlockScanInfo* pScanInfo = pStatus->pTableIter;
H
Hongze Cheng 已提交
2408
    bool                 hasVal = initLastBlockReader(pLastBlockReader, pScanInfo, pReader);
2409
    if (!hasVal) {
2410 2411
      bool hasNexTable = moveToNextTable(pOrderedCheckInfo, pStatus);
      if (!hasNexTable) {
2412 2413 2414
        return TSDB_CODE_SUCCESS;
      }
      continue;
2415 2416
    }

2417 2418 2419 2420 2421 2422 2423 2424
    code = doBuildDataBlock(pReader);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

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

2426
    // current table is exhausted, let's try next table
2427 2428
    bool hasNexTable = moveToNextTable(pOrderedCheckInfo, pStatus);
    if (!hasNexTable) {
2429 2430
      return TSDB_CODE_SUCCESS;
    }
2431 2432 2433
  }
}

2434
static int32_t doBuildDataBlock(STsdbReader* pReader) {
H
Hongze Cheng 已提交
2435 2436
  int32_t   code = TSDB_CODE_SUCCESS;
  SDataBlk* pBlock = NULL;
2437 2438 2439

  SReaderStatus*       pStatus = &pReader->status;
  SDataBlockIter*      pBlockIter = &pStatus->blockIter;
2440 2441 2442
  STableBlockScanInfo* pScanInfo = NULL;
  SFileDataBlockInfo*  pBlockInfo = getCurrentBlockInfo(pBlockIter);
  SLastBlockReader*    pLastBlockReader = pReader->status.fileIter.pLastBlockReader;
2443

2444
  if (pBlockInfo != NULL) {
2445 2446 2447 2448 2449 2450
    pScanInfo = taosHashGet(pReader->status.pTableMap, &pBlockInfo->uid, sizeof(pBlockInfo->uid));
  } else {
    pScanInfo = pReader->status.pTableIter;
  }

  if (pBlockInfo != NULL) {
2451
    pBlock = getCurrentBlock(pBlockIter);
2452 2453
  }

2454
  initLastBlockReader(pLastBlockReader, pScanInfo, pReader);
C
Cary Xu 已提交
2455
  TSDBKEY keyInBuf = getCurrentKeyInBuf(pScanInfo, pReader);
2456

2457 2458 2459
  if (pBlockInfo == NULL) {  // build data block from last data file
    ASSERT(pBlockIter->numOfBlocks == 0);
    code = buildComposedDataBlock(pReader);
C
Cary Xu 已提交
2460
  } else if (fileBlockShouldLoad(pReader, pBlockInfo, pBlock, pScanInfo, keyInBuf, pLastBlockReader)) {
2461 2462 2463
    tBlockDataReset(&pStatus->fileBlockData);
    code = tBlockDataInit(&pStatus->fileBlockData, pReader->suid, pScanInfo->uid, pReader->pSchema);
    if (code != TSDB_CODE_SUCCESS) {
2464
      return code;
2465
    }
2466

2467 2468 2469
    code = doLoadFileBlockData(pReader, pBlockIter, &pStatus->fileBlockData);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
2470 2471 2472
    }

    // build composed data block
2473
    code = buildComposedDataBlock(pReader);
C
Cary Xu 已提交
2474
  } else if (bufferDataInFileBlockGap(pReader->order, keyInBuf, pBlock)) {
2475
    // data in memory that are earlier than current file block
2476
    // rows in buffer should be less than the file block in asc, greater than file block in desc
2477
    int64_t endKey = (ASCENDING_TRAVERSE(pReader->order)) ? pBlock->minKey.ts : pBlock->maxKey.ts;
2478
    code = buildDataBlockFromBuf(pReader, pScanInfo, endKey);
2479 2480 2481 2482
  } else {
    if (hasDataInLastBlock(pLastBlockReader) && !ASCENDING_TRAVERSE(pReader->order)) {
      // only return the rows in last block
      int64_t tsLast = getCurrentKeyInLastBlock(pLastBlockReader);
H
Hongze Cheng 已提交
2483
      ASSERT(tsLast >= pBlock->maxKey.ts);
2484 2485 2486
      tBlockDataReset(&pReader->status.fileBlockData);

      code = buildComposedDataBlock(pReader);
H
Hongze Cheng 已提交
2487
    } else {  // whole block is required, return it directly
2488 2489 2490 2491 2492 2493 2494
      SDataBlockInfo* pInfo = &pReader->pResBlock->info;
      pInfo->rows = pBlock->nRow;
      pInfo->uid = pScanInfo->uid;
      pInfo->window = (STimeWindow){.skey = pBlock->minKey.ts, .ekey = pBlock->maxKey.ts};
      setComposedBlockFlag(pReader, false);
      setBlockAllDumped(&pStatus->fBlockDumpInfo, pBlock->maxKey.ts, pReader->order);
    }
2495 2496 2497 2498 2499
  }

  return code;
}

H
Haojun Liao 已提交
2500
static int32_t buildBlockFromBufferSequentially(STsdbReader* pReader) {
2501 2502
  SReaderStatus* pStatus = &pReader->status;

2503
  while (1) {
2504 2505 2506
    if (pStatus->pTableIter == NULL) {
      pStatus->pTableIter = taosHashIterate(pStatus->pTableMap, NULL);
      if (pStatus->pTableIter == NULL) {
H
Haojun Liao 已提交
2507
        return TSDB_CODE_SUCCESS;
2508 2509 2510 2511
      }
    }

    STableBlockScanInfo* pBlockScanInfo = pStatus->pTableIter;
2512
    initMemDataIterator(pBlockScanInfo, pReader);
2513

2514
    int64_t endKey = (ASCENDING_TRAVERSE(pReader->order)) ? INT64_MAX : INT64_MIN;
2515
    int32_t code = buildDataBlockFromBuf(pReader, pBlockScanInfo, endKey);
H
Haojun Liao 已提交
2516 2517 2518 2519
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

2520
    if (pReader->pResBlock->info.rows > 0) {
H
Haojun Liao 已提交
2521
      return TSDB_CODE_SUCCESS;
2522 2523 2524 2525 2526
    }

    // current table is exhausted, let's try the next table
    pStatus->pTableIter = taosHashIterate(pStatus->pTableMap, pStatus->pTableIter);
    if (pStatus->pTableIter == NULL) {
H
Haojun Liao 已提交
2527
      return TSDB_CODE_SUCCESS;
2528 2529 2530 2531
    }
  }
}

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

2536 2537 2538
  SReaderStatus* pStatus = &pReader->status;

  SFileBlockDumpInfo* pDumpInfo = &pStatus->fBlockDumpInfo;
2539 2540 2541

  pDumpInfo->totalRows = pBlock->nRow;
  pDumpInfo->allDumped = false;
2542
  pDumpInfo->rowIndex = ASCENDING_TRAVERSE(pReader->order) ? 0 : pBlock->nRow - 1;
2543 2544
}

2545
static int32_t initForFirstBlockInFile(STsdbReader* pReader, SDataBlockIter* pBlockIter) {
2546 2547
  SBlockNumber num = {0};

2548
  int32_t code = moveToNextFile(pReader, &num);
2549 2550 2551 2552 2553
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

  // all data files are consumed, try data in buffer
2554
  if (num.numOfBlocks + num.numOfLastFiles == 0) {
2555 2556 2557 2558 2559
    pReader->status.loadFromFile = false;
    return code;
  }

  // initialize the block iterator for a new fileset
2560 2561
  if (num.numOfBlocks > 0) {
    code = initBlockIterator(pReader, pBlockIter, num.numOfBlocks);
H
Hongze Cheng 已提交
2562
  } else {  // no block data, only last block exists
2563
    tBlockDataReset(&pReader->status.fileBlockData);
2564
    resetDataBlockIterator(pBlockIter, pReader->order);
2565
  }
2566 2567

  // set the correct start position according to the query time window
2568
  initBlockDumpInfo(pReader, pBlockIter);
2569 2570 2571
  return code;
}

2572
static bool fileBlockPartiallyRead(SFileBlockDumpInfo* pDumpInfo, bool asc) {
2573 2574
  return (!pDumpInfo->allDumped) &&
         ((pDumpInfo->rowIndex > 0 && asc) || (pDumpInfo->rowIndex < (pDumpInfo->totalRows - 1) && (!asc)));
2575 2576
}

2577
static int32_t buildBlockFromFiles(STsdbReader* pReader) {
H
Haojun Liao 已提交
2578
  int32_t code = TSDB_CODE_SUCCESS;
2579 2580
  bool    asc = ASCENDING_TRAVERSE(pReader->order);

2581 2582
  SDataBlockIter* pBlockIter = &pReader->status.blockIter;

2583
  if (pBlockIter->numOfBlocks == 0) {
H
Hongze Cheng 已提交
2584
  _begin:
2585 2586 2587 2588 2589
    code = doLoadLastBlockSequentially(pReader);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

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

2594
    // all data blocks are checked in this last block file, now let's try the next file
2595 2596 2597 2598 2599 2600 2601 2602
    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;
      }

2603
      // this file does not have data files, let's start check the last block file if exists
2604 2605 2606 2607 2608 2609 2610 2611 2612 2613 2614 2615 2616 2617 2618
      if (pBlockIter->numOfBlocks == 0) {
        goto _begin;
      }
    }

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

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

2619
  while (1) {
2620 2621
    SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;

2622
    if (fileBlockPartiallyRead(pDumpInfo, asc)) {  // file data block is partially loaded
2623
      code = buildComposedDataBlock(pReader);
2624 2625 2626 2627 2628 2629 2630
    } else {
      // current block are exhausted, try the next file block
      if (pDumpInfo->allDumped) {
        // try next data block in current file
        bool hasNext = blockIteratorNext(&pReader->status.blockIter);
        if (hasNext) {  // check for the next block in the block accessed order list
          initBlockDumpInfo(pReader, pBlockIter);
2631
        } else {
H
Haojun Liao 已提交
2632
          if (pReader->status.pCurrentFileset->nSttF > 0) {
2633 2634 2635 2636 2637 2638
            // data blocks in current file are exhausted, let's try the next file now
            tBlockDataReset(&pReader->status.fileBlockData);
            resetDataBlockIterator(pBlockIter, pReader->order);
            goto _begin;
          } else {
            code = initForFirstBlockInFile(pReader, pBlockIter);
2639

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

2645 2646 2647 2648
            // this file does not have blocks, let's start check the last block file
            if (pBlockIter->numOfBlocks == 0) {
              goto _begin;
            }
2649
          }
2650
        }
H
Haojun Liao 已提交
2651
      }
2652 2653

      code = doBuildDataBlock(pReader);
2654 2655
    }

2656 2657 2658 2659 2660 2661 2662 2663
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

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

2666 2667
static STsdb* getTsdbByRetentions(SVnode* pVnode, TSKEY winSKey, SRetention* retentions, const char* idStr,
                                  int8_t* pLevel) {
2668
  if (VND_IS_RSMA(pVnode)) {
2669
    int8_t  level = 0;
2670
    int64_t now = taosGetTimestamp(pVnode->config.tsdbCfg.precision);
2671
    int64_t offset = TSDB_TICK_PER_SECOND(pVnode->config.tsdbCfg.precision);
2672

2673
    for (int8_t i = 0; i < TSDB_RETENTION_MAX; ++i) {
2674 2675 2676 2677 2678 2679 2680
      SRetention* pRetention = retentions + level;
      if (pRetention->keep <= 0) {
        if (level > 0) {
          --level;
        }
        break;
      }
2681
      if ((now - pRetention->keep) <= (winSKey + offset)) {
2682 2683 2684 2685 2686
        break;
      }
      ++level;
    }

2687
    const char* str = (idStr != NULL) ? idStr : "";
2688 2689

    if (level == TSDB_RETENTION_L0) {
2690
      *pLevel = TSDB_RETENTION_L0;
C
Cary Xu 已提交
2691
      tsdbDebug("vgId:%d, rsma level %d is selected to query %s", TD_VID(pVnode), TSDB_RETENTION_L0, str);
2692 2693
      return VND_RSMA0(pVnode);
    } else if (level == TSDB_RETENTION_L1) {
2694
      *pLevel = TSDB_RETENTION_L1;
C
Cary Xu 已提交
2695
      tsdbDebug("vgId:%d, rsma level %d is selected to query %s", TD_VID(pVnode), TSDB_RETENTION_L1, str);
2696 2697
      return VND_RSMA1(pVnode);
    } else {
2698
      *pLevel = TSDB_RETENTION_L2;
C
Cary Xu 已提交
2699
      tsdbDebug("vgId:%d, rsma level %d is selected to query %s", TD_VID(pVnode), TSDB_RETENTION_L2, str);
2700 2701 2702 2703 2704 2705 2706
      return VND_RSMA2(pVnode);
    }
  }

  return VND_TSDB(pVnode);
}

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

  int64_t endVer = 0;
L
Liu Jicong 已提交
2711 2712
  if (pCond->endVersion ==
      -1) {  // user not specified end version, set current maximum version of vnode as the endVersion
H
Haojun Liao 已提交
2713 2714
    endVer = pVnode->state.applied;
  } else {
L
Liu Jicong 已提交
2715
    endVer = (pCond->endVersion > pVnode->state.applied) ? pVnode->state.applied : pCond->endVersion;
2716 2717
  }

H
Haojun Liao 已提交
2718
  return (SVersionRange){.minVer = startVer, .maxVer = endVer};
2719 2720
}

2721
bool hasBeenDropped(const SArray* pDelList, int32_t* index, TSDBKEY* pKey, int32_t order) {
2722 2723 2724 2725
  ASSERT(pKey != NULL);
  if (pDelList == NULL) {
    return false;
  }
L
Liu Jicong 已提交
2726 2727 2728
  size_t  num = taosArrayGetSize(pDelList);
  bool    asc = ASCENDING_TRAVERSE(order);
  int32_t step = asc ? 1 : -1;
2729

2730 2731 2732 2733 2734 2735
  if (asc) {
    if (*index >= num - 1) {
      TSDBKEY* last = taosArrayGetLast(pDelList);
      ASSERT(pKey->ts >= last->ts);

      if (pKey->ts > last->ts) {
2736
        return false;
2737 2738 2739
      } else if (pKey->ts == last->ts) {
        TSDBKEY* prev = taosArrayGet(pDelList, num - 2);
        return (prev->version >= pKey->version);
2740 2741
      }
    } else {
2742 2743 2744 2745 2746 2747 2748 2749 2750 2751 2752 2753 2754 2755 2756 2757 2758 2759 2760 2761 2762 2763 2764 2765 2766 2767 2768 2769 2770 2771
      TSDBKEY* pCurrent = taosArrayGet(pDelList, *index);
      TSDBKEY* pNext = taosArrayGet(pDelList, (*index) + 1);

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

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

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

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

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

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

      return false;
2772 2773
    }
  } else {
2774 2775
    if (*index <= 0) {
      TSDBKEY* pFirst = taosArrayGet(pDelList, 0);
2776

2777 2778 2779 2780 2781 2782 2783
      if (pKey->ts < pFirst->ts) {
        return false;
      } else if (pKey->ts == pFirst->ts) {
        return pFirst->version >= pKey->version;
      } else {
        ASSERT(0);
      }
2784
    } else {
2785 2786 2787 2788 2789 2790 2791 2792 2793 2794 2795 2796 2797 2798 2799 2800 2801 2802 2803 2804 2805 2806 2807 2808 2809 2810 2811
      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;
          }
        }
2812 2813 2814 2815 2816
      }

      return false;
    }
  }
2817 2818

  return false;
2819 2820
}

2821
TSDBROW* getValidMemRow(SIterInfo* pIter, const SArray* pDelList, STsdbReader* pReader) {
2822
  if (!pIter->hasVal) {
H
Haojun Liao 已提交
2823 2824
    return NULL;
  }
H
Hongze Cheng 已提交
2825

2826
  TSDBROW* pRow = tsdbTbDataIterGet(pIter->iter);
2827
  TSDBKEY  key = {.ts = pRow->pTSRow->ts, .version = pRow->version};
2828
  if (outOfTimeWindow(key.ts, &pReader->window)) {
2829
    pIter->hasVal = false;
H
Haojun Liao 已提交
2830 2831
    return NULL;
  }
H
Hongze Cheng 已提交
2832

2833
  // it is a valid data version
dengyihao's avatar
dengyihao 已提交
2834
  if ((key.version <= pReader->verRange.maxVer && key.version >= pReader->verRange.minVer) &&
2835
      (!hasBeenDropped(pDelList, &pIter->index, &key, pReader->order))) {
H
Haojun Liao 已提交
2836 2837
    return pRow;
  }
H
Hongze Cheng 已提交
2838

2839
  while (1) {
2840 2841
    pIter->hasVal = tsdbTbDataIterNext(pIter->iter);
    if (!pIter->hasVal) {
H
Haojun Liao 已提交
2842 2843
      return NULL;
    }
H
Hongze Cheng 已提交
2844

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

H
Haojun Liao 已提交
2847
    key = TSDBROW_KEY(pRow);
2848
    if (outOfTimeWindow(key.ts, &pReader->window)) {
2849
      pIter->hasVal = false;
H
Haojun Liao 已提交
2850 2851
      return NULL;
    }
H
Hongze Cheng 已提交
2852

dengyihao's avatar
dengyihao 已提交
2853
    if (key.version <= pReader->verRange.maxVer && key.version >= pReader->verRange.minVer &&
2854
        (!hasBeenDropped(pDelList, &pIter->index, &key, pReader->order))) {
H
Haojun Liao 已提交
2855 2856 2857 2858
      return pRow;
    }
  }
}
H
Hongze Cheng 已提交
2859

2860 2861
int32_t doMergeRowsInBuf(SIterInfo* pIter, uint64_t uid, int64_t ts, SArray* pDelList, SRowMerger* pMerger,
                         STsdbReader* pReader) {
H
Haojun Liao 已提交
2862
  while (1) {
2863 2864
    pIter->hasVal = tsdbTbDataIterNext(pIter->iter);
    if (!pIter->hasVal) {
H
Haojun Liao 已提交
2865 2866
      break;
    }
H
Hongze Cheng 已提交
2867

2868
    // data exists but not valid
2869
    TSDBROW* pRow = getValidMemRow(pIter, pDelList, pReader);
2870 2871 2872 2873 2874
    if (pRow == NULL) {
      break;
    }

    // ts is not identical, quit
H
Haojun Liao 已提交
2875
    TSDBKEY k = TSDBROW_KEY(pRow);
2876
    if (k.ts != ts) {
H
Haojun Liao 已提交
2877 2878 2879
      break;
    }

H
Haojun Liao 已提交
2880
    STSchema* pTSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(pRow), pReader, uid);
2881
    tRowMergerAdd(pMerger, pRow, pTSchema);
H
Haojun Liao 已提交
2882 2883 2884 2885 2886
  }

  return TSDB_CODE_SUCCESS;
}

2887
static int32_t doMergeRowsInFileBlockImpl(SBlockData* pBlockData, int32_t rowIndex, int64_t key, SRowMerger* pMerger,
2888
                                          SVersionRange* pVerRange, int32_t step) {
2889 2890
  while (pBlockData->aTSKEY[rowIndex] == key && rowIndex < pBlockData->nRow && rowIndex >= 0) {
    if (pBlockData->aVersion[rowIndex] > pVerRange->maxVer || pBlockData->aVersion[rowIndex] < pVerRange->minVer) {
2891
      rowIndex += step;
2892 2893 2894 2895 2896 2897 2898 2899 2900 2901 2902 2903 2904 2905 2906 2907
      continue;
    }

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

  return rowIndex;
}

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

H
Hongze Cheng 已提交
2908
static int32_t checkForNeighborFileBlock(STsdbReader* pReader, STableBlockScanInfo* pScanInfo, SDataBlk* pBlock,
2909 2910
                                         SFileDataBlockInfo* pFBlock, SRowMerger* pMerger, int64_t key,
                                         CHECK_FILEBLOCK_STATE* state) {
2911
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
2912
  SBlockData*         pBlockData = &pReader->status.fileBlockData;
2913

2914
  *state = CHECK_FILEBLOCK_QUIT;
2915
  int32_t step = ASCENDING_TRAVERSE(pReader->order) ? 1 : -1;
2916

H
Hongze Cheng 已提交
2917 2918
  int32_t   nextIndex = -1;
  SDataBlk* pNeighborBlock = getNeighborBlockOfSameTable(pFBlock, pScanInfo, &nextIndex, pReader->order);
2919
  if (pNeighborBlock == NULL) {  // do nothing
2920 2921 2922 2923
    return 0;
  }

  bool overlap = overlapWithNeighborBlock(pBlock, pNeighborBlock, pReader->order);
2924 2925
  taosMemoryFree(pNeighborBlock);

2926
  if (overlap) {  // load next block
2927
    SReaderStatus*  pStatus = &pReader->status;
2928 2929
    SDataBlockIter* pBlockIter = &pStatus->blockIter;

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

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

2937
    // 3. load the neighbor block, and set it to be the currently accessed file data block
H
Haojun Liao 已提交
2938
    tBlockDataReset(&pStatus->fileBlockData);
2939 2940 2941 2942 2943 2944
    int32_t code = tBlockDataInit(&pStatus->fileBlockData, pReader->suid, pFBlock->uid, pReader->pSchema);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

    code = doLoadFileBlockData(pReader, pBlockIter, &pStatus->fileBlockData);
2945 2946 2947 2948
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

2949
    // 4. check the data values
2950 2951 2952 2953
    initBlockDumpInfo(pReader, pBlockIter);

    pDumpInfo->rowIndex =
        doMergeRowsInFileBlockImpl(pBlockData, pDumpInfo->rowIndex, key, pMerger, &pReader->verRange, step);
H
Haojun Liao 已提交
2954
    if (pDumpInfo->rowIndex >= pDumpInfo->totalRows) {
2955 2956 2957 2958 2959 2960 2961
      *state = CHECK_FILEBLOCK_CONT;
    }
  }

  return TSDB_CODE_SUCCESS;
}

2962 2963
int32_t doMergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pScanInfo, STsdbReader* pReader,
                                SRowMerger* pMerger) {
2964 2965
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;

2966
  bool    asc = ASCENDING_TRAVERSE(pReader->order);
2967
  int64_t key = pBlockData->aTSKEY[pDumpInfo->rowIndex];
2968
  int32_t step = asc ? 1 : -1;
2969

2970
  pDumpInfo->rowIndex += step;
2971
  if ((pDumpInfo->rowIndex <= pBlockData->nRow - 1 && asc) || (pDumpInfo->rowIndex >= 0 && !asc)) {
2972 2973 2974
    pDumpInfo->rowIndex =
        doMergeRowsInFileBlockImpl(pBlockData, pDumpInfo->rowIndex, key, pMerger, &pReader->verRange, step);
  }
2975

2976 2977 2978 2979
  // 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;
2980

2981
      SFileDataBlockInfo* pFileBlockInfo = getCurrentBlockInfo(&pReader->status.blockIter);
H
Hongze Cheng 已提交
2982
      SDataBlk*           pCurrentBlock = getCurrentBlock(&pReader->status.blockIter);
2983 2984 2985
      checkForNeighborFileBlock(pReader, pScanInfo, pCurrentBlock, pFileBlockInfo, pMerger, key, &st);
      if (st == CHECK_FILEBLOCK_QUIT) {
        break;
2986
      }
2987
    }
H
Haojun Liao 已提交
2988
  }
2989

H
Haojun Liao 已提交
2990 2991 2992
  return TSDB_CODE_SUCCESS;
}

H
Hongze Cheng 已提交
2993 2994
int32_t doMergeRowsInLastBlock(SLastBlockReader* pLastBlockReader, STableBlockScanInfo* pScanInfo, int64_t ts,
                               SRowMerger* pMerger) {
H
Haojun Liao 已提交
2995
  pScanInfo->lastKey = ts;
2996
  while (nextRowFromLastBlocks(pLastBlockReader, pScanInfo)) {
2997 2998
    int64_t next1 = getCurrentKeyInLastBlock(pLastBlockReader);
    if (next1 == ts) {
2999
      TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
3000 3001 3002 3003 3004 3005 3006 3007 3008
      tRowMerge(pMerger, &fRow1);
    } else {
      break;
    }
  }

  return TSDB_CODE_SUCCESS;
}

3009 3010
int32_t doMergeMemTableMultiRows(TSDBROW* pRow, uint64_t uid, SIterInfo* pIter, SArray* pDelList, STSRow** pTSRow,
                                 STsdbReader* pReader, bool* freeTSRow) {
H
Haojun Liao 已提交
3011
  TSDBROW* pNextRow = NULL;
3012
  TSDBROW  current = *pRow;
3013

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

3017 3018 3019
    if (!pIter->hasVal) {
      *pTSRow = current.pTSRow;
      *freeTSRow = false;
3020
      return TSDB_CODE_SUCCESS;
3021
    } else {  // has next point in mem/imem
3022
      pNextRow = getValidMemRow(pIter, pDelList, pReader);
3023 3024 3025
      if (pNextRow == NULL) {
        *pTSRow = current.pTSRow;
        *freeTSRow = false;
3026
        return TSDB_CODE_SUCCESS;
3027 3028
      }

H
Haojun Liao 已提交
3029
      if (current.pTSRow->ts != pNextRow->pTSRow->ts) {
3030 3031
        *pTSRow = current.pTSRow;
        *freeTSRow = false;
3032
        return TSDB_CODE_SUCCESS;
3033
      }
3034
    }
3035 3036
  }

3037 3038
  SRowMerger merge = {0};

3039
  // get the correct schema for data in memory
H
Haojun Liao 已提交
3040
  STSchema* pTSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(&current), pReader, uid);
H
Haojun Liao 已提交
3041

3042 3043
  if (pReader->pSchema == NULL) {
    pReader->pSchema = pTSchema;
3044
  }
H
Haojun Liao 已提交
3045

H
Haojun Liao 已提交
3046 3047 3048 3049 3050 3051
  tRowMergerInit2(&merge, pReader->pSchema, &current, pTSchema);

  STSchema* pTSchema1 = doGetSchemaForTSRow(TSDBROW_SVERSION(pNextRow), pReader, uid);
  tRowMergerAdd(&merge, pNextRow, pTSchema1);

  doMergeRowsInBuf(pIter, uid, current.pTSRow->ts, pDelList, &merge, pReader);
3052 3053 3054 3055
  int32_t code = tRowMergerGetRow(&merge, pTSRow);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }
M
Minglei Jin 已提交
3056

3057
  tRowMergerClear(&merge);
3058
  *freeTSRow = true;
3059
  return TSDB_CODE_SUCCESS;
3060 3061
}

3062
int32_t doMergeMemIMemRows(TSDBROW* pRow, TSDBROW* piRow, STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader,
H
Hongze Cheng 已提交
3063
                           STSRow** pTSRow) {
H
Haojun Liao 已提交
3064 3065
  SRowMerger merge = {0};

3066 3067 3068
  TSDBKEY k = TSDBROW_KEY(pRow);
  TSDBKEY ik = TSDBROW_KEY(piRow);

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

H
Haojun Liao 已提交
3072
    tRowMergerInit(&merge, piRow, pSchema);
3073
    doMergeRowsInBuf(&pBlockScanInfo->iiter, pBlockScanInfo->uid, ik.ts, pBlockScanInfo->delSkyline, &merge, pReader);
3074

3075
    tRowMerge(&merge, pRow);
3076
    doMergeRowsInBuf(&pBlockScanInfo->iter, pBlockScanInfo->uid, k.ts, pBlockScanInfo->delSkyline, &merge, pReader);
3077
  } else {
H
Haojun Liao 已提交
3078
    STSchema* pSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(pRow), pReader, pBlockScanInfo->uid);
3079

H
Haojun Liao 已提交
3080
    tRowMergerInit(&merge, pRow, pSchema);
3081
    doMergeRowsInBuf(&pBlockScanInfo->iter, pBlockScanInfo->uid, k.ts, pBlockScanInfo->delSkyline, &merge, pReader);
3082 3083

    tRowMerge(&merge, piRow);
3084
    doMergeRowsInBuf(&pBlockScanInfo->iiter, pBlockScanInfo->uid, k.ts, pBlockScanInfo->delSkyline, &merge, pReader);
3085
  }
3086

3087 3088
  int32_t code = tRowMergerGetRow(&merge, pTSRow);
  return code;
3089 3090
}

3091 3092
int32_t tsdbGetNextRowInMem(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader, STSRow** pTSRow, int64_t endKey,
                            bool* freeTSRow) {
3093 3094
  TSDBROW* pRow = getValidMemRow(&pBlockScanInfo->iter, pBlockScanInfo->delSkyline, pReader);
  TSDBROW* piRow = getValidMemRow(&pBlockScanInfo->iiter, pBlockScanInfo->delSkyline, pReader);
dengyihao's avatar
dengyihao 已提交
3095
  SArray*  pDelList = pBlockScanInfo->delSkyline;
3096
  uint64_t uid = pBlockScanInfo->uid;
H
Haojun Liao 已提交
3097

3098 3099
  // todo refactor
  bool asc = ASCENDING_TRAVERSE(pReader->order);
3100
  if (pBlockScanInfo->iter.hasVal) {
3101 3102 3103 3104 3105 3106
    TSDBKEY k = TSDBROW_KEY(pRow);
    if ((k.ts >= endKey && asc) || (k.ts <= endKey && !asc)) {
      pRow = NULL;
    }
  }

3107
  if (pBlockScanInfo->iiter.hasVal) {
3108 3109 3110 3111 3112 3113
    TSDBKEY k = TSDBROW_KEY(piRow);
    if ((k.ts >= endKey && asc) || (k.ts <= endKey && !asc)) {
      piRow = NULL;
    }
  }

3114
  if (pBlockScanInfo->iter.hasVal && pBlockScanInfo->iiter.hasVal && pRow != NULL && piRow != NULL) {
3115
    TSDBKEY k = TSDBROW_KEY(pRow);
3116
    TSDBKEY ik = TSDBROW_KEY(piRow);
H
Haojun Liao 已提交
3117

3118
    int32_t code = TSDB_CODE_SUCCESS;
3119 3120
    if (ik.ts != k.ts) {
      if (((ik.ts < k.ts) && asc) || ((ik.ts > k.ts) && (!asc))) {  // ik.ts < k.ts
3121
        code = doMergeMemTableMultiRows(piRow, uid, &pBlockScanInfo->iiter, pDelList, pTSRow, pReader, freeTSRow);
3122
      } else if (((k.ts < ik.ts) && asc) || ((k.ts > ik.ts) && (!asc))) {
3123
        code = doMergeMemTableMultiRows(pRow, uid, &pBlockScanInfo->iter, pDelList, pTSRow, pReader, freeTSRow);
3124
      }
3125
    } else {  // ik.ts == k.ts
3126
      *freeTSRow = true;
3127 3128 3129 3130
      code = doMergeMemIMemRows(pRow, piRow, pBlockScanInfo, pReader, pTSRow);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
H
Haojun Liao 已提交
3131
    }
3132

3133
    return code;
H
Haojun Liao 已提交
3134 3135
  }

3136
  if (pBlockScanInfo->iter.hasVal && pRow != NULL) {
H
Hongze Cheng 已提交
3137 3138
    return doMergeMemTableMultiRows(pRow, pBlockScanInfo->uid, &pBlockScanInfo->iter, pDelList, pTSRow, pReader,
                                    freeTSRow);
H
Haojun Liao 已提交
3139 3140
  }

3141
  if (pBlockScanInfo->iiter.hasVal && piRow != NULL) {
3142
    return doMergeMemTableMultiRows(piRow, uid, &pBlockScanInfo->iiter, pDelList, pTSRow, pReader, freeTSRow);
H
Haojun Liao 已提交
3143 3144 3145 3146 3147
  }

  return TSDB_CODE_SUCCESS;
}

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

3152
  SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo;
3153
  STSchema*           pSchema = doGetSchemaForTSRow(pTSRow->sver, pReader, uid);
3154

3155
  SColVal colVal = {0};
3156
  int32_t i = 0, j = 0;
H
Haojun Liao 已提交
3157

3158 3159 3160 3161 3162 3163 3164 3165 3166 3167 3168
  SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
  if (pColInfoData->info.colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
    colDataAppend(pColInfoData, numOfRows, (const char*)&pTSRow->ts, false);
    i += 1;
  }

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

    if (colId == pSchema->columns[j].colId) {
H
Haojun Liao 已提交
3169
      tTSRowGetVal(pTSRow, pSchema, j, &colVal);
3170 3171 3172 3173 3174 3175 3176 3177
      doCopyColVal(pColInfoData, numOfRows, i, &colVal, pSupInfo);
      i += 1;
      j += 1;
    } else if (colId < pSchema->columns[j].colId) {
      colDataAppendNULL(pColInfoData, numOfRows);
      i += 1;
    } else if (colId > pSchema->columns[j].colId) {
      j += 1;
3178
    }
3179 3180
  }

3181
  // set null value since current column does not exist in the "pSchema"
3182
  while (i < numOfCols) {
3183 3184 3185 3186 3187
    pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
    colDataAppendNULL(pColInfoData, numOfRows);
    i += 1;
  }

3188 3189 3190 3191
  pBlock->info.rows += 1;
  return TSDB_CODE_SUCCESS;
}

H
Hongze Cheng 已提交
3192 3193
int32_t doAppendRowFromFileBlock(SSDataBlock* pResBlock, STsdbReader* pReader, SBlockData* pBlockData,
                                 int32_t rowIndex) {
3194 3195 3196 3197 3198 3199 3200 3201
  int32_t i = 0, j = 0;
  int32_t outputRowIndex = pResBlock->info.rows;

  SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo;

  SColumnInfoData* pColData = taosArrayGet(pResBlock->pDataBlock, i);
  if (pColData->info.colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
    colDataAppendInt64(pColData, outputRowIndex, &pBlockData->aTSKEY[rowIndex]);
3202
    i += 1;
3203 3204 3205
  }

  SColVal cv = {0};
C
Cary Xu 已提交
3206 3207
  int32_t numOfInputCols = pBlockData->aIdx->size;
  int32_t numOfOutputCols = pResBlock->pDataBlock->size;
3208

3209
  while (i < numOfOutputCols && j < numOfInputCols) {
C
Cary Xu 已提交
3210
    SColumnInfoData* pCol = TARRAY_GET_ELEM(pResBlock->pDataBlock, i);
3211
    SColData*        pData = tBlockDataGetColDataByIdx(pBlockData, j);
3212

3213 3214 3215 3216 3217
    if (pData->cid < pCol->info.colId) {
      j += 1;
      continue;
    }

3218
    if (pData->cid == pCol->info.colId) {
3219 3220
      tColDataGetValue(pData, rowIndex, &cv);
      doCopyColVal(pCol, outputRowIndex, i, &cv, pSupInfo);
3221
      j += 1;
3222
    } else if (pData->cid > pCol->info.colId) {  // the specified column does not exist in file block, fill with null data
3223 3224 3225 3226 3227 3228 3229 3230
      colDataAppendNULL(pCol, outputRowIndex);
    }

    i += 1;
  }

  while (i < numOfOutputCols) {
    SColumnInfoData* pCol = taosArrayGet(pResBlock->pDataBlock, i);
3231
    colDataAppendNULL(pCol, outputRowIndex);
3232 3233 3234 3235 3236 3237 3238
    i += 1;
  }

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

3239 3240
int32_t buildDataBlockFromBufImpl(STableBlockScanInfo* pBlockScanInfo, int64_t endKey, int32_t capacity,
                                  STsdbReader* pReader) {
H
Haojun Liao 已提交
3241 3242 3243 3244
  SSDataBlock* pBlock = pReader->pResBlock;

  do {
    STSRow* pTSRow = NULL;
3245
    bool    freeTSRow = false;
3246
    tsdbGetNextRowInMem(pBlockScanInfo, pReader, &pTSRow, endKey, &freeTSRow);
3247 3248
    if (pTSRow == NULL) {
      break;
H
Haojun Liao 已提交
3249 3250
    }

H
Haojun Liao 已提交
3251
    doAppendRowFromTSRow(pBlock, pReader, pTSRow, pBlockScanInfo->uid);
3252 3253 3254
    if (freeTSRow) {
      taosMemoryFree(pTSRow);
    }
H
Haojun Liao 已提交
3255 3256

    // no data in buffer, return immediately
3257
    if (!(pBlockScanInfo->iter.hasVal || pBlockScanInfo->iiter.hasVal)) {
H
Haojun Liao 已提交
3258 3259 3260
      break;
    }

3261
    if (pBlock->info.rows >= capacity) {
H
Haojun Liao 已提交
3262 3263 3264 3265
      break;
    }
  } while (1);

3266
  ASSERT(pBlock->info.rows <= capacity);
H
Haojun Liao 已提交
3267 3268
  return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
3269

3270
// todo refactor, use arraylist instead
H
Hongze Cheng 已提交
3271
int32_t tsdbSetTableId(STsdbReader* pReader, int64_t uid) {
3272 3273 3274
  ASSERT(pReader != NULL);
  taosHashClear(pReader->status.pTableMap);

3275
  STableBlockScanInfo info = {.lastKey = 0, .uid = uid};
3276
  taosHashPut(pReader->status.pTableMap, &info.uid, sizeof(uint64_t), &info, sizeof(info));
H
Hongze Cheng 已提交
3277 3278 3279
  return TDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
3280 3281 3282 3283 3284 3285
void* tsdbGetIdx(SMeta* pMeta) {
  if (pMeta == NULL) {
    return NULL;
  }
  return metaGetIdx(pMeta);
}
dengyihao's avatar
dengyihao 已提交
3286

dengyihao's avatar
dengyihao 已提交
3287 3288 3289 3290 3291 3292
void* tsdbGetIvtIdx(SMeta* pMeta) {
  if (pMeta == NULL) {
    return NULL;
  }
  return metaGetIvtIdx(pMeta);
}
L
Liu Jicong 已提交
3293

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

3296 3297 3298 3299 3300 3301 3302 3303 3304 3305 3306 3307 3308 3309 3310 3311

static int32_t doOpenReaderImpl(STsdbReader* pReader) {
  SDataBlockIter* pBlockIter = &pReader->status.blockIter;

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

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

H
refact  
Hongze Cheng 已提交
3312
// ====================================== EXPOSED APIs ======================================
3313 3314
int32_t tsdbReaderOpen(SVnode* pVnode, SQueryTableDataCond* pCond, SArray* pTableList, STsdbReader** ppReader,
                       const char* idstr) {
3315 3316 3317 3318 3319 3320
  STimeWindow window = pCond->twindows;
  if (pCond->type == TIMEWINDOW_RANGE_EXTERNAL) {
    pCond->twindows.skey += 1;
    pCond->twindows.ekey -= 1;
  }

3321 3322
  int32_t code = tsdbReaderCreate(pVnode, pCond, ppReader, 4096, idstr);
  if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
3323 3324
    goto _err;
  }
H
Hongze Cheng 已提交
3325

3326
  // check for query time window
H
Haojun Liao 已提交
3327
  STsdbReader* pReader = *ppReader;
3328
  if (isEmptyQueryTimeWindow(&pReader->window) && pCond->type == TIMEWINDOW_RANGE_CONTAINED) {
H
Haojun Liao 已提交
3329 3330 3331
    tsdbDebug("%p query window not overlaps with the data set, no result returned, %s", pReader, pReader->idStr);
    return TSDB_CODE_SUCCESS;
  }
H
Hongze Cheng 已提交
3332

3333 3334
  if (pCond->type == TIMEWINDOW_RANGE_EXTERNAL) {
    // update the SQueryTableDataCond to create inner reader
3335
    int32_t order = pCond->order;
3336
    if (order == TSDB_ORDER_ASC) {
3337
      pCond->twindows.ekey = window.skey;
3338 3339 3340
      pCond->twindows.skey = INT64_MIN;
      pCond->order = TSDB_ORDER_DESC;
    } else {
3341
      pCond->twindows.skey = window.ekey;
3342 3343 3344 3345
      pCond->twindows.ekey = INT64_MAX;
      pCond->order = TSDB_ORDER_ASC;
    }

3346
    // here we only need one more row, so the capacity is set to be ONE.
3347 3348 3349 3350 3351 3352
    code = tsdbReaderCreate(pVnode, pCond, &pReader->innerReader[0], 1, idstr);
    if (code != TSDB_CODE_SUCCESS) {
      goto _err;
    }

    if (order == TSDB_ORDER_ASC) {
3353
      pCond->twindows.skey = window.ekey;
3354
      pCond->twindows.ekey = INT64_MAX;
3355
    } else {
3356
      pCond->twindows.skey = INT64_MIN;
3357
      pCond->twindows.ekey = window.ekey;
3358
    }
3359 3360
    pCond->order = order;

3361 3362 3363 3364 3365 3366
    code = tsdbReaderCreate(pVnode, pCond, &pReader->innerReader[1], 1, idstr);
    if (code != TSDB_CODE_SUCCESS) {
      goto _err;
    }
  }

H
Haojun Liao 已提交
3367
  // NOTE: the endVersion in pCond is the data version not schema version, so pCond->endVersion is not correct here.
3368
  if (pCond->suid != 0) {
3369
    pReader->pSchema = metaGetTbTSchema(pReader->pTsdb->pVnode->pMeta, pReader->suid,  -1);
H
Haojun Liao 已提交
3370 3371 3372
    if (pReader->pSchema == NULL) {
      tsdbError("failed to get table schema, suid:%"PRIu64", ver:%"PRId64" , %s", pReader->suid, -1, pReader->idStr);
    }
3373 3374
  } else if (taosArrayGetSize(pTableList) > 0) {
    STableKeyInfo* pKey = taosArrayGet(pTableList, 0);
3375
    pReader->pSchema = metaGetTbTSchema(pReader->pTsdb->pVnode->pMeta, pKey->uid, -1);
H
Haojun Liao 已提交
3376 3377 3378
    if (pReader->pSchema == NULL) {
      tsdbError("failed to get table schema, uid:%"PRIu64", ver:%"PRId64" , %s", pKey->uid, -1, pReader->idStr);
    }
3379 3380
  }

3381 3382
  STsdbReader* p = pReader->innerReader[0] != NULL? pReader->innerReader[0]:pReader;

3383
  int32_t numOfTables = taosArrayGetSize(pTableList);
3384
  pReader->status.pTableMap = createDataBlockScanInfo(p, pTableList->pData, numOfTables);
H
Haojun Liao 已提交
3385 3386 3387
  if (pReader->status.pTableMap == NULL) {
    tsdbReaderClose(pReader);
    *ppReader = NULL;
H
Haojun Liao 已提交
3388

H
Haojun Liao 已提交
3389 3390 3391
    code = TSDB_CODE_TDB_OUT_OF_MEMORY;
    goto _err;
  }
H
Hongze Cheng 已提交
3392

H
Haojun Liao 已提交
3393
  code = tsdbTakeReadSnap(pReader->pTsdb, &pReader->pReadSnap, pReader->idStr);
3394 3395 3396
  if (code != TSDB_CODE_SUCCESS) {
    goto _err;
  }
H
Hongze Cheng 已提交
3397

3398
  if (pReader->type == TIMEWINDOW_RANGE_CONTAINED) {
3399 3400 3401
    code = doOpenReaderImpl(pReader);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
3402
    }
3403
  } else {
3404 3405 3406 3407 3408 3409 3410 3411 3412 3413 3414
    STsdbReader* pPrevReader = pReader->innerReader[0];
    STsdbReader* pNextReader = pReader->innerReader[1];

    // we need only one row
    pPrevReader->capacity = 1;
    pPrevReader->status.pTableMap = pReader->status.pTableMap;
    pPrevReader->pReadSnap = pReader->pReadSnap;

    pNextReader->capacity = 1;
    pNextReader->status.pTableMap = pReader->status.pTableMap;
    pNextReader->pReadSnap = pReader->pReadSnap;
3415

3416
    code = doOpenReaderImpl(pPrevReader);
3417
    if (code != TSDB_CODE_SUCCESS) {
3418
      return code;
3419 3420
    }

3421 3422 3423 3424
    code = doOpenReaderImpl(pNextReader);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
3425

3426 3427 3428
    code = doOpenReaderImpl(pReader);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
3429 3430 3431
    }
  }

3432
  tsdbDebug("%p total numOfTable:%d in this query %s", pReader, numOfTables, pReader->idStr);
H
Hongze Cheng 已提交
3433
  return code;
H
Hongze Cheng 已提交
3434 3435

_err:
S
Shengliang Guan 已提交
3436
  tsdbError("failed to create data reader, code:%s %s", tstrerror(code), pReader->idStr);
H
Hongze Cheng 已提交
3437
  return code;
H
refact  
Hongze Cheng 已提交
3438 3439 3440
}

void tsdbReaderClose(STsdbReader* pReader) {
3441 3442
  if (pReader == NULL) {
    return;
3443
  }
H
refact  
Hongze Cheng 已提交
3444

3445 3446 3447 3448 3449 3450 3451 3452 3453 3454 3455 3456 3457
  {
    if (pReader->innerReader[0] != NULL) {
      pReader->innerReader[0]->status.pTableMap = NULL;
      pReader->innerReader[0]->pReadSnap = NULL;

      pReader->innerReader[1]->status.pTableMap = NULL;
      pReader->innerReader[1]->pReadSnap = NULL;

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

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

3460 3461 3462 3463
  taosMemoryFreeClear(pSupInfo->plist);
  taosMemoryFree(pSupInfo->colIds);

  taosArrayDestroy(pSupInfo->pColAgg);
L
Liu Jicong 已提交
3464
  for (int32_t i = 0; i < blockDataGetNumOfCols(pReader->pResBlock); ++i) {
3465 3466 3467 3468
    if (pSupInfo->buildBuf[i] != NULL) {
      taosMemoryFreeClear(pSupInfo->buildBuf[i]);
    }
  }
3469

3470
  taosMemoryFree(pSupInfo->buildBuf);
H
Hongze Cheng 已提交
3471
  tBlockDataDestroy(&pReader->status.fileBlockData, true);
3472 3473

  cleanupDataBlockIterator(&pReader->status.blockIter);
3474 3475

  size_t numOfTables = taosHashGetSize(pReader->status.pTableMap);
3476
  destroyBlockScanInfo(pReader->status.pTableMap);
3477
  blockDataDestroy(pReader->pResBlock);
3478

H
Haojun Liao 已提交
3479 3480 3481
  if (pReader->pFileReader != NULL) {
    tsdbDataFReaderClose(&pReader->pFileReader);
  }
H
refact  
Hongze Cheng 已提交
3482

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

3485
  taosMemoryFree(pReader->status.uidCheckInfo.tableUidList);
H
Haojun Liao 已提交
3486
  SIOCostSummary* pCost = &pReader->cost;
3487

H
Haojun Liao 已提交
3488 3489
  SFilesetIter* pFilesetIter = &pReader->status.fileIter;
  if (pFilesetIter->pLastBlockReader != NULL) {
H
Haojun Liao 已提交
3490 3491
    SLastBlockReader* pLReader = pFilesetIter->pLastBlockReader;
    tMergeTreeClose(&pLReader->mergeTree);
H
Haojun Liao 已提交
3492

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

H
Haojun Liao 已提交
3495 3496 3497
    pLReader->pInfo = destroyLastBlockLoadInfo(pLReader->pInfo);
    taosMemoryFree(pLReader);
  }
H
refact  
Hongze Cheng 已提交
3498

H
Haojun Liao 已提交
3499 3500 3501 3502 3503 3504 3505 3506
  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
      ", composed-blocks-time:%.2fms, STableBlockScanInfo size:%.2f Kb %s",
      pReader, pCost->headFileLoad, pCost->headFileLoadTime, pCost->smaDataLoad, pCost->smaLoadTime, pCost->numOfBlocks,
      pCost->blockLoadTime, pCost->buildmemBlock, pCost->lastBlockLoad, pCost->lastBlockLoadTime, pCost->composedBlocks,
      pCost->buildComposedBlockTime, numOfTables * sizeof(STableBlockScanInfo) / 1000.0, pReader->idStr);
H
refact  
Hongze Cheng 已提交
3507

3508 3509
  taosMemoryFree(pReader->idStr);
  taosMemoryFree(pReader->pSchema);
3510 3511 3512
  if (pReader->pMemSchema != pReader->pSchema) {
    taosMemoryFree(pReader->pMemSchema);
  }
3513
  taosMemoryFreeClear(pReader);
H
refact  
Hongze Cheng 已提交
3514 3515
}

3516
static bool doTsdbNextDataBlock(STsdbReader* pReader) {
H
Haojun Liao 已提交
3517
  // cleanup the data that belongs to the previous data block
3518 3519
  SSDataBlock* pBlock = pReader->pResBlock;
  blockDataCleanup(pBlock);
H
Hongze Cheng 已提交
3520

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

3523 3524 3525 3526 3527
  if (pStatus->loadFromFile) {
    int32_t code = buildBlockFromFiles(pReader);
    if (code != TSDB_CODE_SUCCESS) {
      return false;
    }
3528

3529 3530 3531
    if (pBlock->info.rows > 0) {
      return true;
    } else {
H
Haojun Liao 已提交
3532
      buildBlockFromBufferSequentially(pReader);
3533
      return pBlock->info.rows > 0;
H
Haojun Liao 已提交
3534
    }
3535 3536 3537
  } else {  // no data in files, let's try the buffer
    buildBlockFromBufferSequentially(pReader);
    return pBlock->info.rows > 0;
H
Haojun Liao 已提交
3538
  }
3539

3540
  return false;
H
refact  
Hongze Cheng 已提交
3541 3542
}

3543 3544 3545 3546 3547
bool tsdbNextDataBlock(STsdbReader* pReader) {
  if (isEmptyQueryTimeWindow(&pReader->window)) {
    return false;
  }

3548
  if (pReader->innerReader[0] != NULL && pReader->step == 0) {
3549
    bool ret = doTsdbNextDataBlock(pReader->innerReader[0]);
3550
    resetDataBlockScanInfo(pReader->innerReader[0]->status.pTableMap, pReader->innerReader[0]->window.ekey);
3551 3552
    pReader->step = EXTERNAL_ROWS_PREV;

3553 3554 3555
    if (ret) {
      return ret;
    }
3556
  }
3557

3558 3559
  if (pReader->step == EXTERNAL_ROWS_PREV) {
    pReader->step = EXTERNAL_ROWS_MAIN;
3560 3561 3562 3563 3564 3565 3566
  }

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

3567 3568
  if (pReader->innerReader[1] != NULL && pReader->step == EXTERNAL_ROWS_MAIN) {
    resetDataBlockScanInfo(pReader->innerReader[1]->status.pTableMap, pReader->window.ekey);
3569
    bool ret1 = doTsdbNextDataBlock(pReader->innerReader[1]);
3570
    pReader->step = EXTERNAL_ROWS_NEXT;
3571 3572 3573 3574 3575 3576 3577 3578 3579
    if (ret1) {
      return ret1;
    }
  }

  return false;
}

static void setBlockInfo(STsdbReader* pReader, SDataBlockInfo* pDataBlockInfo) {
3580 3581 3582 3583
  ASSERT(pDataBlockInfo != NULL && pReader != NULL);
  pDataBlockInfo->rows = pReader->pResBlock->info.rows;
  pDataBlockInfo->uid = pReader->pResBlock->info.uid;
  pDataBlockInfo->window = pReader->pResBlock->info.window;
H
Hongze Cheng 已提交
3584 3585
}

3586 3587
void tsdbRetrieveDataBlockInfo(STsdbReader* pReader, SDataBlockInfo* pDataBlockInfo) {
  if (pReader->type == TIMEWINDOW_RANGE_EXTERNAL) {
3588
    if (pReader->step == EXTERNAL_ROWS_MAIN) {
3589
      setBlockInfo(pReader, pDataBlockInfo);
3590
    } else if (pReader->step == EXTERNAL_ROWS_PREV) {
3591 3592 3593 3594 3595 3596 3597 3598 3599
      setBlockInfo(pReader->innerReader[0], pDataBlockInfo);
    } else {
      setBlockInfo(pReader->innerReader[1], pDataBlockInfo);
    }
  } else {
    setBlockInfo(pReader, pDataBlockInfo);
  }
}

3600
int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SColumnDataAgg*** pBlockStatis, bool* allHave) {
H
Hongze Cheng 已提交
3601
  int32_t code = 0;
3602
  *allHave = false;
H
Hongze Cheng 已提交
3603

3604
  if (pReader->type == TIMEWINDOW_RANGE_EXTERNAL) {
3605 3606 3607 3608
    *pBlockStatis = NULL;
    return TSDB_CODE_SUCCESS;
  }

3609
  // there is no statistics data for composed block
3610 3611 3612 3613
  if (pReader->status.composedDataBlock) {
    *pBlockStatis = NULL;
    return TSDB_CODE_SUCCESS;
  }
H
Hongze Cheng 已提交
3614

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

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

3620 3621
  SBlockLoadSuppInfo* pSup = &pReader->suppInfo;

H
Hongze Cheng 已提交
3622
  if (tDataBlkHasSma(pBlock)) {
H
Hongze Cheng 已提交
3623
    code = tsdbReadBlockSma(pReader->pFileReader, pBlock, pSup->pColAgg);
3624
    if (code != TSDB_CODE_SUCCESS) {
3625 3626
      tsdbDebug("vgId:%d, failed to load block SMA for uid %" PRIu64 ", code:%s, %s", 0, pFBlock->uid, tstrerror(code),
                pReader->idStr);
3627 3628
      return code;
    }
3629 3630 3631
  } else {
    *pBlockStatis = NULL;
    return TSDB_CODE_SUCCESS;
3632
  }
H
Hongze Cheng 已提交
3633

3634
  *allHave = true;
H
Hongze Cheng 已提交
3635

3636 3637
  // always load the first primary timestamp column data
  SColumnDataAgg* pTsAgg = &pSup->tsColAgg;
3638

3639 3640
  pTsAgg->numOfNull = 0;
  pTsAgg->colId = PRIMARYKEY_TIMESTAMP_COL_ID;
3641 3642 3643 3644 3645 3646 3647 3648 3649 3650 3651 3652 3653 3654 3655 3656
  pTsAgg->min = pReader->pResBlock->info.window.skey;
  pTsAgg->max = pReader->pResBlock->info.window.ekey;
  pSup->plist[0] = pTsAgg;

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

  int32_t i = 0, j = 0;
  while (j < numOfCols && i < taosArrayGetSize(pSup->pColAgg)) {
    SColumnDataAgg* pAgg = taosArrayGet(pSup->pColAgg, i);
    if (pAgg->colId == pSup->colIds[j]) {
      if (IS_BSMA_ON(&(pReader->pSchema->columns[i]))) {
        pSup->plist[j] = pAgg;
      } else {
        *allHave = false;
      }
3657 3658
      i += 1;
      j += 1;
3659 3660 3661 3662 3663 3664 3665
    } else if (pAgg->colId < pSup->colIds[j]) {
      i += 1;
    } else if (pSup->colIds[j] < pAgg->colId) {
      j += 1;
    }
  }

3666
  double elapsed = (taosGetTimestampUs() - stime) / 1000.0;
3667
  pReader->cost.smaLoadTime += elapsed;
3668
  pReader->cost.smaDataLoad += 1;
3669 3670 3671

  *pBlockStatis = pSup->plist;

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

H
Hongze Cheng 已提交
3675
  return code;
H
Hongze Cheng 已提交
3676 3677
}

3678
static SArray* doRetrieveDataBlock(STsdbReader* pReader) {
H
Haojun Liao 已提交
3679 3680 3681
  SReaderStatus* pStatus = &pReader->status;

  if (pStatus->composedDataBlock) {
3682
    return pReader->pResBlock->pDataBlock;
3683
  }
3684

3685
  SFileDataBlockInfo*  pFBlock = getCurrentBlockInfo(&pStatus->blockIter);
3686
  STableBlockScanInfo* pBlockScanInfo = taosHashGet(pStatus->pTableMap, &pFBlock->uid, sizeof(pFBlock->uid));
3687

H
Haojun Liao 已提交
3688
  tBlockDataReset(&pStatus->fileBlockData);
3689 3690
  int32_t code = tBlockDataInit(&pStatus->fileBlockData, pReader->suid, pBlockScanInfo->uid, pReader->pSchema);
  if (code != TSDB_CODE_SUCCESS) {
3691 3692
    terrno = code;
    return NULL;
3693 3694 3695
  }

  code = doLoadFileBlockData(pReader, &pStatus->blockIter, &pStatus->fileBlockData);
3696
  if (code != TSDB_CODE_SUCCESS) {
H
Hongze Cheng 已提交
3697
    tBlockDataDestroy(&pStatus->fileBlockData, 1);
3698 3699
    terrno = code;
    return NULL;
3700
  }
3701 3702 3703

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

3706 3707 3708 3709 3710 3711 3712 3713 3714 3715 3716 3717
SArray* tsdbRetrieveDataBlock(STsdbReader* pReader, SArray* pIdList) {
  if (pReader->type == TIMEWINDOW_RANGE_EXTERNAL) {
    if (pReader->step == EXTERNAL_ROWS_PREV) {
      return doRetrieveDataBlock(pReader->innerReader[0]);
    } else if (pReader->step == EXTERNAL_ROWS_NEXT) {
      return doRetrieveDataBlock(pReader->innerReader[1]);
    }
  }

  return doRetrieveDataBlock(pReader);
}

H
Haojun Liao 已提交
3718
int32_t tsdbReaderReset(STsdbReader* pReader, SQueryTableDataCond* pCond) {
3719 3720 3721
  if (isEmptyQueryTimeWindow(&pReader->window)) {
    return TSDB_CODE_SUCCESS;
  }
H
Hongze Cheng 已提交
3722

L
Liu Jicong 已提交
3723
  pReader->order = pCond->order;
3724
  pReader->type = TIMEWINDOW_RANGE_CONTAINED;
3725
  pReader->status.loadFromFile = true;
dengyihao's avatar
dengyihao 已提交
3726
  pReader->status.pTableIter = NULL;
H
Haojun Liao 已提交
3727
  pReader->window = updateQueryTimeWindow(pReader->pTsdb, &pCond->twindows);
H
Hongze Cheng 已提交
3728

3729
  // allocate buffer in order to load data blocks from file
3730
  memset(&pReader->suppInfo.tsColAgg, 0, sizeof(SColumnDataAgg));
3731 3732
  memset(pReader->suppInfo.plist, 0, POINTER_BYTES);

3733
  pReader->suppInfo.tsColAgg.colId = PRIMARYKEY_TIMESTAMP_COL_ID;
3734
  tsdbDataFReaderClose(&pReader->pFileReader);
3735

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

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

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

3744
  int32_t         code = 0;
3745 3746
  SDataBlockIter* pBlockIter = &pReader->status.blockIter;

3747 3748 3749 3750 3751 3752
  // no data in files, let's try buffer in memory
  if (pReader->status.fileIter.numOfFiles == 0) {
    pReader->status.loadFromFile = false;
  } else {
    code = initForFirstBlockInFile(pReader, pBlockIter);
    if (code != TSDB_CODE_SUCCESS) {
3753 3754
      tsdbError("%p reset reader failed, numOfTables:%d, query range:%" PRId64 " - %" PRId64 " in query %s", pReader,
                numOfTables, pReader->window.skey, pReader->window.ekey, pReader->idStr);
3755 3756 3757
      return code;
    }
  }
H
Hongze Cheng 已提交
3758

H
Haojun Liao 已提交
3759 3760
  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);
3761

3762
  return code;
H
Hongze Cheng 已提交
3763
}
H
Hongze Cheng 已提交
3764

3765 3766 3767
static int32_t getBucketIndex(int32_t startRow, int32_t bucketRange, int32_t numOfRows) {
  return (numOfRows - startRow) / bucketRange;
}
H
Hongze Cheng 已提交
3768

3769 3770 3771 3772
int32_t tsdbGetFileBlocksDistInfo(STsdbReader* pReader, STableBlockDistInfo* pTableBlockInfo) {
  int32_t code = TSDB_CODE_SUCCESS;
  pTableBlockInfo->totalSize = 0;
  pTableBlockInfo->totalRows = 0;
H
Hongze Cheng 已提交
3773

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

3777 3778 3779
  STsdbCfg* pc = &pReader->pTsdb->pVnode->config.tsdbCfg;
  pTableBlockInfo->defMinRows = pc->minRows;
  pTableBlockInfo->defMaxRows = pc->maxRows;
H
Hongze Cheng 已提交
3780

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

3783
  pTableBlockInfo->numOfFiles += 1;
H
Hongze Cheng 已提交
3784

3785 3786
  int32_t numOfTables = (int32_t)taosHashGetSize(pStatus->pTableMap);
  int     defaultRows = 4096;
H
Hongze Cheng 已提交
3787

3788 3789
  SDataBlockIter* pBlockIter = &pStatus->blockIter;
  pTableBlockInfo->numOfFiles += pStatus->fileIter.numOfFiles;
H
Haojun Liao 已提交
3790

3791 3792
  if (pBlockIter->numOfBlocks > 0) {
    pTableBlockInfo->numOfBlocks += pBlockIter->numOfBlocks;
H
Haojun Liao 已提交
3793
  }
H
Hongze Cheng 已提交
3794

3795
  pTableBlockInfo->numOfTables = numOfTables;
3796
  bool hasNext = (pBlockIter->numOfBlocks > 0);
H
Hongze Cheng 已提交
3797

3798 3799
  while (true) {
    if (hasNext) {
H
Hongze Cheng 已提交
3800
      SDataBlk* pBlock = getCurrentBlock(pBlockIter);
H
Hongze Cheng 已提交
3801

3802 3803
      int32_t numOfRows = pBlock->nRow;
      pTableBlockInfo->totalRows += numOfRows;
H
Hongze Cheng 已提交
3804

3805 3806 3807
      if (numOfRows > pTableBlockInfo->maxRows) {
        pTableBlockInfo->maxRows = numOfRows;
      }
H
refact  
Hongze Cheng 已提交
3808

3809 3810 3811
      if (numOfRows < pTableBlockInfo->minRows) {
        pTableBlockInfo->minRows = numOfRows;
      }
H
refact  
Hongze Cheng 已提交
3812

3813 3814 3815
      if (numOfRows < defaultRows) {
        pTableBlockInfo->numOfSmallBlocks += 1;
      }
H
refact  
Hongze Cheng 已提交
3816

3817 3818
      int32_t bucketIndex = getBucketIndex(pTableBlockInfo->defMinRows, bucketRange, numOfRows);
      pTableBlockInfo->blockRowsHisto[bucketIndex]++;
3819 3820

      hasNext = blockIteratorNext(&pStatus->blockIter);
3821 3822 3823 3824 3825
    } else {
      code = initForFirstBlockInFile(pReader, pBlockIter);
      if ((code != TSDB_CODE_SUCCESS) || (pReader->status.loadFromFile == false)) {
        break;
      }
H
refact  
Hongze Cheng 已提交
3826

3827 3828
      pTableBlockInfo->numOfBlocks += pBlockIter->numOfBlocks;
      hasNext = (pBlockIter->numOfBlocks > 0);
3829
    }
H
refact  
Hongze Cheng 已提交
3830

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

H
refact  
Hongze Cheng 已提交
3835 3836
  return code;
}
H
Hongze Cheng 已提交
3837

H
refact  
Hongze Cheng 已提交
3838
int64_t tsdbGetNumOfRowsInMemTable(STsdbReader* pReader) {
3839
  int64_t rows = 0;
H
Hongze Cheng 已提交
3840

3841 3842
  SReaderStatus* pStatus = &pReader->status;
  pStatus->pTableIter = taosHashIterate(pStatus->pTableMap, NULL);
H
Hongze Cheng 已提交
3843

3844 3845 3846 3847 3848
  while (pStatus->pTableIter != NULL) {
    STableBlockScanInfo* pBlockScanInfo = pStatus->pTableIter;

    STbData* d = NULL;
    if (pReader->pTsdb->mem != NULL) {
H
Hongze Cheng 已提交
3849
      d = tsdbGetTbDataFromMemTable(pReader->pReadSnap->pMem, pReader->suid, pBlockScanInfo->uid);
3850 3851 3852 3853 3854 3855 3856
      if (d != NULL) {
        rows += tsdbGetNRowsInTbData(d);
      }
    }

    STbData* di = NULL;
    if (pReader->pTsdb->imem != NULL) {
H
Hongze Cheng 已提交
3857
      di = tsdbGetTbDataFromMemTable(pReader->pReadSnap->pIMem, pReader->suid, pBlockScanInfo->uid);
3858 3859 3860 3861 3862 3863 3864 3865
      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 已提交
3866

H
refact  
Hongze Cheng 已提交
3867
  return rows;
H
Hongze Cheng 已提交
3868
}
D
dapan1121 已提交
3869

L
Liu Jicong 已提交
3870
int32_t tsdbGetTableSchema(SVnode* pVnode, int64_t uid, STSchema** pSchema, int64_t* suid) {
D
dapan1121 已提交
3871 3872 3873 3874 3875 3876 3877 3878 3879 3880 3881 3882
  int32_t sversion = 1;

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

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

D
dapan1121 已提交
3884
  if (mr.me.type == TSDB_CHILD_TABLE) {
D
dapan1121 已提交
3885
    tDecoderClear(&mr.coder);
D
dapan1121 已提交
3886 3887 3888 3889 3890 3891 3892 3893 3894 3895 3896 3897 3898 3899 3900
    *suid = mr.me.ctbEntry.suid;
    code = metaGetTableEntryByUid(&mr, *suid);
    if (code != TSDB_CODE_SUCCESS) {
      terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
      metaReaderClear(&mr);
      return terrno;
    }
    sversion = mr.me.stbEntry.schemaRow.version;
  } else {
    ASSERT(mr.me.type == TSDB_NORMAL_TABLE);
    sversion = mr.me.ntbEntry.schemaRow.version;
  }

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

D
dapan1121 已提交
3902 3903
  return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
3904

H
Haojun Liao 已提交
3905
int32_t tsdbTakeReadSnap(STsdb* pTsdb, STsdbReadSnap** ppSnap, const char* idStr) {
H
Hongze Cheng 已提交
3906 3907 3908 3909 3910 3911 3912 3913 3914 3915 3916 3917 3918 3919 3920 3921 3922 3923 3924 3925 3926 3927 3928 3929 3930 3931 3932 3933
  int32_t code = 0;

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

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

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

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

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

H
Hongze Cheng 已提交
3934
  // fs
H
Hongze Cheng 已提交
3935 3936 3937 3938 3939
  code = tsdbFSRef(pTsdb, &(*ppSnap)->fs);
  if (code) {
    taosThreadRwlockUnlock(&pTsdb->rwLock);
    goto _exit;
  }
H
Hongze Cheng 已提交
3940 3941 3942 3943 3944 3945 3946 3947

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

H
Haojun Liao 已提交
3948
  tsdbTrace("vgId:%d, take read snapshot, %s", TD_VID(pTsdb->pVnode), idStr);
H
Hongze Cheng 已提交
3949 3950 3951 3952
_exit:
  return code;
}

H
Haojun Liao 已提交
3953
void tsdbUntakeReadSnap(STsdb* pTsdb, STsdbReadSnap* pSnap, const char* idStr) {
H
Hongze Cheng 已提交
3954 3955 3956 3957 3958 3959 3960 3961 3962
  if (pSnap) {
    if (pSnap->pMem) {
      tsdbUnrefMemTable(pSnap->pMem);
    }

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

H
Hongze Cheng 已提交
3963
    tsdbFSUnref(pTsdb, &pSnap->fs);
H
Hongze Cheng 已提交
3964
    taosMemoryFree(pSnap);
H
Hongze Cheng 已提交
3965
  }
H
Haojun Liao 已提交
3966
  tsdbTrace("vgId:%d, untake read snapshot, %s", TD_VID(pTsdb->pVnode), idStr);
H
Hongze Cheng 已提交
3967
}