tsdbRead.c 123.6 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
Hongze Cheng 已提交
16
#include "tsdb.h"
H
Hongze Cheng 已提交
17
#define ASCENDING_TRAVERSE(o)      (o == TSDB_ORDER_ASC)
18
#define QH_GET_NUM_OF_COLS(handle) ((size_t)(taosArrayGetSize((handle)->pResBlock->pDataBlock)))
H
Hongze Cheng 已提交
19

H
Hongze Cheng 已提交
20 21 22 23 24 25 26 27 28 29 30
#define GET_FILE_DATA_BLOCK_INFO(_checkInfo, _block)                                      \
  ((SDataBlockInfo){.window = {.skey = (_block)->minKey.ts, .ekey = (_block)->maxKey.ts}, \
                    .rows = (_block)->numOfRows,                                          \
                    .uid = (_checkInfo)->tableId})

typedef struct SQueryFilePos {
  int32_t     fid;
  int32_t     slot;
  int32_t     pos;
  int64_t     lastKey;
  int32_t     rows;
H
Haojun Liao 已提交
31
  bool        composedBlock;
H
Hongze Cheng 已提交
32 33 34 35 36 37 38 39 40 41 42
  bool        blockCompleted;
  STimeWindow win;
} SQueryFilePos;

typedef struct SDataBlockLoadInfo {
  SDFileSet* fileGroup;
  int32_t    slot;
  uint64_t   uid;
  SArray*    pLoadedCols;
} SDataBlockLoadInfo;

H
Haojun Liao 已提交
43 44 45 46 47 48
typedef struct STableBlockScanInfo {
  uint64_t     uid;
  TSKEY        lastKey;
  SBlockIdx    blockIdx;
  SArray*      pBlockList;        // block data index list
  bool         iterInit;       // whether to initialize the in-memory skip list iterator or not
49 50
  STbDataIter* iter;              // mem buffer skip list iterator
  STbDataIter* iiter;             // imem buffer skip list iterator
H
Haojun Liao 已提交
51 52 53 54 55 56 57 58
  bool         memHasVal;
  bool         imemHasVal;
} STableBlockScanInfo;

typedef struct SBlockOrderWrapper {
  int64_t uid;
  SBlock* pBlock;
} SBlockOrderWrapper;
H
Hongze Cheng 已提交
59 60

typedef struct SBlockOrderSupporter {
H
Haojun Liao 已提交
61 62
  SBlockOrderWrapper**  pDataBlockInfo;
  int32_t*          indexPerTable;
H
Hongze Cheng 已提交
63
  int32_t*          numOfBlocksPerTable;
H
Haojun Liao 已提交
64
  int32_t           numOfTables;
H
Hongze Cheng 已提交
65 66 67
} SBlockOrderSupporter;

typedef struct SIOCostSummary {
H
Haojun Liao 已提交
68 69 70 71 72
  int64_t           blockLoadTime;
  int64_t           statisInfoLoadTime;
  int64_t           checkForNextTime;
  int64_t           headFileLoad;
  int64_t           headFileLoadTime;
H
Hongze Cheng 已提交
73 74 75
} SIOCostSummary;

typedef struct SBlockLoadSuppInfo {
H
Haojun Liao 已提交
76 77
  SColumnDataAgg*   pstatis;
  SColumnDataAgg**  plist;
78
  int16_t*          colIds;      // column ids for loading file block data
H
Haojun Liao 已提交
79
  int32_t*          slotIds;     // colId to slotId
80
  char**            buildBuf;    // build string tmp buffer, todo remove it later after all string format being updated.
H
Hongze Cheng 已提交
81 82
} SBlockLoadSuppInfo;

83
typedef struct SFilesetIter {
H
Haojun Liao 已提交
84 85 86
  int32_t           numOfFiles;  // number of total files
  int32_t           index;       // current accessed index in the list
  SArray*           pFileList;   // data file list
87 88
  int32_t           order;
} SFilesetIter;
H
Haojun Liao 已提交
89 90 91 92 93 94 95 96 97 98

typedef struct SFileDataBlockInfo {
  int32_t tbBlockIdx;   // index position in STableBlockScanInfo in order to check whether neighbor block overlaps with it
  int64_t uid;
} SFileDataBlockInfo;

typedef struct SDataBlockIter {
  int32_t      numOfBlocks;
  int32_t      index;
  SArray*      blockList;   // SArray<SFileDataBlockInfo>
99
  int32_t      order;
H
Haojun Liao 已提交
100 101 102 103 104 105
} SDataBlockIter;

typedef struct SFileBlockDumpInfo {
  int32_t      totalRows;
  int32_t      rowIndex;
  int64_t      lastKey;
106
  bool         allDumped;
H
Haojun Liao 已提交
107 108
} SFileBlockDumpInfo;

H
Haojun Liao 已提交
109 110 111 112 113
typedef struct SVersionRange {
  uint64_t minVer;
  uint64_t maxVer;
} SVersionRange;

H
Haojun Liao 已提交
114
typedef struct SReaderStatus {
115 116 117 118
  SQueryFilePos      cur;             // current position
  bool               loadFromFile;    // check file stage
  SHashObj*          pTableMap;       // SHash<STableBlockScanInfo>
  STableBlockScanInfo* pTableIter;    // table iterator used in building in-memory buffer data blocks.
H
Haojun Liao 已提交
119
  SFileBlockDumpInfo fBlockDumpInfo;
120

121
  SDFileSet*         pCurrentFileset; // current opened file set
122
  SBlockData         fileBlockData;
123
  SFilesetIter       fileIter;
124 125
  SDataBlockIter     blockIter;
  bool               composedDataBlock;// the returned data block is a composed block or not
H
Haojun Liao 已提交
126 127
} SReaderStatus;

H
Hongze Cheng 已提交
128
struct STsdbReader {
H
Haojun Liao 已提交
129 130 131 132 133 134 135
  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;
136 137
  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 已提交
138
  SBlockLoadSuppInfo suppInfo;
139 140
  SArray*            prev;    // previous row which is before than time window
  SArray*            next;    // next row which is after the query time window
H
Hongze Cheng 已提交
141 142
  SIOCostSummary     cost;
  STSchema*          pSchema;
H
Haojun Liao 已提交
143 144

  SDataFReader*      pFileReader;
H
Haojun Liao 已提交
145
  SVersionRange      verRange;
H
Haojun Liao 已提交
146 147 148 149 150 151 152 153 154 155 156 157
#if 0
  SFileBlockInfo*   pDataBlockInfo;
  SDataCols*         pDataCols;         // in order to hold current file data block
  int32_t            allocSize;         // allocated data block size
  SDataBlockLoadInfo dataBlockLoadInfo; /* record current block load information */
  SLoadCompBlockInfo compBlockLoadInfo; /* record current compblock information in SQueryAttr */
  //  SDFileSet* pFileGroup;
  // SFSIter            fileIter;
  // SReadH             rhelper;
  //  SColumnDataAgg* statis;  // query level statistics, only one table block statistics info exists at any time
  //  SColumnDataAgg** pstatis;// the ptr array list to return to caller
#endif
H
Hongze Cheng 已提交
158
};
H
Hongze Cheng 已提交
159

H
Haojun Liao 已提交
160
static SFileDataBlockInfo* getCurrentBlockInfo(SDataBlockIter* pBlockIter);
161
static int buildInmemDataBlockImpl(STableBlockScanInfo* pBlockScanInfo, TSDBKEY maxKey, int32_t capacity, STsdbReader* pReader);
H
Haojun Liao 已提交
162
static TSDBROW* getValidRow(STbDataIter* pIter, bool* hasVal, STsdbReader* pReader);
163 164 165 166 167
static int32_t doLoadRowsOfIdenticalTsInFileBlock(SFileDataBlockInfo* pFBlock, SBlock* pBlock, SBlockData* pBlockData,
                                                  STableBlockScanInfo* pScanInfo, STsdbReader* pReader, SRowMerger* pMerger);
static int32_t doLoadRowsOfIdenticalTs(STbDataIter *pIter, bool* hasVal, int64_t ts, SRowMerger* pMerger, STsdbReader* pReader);
static int32_t doAppendOneRow(SSDataBlock* pBlock, STsdbReader* pReader, STSRow* pTSRow);
static void setComposedBlockFlag(STsdbReader* pReader, bool composed);
H
Haojun Liao 已提交
168

H
Hongze Cheng 已提交
169 170 171 172 173
// static void tsdbInitDataBlockLoadInfo(SDataBlockLoadInfo* pBlockLoadInfo) {
//   pBlockLoadInfo->slot = -1;
//   pBlockLoadInfo->uid = 0;
//   pBlockLoadInfo->fileGroup = NULL;
// }
H
Hongze Cheng 已提交
174

H
Hongze Cheng 已提交
175 176 177 178
// static void tsdbInitCompBlockLoadInfo(SLoadCompBlockInfo* pCompBlockLoadInfo) {
//   pCompBlockLoadInfo->tid = -1;
//   pCompBlockLoadInfo->fileId = -1;
// }
H
Hongze Cheng 已提交
179

180 181 182
static int32_t setColumnIdSlotList(STsdbReader* pReader, SSDataBlock* pBlock) {
  SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo;

183 184 185 186 187
  size_t numOfCols = blockDataGetNumOfCols(pBlock);
//  pSupInfo->slotIds = taosMemoryCalloc(numOfCols, sizeof(int16_t));
//  if (pSupInfo->slotIds == NULL) {
//    return TSDB_CODE_OUT_OF_MEMORY;
//  }
188

189
  pSupInfo->colIds = taosMemoryMalloc(numOfCols * sizeof(int16_t));
190
  pSupInfo->buildBuf = taosMemoryCalloc(numOfCols, POINTER_BYTES);
191 192 193
  if (pSupInfo->buildBuf == NULL || pSupInfo->colIds == NULL) {
    taosMemoryFree(pSupInfo->colIds);
    taosMemoryFree(pSupInfo->buildBuf);
H
Haojun Liao 已提交
194 195
    return TSDB_CODE_OUT_OF_MEMORY;
  }
H
Hongze Cheng 已提交
196

197
//  STSchema* pSchema = pReader->pSchema;
H
Haojun Liao 已提交
198 199
  for (int32_t i = 0; i < numOfCols; ++i) {
    SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, i);
200
    pSupInfo->colIds[i] = pCol->info.colId;
201

202 203 204 205 206 207
//    for (int32_t j = 0; j < pSchema->numOfCols; ++j) {
//      if (pCol->info.colId == pSchema->columns[j].colId) {
//        pSupInfo->slotIds[i] = j;
//        break;
//      }
//    }
208 209 210 211

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

H
Haojun Liao 已提交
214 215
  return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
216

217
static SHashObj* createDataBlockScanInfo(STsdbReader* pTsdbReader, const STableKeyInfo* idList, int32_t numOfTables) {
H
Haojun Liao 已提交
218 219 220 221 222 223 224 225 226
   // allocate buffer in order to load data blocks from file
   // todo use simple hash instead
   SHashObj* pTableMap = taosHashInit(numOfTables, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
   if (pTableMap == NULL) {
     return NULL;
   }

   // todo apply the lastkey of table check to avoid to load header file
   for (int32_t j = 0; j < numOfTables; ++j) {
227
     STableBlockScanInfo info = {.lastKey = 0, .uid = idList[j].uid};
H
Haojun Liao 已提交
228 229 230 231
     if (ASCENDING_TRAVERSE(pTsdbReader->order)) {
       if (info.lastKey == INT64_MIN || info.lastKey < pTsdbReader->window.skey) {
         info.lastKey = pTsdbReader->window.skey;
       }
H
Haojun Liao 已提交
232

H
Haojun Liao 已提交
233 234 235 236 237 238 239 240 241 242 243
       ASSERT(info.lastKey >= pTsdbReader->window.skey && info.lastKey <= pTsdbReader->window.ekey);
     } else {
       info.lastKey = pTsdbReader->window.skey;
     }

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

   return pTableMap;
H
Hongze Cheng 已提交
244
}
H
Hongze Cheng 已提交
245

H
Hongze Cheng 已提交
246 247 248 249 250 251
// static void resetCheckInfo(STsdbReader* pTsdbReadHandle) {
//   size_t numOfTables = taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
//   assert(numOfTables >= 1);

//   // todo apply the lastkey of table check to avoid to load header file
//   for (int32_t i = 0; i < numOfTables; ++i) {
H
Haojun Liao 已提交
252
//     STableBlockScanInfo* pCheckInfo = (STableBlockScanInfo*)taosArrayGet(pTsdbReadHandle->pTableCheckInfo, i);
H
Hongze Cheng 已提交
253 254 255 256 257 258 259 260 261 262 263 264
//     pCheckInfo->lastKey = pTsdbReadHandle->window.skey;
//     pCheckInfo->iter = tsdbTbDataIterDestroy(pCheckInfo->iter);
//     pCheckInfo->iiter = tsdbTbDataIterDestroy(pCheckInfo->iiter);
//     pCheckInfo->initBuf = false;

//     if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
//       assert(pCheckInfo->lastKey >= pTsdbReadHandle->window.skey);
//     } else {
//       assert(pCheckInfo->lastKey <= pTsdbReadHandle->window.skey);
//     }
//   }
// }
H
Hongze Cheng 已提交
265

H
Haojun Liao 已提交
266 267
static bool isEmptyQueryTimeWindow(STimeWindow* pWindow, int32_t order) {
   ASSERT(pWindow != NULL);
268
   return pWindow->skey > pWindow->ekey;
H
Haojun Liao 已提交
269
 }
H
Hongze Cheng 已提交
270

H
Hongze Cheng 已提交
271 272 273 274
// // 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 int64_t getEarliestValidTimestamp(STsdb* pTsdb) {
//   STsdbKeepCfg* pCfg = &pTsdb->keepCfg;
H
Hongze Cheng 已提交
275

H
Hongze Cheng 已提交
276 277 278
//   int64_t now = taosGetTimestamp(pCfg->precision);
//   return now - (tsTickPerMin[pCfg->precision] * pCfg->keep2) + 1;  // needs to add one tick
// }
H
Hongze Cheng 已提交
279

H
Haojun Liao 已提交
280
// todo remove this
H
Hongze Cheng 已提交
281 282
static void setQueryTimewindow(STsdbReader* pReader, SQueryTableDataCond* pCond, int32_t tWinIdx) {
  // pReader->window = pCond->twindows[tWinIdx];
H
Hongze Cheng 已提交
283

H
Hongze Cheng 已提交
284 285 286 287 288 289 290 291 292 293 294 295 296 297 298
  // bool    updateTs = false;
  // int64_t startTs = getEarliestValidTimestamp(pReader->pTsdb);
  // if (ASCENDING_TRAVERSE(pReader->order)) {
  //   if (startTs > pReader->window.skey) {
  //     pReader->window.skey = startTs;
  //     pCond->twindows[tWinIdx].skey = startTs;
  //     updateTs = true;
  //   }
  // } else {
  //   if (startTs > pReader->window.ekey) {
  //     pReader->window.ekey = startTs;
  //     pCond->twindows[tWinIdx].ekey = startTs;
  //     updateTs = true;
  //   }
  // }
H
Hongze Cheng 已提交
299

H
Hongze Cheng 已提交
300 301 302 303 304 305
  // if (updateTs) {
  //   tsdbDebug("%p update the query time window, old:%" PRId64 " - %" PRId64 ", new:%" PRId64 " - %" PRId64 ", %s",
  //             pReader, pCond->twindows[tWinIdx].skey, pCond->twindows[tWinIdx].ekey, pReader->window.skey,
  //             pReader->window.ekey, pReader->idStr);
  // }
}
H
Hongze Cheng 已提交
306

H
Haojun Liao 已提交
307
static void limitOutputBufferSize(const SQueryTableDataCond* pCond, int32_t* capacity) {
H
Haojun Liao 已提交
308 309 310 311 312 313
  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 已提交
314 315 316
  const int32_t TWOMB = 2 * 1024 * 1024;
  if ((*capacity) * rowLen > TWOMB) {
    (*capacity) = TWOMB / rowLen;
H
Haojun Liao 已提交
317 318 319 320
  }
}

// init file iterator
321 322 323 324 325 326 327
static int32_t initFileIterator(SFilesetIter* pIter, const STsdbFSState* pFState, int32_t order, const char* idstr) {
  size_t numOfFileset = taosArrayGetSize(pFState->aDFileSet);

  pIter->index      = ASCENDING_TRAVERSE(order) ? -1 : numOfFileset;
  pIter->order      = order;
  pIter->pFileList  = taosArrayDup(pFState->aDFileSet);
  pIter->numOfFiles = numOfFileset;
H
Haojun Liao 已提交
328

H
Haojun Liao 已提交
329
  tsdbDebug("init fileset iterator, total files:%d %s", pIter->numOfFiles, idstr);
H
Haojun Liao 已提交
330 331 332
  return TSDB_CODE_SUCCESS;
}

333
static bool filesetIteratorNext(SFilesetIter* pIter, STsdbReader* pReader) {
334 335
  bool    asc = ASCENDING_TRAVERSE(pIter->order);
  int32_t step = asc ? 1 : -1;
336 337 338
  pIter->index += step;

  if ((asc && pIter->index >= pIter->numOfFiles) || ((!asc) && pIter->index < 0)) {
H
Haojun Liao 已提交
339 340 341 342 343
    return false;
  }

  // check file the time range of coverage
  STimeWindow win = {0};
H
Hongze Cheng 已提交
344

345 346
  while(1) {
    pReader->status.pCurrentFileset = (SDFileSet*)taosArrayGet(pIter->pFileList, pIter->index);
H
Haojun Liao 已提交
347

348 349 350 351
    int32_t code = tsdbDataFReaderOpen(&pReader->pFileReader, pReader->pTsdb, pReader->status.pCurrentFileset);
    if (code != TSDB_CODE_SUCCESS) {
      goto _err;
    }
H
Haojun Liao 已提交
352

353 354 355 356 357 358 359 360 361 362 363 364 365 366
    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;
      continue;
    }
C
Cary Xu 已提交
367

368 369 370 371
    tsdbDebug("%p file found fid:%d for qrange:%" PRId64 "-%" PRId64 ", ignore, %s", pReader, fid,
              pReader->window.skey, pReader->window.ekey, pReader->idStr);
    return true;
  }
372

373
_err:
H
Haojun Liao 已提交
374 375 376
  return false;
}

377 378 379
static void resetDataBlockIterator(SDataBlockIter* pIter, int32_t order) {
  pIter->order       = order;
  pIter->index       = -1;
H
Haojun Liao 已提交
380
  pIter->numOfBlocks = -1;
381
  pIter->blockList   = taosArrayInit(4, sizeof(SFileDataBlockInfo));
H
Haojun Liao 已提交
382 383
}

H
Haojun Liao 已提交
384 385 386
static void initReaderStatus(SReaderStatus* pStatus) {
  pStatus->cur.fid        = INT32_MIN;
  pStatus->cur.win        = TSWINDOW_INITIALIZER;
387
  pStatus->pTableIter     = NULL;
H
Haojun Liao 已提交
388 389 390
  pStatus->loadFromFile   = true;
}

391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413
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;
}

H
Haojun Liao 已提交
414 415 416
static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, STsdbReader** ppReader, const char* idstr) {
  int32_t      code = 0;
  STsdbReader* pReader = (STsdbReader*)taosMemoryCalloc(1, sizeof(*pReader));
H
Hongze Cheng 已提交
417 418
  if (pReader == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
H
Haojun Liao 已提交
419
    goto _end;
H
Hongze Cheng 已提交
420 421
  }

H
Haojun Liao 已提交
422
  initReaderStatus(&pReader->status);
423

H
Haojun Liao 已提交
424 425 426 427 428
  pReader->pTsdb       = pVnode->pTsdb;
  pReader->suid        = pCond->suid;
  pReader->order       = pCond->order;
  pReader->capacity    = 4096;
  pReader->idStr       = strdup(idstr);
H
Haojun Liao 已提交
429 430
  pReader->verRange.minVer= pCond->startVersion;
  pReader->verRange.maxVer  = 100000;//pCond->endVersion;  // todo for test purpose
431 432 433
  pReader->type        = pCond->type;
  pReader->window      = *pCond->twindows;

434 435 436 437 438 439
#if 1
  if (pReader->window.skey > pReader->window.ekey) {
    TSWAP(pReader->window.skey, pReader->window.ekey);
  }
#endif

440 441 442 443
  if (pReader->suid != 0) {
    pReader->pSchema = metaGetTbTSchema(pReader->pTsdb->pVnode->pMeta, pReader->suid, -1);
    ASSERT(pReader->pSchema);
  }
H
Haojun Liao 已提交
444 445

  // todo remove this
H
Hongze Cheng 已提交
446
  setQueryTimewindow(pReader, pCond, 0);
447
  ASSERT (pCond->numOfCols > 0);
H
Hongze Cheng 已提交
448

449
  limitOutputBufferSize(pCond, &pReader->capacity);
450

451 452 453 454 455 456 457 458
  // allocate buffer in order to load data blocks from file
  SBlockLoadSuppInfo* pSup = &pReader->suppInfo;
  pSup->pstatis = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnDataAgg));
  pSup->plist = taosMemoryCalloc(pCond->numOfCols, POINTER_BYTES);
  if (pSup->pstatis == NULL || pSup->plist == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _end;
  }
H
Haojun Liao 已提交
459

460 461 462 463
  pReader->pResBlock = createResBlock(pCond, pReader->capacity);
  if (pReader->pResBlock == NULL) {
    code = terrno;
    goto _end;
H
Hongze Cheng 已提交
464
  }
H
Hongze Cheng 已提交
465

466 467
  setColumnIdSlotList(pReader, pReader->pResBlock);

H
Haojun Liao 已提交
468
  STsdbFSState* pFState = pReader->pTsdb->fs->cState;
469 470
  initFileIterator(&pReader->status.fileIter, pFState, pReader->order, pReader->idStr);
  resetDataBlockIterator(&pReader->status.blockIter, pReader->order);
H
Hongze Cheng 已提交
471

H
Haojun Liao 已提交
472 473 474 475
  // no data in files, let's try buffer in memory
  if (pReader->status.fileIter.numOfFiles == 0) {
    pReader->status.loadFromFile = false;
  }
H
Hongze Cheng 已提交
476

H
Hongze Cheng 已提交
477 478
  *ppReader = pReader;
  return code;
H
Hongze Cheng 已提交
479

H
Haojun Liao 已提交
480 481
_end:
  tsdbReaderClose(pReader);
H
Hongze Cheng 已提交
482 483 484
  *ppReader = NULL;
  return code;
}
H
Hongze Cheng 已提交
485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517

// void tsdbResetQueryHandleForNewTable(STsdbReader* queryHandle, SQueryTableDataCond* pCond, STableListInfo* tableList,
//                                      int32_t tWinIdx) {
//   STsdbReader* pTsdbReadHandle = queryHandle;

//   pTsdbReadHandle->order = pCond->order;
//   pTsdbReadHandle->window = pCond->twindows[tWinIdx];
//   pTsdbReadHandle->type = TSDB_QUERY_TYPE_ALL;
//   pTsdbReadHandle->cur.fid = -1;
//   pTsdbReadHandle->cur.win = TSWINDOW_INITIALIZER;
//   pTsdbReadHandle->checkFiles = true;
//   pTsdbReadHandle->activeIndex = 0;  // current active table index
//   pTsdbReadHandle->locateStart = false;
//   pTsdbReadHandle->loadExternalRow = pCond->loadExternalRows;

//   if (ASCENDING_TRAVERSE(pCond->order)) {
//     assert(pTsdbReadHandle->window.skey <= pTsdbReadHandle->window.ekey);
//   } else {
//     assert(pTsdbReadHandle->window.skey >= pTsdbReadHandle->window.ekey);
//   }

//   // allocate buffer in order to load data blocks from file
//   memset(pTsdbReadHandle->suppInfo.pstatis, 0, sizeof(SColumnDataAgg));
//   memset(pTsdbReadHandle->suppInfo.plist, 0, POINTER_BYTES);

//   tsdbInitDataBlockLoadInfo(&pTsdbReadHandle->dataBlockLoadInfo);
//   tsdbInitCompBlockLoadInfo(&pTsdbReadHandle->compBlockLoadInfo);

//   SArray* pTable = NULL;
//   //  STsdbMeta* pMeta = tsdbGetMeta(pTsdbReadHandle->pTsdb);

//   //  pTsdbReadHandle->pTableCheckInfo = destroyTableCheckInfo(pTsdbReadHandle->pTableCheckInfo);

H
Haojun Liao 已提交
518
//   pTsdbReadHandle->pTableCheckInfo = NULL;  // createDataBlockScanInfo(pTsdbReadHandle, groupList, pMeta,
H
Hongze Cheng 已提交
519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538
//                                             // &pTable);
//   if (pTsdbReadHandle->pTableCheckInfo == NULL) {
//     //    tsdbReaderClose(pTsdbReadHandle);
//     terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
//   }

//   //  pTsdbReadHandle->prev = doFreeColumnInfoData(pTsdbReadHandle->prev);
//   //  pTsdbReadHandle->next = doFreeColumnInfoData(pTsdbReadHandle->next);
// }

// SArray* tsdbGetQueriedTableList(STsdbReader** pHandle) {
//   assert(pHandle != NULL);

//   STsdbReader* pTsdbReadHandle = (STsdbReader*)pHandle;

//   size_t  size = taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
//   SArray* res = taosArrayInit(size, POINTER_BYTES);
//   return res;
// }

H
Haojun Liao 已提交
539
// static TSKEY extractFirstTraverseKey(STableBlockScanInfo* pCheckInfo, int32_t order, int32_t update, TDRowVerT maxVer) {
H
Hongze Cheng 已提交
540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588
//   TSDBROW row = {0};
//   STSRow *rmem = NULL, *rimem = NULL;

//   if (pCheckInfo->iter) {
//     if (tsdbTbDataIterGet(pCheckInfo->iter, &row)) {
//       rmem = row.pTSRow;
//     }
//   }

//   if (pCheckInfo->iiter) {
//     if (tsdbTbDataIterGet(pCheckInfo->iiter, &row)) {
//       rimem = row.pTSRow;
//     }
//   }

//   if (rmem == NULL && rimem == NULL) {
//     return TSKEY_INITIAL_VAL;
//   }

//   if (rmem != NULL && rimem == NULL) {
//     pCheckInfo->chosen = CHECKINFO_CHOSEN_MEM;
//     return TD_ROW_KEY(rmem);
//   }

//   if (rmem == NULL && rimem != NULL) {
//     pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
//     return TD_ROW_KEY(rimem);
//   }

//   TSKEY r1 = TD_ROW_KEY(rmem);
//   TSKEY r2 = TD_ROW_KEY(rimem);

//   if (r1 == r2) {
//     if (TD_SUPPORT_UPDATE(update)) {
//       pCheckInfo->chosen = CHECKINFO_CHOSEN_BOTH;
//     } else {
//       pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
//       tsdbTbDataIterNext(pCheckInfo->iter);
//     }
//     return r1;
//   } else if (r1 < r2 && ASCENDING_TRAVERSE(order)) {
//     pCheckInfo->chosen = CHECKINFO_CHOSEN_MEM;
//     return r1;
//   } else {
//     pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
//     return r2;
//   }
// }

H
Haojun Liao 已提交
589
// static STSRow* getSRowInTableMem(STableBlockScanInfo* pCheckInfo, int32_t order, int32_t update, STSRow** extraRow,
H
Hongze Cheng 已提交
590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605
//                                  TDRowVerT maxVer) {
//   TSDBROW row;
//   STSRow *rmem = NULL, *rimem = NULL;
//   if (pCheckInfo->iter) {
//     if (tsdbTbDataIterGet(pCheckInfo->iter, &row)) {
//       rmem = row.pTSRow;
//     }
//   }

//   if (pCheckInfo->iiter) {
//     if (tsdbTbDataIterGet(pCheckInfo->iiter, &row)) {
//       rimem = row.pTSRow;
//     }
//   }

//   if (rmem == NULL && rimem == NULL) {
H
Hongze Cheng 已提交
606 607
//     return NULL;
//   }
H
Hongze Cheng 已提交
608

H
Hongze Cheng 已提交
609 610 611
//   if (rmem != NULL && rimem == NULL) {
//     pCheckInfo->chosen = 0;
//     return rmem;
H
Hongze Cheng 已提交
612
//   }
H
Hongze Cheng 已提交
613

H
Hongze Cheng 已提交
614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650
//   if (rmem == NULL && rimem != NULL) {
//     pCheckInfo->chosen = 1;
//     return rimem;
//   }

//   TSKEY r1 = TD_ROW_KEY(rmem);
//   TSKEY r2 = TD_ROW_KEY(rimem);

//   if (r1 == r2) {
//     if (TD_SUPPORT_UPDATE(update)) {
//       pCheckInfo->chosen = CHECKINFO_CHOSEN_BOTH;
//       *extraRow = rimem;
//       return rmem;
//     } else {
//       tsdbTbDataIterNext(pCheckInfo->iter);
//       pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
//       return rimem;
//     }
//   } else {
//     if (ASCENDING_TRAVERSE(order)) {
//       if (r1 < r2) {
//         pCheckInfo->chosen = CHECKINFO_CHOSEN_MEM;
//         return rmem;
//       } else {
//         pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
//         return rimem;
//       }
//     } else {
//       if (r1 < r2) {
//         pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
//         return rimem;
//       } else {
//         pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
//         return rmem;
//       }
//     }
//   }
H
Hongze Cheng 已提交
651
// }
H
Hongze Cheng 已提交
652

H
Haojun Liao 已提交
653
// static bool moveToNextRowInMem(STableBlockScanInfo* pCheckInfo) {
H
Hongze Cheng 已提交
654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686
//   bool hasNext = false;
//   if (pCheckInfo->chosen == CHECKINFO_CHOSEN_MEM) {
//     if (pCheckInfo->iter != NULL) {
//       hasNext = tsdbTbDataIterNext(pCheckInfo->iter);
//     }

//     if (hasNext) {
//       return hasNext;
//     }

//     if (pCheckInfo->iiter != NULL) {
//       return tsdbTbDataIterGet(pCheckInfo->iiter, NULL);
//     }
//   } else if (pCheckInfo->chosen == CHECKINFO_CHOSEN_IMEM) {
//     if (pCheckInfo->iiter != NULL) {
//       hasNext = tsdbTbDataIterNext(pCheckInfo->iiter);
//     }

//     if (hasNext) {
//       return hasNext;
//     }

//     if (pCheckInfo->iter != NULL) {
//       return tsdbTbDataIterGet(pCheckInfo->iter, NULL);
//     }
//   } else {
//     if (pCheckInfo->iter != NULL) {
//       hasNext = tsdbTbDataIterNext(pCheckInfo->iter);
//     }
//     if (pCheckInfo->iiter != NULL) {
//       hasNext = tsdbTbDataIterNext(pCheckInfo->iiter) || hasNext;
//     }
//   }
H
Hongze Cheng 已提交
687

H
Hongze Cheng 已提交
688 689
//   return hasNext;
// }
H
Hongze Cheng 已提交
690

H
Hongze Cheng 已提交
691 692 693 694 695
// static int32_t getFileIdFromKey(TSKEY key, int32_t daysPerFile, int32_t precision) {
//   assert(precision >= TSDB_TIME_PRECISION_MICRO || precision <= TSDB_TIME_PRECISION_NANO);
//   if (key == TSKEY_INITIAL_VAL) {
//     return INT32_MIN;
//   }
H
Hongze Cheng 已提交
696

H
Hongze Cheng 已提交
697 698 699
//   if (key < 0) {
//     key -= (daysPerFile * tsTickPerMin[precision]);
//   }
H
Hongze Cheng 已提交
700

H
Hongze Cheng 已提交
701 702 703 704
//   int64_t fid = (int64_t)(key / (daysPerFile * tsTickPerMin[precision]));  // set the starting fileId
//   if (fid < 0L && llabs(fid) > INT32_MAX) {                                // data value overflow for INT32
//     fid = INT32_MIN;
//   }
H
Hongze Cheng 已提交
705

H
Hongze Cheng 已提交
706 707 708
//   if (fid > 0L && fid > INT32_MAX) {
//     fid = INT32_MAX;
//   }
H
Hongze Cheng 已提交
709

H
Hongze Cheng 已提交
710 711
//   return (int32_t)fid;
// }
H
Hongze Cheng 已提交
712

H
Hongze Cheng 已提交
713 714 715
// static int32_t binarySearchForBlock(SBlock* pBlock, int32_t numOfBlocks, TSKEY skey, int32_t order) {
//   int32_t firstSlot = 0;
//   int32_t lastSlot = numOfBlocks - 1;
H
Hongze Cheng 已提交
716

H
Hongze Cheng 已提交
717
//   int32_t midSlot = firstSlot;
H
Hongze Cheng 已提交
718

H
Hongze Cheng 已提交
719 720 721
//   while (1) {
//     numOfBlocks = lastSlot - firstSlot + 1;
//     midSlot = (firstSlot + (numOfBlocks >> 1));
H
Hongze Cheng 已提交
722

H
Hongze Cheng 已提交
723
//     if (numOfBlocks == 1) break;
H
Hongze Cheng 已提交
724

H
Hongze Cheng 已提交
725 726 727 728 729 730 731 732 733 734 735
//     if (skey > pBlock[midSlot].maxKey.ts) {
//       if (numOfBlocks == 2) break;
//       if ((order == TSDB_ORDER_DESC) && (skey < pBlock[midSlot + 1].minKey.ts)) break;
//       firstSlot = midSlot + 1;
//     } else if (skey < pBlock[midSlot].minKey.ts) {
//       if ((order == TSDB_ORDER_ASC) && (skey > pBlock[midSlot - 1].maxKey.ts)) break;
//       lastSlot = midSlot - 1;
//     } else {
//       break;  // got the slot
//     }
//   }
H
Hongze Cheng 已提交
736

H
Hongze Cheng 已提交
737 738
//   return midSlot;
// }
H
Hongze Cheng 已提交
739

H
Haojun Liao 已提交
740 741
static int32_t doLoadBlockIndex(STsdbReader* pReader, SDataFReader* pFileReader, SArray* pIndexList) {
  int32_t code = 0;
742
  bool asc = ASCENDING_TRAVERSE(pReader->order);
H
Hongze Cheng 已提交
743

H
Hongze Cheng 已提交
744
  SArray *aBlockIdx = taosArrayInit(0, sizeof(SBlockIdx));
H
Hongze Cheng 已提交
745

H
Hongze Cheng 已提交
746
  code = tsdbReadBlockIdx(pFileReader, aBlockIdx, NULL);
H
Haojun Liao 已提交
747 748 749
  if (code != TSDB_CODE_SUCCESS) {
    goto _err;
  }
H
Hongze Cheng 已提交
750

H
Hongze Cheng 已提交
751 752
  if (taosArrayGetSize(aBlockIdx) == 0) {
    taosArrayClear(aBlockIdx);
H
Haojun Liao 已提交
753 754
    return TSDB_CODE_SUCCESS;
  }
H
Hongze Cheng 已提交
755

H
Hongze Cheng 已提交
756 757 758
  SBlockIdx *pBlockIdx;
  for (int32_t i = 0; i < taosArrayGetSize(aBlockIdx); ++i) {
    pBlockIdx = (SBlockIdx *)taosArrayGet(aBlockIdx, i);
H
Haojun Liao 已提交
759

H
Hongze Cheng 已提交
760
    if (pBlockIdx->suid != pReader->suid) {
H
Haojun Liao 已提交
761 762 763 764
      continue;
    }

    // this block belongs to a table that is not queried.
H
Hongze Cheng 已提交
765
    void* p = taosHashGet(pReader->status.pTableMap, &pBlockIdx->uid, sizeof(uint64_t));
H
Haojun Liao 已提交
766 767 768 769
    if (p == NULL) {
      continue;
    }

770 771
    if ((asc && (pBlockIdx->minKey > pReader->window.ekey || pBlockIdx->maxKey < pReader->window.skey)) /*||
        (!asc && (pBlockIdx->minKey > pReader->window.skey || pBlockIdx->maxKey < pReader->window.ekey))*/) {
H
Haojun Liao 已提交
772 773 774 775 776 777 778 779
      continue;
    }

    STableBlockScanInfo* pScanInfo = p;
    if (pScanInfo->pBlockList == NULL) {
      pScanInfo->pBlockList = taosArrayInit(16, sizeof(SBlock));
    }

H
Hongze Cheng 已提交
780 781
    pScanInfo->blockIdx = *pBlockIdx;
    taosArrayPush(pIndexList, pBlockIdx);
H
Haojun Liao 已提交
782
  }
H
Hongze Cheng 已提交
783

H
Hongze Cheng 已提交
784
  taosArrayDestroy(aBlockIdx);
H
Haojun Liao 已提交
785
  return TSDB_CODE_SUCCESS;
H
Hongze Cheng 已提交
786

H
Haojun Liao 已提交
787
_err:
H
Hongze Cheng 已提交
788
  taosArrayDestroy(aBlockIdx);
H
Haojun Liao 已提交
789 790
  return code;
}
H
Hongze Cheng 已提交
791

792
static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, uint32_t* numOfValidTables, int32_t* numOfBlocks) {
H
Haojun Liao 已提交
793
  size_t numOfTables = taosArrayGetSize(pIndexList);
H
Hongze Cheng 已提交
794

H
Haojun Liao 已提交
795
  *numOfValidTables = 0;
H
Hongze Cheng 已提交
796

H
Haojun Liao 已提交
797 798
  for(int32_t i = 0; i < numOfTables; ++i) {
    SBlockIdx* pBlockIdx = taosArrayGet(pIndexList, i);
H
Hongze Cheng 已提交
799

H
Hongze Cheng 已提交
800
    SMapData mapData = {0};
H
Haojun Liao 已提交
801 802
    tMapDataReset(&mapData);
    tsdbReadBlock(pReader->pFileReader, pBlockIdx, &mapData, NULL);
H
Hongze Cheng 已提交
803

H
Haojun Liao 已提交
804
    STableBlockScanInfo* pScanInfo = taosHashGet(pReader->status.pTableMap, &pBlockIdx->uid, sizeof(int64_t));
805 806
    taosArrayClear(pScanInfo->pBlockList);

H
Haojun Liao 已提交
807 808
    for (int32_t j = 0; j < mapData.nItem; ++j) {
      SBlock block = {0};
H
Hongze Cheng 已提交
809

H
Haojun Liao 已提交
810 811 812 813
      int32_t code = tMapDataGetItemByIdx(&mapData, j, &block, tGetBlock);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
H
Hongze Cheng 已提交
814

815
      // 1. time range check
816
      if (block.minKey.ts > pReader->window.ekey || block.maxKey.ts < pReader->window.skey) {
H
Haojun Liao 已提交
817 818
        continue;
      }
H
Hongze Cheng 已提交
819

820
      // 2. version range check
821 822 823
      if (block.minVersion > pReader->verRange.maxVer || block.maxVersion < pReader->verRange.minVer) {
        continue;
      }
824

H
Haojun Liao 已提交
825 826 827 828
      void* p = taosArrayPush(pScanInfo->pBlockList, &block);
      if (p == NULL) {
        return TSDB_CODE_OUT_OF_MEMORY;
      }
829 830

      (*numOfBlocks) += 1;
H
Haojun Liao 已提交
831
    }
H
Hongze Cheng 已提交
832

H
Haojun Liao 已提交
833 834 835 836
    if (pScanInfo->pBlockList != NULL && taosArrayGetSize(pScanInfo->pBlockList) > 0) {
      (*numOfValidTables) += 1;
    }
  }
H
Hongze Cheng 已提交
837

H
Haojun Liao 已提交
838 839
  return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
840

841 842
// todo remove pblock parameter
static void setBlockAllDumped(SFileBlockDumpInfo* pDumpInfo, SBlock* pBlock, int32_t order) {
H
Haojun Liao 已提交
843 844
  int32_t step = ASCENDING_TRAVERSE(order)? 1:-1;

845
  pDumpInfo->allDumped = true;
H
Haojun Liao 已提交
846 847 848 849 850 851 852 853 854 855 856 857 858 859 860
  pDumpInfo->lastKey   = pBlock->maxKey.ts + step;
}

static void doCopyColVal(SColumnInfoData* pColInfoData, int32_t rowIndex, int32_t colIndex, SColVal* pColVal, SBlockLoadSuppInfo* pSup) {
  if (IS_VAR_DATA_TYPE(pColVal->type)) {
    if (pColVal->isNull) {
      colDataAppendNULL(pColInfoData, rowIndex);
    } else {
      varDataSetLen(pSup->buildBuf[colIndex], pColVal->value.nData);
      memcpy(varDataVal(pSup->buildBuf[colIndex]), pColVal->value.pData, pColVal->value.nData);
      colDataAppend(pColInfoData, rowIndex, pSup->buildBuf[colIndex], false);
    }
  } else {
    colDataAppend(pColInfoData, rowIndex, (const char*)&pColVal->value, pColVal->isNull);
  }
H
Haojun Liao 已提交
861 862
}

863 864 865
static int32_t copyBlockDataToSDataBlock(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo) {
  SReaderStatus* pStatus = &pReader->status;
  SDataBlockIter* pBlockIter = &pStatus->blockIter;
H
Hongze Cheng 已提交
866

867
  SBlockData*  pBlockData = &pStatus->fileBlockData;
H
Haojun Liao 已提交
868
  SFileDataBlockInfo* pFBlock = getCurrentBlockInfo(pBlockIter);
H
Haojun Liao 已提交
869 870
  SBlock*             pBlock = taosArrayGet(pBlockScanInfo->pBlockList, pFBlock->tbBlockIdx);
  SSDataBlock*        pResBlock = pReader->pResBlock;
871
  int32_t             numOfCols = blockDataGetNumOfCols(pResBlock);
H
Haojun Liao 已提交
872

H
Haojun Liao 已提交
873
  SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo;
874
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
H
Haojun Liao 已提交
875

876
  int64_t st = taosGetTimestampUs();
H
Haojun Liao 已提交
877

H
Haojun Liao 已提交
878
  SColVal cv = {0};
879 880
  int32_t colIndex = 0;

881 882
  bool    asc = ASCENDING_TRAVERSE(pReader->order);
  int32_t step = asc ? 1 : -1;
883

884
  int32_t rowIndex = 0;
885 886
  int32_t remain = asc ? (pBlockData->nRow - pDumpInfo->rowIndex) : (pDumpInfo->rowIndex + 1);

887 888 889 890 891 892 893 894
  int32_t endIndex = 0;
  if (remain <= pReader->capacity) {
    endIndex = pBlockData->nRow;
  } else {
    endIndex = pDumpInfo->rowIndex + step * pReader->capacity;
    remain = pReader->capacity;
  }

895 896 897
  int32_t i = 0;
  SColumnInfoData* pColData = taosArrayGet(pResBlock->pDataBlock, i);
  if (pColData->info.colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
898
    for (int32_t j = pDumpInfo->rowIndex; j < endIndex && j >= 0; j += step) {
899 900 901 902 903 904 905 906 907 908 909 910
      colDataAppend(pColData, rowIndex++, (const char*)&pBlockData->aTSKEY[j], false);
    }
    i += 1;
  }

  while(i < numOfCols && colIndex < taosArrayGetSize(pBlockData->aColDataP)) {
    rowIndex = 0;
    pColData = taosArrayGet(pResBlock->pDataBlock, i);

    SColData* pData = (SColData*)taosArrayGetP(pBlockData->aColDataP, colIndex);

    if (pData->cid == pColData->info.colId) {
911
      for (int32_t j = pDumpInfo->rowIndex; j < endIndex && j >= 0; j += step) {
912 913
        tColDataGetValue(pData, j, &cv);
        doCopyColVal(pColData, rowIndex++, i, &cv, pSupInfo);
H
Haojun Liao 已提交
914
      }
915 916 917
      colIndex += 1;
    } else {  // the specified column does not exist in file block, fill with null data
      colDataAppendNNULL(pColData, 0, remain);
H
Haojun Liao 已提交
918
    }
919 920 921 922 923 924 925 926 927

    ASSERT(rowIndex == remain);
    i += 1;
  }

  while(i < numOfCols) {
    pColData = taosArrayGet(pResBlock->pDataBlock, i);
    colDataAppendNNULL(pColData, 0, remain);
    i += 1;
H
Haojun Liao 已提交
928
  }
H
Haojun Liao 已提交
929

930 931 932 933
  pResBlock->info.rows = remain;
  pDumpInfo->rowIndex += step*remain;

  setBlockAllDumped(pDumpInfo, pBlock, pReader->order);
H
Haojun Liao 已提交
934

H
Haojun Liao 已提交
935 936
  int64_t elapsedTime = (taosGetTimestampUs() - st);
  pReader->cost.blockLoadTime += elapsedTime;
H
Haojun Liao 已提交
937

938
  int32_t unDumpedRows = asc? pBlock->nRow - pDumpInfo->rowIndex: pDumpInfo->rowIndex + 1;
H
Haojun Liao 已提交
939
  tsdbDebug("%p load file block into buffer, global index:%d, table index:%d, brange:%" PRId64 "-%" PRId64
940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971
                ", rows:%d, remain:%d, minVer:%" PRId64 ", maxVer:%" PRId64 ", elapsed time:%" PRId64 " us, %s",
            pReader, pBlockIter->index, pFBlock->tbBlockIdx, pBlock->minKey.ts, pBlock->maxKey.ts, remain, unDumpedRows,
            pBlock->minVersion, pBlock->maxVersion, elapsedTime, pReader->idStr);

  return TSDB_CODE_SUCCESS;
}

// todo consider the output buffer size
static int32_t doLoadFileBlockData(STsdbReader* pReader, SDataBlockIter* pBlockIter, STableBlockScanInfo* pBlockScanInfo, SBlockData* pBlockData) {
  int64_t st = taosGetTimestampUs();

  SFileDataBlockInfo* pFBlock = getCurrentBlockInfo(pBlockIter);
  SBlock*             pBlock = taosArrayGet(pBlockScanInfo->pBlockList, pFBlock->tbBlockIdx);
  SSDataBlock*        pResBlock = pReader->pResBlock;
  int32_t             numOfCols = blockDataGetNumOfCols(pResBlock);

  SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo;
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;

  uint8_t *pb = NULL, *pb1 = NULL;
  int32_t  code = tsdbReadColData(pReader->pFileReader, &pBlockScanInfo->blockIdx, pBlock, pSupInfo->colIds, numOfCols, pBlockData, &pb, &pb1);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

  int64_t elapsedTime = (taosGetTimestampUs() - st);
  pReader->cost.blockLoadTime += elapsedTime;

  pDumpInfo->allDumped = false;
  tsdbDebug("%p load file block into buffer, global index:%d, table index:%d, brange:%" PRId64 "-%" PRId64
                ", rows:%d, minVer:%" PRId64 ", maxVer:%" PRId64 ", elapsed time:%" PRId64 " us, %s",
            pReader, pBlockIter->index, pFBlock->tbBlockIdx, pBlock->minKey.ts, pBlock->maxKey.ts, pBlock->nRow,
H
Haojun Liao 已提交
972 973
            pBlock->minVersion, pBlock->maxVersion, elapsedTime, pReader->idStr);
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
974 975

_error:
H
Haojun Liao 已提交
976 977 978 979 980
  tsdbError("%p error occurs in loading file block, global index:%d, table index:%d, brange:%" PRId64 "-%" PRId64
            ", rows:%d, %s",
            pReader, pBlockIter->index, pFBlock->tbBlockIdx, pBlock->minKey.ts, pBlock->maxKey.ts, pBlock->nRow,
            pReader->idStr);
  return code;
H
Haojun Liao 已提交
981
}
H
Hongze Cheng 已提交
982

H
Haojun Liao 已提交
983
// static int32_t handleDataMergeIfNeeded(STsdbReader* pTsdbReadHandle, SBlock* pBlock, STableBlockScanInfo* pCheckInfo) {
H
Hongze Cheng 已提交
984 985 986 987 988
//   SQueryFilePos* cur = &pTsdbReadHandle->cur;
//   STsdbCfg*      pCfg = REPO_CFG(pTsdbReadHandle->pTsdb);
//   SDataBlockInfo binfo = GET_FILE_DATA_BLOCK_INFO(pCheckInfo, pBlock);
//   TSKEY          key;
//   int32_t        code = TSDB_CODE_SUCCESS;
H
Hongze Cheng 已提交
989

H
Hongze Cheng 已提交
990 991
//   /*bool hasData = */ initTableMemIterator(pTsdbReadHandle, pCheckInfo);
//   assert(cur->pos >= 0 && cur->pos <= binfo.rows);
H
Hongze Cheng 已提交
992

H
Hongze Cheng 已提交
993
//   key = extractFirstTraverseKey(pCheckInfo, pTsdbReadHandle->order, pCfg->update, TD_VER_MAX);
H
Hongze Cheng 已提交
994

H
Hongze Cheng 已提交
995 996 997 998 999
//   if (key != TSKEY_INITIAL_VAL) {
//     tsdbDebug("%p key in mem:%" PRId64 ", %s", pTsdbReadHandle, key, pTsdbReadHandle->idStr);
//   } else {
//     tsdbDebug("%p no data in mem, %s", pTsdbReadHandle, pTsdbReadHandle->idStr);
//   }
H
Hongze Cheng 已提交
1000

H
Hongze Cheng 已提交
1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012
//   bool ascScan = ASCENDING_TRAVERSE(pTsdbReadHandle->order);

//   if ((ascScan && (key != TSKEY_INITIAL_VAL && key <= binfo.window.ekey)) ||
//       (!ascScan && (key != TSKEY_INITIAL_VAL && key >= binfo.window.skey))) {
//     bool cacheDataInFileBlockHole = (ascScan && (key != TSKEY_INITIAL_VAL && key < binfo.window.skey)) ||
//                                     (!ascScan && (key != TSKEY_INITIAL_VAL && key > binfo.window.ekey));
//     if (cacheDataInFileBlockHole) {
//       // do not load file block into buffer
//       int32_t step = ascScan ? 1 : -1;

//       TSKEY maxKey = ascScan ? (binfo.window.skey - step) : (binfo.window.ekey - step);
//       cur->rows =
1013
//           buildInmemDataBlockImpl(pCheckInfo, maxKey, pTsdbReadHandle->outputCapacity, &cur->win, pTsdbReadHandle);
H
Hongze Cheng 已提交
1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028
//       pTsdbReadHandle->realNumOfRows = cur->rows;

//       // update the last key value
//       pCheckInfo->lastKey = cur->win.ekey + step;

//       if (!ascScan) {
//         TSWAP(cur->win.skey, cur->win.ekey);
//       }

//       cur->mixBlock = true;
//       cur->blockCompleted = false;
//       return code;
//     }

//     // return error, add test cases
H
Haojun Liao 已提交
1029
//     if ((code = doLoadFileBlockData(pTsdbReadHandle, pBlock, pCheckInfo, cur->slot)) != TSDB_CODE_SUCCESS) {
H
Hongze Cheng 已提交
1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061
//       return code;
//     }

//     doMergeTwoLevelData(pTsdbReadHandle, pCheckInfo, pBlock);
//   } else {
//     /*
//      * no data in cache, only load data from file
//      * during the query processing, data in cache will not be checked anymore.
//      * Here the buffer is not enough, so only part of file block can be loaded into memory buffer
//      */
//     int32_t endPos = getEndPosInDataBlock(pTsdbReadHandle, &binfo);

//     bool wholeBlockReturned = ((abs(cur->pos - endPos) + 1) == binfo.rows);
//     if (wholeBlockReturned) {
//       pTsdbReadHandle->realNumOfRows = binfo.rows;

//       cur->rows = binfo.rows;
//       cur->win = binfo.window;
//       cur->mixBlock = false;
//       cur->blockCompleted = true;

//       if (ascScan) {
//         cur->lastKey = binfo.window.ekey + 1;
//         cur->pos = binfo.rows;
//       } else {
//         cur->lastKey = binfo.window.skey - 1;
//         cur->pos = -1;
//       }
//     } else {  // partially copy to dest buffer
//       // make sure to only load once
//       bool firstTimeExtract = ((cur->pos == 0 && ascScan) || (cur->pos == binfo.rows - 1 && (!ascScan)));
//       if (pTsdbReadHandle->outputCapacity < binfo.rows && firstTimeExtract) {
H
Haojun Liao 已提交
1062
//         code = doLoadFileBlockData(pTsdbReadHandle, pBlock, pCheckInfo, cur->slot);
H
Hongze Cheng 已提交
1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085
//         if (code != TSDB_CODE_SUCCESS) {
//           return code;
//         }
//       }

//       copyAllRemainRowsFromFileBlock(pTsdbReadHandle, pCheckInfo, &binfo, endPos);
//       cur->mixBlock = true;
//     }

//     if (pTsdbReadHandle->outputCapacity >= binfo.rows) {
//       ASSERT(cur->blockCompleted || cur->mixBlock);
//     }

//     if (cur->rows == binfo.rows) {
//       tsdbDebug("%p whole file block qualified, brange:%" PRId64 "-%" PRId64 ", rows:%d, lastKey:%" PRId64 ", %s",
//                 pTsdbReadHandle, cur->win.skey, cur->win.ekey, cur->rows, cur->lastKey, pTsdbReadHandle->idStr);
//     } else {
//       tsdbDebug("%p create data block from remain file block, brange:%" PRId64 "-%" PRId64
//                 ", rows:%d, total:%d, lastKey:%" PRId64 ", %s",
//                 pTsdbReadHandle, cur->win.skey, cur->win.ekey, cur->rows, binfo.rows, cur->lastKey,
//                 pTsdbReadHandle->idStr);
//     }
//   }
H
Hongze Cheng 已提交
1086

H
Hongze Cheng 已提交
1087 1088
//   return code;
// }
H
Hongze Cheng 已提交
1089

H
Haojun Liao 已提交
1090
// static int32_t loadFileDataBlock(STsdbReader* pTsdbReadHandle, SBlock* pBlock, STableBlockScanInfo* pCheckInfo,
H
Hongze Cheng 已提交
1091 1092 1093 1094 1095 1096 1097 1098
//                                  bool* exists) {
//   SQueryFilePos* cur = &pTsdbReadHandle->cur;
//   int32_t        code = TSDB_CODE_SUCCESS;
//   bool           asc = ASCENDING_TRAVERSE(pTsdbReadHandle->order);

//   if (asc) {
//     // query ended in/started from current block
//     if (pTsdbReadHandle->window.ekey < pBlock->maxKey.ts || pCheckInfo->lastKey > pBlock->minKey.ts) {
H
Haojun Liao 已提交
1099
//       if ((code = doLoadFileBlockData(pTsdbReadHandle, pBlock, pCheckInfo, cur->slot)) != TSDB_CODE_SUCCESS) {
H
Hongze Cheng 已提交
1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122
//         *exists = false;
//         return code;
//       }

//       SDataCols* pTSCol = pTsdbReadHandle->rhelper.pDCols[0];
//       assert(pTSCol->cols->type == TSDB_DATA_TYPE_TIMESTAMP && pTSCol->numOfRows == pBlock->numOfRows);

//       if (pCheckInfo->lastKey > pBlock->minKey.ts) {
//         cur->pos =
//             binarySearchForKey(pTSCol->cols[0].pData, pBlock->numOfRows, pCheckInfo->lastKey,
//             pTsdbReadHandle->order);
//       } else {
//         cur->pos = 0;
//       }

//       assert(pCheckInfo->lastKey <= pBlock->maxKey.ts);
//       doMergeTwoLevelData(pTsdbReadHandle, pCheckInfo, pBlock);
//     } else {  // the whole block is loaded in to buffer
//       cur->pos = asc ? 0 : (pBlock->numOfRows - 1);
//       code = handleDataMergeIfNeeded(pTsdbReadHandle, pBlock, pCheckInfo);
//     }
//   } else {  // desc order, query ended in current block
//     if (pTsdbReadHandle->window.ekey > pBlock->minKey.ts || pCheckInfo->lastKey < pBlock->maxKey.ts) {
H
Haojun Liao 已提交
1123
//       if ((code = doLoadFileBlockData(pTsdbReadHandle, pBlock, pCheckInfo, cur->slot)) != TSDB_CODE_SUCCESS) {
H
Hongze Cheng 已提交
1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143
//         *exists = false;
//         return code;
//       }

//       SDataCols* pTsCol = pTsdbReadHandle->rhelper.pDCols[0];
//       if (pCheckInfo->lastKey < pBlock->maxKey.ts) {
//         cur->pos =
//             binarySearchForKey(pTsCol->cols[0].pData, pBlock->numOfRows, pCheckInfo->lastKey,
//             pTsdbReadHandle->order);
//       } else {
//         cur->pos = pBlock->numOfRows - 1;
//       }

//       assert(pCheckInfo->lastKey >= pBlock->minKey.ts);
//       doMergeTwoLevelData(pTsdbReadHandle, pCheckInfo, pBlock);
//     } else {
//       cur->pos = asc ? 0 : (pBlock->numOfRows - 1);
//       code = handleDataMergeIfNeeded(pTsdbReadHandle, pBlock, pCheckInfo);
//     }
//   }
H
Hongze Cheng 已提交
1144

H
Hongze Cheng 已提交
1145 1146 1147
//   *exists = pTsdbReadHandle->realNumOfRows > 0;
//   return code;
// }
H
Hongze Cheng 已提交
1148

H
Hongze Cheng 已提交
1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206
// static int doBinarySearchKey(char* pValue, int num, TSKEY key, int order) {
//   int    firstPos, lastPos, midPos = -1;
//   int    numOfRows;
//   TSKEY* keyList;

//   assert(order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC);

//   if (num <= 0) return -1;

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

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

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

//       if (key < keyList[midPos]) {
//         lastPos = midPos - 1;
//       } else if (key > keyList[midPos]) {
//         firstPos = 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 >> 1) + firstPos;

//       if (key < keyList[midPos]) {
//         lastPos = midPos - 1;
//       } else if (key > keyList[midPos]) {
//         firstPos = midPos + 1;
//       } else {
//         break;
//       }
//     }
//   }
H
Hongze Cheng 已提交
1207

H
Hongze Cheng 已提交
1208 1209
//   return midPos;
// }
H
Hongze Cheng 已提交
1210

H
Hongze Cheng 已提交
1211 1212 1213 1214 1215
// static int32_t doCopyRowsFromFileBlock(STsdbReader* pTsdbReadHandle, int32_t capacity, int32_t numOfRows, int32_t
// start,
//                                        int32_t end) {
//   SDataCols* pCols = pTsdbReadHandle->rhelper.pDCols[0];
//   TSKEY*     tsArray = pCols->cols[0].pData;
H
Hongze Cheng 已提交
1216

H
Hongze Cheng 已提交
1217 1218
//   int32_t num = end - start + 1;
//   assert(num >= 0);
H
Hongze Cheng 已提交
1219

H
Hongze Cheng 已提交
1220 1221 1222
//   if (num == 0) {
//     return numOfRows;
//   }
H
Hongze Cheng 已提交
1223

H
Hongze Cheng 已提交
1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284
//   bool    ascScan = ASCENDING_TRAVERSE(pTsdbReadHandle->order);
//   int32_t trueStart = ascScan ? start : end;
//   int32_t trueEnd = ascScan ? end : start;
//   int32_t step = ascScan ? 1 : -1;

//   int32_t requiredNumOfCols = (int32_t)taosArrayGetSize(pTsdbReadHandle->pColumns);

//   // data in buffer has greater timestamp, copy data in file block
//   int32_t i = 0, j = 0;
//   while (i < requiredNumOfCols && j < pCols->numOfCols) {
//     SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);

//     SDataCol* src = &pCols->cols[j];
//     if (src->colId < pColInfo->info.colId) {
//       j++;
//       continue;
//     }

//     if (!isAllRowsNull(src) && pColInfo->info.colId == src->colId) {
//       if (!IS_VAR_DATA_TYPE(pColInfo->info.type)) {  // todo opt performance
//         //        memmove(pData, (char*)src->pData + bytes * start, bytes * num);
//         int32_t rowIndex = numOfRows;
//         for (int32_t k = trueStart; ((ascScan && k <= trueEnd) || (!ascScan && k >= trueEnd)); k += step, ++rowIndex)
//         {
//           SCellVal sVal = {0};
//           if (tdGetColDataOfRow(&sVal, src, k, pCols->bitmapMode) < 0) {
//             TASSERT(0);
//           }

//           if (sVal.valType == TD_VTYPE_NORM) {
//             colDataAppend(pColInfo, rowIndex, sVal.val, false);
//           } else {
//             colDataAppendNULL(pColInfo, rowIndex);
//           }
//         }
//       } else {  // handle the var-string
//         int32_t rowIndex = numOfRows;

//         // todo refactor, only copy one-by-one
//         for (int32_t k = trueStart; ((ascScan && k <= trueEnd) || (!ascScan && k >= trueEnd)); k += step, ++rowIndex)
//         {
//           SCellVal sVal = {0};
//           if (tdGetColDataOfRow(&sVal, src, k, pCols->bitmapMode) < 0) {
//             TASSERT(0);
//           }

//           if (sVal.valType == TD_VTYPE_NORM) {
//             colDataAppend(pColInfo, rowIndex, sVal.val, false);
//           } else {
//             colDataAppendNULL(pColInfo, rowIndex);
//           }
//         }
//       }

//       j++;
//       i++;
//     } else {  // pColInfo->info.colId < src->colId, it is a NULL data
//       colDataAppendNNULL(pColInfo, numOfRows, num);
//       i++;
//     }
//   }
H
Hongze Cheng 已提交
1285

H
Hongze Cheng 已提交
1286 1287 1288 1289 1290
//   while (i < requiredNumOfCols) {  // the remain columns are all null data
//     SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
//     colDataAppendNNULL(pColInfo, numOfRows, num);
//     i++;
//   }
H
Hongze Cheng 已提交
1291

H
Hongze Cheng 已提交
1292 1293
//   pTsdbReadHandle->cur.win.ekey = tsArray[trueEnd];
//   pTsdbReadHandle->cur.lastKey = tsArray[trueEnd] + step;
H
Hongze Cheng 已提交
1294

H
Hongze Cheng 已提交
1295 1296
//   return numOfRows + num;
// }
H
Hongze Cheng 已提交
1297

H
Hongze Cheng 已提交
1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321
// static int32_t mergeTwoRowFromMem(STsdbReader* pTsdbReadHandle, int32_t capacity, int32_t* curRow, STSRow* row1,
//                                   STSRow* row2, int32_t numOfCols, uint64_t uid, STSchema* pSchema1, STSchema*
//                                   pSchema2, bool update, TSKEY* lastRowKey) {
// #if 1
//   STSchema* pSchema;
//   STSRow*   row;
//   int16_t   colId;
//   int16_t   offset;

//   bool     isRow1DataRow = TD_IS_TP_ROW(row1);
//   bool     isRow2DataRow;
//   bool     isChosenRowDataRow;
//   int32_t  chosen_itr;
//   SCellVal sVal = {0};
//   TSKEY    rowKey = TSKEY_INITIAL_VAL;
//   int32_t  nResult = 0;
//   int32_t  mergeOption = 0;  // 0 discard 1 overwrite 2 merge

//   // the schema version info is embeded in STSRow
//   int32_t numOfColsOfRow1 = 0;

//   if (pSchema1 == NULL) {
//     pSchema1 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, TD_ROW_SVER(row1));
//   }
H
Hongze Cheng 已提交
1322

H
Hongze Cheng 已提交
1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335
// #ifdef TD_DEBUG_PRINT_ROW
//   char   flags[70] = {0};
//   STsdb* pTsdb = pTsdbReadHandle->rhelper.pRepo;
//   snprintf(flags, 70, "%s:%d vgId:%d dir:%s row1%s=NULL,row2%s=NULL", __func__, __LINE__, TD_VID(pTsdb->pVnode),
//            pTsdb->dir, row1 ? "!" : "", row2 ? "!" : "");
//   tdSRowPrint(row1, pSchema1, flags);
// #endif

//   if (isRow1DataRow) {
//     numOfColsOfRow1 = schemaNCols(pSchema1);
//   } else {
//     numOfColsOfRow1 = tdRowGetNCols(row1);
//   }
H
Hongze Cheng 已提交
1336

H
Hongze Cheng 已提交
1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348
//   int32_t numOfColsOfRow2 = 0;
//   if (row2) {
//     isRow2DataRow = TD_IS_TP_ROW(row2);
//     if (pSchema2 == NULL) {
//       pSchema2 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, TD_ROW_SVER(row2));
//     }
//     if (isRow2DataRow) {
//       numOfColsOfRow2 = schemaNCols(pSchema2);
//     } else {
//       numOfColsOfRow2 = tdRowGetNCols(row2);
//     }
//   }
H
Hongze Cheng 已提交
1349

H
Hongze Cheng 已提交
1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378 1379 1380 1381 1382 1383 1384 1385 1386 1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452 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
//   int32_t i = 0, j = 0, k = 0;
//   while (i < numOfCols && (j < numOfColsOfRow1 || k < numOfColsOfRow2)) {
//     SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);

//     int32_t colIdOfRow1;
//     if (j >= numOfColsOfRow1) {
//       colIdOfRow1 = INT32_MAX;
//     } else if (isRow1DataRow) {
//       colIdOfRow1 = pSchema1->columns[j].colId;
//     } else {
//       colIdOfRow1 = tdKvRowColIdAt(row1, j);
//     }

//     int32_t colIdOfRow2;
//     if (k >= numOfColsOfRow2) {
//       colIdOfRow2 = INT32_MAX;
//     } else if (isRow2DataRow) {
//       colIdOfRow2 = pSchema2->columns[k].colId;
//     } else {
//       colIdOfRow2 = tdKvRowColIdAt(row2, k);
//     }

//     if (colIdOfRow1 < colIdOfRow2) {  // the most probability
//       if (colIdOfRow1 < pColInfo->info.colId) {
//         ++j;
//         continue;
//       }
//       row = row1;
//       pSchema = pSchema1;
//       isChosenRowDataRow = isRow1DataRow;
//       chosen_itr = j;
//     } else if (colIdOfRow1 == colIdOfRow2) {
//       if (colIdOfRow1 < pColInfo->info.colId) {
//         ++j;
//         ++k;
//         continue;
//       }
//       row = row1;
//       pSchema = pSchema1;
//       isChosenRowDataRow = isRow1DataRow;
//       chosen_itr = j;
//     } else {
//       if (colIdOfRow2 < pColInfo->info.colId) {
//         ++k;
//         continue;
//       }
//       row = row2;
//       pSchema = pSchema2;
//       chosen_itr = k;
//       isChosenRowDataRow = isRow2DataRow;
//     }

//     if (isChosenRowDataRow) {
//       colId = pSchema->columns[chosen_itr].colId;
//       offset = pSchema->columns[chosen_itr].offset;
//       // TODO: use STSRowIter
//       tdSTpRowGetVal(row, colId, pSchema->columns[chosen_itr].type, pSchema->flen, offset, chosen_itr - 1, &sVal);
//       if (colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
//         rowKey = *(TSKEY*)sVal.val;
//         if (rowKey != *lastRowKey) {
//           mergeOption = 1;
//           if (*lastRowKey != TSKEY_INITIAL_VAL) {
//             ++(*curRow);
//           }
//           *lastRowKey = rowKey;
//           ++nResult;
//         } else if (update) {
//           mergeOption = 2;
//         } else {
//           mergeOption = 0;
//           break;
//         }
//       }
//     } else {
//       // TODO: use STSRowIter
//       if (chosen_itr == 0) {
//         colId = PRIMARYKEY_TIMESTAMP_COL_ID;
//         tdSKvRowGetVal(row, PRIMARYKEY_TIMESTAMP_COL_ID, -1, -1, &sVal);
//         rowKey = *(TSKEY*)sVal.val;
//         if (rowKey != *lastRowKey) {
//           mergeOption = 1;
//           if (*lastRowKey != TSKEY_INITIAL_VAL) {
//             ++(*curRow);
//           }
//           *lastRowKey = rowKey;
//           ++nResult;
//         } else if (update) {
//           mergeOption = 2;
//         } else {
//           mergeOption = 0;
//           break;
//         }
//       } else {
//         SKvRowIdx* pColIdx = tdKvRowColIdxAt(row, chosen_itr - 1);
//         colId = pColIdx->colId;
//         offset = pColIdx->offset;
//         tdSKvRowGetVal(row, colId, offset, chosen_itr - 1, &sVal);
//       }
//     }

//     ASSERT(rowKey != TSKEY_INITIAL_VAL);

//     if (colId == pColInfo->info.colId) {
//       if (tdValTypeIsNorm(sVal.valType)) {
//         colDataAppend(pColInfo, *curRow, sVal.val, false);
//       } else if (tdValTypeIsNull(sVal.valType)) {
//         colDataAppend(pColInfo, *curRow, NULL, true);
//       } else if (tdValTypeIsNone(sVal.valType)) {
//         // TODO: Set null if nothing append for this row
//         if (mergeOption == 1) {
//           colDataAppend(pColInfo, *curRow, NULL, true);
//         }
//       } else {
//         ASSERT(0);
//       }

//       ++i;

//       if (row == row1) {
//         ++j;
//       } else {
//         ++k;
//       }
//     } else {
//       if (mergeOption == 1) {
//         colDataAppend(pColInfo, *curRow, NULL, true);
//       }
//       ++i;
//     }
//   }
H
Hongze Cheng 已提交
1480

H
Hongze Cheng 已提交
1481 1482 1483 1484 1485 1486 1487
//   if (mergeOption == 1) {
//     while (i < numOfCols) {  // the remain columns are all null data
//       SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
//       colDataAppend(pColInfo, *curRow, NULL, true);
//       ++i;
//     }
//   }
H
Hongze Cheng 已提交
1488

H
Hongze Cheng 已提交
1489 1490 1491
//   return nResult;
// #endif
// }
H
Hongze Cheng 已提交
1492

H
Hongze Cheng 已提交
1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517
// static void getQualifiedRowsPos(STsdbReader* pTsdbReadHandle, int32_t startPos, int32_t endPos, int32_t numOfExisted,
//                                 int32_t* start, int32_t* end) {
//   *start = -1;

//   if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
//     int32_t remain = endPos - startPos + 1;
//     if (remain + numOfExisted > pTsdbReadHandle->outputCapacity) {
//       *end = (pTsdbReadHandle->outputCapacity - numOfExisted) + startPos - 1;
//     } else {
//       *end = endPos;
//     }

//     *start = startPos;
//   } else {
//     int32_t remain = (startPos - endPos) + 1;
//     if (remain + numOfExisted > pTsdbReadHandle->outputCapacity) {
//       *end = startPos + 1 - (pTsdbReadHandle->outputCapacity - numOfExisted);
//     } else {
//       *end = endPos;
//     }

//     *start = *end;
//     *end = startPos;
//   }
// }
H
Hongze Cheng 已提交
1518

H
Haojun Liao 已提交
1519
// static void updateInfoAfterMerge(STsdbReader* pTsdbReadHandle, STableBlockScanInfo* pCheckInfo, int32_t numOfRows,
H
Hongze Cheng 已提交
1520 1521
//                                  int32_t endPos) {
//   SQueryFilePos* cur = &pTsdbReadHandle->cur;
H
Hongze Cheng 已提交
1522

H
Hongze Cheng 已提交
1523 1524 1525 1526 1527
//   pCheckInfo->lastKey = cur->lastKey;
//   pTsdbReadHandle->realNumOfRows = numOfRows;
//   cur->rows = numOfRows;
//   cur->pos = endPos;
// }
H
Hongze Cheng 已提交
1528

H
Hongze Cheng 已提交
1529 1530
// static void doCheckGeneratedBlockRange(STsdbReader* pTsdbReadHandle) {
//   SQueryFilePos* cur = &pTsdbReadHandle->cur;
H
Hongze Cheng 已提交
1531

H
Hongze Cheng 已提交
1532 1533 1534 1535 1536 1537
//   if (cur->rows > 0) {
//     if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
//       assert(cur->win.skey >= pTsdbReadHandle->window.skey && cur->win.ekey <= pTsdbReadHandle->window.ekey);
//     } else {
//       assert(cur->win.skey >= pTsdbReadHandle->window.ekey && cur->win.ekey <= pTsdbReadHandle->window.skey);
//     }
H
Hongze Cheng 已提交
1538

H
Hongze Cheng 已提交
1539 1540 1541 1542 1543
//     SColumnInfoData* pColInfoData = taosArrayGet(pTsdbReadHandle->pColumns, 0);
//     assert(cur->win.skey == ((TSKEY*)pColInfoData->pData)[0] &&
//            cur->win.ekey == ((TSKEY*)pColInfoData->pData)[cur->rows - 1]);
//   } else {
//     cur->win = pTsdbReadHandle->window;
H
Hongze Cheng 已提交
1544

H
Hongze Cheng 已提交
1545 1546 1547 1548
//     int32_t step = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 1 : -1;
//     cur->lastKey = pTsdbReadHandle->window.ekey + step;
//   }
// }
H
Hongze Cheng 已提交
1549

H
Haojun Liao 已提交
1550
// static void copyAllRemainRowsFromFileBlock(STsdbReader* pTsdbReadHandle, STableBlockScanInfo* pCheckInfo,
H
Hongze Cheng 已提交
1551 1552
//                                            SDataBlockInfo* pBlockInfo, int32_t endPos) {
//   SQueryFilePos* cur = &pTsdbReadHandle->cur;
H
Hongze Cheng 已提交
1553

H
Hongze Cheng 已提交
1554 1555
//   SDataCols* pCols = pTsdbReadHandle->rhelper.pDCols[0];
//   TSKEY*     tsArray = pCols->cols[0].pData;
H
Hongze Cheng 已提交
1556

H
Hongze Cheng 已提交
1557
//   bool ascScan = ASCENDING_TRAVERSE(pTsdbReadHandle->order);
H
Hongze Cheng 已提交
1558

H
Hongze Cheng 已提交
1559
//   int32_t step = ascScan ? 1 : -1;
H
Hongze Cheng 已提交
1560

H
Hongze Cheng 已提交
1561 1562
//   int32_t start = cur->pos;
//   int32_t end = endPos;
H
Hongze Cheng 已提交
1563

H
Hongze Cheng 已提交
1564 1565 1566
//   if (!ascScan) {
//     TSWAP(start, end);
//   }
H
Hongze Cheng 已提交
1567

H
Hongze Cheng 已提交
1568 1569
//   assert(pTsdbReadHandle->outputCapacity >= (end - start + 1));
//   int32_t numOfRows = doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, 0, start, end);
H
Hongze Cheng 已提交
1570

H
Hongze Cheng 已提交
1571 1572 1573 1574 1575
//   // the time window should always be ascending order: skey <= ekey
//   cur->win = (STimeWindow){.skey = tsArray[start], .ekey = tsArray[end]};
//   cur->mixBlock = (numOfRows != pBlockInfo->rows);
//   cur->lastKey = tsArray[endPos] + step;
//   cur->blockCompleted = (ascScan ? (endPos == pBlockInfo->rows - 1) : (endPos == 0));
H
Hongze Cheng 已提交
1576

H
Hongze Cheng 已提交
1577 1578 1579 1580
//   // The value of pos may be -1 or pBlockInfo->rows, and it is invalid in both cases.
//   int32_t pos = endPos + step;
//   updateInfoAfterMerge(pTsdbReadHandle, pCheckInfo, numOfRows, pos);
//   doCheckGeneratedBlockRange(pTsdbReadHandle);
H
Hongze Cheng 已提交
1581

H
Hongze Cheng 已提交
1582 1583 1584 1585
//   tsdbDebug("%p uid:%" PRIu64 ", data block created, mixblock:%d, brange:%" PRIu64 "-%" PRIu64 " rows:%d, %s",
//             pTsdbReadHandle, pCheckInfo->tableId, cur->mixBlock, cur->win.skey, cur->win.ekey, cur->rows,
//             pTsdbReadHandle->idStr);
// }
H
Hongze Cheng 已提交
1586

H
Hongze Cheng 已提交
1587 1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629
// int32_t getEndPosInDataBlock(STsdbReader* pTsdbReadHandle, SDataBlockInfo* pBlockInfo) {
//   // NOTE: reverse the order to find the end position in data block
//   int32_t endPos = -1;
//   bool    ascScan = ASCENDING_TRAVERSE(pTsdbReadHandle->order);
//   int32_t order = ascScan ? TSDB_ORDER_DESC : TSDB_ORDER_ASC;

//   SQueryFilePos* cur = &pTsdbReadHandle->cur;
//   SDataCols*     pCols = pTsdbReadHandle->rhelper.pDCols[0];

//   if (pTsdbReadHandle->outputCapacity >= pBlockInfo->rows) {
//     if (ascScan && pTsdbReadHandle->window.ekey >= pBlockInfo->window.ekey) {
//       endPos = pBlockInfo->rows - 1;
//       cur->mixBlock = (cur->pos != 0);
//     } else if ((!ascScan) && pTsdbReadHandle->window.ekey <= pBlockInfo->window.skey) {
//       endPos = 0;
//       cur->mixBlock = (cur->pos != pBlockInfo->rows - 1);
//     } else {
//       assert(pCols->numOfRows > 0);
//       endPos = doBinarySearchKey(pCols->cols[0].pData, pCols->numOfRows, pTsdbReadHandle->window.ekey, order);
//       cur->mixBlock = true;
//     }
//   } else {
//     if (ascScan && pTsdbReadHandle->window.ekey >= pBlockInfo->window.ekey) {
//       endPos = TMIN(cur->pos + pTsdbReadHandle->outputCapacity - 1, pBlockInfo->rows - 1);
//     } else if ((!ascScan) && pTsdbReadHandle->window.ekey <= pBlockInfo->window.skey) {
//       endPos = TMAX(cur->pos - pTsdbReadHandle->outputCapacity + 1, 0);
//     } else {
//       ASSERT(pCols->numOfRows > 0);
//       endPos = doBinarySearchKey(pCols->cols[0].pData, pCols->numOfRows, pTsdbReadHandle->window.ekey, order);

//       // current data is more than the capacity
//       int32_t size = abs(cur->pos - endPos) + 1;
//       if (size > pTsdbReadHandle->outputCapacity) {
//         int32_t delta = size - pTsdbReadHandle->outputCapacity;
//         if (ascScan) {
//           endPos -= delta;
//         } else {
//           endPos += delta;
//         }
//       }
//     }
//     cur->mixBlock = true;
//   }
H
Hongze Cheng 已提交
1630

H
Hongze Cheng 已提交
1631 1632
//   return endPos;
// }
H
Hongze Cheng 已提交
1633

H
Hongze Cheng 已提交
1634 1635
// // only return the qualified data to client in terms of query time window, data rows in the same block but do not
// // be included in the query time window will be discarded
H
Haojun Liao 已提交
1636
// static void doMergeTwoLevelData(STsdbReader* pTsdbReadHandle, STableBlockScanInfo* pCheckInfo, SBlock* pBlock) {
H
Hongze Cheng 已提交
1637 1638 1639 1640 1641 1642 1643 1644 1645 1646 1647 1648 1649 1650 1651 1652 1653 1654 1655 1656 1657 1658 1659 1660 1661 1662 1663 1664 1665 1666 1667 1668 1669 1670 1671 1672 1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 1686 1687 1688 1689 1690 1691 1692 1693 1694 1695 1696 1697 1698 1699 1700 1701 1702 1703 1704 1705 1706 1707 1708 1709 1710 1711 1712 1713 1714 1715 1716 1717 1718 1719 1720 1721 1722 1723 1724 1725 1726 1727 1728 1729 1730 1731 1732 1733 1734 1735 1736 1737 1738 1739 1740 1741 1742 1743 1744 1745 1746 1747 1748 1749 1750 1751 1752 1753 1754 1755 1756 1757 1758 1759 1760 1761 1762 1763 1764 1765 1766 1767 1768 1769 1770 1771 1772 1773 1774 1775 1776 1777 1778 1779 1780 1781 1782 1783 1784 1785 1786 1787 1788 1789 1790 1791 1792 1793 1794 1795 1796 1797 1798 1799 1800 1801 1802 1803 1804 1805 1806 1807 1808 1809 1810 1811 1812 1813 1814 1815 1816 1817 1818 1819 1820 1821 1822 1823 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837
//   SQueryFilePos* cur = &pTsdbReadHandle->cur;
//   SDataBlockInfo blockInfo = GET_FILE_DATA_BLOCK_INFO(pCheckInfo, pBlock);
//   STsdbCfg*      pCfg = REPO_CFG(pTsdbReadHandle->pTsdb);

//   initTableMemIterator(pTsdbReadHandle, pCheckInfo);

//   SDataCols* pCols = pTsdbReadHandle->rhelper.pDCols[0];
//   assert(pCols->cols[0].type == TSDB_DATA_TYPE_TIMESTAMP && pCols->cols[0].colId == PRIMARYKEY_TIMESTAMP_COL_ID &&
//          cur->pos >= 0 && cur->pos < pBlock->numOfRows);
//   // Even Multi-Version supported, the records with duplicated TSKEY would be merged inside of tsdbLoadData
//   interface. TSKEY* tsArray = pCols->cols[0].pData; assert(pCols->numOfRows == pBlock->numOfRows && tsArray[0] ==
//   pBlock->minKey.ts &&
//          tsArray[pBlock->numOfRows - 1] == pBlock->maxKey.ts);

//   bool    ascScan = ASCENDING_TRAVERSE(pTsdbReadHandle->order);
//   int32_t step = ascScan ? 1 : -1;

//   // for search the endPos, so the order needs to reverse
//   int32_t order = ascScan ? TSDB_ORDER_DESC : TSDB_ORDER_ASC;

//   int32_t numOfCols = (int32_t)(QH_GET_NUM_OF_COLS(pTsdbReadHandle));
//   int32_t endPos = getEndPosInDataBlock(pTsdbReadHandle, &blockInfo);

//   STimeWindow* pWin = &blockInfo.window;
//   tsdbDebug("%p uid:%" PRIu64 " start merge data block, file block range:%" PRIu64 "-%" PRIu64
//             " rows:%d, start:%d, end:%d, %s",
//             pTsdbReadHandle, pCheckInfo->tableId, pWin->skey, pWin->ekey, blockInfo.rows, cur->pos, endPos,
//             pTsdbReadHandle->idStr);

//   // compared with the data from in-memory buffer, to generate the correct timestamp array list
//   int32_t numOfRows = 0;
//   int32_t curRow = 0;

//   int16_t   rv1 = -1;
//   int16_t   rv2 = -1;
//   STSchema* pSchema1 = NULL;
//   STSchema* pSchema2 = NULL;

//   int32_t pos = cur->pos;
//   cur->win = TSWINDOW_INITIALIZER;
//   bool adjustPos = false;

//   // no data in buffer, load data from file directly
//   if (pCheckInfo->iiter == NULL && pCheckInfo->iter == NULL) {
//     copyAllRemainRowsFromFileBlock(pTsdbReadHandle, pCheckInfo, &blockInfo, endPos);
//     return;
//   } else if (pCheckInfo->iter != NULL || pCheckInfo->iiter != NULL) {
//     SSkipListNode* node = NULL;
//     TSKEY          lastKeyAppend = TSKEY_INITIAL_VAL;

//     do {
//       STSRow* row2 = NULL;
//       STSRow* row1 = getSRowInTableMem(pCheckInfo, pTsdbReadHandle->order, pCfg->update, &row2, TD_VER_MAX);
//       if (row1 == NULL) {
//         break;
//       }

//       TSKEY key = TD_ROW_KEY(row1);
//       if ((key > pTsdbReadHandle->window.ekey && ascScan) || (key < pTsdbReadHandle->window.ekey && !ascScan)) {
//         break;
//       }

//       if (adjustPos) {
//         if (key == lastKeyAppend) {
//           pos -= step;
//         }
//         adjustPos = false;
//       }

//       if (((pos > endPos || tsArray[pos] > pTsdbReadHandle->window.ekey) && ascScan) ||
//           ((pos < endPos || tsArray[pos] < pTsdbReadHandle->window.ekey) && !ascScan)) {
//         break;
//       }

//       if ((key < tsArray[pos] && ascScan) || (key > tsArray[pos] && !ascScan)) {
//         if (rv1 != TD_ROW_SVER(row1)) {
//           //          pSchema1 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row1));
//           rv1 = TD_ROW_SVER(row1);
//         }
//         if (row2 && rv2 != TD_ROW_SVER(row2)) {
//           //          pSchema2 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row2));
//           rv2 = TD_ROW_SVER(row2);
//         }

//         numOfRows +=
//             mergeTwoRowFromMem(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, &curRow, row1, row2, numOfCols,
//                                pCheckInfo->tableId, pSchema1, pSchema2, pCfg->update, &lastKeyAppend);
//         if (cur->win.skey == TSKEY_INITIAL_VAL) {
//           cur->win.skey = key;
//         }

//         cur->win.ekey = key;
//         cur->lastKey = key + step;
//         cur->mixBlock = true;
//         moveToNextRowInMem(pCheckInfo);
//       } else if (key == tsArray[pos]) {  // data in buffer has the same timestamp of data in file block, ignore it
//         if (TD_SUPPORT_UPDATE(pCfg->update)) {
//           if (lastKeyAppend != key) {
//             if (lastKeyAppend != TSKEY_INITIAL_VAL) {
//               ++curRow;
//             }
//             lastKeyAppend = key;
//           }
//           // load data from file firstly
//           numOfRows = doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, curRow, pos, pos);

//           if (rv1 != TD_ROW_SVER(row1)) {
//             rv1 = TD_ROW_SVER(row1);
//           }
//           if (row2 && rv2 != TD_ROW_SVER(row2)) {
//             rv2 = TD_ROW_SVER(row2);
//           }

//           // still assign data into current row
//           numOfRows +=
//               mergeTwoRowFromMem(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, &curRow, row1, row2, numOfCols,
//                                  pCheckInfo->tableId, pSchema1, pSchema2, pCfg->update, &lastKeyAppend);

//           if (cur->win.skey == TSKEY_INITIAL_VAL) {
//             cur->win.skey = key;
//           }

//           cur->win.ekey = key;
//           cur->lastKey = key + step;
//           cur->mixBlock = true;

//           moveToNextRowInMem(pCheckInfo);

//           pos += step;
//           adjustPos = true;
//         } else {
//           // discard the memory record
//           moveToNextRowInMem(pCheckInfo);
//         }
//       } else if ((key > tsArray[pos] && ascScan) || (key < tsArray[pos] && !ascScan)) {
//         if (cur->win.skey == TSKEY_INITIAL_VAL) {
//           cur->win.skey = tsArray[pos];
//         }

//         int32_t end = doBinarySearchKey(pCols->cols[0].pData, pCols->numOfRows, key, order);
//         assert(end != -1);

//         if (tsArray[end] == key) {  // the value of key in cache equals to the end timestamp value, ignore it
// #if 0
//           if (pCfg->update == TD_ROW_DISCARD_UPDATE) {
//             moveToNextRowInMem(pCheckInfo);
//           } else {
//             end -= step;
//           }
// #endif
//           if (!TD_SUPPORT_UPDATE(pCfg->update)) {
//             moveToNextRowInMem(pCheckInfo);
//           } else {
//             end -= step;
//           }
//         }

//         int32_t qstart = 0, qend = 0;
//         getQualifiedRowsPos(pTsdbReadHandle, pos, end, numOfRows, &qstart, &qend);

//         if ((lastKeyAppend != TSKEY_INITIAL_VAL) && (lastKeyAppend != (ascScan ? tsArray[qstart] : tsArray[qend]))) {
//           ++curRow;
//         }

//         numOfRows = doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, curRow, qstart, qend);
//         pos += (qend - qstart + 1) * step;
//         if (numOfRows > 0) {
//           curRow = numOfRows - 1;
//         }

//         cur->win.ekey = ascScan ? tsArray[qend] : tsArray[qstart];
//         cur->lastKey = cur->win.ekey + step;
//         lastKeyAppend = cur->win.ekey;
//       }
//     } while (numOfRows < pTsdbReadHandle->outputCapacity);

//     if (numOfRows < pTsdbReadHandle->outputCapacity) {
//       /**
//        * if cache is empty, load remain file block data. In contrast, if there are remain data in cache, do NOT
//        * copy them all to result buffer, since it may be overlapped with file data block.
//        */
//       if (node == NULL || ((TD_ROW_KEY((STSRow*)SL_GET_NODE_DATA(node)) > pTsdbReadHandle->window.ekey) && ascScan)
//       ||
//           ((TD_ROW_KEY((STSRow*)SL_GET_NODE_DATA(node)) < pTsdbReadHandle->window.ekey) && !ascScan)) {
//         // no data in cache or data in cache is greater than the ekey of time window, load data from file block
//         if (cur->win.skey == TSKEY_INITIAL_VAL) {
//           cur->win.skey = tsArray[pos];
//         }

//         int32_t start = -1, end = -1;
//         getQualifiedRowsPos(pTsdbReadHandle, pos, endPos, numOfRows, &start, &end);

//         numOfRows = doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, numOfRows, start, end);
//         pos += (end - start + 1) * step;

//         cur->win.ekey = ascScan ? tsArray[end] : tsArray[start];
//         cur->lastKey = cur->win.ekey + step;
//         cur->mixBlock = true;
//       }
//     }
//   }
H
Hongze Cheng 已提交
1838

H
Hongze Cheng 已提交
1839 1840
//   cur->blockCompleted = (((pos > endPos || cur->lastKey > pTsdbReadHandle->window.ekey) && ascScan) ||
//                          ((pos < endPos || cur->lastKey < pTsdbReadHandle->window.ekey) && !ascScan));
H
Hongze Cheng 已提交
1841

H
Hongze Cheng 已提交
1842 1843 1844
//   if (!ascScan) {
//     TSWAP(cur->win.skey, cur->win.ekey);
//   }
H
Hongze Cheng 已提交
1845

H
Hongze Cheng 已提交
1846 1847
//   updateInfoAfterMerge(pTsdbReadHandle, pCheckInfo, numOfRows, pos);
//   doCheckGeneratedBlockRange(pTsdbReadHandle);
H
Hongze Cheng 已提交
1848

H
Hongze Cheng 已提交
1849 1850 1851 1852
//   tsdbDebug("%p uid:%" PRIu64 ", data block created, mixblock:%d, brange:%" PRIu64 "-%" PRIu64 " rows:%d, %s",
//             pTsdbReadHandle, pCheckInfo->tableId, cur->mixBlock, cur->win.skey, cur->win.ekey, cur->rows,
//             pTsdbReadHandle->idStr);
// }
H
Hongze Cheng 已提交
1853

H
Hongze Cheng 已提交
1854 1855 1856 1857 1858 1859 1860 1861 1862 1863 1864 1865 1866 1867 1868 1869 1870 1871 1872 1873 1874 1875 1876 1877 1878 1879 1880 1881 1882 1883 1884 1885 1886 1887 1888 1889 1890 1891 1892 1893 1894 1895 1896 1897 1898 1899 1900 1901 1902 1903 1904 1905 1906 1907 1908 1909
// int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order) {
//   int    firstPos, lastPos, midPos = -1;
//   int    numOfRows;
//   TSKEY* keyList;

//   if (num <= 0) return -1;

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

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

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

//       if (key < keyList[midPos]) {
//         lastPos = midPos - 1;
//       } else if (key > keyList[midPos]) {
//         firstPos = 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 >> 1) + firstPos;

//       if (key < keyList[midPos]) {
//         lastPos = midPos - 1;
//       } else if (key > keyList[midPos]) {
//         firstPos = midPos + 1;
//       } else {
//         break;
//       }
//     }
//   }
H
Hongze Cheng 已提交
1910

H
Hongze Cheng 已提交
1911 1912
//   return midPos;
// }
H
Hongze Cheng 已提交
1913

H
Haojun Liao 已提交
1914 1915 1916
static void cleanupBlockOrderSupporter(SBlockOrderSupporter* pSup) {
  taosMemoryFreeClear(pSup->numOfBlocksPerTable);
  taosMemoryFreeClear(pSup->indexPerTable);
H
Hongze Cheng 已提交
1917

H
Haojun Liao 已提交
1918 1919 1920 1921
  for (int32_t i = 0; i < pSup->numOfTables; ++i) {
    SBlockOrderWrapper* pBlockInfo = pSup->pDataBlockInfo[i];
    taosMemoryFreeClear(pBlockInfo);
  }
H
Hongze Cheng 已提交
1922

H
Haojun Liao 已提交
1923 1924
  taosMemoryFreeClear(pSup->pDataBlockInfo);
}
H
Hongze Cheng 已提交
1925

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

H
Haojun Liao 已提交
1929 1930 1931
  pSup->numOfBlocksPerTable = taosMemoryCalloc(1, sizeof(int32_t) * numOfTables);
  pSup->indexPerTable       = taosMemoryCalloc(1, sizeof(int32_t) * numOfTables);
  pSup->pDataBlockInfo      = taosMemoryCalloc(1, POINTER_BYTES * numOfTables);
H
Hongze Cheng 已提交
1932

H
Haojun Liao 已提交
1933 1934 1935 1936
  if (pSup->numOfBlocksPerTable == NULL || pSup->indexPerTable == NULL || pSup->pDataBlockInfo == NULL) {
    cleanupBlockOrderSupporter(pSup);
    return TSDB_CODE_OUT_OF_MEMORY;
  }
H
Hongze Cheng 已提交
1937

H
Haojun Liao 已提交
1938 1939
  return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
1940

H
Haojun Liao 已提交
1941 1942 1943
static int32_t fileDataBlockOrderCompar(const void* pLeft, const void* pRight, void* param) {
  int32_t leftIndex  = *(int32_t*)pLeft;
  int32_t rightIndex = *(int32_t*)pRight;
H
Hongze Cheng 已提交
1944

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

H
Haojun Liao 已提交
1947 1948
  int32_t leftTableBlockIndex = pSupporter->indexPerTable[leftIndex];
  int32_t rightTableBlockIndex = pSupporter->indexPerTable[rightIndex];
H
Hongze Cheng 已提交
1949

H
Haojun Liao 已提交
1950 1951 1952 1953 1954 1955 1956
  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 已提交
1957

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

H
Haojun Liao 已提交
1961 1962
  return pLeftBlock->pBlock->aSubBlock[0].offset > pRightBlock->pBlock->aSubBlock[0].offset ? 1 : -1;
}
H
Hongze Cheng 已提交
1963

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

1967
  pBlockIter->numOfBlocks = numOfBlocks;
H
Haojun Liao 已提交
1968 1969 1970 1971 1972 1973 1974 1975 1976 1977 1978 1979
 // access data blocks according to the offset of each block in asc/desc order.
 int32_t numOfTables = (int32_t)taosHashGetSize(pReader->status.pTableMap);

 SBlockOrderSupporter sup = {0};
 int32_t code = initBlockOrderSupporter(&sup, numOfTables);
 if (code != TSDB_CODE_SUCCESS) {
   return code;
 }

 int32_t cnt = 0;
 void* ptr = NULL;
 while(1) {
1980
   ptr = taosHashIterate(pReader->status.pTableMap, ptr);
H
Haojun Liao 已提交
1981 1982 1983 1984 1985 1986 1987 1988 1989 1990 1991 1992 1993 1994 1995 1996 1997 1998 1999 2000 2001 2002 2003 2004 2005 2006 2007 2008 2009 2010 2011 2012 2013 2014 2015 2016 2017 2018 2019 2020 2021
   if (ptr == NULL) {
     break;
   }

   STableBlockScanInfo* pTableScanInfo = (STableBlockScanInfo*)ptr;
   if (pTableScanInfo->pBlockList == NULL || taosArrayGetSize(pTableScanInfo->pBlockList) == 0) {
     continue;
   }

   size_t num = taosArrayGetSize(pTableScanInfo->pBlockList);
   sup.numOfBlocksPerTable[sup.numOfTables] = num;

   char* buf = taosMemoryMalloc(sizeof(SBlockOrderWrapper) * num);
   if (buf == NULL) {
     cleanupBlockOrderSupporter(&sup);
     return TSDB_CODE_TDB_OUT_OF_MEMORY;
   }

   sup.pDataBlockInfo[sup.numOfTables] = (SBlockOrderWrapper*)buf;
   for (int32_t k = 0; k < num; ++k) {
     SBlockOrderWrapper wrapper = {0};
     wrapper.pBlock = (SBlock*)taosArrayGet(pTableScanInfo->pBlockList, k);
     wrapper.uid = pTableScanInfo->uid;

     sup.pDataBlockInfo[sup.numOfTables][k] = wrapper;
     cnt++;
   }

   sup.numOfTables += 1;
 }

 ASSERT(numOfBlocks == cnt);

 // since there is only one table qualified, blocks are not sorted
 if (sup.numOfTables == 1) {
   for(int32_t i = 0; i < numOfBlocks; ++i) {
     SFileDataBlockInfo blockInfo = {.uid = sup.pDataBlockInfo[0][i].uid, .tbBlockIdx = i};
     taosArrayPush(pBlockIter->blockList, &blockInfo);
   }
   tsdbDebug("%p create blocks info struct completed for one table, %d blocks not sorted %s",
       pReader, cnt, pReader->idStr);
2022

2023
   pBlockIter->index = asc ? 0 : (numOfBlocks - 1);
H
Haojun Liao 已提交
2024 2025 2026 2027
   return TSDB_CODE_SUCCESS;
 }

 tsdbDebug("%p create data blocks info struct completed, %d blocks in %d tables %s", pReader, cnt, sup.numOfTables, pReader->idStr);
2028 2029
 // the pTableQueryInfo[j]->numOfBlocks may be 0
 assert(cnt <= numOfBlocks && sup.numOfTables <= numOfTables);
H
Haojun Liao 已提交
2030 2031

 SMultiwayMergeTreeInfo* pTree = NULL;
H
Haojun Liao 已提交
2032

H
Haojun Liao 已提交
2033 2034 2035 2036 2037 2038 2039 2040 2041 2042 2043 2044 2045 2046 2047 2048 2049 2050 2051 2052 2053 2054 2055 2056 2057 2058 2059
 uint8_t ret = tMergeTreeCreate(&pTree, sup.numOfTables, &sup, fileDataBlockOrderCompar);
 if (ret != TSDB_CODE_SUCCESS) {
   cleanupBlockOrderSupporter(&sup);
   return TSDB_CODE_TDB_OUT_OF_MEMORY;
 }

 int32_t numOfTotal = 0;
 while (numOfTotal < cnt) {
   int32_t pos = tMergeTreeGetChosenIndex(pTree);
   int32_t index = sup.indexPerTable[pos]++;

   SFileDataBlockInfo blockInfo = {.uid = sup.pDataBlockInfo[pos][index].uid, .tbBlockIdx = index};
   taosArrayPush(pBlockIter->blockList, &blockInfo);

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

   numOfTotal += 1;
   tMergeTreeAdjust(pTree, tMergeTreeGetAdjustIndex(pTree));
 }

 tsdbDebug("%p %d data blocks sort completed, %s", pReader, cnt, pReader->idStr);
 cleanupBlockOrderSupporter(&sup);
 taosMemoryFree(pTree);

2060
 pBlockIter->index = asc ? 0 : (numOfBlocks - 1);
H
Haojun Liao 已提交
2061 2062
 return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
2063

H
Haojun Liao 已提交
2064
static bool blockIteratorNext(SDataBlockIter* pBlockIter) {
2065 2066 2067 2068
  bool asc = ASCENDING_TRAVERSE(pBlockIter->order);

  int32_t step = asc? 1:-1;
  if ((pBlockIter->index >= pBlockIter->numOfBlocks - 1 && asc) || (pBlockIter->index <= 0 && (!asc))) {
2069 2070 2071
    return false;
  }

2072
  pBlockIter->index += step;
2073 2074 2075
  return true;
}

H
Hongze Cheng 已提交
2076 2077 2078 2079 2080 2081 2082 2083 2084 2085 2086 2087 2088 2089 2090 2091 2092 2093 2094 2095 2096 2097 2098 2099 2100 2101 2102 2103 2104 2105 2106 2107 2108 2109 2110 2111 2112 2113 2114 2115 2116 2117 2118 2119 2120 2121 2122 2123 2124 2125 2126 2127 2128 2129 2130 2131 2132 2133 2134 2135 2136 2137 2138 2139 2140 2141 2142 2143 2144
// static int32_t getFirstFileDataBlock(STsdbReader* pTsdbReadHandle, bool* exists) {
//   pTsdbReadHandle->numOfBlocks = 0;
//   SQueryFilePos* cur = &pTsdbReadHandle->cur;

//   int32_t code = TSDB_CODE_SUCCESS;

//   int32_t numOfBlocks = 0;
//   int32_t numOfTables = (int32_t)taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);

//   STsdbKeepCfg* pCfg = REPO_KEEP_CFG(pTsdbReadHandle->pTsdb);
//   STimeWindow   win = TSWINDOW_INITIALIZER;

//   while (true) {
//     tsdbRLockFS(REPO_FS(pTsdbReadHandle->pTsdb));

//     if ((pTsdbReadHandle->pFileGroup = tsdbFSIterNext(&pTsdbReadHandle->fileIter)) == NULL) {
//       tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
//       break;
//     }

//     tsdbGetFidKeyRange(pCfg->days, pCfg->precision, pTsdbReadHandle->pFileGroup->fid, &win.skey, &win.ekey);

//     // current file are not overlapped with query time window, ignore remain files
//     if ((ASCENDING_TRAVERSE(pTsdbReadHandle->order) && win.skey > pTsdbReadHandle->window.ekey) ||
//         (!ASCENDING_TRAVERSE(pTsdbReadHandle->order) && win.ekey < pTsdbReadHandle->window.ekey)) {
//       tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
//       tsdbDebug("%p remain files are not qualified for qrange:%" PRId64 "-%" PRId64 ", ignore, %s", pTsdbReadHandle,
//                 pTsdbReadHandle->window.skey, pTsdbReadHandle->window.ekey, pTsdbReadHandle->idStr);
//       pTsdbReadHandle->pFileGroup = NULL;
//       assert(pTsdbReadHandle->numOfBlocks == 0);
//       break;
//     }

//     if (tsdbSetAndOpenReadFSet(&pTsdbReadHandle->rhelper, pTsdbReadHandle->pFileGroup) < 0) {
//       tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
//       code = terrno;
//       break;
//     }

//     tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));

//     if (tsdbLoadBlockIdx(&pTsdbReadHandle->rhelper) < 0) {
//       code = terrno;
//       break;
//     }

//     if ((code = getFileCompInfo(pTsdbReadHandle, &numOfBlocks)) != TSDB_CODE_SUCCESS) {
//       break;
//     }

//     tsdbDebug("%p %d blocks found in file for %d table(s), fid:%d, %s", pTsdbReadHandle, numOfBlocks, numOfTables,
//               pTsdbReadHandle->pFileGroup->fid, pTsdbReadHandle->idStr);

//     assert(numOfBlocks >= 0);
//     if (numOfBlocks == 0) {
//       continue;
//     }

//     // todo return error code to query engine
//     if ((code = createDataBlocksInfo(pTsdbReadHandle, numOfBlocks, &pTsdbReadHandle->numOfBlocks)) !=
//         TSDB_CODE_SUCCESS) {
//       break;
//     }

//     assert(numOfBlocks >= pTsdbReadHandle->numOfBlocks);
//     if (pTsdbReadHandle->numOfBlocks > 0) {
//       break;
//     }
//   }
H
Hongze Cheng 已提交
2145

H
Hongze Cheng 已提交
2146 2147 2148 2149 2150
//   // no data in file anymore
//   if (pTsdbReadHandle->numOfBlocks <= 0 || code != TSDB_CODE_SUCCESS) {
//     if (code == TSDB_CODE_SUCCESS) {
//       assert(pTsdbReadHandle->pFileGroup == NULL);
//     }
H
Hongze Cheng 已提交
2151

H
Hongze Cheng 已提交
2152 2153 2154 2155
//     cur->fid = INT32_MIN;  // denote that there are no data in file anymore
//     *exists = false;
//     return code;
//   }
H
Hongze Cheng 已提交
2156

H
Hongze Cheng 已提交
2157 2158 2159
//   assert(pTsdbReadHandle->pFileGroup != NULL && pTsdbReadHandle->numOfBlocks > 0);
//   cur->slot = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 0 : pTsdbReadHandle->numOfBlocks - 1;
//   cur->fid = pTsdbReadHandle->pFileGroup->fid;
H
Hongze Cheng 已提交
2160

H
Haojun Liao 已提交
2161
//   SFileBlockInfo* pBlockInfo = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
H
Hongze Cheng 已提交
2162 2163
//   return getDataBlock(pTsdbReadHandle, pBlockInfo, exists);
// }
H
Hongze Cheng 已提交
2164

H
Hongze Cheng 已提交
2165 2166 2167
// static int32_t getBucketIndex(int32_t startRow, int32_t bucketRange, int32_t numOfRows) {
//   return (numOfRows - startRow) / bucketRange;
// }
H
Hongze Cheng 已提交
2168

2169 2170 2171 2172 2173 2174 2175 2176
/**
 * This is an two rectangles overlap cases.
 */
static int32_t dataBlockPartialRequired(STimeWindow* pWindow, SVersionRange* pVerRange, SBlock* pBlock) {
  return (pWindow->ekey < pBlock->maxKey.ts && pWindow->ekey >= pBlock->minKey.ts) ||
         (pWindow->skey > pBlock->minKey.ts && pWindow->skey <= pBlock->maxKey.ts) ||
         (pVerRange->minVer > pBlock->minVersion && pVerRange->minVer <= pBlock->maxVersion) ||
         (pVerRange->maxVer < pBlock->maxVersion && pVerRange->maxVer >= pBlock->minVersion);
H
Haojun Liao 已提交
2177
}
H
Hongze Cheng 已提交
2178

H
Haojun Liao 已提交
2179 2180 2181 2182
static SFileDataBlockInfo* getCurrentBlockInfo(SDataBlockIter* pBlockIter) {
  SFileDataBlockInfo* pFBlockInfo = taosArrayGet(pBlockIter->blockList, pBlockIter->index);
  return pFBlockInfo;
}
H
Hongze Cheng 已提交
2183

2184 2185 2186 2187 2188 2189 2190 2191
static bool overlapWithNeighborBlock(SFileDataBlockInfo *pFBlockInfo, SBlock* pBlock, STableBlockScanInfo* pTableBlockScanInfo) {
  // it is the last block in current file, no chance to overlap with neighbor blocks.
  if(pFBlockInfo->tbBlockIdx == taosArrayGetSize(pTableBlockScanInfo->pBlockList) - 1) { // last block in current file,
    return false;
  }

  SBlock* pNext = taosArrayGet(pTableBlockScanInfo->pBlockList, pFBlockInfo->tbBlockIdx + 1);
  return (pNext->minKey.ts == pBlock->maxKey.ts);
H
Haojun Liao 已提交
2192
}
H
Hongze Cheng 已提交
2193

2194
static bool bufferDataInFileBlockGap(int32_t order, TSDBKEY key, SBlock* pBlock) {
H
Haojun Liao 已提交
2195
  bool ascScan = ASCENDING_TRAVERSE(order);
H
Hongze Cheng 已提交
2196

2197 2198
  return (ascScan && (key.ts != TSKEY_INITIAL_VAL && key.ts <= pBlock->minKey.ts)) ||
       (!ascScan && (key.ts != TSKEY_INITIAL_VAL && key.ts >= pBlock->maxKey.ts));
H
Haojun Liao 已提交
2199
}
H
Hongze Cheng 已提交
2200

H
Haojun Liao 已提交
2201
static bool keyOverlapFileBlock(TSDBKEY key, SBlock* pBlock, SVersionRange* pVerRange) {
2202
  return (key.ts >= pBlock->minKey.ts && key.ts <= pBlock->maxKey.ts) && (pBlock->maxVersion >= pVerRange->minVer) && (pBlock->minVersion <= pVerRange->maxVer);
H
Haojun Liao 已提交
2203 2204
}

2205

H
Haojun Liao 已提交
2206
static bool fileBlockShouldLoad(STsdbReader* pReader, SFileDataBlockInfo *pFBlock, SBlock* pBlock, STableBlockScanInfo *pScanInfo, TSDBKEY key) {
2207 2208 2209 2210
  return (dataBlockPartialRequired(&pReader->window, &pReader->verRange, pBlock) ||
          overlapWithNeighborBlock(pFBlock, pBlock, pScanInfo) ||
          keyOverlapFileBlock(key, pBlock, &pReader->verRange) ||
          (pBlock->nRow > pReader->capacity));
H
Haojun Liao 已提交
2211 2212
}

2213 2214 2215 2216
static int32_t buildInmemDataBlock(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo, TSDBKEY *key) {
  if (!(pBlockScanInfo->imemHasVal || pBlockScanInfo->memHasVal)) {
    return TSDB_CODE_SUCCESS;
  }
H
Haojun Liao 已提交
2217

2218 2219 2220
  SSDataBlock* pBlock = pReader->pResBlock;

  int64_t st = taosGetTimestampUs();
2221
  int32_t code = buildInmemDataBlockImpl(pBlockScanInfo, *key, pReader->capacity, pReader);
H
Haojun Liao 已提交
2222

2223 2224 2225 2226 2227 2228 2229
  int64_t elapsedTime = taosGetTimestampUs() - st;

  tsdbDebug("%p build data block from cache completed, elapsed time:%" PRId64 " us, numOfRows:%d, numOfCols:%d, %s",
            pReader, elapsedTime, pBlock->info.rows, (int32_t)blockDataGetNumOfCols(pBlock), pReader->idStr);

  pBlock->info.uid = pBlockScanInfo->uid;
  setComposedBlockFlag(pReader, true);
H
Haojun Liao 已提交
2230 2231 2232
  return code;
}

2233
static int32_t buildComposedDataBlockImpl(STsdbReader* pReader, SFileDataBlockInfo* pFBlock, SBlock* pBlock, STableBlockScanInfo* pBlockScanInfo) {
H
Haojun Liao 已提交
2234
  SFileBlockDumpInfo *pDumpInfo = &pReader->status.fBlockDumpInfo;
2235
  SBlockData* pBlockData = &pReader->status.fileBlockData;
H
Haojun Liao 已提交
2236 2237

  SRowMerger merge = {0};
2238 2239
  STSRow*    pTSRow = NULL;
  TSKEY      mergeTs = TSKEY_INITIAL_VAL;
H
Haojun Liao 已提交
2240

2241 2242 2243
  int64_t key = pBlockData->aTSKEY[pDumpInfo->rowIndex];
  TSDBROW* pRow = getValidRow(pBlockScanInfo->iter, &pBlockScanInfo->memHasVal, pReader);
  TSDBROW* piRow = getValidRow(pBlockScanInfo->iiter, &pBlockScanInfo->imemHasVal, pReader);
H
Haojun Liao 已提交
2244 2245 2246 2247 2248

  if (pBlockScanInfo->memHasVal && pBlockScanInfo->imemHasVal) {
    TSDBKEY k = TSDBROW_KEY(pRow);
    TSDBKEY ik = TSDBROW_KEY(piRow);

2249 2250 2251 2252 2253 2254
    // [1&2] key <= [k.ts|ik.ts]
    if (key <= k.ts || key <= ik.ts) {
      TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
      tRowMergerInit(&merge, &fRow, pReader->pSchema);

      doLoadRowsOfIdenticalTsInFileBlock(pFBlock, pBlock, pBlockData, pBlockScanInfo, pReader, &merge);
H
Haojun Liao 已提交
2255

2256 2257 2258 2259 2260 2261 2262 2263 2264 2265 2266 2267 2268 2269 2270 2271 2272 2273 2274 2275 2276 2277 2278 2279 2280 2281 2282 2283 2284 2285 2286 2287 2288 2289 2290 2291 2292 2293 2294 2295 2296 2297 2298 2299 2300 2301 2302 2303 2304 2305 2306 2307 2308 2309 2310 2311 2312 2313
      if (ik.ts == mergeTs) {
        doLoadRowsOfIdenticalTs(pBlockScanInfo->iiter, &pBlockScanInfo->imemHasVal, ik.ts, &merge, pReader);
      }

      if (k.ts == mergeTs) {
        doLoadRowsOfIdenticalTs(pBlockScanInfo->iter, &pBlockScanInfo->memHasVal, k.ts, &merge, pReader);
      }

      tRowMergerGetRow(&merge, &pTSRow);
      doAppendOneRow(pReader->pResBlock, pReader, pTSRow);
    } else {
      // [3] ik.ts < key <= k.ts
      if (ik.ts < k.ts) {
        doLoadRowsOfIdenticalTs(pBlockScanInfo->iiter, &pBlockScanInfo->imemHasVal, ik.ts, &merge, pReader);
        tRowMergerGetRow(&merge, &pTSRow);
        doAppendOneRow(pReader->pResBlock, pReader, pTSRow);
        return TSDB_CODE_SUCCESS;
      }

      // [4] k.ts < key <= ik.ts
      if (k.ts < ik.ts) {
        doLoadRowsOfIdenticalTs(pBlockScanInfo->iter, &pBlockScanInfo->memHasVal, k.ts, &merge, pReader);
        tRowMergerGetRow(&merge, &pTSRow);
        doAppendOneRow(pReader->pResBlock, pReader, pTSRow);
        return TSDB_CODE_SUCCESS;
      }

      // [5] k.ts == ik.ts < key
      if (k.ts == ik.ts) {
        doLoadRowsOfIdenticalTs(pBlockScanInfo->iiter, &pBlockScanInfo->imemHasVal, ik.ts, &merge, pReader);
        tRowMergerGetRow(&merge, &pTSRow);

        if (k.ts == mergeTs) {
          doLoadRowsOfIdenticalTs(pBlockScanInfo->iter, &pBlockScanInfo->memHasVal, k.ts, &merge, pReader);
        }

        tRowMergerGetRow(&merge, &pTSRow);
        doAppendOneRow(pReader->pResBlock, pReader, pTSRow);
        return TSDB_CODE_SUCCESS;
      }

      // [6] k.ts < ik.ts < key
      if (k.ts < ik.ts) {
        doLoadRowsOfIdenticalTs(pBlockScanInfo->iter, &pBlockScanInfo->memHasVal, k.ts, &merge, pReader);
        tRowMergerGetRow(&merge, &pTSRow);
        doAppendOneRow(pReader->pResBlock, pReader, pTSRow);

        return TSDB_CODE_SUCCESS;
      }

      // [6] ik.ts < k.ts < key
      if (ik.ts < k.ts) {
        doLoadRowsOfIdenticalTs(pBlockScanInfo->iiter, &pBlockScanInfo->imemHasVal, ik.ts, &merge, pReader);
        tRowMergerGetRow(&merge, &pTSRow);
        doAppendOneRow(pReader->pResBlock, pReader, pTSRow);
        return TSDB_CODE_SUCCESS;
      }
    }
2314 2315 2316 2317 2318 2319
  } else {
    if (pBlockScanInfo->imemHasVal) {
      TSDBKEY ik = TSDBROW_KEY(piRow);
      if (key <= ik.ts) {
        TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
        tRowMergerInit(&merge, &fRow, pReader->pSchema);
2320

2321
        doLoadRowsOfIdenticalTsInFileBlock(pFBlock, pBlock, pBlockData, pBlockScanInfo, pReader, &merge);
2322

2323 2324 2325 2326 2327 2328 2329 2330 2331 2332
        if (ik.ts == mergeTs) {
          doLoadRowsOfIdenticalTs(pBlockScanInfo->iiter, &pBlockScanInfo->imemHasVal, ik.ts, &merge, pReader);
        }

        tRowMergerGetRow(&merge, &pTSRow);
        doAppendOneRow(pReader->pResBlock, pReader, pTSRow);
        return TSDB_CODE_SUCCESS;
      }

      if (ik.ts < key) {
2333
        doLoadRowsOfIdenticalTs(pBlockScanInfo->iiter, &pBlockScanInfo->imemHasVal, ik.ts, &merge, pReader);
2334 2335
        tRowMergerGetRow(&merge, &pTSRow);
        doAppendOneRow(pReader->pResBlock, pReader, pTSRow);
2336 2337 2338 2339 2340
      }

      return TSDB_CODE_SUCCESS;
    }

2341 2342 2343 2344 2345
    if (pBlockScanInfo->memHasVal) {  // pBlockScanInfo->memHasVal != NULL
      TSDBKEY k = TSDBROW_KEY(pRow);
      if (key <= k.ts) {
        TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
        tRowMergerInit(&merge, &fRow, pReader->pSchema);
2346

2347
        doLoadRowsOfIdenticalTsInFileBlock(pFBlock, pBlock, pBlockData, pBlockScanInfo, pReader, &merge);
2348

2349 2350 2351 2352 2353 2354 2355 2356 2357 2358
        if (k.ts == mergeTs) {
          doLoadRowsOfIdenticalTs(pBlockScanInfo->iter, &pBlockScanInfo->memHasVal, k.ts, &merge, pReader);
        }

        tRowMergerGetRow(&merge, &pTSRow);
        doAppendOneRow(pReader->pResBlock, pReader, pTSRow);
        return TSDB_CODE_SUCCESS;
      }

      if (k.ts < key) {
2359
        doLoadRowsOfIdenticalTs(pBlockScanInfo->iter, &pBlockScanInfo->memHasVal, k.ts, &merge, pReader);
2360 2361
        tRowMergerGetRow(&merge, &pTSRow);
        doAppendOneRow(pReader->pResBlock, pReader, pTSRow);
2362 2363 2364
      }

      return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
2365
    }
2366

2367 2368 2369 2370 2371 2372
    // imem & mem are all empty
    TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
    tRowMergerInit(&merge, &fRow, pReader->pSchema);
    doLoadRowsOfIdenticalTsInFileBlock(pFBlock, pBlock, pBlockData, pBlockScanInfo, pReader, &merge);
    tRowMergerGetRow(&merge, &pTSRow);
    doAppendOneRow(pReader->pResBlock, pReader, pTSRow);
2373 2374 2375 2376 2377 2378 2379 2380 2381 2382 2383 2384 2385 2386
  }

  return TSDB_CODE_SUCCESS;
}

static int32_t buildComposedDataBlock(STsdbReader* pReader, SFileDataBlockInfo* pFBlock, SBlock* pBlock, STableBlockScanInfo* pBlockScanInfo) {
  SSDataBlock* pResBlock = pReader->pResBlock;

  while(1) {
    buildComposedDataBlockImpl(pReader, pFBlock, pBlock, pBlockScanInfo);
    SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;

    SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(&pReader->status.blockIter);
    if (pBlockInfo->tbBlockIdx == pFBlock->tbBlockIdx) {  // still in the same file block now
2387 2388
      if (pDumpInfo->rowIndex >= pBlock->nRow || pDumpInfo->rowIndex < 0) {
        setBlockAllDumped(pDumpInfo, pBlock, pReader->order);
2389 2390 2391 2392 2393 2394 2395 2396 2397 2398 2399 2400 2401 2402 2403 2404 2405 2406 2407 2408 2409 2410 2411 2412 2413 2414
        break;
      }

      if (pResBlock->info.rows >= pReader->capacity) {
        break;
      }
    } else {  // todo traverse to next file due to time window overlap
      if (pResBlock->info.rows >= pReader->capacity) {
        ASSERT(0);
        return TSDB_CODE_SUCCESS;
      }
    }
  }

  pResBlock->info.uid = pBlockScanInfo->uid;
  setComposedBlockFlag(pReader, true);
  return TSDB_CODE_SUCCESS;
}

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

static int32_t initMemIterator(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader) {
  if (pBlockScanInfo->iterInit) {
    return TSDB_CODE_SUCCESS;
  }

2415
  int32_t code = TSDB_CODE_SUCCESS;
2416 2417 2418 2419 2420 2421 2422 2423 2424

  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));
2425 2426 2427 2428

  STbData* d = NULL;
  if (pReader->pTsdb->mem != NULL) {
    tsdbGetTbDataFromMemTable(pReader->pTsdb->mem, pReader->suid, pBlockScanInfo->uid, &d);
2429
    if (d != NULL) {
2430
      code = tsdbTbDataIterCreate(d, &startKey, backward, &pBlockScanInfo->iter);
2431 2432 2433
      if (code == TSDB_CODE_SUCCESS) {
        pBlockScanInfo->memHasVal = (tsdbTbDataIterGet(pBlockScanInfo->iter) != NULL);

H
Haojun Liao 已提交
2434
        tsdbDebug("%p uid:%" PRId64 ", check data in mem from skey:%" PRId64 ", order:%d, ts range in buf:%" PRId64
2435 2436
                  "-%" PRId64 " %s",
                  pReader, pBlockScanInfo->uid, startKey.ts, pReader->order, d->minKey, d->maxKey, pReader->idStr);
2437
      } else {
2438 2439
        tsdbError("%p uid:%" PRId64 ", failed to create iterator for imem, code:%s, %s", pReader, pBlockScanInfo->uid,
                  tstrerror(code), pReader->idStr);
2440
        return code;
2441 2442
      }
    }
H
Haojun Liao 已提交
2443
  } else {
2444
    tsdbDebug("%p uid:%" PRId64 ", no data in mem, %s", pReader, pBlockScanInfo->uid, pReader->idStr);
H
Haojun Liao 已提交
2445 2446
  }

2447 2448 2449
  STbData* di = NULL;
  if (pReader->pTsdb->imem != NULL) {
    tsdbGetTbDataFromMemTable(pReader->pTsdb->imem, pReader->suid, pBlockScanInfo->uid, &di);
2450
    if (di != NULL) {
2451
      code = tsdbTbDataIterCreate(di, &startKey, backward, &pBlockScanInfo->iiter);
2452 2453 2454
      if (code == TSDB_CODE_SUCCESS) {
        pBlockScanInfo->imemHasVal = (tsdbTbDataIterGet(pBlockScanInfo->iiter) != NULL);

H
Haojun Liao 已提交
2455
        tsdbDebug("%p uid:%" PRId64 ", check data in imem from skey:%" PRId64 ", order:%d, ts range in buf:%" PRId64
2456
                  "-%" PRId64 " %s",
2457
                  pReader, pBlockScanInfo->uid, startKey.ts, pReader->order, di->minKey, di->maxKey, pReader->idStr);
2458
      } else {
2459 2460
        tsdbError("%p uid:%" PRId64 ", failed to create iterator for mem, code:%s, %s", pReader, pBlockScanInfo->uid,
                  tstrerror(code), pReader->idStr);
2461
        return code;
2462 2463
      }
    }
H
Haojun Liao 已提交
2464 2465
  } else {
    tsdbDebug("%p uid:%" PRId64 ", no data in imem, %s", pReader, pBlockScanInfo->uid, pReader->idStr);
2466 2467 2468
  }

  pBlockScanInfo->iterInit = true;
H
Haojun Liao 已提交
2469 2470 2471
  return TSDB_CODE_SUCCESS;
}

2472 2473 2474 2475 2476 2477 2478 2479 2480 2481 2482 2483 2484 2485 2486 2487 2488 2489 2490 2491 2492 2493 2494
static TSDBKEY getCurrentKeyInBuf(SDataBlockIter* pBlockIter, STsdbReader* pReader) {
  TSDBKEY key = {.ts = TSKEY_INITIAL_VAL};

  SFileDataBlockInfo* pFBlock = getCurrentBlockInfo(pBlockIter);
  STableBlockScanInfo* pScanInfo = taosHashGet(pReader->status.pTableMap, &pFBlock->uid, sizeof(pFBlock->uid));

  initMemIterator(pScanInfo, pReader);
  if (pScanInfo->memHasVal) {
    TSDBROW* pRow = getValidRow(pScanInfo->iter, &pScanInfo->memHasVal, pReader);
    key = TSDBROW_KEY(pRow);
  }

  if (pScanInfo->imemHasVal) {
    TSDBROW* pRow = getValidRow(pScanInfo->iiter, &pScanInfo->imemHasVal, pReader);
    TSDBKEY k = TSDBROW_KEY(pRow);
    if (key.ts > k.ts) {
      key = k;
    }
  }

  return key;
}

H
Haojun Liao 已提交
2495 2496 2497 2498
static int32_t moveToNextFile(STsdbReader* pReader, int32_t* numOfBlocks) {
  SReaderStatus* pStatus = &pReader->status;

  while (1) {
2499
    bool hasNext = filesetIteratorNext(&pStatus->fileIter, pReader);
H
Haojun Liao 已提交
2500 2501 2502 2503 2504 2505 2506 2507 2508 2509 2510 2511 2512 2513 2514 2515 2516 2517 2518 2519 2520 2521 2522 2523 2524 2525 2526 2527
    if (!hasNext) { // no data files on disk
      break;
    }

    SArray* pIndexList = taosArrayInit(4, sizeof(SBlockIdx));
    int32_t code = doLoadBlockIndex(pReader, pReader->pFileReader, pIndexList);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

    if (taosArrayGetSize(pIndexList) > 0) {
      uint32_t numOfValidTable = 0;
      code = doLoadFileBlock(pReader, pIndexList, &numOfValidTable, numOfBlocks);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }

      if (numOfValidTable > 0) {
        break;
      }
    }

    // no blocks in current file, try next files
  }

  return TSDB_CODE_SUCCESS;
}

2528 2529 2530 2531 2532 2533 2534 2535 2536 2537 2538 2539 2540
static int32_t doBuildDataBlock(STsdbReader* pReader) {
  int32_t code = TSDB_CODE_SUCCESS;

  SReaderStatus* pStatus = &pReader->status;
  SDataBlockIter* pBlockIter = &pStatus->blockIter;

  SFileDataBlockInfo* pFBlock = getCurrentBlockInfo(pBlockIter);
  STableBlockScanInfo* pScanInfo = taosHashGet(pReader->status.pTableMap, &pFBlock->uid, sizeof(pFBlock->uid));

  SBlock* pBlock = taosArrayGet(pScanInfo->pBlockList, pFBlock->tbBlockIdx);

  TSDBKEY key = getCurrentKeyInBuf(pBlockIter, pReader);
  if (fileBlockShouldLoad(pReader, pFBlock, pBlock, pScanInfo, key)) {
2541 2542
    tBlockDataInit(&pStatus->fileBlockData);
    code = doLoadFileBlockData(pReader, pBlockIter, pScanInfo, &pStatus->fileBlockData);
2543 2544 2545 2546 2547 2548 2549 2550 2551 2552 2553 2554 2555 2556 2557 2558 2559 2560 2561 2562
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

    // build composed data block
    code = buildComposedDataBlock(pReader, pFBlock, pBlock, pScanInfo);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
  } else if (bufferDataInFileBlockGap(pReader->order, key, pBlock)) {
    // data in memory that are earlier than current file block
    TSDBKEY maxKey = {.ts = pReader->window.ekey, .version = pReader->verRange.maxVer};
    code = buildInmemDataBlock(pReader, pScanInfo, &maxKey);
    // build data block from in-memory buffer data completed.
  } else { // whole block is required, return it directly
    // todo
    // 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
    SDataBlockInfo* pInfo = &pReader->pResBlock->info;
H
Haojun Liao 已提交
2563 2564 2565
    pInfo->rows   = pBlock->nRow;
    pInfo->uid    = pScanInfo->uid;
    pInfo->window = (STimeWindow) {.skey = pBlock->minKey.ts, .ekey = pBlock->maxKey.ts};
2566
    setComposedBlockFlag(pReader, false);
2567
    setBlockAllDumped(&pStatus->fBlockDumpInfo, pBlock, pReader->order);
2568 2569 2570 2571 2572
  }

  return code;
}

H
Haojun Liao 已提交
2573
static int32_t buildBlockFromBufferSequentially(STsdbReader* pReader) {
2574 2575 2576 2577 2578 2579
  SReaderStatus* pStatus = &pReader->status;

  while(1) {
    if (pStatus->pTableIter == NULL) {
      pStatus->pTableIter = taosHashIterate(pStatus->pTableMap, NULL);
      if (pStatus->pTableIter == NULL) {
H
Haojun Liao 已提交
2580
        return TSDB_CODE_SUCCESS;
2581 2582 2583 2584 2585 2586 2587
      }
    }

    STableBlockScanInfo* pBlockScanInfo = pStatus->pTableIter;
    initMemIterator(pBlockScanInfo, pReader);

    TSDBKEY maxKey = {.ts = pReader->window.ekey, .version = pReader->verRange.maxVer};
H
Haojun Liao 已提交
2588 2589 2590 2591 2592
    int32_t code = buildInmemDataBlock(pReader, pBlockScanInfo, &maxKey);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

2593
    if (pReader->pResBlock->info.rows > 0) {
H
Haojun Liao 已提交
2594
      return TSDB_CODE_SUCCESS;
2595 2596 2597 2598 2599
    }

    // current table is exhausted, let's try the next table
    pStatus->pTableIter = taosHashIterate(pStatus->pTableMap, pStatus->pTableIter);
    if (pStatus->pTableIter == NULL) {
H
Haojun Liao 已提交
2600
      return TSDB_CODE_SUCCESS;
2601 2602 2603 2604
    }
  }
}

2605 2606 2607 2608 2609 2610 2611 2612 2613 2614 2615 2616
static void initBlockDumpInfo(STsdbReader* pReader, SDataBlockIter* pBlockIter) {
  SFileDataBlockInfo* pFBlock = getCurrentBlockInfo(pBlockIter);
  STableBlockScanInfo* pScanInfo = taosHashGet(pReader->status.pTableMap, &pFBlock->uid, sizeof(pFBlock->uid));
  SBlock* pBlock = taosArrayGet(pScanInfo->pBlockList, pFBlock->tbBlockIdx);

  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;

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

2617 2618 2619 2620 2621 2622 2623 2624 2625 2626 2627 2628 2629 2630 2631
static int32_t initForFirstBlockOfFile(STsdbReader* pReader, SDataBlockIter* pBlockIter) {
  int32_t numOfBlocks = 0;
  int32_t code = moveToNextFile(pReader, &numOfBlocks);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

  // all data files are consumed, try data in buffer
  if (numOfBlocks == 0) {
    pReader->status.loadFromFile = false;
    return code;
  }

  // initialize the block iterator for a new fileset
  code = initBlockIterator(pReader, pBlockIter, numOfBlocks);
2632
  initBlockDumpInfo(pReader, pBlockIter);
2633 2634 2635
  return code;
}

2636
static int32_t buildBlockFromFiles(STsdbReader* pReader) {
H
Haojun Liao 已提交
2637
  int32_t code = TSDB_CODE_SUCCESS;
2638 2639
  SDataBlockIter* pBlockIter = &pReader->status.blockIter;

2640 2641 2642 2643 2644 2645 2646 2647 2648 2649 2650 2651 2652 2653 2654 2655 2656
  if (pReader->status.blockIter.index == -1) {
    code = initForFirstBlockOfFile(pReader, pBlockIter);
    if ((code != TSDB_CODE_SUCCESS) || (pReader->status.loadFromFile == false)) {
      return code;
    }
    
    code = doBuildDataBlock(pReader);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
  } else {
    SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
    SFileDataBlockInfo* pFBlock = getCurrentBlockInfo(&pReader->status.blockIter);
    STableBlockScanInfo* pScanInfo = taosHashGet(pReader->status.pTableMap, &pFBlock->uid, sizeof(pFBlock->uid));
    SBlock* pBlock = taosArrayGet(pScanInfo->pBlockList, pFBlock->tbBlockIdx);

    // current block are exhausted, try the next file block
2657
    if (pDumpInfo->allDumped) {
2658 2659
      // try next data block in current file
      bool hasNext = blockIteratorNext(&pReader->status.blockIter);
2660 2661 2662
      if (hasNext) {  // current file is exhausted, let's try the next file
        initBlockDumpInfo(pReader, pBlockIter);
      } else {
2663 2664 2665 2666
        code = initForFirstBlockOfFile(pReader, pBlockIter);
        if ((code != TSDB_CODE_SUCCESS) || (pReader->status.loadFromFile == false)) {
          return code;
        }
H
Haojun Liao 已提交
2667
      }
2668
      
2669
      code = doBuildDataBlock(pReader);
H
Haojun Liao 已提交
2670 2671 2672
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
H
Haojun Liao 已提交
2673
    } else {
2674 2675
      code = buildComposedDataBlock(pReader, pFBlock, pBlock, pScanInfo);
      return code;
2676 2677 2678
    }
  }

H
Haojun Liao 已提交
2679
  return code;
2680
}
H
refact  
Hongze Cheng 已提交
2681

H
Hongze Cheng 已提交
2682 2683 2684 2685
// // todo not unref yet, since it is not support multi-group interpolation query
// static UNUSED_FUNC void changeQueryHandleForInterpQuery(STsdbReader* pHandle) {
//   // filter the queried time stamp in the first place
//   STsdbReader* pTsdbReadHandle = (STsdbReader*)pHandle;
H
refact  
Hongze Cheng 已提交
2686

H
Hongze Cheng 已提交
2687 2688
//   // starts from the buffer in case of descending timestamp order check data blocks
//   size_t numOfTables = taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
H
refact  
Hongze Cheng 已提交
2689

H
Hongze Cheng 已提交
2690 2691
//   int32_t i = 0;
//   while (i < numOfTables) {
H
Haojun Liao 已提交
2692
//     STableBlockScanInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, i);
H
refact  
Hongze Cheng 已提交
2693

H
Hongze Cheng 已提交
2694 2695 2696 2697 2698 2699 2700 2701 2702 2703 2704 2705 2706 2707
//     // the first qualified table for interpolation query
//     //    if ((pTsdbReadHandle->window.skey <= pCheckInfo->pTableObj->lastKey) &&
//     //        (pCheckInfo->pTableObj->lastKey != TSKEY_INITIAL_VAL)) {
//     //      break;
//     //    }

//     i++;
//   }

//   // there are no data in all the tables
//   if (i == numOfTables) {
//     return;
//   }

H
Haojun Liao 已提交
2708
//   STableBlockScanInfo info = *(STableBlockScanInfo*)taosArrayGet(pTsdbReadHandle->pTableCheckInfo, i);
H
Hongze Cheng 已提交
2709 2710 2711 2712 2713 2714
//   taosArrayClear(pTsdbReadHandle->pTableCheckInfo);

//   info.lastKey = pTsdbReadHandle->window.skey;
//   taosArrayPush(pTsdbReadHandle->pTableCheckInfo, &info);
// }

2715 2716 2717 2718
static bool outofTimeWindow(int64_t ts, STimeWindow* pWindow, int32_t order) {
  return (((ts > pWindow->ekey) && ASCENDING_TRAVERSE(order)) || ((ts < pWindow->skey) && ASCENDING_TRAVERSE(order)));
}

H
Haojun Liao 已提交
2719
TSDBROW* getValidRow(STbDataIter* pIter, bool* hasVal, STsdbReader* pReader) {
2720
  if (!(*hasVal)) {
H
Haojun Liao 已提交
2721 2722
    return NULL;
  }
H
Hongze Cheng 已提交
2723

H
Haojun Liao 已提交
2724
  TSDBROW* pRow = tsdbTbDataIterGet(pIter);
H
Hongze Cheng 已提交
2725

H
Haojun Liao 已提交
2726
  TSDBKEY key = TSDBROW_KEY(pRow);
2727
  if (outofTimeWindow(key.ts, &pReader->window, pReader->order)) {
H
Haojun Liao 已提交
2728 2729 2730
    *hasVal = false;
    return NULL;
  }
H
Hongze Cheng 已提交
2731

H
Haojun Liao 已提交
2732
  if (key.version <= pReader->verRange.maxVer) {
H
Haojun Liao 已提交
2733 2734
    return pRow;
  }
H
Hongze Cheng 已提交
2735

H
Haojun Liao 已提交
2736 2737 2738 2739 2740
  while(1) {
    *hasVal = tsdbTbDataIterNext(pIter);
    if (!(*hasVal)) {
      return NULL;
    }
H
Hongze Cheng 已提交
2741

H
Haojun Liao 已提交
2742
    pRow = tsdbTbDataIterGet(pIter);
H
Hongze Cheng 已提交
2743

H
Haojun Liao 已提交
2744
    key = TSDBROW_KEY(pRow);
2745
    if (outofTimeWindow(key.ts, &pReader->window, pReader->order)) {
H
Haojun Liao 已提交
2746 2747 2748
      *hasVal = false;
      return NULL;
    }
H
Hongze Cheng 已提交
2749

H
Haojun Liao 已提交
2750
    if (key.version <= pReader->verRange.maxVer) {
H
Haojun Liao 已提交
2751 2752 2753 2754
      return pRow;
    }
  }
}
H
Hongze Cheng 已提交
2755

H
Haojun Liao 已提交
2756 2757 2758
int32_t doLoadRowsOfIdenticalTs(STbDataIter *pIter, bool* hasVal, int64_t ts, SRowMerger* pMerger, STsdbReader* pReader) {
  while (1) {
    *hasVal = tsdbTbDataIterNext(pIter);
2759
    if (!(*hasVal)) {
H
Haojun Liao 已提交
2760 2761
      break;
    }
H
Hongze Cheng 已提交
2762

H
Haojun Liao 已提交
2763 2764
    TSDBROW* pRow = getValidRow(pIter, hasVal, pReader);
    TSDBKEY k = TSDBROW_KEY(pRow);
2765
    if (k.ts != ts) {
H
Haojun Liao 已提交
2766 2767 2768 2769 2770 2771 2772 2773 2774
      break;
    }

    tRowMerge(pMerger, pRow);
  }

  return TSDB_CODE_SUCCESS;
}

2775 2776 2777
int32_t doLoadRowsOfIdenticalTsInFileBlock(SFileDataBlockInfo* pFBlock, SBlock* pBlock, SBlockData* pBlockData,
                                           STableBlockScanInfo* pScanInfo, STsdbReader* pReader, SRowMerger* pMerger) {
  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
2778
  bool asc = ASCENDING_TRAVERSE(pReader->order);
2779

2780
  int32_t step = asc? 1:-1;
2781 2782
  int64_t key = pBlockData->aTSKEY[pDumpInfo->rowIndex];

2783 2784 2785 2786 2787 2788 2789 2790 2791 2792 2793 2794 2795
  if (asc) {  // todo refactor
    if (pDumpInfo->rowIndex < pBlockData->nRow - 1) {
      if (pBlockData->aTSKEY[pDumpInfo->rowIndex + step] == key) {
        int32_t rowIndex = pDumpInfo->rowIndex + step;

        while (pBlockData->aTSKEY[rowIndex] == key) {
          if (pBlockData->aVersion[rowIndex] > pReader->verRange.maxVer || pBlockData->aVersion[rowIndex] < pReader->verRange.minVer) {
            continue;
          }

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

2798 2799 2800 2801 2802 2803 2804 2805 2806
        pDumpInfo->rowIndex = rowIndex;
      } else {
        pDumpInfo->rowIndex += step;
      }
    } else {  // last row of current block, check if current block is overlapped with neighbor block
      pDumpInfo->rowIndex += step;
      bool overlap = overlapWithNeighborBlock(pFBlock, pBlock, pScanInfo);
      if (overlap) {  // load next block
        ASSERT(0);
H
Haojun Liao 已提交
2807 2808
      }
    }
2809 2810 2811 2812 2813 2814 2815 2816 2817 2818 2819 2820 2821 2822 2823
  } else {
    if (pDumpInfo->rowIndex > 0) {
      if (pBlockData->aTSKEY[pDumpInfo->rowIndex + step] == key) {
        int32_t rowIndex = pDumpInfo->rowIndex + step;

        while (pBlockData->aTSKEY[rowIndex] == key) {
          if (pBlockData->aVersion[rowIndex] > pReader->verRange.maxVer ||
              pBlockData->aVersion[rowIndex] < pReader->verRange.minVer) {
            continue;
          }

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

2825 2826 2827 2828 2829 2830 2831 2832 2833 2834
        pDumpInfo->rowIndex = rowIndex;
      } else {
        pDumpInfo->rowIndex += step;
      }
    } else {  // last row of current block, check if current block is overlapped with previous neighbor block
      pDumpInfo->rowIndex += step;
      bool overlap = overlapWithNeighborBlock(pFBlock, pBlock, pScanInfo);
      if (overlap) {  // load next block
        ASSERT(0);
      }
2835
    }
H
Haojun Liao 已提交
2836
  }
2837

H
Haojun Liao 已提交
2838 2839 2840
  return TSDB_CODE_SUCCESS;
}

2841 2842 2843 2844 2845 2846 2847 2848 2849 2850 2851
static void checkUpdateSchema(TSDBROW* pRow, uint64_t uid, STsdbReader* pReader) {
  int32_t sversion = TSDBROW_SVERSION(pRow);

  if (pReader->pSchema == NULL) {
    pReader->pSchema = metaGetTbTSchema(pReader->pTsdb->pVnode->pMeta, uid, sversion);
  } else if (pReader->pSchema->version != sversion) {
    taosMemoryFreeClear(pReader->pSchema);
    pReader->pSchema = metaGetTbTSchema(pReader->pTsdb->pVnode->pMeta, uid, sversion);
  }
}

H
Haojun Liao 已提交
2852 2853 2854 2855 2856
int32_t tsdbGetNextRowInMem(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader, STSRow** pTSRow) {
  TSKEY     mergeTs = TSKEY_INITIAL_VAL;

  SRowMerger merge = {0};

2857 2858
  TSDBROW* pRow  = getValidRow(pBlockScanInfo->iter, &pBlockScanInfo->memHasVal, pReader);
  TSDBROW* piRow = getValidRow(pBlockScanInfo->iiter, &pBlockScanInfo->imemHasVal, pReader);
H
Haojun Liao 已提交
2859

2860
  TSDBKEY k  = {.ts = TSKEY_INITIAL_VAL};
H
Haojun Liao 已提交
2861 2862 2863 2864 2865 2866 2867
  TSDBKEY ik = {.ts = TSKEY_INITIAL_VAL};

  if (pBlockScanInfo->memHasVal && pBlockScanInfo->imemHasVal) {
    k = TSDBROW_KEY(pRow);
    ik = TSDBROW_KEY(piRow);

    if (ik.ts <= k.ts) {
2868 2869
      checkUpdateSchema(piRow, pBlockScanInfo->uid, pReader);

2870 2871
      tRowMergerInit(&merge, piRow, pReader->pSchema);
      doLoadRowsOfIdenticalTs(pBlockScanInfo->iiter, &pBlockScanInfo->imemHasVal, ik.ts, &merge, pReader);
H
Haojun Liao 已提交
2872 2873

      if (k.ts == mergeTs) {
2874
        doLoadRowsOfIdenticalTs(pBlockScanInfo->iter, &pBlockScanInfo->memHasVal, k.ts, &merge, pReader);
H
Haojun Liao 已提交
2875 2876 2877 2878 2879
      }

      tRowMergerGetRow(&merge, pTSRow);
      return TSDB_CODE_SUCCESS;
    } else {  // k.ts < ik.ts
2880 2881
      checkUpdateSchema(pRow, pBlockScanInfo->uid, pReader);

2882 2883
      tRowMergerInit(&merge, pRow, pReader->pSchema);
      doLoadRowsOfIdenticalTs(pBlockScanInfo->iter, &pBlockScanInfo->memHasVal, k.ts, &merge, pReader);
H
Haojun Liao 已提交
2884 2885 2886 2887 2888
      tRowMergerGetRow(&merge, pTSRow);
      return TSDB_CODE_SUCCESS;
    }
  }

2889

H
Haojun Liao 已提交
2890
  if (pBlockScanInfo->memHasVal) {
2891
    k = TSDBROW_KEY(pRow);
2892 2893
    checkUpdateSchema(pRow, pBlockScanInfo->uid, pReader);

2894 2895
    tRowMergerInit(&merge, pRow, pReader->pSchema);
    doLoadRowsOfIdenticalTs(pBlockScanInfo->iter, &pBlockScanInfo->memHasVal, k.ts, &merge, pReader);
H
Haojun Liao 已提交
2896 2897 2898 2899 2900
    tRowMergerGetRow(&merge, pTSRow);
    return TSDB_CODE_SUCCESS;
  }

  if (pBlockScanInfo->imemHasVal) {
2901
    ik = TSDBROW_KEY(piRow);
2902 2903
    checkUpdateSchema(piRow, pBlockScanInfo->uid, pReader);

2904 2905
    tRowMergerInit(&merge, piRow, pReader->pSchema);
    doLoadRowsOfIdenticalTs(pBlockScanInfo->iiter, &pBlockScanInfo->imemHasVal, ik.ts, &merge, pReader);
H
Haojun Liao 已提交
2906 2907 2908 2909 2910 2911 2912
    tRowMergerGetRow(&merge, pTSRow);
    return TSDB_CODE_SUCCESS;
  }

  return TSDB_CODE_SUCCESS;
}

2913 2914 2915 2916
int32_t doAppendOneRow(SSDataBlock* pBlock, STsdbReader* pReader, STSRow* pTSRow) {
  int32_t numOfRows = pBlock->info.rows;
  int32_t numOfCols = (int32_t)taosArrayGetSize(pBlock->pDataBlock);

2917
  SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo;
2918
  STSchema *pSchema = pReader->pSchema;
2919

2920
  SColVal colVal = {0};
2921
  int32_t i = 0, j = 0;
H
Haojun Liao 已提交
2922

2923 2924 2925 2926 2927 2928 2929 2930 2931 2932 2933 2934 2935 2936 2937 2938 2939 2940 2941 2942
  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) {
      tTSRowGetVal(pTSRow, pReader->pSchema, j, &colVal);
      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;
2943
    }
2944 2945
  }

2946 2947 2948 2949 2950 2951 2952
  // set null value since current column does not exist in the "pSchema"
  while(i < numOfCols) {
    pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
    colDataAppendNULL(pColInfoData, numOfRows);
    i += 1;
  }

2953 2954 2955 2956 2957
  pBlock->info.rows += 1;
  return TSDB_CODE_SUCCESS;
}

int32_t buildInmemDataBlockImpl(STableBlockScanInfo* pBlockScanInfo, TSDBKEY maxKey, int32_t capacity, STsdbReader* pReader) {
H
Haojun Liao 已提交
2958 2959 2960 2961 2962
  SSDataBlock* pBlock = pReader->pResBlock;

  do {
    STSRow* pTSRow = NULL;
    tsdbGetNextRowInMem(pBlockScanInfo, pReader, &pTSRow);
2963 2964
    if (pTSRow == NULL) {
      break;
H
Haojun Liao 已提交
2965 2966
    }

2967
    doAppendOneRow(pBlock, pReader, pTSRow);
H
Haojun Liao 已提交
2968 2969 2970 2971 2972 2973

    // no data in buffer, return immediately
    if (!(pBlockScanInfo->memHasVal || pBlockScanInfo->imemHasVal)) {
      break;
    }

2974
    if (pBlock->info.rows >= capacity) {
H
Haojun Liao 已提交
2975 2976 2977 2978
      break;
    }
  } while (1);

2979
  ASSERT(pBlock->info.rows <= capacity);
H
Haojun Liao 已提交
2980 2981
  return TSDB_CODE_SUCCESS;
}
H
Hongze Cheng 已提交
2982

C
Cary Xu 已提交
2983 2984 2985 2986 2987 2988 2989 2990 2991 2992
/**
 * @brief Get all suids since suid
 *
 * @param pMeta
 * @param suid return all suids in one vnode if suid is 0
 * @param list
 * @return int32_t
 */
int32_t tsdbGetStbIdList(SMeta* pMeta, int64_t suid, SArray* list) {
  SMStbCursor* pCur = metaOpenStbCursor(pMeta, suid);
C
Cary Xu 已提交
2993 2994 2995
  if(!pCur) {
    return TSDB_CODE_FAILED;
  }
C
Cary Xu 已提交
2996 2997 2998 2999 3000 3001 3002 3003 3004 3005 3006 3007 3008 3009

  while (1) {
    tb_uid_t id = metaStbCursorNext(pCur);
    if (id == 0) {
      break;
    }

    taosArrayPush(list, &id);
  }

  metaCloseStbCursor(pCur);
  return TSDB_CODE_SUCCESS;
}

H
Hongze Cheng 已提交
3010 3011 3012 3013 3014 3015 3016 3017 3018 3019 3020 3021 3022 3023 3024 3025 3026 3027 3028 3029 3030 3031 3032
// static void destroyHelper(void* param) {
//   if (param == NULL) {
//     return;
//   }

//   //  tQueryInfo* pInfo = (tQueryInfo*)param;
//   //  if (pInfo->optr != TSDB_RELATION_IN) {
//   //    taosMemoryFreeClear(pInfo->q);
//   //  } else {
//   //    taosHashCleanup((SHashObj *)(pInfo->q));
//   //  }

//   taosMemoryFree(param);
// }

// #define TSDB_PREV_ROW 0x1
// #define TSDB_NEXT_ROW 0x2

// static bool loadBlockOfActiveTable(STsdbReader* pTsdbReadHandle) {
//   if (pTsdbReadHandle->checkFiles) {
//     // check if the query range overlaps with the file data block
//     bool exists = true;

3033
//     int32_t code = buildBlockFromFiles(pTsdbReadHandle, &exists);
H
Hongze Cheng 已提交
3034 3035 3036 3037 3038 3039 3040 3041 3042 3043 3044 3045 3046 3047 3048 3049 3050 3051 3052 3053 3054 3055 3056 3057 3058 3059 3060 3061 3062 3063 3064 3065 3066 3067 3068 3069 3070 3071 3072 3073 3074 3075 3076 3077 3078 3079 3080 3081 3082 3083 3084 3085 3086 3087 3088
//     if (code != TSDB_CODE_SUCCESS) {
//       pTsdbReadHandle->checkFiles = false;
//       return false;
//     }

//     if (exists) {
//       tsdbRetrieveDataBlock((STsdbReader**)pTsdbReadHandle, NULL);
//       if (pTsdbReadHandle->currentLoadExternalRows && pTsdbReadHandle->window.skey == pTsdbReadHandle->window.ekey) {
//         SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, 0);
//         assert(*(int64_t*)pColInfo->pData == pTsdbReadHandle->window.skey);
//       }

//       pTsdbReadHandle->currentLoadExternalRows = false;  // clear the flag, since the exact matched row is found.
//       return exists;
//     }

//     pTsdbReadHandle->checkFiles = false;
//   }

//   if (hasMoreDataInCache(pTsdbReadHandle)) {
//     pTsdbReadHandle->currentLoadExternalRows = false;
//     return true;
//   }

//   // current result is empty
//   if (pTsdbReadHandle->currentLoadExternalRows && pTsdbReadHandle->window.skey == pTsdbReadHandle->window.ekey &&
//       pTsdbReadHandle->cur.rows == 0) {
//     //    SMemTable* pMemRef = pTsdbReadHandle->pMemTable;

//     //    doGetExternalRow(pTsdbReadHandle, TSDB_PREV_ROW, pMemRef);
//     //    doGetExternalRow(pTsdbReadHandle, TSDB_NEXT_ROW, pMemRef);

//     bool result = tsdbGetExternalRow(pTsdbReadHandle);

//     //    pTsdbReadHandle->prev = doFreeColumnInfoData(pTsdbReadHandle->prev);
//     //    pTsdbReadHandle->next = doFreeColumnInfoData(pTsdbReadHandle->next);
//     pTsdbReadHandle->currentLoadExternalRows = false;

//     return result;
//   }

//   return false;
// }

// static bool loadDataBlockFromTableSeq(STsdbReader* pTsdbReadHandle) {
//   size_t numOfTables = taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
//   assert(numOfTables > 0);

//   int64_t stime = taosGetTimestampUs();

//   while (pTsdbReadHandle->activeIndex < numOfTables) {
//     if (loadBlockOfActiveTable(pTsdbReadHandle)) {
//       return true;
//     }

H
Haojun Liao 已提交
3089
//     STableBlockScanInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, pTsdbReadHandle->activeIndex);
H
Hongze Cheng 已提交
3090 3091 3092 3093 3094 3095 3096 3097 3098 3099 3100 3101 3102 3103 3104 3105 3106 3107 3108 3109 3110 3111 3112 3113 3114 3115 3116 3117 3118 3119 3120 3121 3122 3123 3124 3125 3126 3127 3128 3129 3130 3131 3132 3133 3134 3135 3136 3137 3138 3139 3140 3141 3142 3143 3144 3145 3146 3147 3148 3149 3150 3151 3152 3153 3154 3155
//     pCheckInfo->numOfBlocks = 0;

//     pTsdbReadHandle->activeIndex += 1;
//     pTsdbReadHandle->locateStart = false;
//     pTsdbReadHandle->checkFiles = true;
//     pTsdbReadHandle->cur.rows = 0;
//     pTsdbReadHandle->currentLoadExternalRows = pTsdbReadHandle->loadExternalRow;

//     terrno = TSDB_CODE_SUCCESS;

//     int64_t elapsedTime = taosGetTimestampUs() - stime;
//     pTsdbReadHandle->cost.checkForNextTime += elapsedTime;
//   }

//   return false;
// }

// bool tsdbGetExternalRow(STsdbReader* pHandle) {
//   STsdbReader*   pTsdbReadHandle = (STsdbReader*)pHandle;
//   SQueryFilePos* cur = &pTsdbReadHandle->cur;

//   cur->fid = INT32_MIN;
//   cur->mixBlock = true;
//   if (pTsdbReadHandle->prev == NULL || pTsdbReadHandle->next == NULL) {
//     cur->rows = 0;
//     return false;
//   }

//   int32_t numOfCols = (int32_t)QH_GET_NUM_OF_COLS(pTsdbReadHandle);
//   for (int32_t i = 0; i < numOfCols; ++i) {
//     SColumnInfoData* pColInfoData = taosArrayGet(pTsdbReadHandle->pColumns, i);
//     SColumnInfoData* first = taosArrayGet(pTsdbReadHandle->prev, i);

//     memcpy(pColInfoData->pData, first->pData, pColInfoData->info.bytes);

//     SColumnInfoData* sec = taosArrayGet(pTsdbReadHandle->next, i);
//     memcpy(((char*)pColInfoData->pData) + pColInfoData->info.bytes, sec->pData, pColInfoData->info.bytes);

//     if (i == 0 && pColInfoData->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
//       cur->win.skey = *(TSKEY*)pColInfoData->pData;
//       cur->win.ekey = *(TSKEY*)(((char*)pColInfoData->pData) + TSDB_KEYSIZE);
//     }
//   }

//   cur->rows = 2;
//   return true;
// }

// static void* doFreeColumnInfoData(SArray* pColumnInfoData) {
//   if (pColumnInfoData == NULL) {
//     return NULL;
//   }

//   size_t cols = taosArrayGetSize(pColumnInfoData);
//   for (int32_t i = 0; i < cols; ++i) {
//     SColumnInfoData* pColInfo = taosArrayGet(pColumnInfoData, i);
//     colDataDestroy(pColInfo);
//   }

//   taosArrayDestroy(pColumnInfoData);
//   return NULL;
// }

// static void* destroyTableCheckInfo(SArray* pTableCheckInfo) {
//   size_t size = taosArrayGetSize(pTableCheckInfo);
//   for (int32_t i = 0; i < size; ++i) {
H
Haojun Liao 已提交
3156
//     STableBlockScanInfo* p = taosArrayGet(pTableCheckInfo, i);
H
Hongze Cheng 已提交
3157 3158 3159 3160 3161 3162 3163 3164
//     destroyTableMemIterator(p);

//     taosMemoryFreeClear(p->pCompInfo);
//   }

//   taosArrayDestroy(pTableCheckInfo);
//   return NULL;
// }
H
refact  
Hongze Cheng 已提交
3165 3166

// ====================================== EXPOSED APIs ======================================
H
Haojun Liao 已提交
3167 3168
int32_t tsdbReaderOpen(SVnode* pVnode, SQueryTableDataCond* pCond, SArray* pTableList, STsdbReader** ppReader, const char* idstr) {
  int32_t code = tsdbReaderCreate(pVnode, pCond, ppReader, idstr);
H
Haojun Liao 已提交
3169 3170 3171
  if (code) {
    goto _err;
  }
H
Hongze Cheng 已提交
3172

H
Haojun Liao 已提交
3173
  STsdbReader* pReader = *ppReader;
H
Haojun Liao 已提交
3174
  if (isEmptyQueryTimeWindow(&pReader->window, pReader->order)) {
H
Haojun Liao 已提交
3175 3176 3177
    tsdbDebug("%p query window not overlaps with the data set, no result returned, %s", pReader, pReader->idStr);
    return TSDB_CODE_SUCCESS;
  }
H
Hongze Cheng 已提交
3178

3179 3180
  int32_t numOfTables = taosArrayGetSize(pTableList);
  pReader->status.pTableMap = createDataBlockScanInfo(pReader, pTableList->pData, numOfTables);
H
Haojun Liao 已提交
3181 3182 3183
  if (pReader->status.pTableMap == NULL) {
    tsdbReaderClose(pReader);
    *ppReader = NULL;
H
Haojun Liao 已提交
3184

H
Haojun Liao 已提交
3185 3186 3187
    code = TSDB_CODE_TDB_OUT_OF_MEMORY;
    goto _err;
  }
H
Hongze Cheng 已提交
3188

3189
  tsdbDebug("%p total numOfTable:%d in this query %s", pReader, numOfTables, pReader->idStr);
H
Hongze Cheng 已提交
3190
  return code;
H
Hongze Cheng 已提交
3191 3192

_err:
H
Haojun Liao 已提交
3193
  tsdbError("failed to create tsdb reader, code: %s %s", tstrerror(code), pReader->idStr);
H
Hongze Cheng 已提交
3194
  return code;
H
refact  
Hongze Cheng 已提交
3195 3196 3197
}

void tsdbReaderClose(STsdbReader* pReader) {
H
Haojun Liao 已提交
3198 3199 3200
   if (pReader == NULL) {
     return;
   }
H
refact  
Hongze Cheng 已提交
3201

H
Haojun Liao 已提交
3202
   blockDataDestroy(pReader->pResBlock);
H
refact  
Hongze Cheng 已提交
3203

H
Haojun Liao 已提交
3204 3205 3206
   taosMemoryFreeClear(pReader->suppInfo.pstatis);
   taosMemoryFreeClear(pReader->suppInfo.plist);
   taosMemoryFree(pReader->suppInfo.slotIds);
H
refact  
Hongze Cheng 已提交
3207

H
Haojun Liao 已提交
3208
   if (!isEmptyQueryTimeWindow(&pReader->window, pReader->order)) {
H
Haojun Liao 已提交
3209 3210 3211 3212 3213 3214 3215 3216
     //    tsdbMayUnTakeMemSnapshot(pTsdbReadHandle);
   } else {
     ASSERT(pReader->status.pTableMap == NULL);
   }
#if 0
//   if (pReader->status.pTableScanInfo != NULL) {
//     pReader->status.pTableScanInfo = destroyTableCheckInfo(pReader->status.pTableScanInfo);
//   }
H
refact  
Hongze Cheng 已提交
3217

H
Haojun Liao 已提交
3218
//   tsdbDestroyReadH(&pReader->rhelper);
H
refact  
Hongze Cheng 已提交
3219

H
Haojun Liao 已提交
3220 3221 3222 3223 3224 3225
//   tdFreeDataCols(pReader->pDataCols);
//   pReader->pDataCols = NULL;
//
//   pReader->prev = doFreeColumnInfoData(pReader->prev);
//   pReader->next = doFreeColumnInfoData(pReader->next);
#endif
H
refact  
Hongze Cheng 已提交
3226

H
Haojun Liao 已提交
3227
   SIOCostSummary* pCost = &pReader->cost;
H
refact  
Hongze Cheng 已提交
3228

H
Haojun Liao 已提交
3229 3230 3231 3232
   tsdbDebug("%p :io-cost summary: head-file read cnt:%" PRIu64 ", head-file time:%" PRIu64 " us, statis-info:%" PRId64
             " us, datablock:%" PRId64 " us, check data:%" PRId64 " us, %s",
             pReader, pCost->headFileLoad, pCost->headFileLoadTime, pCost->statisInfoLoadTime, pCost->blockLoadTime,
             pCost->checkForNextTime, pReader->idStr);
H
refact  
Hongze Cheng 已提交
3233

H
Haojun Liao 已提交
3234 3235 3236
   taosMemoryFree(pReader->idStr);
   taosMemoryFree(pReader->pSchema);
   taosMemoryFreeClear(pReader);
H
refact  
Hongze Cheng 已提交
3237 3238 3239
}

bool tsdbNextDataBlock(STsdbReader* pReader) {
H
Haojun Liao 已提交
3240
  if (isEmptyQueryTimeWindow(&pReader->window, pReader->order)) {
H
Haojun Liao 已提交
3241 3242
    return false;
  }
H
Hongze Cheng 已提交
3243

H
Haojun Liao 已提交
3244
  // cleanup the data that belongs to the previous data block
3245 3246
  SSDataBlock* pBlock = pReader->pResBlock;
  blockDataCleanup(pBlock);
H
Hongze Cheng 已提交
3247

H
Haojun Liao 已提交
3248 3249
  int64_t stime = taosGetTimestampUs();
  int64_t elapsedTime = stime;
3250
  SReaderStatus* pStatus = &pReader->status;
H
Haojun Liao 已提交
3251 3252

  if (pReader->type == BLOCK_LOAD_OFFSET_ORDER) {
3253
    if (pStatus->loadFromFile) {
3254
      int32_t code = buildBlockFromFiles(pReader);
3255 3256 3257 3258
      if (code != TSDB_CODE_SUCCESS) {
        return false;
      }

3259
      if (pBlock->info.rows > 0) {
3260
        return true;
3261
      } else {
H
Haojun Liao 已提交
3262
        buildBlockFromBufferSequentially(pReader);
3263
        return pBlock->info.rows > 0;
3264
      }
3265
    } else { // no data in files, let's try the buffer
H
Haojun Liao 已提交
3266
      buildBlockFromBufferSequentially(pReader);
3267
      return pBlock->info.rows > 0;
H
Haojun Liao 已提交
3268 3269 3270 3271 3272
    }
  } else if (pReader->type == BLOCK_LOAD_TABLESEQ_ORDER) {

  } else if (pReader->type == BLOCK_LOAD_EXTERN_ORDER) {

3273 3274
  } else {
    ASSERT(0);
H
Haojun Liao 已提交
3275
  }
H
Hongze Cheng 已提交
3276 3277 3278 3279 3280 3281
  // if (pReader->loadType == BLOCK_LOAD_TABLE_SEQ_ORDER) {
  //   return loadDataBlockFromTableSeq(pReader);
  // } else {  // loadType == RR and Offset Order
  //   if (pReader->checkFiles) {
  //     // check if the query range overlaps with the file data block
  //     bool exists = true;
3282
  //     int32_t code = buildBlockFromFiles(pReader, &exists);
H
Hongze Cheng 已提交
3283 3284 3285 3286 3287 3288 3289 3290 3291 3292 3293 3294 3295 3296 3297 3298 3299 3300 3301 3302 3303 3304 3305 3306
  //     if (code != TSDB_CODE_SUCCESS) {
  //       pReader->activeIndex = 0;
  //       pReader->checkFiles = false;

  //       return false;
  //     }

  //     if (exists) {
  //       pReader->cost.checkForNextTime += (taosGetTimestampUs() - stime);
  //       return exists;
  //     }

  //     pReader->activeIndex = 0;
  //     pReader->checkFiles = false;
  //   }

  //   // TODO: opt by consider the scan order
  //   bool ret = doHasDataInBuffer(pReader);
  //   terrno = TSDB_CODE_SUCCESS;

  //   elapsedTime = taosGetTimestampUs() - stime;
  //   pReader->cost.checkForNextTime += elapsedTime;
  //   return ret;
  // }
3307
  return false;
H
refact  
Hongze Cheng 已提交
3308 3309 3310
}

void tsdbRetrieveDataBlockInfo(STsdbReader* pReader, SDataBlockInfo* pDataBlockInfo) {
3311 3312 3313 3314
  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 已提交
3315 3316
}

H
Hongze Cheng 已提交
3317
int32_t tsdbRetrieveDataBlockStatisInfo(STsdbReader* pReader, SColumnDataAgg*** pBlockStatis, bool* allHave) {
H
Hongze Cheng 已提交
3318
  int32_t code = 0;
3319
  *allHave = false;
H
Hongze Cheng 已提交
3320

3321 3322 3323 3324
  if (pReader->status.composedDataBlock) {
    *pBlockStatis = NULL;
    return TSDB_CODE_SUCCESS;
  }
H
Hongze Cheng 已提交
3325

H
Haojun Liao 已提交
3326
  // SFileBlockInfo* pBlockInfo = &pReader->pDataBlockInfo[c->slot];
H
Hongze Cheng 已提交
3327 3328 3329 3330 3331 3332 3333 3334 3335 3336 3337 3338 3339 3340 3341 3342 3343 3344 3345 3346 3347 3348 3349 3350 3351 3352 3353 3354 3355 3356 3357 3358 3359 3360 3361 3362 3363 3364 3365 3366 3367 3368 3369 3370 3371 3372 3373 3374 3375 3376 3377 3378 3379 3380 3381 3382 3383 3384 3385 3386 3387 3388
  // assert((c->slot >= 0 && c->slot < pReader->numOfBlocks) || ((c->slot == pReader->numOfBlocks) && (c->slot == 0)));

  // // file block with sub-blocks has no statistics data
  // if (pBlockInfo->compBlock->numOfSubBlocks > 1) {
  //   *pBlockStatis = NULL;
  //   return TSDB_CODE_SUCCESS;
  // }

  // int64_t stime = taosGetTimestampUs();
  // int     statisStatus = tsdbLoadBlockStatis(&pReader->rhelper, pBlockInfo->compBlock);
  // if (statisStatus < TSDB_STATIS_OK) {
  //   return terrno;
  // } else if (statisStatus > TSDB_STATIS_OK) {
  //   *pBlockStatis = NULL;
  //   return TSDB_CODE_SUCCESS;
  // }

  // tsdbDebug("vgId:%d, succeed to load block statis part for uid %" PRIu64, REPO_ID(pReader->pTsdb),
  //           TSDB_READ_TABLE_UID(&pReader->rhelper));

  // int16_t* colIds = pReader->suppInfo.defaultLoadColumn->pData;

  // size_t numOfCols = QH_GET_NUM_OF_COLS(pReader);
  // memset(pReader->suppInfo.plist, 0, numOfCols * POINTER_BYTES);
  // memset(pReader->suppInfo.pstatis, 0, numOfCols * sizeof(SColumnDataAgg));

  // for (int32_t i = 0; i < numOfCols; ++i) {
  //   pReader->suppInfo.pstatis[i].colId = colIds[i];
  // }

  // *allHave = true;
  // tsdbGetBlockStatis(&pReader->rhelper, pReader->suppInfo.pstatis, (int)numOfCols, pBlockInfo->compBlock);

  // // always load the first primary timestamp column data
  // SColumnDataAgg* pPrimaryColStatis = &pReader->suppInfo.pstatis[0];
  // assert(pPrimaryColStatis->colId == PRIMARYKEY_TIMESTAMP_COL_ID);

  // pPrimaryColStatis->numOfNull = 0;
  // pPrimaryColStatis->min = pBlockInfo->compBlock->minKey.ts;
  // pPrimaryColStatis->max = pBlockInfo->compBlock->maxKey.ts;
  // pReader->suppInfo.plist[0] = &pReader->suppInfo.pstatis[0];

  // // update the number of NULL data rows
  // int32_t* slotIds = pReader->suppInfo.slotIds;
  // for (int32_t i = 1; i < numOfCols; ++i) {
  //   ASSERT(colIds[i] == pReader->pSchema->columns[slotIds[i]].colId);
  //   if (IS_BSMA_ON(&(pReader->pSchema->columns[slotIds[i]]))) {
  //     if (pReader->suppInfo.pstatis[i].numOfNull == -1) {  // set the column data are all NULL
  //       pReader->suppInfo.pstatis[i].numOfNull = pBlockInfo->compBlock->numOfRows;
  //     }

  //     pReader->suppInfo.plist[i] = &pReader->suppInfo.pstatis[i];
  //   } else {
  //     *allHave = false;
  //   }
  // }

  // int64_t elapsed = taosGetTimestampUs() - stime;
  // pReader->cost.statisInfoLoadTime += elapsed;

  // *pBlockStatis = pReader->suppInfo.plist;
  return code;
H
Hongze Cheng 已提交
3389 3390
}

H
Hongze Cheng 已提交
3391
SArray* tsdbRetrieveDataBlock(STsdbReader* pReader, SArray* pIdList) {
H
Haojun Liao 已提交
3392 3393 3394
  SReaderStatus* pStatus = &pReader->status;

  if (pStatus->composedDataBlock) {
3395
    return pReader->pResBlock->pDataBlock;
3396
  }
3397

3398 3399
  SFileDataBlockInfo* pFBlock = getCurrentBlockInfo(&pStatus->blockIter);
  STableBlockScanInfo* pBlockScanInfo = taosHashGet(pStatus->pTableMap, &pFBlock->uid, sizeof(pFBlock->uid));
3400

3401 3402 3403 3404 3405
  int32_t code = tBlockDataInit(&pStatus->fileBlockData);
  if (code != TSDB_CODE_SUCCESS) {
    terrno = code;
    return NULL;
  }
3406

3407 3408 3409 3410
  code = doLoadFileBlockData(pReader, &pStatus->blockIter, pBlockScanInfo, &pStatus->fileBlockData);
  if (code != TSDB_CODE_SUCCESS) {
    terrno = code;
    return NULL;
3411
  }
3412 3413 3414

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

H
refact  
Hongze Cheng 已提交
3417
void tsdbResetReadHandle(STsdbReader* pReader, SQueryTableDataCond* pCond, int32_t tWinIdx) {
H
Haojun Liao 已提交
3418
  // if (isEmptyQueryTimeWindow(pReader)) {
H
Hongze Cheng 已提交
3419 3420 3421 3422 3423 3424 3425 3426 3427 3428 3429 3430 3431 3432 3433 3434 3435 3436 3437 3438 3439 3440 3441 3442 3443 3444 3445 3446 3447 3448 3449 3450
  //   if (pCond->order != pReader->order) {
  //     pReader->order = pCond->order;
  //     TSWAP(pReader->window.skey, pReader->window.ekey);
  //   }

  //   return;
  // }

  // pReader->order = pCond->order;
  // setQueryTimewindow(pReader, pCond, tWinIdx);
  // pReader->type = TSDB_QUERY_TYPE_ALL;
  // pReader->cur.fid = -1;
  // pReader->cur.win = TSWINDOW_INITIALIZER;
  // pReader->checkFiles = true;
  // pReader->activeIndex = 0;  // current active table index
  // pReader->locateStart = false;
  // pReader->loadExternalRow = pCond->loadExternalRows;

  // if (ASCENDING_TRAVERSE(pCond->order)) {
  //   assert(pReader->window.skey <= pReader->window.ekey);
  // } else {
  //   assert(pReader->window.skey >= pReader->window.ekey);
  // }

  // // allocate buffer in order to load data blocks from file
  // memset(pReader->suppInfo.pstatis, 0, sizeof(SColumnDataAgg));
  // memset(pReader->suppInfo.plist, 0, POINTER_BYTES);

  // tsdbInitDataBlockLoadInfo(&pReader->dataBlockLoadInfo);
  // tsdbInitCompBlockLoadInfo(&pReader->compBlockLoadInfo);

  // resetCheckInfo(pReader);
H
Hongze Cheng 已提交
3451
}
H
Hongze Cheng 已提交
3452

H
refact  
Hongze Cheng 已提交
3453
int32_t tsdbGetFileBlocksDistInfo(STsdbReader* pReader, STableBlockDistInfo* pTableBlockInfo) {
H
Hongze Cheng 已提交
3454 3455 3456
  int32_t code = 0;
  // pTableBlockInfo->totalSize = 0;
  // pTableBlockInfo->totalRows = 0;
H
Hongze Cheng 已提交
3457

H
Hongze Cheng 已提交
3458
  // STsdbFS* pFileHandle = REPO_FS(pReader->pTsdb);
H
Hongze Cheng 已提交
3459

H
Hongze Cheng 已提交
3460 3461 3462 3463
  // // find the start data block in file
  // pReader->locateStart = true;
  // STsdbKeepCfg* pCfg = REPO_KEEP_CFG(pReader->pTsdb);
  // int32_t       fid = getFileIdFromKey(pReader->window.skey, pCfg->days, pCfg->precision);
H
Hongze Cheng 已提交
3464

H
Hongze Cheng 已提交
3465 3466 3467 3468
  // tsdbRLockFS(pFileHandle);
  // tsdbFSIterInit(&pReader->fileIter, pFileHandle, pReader->order);
  // tsdbFSIterSeek(&pReader->fileIter, fid);
  // tsdbUnLockFS(pFileHandle);
H
Hongze Cheng 已提交
3469

H
Hongze Cheng 已提交
3470 3471 3472
  // STsdbCfg* pc = REPO_CFG(pReader->pTsdb);
  // pTableBlockInfo->defMinRows = pc->minRows;
  // pTableBlockInfo->defMaxRows = pc->maxRows;
H
Hongze Cheng 已提交
3473

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

H
Hongze Cheng 已提交
3476
  // pTableBlockInfo->numOfFiles += 1;
H
Hongze Cheng 已提交
3477

H
Hongze Cheng 已提交
3478 3479 3480 3481 3482
  // int32_t     code = TSDB_CODE_SUCCESS;
  // int32_t     numOfBlocks = 0;
  // int32_t     numOfTables = (int32_t)taosArrayGetSize(pReader->pTableCheckInfo);
  // int         defaultRows = 4096;
  // STimeWindow win = TSWINDOW_INITIALIZER;
H
Hongze Cheng 已提交
3483

H
Hongze Cheng 已提交
3484 3485 3486
  // while (true) {
  //   numOfBlocks = 0;
  //   tsdbRLockFS(REPO_FS(pReader->pTsdb));
H
Hongze Cheng 已提交
3487

H
Hongze Cheng 已提交
3488 3489 3490 3491
  //   if ((pReader->pFileGroup = tsdbFSIterNext(&pReader->fileIter)) == NULL) {
  //     tsdbUnLockFS(REPO_FS(pReader->pTsdb));
  //     break;
  //   }
H
Hongze Cheng 已提交
3492

H
Hongze Cheng 已提交
3493
  //   tsdbGetFidKeyRange(pCfg->days, pCfg->precision, pReader->pFileGroup->fid, &win.skey, &win.ekey);
H
Hongze Cheng 已提交
3494

H
Hongze Cheng 已提交
3495 3496 3497 3498 3499 3500 3501 3502
  //   // current file are not overlapped with query time window, ignore remain files
  //   if ((win.skey > pReader->window.ekey) /* || (!ascTraverse && win.ekey < pTsdbReadHandle->window.ekey)*/) {
  //     tsdbUnLockFS(REPO_FS(pReader->pTsdb));
  //     tsdbDebug("%p remain files are not qualified for qrange:%" PRId64 "-%" PRId64 ", ignore, %s", pReader,
  //               pReader->window.skey, pReader->window.ekey, pReader->idStr);
  //     pReader->pFileGroup = NULL;
  //     break;
  //   }
H
Hongze Cheng 已提交
3503

H
Hongze Cheng 已提交
3504 3505 3506 3507 3508 3509
  //   pTableBlockInfo->numOfFiles += 1;
  //   if (tsdbSetAndOpenReadFSet(&pReader->rhelper, pReader->pFileGroup) < 0) {
  //     tsdbUnLockFS(REPO_FS(pReader->pTsdb));
  //     code = terrno;
  //     break;
  //   }
H
Hongze Cheng 已提交
3510

H
Hongze Cheng 已提交
3511
  //   tsdbUnLockFS(REPO_FS(pReader->pTsdb));
H
Hongze Cheng 已提交
3512

H
Hongze Cheng 已提交
3513 3514 3515 3516
  //   if (tsdbLoadBlockIdx(&pReader->rhelper) < 0) {
  //     code = terrno;
  //     break;
  //   }
H
Hongze Cheng 已提交
3517

H
Hongze Cheng 已提交
3518 3519 3520
  //   if ((code = getFileCompInfo(pReader, &numOfBlocks)) != TSDB_CODE_SUCCESS) {
  //     break;
  //   }
H
Hongze Cheng 已提交
3521

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

H
Hongze Cheng 已提交
3525 3526 3527
  //   if (numOfBlocks == 0) {
  //     continue;
  //   }
H
Hongze Cheng 已提交
3528

H
Hongze Cheng 已提交
3529
  //   pTableBlockInfo->numOfBlocks += numOfBlocks;
H
Hongze Cheng 已提交
3530

H
Hongze Cheng 已提交
3531
  //   for (int32_t i = 0; i < numOfTables; ++i) {
H
Haojun Liao 已提交
3532
  //     STableBlockScanInfo* pCheckInfo = taosArrayGet(pReader->pTableCheckInfo, i);
H
Hongze Cheng 已提交
3533

H
Hongze Cheng 已提交
3534
  //     SBlock* pBlock = pCheckInfo->pCompInfo->blocks;
H
Hongze Cheng 已提交
3535

H
Hongze Cheng 已提交
3536 3537
  //     for (int32_t j = 0; j < pCheckInfo->numOfBlocks; ++j) {
  //       pTableBlockInfo->totalSize += pBlock[j].len;
H
refact  
Hongze Cheng 已提交
3538

H
Hongze Cheng 已提交
3539 3540
  //       int32_t numOfRows = pBlock[j].numOfRows;
  //       pTableBlockInfo->totalRows += numOfRows;
H
refact  
Hongze Cheng 已提交
3541

H
Hongze Cheng 已提交
3542 3543 3544
  //       if (numOfRows > pTableBlockInfo->maxRows) {
  //         pTableBlockInfo->maxRows = numOfRows;
  //       }
H
refact  
Hongze Cheng 已提交
3545

H
Hongze Cheng 已提交
3546 3547 3548
  //       if (numOfRows < pTableBlockInfo->minRows) {
  //         pTableBlockInfo->minRows = numOfRows;
  //       }
H
refact  
Hongze Cheng 已提交
3549

H
Hongze Cheng 已提交
3550 3551 3552
  //       if (numOfRows < defaultRows) {
  //         pTableBlockInfo->numOfSmallBlocks += 1;
  //       }
H
refact  
Hongze Cheng 已提交
3553

H
Hongze Cheng 已提交
3554 3555 3556 3557 3558
  //       int32_t bucketIndex = getBucketIndex(pTableBlockInfo->defMinRows, bucketRange, numOfRows);
  //       pTableBlockInfo->blockRowsHisto[bucketIndex]++;
  //     }
  //   }
  // }
H
Hongze Cheng 已提交
3559

H
Hongze Cheng 已提交
3560
  // pTableBlockInfo->numOfTables = numOfTables;
H
refact  
Hongze Cheng 已提交
3561 3562
  return code;
}
H
Hongze Cheng 已提交
3563

H
refact  
Hongze Cheng 已提交
3564
int64_t tsdbGetNumOfRowsInMemTable(STsdbReader* pReader) {
3565
  int64_t rows = 0;
H
Hongze Cheng 已提交
3566

3567 3568
  SReaderStatus* pStatus = &pReader->status;
  pStatus->pTableIter = taosHashIterate(pStatus->pTableMap, NULL);
H
Hongze Cheng 已提交
3569

3570 3571 3572 3573 3574 3575 3576 3577 3578 3579 3580 3581 3582 3583 3584 3585 3586 3587 3588 3589 3590 3591
  while (pStatus->pTableIter != NULL) {
    STableBlockScanInfo* pBlockScanInfo = pStatus->pTableIter;

    STbData* d = NULL;
    if (pReader->pTsdb->mem != NULL) {
      tsdbGetTbDataFromMemTable(pReader->pTsdb->mem, pReader->suid, pBlockScanInfo->uid, &d);
      if (d != NULL) {
        rows += tsdbGetNRowsInTbData(d);
      }
    }

    STbData* di = NULL;
    if (pReader->pTsdb->imem != NULL) {
      tsdbGetTbDataFromMemTable(pReader->pTsdb->imem, pReader->suid, pBlockScanInfo->uid, &di);
      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 已提交
3592

H
refact  
Hongze Cheng 已提交
3593
  return rows;
H
Hongze Cheng 已提交
3594
}