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

H
Hongze Cheng 已提交
16
#include "tsdb.h"
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) {
C
Cary Xu 已提交
1641 1642
    // pSchema1 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, TD_ROW_SVER(row1));
    // TODO: use the real schemaVersion
H
Hongze Cheng 已提交
1643
    pSchema1 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, 1);
1644
  }
1645

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

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

1656
  int32_t numOfColsOfRow2 = 0;
dengyihao's avatar
dengyihao 已提交
1657
  if (row2) {
H
Haojun Liao 已提交
1658
    isRow2DataRow = TD_IS_TP_ROW(row2);
1659
    if (pSchema2 == NULL) {
C
Cary Xu 已提交
1660 1661
      // pSchema2 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, TD_ROW_SVER(row2));
      // TODO: use the real schemaVersion
H
Hongze Cheng 已提交
1662
      pSchema2 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, 1);
1663
    }
dengyihao's avatar
dengyihao 已提交
1664
    if (isRow2DataRow) {
1665 1666
      numOfColsOfRow2 = schemaNCols(pSchema2);
    } else {
H
Haojun Liao 已提交
1667
      numOfColsOfRow2 = tdRowGetNCols(row2);
1668 1669
    }
  }
C
Cary Xu 已提交
1670

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

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

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

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

dengyihao's avatar
dengyihao 已提交
1723
    if (isChosenRowDataRow) {
1724 1725
      colId = pSchema->columns[chosen_itr].colId;
      offset = pSchema->columns[chosen_itr].offset;
C
Cary Xu 已提交
1726 1727
      // TODO: use STSRowIter
      tdSTpRowGetVal(row, colId, pSchema->columns[chosen_itr].type, pSchema->flen, offset, chosen_itr - 1, &sVal);
C
Cary Xu 已提交
1728 1729 1730
      if (colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
        rowKey = *(TSKEY*)sVal.val;
        if (rowKey != *lastRowKey) {
C
Cary Xu 已提交
1731
          mergeOption = 1;
C
Cary Xu 已提交
1732 1733 1734 1735
          if (*lastRowKey != TSKEY_INITIAL_VAL) {
            ++(*curRow);
          }
          ++nResult;
C
Cary Xu 已提交
1736
        } else if (update) {
C
Cary Xu 已提交
1737 1738 1739 1740
          mergeOption = 2;
        } else {
          mergeOption = 0;
          break;
C
Cary Xu 已提交
1741
        }
C
Cary Xu 已提交
1742

C
Cary Xu 已提交
1743 1744
        *lastRowKey = rowKey;
      }
1745
    } else {
C
Cary Xu 已提交
1746 1747 1748 1749
      // TODO: use STSRowIter
      if (chosen_itr == 0) {
        colId = PRIMARYKEY_TIMESTAMP_COL_ID;
        tdSKvRowGetVal(row, PRIMARYKEY_TIMESTAMP_COL_ID, -1, -1, &sVal);
C
Cary Xu 已提交
1750 1751
        rowKey = *(TSKEY*)sVal.val;
        if (rowKey != *lastRowKey) {
C
Cary Xu 已提交
1752
          mergeOption = 1;
C
Cary Xu 已提交
1753 1754 1755 1756
          if (*lastRowKey != TSKEY_INITIAL_VAL) {
            ++(*curRow);
          }
          ++nResult;
C
Cary Xu 已提交
1757
        } else if (update) {
C
Cary Xu 已提交
1758 1759 1760 1761
          mergeOption = 2;
        } else {
          mergeOption = 0;
          break;
C
Cary Xu 已提交
1762 1763
        }
        *lastRowKey = rowKey;
C
Cary Xu 已提交
1764 1765 1766 1767 1768 1769
      } else {
        SKvRowIdx* pColIdx = tdKvRowColIdxAt(row, chosen_itr - 1);
        colId = pColIdx->colId;
        offset = pColIdx->offset;
        tdSKvRowGetVal(row, colId, offset, chosen_itr - 1, &sVal);
      }
1770 1771
    }

C
Cary Xu 已提交
1772 1773
    ASSERT(rowKey != TSKEY_INITIAL_VAL);

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

C
Cary Xu 已提交
1788
      ++i;
C
Cary Xu 已提交
1789

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

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

  return nResult;
H
Haojun Liao 已提交
1812
#endif
1813
}
1814

dengyihao's avatar
dengyihao 已提交
1815 1816
static void getQualifiedRowsPos(STsdbReadHandle* pTsdbReadHandle, int32_t startPos, int32_t endPos,
                                int32_t numOfExisted, int32_t* start, int32_t* end) {
1817 1818
  *start = -1;

1819
  if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
1820
    int32_t remain = endPos - startPos + 1;
1821 1822
    if (remain + numOfExisted > pTsdbReadHandle->outputCapacity) {
      *end = (pTsdbReadHandle->outputCapacity - numOfExisted) + startPos - 1;
H
Haojun Liao 已提交
1823 1824
    } else {
      *end = endPos;
1825 1826 1827 1828 1829
    }

    *start = startPos;
  } else {
    int32_t remain = (startPos - endPos) + 1;
1830 1831
    if (remain + numOfExisted > pTsdbReadHandle->outputCapacity) {
      *end = startPos + 1 - (pTsdbReadHandle->outputCapacity - numOfExisted);
H
Haojun Liao 已提交
1832 1833
    } else {
      *end = endPos;
1834 1835 1836 1837 1838 1839 1840
    }

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

dengyihao's avatar
dengyihao 已提交
1841 1842
static void updateInfoAfterMerge(STsdbReadHandle* pTsdbReadHandle, STableCheckInfo* pCheckInfo, int32_t numOfRows,
                                 int32_t endPos) {
1843
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
1844 1845

  pCheckInfo->lastKey = cur->lastKey;
1846
  pTsdbReadHandle->realNumOfRows = numOfRows;
1847 1848 1849 1850
  cur->rows = numOfRows;
  cur->pos = endPos;
}

1851 1852
static void doCheckGeneratedBlockRange(STsdbReadHandle* pTsdbReadHandle) {
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
H
Haojun Liao 已提交
1853 1854

  if (cur->rows > 0) {
1855 1856
    if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
      assert(cur->win.skey >= pTsdbReadHandle->window.skey && cur->win.ekey <= pTsdbReadHandle->window.ekey);
H
Haojun Liao 已提交
1857
    } else {
1858
      assert(cur->win.skey >= pTsdbReadHandle->window.ekey && cur->win.ekey <= pTsdbReadHandle->window.skey);
H
Haojun Liao 已提交
1859 1860
    }

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

dengyihao's avatar
dengyihao 已提交
1867
    int32_t step = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 1 : -1;
1868
    cur->lastKey = pTsdbReadHandle->window.ekey + step;
H
Haojun Liao 已提交
1869 1870 1871
  }
}

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

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

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

1881
  int32_t step = ascScan? 1 : -1;
H
Haojun Liao 已提交
1882 1883 1884 1885

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

1886
  if (!ascScan) {
wafwerar's avatar
wafwerar 已提交
1887
    TSWAP(start, end);
H
Haojun Liao 已提交
1888 1889
  }

1890 1891
  assert(pTsdbReadHandle->outputCapacity >= (end - start + 1));
  int32_t numOfRows = doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, 0, start, end);
H
Haojun Liao 已提交
1892 1893

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

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

dengyihao's avatar
dengyihao 已提交
1904 1905 1906
  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 已提交
1907 1908
}

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

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

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

  return endPos;
}

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

1963
  initTableMemIterator(pTsdbReadHandle, pCheckInfo);
1964

1965 1966
  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 已提交
1967
         cur->pos >= 0 && cur->pos < pBlock->numOfRows);
H
Haojun Liao 已提交
1968

1969
  TSKEY* tsArray = pCols->cols[0].pData;
dengyihao's avatar
dengyihao 已提交
1970 1971
  assert(pCols->numOfRows == pBlock->numOfRows && tsArray[0] == pBlock->keyFirst &&
         tsArray[pBlock->numOfRows - 1] == pBlock->keyLast);
1972

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

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

1979 1980
  int32_t numOfCols = (int32_t)(QH_GET_NUM_OF_COLS(pTsdbReadHandle));
  int32_t endPos = getEndPosInDataBlock(pTsdbReadHandle, &blockInfo);
H
Haojun Liao 已提交
1981

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

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

dengyihao's avatar
dengyihao 已提交
1991 1992
  int16_t   rv1 = -1;
  int16_t   rv2 = -1;
1993 1994
  STSchema* pSchema1 = NULL;
  STSchema* pSchema2 = NULL;
D
fix bug  
dapan1121 已提交
1995

H
Haojun Liao 已提交
1996 1997
  int32_t pos = cur->pos;
  cur->win = TSWINDOW_INITIALIZER;
1998

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

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

H
Haojun Liao 已提交
2014
      TSKEY key = TD_ROW_KEY(row1);
2015
      if ((key > pTsdbReadHandle->window.ekey && ascScan) || (key < pTsdbReadHandle->window.ekey && !ascScan)) {
2016 2017 2018
        break;
      }

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

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

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

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

2060
          bool forceSetNull = pCfg->update != TD_ROW_PARTIAL_UPDATE;
dengyihao's avatar
dengyihao 已提交
2061
          mergeTwoRowFromMem(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, numOfRows, row1, row2, numOfCols,
C
Cary Xu 已提交
2062
                             pCheckInfo->tableId, pSchema1, pSchema2, forceSetNull, &lastRowKey);
H
TD-1439  
Hongze Cheng 已提交
2063 2064 2065 2066 2067 2068 2069 2070 2071
          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 已提交
2072 2073 2074 2075 2076 2077 2078
          moveToNextRowInMem(pCheckInfo);
          pos += step;
        } else {
          moveToNextRowInMem(pCheckInfo);
        }
#endif
        if (TD_SUPPORT_UPDATE(pCfg->update)) {
2079 2080 2081 2082
          if (lastKeyAppend != key) {
            if (lastKeyAppend != TSKEY_INITIAL_VAL) {
              ++curRow;
            }
2083
            lastKeyAppend = key;
2084 2085
          }
          // load data from file firstly
C
Cary Xu 已提交
2086
          numOfRows = doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, curRow, pos, pos);
C
Cary Xu 已提交
2087 2088 2089 2090 2091 2092 2093

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

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

          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 已提交
2107
          moveToNextRowInMem(pCheckInfo);
2108

H
TD-1439  
Hongze Cheng 已提交
2109 2110 2111 2112
          pos += step;
        } else {
          moveToNextRowInMem(pCheckInfo);
        }
2113
      } else if ((key > tsArray[pos] && ascScan) || (key < tsArray[pos] && !ascScan)) {
2114 2115 2116
        if (cur->win.skey == TSKEY_INITIAL_VAL) {
          cur->win.skey = tsArray[pos];
        }
2117

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

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

2136
        int32_t qstart = 0, qend = 0;
2137
        getQualifiedRowsPos(pTsdbReadHandle, pos, end, numOfRows, &qstart, &qend);
2138

2139
        if ((lastKeyAppend != TSKEY_INITIAL_VAL) && (lastKeyAppend != (ascScan ? tsArray[qstart] : tsArray[qend]))) {
C
Cary Xu 已提交
2140 2141
          ++curRow;
        }
2142

C
Cary Xu 已提交
2143
        numOfRows = doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, curRow, qstart, qend);
2144
        pos += (qend - qstart + 1) * step;
C
Cary Xu 已提交
2145 2146 2147
        if (numOfRows > 0) {
          curRow = numOfRows - 1;
        }
2148

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

2155
    if (numOfRows < pTsdbReadHandle->outputCapacity) {
H
Haojun Liao 已提交
2156 2157 2158 2159
      /**
       * 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.
       */
2160
      if (node == NULL ||
2161 2162
          ((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)) {
2163 2164 2165 2166 2167
        // 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];
        }

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

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

2174
        cur->win.ekey = ascScan ? tsArray[end] : tsArray[start];
dengyihao's avatar
dengyihao 已提交
2175
        cur->lastKey = cur->win.ekey + step;
H
Haojun Liao 已提交
2176
        cur->mixBlock = true;
2177
      }
2178 2179
    }
  }
H
Haojun Liao 已提交
2180

2181 2182
  cur->blockCompleted = (((pos > endPos || cur->lastKey > pTsdbReadHandle->window.ekey) && ascScan) ||
       ((pos < endPos || cur->lastKey < pTsdbReadHandle->window.ekey) && !ascScan));
2183

2184
  if (!ascScan) {
wafwerar's avatar
wafwerar 已提交
2185
    TSWAP(cur->win.skey, cur->win.ekey);
2186
  }
2187

2188 2189
  updateInfoAfterMerge(pTsdbReadHandle, pCheckInfo, numOfRows, pos);
  doCheckGeneratedBlockRange(pTsdbReadHandle);
H
Haojun Liao 已提交
2190

dengyihao's avatar
dengyihao 已提交
2191 2192 2193
  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);
2194 2195
}

2196
int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order) {
H
[td-32]  
hjxilinx 已提交
2197
  int    firstPos, lastPos, midPos = -1;
H
Haojun Liao 已提交
2198
  int    numOfRows;
2199 2200
  TSKEY* keyList;

H
[td-32]  
hjxilinx 已提交
2201
  if (num <= 0) return -1;
2202 2203

  keyList = (TSKEY*)pValue;
H
[td-32]  
hjxilinx 已提交
2204 2205
  firstPos = 0;
  lastPos = num - 1;
2206

2207
  if (order == TSDB_ORDER_DESC) {
H
[td-32]  
hjxilinx 已提交
2208 2209 2210 2211 2212
    // 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;
2213

H
Haojun Liao 已提交
2214 2215
      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1) + firstPos;
2216

H
[td-32]  
hjxilinx 已提交
2217 2218 2219 2220 2221 2222 2223 2224
      if (key < keyList[midPos]) {
        lastPos = midPos - 1;
      } else if (key > keyList[midPos]) {
        firstPos = midPos + 1;
      } else {
        break;
      }
    }
2225

H
[td-32]  
hjxilinx 已提交
2226 2227 2228 2229 2230
  } 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;
2231

H
[td-32]  
hjxilinx 已提交
2232 2233 2234 2235 2236 2237 2238
      if (key > keyList[lastPos]) {
        lastPos = lastPos + 1;
        if (lastPos >= num)
          return -1;
        else
          return lastPos;
      }
2239

H
Haojun Liao 已提交
2240 2241
      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1) + firstPos;
2242

H
[td-32]  
hjxilinx 已提交
2243 2244 2245 2246 2247 2248 2249 2250 2251
      if (key < keyList[midPos]) {
        lastPos = midPos - 1;
      } else if (key > keyList[midPos]) {
        firstPos = midPos + 1;
      } else {
        break;
      }
    }
  }
2252

H
[td-32]  
hjxilinx 已提交
2253 2254 2255
  return midPos;
}

2256
static void cleanBlockOrderSupporter(SBlockOrderSupporter* pSupporter, int32_t numOfTables) {
wafwerar's avatar
wafwerar 已提交
2257 2258
  taosMemoryFreeClear(pSupporter->numOfBlocksPerTable);
  taosMemoryFreeClear(pSupporter->blockIndexArray);
2259 2260

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

wafwerar's avatar
wafwerar 已提交
2265
  taosMemoryFreeClear(pSupporter->pDataBlockInfo);
2266 2267 2268 2269 2270 2271 2272 2273 2274 2275 2276
}

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

2277
  if (leftTableBlockIndex > pSupporter->numOfBlocksPerTable[leftTableIndex]) {
2278 2279
    /* left block is empty */
    return 1;
2280
  } else if (rightTableBlockIndex > pSupporter->numOfBlocksPerTable[rightTableIndex]) {
2281 2282 2283 2284 2285 2286 2287
    /* right block is empty */
    return -1;
  }

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

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

H
Haojun Liao 已提交
2296
  return pLeftBlockInfoEx->compBlock->offset > pRightBlockInfoEx->compBlock->offset ? 1 : -1;
2297 2298
}

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

2302 2303
  if (pTsdbReadHandle->allocSize < size) {
    pTsdbReadHandle->allocSize = (int32_t)size;
wafwerar's avatar
wafwerar 已提交
2304
    char* tmp = taosMemoryRealloc(pTsdbReadHandle->pDataBlockInfo, pTsdbReadHandle->allocSize);
H
Haojun Liao 已提交
2305 2306 2307 2308
    if (tmp == NULL) {
      return TSDB_CODE_TDB_OUT_OF_MEMORY;
    }

dengyihao's avatar
dengyihao 已提交
2309
    pTsdbReadHandle->pDataBlockInfo = (STableBlockInfo*)tmp;
2310 2311
  }

2312
  memset(pTsdbReadHandle->pDataBlockInfo, 0, size);
2313 2314
  *numOfAllocBlocks = numOfBlocks;

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

2318 2319
  SBlockOrderSupporter sup = {0};
  sup.numOfTables = numOfTables;
wafwerar's avatar
wafwerar 已提交
2320 2321 2322
  sup.numOfBlocksPerTable = taosMemoryCalloc(1, sizeof(int32_t) * numOfTables);
  sup.blockIndexArray = taosMemoryCalloc(1, sizeof(int32_t) * numOfTables);
  sup.pDataBlockInfo = taosMemoryCalloc(1, POINTER_BYTES * numOfTables);
2323

2324
  if (sup.numOfBlocksPerTable == NULL || sup.blockIndexArray == NULL || sup.pDataBlockInfo == NULL) {
2325
    cleanBlockOrderSupporter(&sup, 0);
2326
    return TSDB_CODE_TDB_OUT_OF_MEMORY;
2327
  }
H
Haojun Liao 已提交
2328

2329
  int32_t cnt = 0;
2330
  int32_t numOfQualTables = 0;
H
Haojun Liao 已提交
2331

2332
  for (int32_t j = 0; j < numOfTables; ++j) {
2333
    STableCheckInfo* pTableCheck = (STableCheckInfo*)taosArrayGet(pTsdbReadHandle->pTableCheckInfo, j);
2334 2335 2336
    if (pTableCheck->numOfBlocks <= 0) {
      continue;
    }
H
Haojun Liao 已提交
2337

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

wafwerar's avatar
wafwerar 已提交
2341
    char* buf = taosMemoryMalloc(sizeof(STableBlockInfo) * pTableCheck->numOfBlocks);
2342
    if (buf == NULL) {
2343
      cleanBlockOrderSupporter(&sup, numOfQualTables);
2344
      return TSDB_CODE_TDB_OUT_OF_MEMORY;
2345 2346
    }

2347
    sup.pDataBlockInfo[numOfQualTables] = (STableBlockInfo*)buf;
2348 2349

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

H
Haojun Liao 已提交
2352 2353
      pBlockInfo->compBlock = &pBlock[k];
      pBlockInfo->pTableCheckInfo = pTableCheck;
2354 2355 2356
      cnt++;
    }

2357
    numOfQualTables++;
2358 2359
  }

H
Haojun Liao 已提交
2360
  assert(numOfBlocks == cnt);
2361

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

H
Haojun Liao 已提交
2367
    tsdbDebug("%p create data blocks info struct completed for 1 table, %d blocks not sorted %s", pTsdbReadHandle, cnt,
dengyihao's avatar
dengyihao 已提交
2368
              pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
2369 2370
    return TSDB_CODE_SUCCESS;
  }
2371

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

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

2378
  SMultiwayMergeTreeInfo* pTree = NULL;
dengyihao's avatar
dengyihao 已提交
2379
  uint8_t                 ret = tMergeTreeCreate(&pTree, sup.numOfTables, &sup, dataBlockOrderCompar);
2380 2381
  if (ret != TSDB_CODE_SUCCESS) {
    cleanBlockOrderSupporter(&sup, numOfTables);
2382
    return TSDB_CODE_TDB_OUT_OF_MEMORY;
2383 2384 2385 2386 2387
  }

  int32_t numOfTotal = 0;

  while (numOfTotal < cnt) {
2388
    int32_t pos = tMergeTreeGetChosenIndex(pTree);
2389 2390
    int32_t index = sup.blockIndexArray[pos]++;

H
Haojun Liao 已提交
2391
    STableBlockInfo* pBlocksInfo = sup.pDataBlockInfo[pos];
2392
    pTsdbReadHandle->pDataBlockInfo[numOfTotal++] = pBlocksInfo[index];
2393 2394

    // set data block index overflow, in order to disable the offset comparator
2395 2396
    if (sup.blockIndexArray[pos] >= sup.numOfBlocksPerTable[pos]) {
      sup.blockIndexArray[pos] = sup.numOfBlocksPerTable[pos] + 1;
2397
    }
2398

H
Haojun Liao 已提交
2399
    tMergeTreeAdjust(pTree, tMergeTreeGetAdjustIndex(pTree));
2400 2401 2402 2403 2404
  }

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

H
Haojun Liao 已提交
2409
  tsdbDebug("%p %d data blocks sort completed, %s", pTsdbReadHandle, cnt, pTsdbReadHandle->idStr);
2410
  cleanBlockOrderSupporter(&sup, numOfTables);
wafwerar's avatar
wafwerar 已提交
2411
  taosMemoryFree(pTree);
2412 2413 2414 2415

  return TSDB_CODE_SUCCESS;
}

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

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

dengyihao's avatar
dengyihao 已提交
2422
  while (1) {
2423
    int32_t code = loadFileDataBlock(pTsdbReadHandle, pNext->compBlock, pNext->pTableCheckInfo, exists);
H
Haojun Liao 已提交
2424 2425 2426 2427
    if (code != TSDB_CODE_SUCCESS || *exists) {
      return code;
    }

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

2441 2442 2443
static int32_t getFirstFileDataBlock(STsdbReadHandle* pTsdbReadHandle, bool* exists) {
  pTsdbReadHandle->numOfBlocks = 0;
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
H
Haojun Liao 已提交
2444 2445 2446

  int32_t code = TSDB_CODE_SUCCESS;

2447
  int32_t numOfBlocks = 0;
2448
  int32_t numOfTables = (int32_t)taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
2449

C
Cary Xu 已提交
2450
  STsdbKeepCfg* pCfg = REPO_KEEP_CFG(pTsdbReadHandle->pTsdb);
C
Cary Xu 已提交
2451
  STimeWindow   win = TSWINDOW_INITIALIZER;
2452

H
Hongze Cheng 已提交
2453
  while (true) {
2454
    tsdbRLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2455

2456 2457
    if ((pTsdbReadHandle->pFileGroup = tsdbFSIterNext(&pTsdbReadHandle->fileIter)) == NULL) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2458 2459 2460
      break;
    }

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

    // current file are not overlapped with query time window, ignore remain files
2464 2465 2466
    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 已提交
2467 2468
      tsdbDebug("%p remain files are not qualified for qrange:%" PRId64 "-%" PRId64 ", ignore, %s", pTsdbReadHandle,
                pTsdbReadHandle->window.skey, pTsdbReadHandle->window.ekey, pTsdbReadHandle->idStr);
2469 2470
      pTsdbReadHandle->pFileGroup = NULL;
      assert(pTsdbReadHandle->numOfBlocks == 0);
2471 2472 2473
      break;
    }

2474 2475
    if (tsdbSetAndOpenReadFSet(&pTsdbReadHandle->rhelper, pTsdbReadHandle->pFileGroup) < 0) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2476 2477 2478 2479
      code = terrno;
      break;
    }

2480
    tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2481

2482
    if (tsdbLoadBlockIdx(&pTsdbReadHandle->rhelper) < 0) {
H
Hongze Cheng 已提交
2483 2484 2485 2486
      code = terrno;
      break;
    }

2487
    if ((code = getFileCompInfo(pTsdbReadHandle, &numOfBlocks)) != TSDB_CODE_SUCCESS) {
2488 2489
      break;
    }
H
Haojun Liao 已提交
2490

H
Haojun Liao 已提交
2491 2492
    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 已提交
2493

2494 2495 2496 2497
    assert(numOfBlocks >= 0);
    if (numOfBlocks == 0) {
      continue;
    }
H
Haojun Liao 已提交
2498

2499
    // todo return error code to query engine
dengyihao's avatar
dengyihao 已提交
2500 2501
    if ((code = createDataBlocksInfo(pTsdbReadHandle, numOfBlocks, &pTsdbReadHandle->numOfBlocks)) !=
        TSDB_CODE_SUCCESS) {
2502 2503
      break;
    }
H
Haojun Liao 已提交
2504

2505 2506
    assert(numOfBlocks >= pTsdbReadHandle->numOfBlocks);
    if (pTsdbReadHandle->numOfBlocks > 0) {
2507 2508 2509
      break;
    }
  }
H
Haojun Liao 已提交
2510

2511
  // no data in file anymore
2512
  if (pTsdbReadHandle->numOfBlocks <= 0 || code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2513
    if (code == TSDB_CODE_SUCCESS) {
2514
      assert(pTsdbReadHandle->pFileGroup == NULL);
H
Haojun Liao 已提交
2515 2516
    }

D
dapan1121 已提交
2517
    cur->fid = INT32_MIN;  // denote that there are no data in file anymore
H
Haojun Liao 已提交
2518 2519
    *exists = false;
    return code;
2520
  }
H
Haojun Liao 已提交
2521

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

2526
  STableBlockInfo* pBlockInfo = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
2527
  return getDataBlock(pTsdbReadHandle, pBlockInfo, exists);
H
Haojun Liao 已提交
2528 2529 2530 2531 2532 2533 2534
}

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

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

2538 2539
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
  assert(cur->slot < pTsdbReadHandle->numOfBlocks && cur->slot >= 0);
H
Haojun Liao 已提交
2540 2541

  cur->slot += step;
dengyihao's avatar
dengyihao 已提交
2542
  cur->mixBlock = false;
H
Haojun Liao 已提交
2543
  cur->blockCompleted = false;
2544
}
H
Haojun Liao 已提交
2545 2546

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

H
Haojun Liao 已提交
2549
  pTableBlockInfo->totalSize = 0;
2550
  pTableBlockInfo->totalRows = 0;
H
Haojun Liao 已提交
2551

2552
  STsdbFS* pFileHandle = REPO_FS(pTsdbReadHandle->pTsdb);
H
Haojun Liao 已提交
2553 2554

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

  tsdbRLockFS(pFileHandle);
2560 2561
  tsdbFSIterInit(&pTsdbReadHandle->fileIter, pFileHandle, pTsdbReadHandle->order);
  tsdbFSIterSeek(&pTsdbReadHandle->fileIter, fid);
H
Haojun Liao 已提交
2562 2563
  tsdbUnLockFS(pFileHandle);

H
Haojun Liao 已提交
2564
  pTableBlockInfo->numOfFiles += 1;
H
Haojun Liao 已提交
2565

H
Haojun Liao 已提交
2566
  int32_t     code = TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
2567
  int32_t     numOfBlocks = 0;
2568
  int32_t     numOfTables = (int32_t)taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
dengyihao's avatar
dengyihao 已提交
2569
  int         defaultRows = 4096;  // TSDB_DEFAULT_BLOCK_ROWS(pCfg->maxRowsPerFileBlock);
H
Haojun Liao 已提交
2570 2571
  STimeWindow win = TSWINDOW_INITIALIZER;

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

H
Haojun Liao 已提交
2574 2575
  while (true) {
    numOfBlocks = 0;
2576
    tsdbRLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2577

2578 2579
    if ((pTsdbReadHandle->pFileGroup = tsdbFSIterNext(&pTsdbReadHandle->fileIter)) == NULL) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2580 2581 2582
      break;
    }

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

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

H
Haojun Liao 已提交
2595
    pTableBlockInfo->numOfFiles += 1;
2596 2597
    if (tsdbSetAndOpenReadFSet(&pTsdbReadHandle->rhelper, pTsdbReadHandle->pFileGroup) < 0) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2598 2599 2600 2601
      code = terrno;
      break;
    }

2602
    tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2603

2604
    if (tsdbLoadBlockIdx(&pTsdbReadHandle->rhelper) < 0) {
H
Haojun Liao 已提交
2605 2606 2607 2608
      code = terrno;
      break;
    }

2609
    if ((code = getFileCompInfo(pTsdbReadHandle, &numOfBlocks)) != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2610 2611 2612
      break;
    }

H
Haojun Liao 已提交
2613 2614
    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 已提交
2615 2616 2617 2618 2619 2620

    if (numOfBlocks == 0) {
      continue;
    }

    for (int32_t i = 0; i < numOfTables; ++i) {
2621
      STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, i);
H
Haojun Liao 已提交
2622 2623 2624

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

H
Haojun Liao 已提交
2627
        int32_t numOfRows = pBlock[j].numOfRows;
2628
        pTableBlockInfo->totalRows += numOfRows;
H
Haojun Liao 已提交
2629 2630 2631 2632 2633 2634 2635 2636 2637 2638 2639
        if (numOfRows > pTableBlockInfo->maxRows) {
          pTableBlockInfo->maxRows = numOfRows;
        }

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

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

  return code;
}

2650 2651 2652
static int32_t getDataBlocksInFiles(STsdbReadHandle* pTsdbReadHandle, bool* exists) {
  STsdbFS*       pFileHandle = REPO_FS(pTsdbReadHandle->pTsdb);
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
2653 2654

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

H
Hongze Cheng 已提交
2660
    tsdbRLockFS(pFileHandle);
2661 2662
    tsdbFSIterInit(&pTsdbReadHandle->fileIter, pFileHandle, pTsdbReadHandle->order);
    tsdbFSIterSeek(&pTsdbReadHandle->fileIter, fid);
H
Hongze Cheng 已提交
2663
    tsdbUnLockFS(pFileHandle);
2664

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

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

H
Haojun Liao 已提交
2680 2681 2682 2683 2684 2685 2686
      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
2687 2688
    if (isEndFileDataBlock(cur, pTsdbReadHandle->numOfBlocks, ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
      return getFirstFileDataBlock(pTsdbReadHandle, exists);
H
Haojun Liao 已提交
2689
    } else {
2690 2691
      moveToNextDataBlockInCurrentFile(pTsdbReadHandle);
      STableBlockInfo* pNext = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
2692
      return getDataBlock(pTsdbReadHandle, pNext, exists);
2693 2694
    }
  }
2695 2696
}

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

2700 2701
  while (pTsdbReadHandle->activeIndex < numOfTables) {
    if (hasMoreDataInCache(pTsdbReadHandle)) {
2702 2703
      return true;
    }
H
Haojun Liao 已提交
2704

2705
    pTsdbReadHandle->activeIndex += 1;
2706
  }
H
Haojun Liao 已提交
2707

2708 2709 2710
  return false;
}

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

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

  int32_t i = 0;
dengyihao's avatar
dengyihao 已提交
2720
  while (i < numOfTables) {
2721
    STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, i);
H
Haojun Liao 已提交
2722 2723

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

    i++;
  }

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

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

2740 2741
  info.lastKey = pTsdbReadHandle->window.skey;
  taosArrayPush(pTsdbReadHandle->pTableCheckInfo, &info);
H
Haojun Liao 已提交
2742 2743 2744
}

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

dengyihao's avatar
dengyihao 已提交
2752 2753
  int64_t   st = taosGetTimestampUs();
  int16_t   rv = -1;
D
fix bug  
dapan1121 已提交
2754
  STSchema* pSchema = NULL;
C
Cary Xu 已提交
2755 2756
  TSKEY     lastRowKey = TSKEY_INITIAL_VAL;

H
Haojun Liao 已提交
2757
  do {
C
Cary Xu 已提交
2758
    STSRow* row = getSRowInTableMem(pCheckInfo, pTsdbReadHandle->order, pCfg->update, NULL, TD_VER_MAX);
H
Haojun Liao 已提交
2759 2760 2761 2762
    if (row == NULL) {
      break;
    }

H
Haojun Liao 已提交
2763
    TSKEY key = TD_ROW_KEY(row);
dengyihao's avatar
dengyihao 已提交
2764 2765 2766 2767
    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 已提交
2768 2769 2770 2771 2772 2773 2774 2775 2776

      break;
    }

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

    win->ekey = key;
H
Haojun Liao 已提交
2777
    if (rv != TD_ROW_SVER(row)) {
H
Hongze Cheng 已提交
2778
      pSchema = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), pCheckInfo->tableId, 1);
H
Haojun Liao 已提交
2779
      rv = TD_ROW_SVER(row);
D
fix bug  
dapan1121 已提交
2780
    }
C
Cary Xu 已提交
2781 2782
    numOfRows += mergeTwoRowFromMem(pTsdbReadHandle, maxRowsToRead, &curRows, row, NULL, numOfCols, pCheckInfo->tableId,
                                    pSchema, NULL, pCfg->update, &lastRowKey);
H
Haojun Liao 已提交
2783

C
Cary Xu 已提交
2784
    if (numOfRows >= maxRowsToRead) {
H
Haojun Liao 已提交
2785 2786 2787 2788
      moveToNextRowInMem(pCheckInfo);
      break;
    }

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

C
Cary Xu 已提交
2791
  taosMemoryFreeClear(pSchema);  // free the STSChema
H
Haojun Liao 已提交
2792 2793 2794
  assert(numOfRows <= maxRowsToRead);

  int64_t elapsedTime = taosGetTimestampUs() - st;
dengyihao's avatar
dengyihao 已提交
2795 2796
  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 已提交
2797 2798 2799 2800

  return numOfRows;
}

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

2804 2805 2806 2807 2808
  while (1) {
    tb_uid_t id = metaCtbCursorNext(pCur);
    if (id == 0) {
      break;
    }
H
Haojun Liao 已提交
2809

2810
    STableKeyInfo info = {.lastKey = TSKEY_INITIAL_VAL, uid = id};
H
Haojun Liao 已提交
2811 2812 2813
    taosArrayPush(list, &info);
  }

C
Cary Xu 已提交
2814
  metaCloseCtbCursor(pCur);
H
Haojun Liao 已提交
2815 2816 2817 2818 2819 2820 2821 2822
  return TSDB_CODE_SUCCESS;
}

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

dengyihao's avatar
dengyihao 已提交
2823 2824 2825 2826 2827 2828
  //  tQueryInfo* pInfo = (tQueryInfo*)param;
  //  if (pInfo->optr != TSDB_RELATION_IN) {
  //    taosMemoryFreeClear(pInfo->q);
  //  } else {
  //    taosHashCleanup((SHashObj *)(pInfo->q));
  //  }
H
Haojun Liao 已提交
2829

wafwerar's avatar
wafwerar 已提交
2830
  taosMemoryFree(param);
H
Haojun Liao 已提交
2831 2832
}

dengyihao's avatar
dengyihao 已提交
2833 2834
#define TSDB_PREV_ROW 0x1
#define TSDB_NEXT_ROW 0x2
2835

dengyihao's avatar
dengyihao 已提交
2836
static bool loadBlockOfActiveTable(STsdbReadHandle* pTsdbReadHandle) {
2837
  if (pTsdbReadHandle->checkFiles) {
H
Haojun Liao 已提交
2838 2839
    // check if the query range overlaps with the file data block
    bool exists = true;
H
Haojun Liao 已提交
2840

2841
    int32_t code = getDataBlocksInFiles(pTsdbReadHandle, &exists);
H
Haojun Liao 已提交
2842
    if (code != TSDB_CODE_SUCCESS) {
2843
      pTsdbReadHandle->checkFiles = false;
H
Haojun Liao 已提交
2844 2845
      return false;
    }
H
Haojun Liao 已提交
2846

H
Haojun Liao 已提交
2847
    if (exists) {
dengyihao's avatar
dengyihao 已提交
2848
      tsdbRetrieveDataBlock((tsdbReaderT*)pTsdbReadHandle, NULL);
2849 2850 2851
      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 已提交
2852 2853
      }

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

2858
    pTsdbReadHandle->checkFiles = false;
H
Haojun Liao 已提交
2859
  }
H
Haojun Liao 已提交
2860

2861 2862
  if (hasMoreDataInCache(pTsdbReadHandle)) {
    pTsdbReadHandle->currentLoadExternalRows = false;
H
Haojun Liao 已提交
2863 2864
    return true;
  }
H
Haojun Liao 已提交
2865

H
Haojun Liao 已提交
2866
  // current result is empty
dengyihao's avatar
dengyihao 已提交
2867 2868 2869
  if (pTsdbReadHandle->currentLoadExternalRows && pTsdbReadHandle->window.skey == pTsdbReadHandle->window.ekey &&
      pTsdbReadHandle->cur.rows == 0) {
    //    STsdbMemTable* pMemRef = pTsdbReadHandle->pMemTable;
H
Haojun Liao 已提交
2870

dengyihao's avatar
dengyihao 已提交
2871 2872
    //    doGetExternalRow(pTsdbReadHandle, TSDB_PREV_ROW, pMemRef);
    //    doGetExternalRow(pTsdbReadHandle, TSDB_NEXT_ROW, pMemRef);
H
Haojun Liao 已提交
2873

2874
    bool result = tsdbGetExternalRow(pTsdbReadHandle);
H
Haojun Liao 已提交
2875

dengyihao's avatar
dengyihao 已提交
2876 2877
    //    pTsdbReadHandle->prev = doFreeColumnInfoData(pTsdbReadHandle->prev);
    //    pTsdbReadHandle->next = doFreeColumnInfoData(pTsdbReadHandle->next);
2878
    pTsdbReadHandle->currentLoadExternalRows = false;
H
Haojun Liao 已提交
2879 2880

    return result;
2881
  }
H
Haojun Liao 已提交
2882

H
Haojun Liao 已提交
2883 2884
  return false;
}
2885

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

2893
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
2894

dengyihao's avatar
dengyihao 已提交
2895 2896 2897
  STSRow* pRow = NULL;
  TSKEY   key = TSKEY_INITIAL_VAL;
  int32_t step = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 1 : -1;
C
Cary Xu 已提交
2898 2899
  TSKEY   lastRowKey = TSKEY_INITIAL_VAL;
  int32_t curRow = 0;
2900 2901 2902

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

H
Haojun Liao 已提交
2911 2912 2913
    // update the last key value
    pCheckInfo->lastKey = key + step;

dengyihao's avatar
dengyihao 已提交
2914 2915
    cur->rows = 1;  // only one row
    cur->lastKey = key + step;
H
Haojun Liao 已提交
2916 2917 2918 2919 2920
    cur->mixBlock = true;
    cur->win.skey = key;
    cur->win.ekey = key;

    return true;
2921
  }
H
Haojun Liao 已提交
2922

H
Haojun Liao 已提交
2923 2924 2925
  return false;
}

dengyihao's avatar
dengyihao 已提交
2926
// static bool loadCachedLast(STsdbReadHandle* pTsdbReadHandle) {
2927 2928 2929 2930 2931 2932 2933 2934 2935 2936 2937 2938 2939 2940 2941 2942 2943 2944 2945
//  // 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 已提交
2946 2947
//      tsdbWarn("no last cached for table %s, uid:%" PRIu64 ",tid:%d", pTable->name->data, pTable->uid,
//      pTable->tableId); continue;
2948 2949 2950 2951 2952 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
//    }
//
//    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 已提交
3078 3079 3080
  assert(numOfTables > 0);

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

dengyihao's avatar
dengyihao 已提交
3082
  while (pTsdbReadHandle->activeIndex < numOfTables) {
3083
    if (loadBlockOfActiveTable(pTsdbReadHandle)) {
H
Haojun Liao 已提交
3084 3085 3086
      return true;
    }

3087
    STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, pTsdbReadHandle->activeIndex);
H
Haojun Liao 已提交
3088 3089
    pCheckInfo->numOfBlocks = 0;

3090 3091
    pTsdbReadHandle->activeIndex += 1;
    pTsdbReadHandle->locateStart = false;
dengyihao's avatar
dengyihao 已提交
3092 3093
    pTsdbReadHandle->checkFiles = true;
    pTsdbReadHandle->cur.rows = 0;
3094
    pTsdbReadHandle->currentLoadExternalRows = pTsdbReadHandle->loadExternalRow;
H
Haojun Liao 已提交
3095 3096 3097 3098

    terrno = TSDB_CODE_SUCCESS;

    int64_t elapsedTime = taosGetTimestampUs() - stime;
3099
    pTsdbReadHandle->cost.checkForNextTime += elapsedTime;
H
Haojun Liao 已提交
3100 3101 3102
  }

  return false;
3103 3104
}

H
Haojun Liao 已提交
3105
// handle data in cache situation
H
Haojun Liao 已提交
3106
bool tsdbNextDataBlock(tsdbReaderT pHandle) {
dengyihao's avatar
dengyihao 已提交
3107
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle;
Y
yihaoDeng 已提交
3108

3109 3110
  size_t numOfCols = taosArrayGetSize(pTsdbReadHandle->pColumns);
  for (int32_t i = 0; i < numOfCols; ++i) {
3111 3112 3113 3114
    SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
    colInfoDataCleanup(pColInfo, pTsdbReadHandle->outputCapacity);
  }

3115
  if (emptyQueryTimewindow(pTsdbReadHandle)) {
dengyihao's avatar
dengyihao 已提交
3116 3117
    tsdbDebug("%p query window not overlaps with the data set, no result returned, %s", pTsdbReadHandle,
              pTsdbReadHandle->idStr);
3118 3119 3120
    return false;
  }

Y
yihaoDeng 已提交
3121 3122 3123
  int64_t stime = taosGetTimestampUs();
  int64_t elapsedTime = stime;

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

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

3140
      int32_t code = getDataBlocksInFiles(pTsdbReadHandle, &exists);
H
Haojun Liao 已提交
3141
      if (code != TSDB_CODE_SUCCESS) {
3142 3143
        pTsdbReadHandle->activeIndex = 0;
        pTsdbReadHandle->checkFiles = false;
H
Haojun Liao 已提交
3144 3145 3146 3147 3148

        return false;
      }

      if (exists) {
3149
        pTsdbReadHandle->cost.checkForNextTime += (taosGetTimestampUs() - stime);
H
Haojun Liao 已提交
3150 3151
        return exists;
      }
Y
yihaoDeng 已提交
3152

3153 3154
      pTsdbReadHandle->activeIndex = 0;
      pTsdbReadHandle->checkFiles = false;
Y
yihaoDeng 已提交
3155 3156
    }

H
Haojun Liao 已提交
3157
    // TODO: opt by consider the scan order
3158
    bool ret = doHasDataInBuffer(pTsdbReadHandle);
H
Haojun Liao 已提交
3159
    terrno = TSDB_CODE_SUCCESS;
Y
yihaoDeng 已提交
3160

H
Haojun Liao 已提交
3161
    elapsedTime = taosGetTimestampUs() - stime;
3162
    pTsdbReadHandle->cost.checkForNextTime += elapsedTime;
H
Haojun Liao 已提交
3163
    return ret;
Y
yihaoDeng 已提交
3164 3165
  }
}
3166

dengyihao's avatar
dengyihao 已提交
3167
// static int32_t doGetExternalRow(STsdbReadHandle* pTsdbReadHandle, int16_t type, STsdbMemTable* pMemRef) {
3168 3169 3170 3171 3172 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
//  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 已提交
3202
//    colInfo.pData = taosMemoryCalloc(1, pCol->info.bytes);
3203 3204 3205 3206 3207 3208 3209 3210 3211
//    if (colInfo.pData == NULL) {
//      terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
//      goto out_of_memory;
//    }
//
//    taosArrayPush(row, &colInfo);
//  }
//
//  // load the previous row
3212
//  SQueryTableDataCond cond = {.numOfCols = numOfCols, .loadExternalRows = false, .type = BLOCK_LOAD_OFFSET_SEQ_ORDER};
3213 3214 3215 3216 3217 3218 3219 3220
//  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 已提交
3221
//  cond.colList = taosMemoryCalloc(cond.numOfCols, sizeof(SColumnInfo));
3222 3223 3224 3225 3226 3227 3228 3229 3230 3231
//  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 已提交
3232
//  pSecQueryHandle = tsdbQueryTablesImpl(pTsdbReadHandle->pTsdb, &cond, pTsdbReadHandle->idStr, pMemRef);
wafwerar's avatar
wafwerar 已提交
3233
//  taosMemoryFreeClear(cond.colList);
3234 3235 3236 3237 3238 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
//
//  // 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 已提交
3272
// out_of_memory:
3273
//  tsdbCleanupReadHandle(pSecQueryHandle);
3274 3275 3276
//  return terrno;
//}

H
Haojun Liao 已提交
3277
bool tsdbGetExternalRow(tsdbReaderT pHandle) {
dengyihao's avatar
dengyihao 已提交
3278 3279
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle;
  SQueryFilePos*   cur = &pTsdbReadHandle->cur;
H
Haojun Liao 已提交
3280

H
Haojun Liao 已提交
3281 3282
  cur->fid = INT32_MIN;
  cur->mixBlock = true;
3283
  if (pTsdbReadHandle->prev == NULL || pTsdbReadHandle->next == NULL) {
H
Haojun Liao 已提交
3284 3285
    cur->rows = 0;
    return false;
H
Haojun Liao 已提交
3286 3287
  }

dengyihao's avatar
dengyihao 已提交
3288
  int32_t numOfCols = (int32_t)QH_GET_NUM_OF_COLS(pTsdbReadHandle);
H
Haojun Liao 已提交
3289
  for (int32_t i = 0; i < numOfCols; ++i) {
3290 3291
    SColumnInfoData* pColInfoData = taosArrayGet(pTsdbReadHandle->pColumns, i);
    SColumnInfoData* first = taosArrayGet(pTsdbReadHandle->prev, i);
H
Haojun Liao 已提交
3292 3293 3294

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

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

    if (i == 0 && pColInfoData->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
H
Haojun Liao 已提交
3299
      cur->win.skey = *(TSKEY*)pColInfoData->pData;
sangshuduo's avatar
sangshuduo 已提交
3300
      cur->win.ekey = *(TSKEY*)(((char*)pColInfoData->pData) + TSDB_KEYSIZE);
H
Haojun Liao 已提交
3301 3302 3303
    }
  }

H
Haojun Liao 已提交
3304 3305
  cur->rows = 2;
  return true;
3306 3307
}

3308
/*
3309
 * if lastRow == NULL, return TSDB_CODE_TDB_NO_CACHE_LAST_ROW
3310
 * else set pRes and return TSDB_CODE_SUCCESS and save lastKey
3311
 */
H
Haojun Liao 已提交
3312
// int32_t tsdbGetCachedLastRow(STable* pTable, STSRow** pRes, TSKEY* lastKey) {
3313 3314 3315 3316 3317 3318 3319 3320 3321 3322 3323 3324 3325 3326 3327 3328
//  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 已提交
3329
// out:
3330 3331 3332 3333
//  TSDB_RUNLOCK_TABLE(pTable);
//  return code;
//}

3334
bool isTsdbCacheLastRow(tsdbReaderT* pReader) {
dengyihao's avatar
dengyihao 已提交
3335
  return ((STsdbReadHandle*)pReader)->cachelastrow > TSDB_CACHED_TYPE_NONE;
D
fix bug  
dapan1121 已提交
3336 3337
}

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

dengyihao's avatar
dengyihao 已提交
3341 3342 3343 3344 3345 3346 3347 3348 3349 3350 3351 3352 3353 3354 3355 3356 3357 3358 3359 3360 3361 3362 3363 3364
  //  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 已提交
3365

3366
  return TSDB_CODE_SUCCESS;
3367 3368
}

3369 3370
int32_t checkForCachedLast(STsdbReadHandle* pTsdbReadHandle) {
  assert(pTsdbReadHandle != NULL);
D
update  
dapan1121 已提交
3371 3372

  int32_t code = 0;
dengyihao's avatar
dengyihao 已提交
3373 3374 3375
  //  if (pTsdbReadHandle->pTsdb && atomic_load_8(&pTsdbReadHandle->pTsdb->hasCachedLastColumn)){
  //    pTsdbReadHandle->cachelastrow = TSDB_CACHED_TYPE_LAST;
  //  }
D
update  
dapan1121 已提交
3376 3377

  // update the tsdb query time range
3378
  if (pTsdbReadHandle->cachelastrow) {
dengyihao's avatar
dengyihao 已提交
3379
    pTsdbReadHandle->checkFiles = false;
3380
    pTsdbReadHandle->activeIndex = -1;  // start from -1
D
update  
dapan1121 已提交
3381 3382 3383 3384 3385
  }

  return code;
}

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

H
Haojun Liao 已提交
3389
  int32_t totalNumOfTable = 0;
3390
  SArray* emptyGroup = taosArrayInit(16, sizeof(int32_t));
H
Haojun Liao 已提交
3391

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

H
Haojun Liao 已提交
3398
    STableKeyInfo keyInfo = {0};
H
Haojun Liao 已提交
3399

H
Haojun Liao 已提交
3400
    size_t numOfTables = taosArrayGetSize(pGroup);
dengyihao's avatar
dengyihao 已提交
3401 3402
    for (int32_t i = 0; i < numOfTables; ++i) {
      STableKeyInfo* pInfo = (STableKeyInfo*)taosArrayGet(pGroup, i);
H
Haojun Liao 已提交
3403

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

dengyihao's avatar
dengyihao 已提交
3409
        //        keyInfo.pTable  = pInfo->pTable;
H
Haojun Liao 已提交
3410
        keyInfo.lastKey = key;
dengyihao's avatar
dengyihao 已提交
3411
        pInfo->lastKey = key;
H
Haojun Liao 已提交
3412

H
Haojun Liao 已提交
3413 3414 3415
        if (key < window.skey) {
          window.skey = key;
        }
3416

H
Haojun Liao 已提交
3417 3418 3419 3420
        if (key > window.ekey) {
          window.ekey = key;
        }
      }
3421
    }
H
Haojun Liao 已提交
3422

H
Haojun Liao 已提交
3423
    // more than one table in each group, only one table left for each group
dengyihao's avatar
dengyihao 已提交
3424 3425 3426 3427 3428 3429 3430 3431 3432 3433 3434 3435
    //    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);
    //    }
3436
  }
H
Haojun Liao 已提交
3437

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

dengyihao's avatar
dengyihao 已提交
3444
  taosArrayRemoveBatch(groupList->pGroupList, TARRAY_GET_START(emptyGroup), (int32_t)taosArrayGetSize(emptyGroup));
3445 3446
  taosArrayDestroy(emptyGroup);

H
Haojun Liao 已提交
3447
  groupList->numOfTables = totalNumOfTable;
H
Haojun Liao 已提交
3448
  return window;
H
hjxilinx 已提交
3449 3450
}

H
Haojun Liao 已提交
3451
void tsdbRetrieveDataBlockInfo(tsdbReaderT* pTsdbReadHandle, SDataBlockInfo* pDataBlockInfo) {
3452
  STsdbReadHandle* pHandle = (STsdbReadHandle*)pTsdbReadHandle;
dengyihao's avatar
dengyihao 已提交
3453
  SQueryFilePos*   cur = &pHandle->cur;
3454 3455

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

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

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

3469
  pDataBlockInfo->uid = uid;
3470 3471 3472 3473 3474 3475

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

dengyihao's avatar
dengyihao 已提交
3476
  pDataBlockInfo->rows = cur->rows;
H
Haojun Liao 已提交
3477
  pDataBlockInfo->window = cur->win;
C
Cary Xu 已提交
3478
  //  ASSERT(pDataBlockInfo->numOfCols >= (int32_t)(QH_GET_NUM_OF_COLS(pHandle));
3479
}
H
hjxilinx 已提交
3480

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

H
Haojun Liao 已提交
3488 3489
  SQueryFilePos* c = &pHandle->cur;
  if (c->mixBlock) {
H
Haojun Liao 已提交
3490 3491 3492
    *pBlockStatis = NULL;
    return TSDB_CODE_SUCCESS;
  }
H
Haojun Liao 已提交
3493

H
Haojun Liao 已提交
3494 3495 3496 3497
  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 已提交
3498 3499 3500 3501
  if (pBlockInfo->compBlock->numOfSubBlocks > 1) {
    *pBlockStatis = NULL;
    return TSDB_CODE_SUCCESS;
  }
H
Haojun Liao 已提交
3502 3503

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

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

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

H
Haojun Liao 已提交
3517
  size_t numOfCols = QH_GET_NUM_OF_COLS(pHandle);
3518 3519 3520
  memset(pHandle->suppInfo.plist, 0, numOfCols * POINTER_BYTES);
  memset(pHandle->suppInfo.pstatis, 0, numOfCols * sizeof(SColumnDataAgg));

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

3525 3526
  *allHave = true;
  tsdbGetBlockStatis(&pHandle->rhelper, pHandle->suppInfo.pstatis, (int)numOfCols, pBlockInfo->compBlock);
H
Haojun Liao 已提交
3527 3528

  // always load the first primary timestamp column data
3529
  SColumnDataAgg* pPrimaryColStatis = &pHandle->suppInfo.pstatis[0];
3530
  assert(pPrimaryColStatis->colId == PRIMARYKEY_TIMESTAMP_COL_ID);
H
Haojun Liao 已提交
3531 3532 3533 3534

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

dengyihao's avatar
dengyihao 已提交
3537
  // update the number of NULL data rows
3538
  int32_t* slotIds = pHandle->suppInfo.slotIds;
dengyihao's avatar
dengyihao 已提交
3539
  for (int32_t i = 1; i < numOfCols; ++i) {
3540
    ASSERT(colIds[i] == pHandle->pSchema->columns[slotIds[i]].colId);
C
Cary Xu 已提交
3541
    if (IS_BSMA_ON(&(pHandle->pSchema->columns[slotIds[i]]))) {
3542 3543 3544 3545 3546 3547 3548
      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 已提交
3549 3550
    }
  }
H
Haojun Liao 已提交
3551 3552 3553 3554

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

3555
  *pBlockStatis = pHandle->suppInfo.plist;
3556
  return TSDB_CODE_SUCCESS;
H
hjxilinx 已提交
3557 3558
}

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

3571
    if (pHandle->cur.mixBlock) {
H
[td-32]  
hjxilinx 已提交
3572 3573
      return pHandle->pColumns;
    } else {
H
Haojun Liao 已提交
3574
      SDataBlockInfo binfo = GET_FILE_DATA_BLOCK_INFO(pCheckInfo, pBlockInfo->compBlock);
3575
      assert(pHandle->realNumOfRows <= binfo.rows);
H
Haojun Liao 已提交
3576

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

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

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

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

3604
  STSchema*   pTSSchema = (STSchema*) param;
H
hjxilinx 已提交
3605 3606 3607
  tQueryInfo* pInfo = pExpr->_node.info;
  tVariant*   pCond = pExpr->_node.pRight->pVal;
  SSchema*    pSchema = pExpr->_node.pLeft->pSchema;
3608

3609 3610
  pInfo->sch      = *pSchema;
  pInfo->optr     = pExpr->_node.optr;
Y
yihaoDeng 已提交
3611
  pInfo->compare  = getComparFunc(pInfo->sch.type, pInfo->optr);
H
Haojun Liao 已提交
3612
  pInfo->indexed  = pTSSchema->columns->colId == pInfo->sch.colId;
H
Haojun Liao 已提交
3613

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

3640
#endif
3641

dengyihao's avatar
dengyihao 已提交
3642
static int32_t tableGroupComparFn(const void* p1, const void* p2, const void* param) {
3643
#if 0
3644
  STableGroupSupporter* pTableGroupSupp = (STableGroupSupporter*) param;
3645 3646
  STable* pTable1 = ((STableKeyInfo*) p1)->uid;
  STable* pTable2 = ((STableKeyInfo*) p2)->uid;
H
Haojun Liao 已提交
3647

3648 3649 3650
  for (int32_t i = 0; i < pTableGroupSupp->numOfCols; ++i) {
    SColIndex* pColIndex = &pTableGroupSupp->pCols[i];
    int32_t colIndex = pColIndex->colIndex;
H
Haojun Liao 已提交
3651

H
Haojun Liao 已提交
3652
    assert(colIndex >= TSDB_TBNAME_COLUMN_INDEX);
H
Haojun Liao 已提交
3653

3654 3655 3656 3657
    char *  f1 = NULL;
    char *  f2 = NULL;
    int32_t type = 0;
    int32_t bytes = 0;
H
Haojun Liao 已提交
3658

H
Haojun Liao 已提交
3659 3660 3661
    if (colIndex == TSDB_TBNAME_COLUMN_INDEX) {
      f1 = (char*) TABLE_NAME(pTable1);
      f2 = (char*) TABLE_NAME(pTable2);
3662
      type = TSDB_DATA_TYPE_BINARY;
3663
      bytes = tGetTbnameColumnSchema()->bytes;
3664
    } else {
Y
yihaoDeng 已提交
3665 3666 3667 3668 3669 3670 3671
      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);
      } 
3672
    }
H
Haojun Liao 已提交
3673 3674 3675 3676 3677 3678 3679 3680 3681 3682 3683 3684 3685 3686

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

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

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

3687 3688 3689 3690 3691 3692 3693
    int32_t ret = doCompare(f1, f2, type, bytes);
    if (ret == 0) {
      continue;
    } else {
      return ret;
    }
  }
3694
#endif
3695 3696 3697
  return 0;
}

H
Haojun Liao 已提交
3698
static int tsdbCheckInfoCompar(const void* key1, const void* key2) {
3699
  if (((STableCheckInfo*)key1)->tableId < ((STableCheckInfo*)key2)->tableId) {
H
Haojun Liao 已提交
3700
    return -1;
3701
  } else if (((STableCheckInfo*)key1)->tableId > ((STableCheckInfo*)key2)->tableId) {
H
Haojun Liao 已提交
3702 3703 3704 3705 3706 3707 3708 3709 3710
    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) {
3711
  STable* pTable = taosArrayGetP(pTableList, 0);
H
Haojun Liao 已提交
3712 3713
  SArray* g = taosArrayInit(16, sizeof(STableKeyInfo));

3714
  STableKeyInfo info = {.lastKey = skey};
H
Haojun Liao 已提交
3715
  taosArrayPush(g, &info);
3716

3717
  for (int32_t i = 1; i < numOfTables; ++i) {
3718 3719
    STable** prev = taosArrayGet(pTableList, i - 1);
    STable** p = taosArrayGet(pTableList, i);
H
Haojun Liao 已提交
3720

H
hjxilinx 已提交
3721
    int32_t ret = compareFn(prev, p, pSupp);
3722
    assert(ret == 0 || ret == -1);
H
Haojun Liao 已提交
3723

3724
    if (ret == 0) {
3725
      STableKeyInfo info1 = {.lastKey = skey};
H
Haojun Liao 已提交
3726
      taosArrayPush(g, &info1);
3727 3728
    } else {
      taosArrayPush(pGroups, &g);  // current group is ended, start a new group
H
Haojun Liao 已提交
3729 3730
      g = taosArrayInit(16, sizeof(STableKeyInfo));

3731
      STableKeyInfo info1 = {.lastKey = skey};
H
Haojun Liao 已提交
3732
      taosArrayPush(g, &info1);
3733 3734
    }
  }
H
Haojun Liao 已提交
3735

3736
  taosArrayPush(pGroups, &g);
3737 3738
}

dengyihao's avatar
dengyihao 已提交
3739 3740
SArray* createTableGroup(SArray* pTableList, SSchemaWrapper* pTagSchema, SColIndex* pCols, int32_t numOfOrderCols,
                         TSKEY skey) {
3741
  assert(pTableList != NULL);
3742
  SArray* pTableGroup = taosArrayInit(1, POINTER_BYTES);
H
Haojun Liao 已提交
3743

3744 3745
  size_t size = taosArrayGetSize(pTableList);
  if (size == 0) {
S
Shengliang Guan 已提交
3746
    tsdbDebug("no qualified tables");
3747 3748
    return pTableGroup;
  }
H
Haojun Liao 已提交
3749

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

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

3765 3766
    taosqsort(pTableList->pData, size, sizeof(STableKeyInfo), &sup, tableGroupComparFn);
    createTableGroupImpl(pTableGroup, pTableList, size, skey, &sup, tableGroupComparFn);
3767
  }
H
Haojun Liao 已提交
3768

3769 3770 3771
  return pTableGroup;
}

dengyihao's avatar
dengyihao 已提交
3772
// static bool tableFilterFp(const void* pNode, void* param) {
3773 3774 3775 3776 3777 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
//  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 已提交
3856

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

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

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

H
Hongze Cheng 已提交
3878
  metaReaderInit(&mr, (SMeta*)pMeta, 0);
H
Hongze Cheng 已提交
3879 3880

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

H
Hongze Cheng 已提交
3889
  if (mr.me.type != TSDB_SUPER_TABLE) {
dengyihao's avatar
dengyihao 已提交
3890 3891 3892
    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 已提交
3893
    metaReaderClear(&mr);
3894
    goto _error;
H
hjxilinx 已提交
3895
  }
3896

H
Hongze Cheng 已提交
3897 3898
  metaReaderClear(&mr);

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

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

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

dengyihao's avatar
dengyihao 已提交
3913 3914 3915
    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);
3916

3917
    taosArrayDestroy(res);
3918 3919
    return ret;
  }
3920

H
hjxilinx 已提交
3921
  int32_t ret = TSDB_CODE_SUCCESS;
3922

dengyihao's avatar
dengyihao 已提交
3923 3924
  SFilterInfo* filterInfo = NULL;
  ret = filterInitFromNode((SNode*)pTagCond, &filterInfo, 0);
dengyihao's avatar
dengyihao 已提交
3925 3926 3927 3928 3929 3930 3931 3932
  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 已提交
3933 3934
  // 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 已提交
3935 3936 3937 3938 3939

  taosArrayDestroy(res);
  return ret;

_error:
3940
  return terrno;
3941
}
3942

dengyihao's avatar
dengyihao 已提交
3943 3944 3945 3946 3947
int32_t tsdbQueryTableList(void* pMeta, SArray* pRes, void* filterInfo) {
  // impl later

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

H
Hongze Cheng 已提交
3951
  metaReaderInit(&mr, (SMeta*)pMeta, 0);
H
Hongze Cheng 已提交
3952 3953

  if (metaGetTableEntryByUid(&mr, uid) < 0) {
D
dapan1121 已提交
3954
    terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST;
3955
    goto _error;
3956
  }
3957

H
Hongze Cheng 已提交
3958 3959
  metaReaderClear(&mr);

3960 3961
  pGroupInfo->numOfTables = 1;
  pGroupInfo->pGroupList = taosArrayInit(1, POINTER_BYTES);
H
Haojun Liao 已提交
3962

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

3965
  STableKeyInfo info = {.lastKey = startKey, .uid = uid};
H
Haojun Liao 已提交
3966
  taosArrayPush(group, &info);
H
Haojun Liao 已提交
3967

3968
  taosArrayPush(pGroupInfo->pGroupList, &group);
3969
  return TSDB_CODE_SUCCESS;
3970

dengyihao's avatar
dengyihao 已提交
3971
_error:
H
Hongze Cheng 已提交
3972
  metaReaderClear(&mr);
3973
  return terrno;
3974
}
3975

3976
#if 0
3977
int32_t tsdbGetTableGroupFromIdListT(STsdb* tsdb, SArray* pTableIdList, STableGroupInfo* pGroupInfo) {
B
Bomin Zhang 已提交
3978 3979 3980
  if (tsdbRLockRepoMeta(tsdb) < 0) {
    return terrno;
  }
3981 3982 3983 3984

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

B
Bomin Zhang 已提交
3987
  for(int32_t i = 0; i < size; ++i) {
3988 3989 3990 3991 3992 3993 3994 3995 3996 3997 3998
    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 已提交
3999 4000 4001
      tsdbUnlockRepoMeta(tsdb);
      taosArrayDestroy(group);
      return terrno;
4002 4003
    }

H
Haojun Liao 已提交
4004 4005
    STableKeyInfo info = {.pTable = pTable, .lastKey = id->key};
    taosArrayPush(group, &info);
4006 4007
  }

B
Bomin Zhang 已提交
4008 4009 4010 4011
  if (tsdbUnlockRepoMeta(tsdb) < 0) {
    taosArrayDestroy(group);
    return terrno;
  }
4012

sangshuduo's avatar
sangshuduo 已提交
4013
  pGroupInfo->numOfTables = (uint32_t) taosArrayGetSize(group);
B
Bomin Zhang 已提交
4014 4015 4016 4017 4018
  if (pGroupInfo->numOfTables > 0) {
    taosArrayPush(pGroupInfo->pGroupList, &group);
  } else {
    taosArrayDestroy(group);
  }
4019 4020 4021

  return TSDB_CODE_SUCCESS;
}
4022
#endif
4023 4024 4025 4026 4027 4028 4029 4030
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 已提交
4031
    taosMemoryFreeClear(pColInfo->pData);
4032 4033 4034 4035 4036 4037
  }

  taosArrayDestroy(pColumnInfoData);
  return NULL;
}

H
Haojun Liao 已提交
4038 4039 4040 4041 4042 4043
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 已提交
4044
    taosMemoryFreeClear(p->pCompInfo);
H
Haojun Liao 已提交
4045 4046 4047 4048 4049 4050
  }

  taosArrayDestroy(pTableCheckInfo);
  return NULL;
}

H
Haojun Liao 已提交
4051
void tsdbCleanupReadHandle(tsdbReaderT queryHandle) {
4052 4053
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)queryHandle;
  if (pTsdbReadHandle == NULL) {
4054 4055
    return;
  }
4056

4057
  pTsdbReadHandle->pColumns = doFreeColumnInfoData(pTsdbReadHandle->pColumns);
4058

4059
  taosArrayDestroy(pTsdbReadHandle->suppInfo.defaultLoadColumn);
wafwerar's avatar
wafwerar 已提交
4060
  taosMemoryFreeClear(pTsdbReadHandle->pDataBlockInfo);
4061 4062
  taosMemoryFreeClear(pTsdbReadHandle->suppInfo.pstatis);
  taosMemoryFreeClear(pTsdbReadHandle->suppInfo.plist);
4063

4064
  if (!emptyQueryTimewindow(pTsdbReadHandle)) {
dengyihao's avatar
dengyihao 已提交
4065
    //    tsdbMayUnTakeMemSnapshot(pTsdbReadHandle);
4066
  } else {
4067
    assert(pTsdbReadHandle->pTableCheckInfo == NULL);
4068 4069
  }

4070 4071
  if (pTsdbReadHandle->pTableCheckInfo != NULL) {
    pTsdbReadHandle->pTableCheckInfo = destroyTableCheckInfo(pTsdbReadHandle->pTableCheckInfo);
4072
  }
4073

4074
  tsdbDestroyReadH(&pTsdbReadHandle->rhelper);
H
Haojun Liao 已提交
4075

4076 4077
  tdFreeDataCols(pTsdbReadHandle->pDataCols);
  pTsdbReadHandle->pDataCols = NULL;
H
Haojun Liao 已提交
4078

4079 4080
  pTsdbReadHandle->prev = doFreeColumnInfoData(pTsdbReadHandle->prev);
  pTsdbReadHandle->next = doFreeColumnInfoData(pTsdbReadHandle->next);
4081

4082
  SIOCostSummary* pCost = &pTsdbReadHandle->cost;
4083

dengyihao's avatar
dengyihao 已提交
4084 4085 4086 4087
  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 已提交
4088

wafwerar's avatar
wafwerar 已提交
4089
  taosMemoryFreeClear(pTsdbReadHandle);
4090
}
4091

4092
#if 0
H
Haojun Liao 已提交
4093
void tsdbDestroyTableGroup(STableGroupInfo *pGroupList) {
4094 4095 4096 4097 4098 4099 4100 4101 4102 4103
  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);
4104 4105 4106 4107
      if (pTable != NULL) { // in case of handling retrieve data from tsdb
        tsdbUnRefTable(pTable);
      }
      //assert(pTable != NULL);
4108 4109 4110 4111 4112
    }

    taosArrayDestroy(p);
  }

4113
  taosHashCleanup(pGroupList->map);
4114
  taosArrayDestroy(pGroupList->pGroupList);
H
Haojun Liao 已提交
4115
  pGroupList->numOfTables = 0;
4116
}
H
Haojun Liao 已提交
4117 4118 4119 4120 4121 4122 4123

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 已提交
4124
    if (exprTreeApplyFilter(pExpr, pNode, param)) {
H
Haojun Liao 已提交
4125 4126 4127 4128 4129 4130 4131 4132 4133 4134 4135 4136 4137 4138 4139 4140 4141 4142 4143 4144 4145 4146 4147
      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 已提交
4148
    pCond->start       = taosMemoryCalloc(1, sizeof(SEndPoint));
H
Haojun Liao 已提交
4149
    pCond->start->optr = queryColInfo->optr;
Y
yihaoDeng 已提交
4150
    pCond->start->v    = queryColInfo->q;
H
Haojun Liao 已提交
4151
  } else if (optr == TSDB_RELATION_LESS || optr == TSDB_RELATION_LESS_EQUAL) {
wafwerar's avatar
wafwerar 已提交
4152
    pCond->end       = taosMemoryCalloc(1, sizeof(SEndPoint));
H
Haojun Liao 已提交
4153
    pCond->end->optr = queryColInfo->optr;
Y
yihaoDeng 已提交
4154 4155
    pCond->end->v    = queryColInfo->q;
  } else if (optr == TSDB_RELATION_IN) {
wafwerar's avatar
wafwerar 已提交
4156
    pCond->start       = taosMemoryCalloc(1, sizeof(SEndPoint));
Y
yihaoDeng 已提交
4157 4158 4159
    pCond->start->optr = queryColInfo->optr;
    pCond->start->v    = queryColInfo->q; 
  } else if (optr == TSDB_RELATION_LIKE) {
H
Haojun Liao 已提交
4160
    assert(0);
4161 4162
  } else if (optr == TSDB_RELATION_MATCH) {
    assert(0);
4163 4164
  } else if (optr == TSDB_RELATION_NMATCH) {
    assert(0);
H
Haojun Liao 已提交
4165 4166 4167 4168 4169 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
  }

  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 已提交
4248 4249 4250 4251 4252 4253 4254 4255 4256 4257 4258 4259 4260
    } 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 已提交
4261 4262 4263 4264 4265 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
    } 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 已提交
4302 4303
  taosMemoryFree(cond.start);
  taosMemoryFree(cond.end);
H
Haojun Liao 已提交
4304 4305 4306 4307 4308 4309 4310 4311 4312 4313 4314 4315 4316 4317 4318 4319 4320 4321
  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);
4322 4323 4324
      } else if (pQueryInfo->optr == TSDB_RELATION_LIKE ||
                 pQueryInfo->optr == TSDB_RELATION_MATCH ||
                 pQueryInfo->optr == TSDB_RELATION_NMATCH) {
H
Haojun Liao 已提交
4325 4326 4327 4328 4329 4330 4331 4332 4333 4334 4335 4336 4337 4338 4339 4340
        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 已提交
4341 4342 4343 4344 4345 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
//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 已提交
4374
#endif