tsdbRead.c 147.7 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"
17

dengyihao's avatar
dengyihao 已提交
18 19
#define EXTRA_BYTES                2
#define ASCENDING_TRAVERSE(o)      (o == TSDB_ORDER_ASC)
20
#define QH_GET_NUM_OF_COLS(handle) ((size_t)(taosArrayGetSize((handle)->pColumns)))
H
hjxilinx 已提交
21

H
Haojun Liao 已提交
22 23 24 25
#define GET_FILE_DATA_BLOCK_INFO(_checkInfo, _block)                                   \
  ((SDataBlockInfo){.window = {.skey = (_block)->keyFirst, .ekey = (_block)->keyLast}, \
                    .numOfCols = (_block)->numOfCols,                                  \
                    .rows = (_block)->numOfRows,                                       \
26
                    .uid = (_checkInfo)->tableId})
H
Haojun Liao 已提交
27

H
hjxilinx 已提交
28
enum {
dengyihao's avatar
dengyihao 已提交
29 30
  TSDB_QUERY_TYPE_ALL = 1,
  TSDB_QUERY_TYPE_LAST = 2,
H
hjxilinx 已提交
31 32
};

33
enum {
dengyihao's avatar
dengyihao 已提交
34
  TSDB_CACHED_TYPE_NONE = 0,
35
  TSDB_CACHED_TYPE_LASTROW = 1,
dengyihao's avatar
dengyihao 已提交
36
  TSDB_CACHED_TYPE_LAST = 2,
37 38
};

39
typedef struct SQueryFilePos {
dengyihao's avatar
dengyihao 已提交
40 41 42 43 44 45 46
  int32_t     fid;
  int32_t     slot;
  int32_t     pos;
  int64_t     lastKey;
  int32_t     rows;
  bool        mixBlock;
  bool        blockCompleted;
47
  STimeWindow win;
48
} SQueryFilePos;
H
hjxilinx 已提交
49

50
typedef struct SDataBlockLoadInfo {
dengyihao's avatar
dengyihao 已提交
51 52 53 54
  SDFileSet* fileGroup;
  int32_t    slot;
  uint64_t   uid;
  SArray*    pLoadedCols;
55
} SDataBlockLoadInfo;
H
hjxilinx 已提交
56

57
typedef struct SLoadCompBlockInfo {
H
hjLiao 已提交
58
  int32_t tid; /* table tid */
59 60
  int32_t fileId;
} SLoadCompBlockInfo;
H
hjxilinx 已提交
61

62
enum {
dengyihao's avatar
dengyihao 已提交
63
  CHECKINFO_CHOSEN_MEM = 0,
64
  CHECKINFO_CHOSEN_IMEM = 1,
dengyihao's avatar
dengyihao 已提交
65
  CHECKINFO_CHOSEN_BOTH = 2  // for update=2(merge case)
66 67
};

68
typedef struct STableCheckInfo {
dengyihao's avatar
dengyihao 已提交
69 70 71 72 73 74 75 76 77
  uint64_t           tableId;
  TSKEY              lastKey;
  SBlockInfo*        pCompInfo;
  int32_t            compSize;
  int32_t            numOfBlocks : 29;  // number of qualified data blocks not the original blocks
  uint8_t            chosen : 2;        // indicate which iterator should move forward
  bool               initBuf : 1;       // whether to initialize the in-memory skip list iterator or not
  SSkipListIterator* iter;              // mem buffer skip list iterator
  SSkipListIterator* iiter;             // imem buffer skip list iterator
78
} STableCheckInfo;
79

80
typedef struct STableBlockInfo {
dengyihao's avatar
dengyihao 已提交
81 82
  SBlock*          compBlock;
  STableCheckInfo* pTableCheckInfo;
83
} STableBlockInfo;
84

85
typedef struct SBlockOrderSupporter {
dengyihao's avatar
dengyihao 已提交
86 87 88 89
  int32_t           numOfTables;
  STableBlockInfo** pDataBlockInfo;
  int32_t*          blockIndexArray;
  int32_t*          numOfBlocksPerTable;
90 91
} SBlockOrderSupporter;

H
Haojun Liao 已提交
92 93 94
typedef struct SIOCostSummary {
  int64_t blockLoadTime;
  int64_t statisInfoLoadTime;
H
Haojun Liao 已提交
95
  int64_t checkForNextTime;
96 97
  int64_t headFileLoad;
  int64_t headFileLoadTime;
H
Haojun Liao 已提交
98 99
} SIOCostSummary;

100
typedef struct SBlockLoadSuppInfo {
C
Cary Xu 已提交
101 102 103 104
  SColumnDataAgg*  pstatis;
  SColumnDataAgg** plist;
  SArray*          defaultLoadColumn;  // default load column
  int32_t*         slotIds;            // colId to slotId
105 106
} SBlockLoadSuppInfo;

107
typedef struct STsdbReadHandle {
C
Cary Xu 已提交
108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126
  STsdb*        pTsdb;
  SQueryFilePos cur;  // current position
  int16_t       order;
  STimeWindow   window;  // the primary query time window that applies to all queries
  //  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
  int32_t numOfBlocks;
  SArray* pColumns;  // column list, SColumnInfoData array list
  bool    locateStart;
  int32_t outputCapacity;
  int32_t realNumOfRows;
  SArray* pTableCheckInfo;  // SArray<STableCheckInfo>
  int32_t activeIndex;
  bool    checkFiles;               // check file stage
  int8_t  cachelastrow;             // check if last row cached
  bool    loadExternalRow;          // load time window external data rows
  bool    currentLoadExternalRows;  // current load external rows
  int32_t loadType;                 // block load type
  char*   idStr;                    // query info handle, for debug purpose
dengyihao's avatar
dengyihao 已提交
127 128 129 130 131
  int32_t type;  // query type: retrieve all data blocks, 2. retrieve only last row, 3. retrieve direct prev|next rows
  SDFileSet*         pFileGroup;
  SFSIter            fileIter;
  SReadH             rhelper;
  STableBlockInfo*   pDataBlockInfo;
C
Cary Xu 已提交
132 133 134 135
  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 */
136
  SBlockLoadSuppInfo suppInfo;
C
Cary Xu 已提交
137 138 139 140
  SArray*            prev;  // previous row which is before than time window
  SArray*            next;  // next row which is after the query time window
  SIOCostSummary     cost;
  STSchema*          pSchema;
141
} STsdbReadHandle;
142

H
Haojun Liao 已提交
143 144 145
typedef struct STableGroupSupporter {
  int32_t    numOfCols;
  SColIndex* pCols;
146
  SSchema*   pTagSchema;
H
Haojun Liao 已提交
147 148
} STableGroupSupporter;

dengyihao's avatar
dengyihao 已提交
149 150 151 152 153
int32_t tsdbQueryTableList(void* pMeta, SArray* pRes, void* filterInfo);

static STimeWindow updateLastrowForEachGroup(STableGroupInfo* groupList);
static int32_t     checkForCachedLastRow(STsdbReadHandle* pTsdbReadHandle, STableGroupInfo* groupList);
static int32_t     checkForCachedLast(STsdbReadHandle* pTsdbReadHandle);
H
Haojun Liao 已提交
154
// static int32_t tsdbGetCachedLastRow(STable* pTable, STSRow** pRes, TSKEY* lastKey);
H
Haojun Liao 已提交
155

H
Haojun Liao 已提交
156
static void    changeQueryHandleForInterpQuery(tsdbReaderT pHandle);
157
static void    doMergeTwoLevelData(STsdbReadHandle* pTsdbReadHandle, STableCheckInfo* pCheckInfo, SBlock* pBlock);
dengyihao's avatar
dengyihao 已提交
158 159
static int32_t tsdbReadRowsFromCache(STableCheckInfo* pCheckInfo, TSKEY maxKey, int maxRowsToRead, STimeWindow* win,
                                     STsdbReadHandle* pTsdbReadHandle);
160
static int32_t tsdbCheckInfoCompar(const void* key1, const void* key2);
dengyihao's avatar
dengyihao 已提交
161 162 163 164
// static int32_t doGetExternalRow(STsdbReadHandle* pTsdbReadHandle, int16_t type, void* pMemRef);
// static void*   doFreeColumnInfoData(SArray* pColumnInfoData);
// static void*   destroyTableCheckInfo(SArray* pTableCheckInfo);
static bool tsdbGetExternalRow(tsdbReaderT pHandle);
Y
TD-1733  
yihaoDeng 已提交
165

C
Cary Xu 已提交
166
static STsdb* getTsdbByRetentions(SVnode* pVnode, STsdbReadHandle* pReadHandle, TSKEY winSKey, SRetention* retentions);
C
Cary Xu 已提交
167

168
static void tsdbInitDataBlockLoadInfo(SDataBlockLoadInfo* pBlockLoadInfo) {
H
hjxilinx 已提交
169
  pBlockLoadInfo->slot = -1;
dengyihao's avatar
dengyihao 已提交
170
  pBlockLoadInfo->uid = 0;
H
hjxilinx 已提交
171
  pBlockLoadInfo->fileGroup = NULL;
H
hjxilinx 已提交
172 173
}

174
static void tsdbInitCompBlockLoadInfo(SLoadCompBlockInfo* pCompBlockLoadInfo) {
H
hjLiao 已提交
175
  pCompBlockLoadInfo->tid = -1;
176 177
  pCompBlockLoadInfo->fileId = -1;
}
H
hjxilinx 已提交
178

179 180
static SArray* getColumnIdList(STsdbReadHandle* pTsdbReadHandle) {
  size_t numOfCols = QH_GET_NUM_OF_COLS(pTsdbReadHandle);
H
Haojun Liao 已提交
181 182 183 184
  assert(numOfCols <= TSDB_MAX_COLUMNS);

  SArray* pIdList = taosArrayInit(numOfCols, sizeof(int16_t));
  for (int32_t i = 0; i < numOfCols; ++i) {
185
    SColumnInfoData* pCol = taosArrayGet(pTsdbReadHandle->pColumns, i);
H
Haojun Liao 已提交
186 187 188 189 190 191
    taosArrayPush(pIdList, &pCol->info.colId);
  }

  return pIdList;
}

192 193
static SArray* getDefaultLoadColumns(STsdbReadHandle* pTsdbReadHandle, bool loadTS) {
  SArray* pLocalIdList = getColumnIdList(pTsdbReadHandle);
H
Haojun Liao 已提交
194 195 196 197 198

  // check if the primary time stamp column needs to load
  int16_t colId = *(int16_t*)taosArrayGet(pLocalIdList, 0);

  // the primary timestamp column does not be included in the the specified load column list, add it
H
Haojun Liao 已提交
199 200
  if (loadTS && colId != PRIMARYKEY_TIMESTAMP_COL_ID) {
    int16_t columnId = PRIMARYKEY_TIMESTAMP_COL_ID;
H
Haojun Liao 已提交
201 202 203 204 205 206
    taosArrayInsert(pLocalIdList, 0, &columnId);
  }

  return pLocalIdList;
}

H
Haojun Liao 已提交
207
int64_t tsdbGetNumOfRowsInMemTable(tsdbReaderT* pHandle) {
dengyihao's avatar
dengyihao 已提交
208
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle;
H
Haojun Liao 已提交
209

dengyihao's avatar
dengyihao 已提交
210 211 212 213 214
  int64_t        rows = 0;
  STsdbMemTable* pMemTable = NULL;  // pTsdbReadHandle->pMemTable;
  if (pMemTable == NULL) {
    return rows;
  }
H
Haojun Liao 已提交
215

dengyihao's avatar
dengyihao 已提交
216 217
  //  STableData* pMem  = NULL;
  //  STableData* pIMem = NULL;
H
Haojun Liao 已提交
218

dengyihao's avatar
dengyihao 已提交
219 220
  //  SMemTable* pMemT = pMemRef->snapshot.mem;
  //  SMemTable* pIMemT = pMemRef->snapshot.imem;
H
Haojun Liao 已提交
221 222 223 224 225

  size_t size = taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
  for (int32_t i = 0; i < size; ++i) {
    STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, i);

dengyihao's avatar
dengyihao 已提交
226 227 228 229 230 231 232 233
    //    if (pMemT && pCheckInfo->tableId < pMemT->maxTables) {
    //      pMem = pMemT->tData[pCheckInfo->tableId];
    //      rows += (pMem && pMem->uid == pCheckInfo->tableId) ? pMem->numOfRows : 0;
    //    }
    //    if (pIMemT && pCheckInfo->tableId < pIMemT->maxTables) {
    //      pIMem = pIMemT->tData[pCheckInfo->tableId];
    //      rows += (pIMem && pIMem->uid == pCheckInfo->tableId) ? pIMem->numOfRows : 0;
    //    }
H
Haojun Liao 已提交
234 235 236
  }
  return rows;
}
237

238 239 240
static SArray* createCheckInfoFromTableGroup(STsdbReadHandle* pTsdbReadHandle, STableGroupInfo* pGroupList) {
  size_t numOfGroup = taosArrayGetSize(pGroupList->pGroupList);
  assert(numOfGroup >= 1);
H
Haojun Liao 已提交
241 242 243 244 245 246 247 248

  // allocate buffer in order to load data blocks from file
  SArray* pTableCheckInfo = taosArrayInit(pGroupList->numOfTables, sizeof(STableCheckInfo));
  if (pTableCheckInfo == NULL) {
    return NULL;
  }

  // todo apply the lastkey of table check to avoid to load header file
249
  for (int32_t i = 0; i < numOfGroup; ++i) {
dengyihao's avatar
dengyihao 已提交
250
    SArray* group = *(SArray**)taosArrayGet(pGroupList->pGroupList, i);
H
Haojun Liao 已提交
251 252 253 254 255

    size_t gsize = taosArrayGetSize(group);
    assert(gsize > 0);

    for (int32_t j = 0; j < gsize; ++j) {
dengyihao's avatar
dengyihao 已提交
256
      STableKeyInfo* pKeyInfo = (STableKeyInfo*)taosArrayGet(group, j);
H
Haojun Liao 已提交
257

dengyihao's avatar
dengyihao 已提交
258
      STableCheckInfo info = {.lastKey = pKeyInfo->lastKey, .tableId = pKeyInfo->uid};
259 260 261
      if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
        if (info.lastKey == INT64_MIN || info.lastKey < pTsdbReadHandle->window.skey) {
          info.lastKey = pTsdbReadHandle->window.skey;
262 263
        }

264
        assert(info.lastKey >= pTsdbReadHandle->window.skey && info.lastKey <= pTsdbReadHandle->window.ekey);
H
Haojun Liao 已提交
265
      } else {
266
        info.lastKey = pTsdbReadHandle->window.skey;
H
Haojun Liao 已提交
267 268 269
      }

      taosArrayPush(pTableCheckInfo, &info);
dengyihao's avatar
dengyihao 已提交
270 271
      tsdbDebug("%p check table uid:%" PRId64 " from lastKey:%" PRId64 " %s", pTsdbReadHandle, info.tableId,
                info.lastKey, pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
272 273 274
    }
  }

275
  // TODO  group table according to the tag value.
276
  taosArraySort(pTableCheckInfo, tsdbCheckInfoCompar);
H
Haojun Liao 已提交
277 278 279
  return pTableCheckInfo;
}

280 281
static void resetCheckInfo(STsdbReadHandle* pTsdbReadHandle) {
  size_t numOfTables = taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
H
Haojun Liao 已提交
282 283 284 285
  assert(numOfTables >= 1);

  // todo apply the lastkey of table check to avoid to load header file
  for (int32_t i = 0; i < numOfTables; ++i) {
dengyihao's avatar
dengyihao 已提交
286
    STableCheckInfo* pCheckInfo = (STableCheckInfo*)taosArrayGet(pTsdbReadHandle->pTableCheckInfo, i);
287
    pCheckInfo->lastKey = pTsdbReadHandle->window.skey;
dengyihao's avatar
dengyihao 已提交
288 289
    pCheckInfo->iter = tSkipListDestroyIter(pCheckInfo->iter);
    pCheckInfo->iiter = tSkipListDestroyIter(pCheckInfo->iiter);
290
    pCheckInfo->initBuf = false;
H
Haojun Liao 已提交
291

292 293
    if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
      assert(pCheckInfo->lastKey >= pTsdbReadHandle->window.skey);
294
    } else {
295
      assert(pCheckInfo->lastKey <= pTsdbReadHandle->window.skey);
296
    }
H
Haojun Liao 已提交
297 298 299
  }
}

H
Haojun Liao 已提交
300 301 302
// only one table, not need to sort again
static SArray* createCheckInfoFromCheckInfo(STableCheckInfo* pCheckInfo, TSKEY skey, SArray** psTable) {
  SArray* pNew = taosArrayInit(1, sizeof(STableCheckInfo));
D
fix bug  
dapan1121 已提交
303

dengyihao's avatar
dengyihao 已提交
304
  STableCheckInfo info = {.lastKey = skey};
H
Haojun Liao 已提交
305

H
Haojun Liao 已提交
306 307
  info.tableId = pCheckInfo->tableId;
  taosArrayPush(pNew, &info);
H
Haojun Liao 已提交
308 309 310
  return pNew;
}

311 312
static bool emptyQueryTimewindow(STsdbReadHandle* pTsdbReadHandle) {
  assert(pTsdbReadHandle != NULL);
313

314
  STimeWindow* w = &pTsdbReadHandle->window;
dengyihao's avatar
dengyihao 已提交
315
  bool         asc = ASCENDING_TRAVERSE(pTsdbReadHandle->order);
316 317 318 319

  return ((asc && w->skey > w->ekey) || (!asc && w->ekey > w->skey));
}

320 321
// 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.
322
static int64_t getEarliestValidTimestamp(STsdb* pTsdb) {
C
Cary Xu 已提交
323
  STsdbKeepCfg* pCfg = REPO_KEEP_CFG(pTsdb);
324 325

  int64_t now = taosGetTimestamp(pCfg->precision);
326
  return now - (tsTickPerMin[pCfg->precision] * pCfg->keep2) + 1;  // needs to add one tick
327 328
}

329
static void setQueryTimewindow(STsdbReadHandle* pTsdbReadHandle, SQueryTableDataCond* pCond) {
330
  pTsdbReadHandle->window = pCond->twindow;
331

332
  bool    updateTs = false;
333 334 335 336
  int64_t startTs = getEarliestValidTimestamp(pTsdbReadHandle->pTsdb);
  if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
    if (startTs > pTsdbReadHandle->window.skey) {
      pTsdbReadHandle->window.skey = startTs;
337 338
      pCond->twindow.skey = startTs;
      updateTs = true;
339 340
    }
  } else {
341 342
    if (startTs > pTsdbReadHandle->window.ekey) {
      pTsdbReadHandle->window.ekey = startTs;
343 344
      pCond->twindow.ekey = startTs;
      updateTs = true;
345 346 347
    }
  }

348
  if (updateTs) {
H
Haojun Liao 已提交
349 350 351
    tsdbDebug("%p update the query time window, old:%" PRId64 " - %" PRId64 ", new:%" PRId64 " - %" PRId64 ", %s",
              pTsdbReadHandle, pCond->twindow.skey, pCond->twindow.ekey, pTsdbReadHandle->window.skey,
              pTsdbReadHandle->window.ekey, pTsdbReadHandle->idStr);
352
  }
353
}
C
Cary Xu 已提交
354

C
Cary Xu 已提交
355
static STsdb* getTsdbByRetentions(SVnode* pVnode, STsdbReadHandle* pReadHandle, TSKEY winSKey, SRetention* retentions) {
C
Cary Xu 已提交
356
  if (vnodeIsRollup(pVnode)) {
C
Cary Xu 已提交
357
    int     level = 0;
C
Cary Xu 已提交
358
    int64_t now = taosGetTimestamp(pVnode->config.tsdbCfg.precision);
C
Cary Xu 已提交
359

C
Cary Xu 已提交
360
    for (int i = 0; i < TSDB_RETENTION_MAX; ++i) {
C
Cary Xu 已提交
361 362 363 364 365
      SRetention* pRetention = retentions + level;
      if (pRetention->keep <= 0) {
        if (level > 0) {
          --level;
        }
C
Cary Xu 已提交
366 367
        break;
      }
C
Cary Xu 已提交
368
      if ((now - pRetention->keep) <= winSKey) {
C
Cary Xu 已提交
369
        break;
C
Cary Xu 已提交
370 371
      }
      ++level;
C
Cary Xu 已提交
372
    }
C
Cary Xu 已提交
373

C
Cary Xu 已提交
374
    if (level == TSDB_RETENTION_L0) {
C
Cary Xu 已提交
375
      tsdbDebug("vgId:%d read handle %p rsma level %d is selected to query", TD_VID(pVnode), pReadHandle, TSDB_RETENTION_L0);
C
Cary Xu 已提交
376 377
      return VND_RSMA0(pVnode);
    } else if (level == TSDB_RETENTION_L1) {
C
Cary Xu 已提交
378
      tsdbDebug("vgId:%d read handle %p rsma level %d is selected to query", TD_VID(pVnode), pReadHandle, TSDB_RETENTION_L1);
C
Cary Xu 已提交
379 380
      return VND_RSMA1(pVnode);
    } else {
C
Cary Xu 已提交
381
      tsdbDebug("vgId:%d read handle %p rsma level %d is selected to query", TD_VID(pVnode), pReadHandle, TSDB_RETENTION_L2);
C
Cary Xu 已提交
382 383
      return VND_RSMA2(pVnode);
    }
C
Cary Xu 已提交
384
  }
C
Cary Xu 已提交
385
  return VND_TSDB(pVnode);
C
Cary Xu 已提交
386 387
}

388
static STsdbReadHandle* tsdbQueryTablesImpl(SVnode* pVnode, SQueryTableDataCond* pCond, uint64_t qId, uint64_t taskId) {
wafwerar's avatar
wafwerar 已提交
389
  STsdbReadHandle* pReadHandle = taosMemoryCalloc(1, sizeof(STsdbReadHandle));
390
  if (pReadHandle == NULL) {
391
    goto _end;
392
  }
H
Haojun Liao 已提交
393

C
Cary Xu 已提交
394
  STsdb* pTsdb = getTsdbByRetentions(pVnode, pReadHandle, pCond->twindow.skey, pVnode->config.tsdbCfg.retentions);
C
Cary Xu 已提交
395

dengyihao's avatar
dengyihao 已提交
396
  pReadHandle->order = pCond->order;
C
Cary Xu 已提交
397
  pReadHandle->pTsdb = pTsdb;
dengyihao's avatar
dengyihao 已提交
398 399 400 401 402 403
  pReadHandle->type = TSDB_QUERY_TYPE_ALL;
  pReadHandle->cur.fid = INT32_MIN;
  pReadHandle->cur.win = TSWINDOW_INITIALIZER;
  pReadHandle->checkFiles = true;
  pReadHandle->activeIndex = 0;  // current active table index
  pReadHandle->allocSize = 0;
404
  pReadHandle->locateStart = false;
dengyihao's avatar
dengyihao 已提交
405
  pReadHandle->loadType = pCond->type;
406

dengyihao's avatar
dengyihao 已提交
407
  pReadHandle->outputCapacity = 4096;  //((STsdb*)tsdb)->config.maxRowsPerFileBlock;
408 409 410
  pReadHandle->loadExternalRow = pCond->loadExternalRows;
  pReadHandle->currentLoadExternalRows = pCond->loadExternalRows;

H
Haojun Liao 已提交
411
  char buf[128] = {0};
dengyihao's avatar
dengyihao 已提交
412
  snprintf(buf, tListLen(buf), "TID:0x%" PRIx64 " QID:0x%" PRIx64, taskId, qId);
H
Haojun Liao 已提交
413 414
  pReadHandle->idStr = strdup(buf);

C
Cary Xu 已提交
415
  if (tsdbInitReadH(&pReadHandle->rhelper, pReadHandle->pTsdb) != 0) {
416
    goto _end;
B
Bomin Zhang 已提交
417
  }
H
Haojun Liao 已提交
418

419 420
  assert(pCond != NULL);
  setQueryTimewindow(pReadHandle, pCond);
421

422
  if (pCond->numOfCols > 0) {
H
Haojun Liao 已提交
423 424 425 426 427
    int32_t rowLen = 0;
    for(int32_t i = 0; i < pCond->numOfCols; ++i) {
      rowLen += pCond->colList[i].bytes;
    }

428 429 430 431 432 433
    // make sure the output SSDataBlock size be less than 2MB.
    int32_t TWOMB = 2 * 1024 * 1024;
    if (pReadHandle->outputCapacity * rowLen > TWOMB) {
      pReadHandle->outputCapacity = TWOMB / rowLen;
    }

434
    // allocate buffer in order to load data blocks from file
435 436
    pReadHandle->suppInfo.pstatis = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnDataAgg));
    if (pReadHandle->suppInfo.pstatis == NULL) {
437
      goto _end;
438
    }
H
Haojun Liao 已提交
439

440
    // todo: use list instead of array?
441 442
    pReadHandle->pColumns = taosArrayInit(pCond->numOfCols, sizeof(SColumnInfoData));
    if (pReadHandle->pColumns == NULL) {
443
      goto _end;
444
    }
H
Haojun Liao 已提交
445

446 447 448
    for (int32_t i = 0; i < pCond->numOfCols; ++i) {
      SColumnInfoData colInfo = {{0}, 0};
      colInfo.info = pCond->colList[i];
H
Haojun Liao 已提交
449

450
      int32_t code = colInfoDataEnsureCapacity(&colInfo, 0, pReadHandle->outputCapacity);
451
      if (code != TSDB_CODE_SUCCESS) {
452
        goto _end;
453
      }
454

455
      taosArrayPush(pReadHandle->pColumns, &colInfo);
B
Bomin Zhang 已提交
456
    }
H
Haojun Liao 已提交
457

458
    pReadHandle->suppInfo.defaultLoadColumn = getDefaultLoadColumns(pReadHandle, true);
C
Cary Xu 已提交
459 460 461 462
    pReadHandle->suppInfo.slotIds =
        taosMemoryMalloc(sizeof(int32_t) * taosArrayGetSize(pReadHandle->suppInfo.defaultLoadColumn));
    pReadHandle->suppInfo.plist =
        taosMemoryCalloc(taosArrayGetSize(pReadHandle->suppInfo.defaultLoadColumn), POINTER_BYTES);
H
Haojun Liao 已提交
463
  }
464

C
Cary Xu 已提交
465
  pReadHandle->pDataCols = tdNewDataCols(1000, pVnode->config.tsdbCfg.maxRows);
466
  if (pReadHandle->pDataCols == NULL) {
H
Haojun Liao 已提交
467
    tsdbError("%p failed to malloc buf for pDataCols, %s", pReadHandle, pReadHandle->idStr);
H
Haojun Liao 已提交
468
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
469
    goto _end;
H
hjxilinx 已提交
470
  }
471

472 473
  tsdbInitDataBlockLoadInfo(&pReadHandle->dataBlockLoadInfo);
  tsdbInitCompBlockLoadInfo(&pReadHandle->compBlockLoadInfo);
474

H
Haojun Liao 已提交
475
  return (tsdbReaderT)pReadHandle;
476

dengyihao's avatar
dengyihao 已提交
477
_end:
478
  tsdbCleanupReadHandle(pReadHandle);
479
  terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
480
  return NULL;
H
hjxilinx 已提交
481 482
}

483
tsdbReaderT* tsdbQueryTables(SVnode* pVnode, SQueryTableDataCond* pCond, STableGroupInfo* groupList, uint64_t qId,
dengyihao's avatar
dengyihao 已提交
484
                             uint64_t taskId) {
485
  STsdbReadHandle* pTsdbReadHandle = tsdbQueryTablesImpl(pVnode, pCond, qId, taskId);
486
  if (pTsdbReadHandle == NULL) {
487 488 489
    return NULL;
  }

490
  if (emptyQueryTimewindow(pTsdbReadHandle)) {
dengyihao's avatar
dengyihao 已提交
491
    return (tsdbReaderT*)pTsdbReadHandle;
492
  }
H
Haojun Liao 已提交
493 494

  // todo apply the lastkey of table check to avoid to load header file
495
  pTsdbReadHandle->pTableCheckInfo = createCheckInfoFromTableGroup(pTsdbReadHandle, groupList);
496
  if (pTsdbReadHandle->pTableCheckInfo == NULL) {
dengyihao's avatar
dengyihao 已提交
497
    //    tsdbCleanupReadHandle(pTsdbReadHandle);
H
Haojun Liao 已提交
498 499 500 501
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    return NULL;
  }

C
Cary Xu 已提交
502
  STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, 0);
503

H
Hongze Cheng 已提交
504
  pTsdbReadHandle->pSchema = metaGetTbTSchema(pVnode->pMeta, pCheckInfo->tableId, 1);
C
Cary Xu 已提交
505
  int32_t  numOfCols = taosArrayGetSize(pTsdbReadHandle->suppInfo.defaultLoadColumn);
506 507 508 509 510
  int16_t* ids = pTsdbReadHandle->suppInfo.defaultLoadColumn->pData;

  STSchema* pSchema = pTsdbReadHandle->pSchema;

  int32_t i = 0, j = 0;
C
Cary Xu 已提交
511
  while (i < numOfCols && j < pSchema->numOfCols) {
512 513 514 515 516 517 518 519 520 521 522 523
    if (ids[i] == pSchema->columns[j].colId) {
      pTsdbReadHandle->suppInfo.slotIds[i] = j;
      i++;
      j++;
    } else if (ids[i] > pSchema->columns[j].colId) {
      j++;
    } else {
      //    tsdbCleanupReadHandle(pTsdbReadHandle);
      terrno = TSDB_CODE_INVALID_PARA;
      return NULL;
    }
  }
524

dengyihao's avatar
dengyihao 已提交
525 526 527
  tsdbDebug("%p total numOfTable:%" PRIzu " in this query, group %" PRIzu " %s", pTsdbReadHandle,
            taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo), taosArrayGetSize(groupList->pGroupList),
            pTsdbReadHandle->idStr);
528

dengyihao's avatar
dengyihao 已提交
529
  return (tsdbReaderT)pTsdbReadHandle;
H
Haojun Liao 已提交
530 531
}

532
void tsdbResetReadHandle(tsdbReaderT queryHandle, SQueryTableDataCond* pCond) {
533
  STsdbReadHandle* pTsdbReadHandle = queryHandle;
H
Haojun Liao 已提交
534

535 536 537
  if (emptyQueryTimewindow(pTsdbReadHandle)) {
    if (pCond->order != pTsdbReadHandle->order) {
      pTsdbReadHandle->order = pCond->order;
wafwerar's avatar
wafwerar 已提交
538
      TSWAP(pTsdbReadHandle->window.skey, pTsdbReadHandle->window.ekey);
539 540 541 542 543
    }

    return;
  }

dengyihao's avatar
dengyihao 已提交
544 545 546 547 548 549 550
  pTsdbReadHandle->order = pCond->order;
  pTsdbReadHandle->window = pCond->twindow;
  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
551 552
  pTsdbReadHandle->locateStart = false;
  pTsdbReadHandle->loadExternalRow = pCond->loadExternalRows;
H
Haojun Liao 已提交
553 554

  if (ASCENDING_TRAVERSE(pCond->order)) {
555
    assert(pTsdbReadHandle->window.skey <= pTsdbReadHandle->window.ekey);
H
Haojun Liao 已提交
556
  } else {
557
    assert(pTsdbReadHandle->window.skey >= pTsdbReadHandle->window.ekey);
H
Haojun Liao 已提交
558 559 560
  }

  // allocate buffer in order to load data blocks from file
561 562
  memset(pTsdbReadHandle->suppInfo.pstatis, 0, sizeof(SColumnDataAgg));
  memset(pTsdbReadHandle->suppInfo.plist, 0, POINTER_BYTES);
H
Haojun Liao 已提交
563

564 565
  tsdbInitDataBlockLoadInfo(&pTsdbReadHandle->dataBlockLoadInfo);
  tsdbInitCompBlockLoadInfo(&pTsdbReadHandle->compBlockLoadInfo);
H
Haojun Liao 已提交
566

567
  resetCheckInfo(pTsdbReadHandle);
H
Haojun Liao 已提交
568 569
}

570
void tsdbResetQueryHandleForNewTable(tsdbReaderT queryHandle, SQueryTableDataCond* pCond, STableGroupInfo* groupList) {
571
  STsdbReadHandle* pTsdbReadHandle = queryHandle;
H
Haojun Liao 已提交
572

dengyihao's avatar
dengyihao 已提交
573 574 575 576 577 578 579
  pTsdbReadHandle->order = pCond->order;
  pTsdbReadHandle->window = pCond->twindow;
  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
580 581
  pTsdbReadHandle->locateStart = false;
  pTsdbReadHandle->loadExternalRow = pCond->loadExternalRows;
H
Haojun Liao 已提交
582 583

  if (ASCENDING_TRAVERSE(pCond->order)) {
584
    assert(pTsdbReadHandle->window.skey <= pTsdbReadHandle->window.ekey);
H
Haojun Liao 已提交
585
  } else {
586
    assert(pTsdbReadHandle->window.skey >= pTsdbReadHandle->window.ekey);
H
Haojun Liao 已提交
587 588 589
  }

  // allocate buffer in order to load data blocks from file
590 591
  memset(pTsdbReadHandle->suppInfo.pstatis, 0, sizeof(SColumnDataAgg));
  memset(pTsdbReadHandle->suppInfo.plist, 0, POINTER_BYTES);
H
Haojun Liao 已提交
592

593 594
  tsdbInitDataBlockLoadInfo(&pTsdbReadHandle->dataBlockLoadInfo);
  tsdbInitCompBlockLoadInfo(&pTsdbReadHandle->compBlockLoadInfo);
H
Haojun Liao 已提交
595

H
Haojun Liao 已提交
596
  SArray* pTable = NULL;
dengyihao's avatar
dengyihao 已提交
597
  //  STsdbMeta* pMeta = tsdbGetMeta(pTsdbReadHandle->pTsdb);
H
Haojun Liao 已提交
598

dengyihao's avatar
dengyihao 已提交
599
  //  pTsdbReadHandle->pTableCheckInfo = destroyTableCheckInfo(pTsdbReadHandle->pTableCheckInfo);
H
Haojun Liao 已提交
600

dengyihao's avatar
dengyihao 已提交
601 602
  pTsdbReadHandle->pTableCheckInfo = NULL;  // createCheckInfoFromTableGroup(pTsdbReadHandle, groupList, pMeta,
                                            // &pTable);
603
  if (pTsdbReadHandle->pTableCheckInfo == NULL) {
dengyihao's avatar
dengyihao 已提交
604
    //    tsdbCleanupReadHandle(pTsdbReadHandle);
H
Haojun Liao 已提交
605 606
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
  }
H
Haojun Liao 已提交
607

dengyihao's avatar
dengyihao 已提交
608 609
  //  pTsdbReadHandle->prev = doFreeColumnInfoData(pTsdbReadHandle->prev);
  //  pTsdbReadHandle->next = doFreeColumnInfoData(pTsdbReadHandle->next);
H
Haojun Liao 已提交
610 611
}

612
tsdbReaderT tsdbQueryLastRow(SVnode* pVnode, SQueryTableDataCond* pCond, STableGroupInfo* groupList, uint64_t qId,
dengyihao's avatar
dengyihao 已提交
613
                             uint64_t taskId) {
614
  pCond->twindow = updateLastrowForEachGroup(groupList);
H
Haojun Liao 已提交
615 616 617 618 619 620

  // no qualified table
  if (groupList->numOfTables == 0) {
    return NULL;
  }

621
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)tsdbQueryTables(pVnode, pCond, groupList, qId, taskId);
622
  if (pTsdbReadHandle == NULL) {
623 624 625
    return NULL;
  }

626
  int32_t code = checkForCachedLastRow(pTsdbReadHandle, groupList);
dengyihao's avatar
dengyihao 已提交
627
  if (code != TSDB_CODE_SUCCESS) {  // set the numOfTables to be 0
H
Haojun Liao 已提交
628 629 630
    terrno = code;
    return NULL;
  }
H
Haojun Liao 已提交
631 632

  assert(pCond->order == TSDB_ORDER_ASC && pCond->twindow.skey <= pCond->twindow.ekey);
633 634
  if (pTsdbReadHandle->cachelastrow) {
    pTsdbReadHandle->type = TSDB_QUERY_TYPE_LAST;
D
init  
dapan1121 已提交
635
  }
dengyihao's avatar
dengyihao 已提交
636

637
  return pTsdbReadHandle;
D
init  
dapan1121 已提交
638 639
}

640
#if 0
641 642
tsdbReaderT tsdbQueryCacheLastT(STsdb *tsdb, SQueryTableDataCond *pCond, STableGroupInfo *groupList, uint64_t qId, STsdbMemTable* pMemRef) {
  STsdbReadHandle *pTsdbReadHandle = (STsdbReadHandle*) tsdbQueryTablesT(tsdb, pCond, groupList, qId, pMemRef);
643
  if (pTsdbReadHandle == NULL) {
644 645 646
    return NULL;
  }

647
  int32_t code = checkForCachedLast(pTsdbReadHandle);
D
init  
dapan1121 已提交
648 649 650 651 652
  if (code != TSDB_CODE_SUCCESS) { // set the numOfTables to be 0
    terrno = code;
    return NULL;
  }

653 654
  if (pTsdbReadHandle->cachelastrow) {
    pTsdbReadHandle->type = TSDB_QUERY_TYPE_LAST;
D
fix bug  
dapan1121 已提交
655
  }
D
init  
dapan1121 已提交
656
  
657
  return pTsdbReadHandle;
H
hjxilinx 已提交
658 659
}

660
#endif
dengyihao's avatar
dengyihao 已提交
661
SArray* tsdbGetQueriedTableList(tsdbReaderT* pHandle) {
662
  assert(pHandle != NULL);
H
Haojun Liao 已提交
663

dengyihao's avatar
dengyihao 已提交
664
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle;
H
Haojun Liao 已提交
665

dengyihao's avatar
dengyihao 已提交
666
  size_t  size = taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
667
  SArray* res = taosArrayInit(size, POINTER_BYTES);
668 669 670
  return res;
}

H
Haojun Liao 已提交
671 672 673 674 675
// leave only one table for each group
static STableGroupInfo* trimTableGroup(STimeWindow* window, STableGroupInfo* pGroupList) {
  assert(pGroupList);
  size_t numOfGroup = taosArrayGetSize(pGroupList->pGroupList);

wafwerar's avatar
wafwerar 已提交
676
  STableGroupInfo* pNew = taosMemoryCalloc(1, sizeof(STableGroupInfo));
Y
yihaoDeng 已提交
677
  pNew->pGroupList = taosArrayInit(numOfGroup, POINTER_BYTES);
H
Haojun Liao 已提交
678

dengyihao's avatar
dengyihao 已提交
679
  for (int32_t i = 0; i < numOfGroup; ++i) {
H
Haojun Liao 已提交
680
    SArray* oneGroup = taosArrayGetP(pGroupList->pGroupList, i);
dengyihao's avatar
dengyihao 已提交
681
    size_t  numOfTables = taosArrayGetSize(oneGroup);
H
Haojun Liao 已提交
682 683 684 685

    SArray* px = taosArrayInit(4, sizeof(STableKeyInfo));
    for (int32_t j = 0; j < numOfTables; ++j) {
      STableKeyInfo* pInfo = (STableKeyInfo*)taosArrayGet(oneGroup, j);
dengyihao's avatar
dengyihao 已提交
686 687 688 689 690
      //      if (window->skey <= pInfo->lastKey && ((STable*)pInfo->pTable)->lastKey != TSKEY_INITIAL_VAL) {
      //        taosArrayPush(px, pInfo);
      //        pNew->numOfTables += 1;
      //        break;
      //      }
H
Haojun Liao 已提交
691 692 693 694 695 696 697 698 699 700 701 702 703
    }

    // there are no data in this group
    if (taosArrayGetSize(px) == 0) {
      taosArrayDestroy(px);
    } else {
      taosArrayPush(pNew->pGroupList, &px);
    }
  }

  return pNew;
}

704
tsdbReaderT tsdbQueryRowsInExternalWindow(SVnode* pVnode, SQueryTableDataCond* pCond, STableGroupInfo* groupList,
705
                                          uint64_t qId, uint64_t taskId) {
H
Haojun Liao 已提交
706 707
  STableGroupInfo* pNew = trimTableGroup(&pCond->twindow, groupList);

708 709 710 711 712 713 714 715 716 717 718 719
  if (pNew->numOfTables == 0) {
    tsdbDebug("update query time range to invalidate time window");

    assert(taosArrayGetSize(pNew->pGroupList) == 0);
    bool asc = ASCENDING_TRAVERSE(pCond->order);
    if (asc) {
      pCond->twindow.ekey = pCond->twindow.skey - 1;
    } else {
      pCond->twindow.skey = pCond->twindow.ekey - 1;
    }
  }

720
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)tsdbQueryTables(pVnode, pCond, pNew, qId, taskId);
721 722
  pTsdbReadHandle->loadExternalRow = true;
  pTsdbReadHandle->currentLoadExternalRows = true;
723

724
  return pTsdbReadHandle;
725 726
}

727
static bool initTableMemIterator(STsdbReadHandle* pHandle, STableCheckInfo* pCheckInfo) {
728
  if (pCheckInfo->initBuf) {
729 730
    return true;
  }
H
Haojun Liao 已提交
731

732
  pCheckInfo->initBuf = true;
733
  int32_t order = pHandle->order;
H
Haojun Liao 已提交
734

735 736 737
  STbData** pMem = NULL;
  STbData** pIMem = NULL;

738
  TSKEY tLastKey = keyToTkey(pCheckInfo->lastKey);
739 740 741
  if (pHandle->pTsdb->mem != NULL) {
    pMem = taosHashGet(pHandle->pTsdb->mem->pHashIdx, &pCheckInfo->tableId, sizeof(pCheckInfo->tableId));
    if (pMem != NULL) {
H
Haojun Liao 已提交
742
      pCheckInfo->iter =
743
          tSkipListCreateIterFromVal((*pMem)->pData, (const char*)&tLastKey, TSDB_DATA_TYPE_TIMESTAMP, order);
H
Haojun Liao 已提交
744
    }
745
  }
H
Haojun Liao 已提交
746

747 748 749
  if (pHandle->pTsdb->imem != NULL) {
    pIMem = taosHashGet(pHandle->pTsdb->imem->pHashIdx, &pCheckInfo->tableId, sizeof(pCheckInfo->tableId));
    if (pIMem != NULL) {
H
Haojun Liao 已提交
750
      pCheckInfo->iiter =
751
          tSkipListCreateIterFromVal((*pIMem)->pData, (const char*)&tLastKey, TSDB_DATA_TYPE_TIMESTAMP, order);
H
Haojun Liao 已提交
752
    }
753
  }
H
Haojun Liao 已提交
754

755 756 757 758
  // both iterators are NULL, no data in buffer right now
  if (pCheckInfo->iter == NULL && pCheckInfo->iiter == NULL) {
    return false;
  }
H
Haojun Liao 已提交
759

dengyihao's avatar
dengyihao 已提交
760
  bool memEmpty = (pCheckInfo->iter == NULL) || (pCheckInfo->iter != NULL && !tSkipListIterNext(pCheckInfo->iter));
761
  bool imemEmpty = (pCheckInfo->iiter == NULL) || (pCheckInfo->iiter != NULL && !tSkipListIterNext(pCheckInfo->iiter));
dengyihao's avatar
dengyihao 已提交
762
  if (memEmpty && imemEmpty) {  // buffer is empty
763 764
    return false;
  }
H
Haojun Liao 已提交
765

766 767 768
  if (!memEmpty) {
    SSkipListNode* node = tSkipListIterGet(pCheckInfo->iter);
    assert(node != NULL);
H
Haojun Liao 已提交
769

H
Haojun Liao 已提交
770 771
    STSRow* row = (STSRow*)SL_GET_NODE_DATA(node);
    TSKEY   key = TD_ROW_KEY(row);  // first timestamp in buffer
772
    tsdbDebug("%p uid:%" PRId64 ", check data in mem from skey:%" PRId64 ", order:%d, ts range in buf:%" PRId64
dengyihao's avatar
dengyihao 已提交
773 774 775
              "-%" PRId64 ", lastKey:%" PRId64 ", numOfRows:%" PRId64 ", %s",
              pHandle, pCheckInfo->tableId, key, order, (*pMem)->keyMin, (*pMem)->keyMax, pCheckInfo->lastKey,
              (*pMem)->nrows, pHandle->idStr);
H
Haojun Liao 已提交
776 777 778 779 780 781 782

    if (ASCENDING_TRAVERSE(order)) {
      assert(pCheckInfo->lastKey <= key);
    } else {
      assert(pCheckInfo->lastKey >= key);
    }

783
  } else {
dengyihao's avatar
dengyihao 已提交
784
    tsdbDebug("%p uid:%" PRId64 ", no data in mem, %s", pHandle, pCheckInfo->tableId, pHandle->idStr);
785
  }
H
Haojun Liao 已提交
786

787 788 789
  if (!imemEmpty) {
    SSkipListNode* node = tSkipListIterGet(pCheckInfo->iiter);
    assert(node != NULL);
H
Haojun Liao 已提交
790

H
Haojun Liao 已提交
791 792
    STSRow* row = (STSRow*)SL_GET_NODE_DATA(node);
    TSKEY   key = TD_ROW_KEY(row);  // first timestamp in buffer
793
    tsdbDebug("%p uid:%" PRId64 ", check data in imem from skey:%" PRId64 ", order:%d, ts range in buf:%" PRId64
dengyihao's avatar
dengyihao 已提交
794 795 796
              "-%" PRId64 ", lastKey:%" PRId64 ", numOfRows:%" PRId64 ", %s",
              pHandle, pCheckInfo->tableId, key, order, (*pIMem)->keyMin, (*pIMem)->keyMax, pCheckInfo->lastKey,
              (*pIMem)->nrows, pHandle->idStr);
H
Haojun Liao 已提交
797 798 799 800 801 802

    if (ASCENDING_TRAVERSE(order)) {
      assert(pCheckInfo->lastKey <= key);
    } else {
      assert(pCheckInfo->lastKey >= key);
    }
803
  } else {
dengyihao's avatar
dengyihao 已提交
804
    tsdbDebug("%p uid:%" PRId64 ", no data in imem, %s", pHandle, pCheckInfo->tableId, pHandle->idStr);
805
  }
H
Haojun Liao 已提交
806

807 808 809
  return true;
}

H
Haojun Liao 已提交
810 811 812 813 814
static void destroyTableMemIterator(STableCheckInfo* pCheckInfo) {
  tSkipListDestroyIter(pCheckInfo->iter);
  tSkipListDestroyIter(pCheckInfo->iiter);
}

C
Cary Xu 已提交
815
static TSKEY extractFirstTraverseKey(STableCheckInfo* pCheckInfo, int32_t order, int32_t update, TDRowVerT maxVer) {
H
Haojun Liao 已提交
816
  STSRow *rmem = NULL, *rimem = NULL;
817 818 819
  if (pCheckInfo->iter) {
    SSkipListNode* node = tSkipListIterGet(pCheckInfo->iter);
    if (node != NULL) {
H
Haojun Liao 已提交
820
      rmem = (STSRow*)SL_GET_NODE_DATA(node);
C
Cary Xu 已提交
821 822 823 824
      // TODO: filter max version
      // if (TD_ROW_VER(rmem) > maxVer) {
      //   rmem = NULL;
      // }
825 826 827 828 829 830
    }
  }

  if (pCheckInfo->iiter) {
    SSkipListNode* node = tSkipListIterGet(pCheckInfo->iiter);
    if (node != NULL) {
H
Haojun Liao 已提交
831
      rimem = (STSRow*)SL_GET_NODE_DATA(node);
C
Cary Xu 已提交
832 833 834 835
      // TODO: filter max version
      // if (TD_ROW_VER(rimem) > maxVer) {
      //   rimem = NULL;
      // }
836 837 838 839 840 841 842 843 844
    }
  }

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

  if (rmem != NULL && rimem == NULL) {
    pCheckInfo->chosen = CHECKINFO_CHOSEN_MEM;
H
Haojun Liao 已提交
845
    return TD_ROW_KEY(rmem);
846 847 848 849
  }

  if (rmem == NULL && rimem != NULL) {
    pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
H
Haojun Liao 已提交
850
    return TD_ROW_KEY(rimem);
851 852
  }

H
Haojun Liao 已提交
853 854
  TSKEY r1 = TD_ROW_KEY(rmem);
  TSKEY r2 = TD_ROW_KEY(rimem);
855 856

  if (r1 == r2) {
C
Cary Xu 已提交
857
#if 0
dengyihao's avatar
dengyihao 已提交
858
    if (update == TD_ROW_DISCARD_UPDATE) {
859 860
      pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
      tSkipListIterNext(pCheckInfo->iter);
dengyihao's avatar
dengyihao 已提交
861
    } else if (update == TD_ROW_OVERWRITE_UPDATE) {
862 863 864 865 866
      pCheckInfo->chosen = CHECKINFO_CHOSEN_MEM;
      tSkipListIterNext(pCheckInfo->iiter);
    } else {
      pCheckInfo->chosen = CHECKINFO_CHOSEN_BOTH;
    }
C
Cary Xu 已提交
867 868 869 870 871 872 873
#endif
    if (TD_SUPPORT_UPDATE(update)) {
      pCheckInfo->chosen = CHECKINFO_CHOSEN_BOTH;
    } else {
      pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
      tSkipListIterNext(pCheckInfo->iter);
    }
874 875 876 877
    return r1;
  } else if (r1 < r2 && ASCENDING_TRAVERSE(order)) {
    pCheckInfo->chosen = CHECKINFO_CHOSEN_MEM;
    return r1;
dengyihao's avatar
dengyihao 已提交
878
  } else {
879 880 881 882 883
    pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
    return r2;
  }
}

C
Cary Xu 已提交
884 885
static STSRow* getSRowInTableMem(STableCheckInfo* pCheckInfo, int32_t order, int32_t update, STSRow** extraRow,
                                 TDRowVerT maxVer) {
H
Haojun Liao 已提交
886
  STSRow *rmem = NULL, *rimem = NULL;
H
Haojun Liao 已提交
887 888 889
  if (pCheckInfo->iter) {
    SSkipListNode* node = tSkipListIterGet(pCheckInfo->iter);
    if (node != NULL) {
H
Haojun Liao 已提交
890
      rmem = (STSRow*)SL_GET_NODE_DATA(node);
C
Cary Xu 已提交
891
#if 0  // TODO: skiplist refactor
C
Cary Xu 已提交
892 893 894
      if (TD_ROW_VER(rmem) > maxVer) {
        rmem = NULL;
      }
C
Cary Xu 已提交
895
#endif
H
Haojun Liao 已提交
896 897
    }
  }
898

H
Haojun Liao 已提交
899 900 901
  if (pCheckInfo->iiter) {
    SSkipListNode* node = tSkipListIterGet(pCheckInfo->iiter);
    if (node != NULL) {
H
Haojun Liao 已提交
902
      rimem = (STSRow*)SL_GET_NODE_DATA(node);
C
Cary Xu 已提交
903
#if 0  // TODO: skiplist refactor
C
Cary Xu 已提交
904 905 906
      if (TD_ROW_VER(rimem) > maxVer) {
        rimem = NULL;
      }
C
Cary Xu 已提交
907
#endif
H
Haojun Liao 已提交
908 909
    }
  }
910

H
Haojun Liao 已提交
911 912
  if (rmem == NULL && rimem == NULL) {
    return NULL;
H
Haojun Liao 已提交
913
  }
914

H
Haojun Liao 已提交
915
  if (rmem != NULL && rimem == NULL) {
H
Haojun Liao 已提交
916 917 918
    pCheckInfo->chosen = 0;
    return rmem;
  }
919

H
Haojun Liao 已提交
920
  if (rmem == NULL && rimem != NULL) {
H
Haojun Liao 已提交
921 922 923
    pCheckInfo->chosen = 1;
    return rimem;
  }
924

H
Haojun Liao 已提交
925 926
  TSKEY r1 = TD_ROW_KEY(rmem);
  TSKEY r2 = TD_ROW_KEY(rimem);
H
Haojun Liao 已提交
927

928
  if (r1 == r2) {
C
Cary Xu 已提交
929
#if 0
930
    if (update == TD_ROW_DISCARD_UPDATE) {
H
TD-1439  
Hongze Cheng 已提交
931
      tSkipListIterNext(pCheckInfo->iter);
932
      pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
H
TD-1439  
Hongze Cheng 已提交
933
      return rimem;
dengyihao's avatar
dengyihao 已提交
934
    } else if (update == TD_ROW_OVERWRITE_UPDATE) {
H
TD-1439  
Hongze Cheng 已提交
935
      tSkipListIterNext(pCheckInfo->iiter);
936 937 938 939
      pCheckInfo->chosen = CHECKINFO_CHOSEN_MEM;
      return rmem;
    } else {
      pCheckInfo->chosen = CHECKINFO_CHOSEN_BOTH;
H
Haojun Liao 已提交
940
      *extraRow = rimem;
H
TD-1439  
Hongze Cheng 已提交
941 942
      return rmem;
    }
C
Cary Xu 已提交
943 944 945 946 947 948 949 950 951 952
#endif
    if (TD_SUPPORT_UPDATE(update)) {
      pCheckInfo->chosen = CHECKINFO_CHOSEN_BOTH;
      *extraRow = rimem;
      return rmem;
    } else {
      tSkipListIterNext(pCheckInfo->iter);
      pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
      return rimem;
    }
H
Haojun Liao 已提交
953 954 955
  } else {
    if (ASCENDING_TRAVERSE(order)) {
      if (r1 < r2) {
956
        pCheckInfo->chosen = CHECKINFO_CHOSEN_MEM;
H
Haojun Liao 已提交
957 958
        return rmem;
      } else {
959
        pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
H
Haojun Liao 已提交
960 961 962 963
        return rimem;
      }
    } else {
      if (r1 < r2) {
964
        pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
H
Haojun Liao 已提交
965 966
        return rimem;
      } else {
967
        pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
H
Haojun Liao 已提交
968 969 970 971
        return rmem;
      }
    }
  }
H
Haojun Liao 已提交
972 973
}

974
static bool moveToNextRowInMem(STableCheckInfo* pCheckInfo) {
H
Haojun Liao 已提交
975
  bool hasNext = false;
976
  if (pCheckInfo->chosen == CHECKINFO_CHOSEN_MEM) {
H
Haojun Liao 已提交
977 978 979
    if (pCheckInfo->iter != NULL) {
      hasNext = tSkipListIterNext(pCheckInfo->iter);
    }
980

H
Haojun Liao 已提交
981 982 983
    if (hasNext) {
      return hasNext;
    }
984

H
Haojun Liao 已提交
985 986 987
    if (pCheckInfo->iiter != NULL) {
      return tSkipListIterGet(pCheckInfo->iiter) != NULL;
    }
dengyihao's avatar
dengyihao 已提交
988
  } else if (pCheckInfo->chosen == CHECKINFO_CHOSEN_IMEM) {
989 990 991
    if (pCheckInfo->iiter != NULL) {
      hasNext = tSkipListIterNext(pCheckInfo->iiter);
    }
992

993 994 995
    if (hasNext) {
      return hasNext;
    }
996

997 998
    if (pCheckInfo->iter != NULL) {
      return tSkipListIterGet(pCheckInfo->iter) != NULL;
H
Haojun Liao 已提交
999
    }
1000 1001 1002 1003 1004 1005 1006
  } else {
    if (pCheckInfo->iter != NULL) {
      hasNext = tSkipListIterNext(pCheckInfo->iter);
    }
    if (pCheckInfo->iiter != NULL) {
      hasNext = tSkipListIterNext(pCheckInfo->iiter) || hasNext;
    }
H
Haojun Liao 已提交
1007
  }
1008

H
Haojun Liao 已提交
1009 1010 1011
  return hasNext;
}

1012
static bool hasMoreDataInCache(STsdbReadHandle* pHandle) {
H
Hongze Cheng 已提交
1013
  STsdbCfg* pCfg = REPO_CFG(pHandle->pTsdb);
dengyihao's avatar
dengyihao 已提交
1014
  size_t    size = taosArrayGetSize(pHandle->pTableCheckInfo);
1015
  assert(pHandle->activeIndex < size && pHandle->activeIndex >= 0 && size >= 1);
D
dapan1121 已提交
1016
  pHandle->cur.fid = INT32_MIN;
H
Haojun Liao 已提交
1017

1018
  STableCheckInfo* pCheckInfo = taosArrayGet(pHandle->pTableCheckInfo, pHandle->activeIndex);
H
Haojun Liao 已提交
1019 1020 1021 1022
  if (!pCheckInfo->initBuf) {
    initTableMemIterator(pHandle, pCheckInfo);
  }

C
Cary Xu 已提交
1023
  STSRow* row = getSRowInTableMem(pCheckInfo, pHandle->order, pCfg->update, NULL, TD_VER_MAX);
H
Haojun Liao 已提交
1024
  if (row == NULL) {
1025 1026
    return false;
  }
1027

H
Haojun Liao 已提交
1028
  pCheckInfo->lastKey = TD_ROW_KEY(row);  // first timestamp in buffer
dengyihao's avatar
dengyihao 已提交
1029 1030
  tsdbDebug("%p uid:%" PRId64 ", check data in buffer from skey:%" PRId64 ", order:%d, %s", pHandle,
            pCheckInfo->tableId, pCheckInfo->lastKey, pHandle->order, pHandle->idStr);
H
Haojun Liao 已提交
1031

1032
  // all data in mem are checked already.
1033 1034
  if ((pCheckInfo->lastKey > pHandle->window.ekey && ASCENDING_TRAVERSE(pHandle->order)) ||
      (pCheckInfo->lastKey < pHandle->window.ekey && !ASCENDING_TRAVERSE(pHandle->order))) {
1035 1036
    return false;
  }
H
Haojun Liao 已提交
1037

dengyihao's avatar
dengyihao 已提交
1038
  int32_t      step = ASCENDING_TRAVERSE(pHandle->order) ? 1 : -1;
1039
  STimeWindow* win = &pHandle->cur.win;
H
Haojun Liao 已提交
1040
  pHandle->cur.rows = tsdbReadRowsFromCache(pCheckInfo, pHandle->window.ekey, pHandle->outputCapacity, win, pHandle);
H
Haojun Liao 已提交
1041

1042 1043 1044 1045
  // update the last key value
  pCheckInfo->lastKey = win->ekey + step;
  pHandle->cur.lastKey = win->ekey + step;
  pHandle->cur.mixBlock = true;
1046

1047
  if (!ASCENDING_TRAVERSE(pHandle->order)) {
wafwerar's avatar
wafwerar 已提交
1048
    TSWAP(win->skey, win->ekey);
1049
  }
H
Haojun Liao 已提交
1050

1051
  return true;
1052
}
H
hjxilinx 已提交
1053

1054 1055
static int32_t getFileIdFromKey(TSKEY key, int32_t daysPerFile, int32_t precision) {
  assert(precision >= TSDB_TIME_PRECISION_MICRO || precision <= TSDB_TIME_PRECISION_NANO);
1056 1057 1058
  if (key == TSKEY_INITIAL_VAL) {
    return INT32_MIN;
  }
H
Haojun Liao 已提交
1059

D
dapan1121 已提交
1060
  if (key < 0) {
1061
    key -= (daysPerFile * tsTickPerMin[precision]);
D
dapan1121 已提交
1062
  }
dengyihao's avatar
dengyihao 已提交
1063

1064
  int64_t fid = (int64_t)(key / (daysPerFile * tsTickPerMin[precision]));  // set the starting fileId
dengyihao's avatar
dengyihao 已提交
1065
  if (fid < 0L && llabs(fid) > INT32_MAX) {                                // data value overflow for INT32
1066 1067
    fid = INT32_MIN;
  }
H
Haojun Liao 已提交
1068

1069
  if (fid > 0L && fid > INT32_MAX) {
1070 1071
    fid = INT32_MAX;
  }
H
Haojun Liao 已提交
1072

S
TD-1057  
Shengliang Guan 已提交
1073
  return (int32_t)fid;
1074 1075
}

H
refact  
Hongze Cheng 已提交
1076
static int32_t binarySearchForBlock(SBlock* pBlock, int32_t numOfBlocks, TSKEY skey, int32_t order) {
1077 1078
  int32_t firstSlot = 0;
  int32_t lastSlot = numOfBlocks - 1;
H
Haojun Liao 已提交
1079

1080
  int32_t midSlot = firstSlot;
H
Haojun Liao 已提交
1081

1082 1083 1084
  while (1) {
    numOfBlocks = lastSlot - firstSlot + 1;
    midSlot = (firstSlot + (numOfBlocks >> 1));
H
Haojun Liao 已提交
1085

1086
    if (numOfBlocks == 1) break;
H
Haojun Liao 已提交
1087

1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098
    if (skey > pBlock[midSlot].keyLast) {
      if (numOfBlocks == 2) break;
      if ((order == TSDB_ORDER_DESC) && (skey < pBlock[midSlot + 1].keyFirst)) break;
      firstSlot = midSlot + 1;
    } else if (skey < pBlock[midSlot].keyFirst) {
      if ((order == TSDB_ORDER_ASC) && (skey > pBlock[midSlot - 1].keyLast)) break;
      lastSlot = midSlot - 1;
    } else {
      break;  // got the slot
    }
  }
H
Haojun Liao 已提交
1099

1100 1101
  return midSlot;
}
1102

dengyihao's avatar
dengyihao 已提交
1103
static int32_t loadBlockInfo(STsdbReadHandle* pTsdbReadHandle, int32_t index, int32_t* numOfBlocks) {
H
Haojun Liao 已提交
1104
  int32_t code = 0;
H
Haojun Liao 已提交
1105

1106
  STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, index);
H
Haojun Liao 已提交
1107
  pCheckInfo->numOfBlocks = 0;
1108

H
Haojun Liao 已提交
1109
  STable table = {.uid = pCheckInfo->tableId, .tid = pCheckInfo->tableId};
1110
  table.pSchema = pTsdbReadHandle->pSchema;
H
Haojun Liao 已提交
1111 1112

  if (tsdbSetReadTable(&pTsdbReadHandle->rhelper, &table) != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
1113 1114 1115
    code = terrno;
    return code;
  }
1116

1117
  SBlockIdx* compIndex = pTsdbReadHandle->rhelper.pBlkIdx;
H
Hongze Cheng 已提交
1118

H
Haojun Liao 已提交
1119
  // no data block in this file, try next file
1120
  if (compIndex == NULL || compIndex->uid != pCheckInfo->tableId) {
H
Haojun Liao 已提交
1121 1122
    return 0;  // no data blocks in the file belongs to pCheckInfo->pTable
  }
1123

H
Haojun Liao 已提交
1124 1125 1126
  if (pCheckInfo->compSize < (int32_t)compIndex->len) {
    assert(compIndex->len > 0);

wafwerar's avatar
wafwerar 已提交
1127
    char* t = taosMemoryRealloc(pCheckInfo->pCompInfo, compIndex->len);
H
Haojun Liao 已提交
1128 1129 1130 1131
    if (t == NULL) {
      terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
      code = TSDB_CODE_TDB_OUT_OF_MEMORY;
      return code;
1132 1133
    }

H
Haojun Liao 已提交
1134 1135 1136
    pCheckInfo->pCompInfo = (SBlockInfo*)t;
    pCheckInfo->compSize = compIndex->len;
  }
1137

1138
  if (tsdbLoadBlockInfo(&(pTsdbReadHandle->rhelper), (void*)(pCheckInfo->pCompInfo)) < 0) {
H
Hongze Cheng 已提交
1139 1140
    return terrno;
  }
H
Haojun Liao 已提交
1141
  SBlockInfo* pCompInfo = pCheckInfo->pCompInfo;
1142

H
Haojun Liao 已提交
1143
  TSKEY s = TSKEY_INITIAL_VAL, e = TSKEY_INITIAL_VAL;
1144

1145
  if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
dengyihao's avatar
dengyihao 已提交
1146 1147
    assert(pCheckInfo->lastKey <= pTsdbReadHandle->window.ekey &&
           pTsdbReadHandle->window.skey <= pTsdbReadHandle->window.ekey);
H
Haojun Liao 已提交
1148
  } else {
dengyihao's avatar
dengyihao 已提交
1149 1150
    assert(pCheckInfo->lastKey >= pTsdbReadHandle->window.ekey &&
           pTsdbReadHandle->window.skey >= pTsdbReadHandle->window.ekey);
H
Haojun Liao 已提交
1151
  }
1152

dengyihao's avatar
dengyihao 已提交
1153 1154
  s = TMIN(pCheckInfo->lastKey, pTsdbReadHandle->window.ekey);
  e = TMAX(pCheckInfo->lastKey, pTsdbReadHandle->window.ekey);
1155

H
Haojun Liao 已提交
1156 1157 1158
  // discard the unqualified data block based on the query time window
  int32_t start = binarySearchForBlock(pCompInfo->blocks, compIndex->numOfBlocks, s, TSDB_ORDER_ASC);
  int32_t end = start;
H
TD-100  
hzcheng 已提交
1159

H
Haojun Liao 已提交
1160 1161 1162
  if (s > pCompInfo->blocks[start].keyLast) {
    return 0;
  }
1163

H
Haojun Liao 已提交
1164 1165 1166 1167
  // todo speedup the procedure of located end block
  while (end < (int32_t)compIndex->numOfBlocks && (pCompInfo->blocks[end].keyFirst <= e)) {
    end += 1;
  }
1168

H
Haojun Liao 已提交
1169
  pCheckInfo->numOfBlocks = (end - start);
1170

H
Haojun Liao 已提交
1171 1172 1173
  if (start > 0) {
    memmove(pCompInfo->blocks, &pCompInfo->blocks[start], pCheckInfo->numOfBlocks * sizeof(SBlock));
  }
1174

H
Haojun Liao 已提交
1175 1176 1177
  (*numOfBlocks) += pCheckInfo->numOfBlocks;
  return 0;
}
1178

1179
static int32_t getFileCompInfo(STsdbReadHandle* pTsdbReadHandle, int32_t* numOfBlocks) {
H
Haojun Liao 已提交
1180 1181 1182 1183
  // load all the comp offset value for all tables in this file
  int32_t code = TSDB_CODE_SUCCESS;
  *numOfBlocks = 0;

1184
  pTsdbReadHandle->cost.headFileLoad += 1;
1185 1186
  int64_t s = taosGetTimestampUs();

H
Haojun Liao 已提交
1187
  size_t numOfTables = 0;
1188 1189 1190 1191
  if (pTsdbReadHandle->loadType == BLOCK_LOAD_TABLE_SEQ_ORDER) {
    code = loadBlockInfo(pTsdbReadHandle, pTsdbReadHandle->activeIndex, numOfBlocks);
  } else if (pTsdbReadHandle->loadType == BLOCK_LOAD_OFFSET_SEQ_ORDER) {
    numOfTables = taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
1192

H
Haojun Liao 已提交
1193
    for (int32_t i = 0; i < numOfTables; ++i) {
1194
      code = loadBlockInfo(pTsdbReadHandle, i, numOfBlocks);
H
Haojun Liao 已提交
1195
      if (code != TSDB_CODE_SUCCESS) {
1196 1197
        int64_t e = taosGetTimestampUs();

1198
        pTsdbReadHandle->cost.headFileLoadTime += (e - s);
H
Haojun Liao 已提交
1199 1200 1201 1202 1203
        return code;
      }
    }
  } else {
    assert(0);
1204
  }
1205

1206
  int64_t e = taosGetTimestampUs();
1207
  pTsdbReadHandle->cost.headFileLoadTime += (e - s);
H
Haojun Liao 已提交
1208
  return code;
1209 1210
}

dengyihao's avatar
dengyihao 已提交
1211 1212
static int32_t doLoadFileDataBlock(STsdbReadHandle* pTsdbReadHandle, SBlock* pBlock, STableCheckInfo* pCheckInfo,
                                   int32_t slotIndex) {
H
Haojun Liao 已提交
1213
  int64_t st = taosGetTimestampUs();
1214

C
Cary Xu 已提交
1215
  int32_t code = tdInitDataCols(pTsdbReadHandle->pDataCols, pTsdbReadHandle->pSchema);
H
Haojun Liao 已提交
1216
  if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
1217
    tsdbError("%p failed to malloc buf for pDataCols, %s", pTsdbReadHandle, pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
1218 1219 1220 1221
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    goto _error;
  }

1222
  code = tdInitDataCols(pTsdbReadHandle->rhelper.pDCols[0], pTsdbReadHandle->pSchema);
H
Haojun Liao 已提交
1223
  if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
1224
    tsdbError("%p failed to malloc buf for rhelper.pDataCols[0], %s", pTsdbReadHandle, pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
1225 1226 1227 1228
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    goto _error;
  }

1229
  code = tdInitDataCols(pTsdbReadHandle->rhelper.pDCols[1], pTsdbReadHandle->pSchema);
H
Haojun Liao 已提交
1230
  if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
1231
    tsdbError("%p failed to malloc buf for rhelper.pDataCols[1], %s", pTsdbReadHandle, pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
1232 1233 1234
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    goto _error;
  }
1235

1236
  int16_t* colIds = pTsdbReadHandle->suppInfo.defaultLoadColumn->pData;
H
Haojun Liao 已提交
1237

dengyihao's avatar
dengyihao 已提交
1238 1239
  int32_t ret = tsdbLoadBlockDataCols(&(pTsdbReadHandle->rhelper), pBlock, pCheckInfo->pCompInfo, colIds,
                                      (int)(QH_GET_NUM_OF_COLS(pTsdbReadHandle)), true);
H
Haojun Liao 已提交
1240
  if (ret != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
1241 1242 1243
    int32_t c = terrno;
    assert(c != TSDB_CODE_SUCCESS);
    goto _error;
H
Haojun Liao 已提交
1244
  }
1245

1246
  SDataBlockLoadInfo* pBlockLoadInfo = &pTsdbReadHandle->dataBlockLoadInfo;
1247

1248 1249
  pBlockLoadInfo->fileGroup = pTsdbReadHandle->pFileGroup;
  pBlockLoadInfo->slot = pTsdbReadHandle->cur.slot;
H
Haojun Liao 已提交
1250
  pBlockLoadInfo->uid = pCheckInfo->tableId;
1251

1252
  SDataCols* pCols = pTsdbReadHandle->rhelper.pDCols[0];
1253
  assert(pCols->numOfRows != 0 && pCols->numOfRows <= pBlock->numOfRows);
1254

1255
  pBlock->numOfRows = pCols->numOfRows;
H
Haojun Liao 已提交
1256

1257
  // Convert from TKEY to TSKEY for primary timestamp column if current block has timestamp before 1970-01-01T00:00:00Z
dengyihao's avatar
dengyihao 已提交
1258
  if (pBlock->keyFirst < 0 && colIds[0] == PRIMARYKEY_TIMESTAMP_COL_ID) {
1259
    int64_t* src = pCols->cols[0].pData;
dengyihao's avatar
dengyihao 已提交
1260
    for (int32_t i = 0; i < pBlock->numOfRows; ++i) {
1261 1262 1263 1264
      src[i] = tdGetKey(src[i]);
    }
  }

H
Haojun Liao 已提交
1265
  int64_t elapsedTime = (taosGetTimestampUs() - st);
1266
  pTsdbReadHandle->cost.blockLoadTime += elapsedTime;
1267

dengyihao's avatar
dengyihao 已提交
1268 1269 1270 1271
  tsdbDebug("%p load file block into buffer, index:%d, brange:%" PRId64 "-%" PRId64 ", rows:%d, elapsed time:%" PRId64
            " us, %s",
            pTsdbReadHandle, slotIndex, pBlock->keyFirst, pBlock->keyLast, pBlock->numOfRows, elapsedTime,
            pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
1272
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1273 1274 1275 1276

_error:
  pBlock->numOfRows = 0;

dengyihao's avatar
dengyihao 已提交
1277
  tsdbError("%p error occurs in loading file block, index:%d, brange:%" PRId64 "-%" PRId64 ", rows:%d, %s",
H
Haojun Liao 已提交
1278
            pTsdbReadHandle, slotIndex, pBlock->keyFirst, pBlock->keyLast, pBlock->numOfRows, pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
1279
  return terrno;
H
hjxilinx 已提交
1280 1281
}

1282
static int32_t getEndPosInDataBlock(STsdbReadHandle* pTsdbReadHandle, SDataBlockInfo* pBlockInfo);
dengyihao's avatar
dengyihao 已提交
1283 1284 1285 1286 1287 1288 1289
static int32_t doCopyRowsFromFileBlock(STsdbReadHandle* pTsdbReadHandle, int32_t capacity, int32_t numOfRows,
                                       int32_t start, int32_t end);
static void    doCheckGeneratedBlockRange(STsdbReadHandle* pTsdbReadHandle);
static void    copyAllRemainRowsFromFileBlock(STsdbReadHandle* pTsdbReadHandle, STableCheckInfo* pCheckInfo,
                                              SDataBlockInfo* pBlockInfo, int32_t endPos);

static int32_t handleDataMergeIfNeeded(STsdbReadHandle* pTsdbReadHandle, SBlock* pBlock, STableCheckInfo* pCheckInfo) {
1290
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
H
Hongze Cheng 已提交
1291
  STsdbCfg*      pCfg = REPO_CFG(pTsdbReadHandle->pTsdb);
H
Haojun Liao 已提交
1292
  SDataBlockInfo binfo = GET_FILE_DATA_BLOCK_INFO(pCheckInfo, pBlock);
1293
  TSKEY          key;
dengyihao's avatar
dengyihao 已提交
1294
  int32_t        code = TSDB_CODE_SUCCESS;
1295

1296
  /*bool hasData = */ initTableMemIterator(pTsdbReadHandle, pCheckInfo);
H
Haojun Liao 已提交
1297 1298
  assert(cur->pos >= 0 && cur->pos <= binfo.rows);

C
Cary Xu 已提交
1299
  key = extractFirstTraverseKey(pCheckInfo, pTsdbReadHandle->order, pCfg->update, TD_VER_MAX);
1300

H
Haojun Liao 已提交
1301
  if (key != TSKEY_INITIAL_VAL) {
dengyihao's avatar
dengyihao 已提交
1302
    tsdbDebug("%p key in mem:%" PRId64 ", %s", pTsdbReadHandle, key, pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
1303
  } else {
H
Haojun Liao 已提交
1304
    tsdbDebug("%p no data in mem, %s", pTsdbReadHandle, pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
1305
  }
H
Haojun Liao 已提交
1306

1307 1308
  bool ascScan = ASCENDING_TRAVERSE(pTsdbReadHandle->order);

1309 1310
  if ((ascScan && (key != TSKEY_INITIAL_VAL && key <= binfo.window.ekey)) ||
      (!ascScan && (key != TSKEY_INITIAL_VAL && key >= binfo.window.skey))) {
1311 1312 1313 1314

    bool cacheDataInFileBlockHole = (ascScan && (key != TSKEY_INITIAL_VAL && key < binfo.window.skey)) ||
                                    (!ascScan && (key != TSKEY_INITIAL_VAL && key > binfo.window.ekey));
    if (cacheDataInFileBlockHole) {
H
Haojun Liao 已提交
1315
      // do not load file block into buffer
1316
      int32_t step = ascScan ? 1 : -1;
H
Haojun Liao 已提交
1317

1318
      TSKEY maxKey = ascScan ? (binfo.window.skey - step) : (binfo.window.ekey - step);
dengyihao's avatar
dengyihao 已提交
1319 1320
      cur->rows =
          tsdbReadRowsFromCache(pCheckInfo, maxKey, pTsdbReadHandle->outputCapacity, &cur->win, pTsdbReadHandle);
1321
      pTsdbReadHandle->realNumOfRows = cur->rows;
H
Haojun Liao 已提交
1322 1323 1324

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

      if (!ascScan) {
wafwerar's avatar
wafwerar 已提交
1327
        TSWAP(cur->win.skey, cur->win.ekey);
H
Haojun Liao 已提交
1328
      }
H
Haojun Liao 已提交
1329

H
Haojun Liao 已提交
1330 1331
      cur->mixBlock = true;
      cur->blockCompleted = false;
H
Haojun Liao 已提交
1332
      return code;
H
Haojun Liao 已提交
1333
    }
H
Haojun Liao 已提交
1334

1335
    // return error, add test cases
1336
    if ((code = doLoadFileDataBlock(pTsdbReadHandle, pBlock, pCheckInfo, cur->slot)) != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
1337
      return code;
1338 1339
    }

1340
    doMergeTwoLevelData(pTsdbReadHandle, pCheckInfo, pBlock);
1341
  } else {
1342 1343 1344 1345 1346
    /*
     * 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
     */
1347
    int32_t endPos = getEndPosInDataBlock(pTsdbReadHandle, &binfo);
1348

1349 1350
    bool wholeBlockReturned = ((abs(cur->pos - endPos) + 1) == binfo.rows);
    if (wholeBlockReturned) {
1351
      pTsdbReadHandle->realNumOfRows = binfo.rows;
1352 1353

      cur->rows = binfo.rows;
1354
      cur->win  = binfo.window;
1355
      cur->mixBlock = false;
H
Haojun Liao 已提交
1356 1357
      cur->blockCompleted = true;

1358
      if (ascScan) {
H
Haojun Liao 已提交
1359 1360 1361 1362 1363 1364
        cur->lastKey = binfo.window.ekey + 1;
        cur->pos = binfo.rows;
      } else {
        cur->lastKey = binfo.window.skey - 1;
        cur->pos = -1;
      }
1365 1366 1367 1368 1369 1370 1371 1372 1373 1374
    } 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) {
        code = doLoadFileDataBlock(pTsdbReadHandle, pBlock, pCheckInfo, cur->slot);
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
      }

1375
      copyAllRemainRowsFromFileBlock(pTsdbReadHandle, pCheckInfo, &binfo, endPos);
1376 1377
      cur->mixBlock = true;
    }
1378

1379 1380 1381 1382
    if (pTsdbReadHandle->outputCapacity >= binfo.rows) {
      ASSERT(cur->blockCompleted);
    }

H
Haojun Liao 已提交
1383
    if (cur->rows == binfo.rows) {
dengyihao's avatar
dengyihao 已提交
1384
      tsdbDebug("%p whole file block qualified, brange:%" PRId64 "-%" PRId64 ", rows:%d, lastKey:%" PRId64 ", %s",
H
Haojun Liao 已提交
1385
                pTsdbReadHandle, cur->win.skey, cur->win.ekey, cur->rows, cur->lastKey, pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
1386
    } else {
dengyihao's avatar
dengyihao 已提交
1387 1388 1389 1390
      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
Haojun Liao 已提交
1391
    }
1392
  }
H
Haojun Liao 已提交
1393 1394

  return code;
1395 1396
}

1397 1398
static int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order);

dengyihao's avatar
dengyihao 已提交
1399 1400
static int32_t loadFileDataBlock(STsdbReadHandle* pTsdbReadHandle, SBlock* pBlock, STableCheckInfo* pCheckInfo,
                                 bool* exists) {
1401
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
dengyihao's avatar
dengyihao 已提交
1402 1403
  int32_t        code = TSDB_CODE_SUCCESS;
  bool           asc = ASCENDING_TRAVERSE(pTsdbReadHandle->order);
1404

1405
  if (asc) {
H
Haojun Liao 已提交
1406
    // query ended in/started from current block
1407 1408
    if (pTsdbReadHandle->window.ekey < pBlock->keyLast || pCheckInfo->lastKey > pBlock->keyFirst) {
      if ((code = doLoadFileDataBlock(pTsdbReadHandle, pBlock, pCheckInfo, cur->slot)) != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
1409 1410
        *exists = false;
        return code;
1411
      }
1412

1413
      SDataCols* pTSCol = pTsdbReadHandle->rhelper.pDCols[0];
H
Haojun Liao 已提交
1414
      assert(pTSCol->cols->type == TSDB_DATA_TYPE_TIMESTAMP && pTSCol->numOfRows == pBlock->numOfRows);
H
Haojun Liao 已提交
1415

1416 1417
      if (pCheckInfo->lastKey > pBlock->keyFirst) {
        cur->pos =
1418
            binarySearchForKey(pTSCol->cols[0].pData, pBlock->numOfRows, pCheckInfo->lastKey, pTsdbReadHandle->order);
1419 1420 1421
      } else {
        cur->pos = 0;
      }
H
Haojun Liao 已提交
1422

H
Haojun Liao 已提交
1423
      assert(pCheckInfo->lastKey <= pBlock->keyLast);
1424
      doMergeTwoLevelData(pTsdbReadHandle, pCheckInfo, pBlock);
1425
    } else {  // the whole block is loaded in to buffer
dengyihao's avatar
dengyihao 已提交
1426
      cur->pos = asc ? 0 : (pBlock->numOfRows - 1);
1427
      code = handleDataMergeIfNeeded(pTsdbReadHandle, pBlock, pCheckInfo);
H
[td-32]  
hjxilinx 已提交
1428
    }
dengyihao's avatar
dengyihao 已提交
1429
  } else {  // desc order, query ended in current block
1430 1431
    if (pTsdbReadHandle->window.ekey > pBlock->keyFirst || pCheckInfo->lastKey < pBlock->keyLast) {
      if ((code = doLoadFileDataBlock(pTsdbReadHandle, pBlock, pCheckInfo, cur->slot)) != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
1432 1433
        *exists = false;
        return code;
1434
      }
H
Haojun Liao 已提交
1435

1436
      SDataCols* pTsCol = pTsdbReadHandle->rhelper.pDCols[0];
1437
      if (pCheckInfo->lastKey < pBlock->keyLast) {
dengyihao's avatar
dengyihao 已提交
1438 1439
        cur->pos =
            binarySearchForKey(pTsCol->cols[0].pData, pBlock->numOfRows, pCheckInfo->lastKey, pTsdbReadHandle->order);
1440
      } else {
H
Haojun Liao 已提交
1441
        cur->pos = pBlock->numOfRows - 1;
1442
      }
H
Haojun Liao 已提交
1443

H
Haojun Liao 已提交
1444
      assert(pCheckInfo->lastKey >= pBlock->keyFirst);
1445
      doMergeTwoLevelData(pTsdbReadHandle, pCheckInfo, pBlock);
1446
    } else {
dengyihao's avatar
dengyihao 已提交
1447
      cur->pos = asc ? 0 : (pBlock->numOfRows - 1);
1448
      code = handleDataMergeIfNeeded(pTsdbReadHandle, pBlock, pCheckInfo);
H
[td-32]  
hjxilinx 已提交
1449
    }
1450
  }
1451

1452
  *exists = pTsdbReadHandle->realNumOfRows > 0;
H
Haojun Liao 已提交
1453
  return code;
H
[td-32]  
hjxilinx 已提交
1454 1455
}

1456
static int doBinarySearchKey(char* pValue, int num, TSKEY key, int order) {
1457
  int    firstPos, lastPos, midPos = -1;
H
Haojun Liao 已提交
1458
  int    numOfRows;
1459 1460
  TSKEY* keyList;

1461
  assert(order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC);
H
Haojun Liao 已提交
1462

1463
  if (num <= 0) return -1;
1464 1465

  keyList = (TSKEY*)pValue;
1466 1467
  firstPos = 0;
  lastPos = num - 1;
1468

1469
  if (order == TSDB_ORDER_DESC) {
1470 1471 1472 1473 1474
    // 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;
1475

H
Haojun Liao 已提交
1476 1477
      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1) + firstPos;
1478

1479 1480 1481 1482 1483 1484 1485 1486
      if (key < keyList[midPos]) {
        lastPos = midPos - 1;
      } else if (key > keyList[midPos]) {
        firstPos = midPos + 1;
      } else {
        break;
      }
    }
1487

1488 1489 1490 1491 1492
  } 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;
1493

1494 1495 1496 1497 1498 1499 1500
      if (key > keyList[lastPos]) {
        lastPos = lastPos + 1;
        if (lastPos >= num)
          return -1;
        else
          return lastPos;
      }
1501

H
Haojun Liao 已提交
1502 1503
      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1) + firstPos;
1504

1505 1506 1507 1508 1509 1510 1511 1512 1513
      if (key < keyList[midPos]) {
        lastPos = midPos - 1;
      } else if (key > keyList[midPos]) {
        firstPos = midPos + 1;
      } else {
        break;
      }
    }
  }
1514

1515 1516 1517
  return midPos;
}

dengyihao's avatar
dengyihao 已提交
1518 1519
static int32_t doCopyRowsFromFileBlock(STsdbReadHandle* pTsdbReadHandle, int32_t capacity, int32_t numOfRows,
                                       int32_t start, int32_t end) {
1520
  SDataCols* pCols = pTsdbReadHandle->rhelper.pDCols[0];
dengyihao's avatar
dengyihao 已提交
1521
  TSKEY*     tsArray = pCols->cols[0].pData;
H
Haojun Liao 已提交
1522

1523
  int32_t num = end - start + 1;
H
Haojun Liao 已提交
1524 1525 1526 1527 1528 1529
  assert(num >= 0);

  if (num == 0) {
    return numOfRows;
  }

1530 1531
  bool    ascScan = ASCENDING_TRAVERSE(pTsdbReadHandle->order);
  int32_t trueStart = ascScan ? start : end;
1532
  int32_t trueEnd = ascScan ? end : start;
1533 1534
  int32_t step = ascScan ? 1 : -1;

1535
  int32_t requiredNumOfCols = (int32_t)taosArrayGetSize(pTsdbReadHandle->pColumns);
H
Haojun Liao 已提交
1536

dengyihao's avatar
dengyihao 已提交
1537
  // data in buffer has greater timestamp, copy data in file block
1538
  int32_t i = 0, j = 0;
dengyihao's avatar
dengyihao 已提交
1539
  while (i < requiredNumOfCols && j < pCols->numOfCols) {
1540
    SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
1541 1542 1543 1544 1545 1546 1547

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

L
Liu Jicong 已提交
1548
    if (!isAllRowsNull(src) && pColInfo->info.colId == src->colId) {
1549
      if (!IS_VAR_DATA_TYPE(pColInfo->info.type)) {  // todo opt performance
dengyihao's avatar
dengyihao 已提交
1550
        //        memmove(pData, (char*)src->pData + bytes * start, bytes * num);
1551
        int32_t rowIndex = numOfRows;
1552
        for (int32_t k = trueStart; ((ascScan && k <= trueEnd) || (!ascScan && k >= trueEnd)); k += step, ++rowIndex) {
1553
          SCellVal sVal = {0};
C
Cary Xu 已提交
1554
          if (tdGetColDataOfRow(&sVal, src, k, pCols->bitmapMode) < 0) {
1555 1556
            TASSERT(0);
          }
1557

C
Cary Xu 已提交
1558
          if (sVal.valType == TD_VTYPE_NORM) {
1559
            colDataAppend(pColInfo, rowIndex, sVal.val, false);
C
Cary Xu 已提交
1560 1561
          } else {
            colDataAppendNULL(pColInfo, rowIndex);
1562 1563 1564
          }
        }
      } else {  // handle the var-string
1565 1566
        int32_t rowIndex = numOfRows;

1567
        // todo refactor, only copy one-by-one
1568
        for (int32_t k = trueStart; ((ascScan && k <= trueEnd) || (!ascScan && k >= trueEnd)); k += step, ++rowIndex) {
1569
          SCellVal sVal = {0};
dengyihao's avatar
dengyihao 已提交
1570
          if (tdGetColDataOfRow(&sVal, src, k, pCols->bitmapMode) < 0) {
H
Haojun Liao 已提交
1571 1572
            TASSERT(0);
          }
1573

C
Cary Xu 已提交
1574
          if (sVal.valType == TD_VTYPE_NORM) {
1575
            colDataAppend(pColInfo, rowIndex, sVal.val, false);
C
Cary Xu 已提交
1576 1577
          } else {
            colDataAppendNULL(pColInfo, rowIndex);
1578
          }
1579 1580
        }
      }
1581 1582 1583

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

dengyihao's avatar
dengyihao 已提交
1590
  while (i < requiredNumOfCols) {  // the remain columns are all null data
1591
    SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
1592
    colDataAppendNNULL(pColInfo, numOfRows, num);
1593
    i++;
1594
  }
H
Haojun Liao 已提交
1595

1596 1597
  pTsdbReadHandle->cur.win.ekey = tsArray[trueEnd];
  pTsdbReadHandle->cur.lastKey = tsArray[trueEnd] + step;
1598

1599
  return numOfRows + num;
1600 1601
}

C
Cary Xu 已提交
1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621
/**
 * @brief  // TODO fix bug for reverse copy data problem
 *        Note: row1 always has high priority
 *
 * @param pTsdbReadHandle
 * @param capacity
 * @param curRow
 * @param row1
 * @param row2
 * @param numOfCols
 * @param uid
 * @param pSchema1
 * @param pSchema2
 * @param update
 * @param lastRowKey
 * @return int32_t The quantity of rows appended
 */
static int32_t mergeTwoRowFromMem(STsdbReadHandle* 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) {
H
Haojun Liao 已提交
1622
#if 1
dengyihao's avatar
dengyihao 已提交
1623 1624 1625 1626 1627 1628 1629 1630 1631
  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;
H
Haojun Liao 已提交
1632
  SCellVal sVal = {0};
C
Cary Xu 已提交
1633 1634
  TSKEY    rowKey = TSKEY_INITIAL_VAL;
  int32_t  nResult = 0;
C
Cary Xu 已提交
1635
  int32_t  mergeOption = 0;  // 0 discard 1 overwrite 2 merge
1636

H
Haojun Liao 已提交
1637
  // the schema version info is embeded in STSRow
1638 1639 1640
  int32_t numOfColsOfRow1 = 0;

  if (pSchema1 == NULL) {
1641
    pSchema1 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, TD_ROW_SVER(row1));
1642
  }
1643

C
Cary Xu 已提交
1644 1645 1646 1647
#ifdef TD_DEBUG_PRINT_ROW
  tdSRowPrint(row1, pSchema1, __func__);
#endif

dengyihao's avatar
dengyihao 已提交
1648
  if (isRow1DataRow) {
1649
    numOfColsOfRow1 = schemaNCols(pSchema1);
H
Haojun Liao 已提交
1650
  } else {
H
Haojun Liao 已提交
1651
    numOfColsOfRow1 = tdRowGetNCols(row1);
D
fix bug  
dapan1121 已提交
1652
  }
1653

1654
  int32_t numOfColsOfRow2 = 0;
dengyihao's avatar
dengyihao 已提交
1655
  if (row2) {
H
Haojun Liao 已提交
1656
    isRow2DataRow = TD_IS_TP_ROW(row2);
1657
    if (pSchema2 == NULL) {
1658
      pSchema2 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, TD_ROW_SVER(row2));
1659
    }
dengyihao's avatar
dengyihao 已提交
1660
    if (isRow2DataRow) {
1661 1662
      numOfColsOfRow2 = schemaNCols(pSchema2);
    } else {
H
Haojun Liao 已提交
1663
      numOfColsOfRow2 = tdRowGetNCols(row2);
1664 1665
    }
  }
C
Cary Xu 已提交
1666

1667
  int32_t i = 0, j = 0, k = 0;
dengyihao's avatar
dengyihao 已提交
1668
  while (i < numOfCols && (j < numOfColsOfRow1 || k < numOfColsOfRow2)) {
1669
    SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
1670 1671

    int32_t colIdOfRow1;
dengyihao's avatar
dengyihao 已提交
1672
    if (j >= numOfColsOfRow1) {
1673
      colIdOfRow1 = INT32_MAX;
dengyihao's avatar
dengyihao 已提交
1674
    } else if (isRow1DataRow) {
1675 1676
      colIdOfRow1 = pSchema1->columns[j].colId;
    } else {
C
Cary Xu 已提交
1677
      colIdOfRow1 = tdKvRowColIdAt(row1, j);
1678 1679 1680
    }

    int32_t colIdOfRow2;
dengyihao's avatar
dengyihao 已提交
1681
    if (k >= numOfColsOfRow2) {
1682
      colIdOfRow2 = INT32_MAX;
dengyihao's avatar
dengyihao 已提交
1683
    } else if (isRow2DataRow) {
1684 1685
      colIdOfRow2 = pSchema2->columns[k].colId;
    } else {
C
Cary Xu 已提交
1686
      colIdOfRow2 = tdKvRowColIdAt(row2, k);
1687 1688
    }

C
Cary Xu 已提交
1689
    if (colIdOfRow1 < colIdOfRow2) {  // the most probability
dengyihao's avatar
dengyihao 已提交
1690
      if (colIdOfRow1 < pColInfo->info.colId) {
C
Cary Xu 已提交
1691
        ++j;
C
Cary Xu 已提交
1692 1693
        continue;
      }
1694 1695 1696 1697
      row = row1;
      pSchema = pSchema1;
      isChosenRowDataRow = isRow1DataRow;
      chosen_itr = j;
C
Cary Xu 已提交
1698
    } else if (colIdOfRow1 == colIdOfRow2) {
dengyihao's avatar
dengyihao 已提交
1699
      if (colIdOfRow1 < pColInfo->info.colId) {
C
Cary Xu 已提交
1700 1701
        ++j;
        ++k;
1702
        continue;
C
Cary Xu 已提交
1703
      }
1704 1705 1706 1707 1708
      row = row1;
      pSchema = pSchema1;
      isChosenRowDataRow = isRow1DataRow;
      chosen_itr = j;
    } else {
dengyihao's avatar
dengyihao 已提交
1709
      if (colIdOfRow2 < pColInfo->info.colId) {
C
Cary Xu 已提交
1710
        ++k;
1711 1712 1713 1714 1715 1716 1717
        continue;
      }
      row = row2;
      pSchema = pSchema2;
      chosen_itr = k;
      isChosenRowDataRow = isRow2DataRow;
    }
C
Cary Xu 已提交
1718

dengyihao's avatar
dengyihao 已提交
1719
    if (isChosenRowDataRow) {
1720 1721
      colId = pSchema->columns[chosen_itr].colId;
      offset = pSchema->columns[chosen_itr].offset;
C
Cary Xu 已提交
1722 1723
      // TODO: use STSRowIter
      tdSTpRowGetVal(row, colId, pSchema->columns[chosen_itr].type, pSchema->flen, offset, chosen_itr - 1, &sVal);
C
Cary Xu 已提交
1724 1725 1726
      if (colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
        rowKey = *(TSKEY*)sVal.val;
        if (rowKey != *lastRowKey) {
C
Cary Xu 已提交
1727
          mergeOption = 1;
C
Cary Xu 已提交
1728 1729 1730
          if (*lastRowKey != TSKEY_INITIAL_VAL) {
            ++(*curRow);
          }
1731
          *lastRowKey = rowKey;
C
Cary Xu 已提交
1732
          ++nResult;
C
Cary Xu 已提交
1733
        } else if (update) {
C
Cary Xu 已提交
1734 1735 1736 1737
          mergeOption = 2;
        } else {
          mergeOption = 0;
          break;
C
Cary Xu 已提交
1738 1739
        }
      }
1740
    } else {
C
Cary Xu 已提交
1741 1742 1743 1744
      // TODO: use STSRowIter
      if (chosen_itr == 0) {
        colId = PRIMARYKEY_TIMESTAMP_COL_ID;
        tdSKvRowGetVal(row, PRIMARYKEY_TIMESTAMP_COL_ID, -1, -1, &sVal);
C
Cary Xu 已提交
1745 1746
        rowKey = *(TSKEY*)sVal.val;
        if (rowKey != *lastRowKey) {
C
Cary Xu 已提交
1747
          mergeOption = 1;
C
Cary Xu 已提交
1748 1749 1750
          if (*lastRowKey != TSKEY_INITIAL_VAL) {
            ++(*curRow);
          }
1751
          *lastRowKey = rowKey;
C
Cary Xu 已提交
1752
          ++nResult;
C
Cary Xu 已提交
1753
        } else if (update) {
C
Cary Xu 已提交
1754 1755 1756 1757
          mergeOption = 2;
        } else {
          mergeOption = 0;
          break;
C
Cary Xu 已提交
1758
        }
C
Cary Xu 已提交
1759 1760 1761 1762 1763 1764
      } else {
        SKvRowIdx* pColIdx = tdKvRowColIdxAt(row, chosen_itr - 1);
        colId = pColIdx->colId;
        offset = pColIdx->offset;
        tdSKvRowGetVal(row, colId, offset, chosen_itr - 1, &sVal);
      }
1765 1766
    }

C
Cary Xu 已提交
1767 1768
    ASSERT(rowKey != TSKEY_INITIAL_VAL);

1769
    if (colId == pColInfo->info.colId) {
H
Haojun Liao 已提交
1770
      if (tdValTypeIsNorm(sVal.valType)) {
C
Cary Xu 已提交
1771 1772 1773 1774 1775
        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
C
Cary Xu 已提交
1776
        if (mergeOption == 1) {
C
Cary Xu 已提交
1777 1778 1779 1780
          colDataAppend(pColInfo, *curRow, NULL, true);
        }
      } else {
        ASSERT(0);
1781
      }
H
Haojun Liao 已提交
1782

C
Cary Xu 已提交
1783
      ++i;
C
Cary Xu 已提交
1784

dengyihao's avatar
dengyihao 已提交
1785
      if (row == row1) {
C
Cary Xu 已提交
1786
        ++j;
1787
      } else {
C
Cary Xu 已提交
1788
        ++k;
1789 1790
      }
    } else {
C
Cary Xu 已提交
1791
      if (mergeOption == 1) {
C
Cary Xu 已提交
1792
        colDataAppend(pColInfo, *curRow, NULL, true);
C
Cary Xu 已提交
1793
      }
C
Cary Xu 已提交
1794
      ++i;
1795
    }
1796
  }
1797

C
Cary Xu 已提交
1798
  if (mergeOption == 1) {
dengyihao's avatar
dengyihao 已提交
1799
    while (i < numOfCols) {  // the remain columns are all null data
1800
      SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
C
Cary Xu 已提交
1801 1802
      colDataAppend(pColInfo, *curRow, NULL, true);
      ++i;
1803 1804
    }
  }
C
Cary Xu 已提交
1805 1806

  return nResult;
H
Haojun Liao 已提交
1807
#endif
1808
}
1809

dengyihao's avatar
dengyihao 已提交
1810 1811
static void getQualifiedRowsPos(STsdbReadHandle* pTsdbReadHandle, int32_t startPos, int32_t endPos,
                                int32_t numOfExisted, int32_t* start, int32_t* end) {
1812 1813
  *start = -1;

1814
  if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
1815
    int32_t remain = endPos - startPos + 1;
1816 1817
    if (remain + numOfExisted > pTsdbReadHandle->outputCapacity) {
      *end = (pTsdbReadHandle->outputCapacity - numOfExisted) + startPos - 1;
H
Haojun Liao 已提交
1818 1819
    } else {
      *end = endPos;
1820 1821 1822 1823 1824
    }

    *start = startPos;
  } else {
    int32_t remain = (startPos - endPos) + 1;
1825 1826
    if (remain + numOfExisted > pTsdbReadHandle->outputCapacity) {
      *end = startPos + 1 - (pTsdbReadHandle->outputCapacity - numOfExisted);
H
Haojun Liao 已提交
1827 1828
    } else {
      *end = endPos;
1829 1830 1831 1832 1833 1834 1835
    }

    *start = *end;
    *end = startPos;
  }
}

dengyihao's avatar
dengyihao 已提交
1836 1837
static void updateInfoAfterMerge(STsdbReadHandle* pTsdbReadHandle, STableCheckInfo* pCheckInfo, int32_t numOfRows,
                                 int32_t endPos) {
1838
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
1839 1840

  pCheckInfo->lastKey = cur->lastKey;
1841
  pTsdbReadHandle->realNumOfRows = numOfRows;
1842 1843 1844 1845
  cur->rows = numOfRows;
  cur->pos = endPos;
}

1846 1847
static void doCheckGeneratedBlockRange(STsdbReadHandle* pTsdbReadHandle) {
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
H
Haojun Liao 已提交
1848 1849

  if (cur->rows > 0) {
1850 1851
    if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
      assert(cur->win.skey >= pTsdbReadHandle->window.skey && cur->win.ekey <= pTsdbReadHandle->window.ekey);
H
Haojun Liao 已提交
1852
    } else {
1853
      assert(cur->win.skey >= pTsdbReadHandle->window.ekey && cur->win.ekey <= pTsdbReadHandle->window.skey);
H
Haojun Liao 已提交
1854 1855
    }

1856
    SColumnInfoData* pColInfoData = taosArrayGet(pTsdbReadHandle->pColumns, 0);
dengyihao's avatar
dengyihao 已提交
1857 1858
    assert(cur->win.skey == ((TSKEY*)pColInfoData->pData)[0] &&
           cur->win.ekey == ((TSKEY*)pColInfoData->pData)[cur->rows - 1]);
H
Haojun Liao 已提交
1859
  } else {
1860
    cur->win = pTsdbReadHandle->window;
H
Haojun Liao 已提交
1861

dengyihao's avatar
dengyihao 已提交
1862
    int32_t step = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 1 : -1;
1863
    cur->lastKey = pTsdbReadHandle->window.ekey + step;
H
Haojun Liao 已提交
1864 1865 1866
  }
}

dengyihao's avatar
dengyihao 已提交
1867 1868
static void copyAllRemainRowsFromFileBlock(STsdbReadHandle* pTsdbReadHandle, STableCheckInfo* pCheckInfo,
                                           SDataBlockInfo* pBlockInfo, int32_t endPos) {
1869
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
H
Haojun Liao 已提交
1870

1871
  SDataCols* pCols = pTsdbReadHandle->rhelper.pDCols[0];
dengyihao's avatar
dengyihao 已提交
1872
  TSKEY*     tsArray = pCols->cols[0].pData;
H
Haojun Liao 已提交
1873

1874
  bool ascScan = ASCENDING_TRAVERSE(pTsdbReadHandle->order);
H
Haojun Liao 已提交
1875

1876
  int32_t step = ascScan? 1 : -1;
H
Haojun Liao 已提交
1877 1878 1879 1880

  int32_t start = cur->pos;
  int32_t end = endPos;

1881
  if (!ascScan) {
wafwerar's avatar
wafwerar 已提交
1882
    TSWAP(start, end);
H
Haojun Liao 已提交
1883 1884
  }

1885 1886
  assert(pTsdbReadHandle->outputCapacity >= (end - start + 1));
  int32_t numOfRows = doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, 0, start, end);
H
Haojun Liao 已提交
1887 1888

  // the time window should always be ascending order: skey <= ekey
dengyihao's avatar
dengyihao 已提交
1889
  cur->win = (STimeWindow){.skey = tsArray[start], .ekey = tsArray[end]};
H
Haojun Liao 已提交
1890
  cur->mixBlock = (numOfRows != pBlockInfo->rows);
1891 1892
  cur->lastKey  = tsArray[endPos] + step;
  cur->blockCompleted = (ascScan? (endPos == pBlockInfo->rows - 1):(endPos == 0));
H
Haojun Liao 已提交
1893

H
Haojun Liao 已提交
1894
  // The value of pos may be -1 or pBlockInfo->rows, and it is invalid in both cases.
1895
  int32_t pos = endPos + step;
1896 1897
  updateInfoAfterMerge(pTsdbReadHandle, pCheckInfo, numOfRows, pos);
  doCheckGeneratedBlockRange(pTsdbReadHandle);
H
Haojun Liao 已提交
1898

dengyihao's avatar
dengyihao 已提交
1899 1900 1901
  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
Haojun Liao 已提交
1902 1903
}

1904
int32_t getEndPosInDataBlock(STsdbReadHandle* pTsdbReadHandle, SDataBlockInfo* pBlockInfo) {
H
Haojun Liao 已提交
1905 1906
  // NOTE: reverse the order to find the end position in data block
  int32_t endPos = -1;
1907 1908
  bool    ascScan = ASCENDING_TRAVERSE(pTsdbReadHandle->order);
  int32_t order = ascScan? TSDB_ORDER_DESC : TSDB_ORDER_ASC;
H
Haojun Liao 已提交
1909

1910
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
dengyihao's avatar
dengyihao 已提交
1911
  SDataCols*     pCols = pTsdbReadHandle->rhelper.pDCols[0];
H
Haojun Liao 已提交
1912

1913 1914 1915 1916 1917 1918 1919 1920 1921 1922 1923 1924
  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;
    }
H
Haojun Liao 已提交
1925
  } else {
1926
    if (ascScan && pTsdbReadHandle->window.ekey >= pBlockInfo->window.ekey) {
1927
      endPos = TMIN(cur->pos + pTsdbReadHandle->outputCapacity - 1, pBlockInfo->rows - 1);
1928
    } else if ((!ascScan) && pTsdbReadHandle->window.ekey <= pBlockInfo->window.skey) {
1929
      endPos = TMAX(cur->pos - pTsdbReadHandle->outputCapacity + 1, 0);
1930 1931 1932 1933 1934 1935 1936 1937 1938 1939 1940 1941 1942 1943 1944
    } 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;
        }
      }
    }
H
Haojun Liao 已提交
1945 1946 1947 1948 1949 1950
    cur->mixBlock = true;
  }

  return endPos;
}

H
[td-32]  
hjxilinx 已提交
1951 1952
// 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
1953 1954
static void doMergeTwoLevelData(STsdbReadHandle* pTsdbReadHandle, STableCheckInfo* pCheckInfo, SBlock* pBlock) {
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
1955
  SDataBlockInfo blockInfo = GET_FILE_DATA_BLOCK_INFO(pCheckInfo, pBlock);
H
Hongze Cheng 已提交
1956
  STsdbCfg*      pCfg = REPO_CFG(pTsdbReadHandle->pTsdb);
H
Haojun Liao 已提交
1957

1958
  initTableMemIterator(pTsdbReadHandle, pCheckInfo);
1959

1960 1961
  SDataCols* pCols = pTsdbReadHandle->rhelper.pDCols[0];
  assert(pCols->cols[0].type == TSDB_DATA_TYPE_TIMESTAMP && pCols->cols[0].colId == PRIMARYKEY_TIMESTAMP_COL_ID &&
dengyihao's avatar
dengyihao 已提交
1962
         cur->pos >= 0 && cur->pos < pBlock->numOfRows);
1963
  // Even Multi-Version supported, the records with duplicated TSKEY would be merged inside of tsdbLoadData interface.
1964
  TSKEY* tsArray = pCols->cols[0].pData;
dengyihao's avatar
dengyihao 已提交
1965 1966
  assert(pCols->numOfRows == pBlock->numOfRows && tsArray[0] == pBlock->keyFirst &&
         tsArray[pBlock->numOfRows - 1] == pBlock->keyLast);
1967

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

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

1974 1975
  int32_t numOfCols = (int32_t)(QH_GET_NUM_OF_COLS(pTsdbReadHandle));
  int32_t endPos = getEndPosInDataBlock(pTsdbReadHandle, &blockInfo);
H
Haojun Liao 已提交
1976

1977
  STimeWindow* pWin = &blockInfo.window;
H
Hongze Cheng 已提交
1978
  tsdbDebug("%p uid:%" PRIu64 " start merge data block, file block range:%" PRIu64 "-%" PRIu64
1979
            " rows:%d, start:%d, end:%d, %s", pTsdbReadHandle, pCheckInfo->tableId, pWin->skey, pWin->ekey, blockInfo.rows,
dengyihao's avatar
dengyihao 已提交
1980
            cur->pos, endPos, pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
1981

1982 1983
  // compared with the data from in-memory buffer, to generate the correct timestamp array list
  int32_t numOfRows = 0;
C
Cary Xu 已提交
1984
  int32_t curRow = 0;
H
Haojun Liao 已提交
1985

dengyihao's avatar
dengyihao 已提交
1986 1987
  int16_t   rv1 = -1;
  int16_t   rv2 = -1;
1988 1989
  STSchema* pSchema1 = NULL;
  STSchema* pSchema2 = NULL;
D
fix bug  
dapan1121 已提交
1990

H
Haojun Liao 已提交
1991 1992
  int32_t pos = cur->pos;
  cur->win = TSWINDOW_INITIALIZER;
1993
  bool adjustPos = false;
1994

1995 1996
  // no data in buffer, load data from file directly
  if (pCheckInfo->iiter == NULL && pCheckInfo->iter == NULL) {
1997
    copyAllRemainRowsFromFileBlock(pTsdbReadHandle, pCheckInfo, &blockInfo, endPos);
1998
    return;
1999
  } else if (pCheckInfo->iter != NULL || pCheckInfo->iiter != NULL) {
2000
    SSkipListNode* node = NULL;
C
Cary Xu 已提交
2001
    TSKEY          lastKeyAppend = TSKEY_INITIAL_VAL;
C
Cary Xu 已提交
2002

2003
    do {
H
Haojun Liao 已提交
2004
      STSRow* row2 = NULL;
C
Cary Xu 已提交
2005
      STSRow* row1 = getSRowInTableMem(pCheckInfo, pTsdbReadHandle->order, pCfg->update, &row2, TD_VER_MAX);
2006
      if (row1 == NULL) {
H
[td-32]  
hjxilinx 已提交
2007
        break;
2008
      }
2009

H
Haojun Liao 已提交
2010
      TSKEY key = TD_ROW_KEY(row1);
2011
      if ((key > pTsdbReadHandle->window.ekey && ascScan) || (key < pTsdbReadHandle->window.ekey && !ascScan)) {
2012 2013 2014
        break;
      }

2015 2016 2017 2018 2019 2020 2021
      if (adjustPos) {
        if (key == lastKeyAppend) {
          pos -= step;
        }
        adjustPos = false;
      }

2022 2023
      if (((pos > endPos || tsArray[pos] > pTsdbReadHandle->window.ekey) && ascScan) ||
          ((pos < endPos || tsArray[pos] < pTsdbReadHandle->window.ekey) && !ascScan)) {
2024 2025 2026
        break;
      }

2027
      if ((key < tsArray[pos] && ascScan) || (key > tsArray[pos] && !ascScan)) {
H
Haojun Liao 已提交
2028
        if (rv1 != TD_ROW_SVER(row1)) {
dengyihao's avatar
dengyihao 已提交
2029
          //          pSchema1 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row1));
H
Haojun Liao 已提交
2030
          rv1 = TD_ROW_SVER(row1);
C
Cary Xu 已提交
2031
        }
dengyihao's avatar
dengyihao 已提交
2032 2033
        if (row2 && rv2 != TD_ROW_SVER(row2)) {
          //          pSchema2 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row2));
H
Haojun Liao 已提交
2034
          rv2 = TD_ROW_SVER(row2);
2035
        }
dengyihao's avatar
dengyihao 已提交
2036

C
Cary Xu 已提交
2037 2038 2039
        numOfRows +=
            mergeTwoRowFromMem(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, &curRow, row1, row2, numOfCols,
                               pCheckInfo->tableId, pSchema1, pSchema2, pCfg->update, &lastKeyAppend);
2040 2041 2042
        if (cur->win.skey == TSKEY_INITIAL_VAL) {
          cur->win.skey = key;
        }
2043

2044
        cur->win.ekey = key;
dengyihao's avatar
dengyihao 已提交
2045
        cur->lastKey = key + step;
2046
        cur->mixBlock = true;
2047
        moveToNextRowInMem(pCheckInfo);
2048
      } else if (key == tsArray[pos]) {  // data in buffer has the same timestamp of data in file block, ignore it
C
Cary Xu 已提交
2049
#if 0
H
TD-1439  
Hongze Cheng 已提交
2050
        if (pCfg->update) {
dengyihao's avatar
dengyihao 已提交
2051
          if (pCfg->update == TD_ROW_PARTIAL_UPDATE) {
2052
            doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, numOfRows, pos, pos);
D
fix bug  
dapan1121 已提交
2053
          }
H
Haojun Liao 已提交
2054
          if (rv1 != TD_ROW_SVER(row1)) {
dengyihao's avatar
dengyihao 已提交
2055
            //            pSchema1 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row1));
H
Haojun Liao 已提交
2056
            rv1 = TD_ROW_SVER(row1);
2057
          }
dengyihao's avatar
dengyihao 已提交
2058 2059
          if (row2 && rv2 != TD_ROW_SVER(row2)) {
            //            pSchema2 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row2));
H
Haojun Liao 已提交
2060
            rv2 = TD_ROW_SVER(row2);
2061
          }
dengyihao's avatar
dengyihao 已提交
2062

2063
          bool forceSetNull = pCfg->update != TD_ROW_PARTIAL_UPDATE;
dengyihao's avatar
dengyihao 已提交
2064
          mergeTwoRowFromMem(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, numOfRows, row1, row2, numOfCols,
C
Cary Xu 已提交
2065
                             pCheckInfo->tableId, pSchema1, pSchema2, forceSetNull, &lastRowKey);
H
TD-1439  
Hongze Cheng 已提交
2066 2067 2068 2069 2070 2071 2072 2073 2074
          numOfRows += 1;
          if (cur->win.skey == TSKEY_INITIAL_VAL) {
            cur->win.skey = key;
          }

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

C
Cary Xu 已提交
2075 2076 2077 2078 2079 2080 2081
          moveToNextRowInMem(pCheckInfo);
          pos += step;
        } else {
          moveToNextRowInMem(pCheckInfo);
        }
#endif
        if (TD_SUPPORT_UPDATE(pCfg->update)) {
2082 2083 2084 2085
          if (lastKeyAppend != key) {
            if (lastKeyAppend != TSKEY_INITIAL_VAL) {
              ++curRow;
            }
2086
            lastKeyAppend = key;
2087 2088
          }
          // load data from file firstly
C
Cary Xu 已提交
2089
          numOfRows = doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, curRow, pos, pos);
C
Cary Xu 已提交
2090 2091 2092 2093 2094 2095 2096

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

          // still assign data into current row
2099
          numOfRows += mergeTwoRowFromMem(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, &curRow, row1, row2, numOfCols,
2100
                             pCheckInfo->tableId, pSchema1, pSchema2, pCfg->update, &lastKeyAppend);
C
Cary Xu 已提交
2101 2102 2103 2104 2105 2106 2107 2108 2109

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

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

H
TD-1439  
Hongze Cheng 已提交
2110
          moveToNextRowInMem(pCheckInfo);
2111

H
TD-1439  
Hongze Cheng 已提交
2112
          pos += step;
2113
          adjustPos = true;
H
TD-1439  
Hongze Cheng 已提交
2114
        } else {
2115
          // discard the memory record
H
TD-1439  
Hongze Cheng 已提交
2116 2117
          moveToNextRowInMem(pCheckInfo);
        }
2118
      } else if ((key > tsArray[pos] && ascScan) || (key < tsArray[pos] && !ascScan)) {
2119 2120 2121
        if (cur->win.skey == TSKEY_INITIAL_VAL) {
          cur->win.skey = tsArray[pos];
        }
2122

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

dengyihao's avatar
dengyihao 已提交
2126
        if (tsArray[end] == key) {  // the value of key in cache equals to the end timestamp value, ignore it
C
Cary Xu 已提交
2127
#if 0
2128
          if (pCfg->update == TD_ROW_DISCARD_UPDATE) {
H
Hongze Cheng 已提交
2129 2130 2131 2132
            moveToNextRowInMem(pCheckInfo);
          } else {
            end -= step;
          }
C
Cary Xu 已提交
2133 2134 2135 2136 2137 2138
#endif
          if (!TD_SUPPORT_UPDATE(pCfg->update)) {
            moveToNextRowInMem(pCheckInfo);
          } else {
            end -= step;
          }
H
Haojun Liao 已提交
2139
        }
2140

2141
        int32_t qstart = 0, qend = 0;
2142
        getQualifiedRowsPos(pTsdbReadHandle, pos, end, numOfRows, &qstart, &qend);
2143

2144
        if ((lastKeyAppend != TSKEY_INITIAL_VAL) && (lastKeyAppend != (ascScan ? tsArray[qstart] : tsArray[qend]))) {
C
Cary Xu 已提交
2145 2146
          ++curRow;
        }
2147

C
Cary Xu 已提交
2148
        numOfRows = doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, curRow, qstart, qend);
2149
        pos += (qend - qstart + 1) * step;
C
Cary Xu 已提交
2150 2151 2152
        if (numOfRows > 0) {
          curRow = numOfRows - 1;
        }
2153

2154
        cur->win.ekey = ascScan ? tsArray[qend] : tsArray[qstart];
dengyihao's avatar
dengyihao 已提交
2155
        cur->lastKey = cur->win.ekey + step;
C
Cary Xu 已提交
2156
        lastKeyAppend = cur->win.ekey;
2157
      }
2158
    } while (numOfRows < pTsdbReadHandle->outputCapacity);
H
Haojun Liao 已提交
2159

2160
    if (numOfRows < pTsdbReadHandle->outputCapacity) {
H
Haojun Liao 已提交
2161 2162 2163 2164
      /**
       * 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.
       */
2165
      if (node == NULL ||
2166 2167
          ((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)) {
2168 2169 2170 2171 2172
        // 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];
        }

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

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

2179
        cur->win.ekey = ascScan ? tsArray[end] : tsArray[start];
dengyihao's avatar
dengyihao 已提交
2180
        cur->lastKey = cur->win.ekey + step;
H
Haojun Liao 已提交
2181
        cur->mixBlock = true;
2182
      }
2183 2184
    }
  }
H
Haojun Liao 已提交
2185

2186 2187
  cur->blockCompleted = (((pos > endPos || cur->lastKey > pTsdbReadHandle->window.ekey) && ascScan) ||
       ((pos < endPos || cur->lastKey < pTsdbReadHandle->window.ekey) && !ascScan));
2188

2189
  if (!ascScan) {
wafwerar's avatar
wafwerar 已提交
2190
    TSWAP(cur->win.skey, cur->win.ekey);
2191
  }
2192

2193 2194
  updateInfoAfterMerge(pTsdbReadHandle, pCheckInfo, numOfRows, pos);
  doCheckGeneratedBlockRange(pTsdbReadHandle);
H
Haojun Liao 已提交
2195

dengyihao's avatar
dengyihao 已提交
2196 2197 2198
  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);
2199 2200
}

2201
int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order) {
H
[td-32]  
hjxilinx 已提交
2202
  int    firstPos, lastPos, midPos = -1;
H
Haojun Liao 已提交
2203
  int    numOfRows;
2204 2205
  TSKEY* keyList;

H
[td-32]  
hjxilinx 已提交
2206
  if (num <= 0) return -1;
2207 2208

  keyList = (TSKEY*)pValue;
H
[td-32]  
hjxilinx 已提交
2209 2210
  firstPos = 0;
  lastPos = num - 1;
2211

2212
  if (order == TSDB_ORDER_DESC) {
H
[td-32]  
hjxilinx 已提交
2213 2214 2215 2216 2217
    // 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;
2218

H
Haojun Liao 已提交
2219 2220
      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1) + firstPos;
2221

H
[td-32]  
hjxilinx 已提交
2222 2223 2224 2225 2226 2227 2228 2229
      if (key < keyList[midPos]) {
        lastPos = midPos - 1;
      } else if (key > keyList[midPos]) {
        firstPos = midPos + 1;
      } else {
        break;
      }
    }
2230

H
[td-32]  
hjxilinx 已提交
2231 2232 2233 2234 2235
  } 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;
2236

H
[td-32]  
hjxilinx 已提交
2237 2238 2239 2240 2241 2242 2243
      if (key > keyList[lastPos]) {
        lastPos = lastPos + 1;
        if (lastPos >= num)
          return -1;
        else
          return lastPos;
      }
2244

H
Haojun Liao 已提交
2245 2246
      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1) + firstPos;
2247

H
[td-32]  
hjxilinx 已提交
2248 2249 2250 2251 2252 2253 2254 2255 2256
      if (key < keyList[midPos]) {
        lastPos = midPos - 1;
      } else if (key > keyList[midPos]) {
        firstPos = midPos + 1;
      } else {
        break;
      }
    }
  }
2257

H
[td-32]  
hjxilinx 已提交
2258 2259 2260
  return midPos;
}

2261
static void cleanBlockOrderSupporter(SBlockOrderSupporter* pSupporter, int32_t numOfTables) {
wafwerar's avatar
wafwerar 已提交
2262 2263
  taosMemoryFreeClear(pSupporter->numOfBlocksPerTable);
  taosMemoryFreeClear(pSupporter->blockIndexArray);
2264 2265

  for (int32_t i = 0; i < numOfTables; ++i) {
H
Haojun Liao 已提交
2266
    STableBlockInfo* pBlockInfo = pSupporter->pDataBlockInfo[i];
wafwerar's avatar
wafwerar 已提交
2267
    taosMemoryFreeClear(pBlockInfo);
2268 2269
  }

wafwerar's avatar
wafwerar 已提交
2270
  taosMemoryFreeClear(pSupporter->pDataBlockInfo);
2271 2272 2273 2274 2275 2276 2277 2278 2279 2280 2281
}

static int32_t dataBlockOrderCompar(const void* pLeft, const void* pRight, void* param) {
  int32_t leftTableIndex = *(int32_t*)pLeft;
  int32_t rightTableIndex = *(int32_t*)pRight;

  SBlockOrderSupporter* pSupporter = (SBlockOrderSupporter*)param;

  int32_t leftTableBlockIndex = pSupporter->blockIndexArray[leftTableIndex];
  int32_t rightTableBlockIndex = pSupporter->blockIndexArray[rightTableIndex];

2282
  if (leftTableBlockIndex > pSupporter->numOfBlocksPerTable[leftTableIndex]) {
2283 2284
    /* left block is empty */
    return 1;
2285
  } else if (rightTableBlockIndex > pSupporter->numOfBlocksPerTable[rightTableIndex]) {
2286 2287 2288 2289 2290 2291 2292
    /* right block is empty */
    return -1;
  }

  STableBlockInfo* pLeftBlockInfoEx = &pSupporter->pDataBlockInfo[leftTableIndex][leftTableBlockIndex];
  STableBlockInfo* pRightBlockInfoEx = &pSupporter->pDataBlockInfo[rightTableIndex][rightTableBlockIndex];

H
Haojun Liao 已提交
2293
  //    assert(pLeftBlockInfoEx->compBlock->offset != pRightBlockInfoEx->compBlock->offset);
dengyihao's avatar
dengyihao 已提交
2294
#if 0  // TODO: temporarily comment off requested by Dr. Liao
H
Haojun Liao 已提交
2295 2296
  if (pLeftBlockInfoEx->compBlock->offset == pRightBlockInfoEx->compBlock->offset &&
      pLeftBlockInfoEx->compBlock->last == pRightBlockInfoEx->compBlock->last) {
B
Bomin Zhang 已提交
2297
    tsdbError("error in header file, two block with same offset:%" PRId64, (int64_t)pLeftBlockInfoEx->compBlock->offset);
2298
  }
H
Haojun Liao 已提交
2299
#endif
2300

H
Haojun Liao 已提交
2301
  return pLeftBlockInfoEx->compBlock->offset > pRightBlockInfoEx->compBlock->offset ? 1 : -1;
2302 2303
}

2304
static int32_t createDataBlocksInfo(STsdbReadHandle* pTsdbReadHandle, int32_t numOfBlocks, int32_t* numOfAllocBlocks) {
H
Haojun Liao 已提交
2305 2306
  size_t size = sizeof(STableBlockInfo) * numOfBlocks;

2307 2308
  if (pTsdbReadHandle->allocSize < size) {
    pTsdbReadHandle->allocSize = (int32_t)size;
wafwerar's avatar
wafwerar 已提交
2309
    char* tmp = taosMemoryRealloc(pTsdbReadHandle->pDataBlockInfo, pTsdbReadHandle->allocSize);
H
Haojun Liao 已提交
2310 2311 2312 2313
    if (tmp == NULL) {
      return TSDB_CODE_TDB_OUT_OF_MEMORY;
    }

dengyihao's avatar
dengyihao 已提交
2314
    pTsdbReadHandle->pDataBlockInfo = (STableBlockInfo*)tmp;
2315 2316
  }

2317
  memset(pTsdbReadHandle->pDataBlockInfo, 0, size);
2318 2319
  *numOfAllocBlocks = numOfBlocks;

H
Haojun Liao 已提交
2320
  // access data blocks according to the offset of each block in asc/desc order.
2321
  int32_t numOfTables = (int32_t)taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
2322

2323 2324
  SBlockOrderSupporter sup = {0};
  sup.numOfTables = numOfTables;
wafwerar's avatar
wafwerar 已提交
2325 2326 2327
  sup.numOfBlocksPerTable = taosMemoryCalloc(1, sizeof(int32_t) * numOfTables);
  sup.blockIndexArray = taosMemoryCalloc(1, sizeof(int32_t) * numOfTables);
  sup.pDataBlockInfo = taosMemoryCalloc(1, POINTER_BYTES * numOfTables);
2328

2329
  if (sup.numOfBlocksPerTable == NULL || sup.blockIndexArray == NULL || sup.pDataBlockInfo == NULL) {
2330
    cleanBlockOrderSupporter(&sup, 0);
2331
    return TSDB_CODE_TDB_OUT_OF_MEMORY;
2332
  }
H
Haojun Liao 已提交
2333

2334
  int32_t cnt = 0;
2335
  int32_t numOfQualTables = 0;
H
Haojun Liao 已提交
2336

2337
  for (int32_t j = 0; j < numOfTables; ++j) {
2338
    STableCheckInfo* pTableCheck = (STableCheckInfo*)taosArrayGet(pTsdbReadHandle->pTableCheckInfo, j);
2339 2340 2341
    if (pTableCheck->numOfBlocks <= 0) {
      continue;
    }
H
Haojun Liao 已提交
2342

H
refact  
Hongze Cheng 已提交
2343
    SBlock* pBlock = pTableCheck->pCompInfo->blocks;
2344
    sup.numOfBlocksPerTable[numOfQualTables] = pTableCheck->numOfBlocks;
2345

wafwerar's avatar
wafwerar 已提交
2346
    char* buf = taosMemoryMalloc(sizeof(STableBlockInfo) * pTableCheck->numOfBlocks);
2347
    if (buf == NULL) {
2348
      cleanBlockOrderSupporter(&sup, numOfQualTables);
2349
      return TSDB_CODE_TDB_OUT_OF_MEMORY;
2350 2351
    }

2352
    sup.pDataBlockInfo[numOfQualTables] = (STableBlockInfo*)buf;
2353 2354

    for (int32_t k = 0; k < pTableCheck->numOfBlocks; ++k) {
H
Haojun Liao 已提交
2355
      STableBlockInfo* pBlockInfo = &sup.pDataBlockInfo[numOfQualTables][k];
2356

H
Haojun Liao 已提交
2357 2358
      pBlockInfo->compBlock = &pBlock[k];
      pBlockInfo->pTableCheckInfo = pTableCheck;
2359 2360 2361
      cnt++;
    }

2362
    numOfQualTables++;
2363 2364
  }

H
Haojun Liao 已提交
2365
  assert(numOfBlocks == cnt);
2366

H
Haojun Liao 已提交
2367 2368
  // since there is only one table qualified, blocks are not sorted
  if (numOfQualTables == 1) {
2369
    memcpy(pTsdbReadHandle->pDataBlockInfo, sup.pDataBlockInfo[0], sizeof(STableBlockInfo) * numOfBlocks);
H
Haojun Liao 已提交
2370
    cleanBlockOrderSupporter(&sup, numOfQualTables);
2371

H
Haojun Liao 已提交
2372
    tsdbDebug("%p create data blocks info struct completed for 1 table, %d blocks not sorted %s", pTsdbReadHandle, cnt,
dengyihao's avatar
dengyihao 已提交
2373
              pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
2374 2375
    return TSDB_CODE_SUCCESS;
  }
2376

H
Haojun Liao 已提交
2377
  tsdbDebug("%p create data blocks info struct completed, %d blocks in %d tables %s", pTsdbReadHandle, cnt,
dengyihao's avatar
dengyihao 已提交
2378
            numOfQualTables, pTsdbReadHandle->idStr);
2379

2380
  assert(cnt <= numOfBlocks && numOfQualTables <= numOfTables);  // the pTableQueryInfo[j]->numOfBlocks may be 0
2381
  sup.numOfTables = numOfQualTables;
2382

2383
  SMultiwayMergeTreeInfo* pTree = NULL;
dengyihao's avatar
dengyihao 已提交
2384
  uint8_t                 ret = tMergeTreeCreate(&pTree, sup.numOfTables, &sup, dataBlockOrderCompar);
2385 2386
  if (ret != TSDB_CODE_SUCCESS) {
    cleanBlockOrderSupporter(&sup, numOfTables);
2387
    return TSDB_CODE_TDB_OUT_OF_MEMORY;
2388 2389 2390 2391 2392
  }

  int32_t numOfTotal = 0;

  while (numOfTotal < cnt) {
2393
    int32_t pos = tMergeTreeGetChosenIndex(pTree);
2394 2395
    int32_t index = sup.blockIndexArray[pos]++;

H
Haojun Liao 已提交
2396
    STableBlockInfo* pBlocksInfo = sup.pDataBlockInfo[pos];
2397
    pTsdbReadHandle->pDataBlockInfo[numOfTotal++] = pBlocksInfo[index];
2398 2399

    // set data block index overflow, in order to disable the offset comparator
2400 2401
    if (sup.blockIndexArray[pos] >= sup.numOfBlocksPerTable[pos]) {
      sup.blockIndexArray[pos] = sup.numOfBlocksPerTable[pos] + 1;
2402
    }
2403

H
Haojun Liao 已提交
2404
    tMergeTreeAdjust(pTree, tMergeTreeGetAdjustIndex(pTree));
2405 2406 2407 2408 2409
  }

  /*
   * available when no import exists
   * for(int32_t i = 0; i < cnt - 1; ++i) {
H
Haojun Liao 已提交
2410
   *   assert((*pDataBlockInfo)[i].compBlock->offset < (*pDataBlockInfo)[i+1].compBlock->offset);
2411 2412 2413
   * }
   */

H
Haojun Liao 已提交
2414
  tsdbDebug("%p %d data blocks sort completed, %s", pTsdbReadHandle, cnt, pTsdbReadHandle->idStr);
2415
  cleanBlockOrderSupporter(&sup, numOfTables);
wafwerar's avatar
wafwerar 已提交
2416
  taosMemoryFree(pTree);
2417 2418 2419 2420

  return TSDB_CODE_SUCCESS;
}

2421
static int32_t getFirstFileDataBlock(STsdbReadHandle* pTsdbReadHandle, bool* exists);
H
Haojun Liao 已提交
2422

2423
static int32_t getDataBlock(STsdbReadHandle* pTsdbReadHandle, STableBlockInfo* pNext, bool* exists) {
dengyihao's avatar
dengyihao 已提交
2424
  int32_t        step = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 1 : -1;
2425
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
H
Haojun Liao 已提交
2426

dengyihao's avatar
dengyihao 已提交
2427
  while (1) {
2428
    int32_t code = loadFileDataBlock(pTsdbReadHandle, pNext->compBlock, pNext->pTableCheckInfo, exists);
H
Haojun Liao 已提交
2429 2430 2431 2432
    if (code != TSDB_CODE_SUCCESS || *exists) {
      return code;
    }

2433 2434
    if ((cur->slot == pTsdbReadHandle->numOfBlocks - 1 && ASCENDING_TRAVERSE(pTsdbReadHandle->order)) ||
        (cur->slot == 0 && !ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
H
Haojun Liao 已提交
2435
      // all data blocks in current file has been checked already, try next file if exists
2436
      return getFirstFileDataBlock(pTsdbReadHandle, exists);
H
Haojun Liao 已提交
2437 2438 2439 2440
    } else {  // next block of the same file
      cur->slot += step;
      cur->mixBlock = false;
      cur->blockCompleted = false;
2441
      pNext = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
H
Haojun Liao 已提交
2442 2443 2444 2445
    }
  }
}

2446 2447 2448
static int32_t getFirstFileDataBlock(STsdbReadHandle* pTsdbReadHandle, bool* exists) {
  pTsdbReadHandle->numOfBlocks = 0;
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
H
Haojun Liao 已提交
2449 2450 2451

  int32_t code = TSDB_CODE_SUCCESS;

2452
  int32_t numOfBlocks = 0;
2453
  int32_t numOfTables = (int32_t)taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
2454

C
Cary Xu 已提交
2455
  STsdbKeepCfg* pCfg = REPO_KEEP_CFG(pTsdbReadHandle->pTsdb);
C
Cary Xu 已提交
2456
  STimeWindow   win = TSWINDOW_INITIALIZER;
2457

H
Hongze Cheng 已提交
2458
  while (true) {
2459
    tsdbRLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2460

2461 2462
    if ((pTsdbReadHandle->pFileGroup = tsdbFSIterNext(&pTsdbReadHandle->fileIter)) == NULL) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2463 2464 2465
      break;
    }

H
refact  
Hongze Cheng 已提交
2466
    tsdbGetFidKeyRange(pCfg->days, pCfg->precision, pTsdbReadHandle->pFileGroup->fid, &win.skey, &win.ekey);
2467 2468

    // current file are not overlapped with query time window, ignore remain files
2469 2470 2471
    if ((ASCENDING_TRAVERSE(pTsdbReadHandle->order) && win.skey > pTsdbReadHandle->window.ekey) ||
        (!ASCENDING_TRAVERSE(pTsdbReadHandle->order) && win.ekey < pTsdbReadHandle->window.ekey)) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2472 2473
      tsdbDebug("%p remain files are not qualified for qrange:%" PRId64 "-%" PRId64 ", ignore, %s", pTsdbReadHandle,
                pTsdbReadHandle->window.skey, pTsdbReadHandle->window.ekey, pTsdbReadHandle->idStr);
2474 2475
      pTsdbReadHandle->pFileGroup = NULL;
      assert(pTsdbReadHandle->numOfBlocks == 0);
2476 2477 2478
      break;
    }

2479 2480
    if (tsdbSetAndOpenReadFSet(&pTsdbReadHandle->rhelper, pTsdbReadHandle->pFileGroup) < 0) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2481 2482 2483 2484
      code = terrno;
      break;
    }

2485
    tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2486

2487
    if (tsdbLoadBlockIdx(&pTsdbReadHandle->rhelper) < 0) {
H
Hongze Cheng 已提交
2488 2489 2490 2491
      code = terrno;
      break;
    }

2492
    if ((code = getFileCompInfo(pTsdbReadHandle, &numOfBlocks)) != TSDB_CODE_SUCCESS) {
2493 2494
      break;
    }
H
Haojun Liao 已提交
2495

H
Haojun Liao 已提交
2496 2497
    tsdbDebug("%p %d blocks found in file for %d table(s), fid:%d, %s", pTsdbReadHandle, numOfBlocks, numOfTables,
              pTsdbReadHandle->pFileGroup->fid, pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
2498

2499 2500 2501 2502
    assert(numOfBlocks >= 0);
    if (numOfBlocks == 0) {
      continue;
    }
H
Haojun Liao 已提交
2503

2504
    // todo return error code to query engine
dengyihao's avatar
dengyihao 已提交
2505 2506
    if ((code = createDataBlocksInfo(pTsdbReadHandle, numOfBlocks, &pTsdbReadHandle->numOfBlocks)) !=
        TSDB_CODE_SUCCESS) {
2507 2508
      break;
    }
H
Haojun Liao 已提交
2509

2510 2511
    assert(numOfBlocks >= pTsdbReadHandle->numOfBlocks);
    if (pTsdbReadHandle->numOfBlocks > 0) {
2512 2513 2514
      break;
    }
  }
H
Haojun Liao 已提交
2515

2516
  // no data in file anymore
2517
  if (pTsdbReadHandle->numOfBlocks <= 0 || code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2518
    if (code == TSDB_CODE_SUCCESS) {
2519
      assert(pTsdbReadHandle->pFileGroup == NULL);
H
Haojun Liao 已提交
2520 2521
    }

D
dapan1121 已提交
2522
    cur->fid = INT32_MIN;  // denote that there are no data in file anymore
H
Haojun Liao 已提交
2523 2524
    *exists = false;
    return code;
2525
  }
H
Haojun Liao 已提交
2526

2527
  assert(pTsdbReadHandle->pFileGroup != NULL && pTsdbReadHandle->numOfBlocks > 0);
dengyihao's avatar
dengyihao 已提交
2528
  cur->slot = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 0 : pTsdbReadHandle->numOfBlocks - 1;
2529
  cur->fid = pTsdbReadHandle->pFileGroup->fid;
H
Haojun Liao 已提交
2530

2531
  STableBlockInfo* pBlockInfo = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
2532
  return getDataBlock(pTsdbReadHandle, pBlockInfo, exists);
H
Haojun Liao 已提交
2533 2534 2535 2536 2537 2538 2539
}

static bool isEndFileDataBlock(SQueryFilePos* cur, int32_t numOfBlocks, bool ascTrav) {
  assert(cur != NULL && numOfBlocks > 0);
  return (cur->slot == numOfBlocks - 1 && ascTrav) || (cur->slot == 0 && !ascTrav);
}

2540
static void moveToNextDataBlockInCurrentFile(STsdbReadHandle* pTsdbReadHandle) {
dengyihao's avatar
dengyihao 已提交
2541
  int32_t step = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 1 : -1;
H
Haojun Liao 已提交
2542

2543 2544
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
  assert(cur->slot < pTsdbReadHandle->numOfBlocks && cur->slot >= 0);
H
Haojun Liao 已提交
2545 2546

  cur->slot += step;
dengyihao's avatar
dengyihao 已提交
2547
  cur->mixBlock = false;
H
Haojun Liao 已提交
2548
  cur->blockCompleted = false;
2549
}
H
Haojun Liao 已提交
2550 2551

int32_t tsdbGetFileBlocksDistInfo(tsdbReaderT* queryHandle, STableBlockDistInfo* pTableBlockInfo) {
dengyihao's avatar
dengyihao 已提交
2552
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)queryHandle;
H
Haojun Liao 已提交
2553

H
Haojun Liao 已提交
2554
  pTableBlockInfo->totalSize = 0;
2555
  pTableBlockInfo->totalRows = 0;
H
Haojun Liao 已提交
2556

2557
  STsdbFS* pFileHandle = REPO_FS(pTsdbReadHandle->pTsdb);
H
Haojun Liao 已提交
2558 2559

  // find the start data block in file
2560
  pTsdbReadHandle->locateStart = true;
C
Cary Xu 已提交
2561 2562
  STsdbKeepCfg* pCfg = REPO_KEEP_CFG(pTsdbReadHandle->pTsdb);
  int32_t       fid = getFileIdFromKey(pTsdbReadHandle->window.skey, pCfg->days, pCfg->precision);
H
Haojun Liao 已提交
2563 2564

  tsdbRLockFS(pFileHandle);
2565 2566
  tsdbFSIterInit(&pTsdbReadHandle->fileIter, pFileHandle, pTsdbReadHandle->order);
  tsdbFSIterSeek(&pTsdbReadHandle->fileIter, fid);
H
Haojun Liao 已提交
2567 2568
  tsdbUnLockFS(pFileHandle);

H
Haojun Liao 已提交
2569
  pTableBlockInfo->numOfFiles += 1;
H
Haojun Liao 已提交
2570

H
Haojun Liao 已提交
2571
  int32_t     code = TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
2572
  int32_t     numOfBlocks = 0;
2573
  int32_t     numOfTables = (int32_t)taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
dengyihao's avatar
dengyihao 已提交
2574
  int         defaultRows = 4096;  // TSDB_DEFAULT_BLOCK_ROWS(pCfg->maxRowsPerFileBlock);
H
Haojun Liao 已提交
2575 2576
  STimeWindow win = TSWINDOW_INITIALIZER;

H
Haojun Liao 已提交
2577 2578
  bool ascTraverse = ASCENDING_TRAVERSE(pTsdbReadHandle->order);

H
Haojun Liao 已提交
2579 2580
  while (true) {
    numOfBlocks = 0;
2581
    tsdbRLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2582

2583 2584
    if ((pTsdbReadHandle->pFileGroup = tsdbFSIterNext(&pTsdbReadHandle->fileIter)) == NULL) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2585 2586 2587
      break;
    }

H
refact  
Hongze Cheng 已提交
2588
    tsdbGetFidKeyRange(pCfg->days, pCfg->precision, pTsdbReadHandle->pFileGroup->fid, &win.skey, &win.ekey);
H
Haojun Liao 已提交
2589 2590

    // current file are not overlapped with query time window, ignore remain files
dengyihao's avatar
dengyihao 已提交
2591 2592
    if ((ascTraverse && win.skey > pTsdbReadHandle->window.ekey) ||
        (!ascTraverse && win.ekey < pTsdbReadHandle->window.ekey)) {
2593
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2594 2595
      tsdbDebug("%p remain files are not qualified for qrange:%" PRId64 "-%" PRId64 ", ignore, %s", pTsdbReadHandle,
                pTsdbReadHandle->window.skey, pTsdbReadHandle->window.ekey, pTsdbReadHandle->idStr);
2596
      pTsdbReadHandle->pFileGroup = NULL;
H
Haojun Liao 已提交
2597 2598 2599
      break;
    }

H
Haojun Liao 已提交
2600
    pTableBlockInfo->numOfFiles += 1;
2601 2602
    if (tsdbSetAndOpenReadFSet(&pTsdbReadHandle->rhelper, pTsdbReadHandle->pFileGroup) < 0) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2603 2604 2605 2606
      code = terrno;
      break;
    }

2607
    tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2608

2609
    if (tsdbLoadBlockIdx(&pTsdbReadHandle->rhelper) < 0) {
H
Haojun Liao 已提交
2610 2611 2612 2613
      code = terrno;
      break;
    }

2614
    if ((code = getFileCompInfo(pTsdbReadHandle, &numOfBlocks)) != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2615 2616 2617
      break;
    }

H
Haojun Liao 已提交
2618 2619
    tsdbDebug("%p %d blocks found in file for %d table(s), fid:%d, %s", pTsdbReadHandle, numOfBlocks, numOfTables,
              pTsdbReadHandle->pFileGroup->fid, pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
2620 2621 2622 2623 2624 2625

    if (numOfBlocks == 0) {
      continue;
    }

    for (int32_t i = 0; i < numOfTables; ++i) {
2626
      STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, i);
H
Haojun Liao 已提交
2627 2628 2629

      SBlock* pBlock = pCheckInfo->pCompInfo->blocks;
      for (int32_t j = 0; j < pCheckInfo->numOfBlocks; ++j) {
H
Haojun Liao 已提交
2630
        pTableBlockInfo->totalSize += pBlock[j].len;
H
Haojun Liao 已提交
2631

H
Haojun Liao 已提交
2632
        int32_t numOfRows = pBlock[j].numOfRows;
2633
        pTableBlockInfo->totalRows += numOfRows;
H
Haojun Liao 已提交
2634 2635 2636 2637 2638 2639 2640 2641 2642 2643 2644
        if (numOfRows > pTableBlockInfo->maxRows) {
          pTableBlockInfo->maxRows = numOfRows;
        }

        if (numOfRows < pTableBlockInfo->minRows) {
          pTableBlockInfo->minRows = numOfRows;
        }

        if (numOfRows < defaultRows) {
          pTableBlockInfo->numOfSmallBlocks += 1;
        }
dengyihao's avatar
dengyihao 已提交
2645 2646 2647
        //        int32_t  stepIndex = (numOfRows-1)/TSDB_BLOCK_DIST_STEP_ROWS;
        //        SFileBlockInfo *blockInfo = (SFileBlockInfo*)taosArrayGet(pTableBlockInfo->dataBlockInfos, stepIndex);
        //        blockInfo->numBlocksOfStep++;
H
Haojun Liao 已提交
2648 2649 2650 2651 2652 2653 2654
      }
    }
  }

  return code;
}

2655 2656 2657
static int32_t getDataBlocksInFiles(STsdbReadHandle* pTsdbReadHandle, bool* exists) {
  STsdbFS*       pFileHandle = REPO_FS(pTsdbReadHandle->pTsdb);
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
2658 2659

  // find the start data block in file
2660 2661
  if (!pTsdbReadHandle->locateStart) {
    pTsdbReadHandle->locateStart = true;
C
Cary Xu 已提交
2662 2663
    STsdbKeepCfg* pCfg = REPO_KEEP_CFG(pTsdbReadHandle->pTsdb);
    int32_t       fid = getFileIdFromKey(pTsdbReadHandle->window.skey, pCfg->days, pCfg->precision);
H
Haojun Liao 已提交
2664

H
Hongze Cheng 已提交
2665
    tsdbRLockFS(pFileHandle);
2666 2667
    tsdbFSIterInit(&pTsdbReadHandle->fileIter, pFileHandle, pTsdbReadHandle->order);
    tsdbFSIterSeek(&pTsdbReadHandle->fileIter, fid);
H
Hongze Cheng 已提交
2668
    tsdbUnLockFS(pFileHandle);
2669

2670
    return getFirstFileDataBlock(pTsdbReadHandle, exists);
2671
  } else {
2672
    // check if current file block is all consumed
2673
    STableBlockInfo* pBlockInfo = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
2674
    STableCheckInfo* pCheckInfo = pBlockInfo->pTableCheckInfo;
H
Haojun Liao 已提交
2675

2676
    // current block is done, try next
H
Haojun Liao 已提交
2677
    if ((!cur->mixBlock) || cur->blockCompleted) {
H
Haojun Liao 已提交
2678
      // all data blocks in current file has been checked already, try next file if exists
2679
    } else {
H
Haojun Liao 已提交
2680 2681
      tsdbDebug("%p continue in current data block, index:%d, pos:%d, %s", pTsdbReadHandle, cur->slot, cur->pos,
                pTsdbReadHandle->idStr);
2682 2683
      int32_t code = handleDataMergeIfNeeded(pTsdbReadHandle, pBlockInfo->compBlock, pCheckInfo);
      *exists = (pTsdbReadHandle->realNumOfRows > 0);
H
Haojun Liao 已提交
2684

H
Haojun Liao 已提交
2685 2686 2687 2688 2689 2690 2691
      if (code != TSDB_CODE_SUCCESS || *exists) {
        return code;
      }
    }

    // current block is empty, try next block in file
    // all data blocks in current file has been checked already, try next file if exists
2692 2693
    if (isEndFileDataBlock(cur, pTsdbReadHandle->numOfBlocks, ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
      return getFirstFileDataBlock(pTsdbReadHandle, exists);
H
Haojun Liao 已提交
2694
    } else {
2695 2696
      moveToNextDataBlockInCurrentFile(pTsdbReadHandle);
      STableBlockInfo* pNext = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
2697
      return getDataBlock(pTsdbReadHandle, pNext, exists);
2698 2699
    }
  }
2700 2701
}

2702 2703
static bool doHasDataInBuffer(STsdbReadHandle* pTsdbReadHandle) {
  size_t numOfTables = taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
dengyihao's avatar
dengyihao 已提交
2704

2705 2706
  while (pTsdbReadHandle->activeIndex < numOfTables) {
    if (hasMoreDataInCache(pTsdbReadHandle)) {
2707 2708
      return true;
    }
H
Haojun Liao 已提交
2709

2710
    pTsdbReadHandle->activeIndex += 1;
2711
  }
H
Haojun Liao 已提交
2712

2713 2714 2715
  return false;
}

dengyihao's avatar
dengyihao 已提交
2716
// todo not unref yet, since it is not support multi-group interpolation query
H
Haojun Liao 已提交
2717
static UNUSED_FUNC void changeQueryHandleForInterpQuery(tsdbReaderT pHandle) {
H
Haojun Liao 已提交
2718
  // filter the queried time stamp in the first place
dengyihao's avatar
dengyihao 已提交
2719
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle;
H
Haojun Liao 已提交
2720 2721

  // starts from the buffer in case of descending timestamp order check data blocks
2722
  size_t numOfTables = taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
H
Haojun Liao 已提交
2723 2724

  int32_t i = 0;
dengyihao's avatar
dengyihao 已提交
2725
  while (i < numOfTables) {
2726
    STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, i);
H
Haojun Liao 已提交
2727 2728

    // the first qualified table for interpolation query
dengyihao's avatar
dengyihao 已提交
2729 2730 2731 2732
    //    if ((pTsdbReadHandle->window.skey <= pCheckInfo->pTableObj->lastKey) &&
    //        (pCheckInfo->pTableObj->lastKey != TSKEY_INITIAL_VAL)) {
    //      break;
    //    }
H
Haojun Liao 已提交
2733 2734 2735 2736 2737 2738 2739 2740 2741

    i++;
  }

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

dengyihao's avatar
dengyihao 已提交
2742
  STableCheckInfo info = *(STableCheckInfo*)taosArrayGet(pTsdbReadHandle->pTableCheckInfo, i);
2743
  taosArrayClear(pTsdbReadHandle->pTableCheckInfo);
H
Haojun Liao 已提交
2744

2745 2746
  info.lastKey = pTsdbReadHandle->window.skey;
  taosArrayPush(pTsdbReadHandle->pTableCheckInfo, &info);
H
Haojun Liao 已提交
2747 2748 2749
}

static int tsdbReadRowsFromCache(STableCheckInfo* pCheckInfo, TSKEY maxKey, int maxRowsToRead, STimeWindow* win,
2750
                                 STsdbReadHandle* pTsdbReadHandle) {
dengyihao's avatar
dengyihao 已提交
2751
  int       numOfRows = 0;
C
Cary Xu 已提交
2752
  int       curRows = 0;
dengyihao's avatar
dengyihao 已提交
2753
  int32_t   numOfCols = (int32_t)taosArrayGetSize(pTsdbReadHandle->pColumns);
H
Hongze Cheng 已提交
2754
  STsdbCfg* pCfg = REPO_CFG(pTsdbReadHandle->pTsdb);
H
Haojun Liao 已提交
2755 2756
  win->skey = TSKEY_INITIAL_VAL;

dengyihao's avatar
dengyihao 已提交
2757 2758
  int64_t   st = taosGetTimestampUs();
  int16_t   rv = -1;
D
fix bug  
dapan1121 已提交
2759
  STSchema* pSchema = NULL;
C
Cary Xu 已提交
2760 2761
  TSKEY     lastRowKey = TSKEY_INITIAL_VAL;

H
Haojun Liao 已提交
2762
  do {
C
Cary Xu 已提交
2763
    STSRow* row = getSRowInTableMem(pCheckInfo, pTsdbReadHandle->order, pCfg->update, NULL, TD_VER_MAX);
H
Haojun Liao 已提交
2764 2765 2766 2767
    if (row == NULL) {
      break;
    }

H
Haojun Liao 已提交
2768
    TSKEY key = TD_ROW_KEY(row);
dengyihao's avatar
dengyihao 已提交
2769 2770 2771 2772
    if ((key > maxKey && ASCENDING_TRAVERSE(pTsdbReadHandle->order)) ||
        (key < maxKey && !ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
      tsdbDebug("%p key:%" PRIu64 " beyond qrange:%" PRId64 " - %" PRId64 ", no more data in buffer", pTsdbReadHandle,
                key, pTsdbReadHandle->window.skey, pTsdbReadHandle->window.ekey);
H
Haojun Liao 已提交
2773 2774 2775 2776 2777 2778 2779 2780 2781

      break;
    }

    if (win->skey == INT64_MIN) {
      win->skey = key;
    }

    win->ekey = key;
H
Haojun Liao 已提交
2782
    if (rv != TD_ROW_SVER(row)) {
2783
      pSchema = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), pCheckInfo->tableId, TD_ROW_SVER(row));
H
Haojun Liao 已提交
2784
      rv = TD_ROW_SVER(row);
D
fix bug  
dapan1121 已提交
2785
    }
C
Cary Xu 已提交
2786 2787
    numOfRows += mergeTwoRowFromMem(pTsdbReadHandle, maxRowsToRead, &curRows, row, NULL, numOfCols, pCheckInfo->tableId,
                                    pSchema, NULL, pCfg->update, &lastRowKey);
H
Haojun Liao 已提交
2788

C
Cary Xu 已提交
2789
    if (numOfRows >= maxRowsToRead) {
H
Haojun Liao 已提交
2790 2791 2792 2793
      moveToNextRowInMem(pCheckInfo);
      break;
    }

dengyihao's avatar
dengyihao 已提交
2794
  } while (moveToNextRowInMem(pCheckInfo));
H
Haojun Liao 已提交
2795

C
Cary Xu 已提交
2796
  taosMemoryFreeClear(pSchema);  // free the STSChema
H
Haojun Liao 已提交
2797 2798 2799
  assert(numOfRows <= maxRowsToRead);

  int64_t elapsedTime = taosGetTimestampUs() - st;
dengyihao's avatar
dengyihao 已提交
2800 2801
  tsdbDebug("%p build data block from cache completed, elapsed time:%" PRId64 " us, numOfRows:%d, numOfCols:%d, %s",
            pTsdbReadHandle, elapsedTime, numOfRows, numOfCols, pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
2802 2803 2804 2805

  return numOfRows;
}

2806 2807
static int32_t getAllTableList(SMeta* pMeta, uint64_t uid, SArray* list) {
  SMCtbCursor* pCur = metaOpenCtbCursor(pMeta, uid);
H
Haojun Liao 已提交
2808

2809 2810 2811 2812 2813
  while (1) {
    tb_uid_t id = metaCtbCursorNext(pCur);
    if (id == 0) {
      break;
    }
H
Haojun Liao 已提交
2814

2815
    STableKeyInfo info = {.lastKey = TSKEY_INITIAL_VAL, uid = id};
H
Haojun Liao 已提交
2816 2817 2818
    taosArrayPush(list, &info);
  }

C
Cary Xu 已提交
2819
  metaCloseCtbCursor(pCur);
H
Haojun Liao 已提交
2820 2821 2822 2823 2824 2825 2826 2827
  return TSDB_CODE_SUCCESS;
}

static void destroyHelper(void* param) {
  if (param == NULL) {
    return;
  }

dengyihao's avatar
dengyihao 已提交
2828 2829 2830 2831 2832 2833
  //  tQueryInfo* pInfo = (tQueryInfo*)param;
  //  if (pInfo->optr != TSDB_RELATION_IN) {
  //    taosMemoryFreeClear(pInfo->q);
  //  } else {
  //    taosHashCleanup((SHashObj *)(pInfo->q));
  //  }
H
Haojun Liao 已提交
2834

wafwerar's avatar
wafwerar 已提交
2835
  taosMemoryFree(param);
H
Haojun Liao 已提交
2836 2837
}

dengyihao's avatar
dengyihao 已提交
2838 2839
#define TSDB_PREV_ROW 0x1
#define TSDB_NEXT_ROW 0x2
2840

dengyihao's avatar
dengyihao 已提交
2841
static bool loadBlockOfActiveTable(STsdbReadHandle* pTsdbReadHandle) {
2842
  if (pTsdbReadHandle->checkFiles) {
H
Haojun Liao 已提交
2843 2844
    // check if the query range overlaps with the file data block
    bool exists = true;
H
Haojun Liao 已提交
2845

2846
    int32_t code = getDataBlocksInFiles(pTsdbReadHandle, &exists);
H
Haojun Liao 已提交
2847
    if (code != TSDB_CODE_SUCCESS) {
2848
      pTsdbReadHandle->checkFiles = false;
H
Haojun Liao 已提交
2849 2850
      return false;
    }
H
Haojun Liao 已提交
2851

H
Haojun Liao 已提交
2852
    if (exists) {
dengyihao's avatar
dengyihao 已提交
2853
      tsdbRetrieveDataBlock((tsdbReaderT*)pTsdbReadHandle, NULL);
2854 2855 2856
      if (pTsdbReadHandle->currentLoadExternalRows && pTsdbReadHandle->window.skey == pTsdbReadHandle->window.ekey) {
        SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, 0);
        assert(*(int64_t*)pColInfo->pData == pTsdbReadHandle->window.skey);
H
Haojun Liao 已提交
2857 2858
      }

dengyihao's avatar
dengyihao 已提交
2859
      pTsdbReadHandle->currentLoadExternalRows = false;  // clear the flag, since the exact matched row is found.
H
Haojun Liao 已提交
2860 2861
      return exists;
    }
H
Haojun Liao 已提交
2862

2863
    pTsdbReadHandle->checkFiles = false;
H
Haojun Liao 已提交
2864
  }
H
Haojun Liao 已提交
2865

2866 2867
  if (hasMoreDataInCache(pTsdbReadHandle)) {
    pTsdbReadHandle->currentLoadExternalRows = false;
H
Haojun Liao 已提交
2868 2869
    return true;
  }
H
Haojun Liao 已提交
2870

H
Haojun Liao 已提交
2871
  // current result is empty
dengyihao's avatar
dengyihao 已提交
2872 2873 2874
  if (pTsdbReadHandle->currentLoadExternalRows && pTsdbReadHandle->window.skey == pTsdbReadHandle->window.ekey &&
      pTsdbReadHandle->cur.rows == 0) {
    //    STsdbMemTable* pMemRef = pTsdbReadHandle->pMemTable;
H
Haojun Liao 已提交
2875

dengyihao's avatar
dengyihao 已提交
2876 2877
    //    doGetExternalRow(pTsdbReadHandle, TSDB_PREV_ROW, pMemRef);
    //    doGetExternalRow(pTsdbReadHandle, TSDB_NEXT_ROW, pMemRef);
H
Haojun Liao 已提交
2878

2879
    bool result = tsdbGetExternalRow(pTsdbReadHandle);
H
Haojun Liao 已提交
2880

dengyihao's avatar
dengyihao 已提交
2881 2882
    //    pTsdbReadHandle->prev = doFreeColumnInfoData(pTsdbReadHandle->prev);
    //    pTsdbReadHandle->next = doFreeColumnInfoData(pTsdbReadHandle->next);
2883
    pTsdbReadHandle->currentLoadExternalRows = false;
H
Haojun Liao 已提交
2884 2885

    return result;
2886
  }
H
Haojun Liao 已提交
2887

H
Haojun Liao 已提交
2888 2889
  return false;
}
2890

2891
static bool loadCachedLastRow(STsdbReadHandle* pTsdbReadHandle) {
H
Haojun Liao 已提交
2892
  // the last row is cached in buffer, return it directly.
2893
  // here note that the pTsdbReadHandle->window must be the TS_INITIALIZER
dengyihao's avatar
dengyihao 已提交
2894
  int32_t numOfCols = (int32_t)(QH_GET_NUM_OF_COLS(pTsdbReadHandle));
2895
  size_t  numOfTables = taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
H
Haojun Liao 已提交
2896 2897
  assert(numOfTables > 0 && numOfCols > 0);

2898
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
2899

dengyihao's avatar
dengyihao 已提交
2900 2901 2902
  STSRow* pRow = NULL;
  TSKEY   key = TSKEY_INITIAL_VAL;
  int32_t step = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 1 : -1;
C
Cary Xu 已提交
2903 2904
  TSKEY   lastRowKey = TSKEY_INITIAL_VAL;
  int32_t curRow = 0;
2905 2906 2907

  if (++pTsdbReadHandle->activeIndex < numOfTables) {
    STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, pTsdbReadHandle->activeIndex);
dengyihao's avatar
dengyihao 已提交
2908 2909 2910 2911
    //    int32_t ret = tsdbGetCachedLastRow(pCheckInfo->pTableObj, &pRow, &key);
    //    if (ret != TSDB_CODE_SUCCESS) {
    //      return false;
    //    }
C
Cary Xu 已提交
2912 2913
    mergeTwoRowFromMem(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, &curRow, pRow, NULL, numOfCols,
                       pCheckInfo->tableId, NULL, NULL, true, &lastRowKey);
wafwerar's avatar
wafwerar 已提交
2914
    taosMemoryFreeClear(pRow);
H
Haojun Liao 已提交
2915

H
Haojun Liao 已提交
2916 2917 2918
    // update the last key value
    pCheckInfo->lastKey = key + step;

dengyihao's avatar
dengyihao 已提交
2919 2920
    cur->rows = 1;  // only one row
    cur->lastKey = key + step;
H
Haojun Liao 已提交
2921 2922 2923 2924 2925
    cur->mixBlock = true;
    cur->win.skey = key;
    cur->win.ekey = key;

    return true;
2926
  }
H
Haojun Liao 已提交
2927

H
Haojun Liao 已提交
2928 2929 2930
  return false;
}

dengyihao's avatar
dengyihao 已提交
2931
// static bool loadCachedLast(STsdbReadHandle* pTsdbReadHandle) {
2932 2933 2934 2935 2936 2937 2938 2939 2940 2941 2942 2943 2944 2945 2946 2947 2948 2949 2950
//  // the last row is cached in buffer, return it directly.
//  // here note that the pTsdbReadHandle->window must be the TS_INITIALIZER
//  int32_t tgNumOfCols = (int32_t)QH_GET_NUM_OF_COLS(pTsdbReadHandle);
//  size_t  numOfTables = taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
//  int32_t numOfRows = 0;
//  assert(numOfTables > 0 && tgNumOfCols > 0);
//  SQueryFilePos* cur = &pTsdbReadHandle->cur;
//  TSKEY priKey = TSKEY_INITIAL_VAL;
//  int32_t priIdx = -1;
//  SColumnInfoData* pColInfo = NULL;
//
//  while (++pTsdbReadHandle->activeIndex < numOfTables) {
//    STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, pTsdbReadHandle->activeIndex);
//    STable* pTable = pCheckInfo->pTableObj;
//    char* pData = NULL;
//
//    int32_t numOfCols = pTable->maxColNum;
//
//    if (pTable->lastCols == NULL || pTable->maxColNum <= 0) {
dengyihao's avatar
dengyihao 已提交
2951 2952
//      tsdbWarn("no last cached for table %s, uid:%" PRIu64 ",tid:%d", pTable->name->data, pTable->uid,
//      pTable->tableId); continue;
2953 2954 2955 2956 2957 2958 2959 2960 2961 2962 2963 2964 2965 2966 2967 2968 2969 2970 2971 2972 2973 2974 2975 2976 2977 2978 2979 2980 2981 2982 2983 2984 2985 2986 2987 2988 2989 2990 2991 2992 2993 2994 2995 2996 2997 2998 2999 3000 3001 3002 3003 3004 3005 3006 3007 3008 3009 3010 3011 3012 3013 3014 3015 3016 3017 3018 3019 3020 3021 3022 3023 3024 3025 3026 3027 3028 3029 3030 3031 3032 3033 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
//    }
//
//    int32_t i = 0, j = 0;
//    while(i < tgNumOfCols && j < numOfCols) {
//      pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
//      if (pTable->lastCols[j].colId < pColInfo->info.colId) {
//        j++;
//        continue;
//      } else if (pTable->lastCols[j].colId > pColInfo->info.colId) {
//        i++;
//        continue;
//      }
//
//      pData = (char*)pColInfo->pData + numOfRows * pColInfo->info.bytes;
//
//      if (pTable->lastCols[j].bytes > 0) {
//        void* value = pTable->lastCols[j].pData;
//        switch (pColInfo->info.type) {
//          case TSDB_DATA_TYPE_BINARY:
//          case TSDB_DATA_TYPE_NCHAR:
//            memcpy(pData, value, varDataTLen(value));
//            break;
//          case TSDB_DATA_TYPE_NULL:
//          case TSDB_DATA_TYPE_BOOL:
//          case TSDB_DATA_TYPE_TINYINT:
//          case TSDB_DATA_TYPE_UTINYINT:
//            *(uint8_t *)pData = *(uint8_t *)value;
//            break;
//          case TSDB_DATA_TYPE_SMALLINT:
//          case TSDB_DATA_TYPE_USMALLINT:
//            *(uint16_t *)pData = *(uint16_t *)value;
//            break;
//          case TSDB_DATA_TYPE_INT:
//          case TSDB_DATA_TYPE_UINT:
//            *(uint32_t *)pData = *(uint32_t *)value;
//            break;
//          case TSDB_DATA_TYPE_BIGINT:
//          case TSDB_DATA_TYPE_UBIGINT:
//            *(uint64_t *)pData = *(uint64_t *)value;
//            break;
//          case TSDB_DATA_TYPE_FLOAT:
//            SET_FLOAT_PTR(pData, value);
//            break;
//          case TSDB_DATA_TYPE_DOUBLE:
//            SET_DOUBLE_PTR(pData, value);
//            break;
//          case TSDB_DATA_TYPE_TIMESTAMP:
//            if (pColInfo->info.colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
//              priKey = tdGetKey(*(TKEY *)value);
//              priIdx = i;
//
//              i++;
//              j++;
//              continue;
//            } else {
//              *(TSKEY *)pData = *(TSKEY *)value;
//            }
//            break;
//          default:
//            memcpy(pData, value, pColInfo->info.bytes);
//        }
//
//        for (int32_t n = 0; n < tgNumOfCols; ++n) {
//          if (n == i) {
//            continue;
//          }
//
//          pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, n);
//          pData = (char*)pColInfo->pData + numOfRows * pColInfo->info.bytes;;
//
//          if (pColInfo->info.colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
////            *(TSKEY *)pData = pTable->lastCols[j].ts;
//            continue;
//          }
//
//          if (pColInfo->info.type == TSDB_DATA_TYPE_BINARY || pColInfo->info.type == TSDB_DATA_TYPE_NCHAR) {
//            setVardataNull(pData, pColInfo->info.type);
//          } else {
//            setNull(pData, pColInfo->info.type, pColInfo->info.bytes);
//          }
//        }
//
//        numOfRows++;
//        assert(numOfRows < pTsdbReadHandle->outputCapacity);
//      }
//
//      i++;
//      j++;
//    }
//
//    // leave the real ts column as the last row, because last function only (not stable) use the last row as res
//    if (priKey != TSKEY_INITIAL_VAL) {
//      pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, priIdx);
//      pData = (char*)pColInfo->pData + numOfRows * pColInfo->info.bytes;
//
//      *(TSKEY *)pData = priKey;
//
//      for (int32_t n = 0; n < tgNumOfCols; ++n) {
//        if (n == priIdx) {
//          continue;
//        }
//
//        pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, n);
//        pData = (char*)pColInfo->pData + numOfRows * pColInfo->info.bytes;;
//
//        assert (pColInfo->info.colId != PRIMARYKEY_TIMESTAMP_COL_ID);
//
//        if (pColInfo->info.type == TSDB_DATA_TYPE_BINARY || pColInfo->info.type == TSDB_DATA_TYPE_NCHAR) {
//          setVardataNull(pData, pColInfo->info.type);
//        } else {
//          setNull(pData, pColInfo->info.type, pColInfo->info.bytes);
//        }
//      }
//
//      numOfRows++;
//    }
//
//    if (numOfRows > 0) {
//      cur->rows     = numOfRows;
//      cur->mixBlock = true;
//
//      return true;
//    }
//  }
//
//  return false;
//}

static bool loadDataBlockFromTableSeq(STsdbReadHandle* pTsdbReadHandle) {
  size_t numOfTables = taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
H
Haojun Liao 已提交
3083 3084 3085
  assert(numOfTables > 0);

  int64_t stime = taosGetTimestampUs();
H
Haojun Liao 已提交
3086

dengyihao's avatar
dengyihao 已提交
3087
  while (pTsdbReadHandle->activeIndex < numOfTables) {
3088
    if (loadBlockOfActiveTable(pTsdbReadHandle)) {
H
Haojun Liao 已提交
3089 3090 3091
      return true;
    }

3092
    STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, pTsdbReadHandle->activeIndex);
H
Haojun Liao 已提交
3093 3094
    pCheckInfo->numOfBlocks = 0;

3095 3096
    pTsdbReadHandle->activeIndex += 1;
    pTsdbReadHandle->locateStart = false;
dengyihao's avatar
dengyihao 已提交
3097 3098
    pTsdbReadHandle->checkFiles = true;
    pTsdbReadHandle->cur.rows = 0;
3099
    pTsdbReadHandle->currentLoadExternalRows = pTsdbReadHandle->loadExternalRow;
H
Haojun Liao 已提交
3100 3101 3102 3103

    terrno = TSDB_CODE_SUCCESS;

    int64_t elapsedTime = taosGetTimestampUs() - stime;
3104
    pTsdbReadHandle->cost.checkForNextTime += elapsedTime;
H
Haojun Liao 已提交
3105 3106 3107
  }

  return false;
3108 3109
}

H
Haojun Liao 已提交
3110
// handle data in cache situation
H
Haojun Liao 已提交
3111
bool tsdbNextDataBlock(tsdbReaderT pHandle) {
dengyihao's avatar
dengyihao 已提交
3112
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle;
Y
yihaoDeng 已提交
3113

3114 3115
  size_t numOfCols = taosArrayGetSize(pTsdbReadHandle->pColumns);
  for (int32_t i = 0; i < numOfCols; ++i) {
3116 3117 3118 3119
    SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
    colInfoDataCleanup(pColInfo, pTsdbReadHandle->outputCapacity);
  }

3120
  if (emptyQueryTimewindow(pTsdbReadHandle)) {
dengyihao's avatar
dengyihao 已提交
3121 3122
    tsdbDebug("%p query window not overlaps with the data set, no result returned, %s", pTsdbReadHandle,
              pTsdbReadHandle->idStr);
3123 3124 3125
    return false;
  }

Y
yihaoDeng 已提交
3126 3127 3128
  int64_t stime = taosGetTimestampUs();
  int64_t elapsedTime = stime;

3129
  // TODO refactor: remove "type"
3130 3131
  if (pTsdbReadHandle->type == TSDB_QUERY_TYPE_LAST) {
    if (pTsdbReadHandle->cachelastrow == TSDB_CACHED_TYPE_LASTROW) {
dengyihao's avatar
dengyihao 已提交
3132
      //      return loadCachedLastRow(pTsdbReadHandle);
3133
    } else if (pTsdbReadHandle->cachelastrow == TSDB_CACHED_TYPE_LAST) {
dengyihao's avatar
dengyihao 已提交
3134
      //      return loadCachedLast(pTsdbReadHandle);
D
init  
dapan1121 已提交
3135
    }
H
Haojun Liao 已提交
3136
  }
Y
yihaoDeng 已提交
3137

3138 3139
  if (pTsdbReadHandle->loadType == BLOCK_LOAD_TABLE_SEQ_ORDER) {
    return loadDataBlockFromTableSeq(pTsdbReadHandle);
dengyihao's avatar
dengyihao 已提交
3140
  } else {  // loadType == RR and Offset Order
3141
    if (pTsdbReadHandle->checkFiles) {
H
Haojun Liao 已提交
3142 3143 3144
      // check if the query range overlaps with the file data block
      bool exists = true;

3145
      int32_t code = getDataBlocksInFiles(pTsdbReadHandle, &exists);
H
Haojun Liao 已提交
3146
      if (code != TSDB_CODE_SUCCESS) {
3147 3148
        pTsdbReadHandle->activeIndex = 0;
        pTsdbReadHandle->checkFiles = false;
H
Haojun Liao 已提交
3149 3150 3151 3152 3153

        return false;
      }

      if (exists) {
3154
        pTsdbReadHandle->cost.checkForNextTime += (taosGetTimestampUs() - stime);
H
Haojun Liao 已提交
3155 3156
        return exists;
      }
Y
yihaoDeng 已提交
3157

3158 3159
      pTsdbReadHandle->activeIndex = 0;
      pTsdbReadHandle->checkFiles = false;
Y
yihaoDeng 已提交
3160 3161
    }

H
Haojun Liao 已提交
3162
    // TODO: opt by consider the scan order
3163
    bool ret = doHasDataInBuffer(pTsdbReadHandle);
H
Haojun Liao 已提交
3164
    terrno = TSDB_CODE_SUCCESS;
Y
yihaoDeng 已提交
3165

H
Haojun Liao 已提交
3166
    elapsedTime = taosGetTimestampUs() - stime;
3167
    pTsdbReadHandle->cost.checkForNextTime += elapsedTime;
H
Haojun Liao 已提交
3168
    return ret;
Y
yihaoDeng 已提交
3169 3170
  }
}
3171

dengyihao's avatar
dengyihao 已提交
3172
// static int32_t doGetExternalRow(STsdbReadHandle* pTsdbReadHandle, int16_t type, STsdbMemTable* pMemRef) {
3173 3174 3175 3176 3177 3178 3179 3180 3181 3182 3183 3184 3185 3186 3187 3188 3189 3190 3191 3192 3193 3194 3195 3196 3197 3198 3199 3200 3201 3202 3203 3204 3205 3206
//  STsdbReadHandle* pSecQueryHandle = NULL;
//
//  if (type == TSDB_PREV_ROW && pTsdbReadHandle->prev) {
//    return TSDB_CODE_SUCCESS;
//  }
//
//  if (type == TSDB_NEXT_ROW && pTsdbReadHandle->next) {
//    return TSDB_CODE_SUCCESS;
//  }
//
//  // prepare the structure
//  int32_t numOfCols = (int32_t) QH_GET_NUM_OF_COLS(pTsdbReadHandle);
//
//  if (type == TSDB_PREV_ROW) {
//    pTsdbReadHandle->prev = taosArrayInit(numOfCols, sizeof(SColumnInfoData));
//    if (pTsdbReadHandle->prev == NULL) {
//      terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
//      goto out_of_memory;
//    }
//  } else {
//    pTsdbReadHandle->next = taosArrayInit(numOfCols, sizeof(SColumnInfoData));
//    if (pTsdbReadHandle->next == NULL) {
//      terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
//      goto out_of_memory;
//    }
//  }
//
//  SArray* row = (type == TSDB_PREV_ROW)? pTsdbReadHandle->prev : pTsdbReadHandle->next;
//
//  for (int32_t i = 0; i < numOfCols; ++i) {
//    SColumnInfoData* pCol = taosArrayGet(pTsdbReadHandle->pColumns, i);
//
//    SColumnInfoData colInfo = {{0}, 0};
//    colInfo.info = pCol->info;
wafwerar's avatar
wafwerar 已提交
3207
//    colInfo.pData = taosMemoryCalloc(1, pCol->info.bytes);
3208 3209 3210 3211 3212 3213 3214 3215 3216
//    if (colInfo.pData == NULL) {
//      terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
//      goto out_of_memory;
//    }
//
//    taosArrayPush(row, &colInfo);
//  }
//
//  // load the previous row
3217
//  SQueryTableDataCond cond = {.numOfCols = numOfCols, .loadExternalRows = false, .type = BLOCK_LOAD_OFFSET_SEQ_ORDER};
3218 3219 3220 3221 3222 3223 3224 3225
//  if (type == TSDB_PREV_ROW) {
//    cond.order = TSDB_ORDER_DESC;
//    cond.twindow = (STimeWindow){pTsdbReadHandle->window.skey, INT64_MIN};
//  } else {
//    cond.order = TSDB_ORDER_ASC;
//    cond.twindow = (STimeWindow){pTsdbReadHandle->window.skey, INT64_MAX};
//  }
//
wafwerar's avatar
wafwerar 已提交
3226
//  cond.colList = taosMemoryCalloc(cond.numOfCols, sizeof(SColumnInfo));
3227 3228 3229 3230 3231 3232 3233 3234 3235 3236
//  if (cond.colList == NULL) {
//    terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
//    goto out_of_memory;
//  }
//
//  for (int32_t i = 0; i < cond.numOfCols; ++i) {
//    SColumnInfoData* pColInfoData = taosArrayGet(pTsdbReadHandle->pColumns, i);
//    memcpy(&cond.colList[i], &pColInfoData->info, sizeof(SColumnInfo));
//  }
//
H
Haojun Liao 已提交
3237
//  pSecQueryHandle = tsdbQueryTablesImpl(pTsdbReadHandle->pTsdb, &cond, pTsdbReadHandle->idStr, pMemRef);
wafwerar's avatar
wafwerar 已提交
3238
//  taosMemoryFreeClear(cond.colList);
3239 3240 3241 3242 3243 3244 3245 3246 3247 3248 3249 3250 3251 3252 3253 3254 3255 3256 3257 3258 3259 3260 3261 3262 3263 3264 3265 3266 3267 3268 3269 3270 3271 3272 3273 3274 3275 3276
//
//  // current table, only one table
//  STableCheckInfo* pCurrent = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, pTsdbReadHandle->activeIndex);
//
//  SArray* psTable = NULL;
//  pSecQueryHandle->pTableCheckInfo = createCheckInfoFromCheckInfo(pCurrent, pSecQueryHandle->window.skey, &psTable);
//  if (pSecQueryHandle->pTableCheckInfo == NULL) {
//    taosArrayDestroy(psTable);
//    terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
//    goto out_of_memory;
//  }
//
//
//  tsdbMayTakeMemSnapshot(pSecQueryHandle, psTable);
//  if (!tsdbNextDataBlock((void*)pSecQueryHandle)) {
//    // no result in current query, free the corresponding result rows structure
//    if (type == TSDB_PREV_ROW) {
//      pTsdbReadHandle->prev = doFreeColumnInfoData(pTsdbReadHandle->prev);
//    } else {
//      pTsdbReadHandle->next = doFreeColumnInfoData(pTsdbReadHandle->next);
//    }
//
//    goto out_of_memory;
//  }
//
//  SDataBlockInfo blockInfo = {{0}, 0};
//  tsdbRetrieveDataBlockInfo((void*)pSecQueryHandle, &blockInfo);
//  tsdbRetrieveDataBlock((void*)pSecQueryHandle, pSecQueryHandle->defaultLoadColumn);
//
//  row = (type == TSDB_PREV_ROW)? pTsdbReadHandle->prev:pTsdbReadHandle->next;
//  int32_t pos = (type == TSDB_PREV_ROW)?pSecQueryHandle->cur.rows - 1:0;
//
//  for (int32_t i = 0; i < numOfCols; ++i) {
//    SColumnInfoData* pCol = taosArrayGet(row, i);
//    SColumnInfoData* s = taosArrayGet(pSecQueryHandle->pColumns, i);
//    memcpy((char*)pCol->pData, (char*)s->pData + s->info.bytes * pos, pCol->info.bytes);
//  }
//
dengyihao's avatar
dengyihao 已提交
3277
// out_of_memory:
3278
//  tsdbCleanupReadHandle(pSecQueryHandle);
3279 3280 3281
//  return terrno;
//}

H
Haojun Liao 已提交
3282
bool tsdbGetExternalRow(tsdbReaderT pHandle) {
dengyihao's avatar
dengyihao 已提交
3283 3284
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle;
  SQueryFilePos*   cur = &pTsdbReadHandle->cur;
H
Haojun Liao 已提交
3285

H
Haojun Liao 已提交
3286 3287
  cur->fid = INT32_MIN;
  cur->mixBlock = true;
3288
  if (pTsdbReadHandle->prev == NULL || pTsdbReadHandle->next == NULL) {
H
Haojun Liao 已提交
3289 3290
    cur->rows = 0;
    return false;
H
Haojun Liao 已提交
3291 3292
  }

dengyihao's avatar
dengyihao 已提交
3293
  int32_t numOfCols = (int32_t)QH_GET_NUM_OF_COLS(pTsdbReadHandle);
H
Haojun Liao 已提交
3294
  for (int32_t i = 0; i < numOfCols; ++i) {
3295 3296
    SColumnInfoData* pColInfoData = taosArrayGet(pTsdbReadHandle->pColumns, i);
    SColumnInfoData* first = taosArrayGet(pTsdbReadHandle->prev, i);
H
Haojun Liao 已提交
3297 3298 3299

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

3300
    SColumnInfoData* sec = taosArrayGet(pTsdbReadHandle->next, i);
sangshuduo's avatar
sangshuduo 已提交
3301
    memcpy(((char*)pColInfoData->pData) + pColInfoData->info.bytes, sec->pData, pColInfoData->info.bytes);
H
Haojun Liao 已提交
3302 3303

    if (i == 0 && pColInfoData->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
H
Haojun Liao 已提交
3304
      cur->win.skey = *(TSKEY*)pColInfoData->pData;
sangshuduo's avatar
sangshuduo 已提交
3305
      cur->win.ekey = *(TSKEY*)(((char*)pColInfoData->pData) + TSDB_KEYSIZE);
H
Haojun Liao 已提交
3306 3307 3308
    }
  }

H
Haojun Liao 已提交
3309 3310
  cur->rows = 2;
  return true;
3311 3312
}

3313
/*
3314
 * if lastRow == NULL, return TSDB_CODE_TDB_NO_CACHE_LAST_ROW
3315
 * else set pRes and return TSDB_CODE_SUCCESS and save lastKey
3316
 */
H
Haojun Liao 已提交
3317
// int32_t tsdbGetCachedLastRow(STable* pTable, STSRow** pRes, TSKEY* lastKey) {
3318 3319 3320 3321 3322 3323 3324 3325 3326 3327 3328 3329 3330 3331 3332 3333
//  int32_t code = TSDB_CODE_SUCCESS;
//
//  TSDB_RLOCK_TABLE(pTable);
//
//  if (!pTable->lastRow) {
//    code = TSDB_CODE_TDB_NO_CACHE_LAST_ROW;
//    goto out;
//  }
//
//  if (pRes) {
//    *pRes = tdMemRowDup(pTable->lastRow);
//    if (*pRes == NULL) {
//      code = TSDB_CODE_TDB_OUT_OF_MEMORY;
//    }
//  }
//
H
Haojun Liao 已提交
3334
// out:
3335 3336 3337 3338
//  TSDB_RUNLOCK_TABLE(pTable);
//  return code;
//}

3339
bool isTsdbCacheLastRow(tsdbReaderT* pReader) {
dengyihao's avatar
dengyihao 已提交
3340
  return ((STsdbReadHandle*)pReader)->cachelastrow > TSDB_CACHED_TYPE_NONE;
D
fix bug  
dapan1121 已提交
3341 3342
}

dengyihao's avatar
dengyihao 已提交
3343
int32_t checkForCachedLastRow(STsdbReadHandle* pTsdbReadHandle, STableGroupInfo* groupList) {
3344 3345
  assert(pTsdbReadHandle != NULL && groupList != NULL);

dengyihao's avatar
dengyihao 已提交
3346 3347 3348 3349 3350 3351 3352 3353 3354 3355 3356 3357 3358 3359 3360 3361 3362 3363 3364 3365 3366 3367 3368 3369
  //  TSKEY    key = TSKEY_INITIAL_VAL;
  //
  //  SArray* group = taosArrayGetP(groupList->pGroupList, 0);
  //  assert(group != NULL);
  //
  //  STableKeyInfo* pInfo = (STableKeyInfo*)taosArrayGet(group, 0);
  //
  //  int32_t code = 0;
  //
  //  if (((STable*)pInfo->pTable)->lastRow) {
  //    code = tsdbGetCachedLastRow(pInfo->pTable, NULL, &key);
  //    if (code != TSDB_CODE_SUCCESS) {
  //      pTsdbReadHandle->cachelastrow = TSDB_CACHED_TYPE_NONE;
  //    } else {
  //      pTsdbReadHandle->cachelastrow = TSDB_CACHED_TYPE_LASTROW;
  //    }
  //  }
  //
  //  // update the tsdb query time range
  //  if (pTsdbReadHandle->cachelastrow != TSDB_CACHED_TYPE_NONE) {
  //    pTsdbReadHandle->window      = TSWINDOW_INITIALIZER;
  //    pTsdbReadHandle->checkFiles  = false;
  //    pTsdbReadHandle->activeIndex = -1;  // start from -1
  //  }
H
Haojun Liao 已提交
3370

3371
  return TSDB_CODE_SUCCESS;
3372 3373
}

3374 3375
int32_t checkForCachedLast(STsdbReadHandle* pTsdbReadHandle) {
  assert(pTsdbReadHandle != NULL);
D
update  
dapan1121 已提交
3376 3377

  int32_t code = 0;
dengyihao's avatar
dengyihao 已提交
3378 3379 3380
  //  if (pTsdbReadHandle->pTsdb && atomic_load_8(&pTsdbReadHandle->pTsdb->hasCachedLastColumn)){
  //    pTsdbReadHandle->cachelastrow = TSDB_CACHED_TYPE_LAST;
  //  }
D
update  
dapan1121 已提交
3381 3382

  // update the tsdb query time range
3383
  if (pTsdbReadHandle->cachelastrow) {
dengyihao's avatar
dengyihao 已提交
3384
    pTsdbReadHandle->checkFiles = false;
3385
    pTsdbReadHandle->activeIndex = -1;  // start from -1
D
update  
dapan1121 已提交
3386 3387 3388 3389 3390
  }

  return code;
}

dengyihao's avatar
dengyihao 已提交
3391
STimeWindow updateLastrowForEachGroup(STableGroupInfo* groupList) {
H
Haojun Liao 已提交
3392
  STimeWindow window = {INT64_MAX, INT64_MIN};
H
Haojun Liao 已提交
3393

H
Haojun Liao 已提交
3394
  int32_t totalNumOfTable = 0;
3395
  SArray* emptyGroup = taosArrayInit(16, sizeof(int32_t));
H
Haojun Liao 已提交
3396

H
Haojun Liao 已提交
3397 3398
  // NOTE: starts from the buffer in case of descending timestamp order check data blocks
  size_t numOfGroups = taosArrayGetSize(groupList->pGroupList);
dengyihao's avatar
dengyihao 已提交
3399
  for (int32_t j = 0; j < numOfGroups; ++j) {
H
Haojun Liao 已提交
3400 3401
    SArray* pGroup = taosArrayGetP(groupList->pGroupList, j);
    TSKEY   key = TSKEY_INITIAL_VAL;
H
Haojun Liao 已提交
3402

H
Haojun Liao 已提交
3403
    STableKeyInfo keyInfo = {0};
H
Haojun Liao 已提交
3404

H
Haojun Liao 已提交
3405
    size_t numOfTables = taosArrayGetSize(pGroup);
dengyihao's avatar
dengyihao 已提交
3406 3407
    for (int32_t i = 0; i < numOfTables; ++i) {
      STableKeyInfo* pInfo = (STableKeyInfo*)taosArrayGet(pGroup, i);
H
Haojun Liao 已提交
3408

H
Haojun Liao 已提交
3409
      // if the lastKey equals to INT64_MIN, there is no data in this table
dengyihao's avatar
dengyihao 已提交
3410
      TSKEY lastKey = 0;  //((STable*)(pInfo->pTable))->lastKey;
H
Haojun Liao 已提交
3411 3412
      if (key < lastKey) {
        key = lastKey;
H
Haojun Liao 已提交
3413

dengyihao's avatar
dengyihao 已提交
3414
        //        keyInfo.pTable  = pInfo->pTable;
H
Haojun Liao 已提交
3415
        keyInfo.lastKey = key;
dengyihao's avatar
dengyihao 已提交
3416
        pInfo->lastKey = key;
H
Haojun Liao 已提交
3417

H
Haojun Liao 已提交
3418 3419 3420
        if (key < window.skey) {
          window.skey = key;
        }
3421

H
Haojun Liao 已提交
3422 3423 3424 3425
        if (key > window.ekey) {
          window.ekey = key;
        }
      }
3426
    }
H
Haojun Liao 已提交
3427

H
Haojun Liao 已提交
3428
    // more than one table in each group, only one table left for each group
dengyihao's avatar
dengyihao 已提交
3429 3430 3431 3432 3433 3434 3435 3436 3437 3438 3439 3440
    //    if (keyInfo.pTable != NULL) {
    //      totalNumOfTable++;
    //      if (taosArrayGetSize(pGroup) == 1) {
    //        // do nothing
    //      } else {
    //        taosArrayClear(pGroup);
    //        taosArrayPush(pGroup, &keyInfo);
    //      }
    //    } else {  // mark all the empty groups, and remove it later
    //      taosArrayDestroy(pGroup);
    //      taosArrayPush(emptyGroup, &j);
    //    }
3441
  }
H
Haojun Liao 已提交
3442

H
Haojun Liao 已提交
3443 3444 3445
  // window does not being updated, so set the original
  if (window.skey == INT64_MAX && window.ekey == INT64_MIN) {
    window = TSWINDOW_INITIALIZER;
H
Haojun Liao 已提交
3446
    assert(totalNumOfTable == 0 && taosArrayGetSize(groupList->pGroupList) == numOfGroups);
H
Haojun Liao 已提交
3447 3448
  }

dengyihao's avatar
dengyihao 已提交
3449
  taosArrayRemoveBatch(groupList->pGroupList, TARRAY_GET_START(emptyGroup), (int32_t)taosArrayGetSize(emptyGroup));
3450 3451
  taosArrayDestroy(emptyGroup);

H
Haojun Liao 已提交
3452
  groupList->numOfTables = totalNumOfTable;
H
Haojun Liao 已提交
3453
  return window;
H
hjxilinx 已提交
3454 3455
}

H
Haojun Liao 已提交
3456
void tsdbRetrieveDataBlockInfo(tsdbReaderT* pTsdbReadHandle, SDataBlockInfo* pDataBlockInfo) {
3457
  STsdbReadHandle* pHandle = (STsdbReadHandle*)pTsdbReadHandle;
dengyihao's avatar
dengyihao 已提交
3458
  SQueryFilePos*   cur = &pHandle->cur;
3459 3460

  uint64_t uid = 0;
H
Haojun Liao 已提交
3461

3462
  // there are data in file
D
dapan1121 已提交
3463
  if (pHandle->cur.fid != INT32_MIN) {
3464
    STableBlockInfo* pBlockInfo = &pHandle->pDataBlockInfo[cur->slot];
3465
    uid = pBlockInfo->pTableCheckInfo->tableId;
H
[td-32]  
hjxilinx 已提交
3466
  } else {
3467
    STableCheckInfo* pCheckInfo = taosArrayGet(pHandle->pTableCheckInfo, pHandle->activeIndex);
3468
    uid = pCheckInfo->tableId;
3469
  }
3470

dengyihao's avatar
dengyihao 已提交
3471 3472
  tsdbDebug("data block generated, uid:%" PRIu64 " numOfRows:%d, tsrange:%" PRId64 " - %" PRId64 " %s", uid, cur->rows,
            cur->win.skey, cur->win.ekey, pHandle->idStr);
3473

3474
  pDataBlockInfo->uid = uid;
3475 3476 3477 3478 3479 3480

#if 0
  // for multi-group data query processing test purpose
  pDataBlockInfo->groupId = uid;
#endif

dengyihao's avatar
dengyihao 已提交
3481
  pDataBlockInfo->rows = cur->rows;
H
Haojun Liao 已提交
3482
  pDataBlockInfo->window = cur->win;
C
Cary Xu 已提交
3483
  //  ASSERT(pDataBlockInfo->numOfCols >= (int32_t)(QH_GET_NUM_OF_COLS(pHandle));
3484
}
H
hjxilinx 已提交
3485

H
Haojun Liao 已提交
3486 3487 3488
/*
 * return null for mixed data block, if not a complete file data block, the statistics value will always return NULL
 */
3489
int32_t tsdbRetrieveDataBlockStatisInfo(tsdbReaderT* pTsdbReadHandle, SColumnDataAgg*** pBlockStatis, bool* allHave) {
dengyihao's avatar
dengyihao 已提交
3490
  STsdbReadHandle* pHandle = (STsdbReadHandle*)pTsdbReadHandle;
3491
  *allHave = false;
H
Haojun Liao 已提交
3492

H
Haojun Liao 已提交
3493 3494
  SQueryFilePos* c = &pHandle->cur;
  if (c->mixBlock) {
H
Haojun Liao 已提交
3495 3496 3497
    *pBlockStatis = NULL;
    return TSDB_CODE_SUCCESS;
  }
H
Haojun Liao 已提交
3498

H
Haojun Liao 已提交
3499 3500 3501 3502
  STableBlockInfo* pBlockInfo = &pHandle->pDataBlockInfo[c->slot];
  assert((c->slot >= 0 && c->slot < pHandle->numOfBlocks) || ((c->slot == pHandle->numOfBlocks) && (c->slot == 0)));

  // file block with sub-blocks has no statistics data
H
Haojun Liao 已提交
3503 3504 3505 3506
  if (pBlockInfo->compBlock->numOfSubBlocks > 1) {
    *pBlockStatis = NULL;
    return TSDB_CODE_SUCCESS;
  }
H
Haojun Liao 已提交
3507 3508

  int64_t stime = taosGetTimestampUs();
3509 3510
  int     statisStatus = tsdbLoadBlockStatis(&pHandle->rhelper, pBlockInfo->compBlock);
  if (statisStatus < TSDB_STATIS_OK) {
H
Hongze Cheng 已提交
3511
    return terrno;
3512 3513 3514
  } else if (statisStatus > TSDB_STATIS_OK) {
    *pBlockStatis = NULL;
    return TSDB_CODE_SUCCESS;
H
Hongze Cheng 已提交
3515
  }
H
Haojun Liao 已提交
3516

C
Cary Xu 已提交
3517 3518 3519
  tsdbDebug("vgId:%d succeed to load block statis part for uid %" PRIu64, REPO_ID(pHandle->pTsdb),
            TSDB_READ_TABLE_UID(&pHandle->rhelper));

3520
  int16_t* colIds = pHandle->suppInfo.defaultLoadColumn->pData;
H
Haojun Liao 已提交
3521

H
Haojun Liao 已提交
3522
  size_t numOfCols = QH_GET_NUM_OF_COLS(pHandle);
3523 3524 3525
  memset(pHandle->suppInfo.plist, 0, numOfCols * POINTER_BYTES);
  memset(pHandle->suppInfo.pstatis, 0, numOfCols * sizeof(SColumnDataAgg));

dengyihao's avatar
dengyihao 已提交
3526
  for (int32_t i = 0; i < numOfCols; ++i) {
3527
    pHandle->suppInfo.pstatis[i].colId = colIds[i];
3528
  }
H
Haojun Liao 已提交
3529

3530 3531
  *allHave = true;
  tsdbGetBlockStatis(&pHandle->rhelper, pHandle->suppInfo.pstatis, (int)numOfCols, pBlockInfo->compBlock);
H
Haojun Liao 已提交
3532 3533

  // always load the first primary timestamp column data
3534
  SColumnDataAgg* pPrimaryColStatis = &pHandle->suppInfo.pstatis[0];
3535
  assert(pPrimaryColStatis->colId == PRIMARYKEY_TIMESTAMP_COL_ID);
H
Haojun Liao 已提交
3536 3537 3538 3539

  pPrimaryColStatis->numOfNull = 0;
  pPrimaryColStatis->min = pBlockInfo->compBlock->keyFirst;
  pPrimaryColStatis->max = pBlockInfo->compBlock->keyLast;
3540
  pHandle->suppInfo.plist[0] = &pHandle->suppInfo.pstatis[0];
H
Haojun Liao 已提交
3541

dengyihao's avatar
dengyihao 已提交
3542
  // update the number of NULL data rows
3543
  int32_t* slotIds = pHandle->suppInfo.slotIds;
dengyihao's avatar
dengyihao 已提交
3544
  for (int32_t i = 1; i < numOfCols; ++i) {
3545
    ASSERT(colIds[i] == pHandle->pSchema->columns[slotIds[i]].colId);
C
Cary Xu 已提交
3546
    if (IS_BSMA_ON(&(pHandle->pSchema->columns[slotIds[i]]))) {
3547 3548 3549 3550 3551 3552 3553
      if (pHandle->suppInfo.pstatis[i].numOfNull == -1) {  // set the column data are all NULL
        pHandle->suppInfo.pstatis[i].numOfNull = pBlockInfo->compBlock->numOfRows;
      } else {
        pHandle->suppInfo.plist[i] = &pHandle->suppInfo.pstatis[i];
      }
    } else {
      *allHave = false;
H
Haojun Liao 已提交
3554 3555
    }
  }
H
Haojun Liao 已提交
3556 3557 3558 3559

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

3560
  *pBlockStatis = pHandle->suppInfo.plist;
3561
  return TSDB_CODE_SUCCESS;
H
hjxilinx 已提交
3562 3563
}

H
Haojun Liao 已提交
3564
SArray* tsdbRetrieveDataBlock(tsdbReaderT* pTsdbReadHandle, SArray* pIdList) {
H
[td-32]  
hjxilinx 已提交
3565
  /**
H
hjxilinx 已提交
3566
   * In the following two cases, the data has been loaded to SColumnInfoData.
H
[td-32]  
hjxilinx 已提交
3567 3568
   * 1. data is from cache, 2. data block is not completed qualified to query time range
   */
3569
  STsdbReadHandle* pHandle = (STsdbReadHandle*)pTsdbReadHandle;
D
dapan1121 已提交
3570
  if (pHandle->cur.fid == INT32_MIN) {
H
[td-32]  
hjxilinx 已提交
3571 3572
    return pHandle->pColumns;
  } else {
H
Haojun Liao 已提交
3573 3574
    STableBlockInfo* pBlockInfo = &pHandle->pDataBlockInfo[pHandle->cur.slot];
    STableCheckInfo* pCheckInfo = pBlockInfo->pTableCheckInfo;
3575

3576
    if (pHandle->cur.mixBlock) {
H
[td-32]  
hjxilinx 已提交
3577 3578
      return pHandle->pColumns;
    } else {
H
Haojun Liao 已提交
3579
      SDataBlockInfo binfo = GET_FILE_DATA_BLOCK_INFO(pCheckInfo, pBlockInfo->compBlock);
3580
      assert(pHandle->realNumOfRows <= binfo.rows);
H
Haojun Liao 已提交
3581

H
hjxilinx 已提交
3582 3583
      // data block has been loaded, todo extract method
      SDataBlockLoadInfo* pBlockLoadInfo = &pHandle->dataBlockLoadInfo;
H
Haojun Liao 已提交
3584

H
Hongze Cheng 已提交
3585
      if (pBlockLoadInfo->slot == pHandle->cur.slot && pBlockLoadInfo->fileGroup->fid == pHandle->cur.fid &&
H
Haojun Liao 已提交
3586
          pBlockLoadInfo->uid == pCheckInfo->tableId) {
H
hjxilinx 已提交
3587
        return pHandle->pColumns;
H
Haojun Liao 已提交
3588
      } else {  // only load the file block
H
refact  
Hongze Cheng 已提交
3589
        SBlock* pBlock = pBlockInfo->compBlock;
H
Haojun Liao 已提交
3590
        if (doLoadFileDataBlock(pHandle, pBlock, pCheckInfo, pHandle->cur.slot) != TSDB_CODE_SUCCESS) {
3591 3592
          return NULL;
        }
H
Haojun Liao 已提交
3593

H
Haojun Liao 已提交
3594
        int32_t numOfRows = doCopyRowsFromFileBlock(pHandle, pHandle->outputCapacity, 0, 0, pBlock->numOfRows - 1);
H
hjxilinx 已提交
3595 3596
        return pHandle->pColumns;
      }
H
[td-32]  
hjxilinx 已提交
3597 3598
    }
  }
H
hjxilinx 已提交
3599
}
3600
#if 0
3601
void filterPrepare(void* expr, void* param) {
3602
  tExprNode* pExpr = (tExprNode*)expr;
H
[td-32]  
hjxilinx 已提交
3603
  if (pExpr->_node.info != NULL) {
3604 3605
    return;
  }
3606

wafwerar's avatar
wafwerar 已提交
3607
  pExpr->_node.info = taosMemoryCalloc(1, sizeof(tQueryInfo));
H
Haojun Liao 已提交
3608

3609
  STSchema*   pTSSchema = (STSchema*) param;
H
hjxilinx 已提交
3610 3611 3612
  tQueryInfo* pInfo = pExpr->_node.info;
  tVariant*   pCond = pExpr->_node.pRight->pVal;
  SSchema*    pSchema = pExpr->_node.pLeft->pSchema;
3613

3614 3615
  pInfo->sch      = *pSchema;
  pInfo->optr     = pExpr->_node.optr;
Y
yihaoDeng 已提交
3616
  pInfo->compare  = getComparFunc(pInfo->sch.type, pInfo->optr);
H
Haojun Liao 已提交
3617
  pInfo->indexed  = pTSSchema->columns->colId == pInfo->sch.colId;
H
Haojun Liao 已提交
3618

H
hjxilinx 已提交
3619
  if (pInfo->optr == TSDB_RELATION_IN) {
Y
yihaoDeng 已提交
3620
     int dummy = -1;
3621
     SHashObj *pObj = NULL;
Y
yihaoDeng 已提交
3622 3623 3624 3625
     if (pInfo->sch.colId == TSDB_TBNAME_COLUMN_INDEX) {
        pObj = taosHashInit(256, taosGetDefaultHashFunction(pInfo->sch.type), true, false);
        SArray *arr = (SArray *)(pCond->arr);
        for (size_t i = 0; i < taosArrayGetSize(arr); i++) {
Y
yihaoDeng 已提交
3626
          char* p = taosArrayGetP(arr, i);
3627 3628
          strntolower_s(varDataVal(p), varDataVal(p), varDataLen(p));
          taosHashPut(pObj, varDataVal(p), varDataLen(p), &dummy, sizeof(dummy));
Y
yihaoDeng 已提交
3629 3630 3631 3632
        }
     } else {
       buildFilterSetFromBinary((void **)&pObj, pCond->pz, pCond->nLen);
     }
3633
     pInfo->q = (char *)pObj;
H
Haojun Liao 已提交
3634
  } else if (pCond != NULL) {
3635 3636 3637 3638
    uint32_t size = pCond->nLen * TSDB_NCHAR_SIZE;
    if (size < (uint32_t)pSchema->bytes) {
      size = pSchema->bytes;
    }
wafwerar's avatar
wafwerar 已提交
3639
    // to make sure tonchar does not cause invalid write, since the '\0' needs at least sizeof(TdUcs4) space.
wafwerar's avatar
wafwerar 已提交
3640
    pInfo->q = taosMemoryCalloc(1, size + TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE);
3641
    tVariantDump(pCond, pInfo->q, pSchema->type, true);
weixin_48148422's avatar
weixin_48148422 已提交
3642
  }
3643 3644
}

3645
#endif
3646

dengyihao's avatar
dengyihao 已提交
3647
static int32_t tableGroupComparFn(const void* p1, const void* p2, const void* param) {
3648
#if 0
3649
  STableGroupSupporter* pTableGroupSupp = (STableGroupSupporter*) param;
3650 3651
  STable* pTable1 = ((STableKeyInfo*) p1)->uid;
  STable* pTable2 = ((STableKeyInfo*) p2)->uid;
H
Haojun Liao 已提交
3652

3653 3654 3655
  for (int32_t i = 0; i < pTableGroupSupp->numOfCols; ++i) {
    SColIndex* pColIndex = &pTableGroupSupp->pCols[i];
    int32_t colIndex = pColIndex->colIndex;
H
Haojun Liao 已提交
3656

H
Haojun Liao 已提交
3657
    assert(colIndex >= TSDB_TBNAME_COLUMN_INDEX);
H
Haojun Liao 已提交
3658

3659 3660 3661 3662
    char *  f1 = NULL;
    char *  f2 = NULL;
    int32_t type = 0;
    int32_t bytes = 0;
H
Haojun Liao 已提交
3663

H
Haojun Liao 已提交
3664 3665 3666
    if (colIndex == TSDB_TBNAME_COLUMN_INDEX) {
      f1 = (char*) TABLE_NAME(pTable1);
      f2 = (char*) TABLE_NAME(pTable2);
3667
      type = TSDB_DATA_TYPE_BINARY;
3668
      bytes = tGetTbnameColumnSchema()->bytes;
3669
    } else {
Y
yihaoDeng 已提交
3670 3671 3672 3673 3674 3675 3676
      if (pTableGroupSupp->pTagSchema && colIndex < pTableGroupSupp->pTagSchema->numOfCols) {
        STColumn* pCol = schemaColAt(pTableGroupSupp->pTagSchema, colIndex);
        bytes = pCol->bytes;
        type = pCol->type;
        f1 = tdGetKVRowValOfCol(pTable1->tagVal, pCol->colId);
        f2 = tdGetKVRowValOfCol(pTable2->tagVal, pCol->colId);
      } 
3677
    }
H
Haojun Liao 已提交
3678 3679 3680 3681 3682 3683 3684 3685 3686 3687 3688 3689 3690 3691

    // this tags value may be NULL
    if (f1 == NULL && f2 == NULL) {
      continue;
    }

    if (f1 == NULL) {
      return -1;
    }

    if (f2 == NULL) {
      return 1;
    }

3692 3693 3694 3695 3696 3697 3698
    int32_t ret = doCompare(f1, f2, type, bytes);
    if (ret == 0) {
      continue;
    } else {
      return ret;
    }
  }
3699
#endif
3700 3701 3702
  return 0;
}

H
Haojun Liao 已提交
3703
static int tsdbCheckInfoCompar(const void* key1, const void* key2) {
3704
  if (((STableCheckInfo*)key1)->tableId < ((STableCheckInfo*)key2)->tableId) {
H
Haojun Liao 已提交
3705
    return -1;
3706
  } else if (((STableCheckInfo*)key1)->tableId > ((STableCheckInfo*)key2)->tableId) {
H
Haojun Liao 已提交
3707 3708 3709 3710 3711 3712 3713 3714 3715
    return 1;
  } else {
    ASSERT(false);
    return 0;
  }
}

void createTableGroupImpl(SArray* pGroups, SArray* pTableList, size_t numOfTables, TSKEY skey,
                          STableGroupSupporter* pSupp, __ext_compar_fn_t compareFn) {
3716
  STable* pTable = taosArrayGetP(pTableList, 0);
H
Haojun Liao 已提交
3717 3718
  SArray* g = taosArrayInit(16, sizeof(STableKeyInfo));

3719
  STableKeyInfo info = {.lastKey = skey};
H
Haojun Liao 已提交
3720
  taosArrayPush(g, &info);
3721

3722
  for (int32_t i = 1; i < numOfTables; ++i) {
3723 3724
    STable** prev = taosArrayGet(pTableList, i - 1);
    STable** p = taosArrayGet(pTableList, i);
H
Haojun Liao 已提交
3725

H
hjxilinx 已提交
3726
    int32_t ret = compareFn(prev, p, pSupp);
3727
    assert(ret == 0 || ret == -1);
H
Haojun Liao 已提交
3728

3729
    if (ret == 0) {
3730
      STableKeyInfo info1 = {.lastKey = skey};
H
Haojun Liao 已提交
3731
      taosArrayPush(g, &info1);
3732 3733
    } else {
      taosArrayPush(pGroups, &g);  // current group is ended, start a new group
H
Haojun Liao 已提交
3734 3735
      g = taosArrayInit(16, sizeof(STableKeyInfo));

3736
      STableKeyInfo info1 = {.lastKey = skey};
H
Haojun Liao 已提交
3737
      taosArrayPush(g, &info1);
3738 3739
    }
  }
H
Haojun Liao 已提交
3740

3741
  taosArrayPush(pGroups, &g);
3742 3743
}

dengyihao's avatar
dengyihao 已提交
3744 3745
SArray* createTableGroup(SArray* pTableList, SSchemaWrapper* pTagSchema, SColIndex* pCols, int32_t numOfOrderCols,
                         TSKEY skey) {
3746
  assert(pTableList != NULL);
3747
  SArray* pTableGroup = taosArrayInit(1, POINTER_BYTES);
H
Haojun Liao 已提交
3748

3749 3750
  size_t size = taosArrayGetSize(pTableList);
  if (size == 0) {
S
Shengliang Guan 已提交
3751
    tsdbDebug("no qualified tables");
3752 3753
    return pTableGroup;
  }
H
Haojun Liao 已提交
3754

dengyihao's avatar
dengyihao 已提交
3755
  if (numOfOrderCols == 0 || size == 1) {  // no group by tags clause or only one table
3756
    SArray* sa = taosArrayDup(pTableList);
H
Haojun Liao 已提交
3757 3758 3759 3760
    if (sa == NULL) {
      taosArrayDestroy(pTableGroup);
      return NULL;
    }
H
Haojun Liao 已提交
3761

3762
    taosArrayPush(pTableGroup, &sa);
S
TD-1057  
Shengliang Guan 已提交
3763
    tsdbDebug("all %" PRIzu " tables belong to one group", size);
3764
  } else {
H
Haojun Liao 已提交
3765 3766
    STableGroupSupporter sup = {0};
    sup.numOfCols = numOfOrderCols;
3767
    sup.pTagSchema = pTagSchema->pSchema;
H
Haojun Liao 已提交
3768 3769
    sup.pCols = pCols;

3770 3771
    taosqsort(pTableList->pData, size, sizeof(STableKeyInfo), &sup, tableGroupComparFn);
    createTableGroupImpl(pTableGroup, pTableList, size, skey, &sup, tableGroupComparFn);
3772
  }
H
Haojun Liao 已提交
3773

3774 3775 3776
  return pTableGroup;
}

dengyihao's avatar
dengyihao 已提交
3777
// static bool tableFilterFp(const void* pNode, void* param) {
3778 3779 3780 3781 3782 3783 3784 3785 3786 3787 3788 3789 3790 3791 3792 3793 3794 3795 3796 3797 3798 3799 3800 3801 3802 3803 3804 3805 3806 3807 3808 3809 3810 3811 3812 3813 3814 3815 3816 3817 3818 3819 3820 3821 3822 3823 3824 3825 3826 3827 3828 3829 3830 3831 3832 3833 3834 3835 3836 3837 3838 3839 3840 3841 3842 3843 3844 3845 3846 3847 3848 3849 3850 3851 3852 3853 3854 3855 3856 3857 3858 3859 3860
//  tQueryInfo* pInfo = (tQueryInfo*) param;
//
//  STable* pTable = (STable*)(SL_GET_NODE_DATA((SSkipListNode*)pNode));
//
//  char* val = NULL;
//  if (pInfo->sch.colId == TSDB_TBNAME_COLUMN_INDEX) {
//    val = (char*) TABLE_NAME(pTable);
//  } else {
//    val = tdGetKVRowValOfCol(pTable->tagVal, pInfo->sch.colId);
//  }
//
//  if (pInfo->optr == TSDB_RELATION_ISNULL || pInfo->optr == TSDB_RELATION_NOTNULL) {
//    if (pInfo->optr == TSDB_RELATION_ISNULL) {
//      return (val == NULL) || isNull(val, pInfo->sch.type);
//    } else if (pInfo->optr == TSDB_RELATION_NOTNULL) {
//      return (val != NULL) && (!isNull(val, pInfo->sch.type));
//    }
//  } else if (pInfo->optr == TSDB_RELATION_IN) {
//     int type = pInfo->sch.type;
//     if (type == TSDB_DATA_TYPE_BOOL || IS_SIGNED_NUMERIC_TYPE(type) || type == TSDB_DATA_TYPE_TIMESTAMP) {
//       int64_t v;
//       GET_TYPED_DATA(v, int64_t, pInfo->sch.type, val);
//       return NULL != taosHashGet((SHashObj *)pInfo->q, (char *)&v, sizeof(v));
//     } else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
//       uint64_t v;
//       GET_TYPED_DATA(v, uint64_t, pInfo->sch.type, val);
//       return NULL != taosHashGet((SHashObj *)pInfo->q, (char *)&v, sizeof(v));
//     }
//     else if (type == TSDB_DATA_TYPE_DOUBLE || type == TSDB_DATA_TYPE_FLOAT) {
//       double v;
//       GET_TYPED_DATA(v, double, pInfo->sch.type, val);
//       return NULL != taosHashGet((SHashObj *)pInfo->q, (char *)&v, sizeof(v));
//     } else if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR){
//       return NULL != taosHashGet((SHashObj *)pInfo->q, varDataVal(val), varDataLen(val));
//     }
//
//  }
//
//  int32_t ret = 0;
//  if (val == NULL) { //the val is possible to be null, so check it out carefully
//    ret = -1; // val is missing in table tags value pairs
//  } else {
//    ret = pInfo->compare(val, pInfo->q);
//  }
//
//  switch (pInfo->optr) {
//    case TSDB_RELATION_EQUAL: {
//      return ret == 0;
//    }
//    case TSDB_RELATION_NOT_EQUAL: {
//      return ret != 0;
//    }
//    case TSDB_RELATION_GREATER_EQUAL: {
//      return ret >= 0;
//    }
//    case TSDB_RELATION_GREATER: {
//      return ret > 0;
//    }
//    case TSDB_RELATION_LESS_EQUAL: {
//      return ret <= 0;
//    }
//    case TSDB_RELATION_LESS: {
//      return ret < 0;
//    }
//    case TSDB_RELATION_LIKE: {
//      return ret == 0;
//    }
//    case TSDB_RELATION_MATCH: {
//      return ret == 0;
//    }
//    case TSDB_RELATION_NMATCH: {
//      return ret == 0;
//    }
//    case TSDB_RELATION_IN: {
//      return ret == 1;
//    }
//
//    default:
//      assert(false);
//  }
//
//  return true;
//}
H
Haojun Liao 已提交
3861

dengyihao's avatar
dengyihao 已提交
3862 3863
// static void getTableListfromSkipList(tExprNode *pExpr, SSkipList *pSkipList, SArray *result, SExprTraverseSupp
// *param);
3864

dengyihao's avatar
dengyihao 已提交
3865 3866
// static int32_t doQueryTableList(STable* pSTable, SArray* pRes, tExprNode* pExpr) {
//  //  // query according to the expression tree
3867
//  SExprTraverseSupp supp = {
dengyihao's avatar
dengyihao 已提交
3868
//      .nodeFilterFn = (__result_filter_fn_t)tableFilterFp,
3869 3870
//      .setupInfoFn = filterPrepare,
//      .pExtInfo = pSTable->tagSchema,
dengyihao's avatar
dengyihao 已提交
3871
//  };
3872 3873 3874 3875 3876
//
//  getTableListfromSkipList(pExpr, pSTable->pIndex, pRes, &supp);
//  tExprTreeDestroy(pExpr, destroyHelper);
//  return TSDB_CODE_SUCCESS;
//}
3877

H
Haojun Liao 已提交
3878
int32_t tsdbQuerySTableByTagCond(void* pMeta, uint64_t uid, TSKEY skey, const char* pTagCond, size_t len,
3879
                                 int16_t tagNameRelType, const char* tbnameCond, STableGroupInfo* pGroupInfo,
3880
                                 SColIndex* pColIndex, int32_t numOfCols, uint64_t reqId, uint64_t taskId) {
H
Hongze Cheng 已提交
3881 3882
  SMetaReader mr = {0};

H
Hongze Cheng 已提交
3883
  metaReaderInit(&mr, (SMeta*)pMeta, 0);
H
Hongze Cheng 已提交
3884 3885

  if (metaGetTableEntryByUid(&mr, uid) < 0) {
dengyihao's avatar
dengyihao 已提交
3886
    tsdbError("%p failed to get stable, uid:%" PRIu64 ", TID:0x%" PRIx64 " QID:0x%" PRIx64, pMeta, uid, taskId, reqId);
H
Hongze Cheng 已提交
3887
    metaReaderClear(&mr);
D
dapan1121 已提交
3888
    terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST;
3889
    goto _error;
C
Cary Xu 已提交
3890 3891
  } else {
    tsdbDebug("%p succeed to get stable, uid:%" PRIu64 ", TID:0x%" PRIx64 " QID:0x%" PRIx64, pMeta, uid, taskId, reqId);
3892
  }
H
Haojun Liao 已提交
3893

H
Hongze Cheng 已提交
3894
  if (mr.me.type != TSDB_SUPER_TABLE) {
dengyihao's avatar
dengyihao 已提交
3895 3896 3897
    tsdbError("%p query normal tag not allowed, uid:%" PRIu64 ", TID:0x%" PRIx64 " QID:0x%" PRIx64, pMeta, uid, taskId,
              reqId);
    terrno = TSDB_CODE_OPS_NOT_SUPPORT;  // basically, this error is caused by invalid sql issued by client
H
Hongze Cheng 已提交
3898
    metaReaderClear(&mr);
3899
    goto _error;
H
hjxilinx 已提交
3900
  }
3901

H
Hongze Cheng 已提交
3902 3903
  metaReaderClear(&mr);

dengyihao's avatar
dengyihao 已提交
3904 3905
  // NOTE: not add ref count for super table
  SArray*         res = taosArrayInit(8, sizeof(STableKeyInfo));
H
Hongze Cheng 已提交
3906
  SSchemaWrapper* pTagSchema = metaGetTableSchema(pMeta, uid, 1, true);
H
Haojun Liao 已提交
3907

weixin_48148422's avatar
weixin_48148422 已提交
3908 3909
  // no tags and tbname condition, all child tables of this stable are involved
  if (tbnameCond == NULL && (pTagCond == NULL || len == 0)) {
H
Haojun Liao 已提交
3910
    int32_t ret = getAllTableList(pMeta, uid, res);
3911 3912
    if (ret != TSDB_CODE_SUCCESS) {
      goto _error;
3913
    }
3914

dengyihao's avatar
dengyihao 已提交
3915 3916
    pGroupInfo->numOfTables = (uint32_t)taosArrayGetSize(res);
    pGroupInfo->pGroupList = createTableGroup(res, pTagSchema, pColIndex, numOfCols, skey);
H
Haojun Liao 已提交
3917

dengyihao's avatar
dengyihao 已提交
3918 3919 3920
    tsdbDebug("%p no table name/tag condition, all tables qualified, numOfTables:%u, group:%zu, TID:0x%" PRIx64
              " QID:0x%" PRIx64,
              pMeta, pGroupInfo->numOfTables, taosArrayGetSize(pGroupInfo->pGroupList), taskId, reqId);
3921

3922
    taosArrayDestroy(res);
3923 3924
    return ret;
  }
3925

H
hjxilinx 已提交
3926
  int32_t ret = TSDB_CODE_SUCCESS;
3927

dengyihao's avatar
dengyihao 已提交
3928 3929
  SFilterInfo* filterInfo = NULL;
  ret = filterInitFromNode((SNode*)pTagCond, &filterInfo, 0);
dengyihao's avatar
dengyihao 已提交
3930 3931 3932 3933 3934 3935 3936 3937
  if (ret != TSDB_CODE_SUCCESS) {
    terrno = ret;
    return ret;
  }
  ret = tsdbQueryTableList(pMeta, res, filterInfo);
  pGroupInfo->numOfTables = (uint32_t)taosArrayGetSize(res);
  pGroupInfo->pGroupList = createTableGroup(res, pTagSchema, pColIndex, numOfCols, skey);

dengyihao's avatar
dengyihao 已提交
3938 3939
  // tsdbDebug("%p stable tid:%d, uid:%" PRIu64 " query, numOfTables:%u, belong to %" PRIzu " groups", tsdb,
  //          pTable->tableId, pTable->uid, pGroupInfo->numOfTables, taosArrayGetSize(pGroupInfo->pGroupList));
dengyihao's avatar
dengyihao 已提交
3940 3941 3942 3943 3944

  taosArrayDestroy(res);
  return ret;

_error:
3945
  return terrno;
3946
}
3947

dengyihao's avatar
dengyihao 已提交
3948 3949 3950 3951 3952
int32_t tsdbQueryTableList(void* pMeta, SArray* pRes, void* filterInfo) {
  // impl later

  return TSDB_CODE_SUCCESS;
}
H
Haojun Liao 已提交
3953
int32_t tsdbGetOneTableGroup(void* pMeta, uint64_t uid, TSKEY startKey, STableGroupInfo* pGroupInfo) {
H
Hongze Cheng 已提交
3954 3955
  SMetaReader mr = {0};

H
Hongze Cheng 已提交
3956
  metaReaderInit(&mr, (SMeta*)pMeta, 0);
H
Hongze Cheng 已提交
3957 3958

  if (metaGetTableEntryByUid(&mr, uid) < 0) {
D
dapan1121 已提交
3959
    terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST;
3960
    goto _error;
3961
  }
3962

H
Hongze Cheng 已提交
3963 3964
  metaReaderClear(&mr);

3965 3966
  pGroupInfo->numOfTables = 1;
  pGroupInfo->pGroupList = taosArrayInit(1, POINTER_BYTES);
H
Haojun Liao 已提交
3967

H
Haojun Liao 已提交
3968 3969
  SArray* group = taosArrayInit(1, sizeof(STableKeyInfo));

3970
  STableKeyInfo info = {.lastKey = startKey, .uid = uid};
H
Haojun Liao 已提交
3971
  taosArrayPush(group, &info);
H
Haojun Liao 已提交
3972

3973
  taosArrayPush(pGroupInfo->pGroupList, &group);
3974
  return TSDB_CODE_SUCCESS;
3975

dengyihao's avatar
dengyihao 已提交
3976
_error:
H
Hongze Cheng 已提交
3977
  metaReaderClear(&mr);
3978
  return terrno;
3979
}
3980

3981
#if 0
3982
int32_t tsdbGetTableGroupFromIdListT(STsdb* tsdb, SArray* pTableIdList, STableGroupInfo* pGroupInfo) {
B
Bomin Zhang 已提交
3983 3984 3985
  if (tsdbRLockRepoMeta(tsdb) < 0) {
    return terrno;
  }
3986 3987 3988 3989

  assert(pTableIdList != NULL);
  size_t size = taosArrayGetSize(pTableIdList);
  pGroupInfo->pGroupList = taosArrayInit(1, POINTER_BYTES);
H
Haojun Liao 已提交
3990
  SArray* group = taosArrayInit(1, sizeof(STableKeyInfo));
3991

B
Bomin Zhang 已提交
3992
  for(int32_t i = 0; i < size; ++i) {
3993 3994 3995 3996 3997 3998 3999 4000 4001 4002 4003
    STableIdInfo *id = taosArrayGet(pTableIdList, i);

    STable* pTable = tsdbGetTableByUid(tsdbGetMeta(tsdb), id->uid);
    if (pTable == NULL) {
      tsdbWarn("table uid:%"PRIu64", tid:%d has been drop already", id->uid, id->tid);
      continue;
    }

    if (pTable->type == TSDB_SUPER_TABLE) {
      tsdbError("direct query on super tale is not allowed, table uid:%"PRIu64", tid:%d", id->uid, id->tid);
      terrno = TSDB_CODE_QRY_INVALID_MSG;
D
fix bug  
dapan1121 已提交
4004 4005 4006
      tsdbUnlockRepoMeta(tsdb);
      taosArrayDestroy(group);
      return terrno;
4007 4008
    }

H
Haojun Liao 已提交
4009 4010
    STableKeyInfo info = {.pTable = pTable, .lastKey = id->key};
    taosArrayPush(group, &info);
4011 4012
  }

B
Bomin Zhang 已提交
4013 4014 4015 4016
  if (tsdbUnlockRepoMeta(tsdb) < 0) {
    taosArrayDestroy(group);
    return terrno;
  }
4017

sangshuduo's avatar
sangshuduo 已提交
4018
  pGroupInfo->numOfTables = (uint32_t) taosArrayGetSize(group);
B
Bomin Zhang 已提交
4019 4020 4021 4022 4023
  if (pGroupInfo->numOfTables > 0) {
    taosArrayPush(pGroupInfo->pGroupList, &group);
  } else {
    taosArrayDestroy(group);
  }
4024 4025 4026

  return TSDB_CODE_SUCCESS;
}
4027
#endif
4028 4029 4030 4031 4032 4033 4034 4035
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);
wafwerar's avatar
wafwerar 已提交
4036
    taosMemoryFreeClear(pColInfo->pData);
4037 4038 4039 4040 4041 4042
  }

  taosArrayDestroy(pColumnInfoData);
  return NULL;
}

H
Haojun Liao 已提交
4043 4044 4045 4046 4047 4048
static void* destroyTableCheckInfo(SArray* pTableCheckInfo) {
  size_t size = taosArrayGetSize(pTableCheckInfo);
  for (int32_t i = 0; i < size; ++i) {
    STableCheckInfo* p = taosArrayGet(pTableCheckInfo, i);
    destroyTableMemIterator(p);

wafwerar's avatar
wafwerar 已提交
4049
    taosMemoryFreeClear(p->pCompInfo);
H
Haojun Liao 已提交
4050 4051 4052 4053 4054 4055
  }

  taosArrayDestroy(pTableCheckInfo);
  return NULL;
}

H
Haojun Liao 已提交
4056
void tsdbCleanupReadHandle(tsdbReaderT queryHandle) {
4057 4058
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)queryHandle;
  if (pTsdbReadHandle == NULL) {
4059 4060
    return;
  }
4061

4062
  pTsdbReadHandle->pColumns = doFreeColumnInfoData(pTsdbReadHandle->pColumns);
4063

4064
  taosArrayDestroy(pTsdbReadHandle->suppInfo.defaultLoadColumn);
wafwerar's avatar
wafwerar 已提交
4065
  taosMemoryFreeClear(pTsdbReadHandle->pDataBlockInfo);
4066 4067
  taosMemoryFreeClear(pTsdbReadHandle->suppInfo.pstatis);
  taosMemoryFreeClear(pTsdbReadHandle->suppInfo.plist);
4068

4069
  if (!emptyQueryTimewindow(pTsdbReadHandle)) {
dengyihao's avatar
dengyihao 已提交
4070
    //    tsdbMayUnTakeMemSnapshot(pTsdbReadHandle);
4071
  } else {
4072
    assert(pTsdbReadHandle->pTableCheckInfo == NULL);
4073 4074
  }

4075 4076
  if (pTsdbReadHandle->pTableCheckInfo != NULL) {
    pTsdbReadHandle->pTableCheckInfo = destroyTableCheckInfo(pTsdbReadHandle->pTableCheckInfo);
4077
  }
4078

4079
  tsdbDestroyReadH(&pTsdbReadHandle->rhelper);
H
Haojun Liao 已提交
4080

4081 4082
  tdFreeDataCols(pTsdbReadHandle->pDataCols);
  pTsdbReadHandle->pDataCols = NULL;
H
Haojun Liao 已提交
4083

4084 4085
  pTsdbReadHandle->prev = doFreeColumnInfoData(pTsdbReadHandle->prev);
  pTsdbReadHandle->next = doFreeColumnInfoData(pTsdbReadHandle->next);
4086

4087
  SIOCostSummary* pCost = &pTsdbReadHandle->cost;
4088

dengyihao's avatar
dengyihao 已提交
4089 4090 4091 4092
  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",
            pTsdbReadHandle, pCost->headFileLoad, pCost->headFileLoadTime, pCost->statisInfoLoadTime,
            pCost->blockLoadTime, pCost->checkForNextTime, pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
4093

wafwerar's avatar
wafwerar 已提交
4094
  taosMemoryFreeClear(pTsdbReadHandle);
4095
}
4096

4097
#if 0
H
Haojun Liao 已提交
4098
void tsdbDestroyTableGroup(STableGroupInfo *pGroupList) {
4099 4100 4101 4102 4103 4104 4105 4106 4107 4108
  assert(pGroupList != NULL);

  size_t numOfGroup = taosArrayGetSize(pGroupList->pGroupList);

  for(int32_t i = 0; i < numOfGroup; ++i) {
    SArray* p = taosArrayGetP(pGroupList->pGroupList, i);

    size_t numOfTables = taosArrayGetSize(p);
    for(int32_t j = 0; j < numOfTables; ++j) {
      STable* pTable = taosArrayGetP(p, j);
4109 4110 4111 4112
      if (pTable != NULL) { // in case of handling retrieve data from tsdb
        tsdbUnRefTable(pTable);
      }
      //assert(pTable != NULL);
4113 4114 4115 4116 4117
    }

    taosArrayDestroy(p);
  }

4118
  taosHashCleanup(pGroupList->map);
4119
  taosArrayDestroy(pGroupList->pGroupList);
H
Haojun Liao 已提交
4120
  pGroupList->numOfTables = 0;
4121
}
H
Haojun Liao 已提交
4122 4123 4124 4125 4126 4127 4128

static void applyFilterToSkipListNode(SSkipList *pSkipList, tExprNode *pExpr, SArray *pResult, SExprTraverseSupp *param) {
  SSkipListIterator* iter = tSkipListCreateIter(pSkipList);

  // Scan each node in the skiplist by using iterator
  while (tSkipListIterNext(iter)) {
    SSkipListNode *pNode = tSkipListIterGet(iter);
H
Haojun Liao 已提交
4129
    if (exprTreeApplyFilter(pExpr, pNode, param)) {
H
Haojun Liao 已提交
4130 4131 4132 4133 4134 4135 4136 4137 4138 4139 4140 4141 4142 4143 4144 4145 4146 4147 4148 4149 4150 4151 4152
      taosArrayPush(pResult, &(SL_GET_NODE_DATA(pNode)));
    }
  }

  tSkipListDestroyIter(iter);
}

typedef struct {
  char*    v;
  int32_t  optr;
} SEndPoint;

typedef struct {
  SEndPoint* start;
  SEndPoint* end;
} SQueryCond;

// todo check for malloc failure
static int32_t setQueryCond(tQueryInfo *queryColInfo, SQueryCond* pCond) {
  int32_t optr = queryColInfo->optr;

  if (optr == TSDB_RELATION_GREATER || optr == TSDB_RELATION_GREATER_EQUAL ||
      optr == TSDB_RELATION_EQUAL || optr == TSDB_RELATION_NOT_EQUAL) {
wafwerar's avatar
wafwerar 已提交
4153
    pCond->start       = taosMemoryCalloc(1, sizeof(SEndPoint));
H
Haojun Liao 已提交
4154
    pCond->start->optr = queryColInfo->optr;
Y
yihaoDeng 已提交
4155
    pCond->start->v    = queryColInfo->q;
H
Haojun Liao 已提交
4156
  } else if (optr == TSDB_RELATION_LESS || optr == TSDB_RELATION_LESS_EQUAL) {
wafwerar's avatar
wafwerar 已提交
4157
    pCond->end       = taosMemoryCalloc(1, sizeof(SEndPoint));
H
Haojun Liao 已提交
4158
    pCond->end->optr = queryColInfo->optr;
Y
yihaoDeng 已提交
4159 4160
    pCond->end->v    = queryColInfo->q;
  } else if (optr == TSDB_RELATION_IN) {
wafwerar's avatar
wafwerar 已提交
4161
    pCond->start       = taosMemoryCalloc(1, sizeof(SEndPoint));
Y
yihaoDeng 已提交
4162 4163 4164
    pCond->start->optr = queryColInfo->optr;
    pCond->start->v    = queryColInfo->q; 
  } else if (optr == TSDB_RELATION_LIKE) {
H
Haojun Liao 已提交
4165
    assert(0);
4166 4167
  } else if (optr == TSDB_RELATION_MATCH) {
    assert(0);
4168 4169
  } else if (optr == TSDB_RELATION_NMATCH) {
    assert(0);
H
Haojun Liao 已提交
4170 4171 4172 4173 4174 4175 4176 4177 4178 4179 4180 4181 4182 4183 4184 4185 4186 4187 4188 4189 4190 4191 4192 4193 4194 4195 4196 4197 4198 4199 4200 4201 4202 4203 4204 4205 4206 4207 4208 4209 4210 4211 4212 4213 4214 4215 4216 4217 4218 4219 4220 4221 4222 4223 4224 4225 4226 4227 4228 4229 4230 4231 4232 4233 4234 4235 4236 4237 4238 4239 4240 4241 4242 4243 4244 4245 4246 4247 4248 4249 4250 4251 4252
  }

  return TSDB_CODE_SUCCESS;
}

static void queryIndexedColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo, SArray* result) {
  SSkipListIterator* iter = NULL;

  SQueryCond cond = {0};
  if (setQueryCond(pQueryInfo, &cond) != TSDB_CODE_SUCCESS) {
    //todo handle error
  }

  if (cond.start != NULL) {
    iter = tSkipListCreateIterFromVal(pSkipList, (char*) cond.start->v, pSkipList->type, TSDB_ORDER_ASC);
  } else {
    iter = tSkipListCreateIterFromVal(pSkipList, (char*)(cond.end ? cond.end->v: NULL), pSkipList->type, TSDB_ORDER_DESC);
  }

  if (cond.start != NULL) {
    int32_t optr = cond.start->optr;

    if (optr == TSDB_RELATION_EQUAL) {   // equals
      while(tSkipListIterNext(iter)) {
        SSkipListNode* pNode = tSkipListIterGet(iter);

        int32_t ret = pQueryInfo->compare(SL_GET_NODE_KEY(pSkipList, pNode), cond.start->v);
        if (ret != 0) {
          break;
        }

        STableKeyInfo info = {.pTable = (void*)SL_GET_NODE_DATA(pNode), .lastKey = TSKEY_INITIAL_VAL};
        taosArrayPush(result, &info);
      }
    } else if (optr == TSDB_RELATION_GREATER || optr == TSDB_RELATION_GREATER_EQUAL) { // greater equal
      bool comp = true;
      int32_t ret = 0;

      while(tSkipListIterNext(iter)) {
        SSkipListNode* pNode = tSkipListIterGet(iter);

        if (comp) {
          ret = pQueryInfo->compare(SL_GET_NODE_KEY(pSkipList, pNode), cond.start->v);
          assert(ret >= 0);
        }

        if (ret == 0 && optr == TSDB_RELATION_GREATER) {
          continue;
        } else {
          STableKeyInfo info = {.pTable = (void*)SL_GET_NODE_DATA(pNode), .lastKey = TSKEY_INITIAL_VAL};
          taosArrayPush(result, &info);
          comp = false;
        }
      }
    } else if (optr == TSDB_RELATION_NOT_EQUAL) {   // not equal
      bool comp = true;

      while(tSkipListIterNext(iter)) {
        SSkipListNode* pNode = tSkipListIterGet(iter);
        comp = comp && (pQueryInfo->compare(SL_GET_NODE_KEY(pSkipList, pNode), cond.start->v) == 0);
        if (comp) {
          continue;
        }

        STableKeyInfo info = {.pTable = (void*)SL_GET_NODE_DATA(pNode), .lastKey = TSKEY_INITIAL_VAL};
        taosArrayPush(result, &info);
      }

      tSkipListDestroyIter(iter);

      comp = true;
      iter = tSkipListCreateIterFromVal(pSkipList, (char*) cond.start->v, pSkipList->type, TSDB_ORDER_DESC);
      while(tSkipListIterNext(iter)) {
        SSkipListNode* pNode = tSkipListIterGet(iter);
        comp = comp && (pQueryInfo->compare(SL_GET_NODE_KEY(pSkipList, pNode), cond.start->v) == 0);
        if (comp) {
          continue;
        }

        STableKeyInfo info = {.pTable = (void*)SL_GET_NODE_DATA(pNode), .lastKey = TSKEY_INITIAL_VAL};
        taosArrayPush(result, &info);
      }

Y
yihaoDeng 已提交
4253 4254 4255 4256 4257 4258 4259 4260 4261 4262 4263 4264 4265
    } else if (optr == TSDB_RELATION_IN) {
      while(tSkipListIterNext(iter)) {
        SSkipListNode* pNode = tSkipListIterGet(iter);

        int32_t ret = pQueryInfo->compare(SL_GET_NODE_KEY(pSkipList, pNode), cond.start->v);
        if (ret != 0) {
          break;
        }

        STableKeyInfo info = {.pTable = (void*)SL_GET_NODE_DATA(pNode), .lastKey = TSKEY_INITIAL_VAL};
        taosArrayPush(result, &info);
      }
      
H
Haojun Liao 已提交
4266 4267 4268 4269 4270 4271 4272 4273 4274 4275 4276 4277 4278 4279 4280 4281 4282 4283 4284 4285 4286 4287 4288 4289 4290 4291 4292 4293 4294 4295 4296 4297 4298 4299 4300 4301 4302 4303 4304 4305 4306
    } else {
      assert(0);
    }
  } else {
    int32_t optr = cond.end ? cond.end->optr : TSDB_RELATION_INVALID;
    if (optr == TSDB_RELATION_LESS || optr == TSDB_RELATION_LESS_EQUAL) {
      bool    comp = true;
      int32_t ret = 0;

      while (tSkipListIterNext(iter)) {
        SSkipListNode *pNode = tSkipListIterGet(iter);

        if (comp) {
          ret = pQueryInfo->compare(SL_GET_NODE_KEY(pSkipList, pNode), cond.end->v);
          assert(ret <= 0);
        }

        if (ret == 0 && optr == TSDB_RELATION_LESS) {
          continue;
        } else {
          STableKeyInfo info = {.pTable = (void *)SL_GET_NODE_DATA(pNode), .lastKey = TSKEY_INITIAL_VAL};
          taosArrayPush(result, &info);
          comp = false;  // no need to compare anymore
        }
      }
    } else {
      assert(pQueryInfo->optr == TSDB_RELATION_ISNULL || pQueryInfo->optr == TSDB_RELATION_NOTNULL);

      while (tSkipListIterNext(iter)) {
        SSkipListNode *pNode = tSkipListIterGet(iter);

        bool isnull = isNull(SL_GET_NODE_KEY(pSkipList, pNode), pQueryInfo->sch.type);
        if ((pQueryInfo->optr == TSDB_RELATION_ISNULL && isnull) ||
            (pQueryInfo->optr == TSDB_RELATION_NOTNULL && (!isnull))) {
          STableKeyInfo info = {.pTable = (void *)SL_GET_NODE_DATA(pNode), .lastKey = TSKEY_INITIAL_VAL};
          taosArrayPush(result, &info);
        }
      }
    }
  }

wafwerar's avatar
wafwerar 已提交
4307 4308
  taosMemoryFree(cond.start);
  taosMemoryFree(cond.end);
H
Haojun Liao 已提交
4309 4310 4311 4312 4313 4314 4315 4316 4317 4318 4319 4320 4321 4322 4323 4324 4325 4326
  tSkipListDestroyIter(iter);
}

static void queryIndexlessColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo, SArray* res, __result_filter_fn_t filterFp) {
  SSkipListIterator* iter = tSkipListCreateIter(pSkipList);

  while (tSkipListIterNext(iter)) {
    bool addToResult = false;

    SSkipListNode *pNode = tSkipListIterGet(iter);

    char *pData = SL_GET_NODE_DATA(pNode);
    tstr *name = (tstr*) tsdbGetTableName((void*) pData);

    // todo speed up by using hash
    if (pQueryInfo->sch.colId == TSDB_TBNAME_COLUMN_INDEX) {
      if (pQueryInfo->optr == TSDB_RELATION_IN) {
        addToResult = pQueryInfo->compare(name, pQueryInfo->q);
4327 4328 4329
      } else if (pQueryInfo->optr == TSDB_RELATION_LIKE ||
                 pQueryInfo->optr == TSDB_RELATION_MATCH ||
                 pQueryInfo->optr == TSDB_RELATION_NMATCH) {
H
Haojun Liao 已提交
4330 4331 4332 4333 4334 4335 4336 4337 4338 4339 4340 4341 4342 4343 4344 4345
        addToResult = !pQueryInfo->compare(name, pQueryInfo->q);
      }
    } else {
      addToResult = filterFp(pNode, pQueryInfo);
    }

    if (addToResult) {
      STableKeyInfo info = {.pTable = (void*)pData, .lastKey = TSKEY_INITIAL_VAL};
      taosArrayPush(res, &info);
    }
  }

  tSkipListDestroyIter(iter);
}

// Apply the filter expression to each node in the skiplist to acquire the qualified nodes in skip list
dengyihao's avatar
dengyihao 已提交
4346 4347 4348 4349 4350 4351 4352 4353 4354 4355 4356 4357 4358 4359 4360 4361 4362 4363 4364 4365 4366 4367 4368 4369 4370 4371 4372 4373 4374 4375 4376 4377 4378
//void getTableListfromSkipList(tExprNode *pExpr, SSkipList *pSkipList, SArray *result, SExprTraverseSupp *param) {
//  if (pExpr == NULL) {
//    return;
//  }
//
//  tExprNode *pLeft  = pExpr->_node.pLeft;
//  tExprNode *pRight = pExpr->_node.pRight;
//
//  // column project
//  if (pLeft->nodeType != TSQL_NODE_EXPR && pRight->nodeType != TSQL_NODE_EXPR) {
//    assert(pLeft->nodeType == TSQL_NODE_COL && (pRight->nodeType == TSQL_NODE_VALUE || pRight->nodeType == TSQL_NODE_DUMMY));
//
//    param->setupInfoFn(pExpr, param->pExtInfo);
//
//    tQueryInfo *pQueryInfo = pExpr->_node.info;
//    if (pQueryInfo->indexed && (pQueryInfo->optr != TSDB_RELATION_LIKE
//                                && pQueryInfo->optr != TSDB_RELATION_MATCH && pQueryInfo->optr != TSDB_RELATION_NMATCH
//                                && pQueryInfo->optr != TSDB_RELATION_IN)) {
//      queryIndexedColumn(pSkipList, pQueryInfo, result);
//    } else {
//      queryIndexlessColumn(pSkipList, pQueryInfo, result, param->nodeFilterFn);
//    }
//
//    return;
//  }
//
//  // The value of hasPK is always 0.
//  uint8_t weight = pLeft->_node.hasPK + pRight->_node.hasPK;
//  assert(weight == 0 && pSkipList != NULL && taosArrayGetSize(result) == 0);
//
//  //apply the hierarchical filter expression to every node in skiplist to find the qualified nodes
//  applyFilterToSkipListNode(pSkipList, pExpr, result, param);
//}
L
Liu Jicong 已提交
4379
#endif