tsdbRead.c 145.9 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 423
  if (pCond->numOfCols > 0) {
    // allocate buffer in order to load data blocks from file
424 425
    pReadHandle->suppInfo.pstatis = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnDataAgg));
    if (pReadHandle->suppInfo.pstatis == NULL) {
426
      goto _end;
427
    }
H
Haojun Liao 已提交
428

429
    // todo: use list instead of array?
430 431
    pReadHandle->pColumns = taosArrayInit(pCond->numOfCols, sizeof(SColumnInfoData));
    if (pReadHandle->pColumns == NULL) {
432
      goto _end;
433
    }
H
Haojun Liao 已提交
434

435 436 437
    for (int32_t i = 0; i < pCond->numOfCols; ++i) {
      SColumnInfoData colInfo = {{0}, 0};
      colInfo.info = pCond->colList[i];
H
Haojun Liao 已提交
438

439
      int32_t code = colInfoDataEnsureCapacity(&colInfo, 0, pReadHandle->outputCapacity);
440
      if (code != TSDB_CODE_SUCCESS) {
441
        goto _end;
442
      }
443

444
      taosArrayPush(pReadHandle->pColumns, &colInfo);
B
Bomin Zhang 已提交
445
    }
H
Haojun Liao 已提交
446

447
    pReadHandle->suppInfo.defaultLoadColumn = getDefaultLoadColumns(pReadHandle, true);
C
Cary Xu 已提交
448 449 450 451
    pReadHandle->suppInfo.slotIds =
        taosMemoryMalloc(sizeof(int32_t) * taosArrayGetSize(pReadHandle->suppInfo.defaultLoadColumn));
    pReadHandle->suppInfo.plist =
        taosMemoryCalloc(taosArrayGetSize(pReadHandle->suppInfo.defaultLoadColumn), POINTER_BYTES);
H
Haojun Liao 已提交
452
  }
453

C
Cary Xu 已提交
454
  pReadHandle->pDataCols = tdNewDataCols(1000, pVnode->config.tsdbCfg.maxRows);
455
  if (pReadHandle->pDataCols == NULL) {
H
Haojun Liao 已提交
456
    tsdbError("%p failed to malloc buf for pDataCols, %s", pReadHandle, pReadHandle->idStr);
H
Haojun Liao 已提交
457
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
458
    goto _end;
H
hjxilinx 已提交
459
  }
460

461 462
  tsdbInitDataBlockLoadInfo(&pReadHandle->dataBlockLoadInfo);
  tsdbInitCompBlockLoadInfo(&pReadHandle->compBlockLoadInfo);
463

H
Haojun Liao 已提交
464
  return (tsdbReaderT)pReadHandle;
465

dengyihao's avatar
dengyihao 已提交
466
_end:
467
  tsdbCleanupReadHandle(pReadHandle);
468
  terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
469
  return NULL;
H
hjxilinx 已提交
470 471
}

472
tsdbReaderT* tsdbQueryTables(SVnode* pVnode, SQueryTableDataCond* pCond, STableGroupInfo* groupList, uint64_t qId,
dengyihao's avatar
dengyihao 已提交
473
                             uint64_t taskId) {
474
  STsdbReadHandle* pTsdbReadHandle = tsdbQueryTablesImpl(pVnode, pCond, qId, taskId);
475
  if (pTsdbReadHandle == NULL) {
476 477 478
    return NULL;
  }

479
  if (emptyQueryTimewindow(pTsdbReadHandle)) {
dengyihao's avatar
dengyihao 已提交
480
    return (tsdbReaderT*)pTsdbReadHandle;
481
  }
H
Haojun Liao 已提交
482 483

  // todo apply the lastkey of table check to avoid to load header file
484
  pTsdbReadHandle->pTableCheckInfo = createCheckInfoFromTableGroup(pTsdbReadHandle, groupList);
485
  if (pTsdbReadHandle->pTableCheckInfo == NULL) {
dengyihao's avatar
dengyihao 已提交
486
    //    tsdbCleanupReadHandle(pTsdbReadHandle);
H
Haojun Liao 已提交
487 488 489 490
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    return NULL;
  }

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

H
Hongze Cheng 已提交
493
  pTsdbReadHandle->pSchema = metaGetTbTSchema(pVnode->pMeta, pCheckInfo->tableId, 1);
C
Cary Xu 已提交
494
  int32_t  numOfCols = taosArrayGetSize(pTsdbReadHandle->suppInfo.defaultLoadColumn);
495 496 497 498 499
  int16_t* ids = pTsdbReadHandle->suppInfo.defaultLoadColumn->pData;

  STSchema* pSchema = pTsdbReadHandle->pSchema;

  int32_t i = 0, j = 0;
C
Cary Xu 已提交
500
  while (i < numOfCols && j < pSchema->numOfCols) {
501 502 503 504 505 506 507 508 509 510 511 512
    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;
    }
  }
513

dengyihao's avatar
dengyihao 已提交
514 515 516
  tsdbDebug("%p total numOfTable:%" PRIzu " in this query, group %" PRIzu " %s", pTsdbReadHandle,
            taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo), taosArrayGetSize(groupList->pGroupList),
            pTsdbReadHandle->idStr);
517

dengyihao's avatar
dengyihao 已提交
518
  return (tsdbReaderT)pTsdbReadHandle;
H
Haojun Liao 已提交
519 520
}

521
void tsdbResetReadHandle(tsdbReaderT queryHandle, SQueryTableDataCond* pCond) {
522
  STsdbReadHandle* pTsdbReadHandle = queryHandle;
H
Haojun Liao 已提交
523

524 525 526
  if (emptyQueryTimewindow(pTsdbReadHandle)) {
    if (pCond->order != pTsdbReadHandle->order) {
      pTsdbReadHandle->order = pCond->order;
wafwerar's avatar
wafwerar 已提交
527
      TSWAP(pTsdbReadHandle->window.skey, pTsdbReadHandle->window.ekey);
528 529 530 531 532
    }

    return;
  }

dengyihao's avatar
dengyihao 已提交
533 534 535 536 537 538 539
  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
540 541
  pTsdbReadHandle->locateStart = false;
  pTsdbReadHandle->loadExternalRow = pCond->loadExternalRows;
H
Haojun Liao 已提交
542 543

  if (ASCENDING_TRAVERSE(pCond->order)) {
544
    assert(pTsdbReadHandle->window.skey <= pTsdbReadHandle->window.ekey);
H
Haojun Liao 已提交
545
  } else {
546
    assert(pTsdbReadHandle->window.skey >= pTsdbReadHandle->window.ekey);
H
Haojun Liao 已提交
547 548 549
  }

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

553 554
  tsdbInitDataBlockLoadInfo(&pTsdbReadHandle->dataBlockLoadInfo);
  tsdbInitCompBlockLoadInfo(&pTsdbReadHandle->compBlockLoadInfo);
H
Haojun Liao 已提交
555

556
  resetCheckInfo(pTsdbReadHandle);
H
Haojun Liao 已提交
557 558
}

559
void tsdbResetQueryHandleForNewTable(tsdbReaderT queryHandle, SQueryTableDataCond* pCond, STableGroupInfo* groupList) {
560
  STsdbReadHandle* pTsdbReadHandle = queryHandle;
H
Haojun Liao 已提交
561

dengyihao's avatar
dengyihao 已提交
562 563 564 565 566 567 568
  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
569 570
  pTsdbReadHandle->locateStart = false;
  pTsdbReadHandle->loadExternalRow = pCond->loadExternalRows;
H
Haojun Liao 已提交
571 572

  if (ASCENDING_TRAVERSE(pCond->order)) {
573
    assert(pTsdbReadHandle->window.skey <= pTsdbReadHandle->window.ekey);
H
Haojun Liao 已提交
574
  } else {
575
    assert(pTsdbReadHandle->window.skey >= pTsdbReadHandle->window.ekey);
H
Haojun Liao 已提交
576 577 578
  }

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

582 583
  tsdbInitDataBlockLoadInfo(&pTsdbReadHandle->dataBlockLoadInfo);
  tsdbInitCompBlockLoadInfo(&pTsdbReadHandle->compBlockLoadInfo);
H
Haojun Liao 已提交
584

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

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

dengyihao's avatar
dengyihao 已提交
590 591
  pTsdbReadHandle->pTableCheckInfo = NULL;  // createCheckInfoFromTableGroup(pTsdbReadHandle, groupList, pMeta,
                                            // &pTable);
592
  if (pTsdbReadHandle->pTableCheckInfo == NULL) {
dengyihao's avatar
dengyihao 已提交
593
    //    tsdbCleanupReadHandle(pTsdbReadHandle);
H
Haojun Liao 已提交
594 595
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
  }
H
Haojun Liao 已提交
596

dengyihao's avatar
dengyihao 已提交
597 598
  //  pTsdbReadHandle->prev = doFreeColumnInfoData(pTsdbReadHandle->prev);
  //  pTsdbReadHandle->next = doFreeColumnInfoData(pTsdbReadHandle->next);
H
Haojun Liao 已提交
599 600
}

601
tsdbReaderT tsdbQueryLastRow(SVnode* pVnode, SQueryTableDataCond* pCond, STableGroupInfo* groupList, uint64_t qId,
dengyihao's avatar
dengyihao 已提交
602
                             uint64_t taskId) {
603
  pCond->twindow = updateLastrowForEachGroup(groupList);
H
Haojun Liao 已提交
604 605 606 607 608 609

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

610
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)tsdbQueryTables(pVnode, pCond, groupList, qId, taskId);
611
  if (pTsdbReadHandle == NULL) {
612 613 614
    return NULL;
  }

615
  int32_t code = checkForCachedLastRow(pTsdbReadHandle, groupList);
dengyihao's avatar
dengyihao 已提交
616
  if (code != TSDB_CODE_SUCCESS) {  // set the numOfTables to be 0
H
Haojun Liao 已提交
617 618 619
    terrno = code;
    return NULL;
  }
H
Haojun Liao 已提交
620 621

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

626
  return pTsdbReadHandle;
D
init  
dapan1121 已提交
627 628
}

629
#if 0
630 631
tsdbReaderT tsdbQueryCacheLastT(STsdb *tsdb, SQueryTableDataCond *pCond, STableGroupInfo *groupList, uint64_t qId, STsdbMemTable* pMemRef) {
  STsdbReadHandle *pTsdbReadHandle = (STsdbReadHandle*) tsdbQueryTablesT(tsdb, pCond, groupList, qId, pMemRef);
632
  if (pTsdbReadHandle == NULL) {
633 634 635
    return NULL;
  }

636
  int32_t code = checkForCachedLast(pTsdbReadHandle);
D
init  
dapan1121 已提交
637 638 639 640 641
  if (code != TSDB_CODE_SUCCESS) { // set the numOfTables to be 0
    terrno = code;
    return NULL;
  }

642 643
  if (pTsdbReadHandle->cachelastrow) {
    pTsdbReadHandle->type = TSDB_QUERY_TYPE_LAST;
D
fix bug  
dapan1121 已提交
644
  }
D
init  
dapan1121 已提交
645
  
646
  return pTsdbReadHandle;
H
hjxilinx 已提交
647 648
}

649
#endif
dengyihao's avatar
dengyihao 已提交
650
SArray* tsdbGetQueriedTableList(tsdbReaderT* pHandle) {
651
  assert(pHandle != NULL);
H
Haojun Liao 已提交
652

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

dengyihao's avatar
dengyihao 已提交
655
  size_t  size = taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
656
  SArray* res = taosArrayInit(size, POINTER_BYTES);
657 658 659
  return res;
}

H
Haojun Liao 已提交
660 661 662 663 664
// 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 已提交
665
  STableGroupInfo* pNew = taosMemoryCalloc(1, sizeof(STableGroupInfo));
Y
yihaoDeng 已提交
666
  pNew->pGroupList = taosArrayInit(numOfGroup, POINTER_BYTES);
H
Haojun Liao 已提交
667

dengyihao's avatar
dengyihao 已提交
668
  for (int32_t i = 0; i < numOfGroup; ++i) {
H
Haojun Liao 已提交
669
    SArray* oneGroup = taosArrayGetP(pGroupList->pGroupList, i);
dengyihao's avatar
dengyihao 已提交
670
    size_t  numOfTables = taosArrayGetSize(oneGroup);
H
Haojun Liao 已提交
671 672 673 674

    SArray* px = taosArrayInit(4, sizeof(STableKeyInfo));
    for (int32_t j = 0; j < numOfTables; ++j) {
      STableKeyInfo* pInfo = (STableKeyInfo*)taosArrayGet(oneGroup, j);
dengyihao's avatar
dengyihao 已提交
675 676 677 678 679
      //      if (window->skey <= pInfo->lastKey && ((STable*)pInfo->pTable)->lastKey != TSKEY_INITIAL_VAL) {
      //        taosArrayPush(px, pInfo);
      //        pNew->numOfTables += 1;
      //        break;
      //      }
H
Haojun Liao 已提交
680 681 682 683 684 685 686 687 688 689 690 691 692
    }

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

  return pNew;
}

693
tsdbReaderT tsdbQueryRowsInExternalWindow(SVnode* pVnode, SQueryTableDataCond* pCond, STableGroupInfo* groupList,
694
                                          uint64_t qId, uint64_t taskId) {
H
Haojun Liao 已提交
695 696
  STableGroupInfo* pNew = trimTableGroup(&pCond->twindow, groupList);

697 698 699 700 701 702 703 704 705 706 707 708
  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;
    }
  }

709
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)tsdbQueryTables(pVnode, pCond, pNew, qId, taskId);
710 711
  pTsdbReadHandle->loadExternalRow = true;
  pTsdbReadHandle->currentLoadExternalRows = true;
712

713
  return pTsdbReadHandle;
714 715
}

716
static bool initTableMemIterator(STsdbReadHandle* pHandle, STableCheckInfo* pCheckInfo) {
717
  if (pCheckInfo->initBuf) {
718 719
    return true;
  }
H
Haojun Liao 已提交
720

721
  pCheckInfo->initBuf = true;
722
  int32_t order = pHandle->order;
H
Haojun Liao 已提交
723

724 725 726
  STbData** pMem = NULL;
  STbData** pIMem = NULL;

727
  TSKEY tLastKey = keyToTkey(pCheckInfo->lastKey);
728 729 730
  if (pHandle->pTsdb->mem != NULL) {
    pMem = taosHashGet(pHandle->pTsdb->mem->pHashIdx, &pCheckInfo->tableId, sizeof(pCheckInfo->tableId));
    if (pMem != NULL) {
H
Haojun Liao 已提交
731
      pCheckInfo->iter =
732
          tSkipListCreateIterFromVal((*pMem)->pData, (const char*)&tLastKey, TSDB_DATA_TYPE_TIMESTAMP, order);
H
Haojun Liao 已提交
733
    }
734
  }
H
Haojun Liao 已提交
735

736 737 738
  if (pHandle->pTsdb->imem != NULL) {
    pIMem = taosHashGet(pHandle->pTsdb->imem->pHashIdx, &pCheckInfo->tableId, sizeof(pCheckInfo->tableId));
    if (pIMem != NULL) {
H
Haojun Liao 已提交
739
      pCheckInfo->iiter =
740
          tSkipListCreateIterFromVal((*pIMem)->pData, (const char*)&tLastKey, TSDB_DATA_TYPE_TIMESTAMP, order);
H
Haojun Liao 已提交
741
    }
742
  }
H
Haojun Liao 已提交
743

744 745 746 747
  // both iterators are NULL, no data in buffer right now
  if (pCheckInfo->iter == NULL && pCheckInfo->iiter == NULL) {
    return false;
  }
H
Haojun Liao 已提交
748

dengyihao's avatar
dengyihao 已提交
749
  bool memEmpty = (pCheckInfo->iter == NULL) || (pCheckInfo->iter != NULL && !tSkipListIterNext(pCheckInfo->iter));
750
  bool imemEmpty = (pCheckInfo->iiter == NULL) || (pCheckInfo->iiter != NULL && !tSkipListIterNext(pCheckInfo->iiter));
dengyihao's avatar
dengyihao 已提交
751
  if (memEmpty && imemEmpty) {  // buffer is empty
752 753
    return false;
  }
H
Haojun Liao 已提交
754

755 756 757
  if (!memEmpty) {
    SSkipListNode* node = tSkipListIterGet(pCheckInfo->iter);
    assert(node != NULL);
H
Haojun Liao 已提交
758

H
Haojun Liao 已提交
759 760
    STSRow* row = (STSRow*)SL_GET_NODE_DATA(node);
    TSKEY   key = TD_ROW_KEY(row);  // first timestamp in buffer
761
    tsdbDebug("%p uid:%" PRId64 ", check data in mem from skey:%" PRId64 ", order:%d, ts range in buf:%" PRId64
dengyihao's avatar
dengyihao 已提交
762 763 764
              "-%" PRId64 ", lastKey:%" PRId64 ", numOfRows:%" PRId64 ", %s",
              pHandle, pCheckInfo->tableId, key, order, (*pMem)->keyMin, (*pMem)->keyMax, pCheckInfo->lastKey,
              (*pMem)->nrows, pHandle->idStr);
H
Haojun Liao 已提交
765 766 767 768 769 770 771

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

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

776 777 778
  if (!imemEmpty) {
    SSkipListNode* node = tSkipListIterGet(pCheckInfo->iiter);
    assert(node != NULL);
H
Haojun Liao 已提交
779

H
Haojun Liao 已提交
780 781
    STSRow* row = (STSRow*)SL_GET_NODE_DATA(node);
    TSKEY   key = TD_ROW_KEY(row);  // first timestamp in buffer
782
    tsdbDebug("%p uid:%" PRId64 ", check data in imem from skey:%" PRId64 ", order:%d, ts range in buf:%" PRId64
dengyihao's avatar
dengyihao 已提交
783 784 785
              "-%" PRId64 ", lastKey:%" PRId64 ", numOfRows:%" PRId64 ", %s",
              pHandle, pCheckInfo->tableId, key, order, (*pIMem)->keyMin, (*pIMem)->keyMax, pCheckInfo->lastKey,
              (*pIMem)->nrows, pHandle->idStr);
H
Haojun Liao 已提交
786 787 788 789 790 791

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

796 797 798
  return true;
}

H
Haojun Liao 已提交
799 800 801 802 803
static void destroyTableMemIterator(STableCheckInfo* pCheckInfo) {
  tSkipListDestroyIter(pCheckInfo->iter);
  tSkipListDestroyIter(pCheckInfo->iiter);
}

C
Cary Xu 已提交
804
static TSKEY extractFirstTraverseKey(STableCheckInfo* pCheckInfo, int32_t order, int32_t update, TDRowVerT maxVer) {
H
Haojun Liao 已提交
805
  STSRow *rmem = NULL, *rimem = NULL;
806 807 808
  if (pCheckInfo->iter) {
    SSkipListNode* node = tSkipListIterGet(pCheckInfo->iter);
    if (node != NULL) {
H
Haojun Liao 已提交
809
      rmem = (STSRow*)SL_GET_NODE_DATA(node);
C
Cary Xu 已提交
810 811 812 813
      // TODO: filter max version
      // if (TD_ROW_VER(rmem) > maxVer) {
      //   rmem = NULL;
      // }
814 815 816 817 818 819
    }
  }

  if (pCheckInfo->iiter) {
    SSkipListNode* node = tSkipListIterGet(pCheckInfo->iiter);
    if (node != NULL) {
H
Haojun Liao 已提交
820
      rimem = (STSRow*)SL_GET_NODE_DATA(node);
C
Cary Xu 已提交
821 822 823 824
      // TODO: filter max version
      // if (TD_ROW_VER(rimem) > maxVer) {
      //   rimem = NULL;
      // }
825 826 827 828 829 830 831 832 833
    }
  }

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

  if (rmem != NULL && rimem == NULL) {
    pCheckInfo->chosen = CHECKINFO_CHOSEN_MEM;
H
Haojun Liao 已提交
834
    return TD_ROW_KEY(rmem);
835 836 837 838
  }

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

H
Haojun Liao 已提交
842 843
  TSKEY r1 = TD_ROW_KEY(rmem);
  TSKEY r2 = TD_ROW_KEY(rimem);
844 845

  if (r1 == r2) {
C
Cary Xu 已提交
846
#if 0
dengyihao's avatar
dengyihao 已提交
847
    if (update == TD_ROW_DISCARD_UPDATE) {
848 849
      pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
      tSkipListIterNext(pCheckInfo->iter);
dengyihao's avatar
dengyihao 已提交
850
    } else if (update == TD_ROW_OVERWRITE_UPDATE) {
851 852 853 854 855
      pCheckInfo->chosen = CHECKINFO_CHOSEN_MEM;
      tSkipListIterNext(pCheckInfo->iiter);
    } else {
      pCheckInfo->chosen = CHECKINFO_CHOSEN_BOTH;
    }
C
Cary Xu 已提交
856 857 858 859 860 861 862
#endif
    if (TD_SUPPORT_UPDATE(update)) {
      pCheckInfo->chosen = CHECKINFO_CHOSEN_BOTH;
    } else {
      pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
      tSkipListIterNext(pCheckInfo->iter);
    }
863 864 865 866
    return r1;
  } else if (r1 < r2 && ASCENDING_TRAVERSE(order)) {
    pCheckInfo->chosen = CHECKINFO_CHOSEN_MEM;
    return r1;
dengyihao's avatar
dengyihao 已提交
867
  } else {
868 869 870 871 872
    pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
    return r2;
  }
}

C
Cary Xu 已提交
873 874
static STSRow* getSRowInTableMem(STableCheckInfo* pCheckInfo, int32_t order, int32_t update, STSRow** extraRow,
                                 TDRowVerT maxVer) {
H
Haojun Liao 已提交
875
  STSRow *rmem = NULL, *rimem = NULL;
H
Haojun Liao 已提交
876 877 878
  if (pCheckInfo->iter) {
    SSkipListNode* node = tSkipListIterGet(pCheckInfo->iter);
    if (node != NULL) {
H
Haojun Liao 已提交
879
      rmem = (STSRow*)SL_GET_NODE_DATA(node);
C
Cary Xu 已提交
880
#if 0  // TODO: skiplist refactor
C
Cary Xu 已提交
881 882 883
      if (TD_ROW_VER(rmem) > maxVer) {
        rmem = NULL;
      }
C
Cary Xu 已提交
884
#endif
H
Haojun Liao 已提交
885 886
    }
  }
887

H
Haojun Liao 已提交
888 889 890
  if (pCheckInfo->iiter) {
    SSkipListNode* node = tSkipListIterGet(pCheckInfo->iiter);
    if (node != NULL) {
H
Haojun Liao 已提交
891
      rimem = (STSRow*)SL_GET_NODE_DATA(node);
C
Cary Xu 已提交
892
#if 0  // TODO: skiplist refactor
C
Cary Xu 已提交
893 894 895
      if (TD_ROW_VER(rimem) > maxVer) {
        rimem = NULL;
      }
C
Cary Xu 已提交
896
#endif
H
Haojun Liao 已提交
897 898
    }
  }
899

H
Haojun Liao 已提交
900 901
  if (rmem == NULL && rimem == NULL) {
    return NULL;
H
Haojun Liao 已提交
902
  }
903

H
Haojun Liao 已提交
904
  if (rmem != NULL && rimem == NULL) {
H
Haojun Liao 已提交
905 906 907
    pCheckInfo->chosen = 0;
    return rmem;
  }
908

H
Haojun Liao 已提交
909
  if (rmem == NULL && rimem != NULL) {
H
Haojun Liao 已提交
910 911 912
    pCheckInfo->chosen = 1;
    return rimem;
  }
913

H
Haojun Liao 已提交
914 915
  TSKEY r1 = TD_ROW_KEY(rmem);
  TSKEY r2 = TD_ROW_KEY(rimem);
H
Haojun Liao 已提交
916

917
  if (r1 == r2) {
C
Cary Xu 已提交
918
#if 0
919
    if (update == TD_ROW_DISCARD_UPDATE) {
H
TD-1439  
Hongze Cheng 已提交
920
      tSkipListIterNext(pCheckInfo->iter);
921
      pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
H
TD-1439  
Hongze Cheng 已提交
922
      return rimem;
dengyihao's avatar
dengyihao 已提交
923
    } else if (update == TD_ROW_OVERWRITE_UPDATE) {
H
TD-1439  
Hongze Cheng 已提交
924
      tSkipListIterNext(pCheckInfo->iiter);
925 926 927 928
      pCheckInfo->chosen = CHECKINFO_CHOSEN_MEM;
      return rmem;
    } else {
      pCheckInfo->chosen = CHECKINFO_CHOSEN_BOTH;
H
Haojun Liao 已提交
929
      *extraRow = rimem;
H
TD-1439  
Hongze Cheng 已提交
930 931
      return rmem;
    }
C
Cary Xu 已提交
932 933 934 935 936 937 938 939 940 941
#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 已提交
942 943 944
  } else {
    if (ASCENDING_TRAVERSE(order)) {
      if (r1 < r2) {
945
        pCheckInfo->chosen = CHECKINFO_CHOSEN_MEM;
H
Haojun Liao 已提交
946 947
        return rmem;
      } else {
948
        pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
H
Haojun Liao 已提交
949 950 951 952
        return rimem;
      }
    } else {
      if (r1 < r2) {
953
        pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
H
Haojun Liao 已提交
954 955
        return rimem;
      } else {
956
        pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
H
Haojun Liao 已提交
957 958 959 960
        return rmem;
      }
    }
  }
H
Haojun Liao 已提交
961 962
}

963
static bool moveToNextRowInMem(STableCheckInfo* pCheckInfo) {
H
Haojun Liao 已提交
964
  bool hasNext = false;
965
  if (pCheckInfo->chosen == CHECKINFO_CHOSEN_MEM) {
H
Haojun Liao 已提交
966 967 968
    if (pCheckInfo->iter != NULL) {
      hasNext = tSkipListIterNext(pCheckInfo->iter);
    }
969

H
Haojun Liao 已提交
970 971 972
    if (hasNext) {
      return hasNext;
    }
973

H
Haojun Liao 已提交
974 975 976
    if (pCheckInfo->iiter != NULL) {
      return tSkipListIterGet(pCheckInfo->iiter) != NULL;
    }
dengyihao's avatar
dengyihao 已提交
977
  } else if (pCheckInfo->chosen == CHECKINFO_CHOSEN_IMEM) {
978 979 980
    if (pCheckInfo->iiter != NULL) {
      hasNext = tSkipListIterNext(pCheckInfo->iiter);
    }
981

982 983 984
    if (hasNext) {
      return hasNext;
    }
985

986 987
    if (pCheckInfo->iter != NULL) {
      return tSkipListIterGet(pCheckInfo->iter) != NULL;
H
Haojun Liao 已提交
988
    }
989 990 991 992 993 994 995
  } else {
    if (pCheckInfo->iter != NULL) {
      hasNext = tSkipListIterNext(pCheckInfo->iter);
    }
    if (pCheckInfo->iiter != NULL) {
      hasNext = tSkipListIterNext(pCheckInfo->iiter) || hasNext;
    }
H
Haojun Liao 已提交
996
  }
997

H
Haojun Liao 已提交
998 999 1000
  return hasNext;
}

1001
static bool hasMoreDataInCache(STsdbReadHandle* pHandle) {
H
Hongze Cheng 已提交
1002
  STsdbCfg* pCfg = REPO_CFG(pHandle->pTsdb);
dengyihao's avatar
dengyihao 已提交
1003
  size_t    size = taosArrayGetSize(pHandle->pTableCheckInfo);
1004
  assert(pHandle->activeIndex < size && pHandle->activeIndex >= 0 && size >= 1);
D
dapan1121 已提交
1005
  pHandle->cur.fid = INT32_MIN;
H
Haojun Liao 已提交
1006

1007
  STableCheckInfo* pCheckInfo = taosArrayGet(pHandle->pTableCheckInfo, pHandle->activeIndex);
H
Haojun Liao 已提交
1008 1009 1010 1011
  if (!pCheckInfo->initBuf) {
    initTableMemIterator(pHandle, pCheckInfo);
  }

C
Cary Xu 已提交
1012
  STSRow* row = getSRowInTableMem(pCheckInfo, pHandle->order, pCfg->update, NULL, TD_VER_MAX);
H
Haojun Liao 已提交
1013
  if (row == NULL) {
1014 1015
    return false;
  }
1016

H
Haojun Liao 已提交
1017
  pCheckInfo->lastKey = TD_ROW_KEY(row);  // first timestamp in buffer
dengyihao's avatar
dengyihao 已提交
1018 1019
  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 已提交
1020

1021
  // all data in mem are checked already.
1022 1023
  if ((pCheckInfo->lastKey > pHandle->window.ekey && ASCENDING_TRAVERSE(pHandle->order)) ||
      (pCheckInfo->lastKey < pHandle->window.ekey && !ASCENDING_TRAVERSE(pHandle->order))) {
1024 1025
    return false;
  }
H
Haojun Liao 已提交
1026

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

1031 1032 1033 1034
  // update the last key value
  pCheckInfo->lastKey = win->ekey + step;
  pHandle->cur.lastKey = win->ekey + step;
  pHandle->cur.mixBlock = true;
1035

1036
  if (!ASCENDING_TRAVERSE(pHandle->order)) {
wafwerar's avatar
wafwerar 已提交
1037
    TSWAP(win->skey, win->ekey);
1038
  }
H
Haojun Liao 已提交
1039

1040
  return true;
1041
}
H
hjxilinx 已提交
1042

1043 1044
static int32_t getFileIdFromKey(TSKEY key, int32_t daysPerFile, int32_t precision) {
  assert(precision >= TSDB_TIME_PRECISION_MICRO || precision <= TSDB_TIME_PRECISION_NANO);
1045 1046 1047
  if (key == TSKEY_INITIAL_VAL) {
    return INT32_MIN;
  }
H
Haojun Liao 已提交
1048

D
dapan1121 已提交
1049
  if (key < 0) {
1050
    key -= (daysPerFile * tsTickPerMin[precision]);
D
dapan1121 已提交
1051
  }
dengyihao's avatar
dengyihao 已提交
1052

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

1058
  if (fid > 0L && fid > INT32_MAX) {
1059 1060
    fid = INT32_MAX;
  }
H
Haojun Liao 已提交
1061

S
TD-1057  
Shengliang Guan 已提交
1062
  return (int32_t)fid;
1063 1064
}

H
refact  
Hongze Cheng 已提交
1065
static int32_t binarySearchForBlock(SBlock* pBlock, int32_t numOfBlocks, TSKEY skey, int32_t order) {
1066 1067
  int32_t firstSlot = 0;
  int32_t lastSlot = numOfBlocks - 1;
H
Haojun Liao 已提交
1068

1069
  int32_t midSlot = firstSlot;
H
Haojun Liao 已提交
1070

1071 1072 1073
  while (1) {
    numOfBlocks = lastSlot - firstSlot + 1;
    midSlot = (firstSlot + (numOfBlocks >> 1));
H
Haojun Liao 已提交
1074

1075
    if (numOfBlocks == 1) break;
H
Haojun Liao 已提交
1076

1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087
    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 已提交
1088

1089 1090
  return midSlot;
}
1091

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

1095
  STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, index);
H
Haojun Liao 已提交
1096
  pCheckInfo->numOfBlocks = 0;
1097

H
Haojun Liao 已提交
1098
  STable table = {.uid = pCheckInfo->tableId, .tid = pCheckInfo->tableId};
1099
  table.pSchema = pTsdbReadHandle->pSchema;
H
Haojun Liao 已提交
1100 1101

  if (tsdbSetReadTable(&pTsdbReadHandle->rhelper, &table) != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
1102 1103 1104
    code = terrno;
    return code;
  }
1105

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

H
Haojun Liao 已提交
1108
  // no data block in this file, try next file
1109
  if (compIndex == NULL || compIndex->uid != pCheckInfo->tableId) {
H
Haojun Liao 已提交
1110 1111
    return 0;  // no data blocks in the file belongs to pCheckInfo->pTable
  }
1112

H
Haojun Liao 已提交
1113 1114 1115
  if (pCheckInfo->compSize < (int32_t)compIndex->len) {
    assert(compIndex->len > 0);

wafwerar's avatar
wafwerar 已提交
1116
    char* t = taosMemoryRealloc(pCheckInfo->pCompInfo, compIndex->len);
H
Haojun Liao 已提交
1117 1118 1119 1120
    if (t == NULL) {
      terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
      code = TSDB_CODE_TDB_OUT_OF_MEMORY;
      return code;
1121 1122
    }

H
Haojun Liao 已提交
1123 1124 1125
    pCheckInfo->pCompInfo = (SBlockInfo*)t;
    pCheckInfo->compSize = compIndex->len;
  }
1126

1127
  if (tsdbLoadBlockInfo(&(pTsdbReadHandle->rhelper), (void*)(pCheckInfo->pCompInfo)) < 0) {
H
Hongze Cheng 已提交
1128 1129
    return terrno;
  }
H
Haojun Liao 已提交
1130
  SBlockInfo* pCompInfo = pCheckInfo->pCompInfo;
1131

H
Haojun Liao 已提交
1132
  TSKEY s = TSKEY_INITIAL_VAL, e = TSKEY_INITIAL_VAL;
1133

1134
  if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
dengyihao's avatar
dengyihao 已提交
1135 1136
    assert(pCheckInfo->lastKey <= pTsdbReadHandle->window.ekey &&
           pTsdbReadHandle->window.skey <= pTsdbReadHandle->window.ekey);
H
Haojun Liao 已提交
1137
  } else {
dengyihao's avatar
dengyihao 已提交
1138 1139
    assert(pCheckInfo->lastKey >= pTsdbReadHandle->window.ekey &&
           pTsdbReadHandle->window.skey >= pTsdbReadHandle->window.ekey);
H
Haojun Liao 已提交
1140
  }
1141

dengyihao's avatar
dengyihao 已提交
1142 1143
  s = TMIN(pCheckInfo->lastKey, pTsdbReadHandle->window.ekey);
  e = TMAX(pCheckInfo->lastKey, pTsdbReadHandle->window.ekey);
1144

H
Haojun Liao 已提交
1145 1146 1147
  // 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 已提交
1148

H
Haojun Liao 已提交
1149 1150 1151
  if (s > pCompInfo->blocks[start].keyLast) {
    return 0;
  }
1152

H
Haojun Liao 已提交
1153 1154 1155 1156
  // todo speedup the procedure of located end block
  while (end < (int32_t)compIndex->numOfBlocks && (pCompInfo->blocks[end].keyFirst <= e)) {
    end += 1;
  }
1157

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

H
Haojun Liao 已提交
1160 1161 1162
  if (start > 0) {
    memmove(pCompInfo->blocks, &pCompInfo->blocks[start], pCheckInfo->numOfBlocks * sizeof(SBlock));
  }
1163

H
Haojun Liao 已提交
1164 1165 1166
  (*numOfBlocks) += pCheckInfo->numOfBlocks;
  return 0;
}
1167

1168
static int32_t getFileCompInfo(STsdbReadHandle* pTsdbReadHandle, int32_t* numOfBlocks) {
H
Haojun Liao 已提交
1169 1170 1171 1172
  // load all the comp offset value for all tables in this file
  int32_t code = TSDB_CODE_SUCCESS;
  *numOfBlocks = 0;

1173
  pTsdbReadHandle->cost.headFileLoad += 1;
1174 1175
  int64_t s = taosGetTimestampUs();

H
Haojun Liao 已提交
1176
  size_t numOfTables = 0;
1177 1178 1179 1180
  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);
1181

H
Haojun Liao 已提交
1182
    for (int32_t i = 0; i < numOfTables; ++i) {
1183
      code = loadBlockInfo(pTsdbReadHandle, i, numOfBlocks);
H
Haojun Liao 已提交
1184
      if (code != TSDB_CODE_SUCCESS) {
1185 1186
        int64_t e = taosGetTimestampUs();

1187
        pTsdbReadHandle->cost.headFileLoadTime += (e - s);
H
Haojun Liao 已提交
1188 1189 1190 1191 1192
        return code;
      }
    }
  } else {
    assert(0);
1193
  }
1194

1195
  int64_t e = taosGetTimestampUs();
1196
  pTsdbReadHandle->cost.headFileLoadTime += (e - s);
H
Haojun Liao 已提交
1197
  return code;
1198 1199
}

dengyihao's avatar
dengyihao 已提交
1200 1201
static int32_t doLoadFileDataBlock(STsdbReadHandle* pTsdbReadHandle, SBlock* pBlock, STableCheckInfo* pCheckInfo,
                                   int32_t slotIndex) {
H
Haojun Liao 已提交
1202
  int64_t st = taosGetTimestampUs();
1203

C
Cary Xu 已提交
1204
  int32_t code = tdInitDataCols(pTsdbReadHandle->pDataCols, pTsdbReadHandle->pSchema);
H
Haojun Liao 已提交
1205
  if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
1206
    tsdbError("%p failed to malloc buf for pDataCols, %s", pTsdbReadHandle, pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
1207 1208 1209 1210
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    goto _error;
  }

1211
  code = tdInitDataCols(pTsdbReadHandle->rhelper.pDCols[0], pTsdbReadHandle->pSchema);
H
Haojun Liao 已提交
1212
  if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
1213
    tsdbError("%p failed to malloc buf for rhelper.pDataCols[0], %s", pTsdbReadHandle, pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
1214 1215 1216 1217
    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
    goto _error;
  }

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

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

dengyihao's avatar
dengyihao 已提交
1227 1228
  int32_t ret = tsdbLoadBlockDataCols(&(pTsdbReadHandle->rhelper), pBlock, pCheckInfo->pCompInfo, colIds,
                                      (int)(QH_GET_NUM_OF_COLS(pTsdbReadHandle)), true);
H
Haojun Liao 已提交
1229
  if (ret != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
1230 1231 1232
    int32_t c = terrno;
    assert(c != TSDB_CODE_SUCCESS);
    goto _error;
H
Haojun Liao 已提交
1233
  }
1234

1235
  SDataBlockLoadInfo* pBlockLoadInfo = &pTsdbReadHandle->dataBlockLoadInfo;
1236

1237 1238
  pBlockLoadInfo->fileGroup = pTsdbReadHandle->pFileGroup;
  pBlockLoadInfo->slot = pTsdbReadHandle->cur.slot;
H
Haojun Liao 已提交
1239
  pBlockLoadInfo->uid = pCheckInfo->tableId;
1240

1241
  SDataCols* pCols = pTsdbReadHandle->rhelper.pDCols[0];
1242
  assert(pCols->numOfRows != 0 && pCols->numOfRows <= pBlock->numOfRows);
1243

1244
  pBlock->numOfRows = pCols->numOfRows;
H
Haojun Liao 已提交
1245

1246
  // Convert from TKEY to TSKEY for primary timestamp column if current block has timestamp before 1970-01-01T00:00:00Z
dengyihao's avatar
dengyihao 已提交
1247
  if (pBlock->keyFirst < 0 && colIds[0] == PRIMARYKEY_TIMESTAMP_COL_ID) {
1248
    int64_t* src = pCols->cols[0].pData;
dengyihao's avatar
dengyihao 已提交
1249
    for (int32_t i = 0; i < pBlock->numOfRows; ++i) {
1250 1251 1252 1253
      src[i] = tdGetKey(src[i]);
    }
  }

H
Haojun Liao 已提交
1254
  int64_t elapsedTime = (taosGetTimestampUs() - st);
1255
  pTsdbReadHandle->cost.blockLoadTime += elapsedTime;
1256

dengyihao's avatar
dengyihao 已提交
1257 1258 1259 1260
  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 已提交
1261
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1262 1263 1264 1265

_error:
  pBlock->numOfRows = 0;

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

1271
static int32_t getEndPosInDataBlock(STsdbReadHandle* pTsdbReadHandle, SDataBlockInfo* pBlockInfo);
dengyihao's avatar
dengyihao 已提交
1272 1273 1274 1275 1276 1277 1278
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) {
1279
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
H
Hongze Cheng 已提交
1280
  STsdbCfg*      pCfg = REPO_CFG(pTsdbReadHandle->pTsdb);
H
Haojun Liao 已提交
1281
  SDataBlockInfo binfo = GET_FILE_DATA_BLOCK_INFO(pCheckInfo, pBlock);
1282
  TSKEY          key;
dengyihao's avatar
dengyihao 已提交
1283
  int32_t        code = TSDB_CODE_SUCCESS;
1284

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

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

H
Haojun Liao 已提交
1290
  if (key != TSKEY_INITIAL_VAL) {
dengyihao's avatar
dengyihao 已提交
1291
    tsdbDebug("%p key in mem:%" PRId64 ", %s", pTsdbReadHandle, key, pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
1292
  } else {
H
Haojun Liao 已提交
1293
    tsdbDebug("%p no data in mem, %s", pTsdbReadHandle, pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
1294
  }
H
Haojun Liao 已提交
1295

1296 1297
  bool ascScan = ASCENDING_TRAVERSE(pTsdbReadHandle->order);

1298 1299 1300 1301
  if ((ascScan && (key != TSKEY_INITIAL_VAL && key <= binfo.window.ekey)) ||
      (!ascScan && (key != TSKEY_INITIAL_VAL && key >= binfo.window.skey))) {
    if ((ascScan && (key != TSKEY_INITIAL_VAL && key < binfo.window.skey)) ||
        (!ascScan && (key != TSKEY_INITIAL_VAL && key > binfo.window.ekey))) {
H
Haojun Liao 已提交
1302
      // do not load file block into buffer
1303
      int32_t step = ascScan ? 1 : -1;
H
Haojun Liao 已提交
1304

dengyihao's avatar
dengyihao 已提交
1305 1306 1307 1308
      TSKEY maxKey =
          ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? (binfo.window.skey - step) : (binfo.window.ekey - step);
      cur->rows =
          tsdbReadRowsFromCache(pCheckInfo, maxKey, pTsdbReadHandle->outputCapacity, &cur->win, pTsdbReadHandle);
1309
      pTsdbReadHandle->realNumOfRows = cur->rows;
H
Haojun Liao 已提交
1310 1311 1312

      // update the last key value
      pCheckInfo->lastKey = cur->win.ekey + step;
1313
      if (!ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
wafwerar's avatar
wafwerar 已提交
1314
        TSWAP(cur->win.skey, cur->win.ekey);
H
Haojun Liao 已提交
1315
      }
H
Haojun Liao 已提交
1316

H
Haojun Liao 已提交
1317 1318
      cur->mixBlock = true;
      cur->blockCompleted = false;
H
Haojun Liao 已提交
1319
      return code;
H
Haojun Liao 已提交
1320
    }
H
Haojun Liao 已提交
1321

1322
    // return error, add test cases
1323
    if ((code = doLoadFileDataBlock(pTsdbReadHandle, pBlock, pCheckInfo, cur->slot)) != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
1324
      return code;
1325 1326
    }

1327
    doMergeTwoLevelData(pTsdbReadHandle, pCheckInfo, pBlock);
1328
  } else {
1329 1330 1331 1332 1333 1334
    /*
     * 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
     */
1335 1336
    assert(pTsdbReadHandle->outputCapacity >= binfo.rows);
    int32_t endPos = getEndPosInDataBlock(pTsdbReadHandle, &binfo);
1337

1338 1339
    if ((cur->pos == 0 && endPos == binfo.rows - 1 && ascScan) ||
        (cur->pos == (binfo.rows - 1) && endPos == 0 && (!ascScan))) {
1340
      pTsdbReadHandle->realNumOfRows = binfo.rows;
1341 1342

      cur->rows = binfo.rows;
dengyihao's avatar
dengyihao 已提交
1343
      cur->win = binfo.window;
1344
      cur->mixBlock = false;
H
Haojun Liao 已提交
1345 1346
      cur->blockCompleted = true;

1347
      if (ascScan) {
H
Haojun Liao 已提交
1348 1349 1350 1351 1352 1353
        cur->lastKey = binfo.window.ekey + 1;
        cur->pos = binfo.rows;
      } else {
        cur->lastKey = binfo.window.skey - 1;
        cur->pos = -1;
      }
dengyihao's avatar
dengyihao 已提交
1354
    } else {  // partially copy to dest buffer
1355
      copyAllRemainRowsFromFileBlock(pTsdbReadHandle, pCheckInfo, &binfo, endPos);
1356 1357
      cur->mixBlock = true;
    }
1358

H
Haojun Liao 已提交
1359
    assert(cur->blockCompleted);
H
Haojun Liao 已提交
1360
    if (cur->rows == binfo.rows) {
dengyihao's avatar
dengyihao 已提交
1361
      tsdbDebug("%p whole file block qualified, brange:%" PRId64 "-%" PRId64 ", rows:%d, lastKey:%" PRId64 ", %s",
H
Haojun Liao 已提交
1362
                pTsdbReadHandle, cur->win.skey, cur->win.ekey, cur->rows, cur->lastKey, pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
1363
    } else {
dengyihao's avatar
dengyihao 已提交
1364 1365 1366 1367
      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 已提交
1368
    }
1369
  }
H
Haojun Liao 已提交
1370 1371

  return code;
1372 1373
}

1374 1375
static int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order);

dengyihao's avatar
dengyihao 已提交
1376 1377
static int32_t loadFileDataBlock(STsdbReadHandle* pTsdbReadHandle, SBlock* pBlock, STableCheckInfo* pCheckInfo,
                                 bool* exists) {
1378
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
dengyihao's avatar
dengyihao 已提交
1379 1380
  int32_t        code = TSDB_CODE_SUCCESS;
  bool           asc = ASCENDING_TRAVERSE(pTsdbReadHandle->order);
1381

1382
  if (asc) {
H
Haojun Liao 已提交
1383
    // query ended in/started from current block
1384 1385
    if (pTsdbReadHandle->window.ekey < pBlock->keyLast || pCheckInfo->lastKey > pBlock->keyFirst) {
      if ((code = doLoadFileDataBlock(pTsdbReadHandle, pBlock, pCheckInfo, cur->slot)) != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
1386 1387
        *exists = false;
        return code;
1388
      }
1389

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

1393 1394
      if (pCheckInfo->lastKey > pBlock->keyFirst) {
        cur->pos =
1395
            binarySearchForKey(pTSCol->cols[0].pData, pBlock->numOfRows, pCheckInfo->lastKey, pTsdbReadHandle->order);
1396 1397 1398
      } else {
        cur->pos = 0;
      }
H
Haojun Liao 已提交
1399

H
Haojun Liao 已提交
1400
      assert(pCheckInfo->lastKey <= pBlock->keyLast);
1401
      doMergeTwoLevelData(pTsdbReadHandle, pCheckInfo, pBlock);
1402
    } else {  // the whole block is loaded in to buffer
dengyihao's avatar
dengyihao 已提交
1403
      cur->pos = asc ? 0 : (pBlock->numOfRows - 1);
1404
      code = handleDataMergeIfNeeded(pTsdbReadHandle, pBlock, pCheckInfo);
H
[td-32]  
hjxilinx 已提交
1405
    }
dengyihao's avatar
dengyihao 已提交
1406
  } else {  // desc order, query ended in current block
1407 1408
    if (pTsdbReadHandle->window.ekey > pBlock->keyFirst || pCheckInfo->lastKey < pBlock->keyLast) {
      if ((code = doLoadFileDataBlock(pTsdbReadHandle, pBlock, pCheckInfo, cur->slot)) != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
1409 1410
        *exists = false;
        return code;
1411
      }
H
Haojun Liao 已提交
1412

1413
      SDataCols* pTsCol = pTsdbReadHandle->rhelper.pDCols[0];
1414
      if (pCheckInfo->lastKey < pBlock->keyLast) {
dengyihao's avatar
dengyihao 已提交
1415 1416
        cur->pos =
            binarySearchForKey(pTsCol->cols[0].pData, pBlock->numOfRows, pCheckInfo->lastKey, pTsdbReadHandle->order);
1417
      } else {
H
Haojun Liao 已提交
1418
        cur->pos = pBlock->numOfRows - 1;
1419
      }
H
Haojun Liao 已提交
1420

H
Haojun Liao 已提交
1421
      assert(pCheckInfo->lastKey >= pBlock->keyFirst);
1422
      doMergeTwoLevelData(pTsdbReadHandle, pCheckInfo, pBlock);
1423
    } else {
dengyihao's avatar
dengyihao 已提交
1424
      cur->pos = asc ? 0 : (pBlock->numOfRows - 1);
1425
      code = handleDataMergeIfNeeded(pTsdbReadHandle, pBlock, pCheckInfo);
H
[td-32]  
hjxilinx 已提交
1426
    }
1427
  }
1428

1429
  *exists = pTsdbReadHandle->realNumOfRows > 0;
H
Haojun Liao 已提交
1430
  return code;
H
[td-32]  
hjxilinx 已提交
1431 1432
}

1433
static int doBinarySearchKey(char* pValue, int num, TSKEY key, int order) {
1434
  int    firstPos, lastPos, midPos = -1;
H
Haojun Liao 已提交
1435
  int    numOfRows;
1436 1437
  TSKEY* keyList;

1438
  assert(order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC);
H
Haojun Liao 已提交
1439

1440
  if (num <= 0) return -1;
1441 1442

  keyList = (TSKEY*)pValue;
1443 1444
  firstPos = 0;
  lastPos = num - 1;
1445

1446
  if (order == TSDB_ORDER_DESC) {
1447 1448 1449 1450 1451
    // 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;
1452

H
Haojun Liao 已提交
1453 1454
      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1) + firstPos;
1455

1456 1457 1458 1459 1460 1461 1462 1463
      if (key < keyList[midPos]) {
        lastPos = midPos - 1;
      } else if (key > keyList[midPos]) {
        firstPos = midPos + 1;
      } else {
        break;
      }
    }
1464

1465 1466 1467 1468 1469
  } 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;
1470

1471 1472 1473 1474 1475 1476 1477
      if (key > keyList[lastPos]) {
        lastPos = lastPos + 1;
        if (lastPos >= num)
          return -1;
        else
          return lastPos;
      }
1478

H
Haojun Liao 已提交
1479 1480
      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1) + firstPos;
1481

1482 1483 1484 1485 1486 1487 1488 1489 1490
      if (key < keyList[midPos]) {
        lastPos = midPos - 1;
      } else if (key > keyList[midPos]) {
        firstPos = midPos + 1;
      } else {
        break;
      }
    }
  }
1491

1492 1493 1494
  return midPos;
}

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

1500
  int32_t num = end - start + 1;
H
Haojun Liao 已提交
1501 1502 1503 1504 1505 1506
  assert(num >= 0);

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

1507 1508
  bool    ascScan = ASCENDING_TRAVERSE(pTsdbReadHandle->order);
  int32_t trueStart = ascScan ? start : end;
1509
  int32_t trueEnd = ascScan ? end : start;
1510 1511
  int32_t step = ascScan ? 1 : -1;

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

dengyihao's avatar
dengyihao 已提交
1514
  // data in buffer has greater timestamp, copy data in file block
1515
  int32_t i = 0, j = 0;
dengyihao's avatar
dengyihao 已提交
1516
  while (i < requiredNumOfCols && j < pCols->numOfCols) {
1517
    SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
1518 1519 1520 1521 1522 1523 1524

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

L
Liu Jicong 已提交
1525
    if (!isAllRowsNull(src) && pColInfo->info.colId == src->colId) {
1526
      if (!IS_VAR_DATA_TYPE(pColInfo->info.type)) {  // todo opt performance
dengyihao's avatar
dengyihao 已提交
1527
        //        memmove(pData, (char*)src->pData + bytes * start, bytes * num);
1528
        int32_t rowIndex = numOfRows;
1529
        for (int32_t k = trueStart; ((ascScan && k <= trueEnd) || (!ascScan && k >= trueEnd)); k += step, ++rowIndex) {
1530
          SCellVal sVal = {0};
C
Cary Xu 已提交
1531
          if (tdGetColDataOfRow(&sVal, src, k, pCols->bitmapMode) < 0) {
1532 1533
            TASSERT(0);
          }
1534

C
Cary Xu 已提交
1535
          if (sVal.valType == TD_VTYPE_NORM) {
1536
            colDataAppend(pColInfo, rowIndex, sVal.val, false);
C
Cary Xu 已提交
1537 1538
          } else {
            colDataAppendNULL(pColInfo, rowIndex);
1539 1540 1541
          }
        }
      } else {  // handle the var-string
1542 1543
        int32_t rowIndex = numOfRows;

1544
        // todo refactor, only copy one-by-one
1545
        for (int32_t k = trueStart; ((ascScan && k <= trueEnd) || (!ascScan && k >= trueEnd)); k += step, ++rowIndex) {
1546
          SCellVal sVal = {0};
dengyihao's avatar
dengyihao 已提交
1547
          if (tdGetColDataOfRow(&sVal, src, k, pCols->bitmapMode) < 0) {
H
Haojun Liao 已提交
1548 1549
            TASSERT(0);
          }
1550

C
Cary Xu 已提交
1551
          if (sVal.valType == TD_VTYPE_NORM) {
1552
            colDataAppend(pColInfo, rowIndex, sVal.val, false);
C
Cary Xu 已提交
1553 1554
          } else {
            colDataAppendNULL(pColInfo, rowIndex);
1555
          }
1556 1557
        }
      }
1558 1559 1560

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

dengyihao's avatar
dengyihao 已提交
1567
  while (i < requiredNumOfCols) {  // the remain columns are all null data
1568
    SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
1569
    colDataAppendNNULL(pColInfo, numOfRows, num);
1570
    i++;
1571
  }
H
Haojun Liao 已提交
1572

1573 1574
  pTsdbReadHandle->cur.win.ekey = tsArray[trueEnd];
  pTsdbReadHandle->cur.lastKey = tsArray[trueEnd] + step;
1575

1576
  return numOfRows + num;
1577 1578
}

C
Cary Xu 已提交
1579 1580 1581 1582 1583 1584 1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598
/**
 * @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 已提交
1599
#if 1
dengyihao's avatar
dengyihao 已提交
1600 1601 1602 1603 1604 1605 1606 1607 1608
  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 已提交
1609
  SCellVal sVal = {0};
C
Cary Xu 已提交
1610 1611
  TSKEY    rowKey = TSKEY_INITIAL_VAL;
  int32_t  nResult = 0;
C
Cary Xu 已提交
1612
  int32_t  mergeOption = 0;  // 0 discard 1 overwrite 2 merge
1613

H
Haojun Liao 已提交
1614
  // the schema version info is embeded in STSRow
1615 1616 1617
  int32_t numOfColsOfRow1 = 0;

  if (pSchema1 == NULL) {
C
Cary Xu 已提交
1618 1619
    // pSchema1 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, TD_ROW_SVER(row1));
    // TODO: use the real schemaVersion
H
Hongze Cheng 已提交
1620
    pSchema1 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, 1);
1621
  }
1622

C
Cary Xu 已提交
1623 1624 1625 1626
#ifdef TD_DEBUG_PRINT_ROW
  tdSRowPrint(row1, pSchema1, __func__);
#endif

dengyihao's avatar
dengyihao 已提交
1627
  if (isRow1DataRow) {
1628
    numOfColsOfRow1 = schemaNCols(pSchema1);
H
Haojun Liao 已提交
1629
  } else {
H
Haojun Liao 已提交
1630
    numOfColsOfRow1 = tdRowGetNCols(row1);
D
fix bug  
dapan1121 已提交
1631
  }
1632

1633
  int32_t numOfColsOfRow2 = 0;
dengyihao's avatar
dengyihao 已提交
1634
  if (row2) {
H
Haojun Liao 已提交
1635
    isRow2DataRow = TD_IS_TP_ROW(row2);
1636
    if (pSchema2 == NULL) {
C
Cary Xu 已提交
1637 1638
      // pSchema2 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, TD_ROW_SVER(row2));
      // TODO: use the real schemaVersion
H
Hongze Cheng 已提交
1639
      pSchema2 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, 1);
1640
    }
dengyihao's avatar
dengyihao 已提交
1641
    if (isRow2DataRow) {
1642 1643
      numOfColsOfRow2 = schemaNCols(pSchema2);
    } else {
H
Haojun Liao 已提交
1644
      numOfColsOfRow2 = tdRowGetNCols(row2);
1645 1646
    }
  }
C
Cary Xu 已提交
1647

1648
  int32_t i = 0, j = 0, k = 0;
dengyihao's avatar
dengyihao 已提交
1649
  while (i < numOfCols && (j < numOfColsOfRow1 || k < numOfColsOfRow2)) {
1650
    SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
1651 1652

    int32_t colIdOfRow1;
dengyihao's avatar
dengyihao 已提交
1653
    if (j >= numOfColsOfRow1) {
1654
      colIdOfRow1 = INT32_MAX;
dengyihao's avatar
dengyihao 已提交
1655
    } else if (isRow1DataRow) {
1656 1657
      colIdOfRow1 = pSchema1->columns[j].colId;
    } else {
C
Cary Xu 已提交
1658
      colIdOfRow1 = tdKvRowColIdAt(row1, j);
1659 1660 1661
    }

    int32_t colIdOfRow2;
dengyihao's avatar
dengyihao 已提交
1662
    if (k >= numOfColsOfRow2) {
1663
      colIdOfRow2 = INT32_MAX;
dengyihao's avatar
dengyihao 已提交
1664
    } else if (isRow2DataRow) {
1665 1666
      colIdOfRow2 = pSchema2->columns[k].colId;
    } else {
C
Cary Xu 已提交
1667
      colIdOfRow2 = tdKvRowColIdAt(row2, k);
1668 1669
    }

C
Cary Xu 已提交
1670
    if (colIdOfRow1 < colIdOfRow2) {  // the most probability
dengyihao's avatar
dengyihao 已提交
1671
      if (colIdOfRow1 < pColInfo->info.colId) {
C
Cary Xu 已提交
1672
        ++j;
C
Cary Xu 已提交
1673 1674
        continue;
      }
1675 1676 1677 1678
      row = row1;
      pSchema = pSchema1;
      isChosenRowDataRow = isRow1DataRow;
      chosen_itr = j;
C
Cary Xu 已提交
1679
    } else if (colIdOfRow1 == colIdOfRow2) {
dengyihao's avatar
dengyihao 已提交
1680
      if (colIdOfRow1 < pColInfo->info.colId) {
C
Cary Xu 已提交
1681 1682
        ++j;
        ++k;
1683
        continue;
C
Cary Xu 已提交
1684
      }
1685 1686 1687 1688 1689
      row = row1;
      pSchema = pSchema1;
      isChosenRowDataRow = isRow1DataRow;
      chosen_itr = j;
    } else {
dengyihao's avatar
dengyihao 已提交
1690
      if (colIdOfRow2 < pColInfo->info.colId) {
C
Cary Xu 已提交
1691
        ++k;
1692 1693 1694 1695 1696 1697 1698
        continue;
      }
      row = row2;
      pSchema = pSchema2;
      chosen_itr = k;
      isChosenRowDataRow = isRow2DataRow;
    }
C
Cary Xu 已提交
1699

dengyihao's avatar
dengyihao 已提交
1700
    if (isChosenRowDataRow) {
1701 1702
      colId = pSchema->columns[chosen_itr].colId;
      offset = pSchema->columns[chosen_itr].offset;
C
Cary Xu 已提交
1703 1704
      // TODO: use STSRowIter
      tdSTpRowGetVal(row, colId, pSchema->columns[chosen_itr].type, pSchema->flen, offset, chosen_itr - 1, &sVal);
C
Cary Xu 已提交
1705 1706 1707
      if (colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
        rowKey = *(TSKEY*)sVal.val;
        if (rowKey != *lastRowKey) {
C
Cary Xu 已提交
1708
          mergeOption = 1;
C
Cary Xu 已提交
1709 1710 1711 1712
          if (*lastRowKey != TSKEY_INITIAL_VAL) {
            ++(*curRow);
          }
          ++nResult;
C
Cary Xu 已提交
1713
        } else if (update) {
C
Cary Xu 已提交
1714 1715 1716 1717
          mergeOption = 2;
        } else {
          mergeOption = 0;
          break;
C
Cary Xu 已提交
1718
        }
C
Cary Xu 已提交
1719

C
Cary Xu 已提交
1720 1721
        *lastRowKey = rowKey;
      }
1722
    } else {
C
Cary Xu 已提交
1723 1724 1725 1726
      // TODO: use STSRowIter
      if (chosen_itr == 0) {
        colId = PRIMARYKEY_TIMESTAMP_COL_ID;
        tdSKvRowGetVal(row, PRIMARYKEY_TIMESTAMP_COL_ID, -1, -1, &sVal);
C
Cary Xu 已提交
1727 1728
        rowKey = *(TSKEY*)sVal.val;
        if (rowKey != *lastRowKey) {
C
Cary Xu 已提交
1729
          mergeOption = 1;
C
Cary Xu 已提交
1730 1731 1732 1733
          if (*lastRowKey != TSKEY_INITIAL_VAL) {
            ++(*curRow);
          }
          ++nResult;
C
Cary Xu 已提交
1734
        } else if (update) {
C
Cary Xu 已提交
1735 1736 1737 1738
          mergeOption = 2;
        } else {
          mergeOption = 0;
          break;
C
Cary Xu 已提交
1739 1740
        }
        *lastRowKey = rowKey;
C
Cary Xu 已提交
1741 1742 1743 1744 1745 1746
      } else {
        SKvRowIdx* pColIdx = tdKvRowColIdxAt(row, chosen_itr - 1);
        colId = pColIdx->colId;
        offset = pColIdx->offset;
        tdSKvRowGetVal(row, colId, offset, chosen_itr - 1, &sVal);
      }
1747 1748
    }

C
Cary Xu 已提交
1749 1750
    ASSERT(rowKey != TSKEY_INITIAL_VAL);

1751
    if (colId == pColInfo->info.colId) {
H
Haojun Liao 已提交
1752
      if (tdValTypeIsNorm(sVal.valType)) {
C
Cary Xu 已提交
1753 1754 1755 1756 1757
        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 已提交
1758
        if (mergeOption == 1) {
C
Cary Xu 已提交
1759 1760 1761 1762
          colDataAppend(pColInfo, *curRow, NULL, true);
        }
      } else {
        ASSERT(0);
1763
      }
H
Haojun Liao 已提交
1764

C
Cary Xu 已提交
1765
      ++i;
C
Cary Xu 已提交
1766

dengyihao's avatar
dengyihao 已提交
1767
      if (row == row1) {
C
Cary Xu 已提交
1768
        ++j;
1769
      } else {
C
Cary Xu 已提交
1770
        ++k;
1771 1772
      }
    } else {
C
Cary Xu 已提交
1773
      if (mergeOption == 1) {
C
Cary Xu 已提交
1774
        colDataAppend(pColInfo, *curRow, NULL, true);
C
Cary Xu 已提交
1775
      }
C
Cary Xu 已提交
1776
      ++i;
1777
    }
1778
  }
1779

C
Cary Xu 已提交
1780
  if (mergeOption == 1) {
dengyihao's avatar
dengyihao 已提交
1781
    while (i < numOfCols) {  // the remain columns are all null data
1782
      SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
C
Cary Xu 已提交
1783 1784
      colDataAppend(pColInfo, *curRow, NULL, true);
      ++i;
1785 1786
    }
  }
C
Cary Xu 已提交
1787 1788

  return nResult;
H
Haojun Liao 已提交
1789
#endif
1790
}
1791

dengyihao's avatar
dengyihao 已提交
1792 1793
static void getQualifiedRowsPos(STsdbReadHandle* pTsdbReadHandle, int32_t startPos, int32_t endPos,
                                int32_t numOfExisted, int32_t* start, int32_t* end) {
1794 1795
  *start = -1;

1796
  if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
1797
    int32_t remain = endPos - startPos + 1;
1798 1799
    if (remain + numOfExisted > pTsdbReadHandle->outputCapacity) {
      *end = (pTsdbReadHandle->outputCapacity - numOfExisted) + startPos - 1;
H
Haojun Liao 已提交
1800 1801
    } else {
      *end = endPos;
1802 1803 1804 1805 1806
    }

    *start = startPos;
  } else {
    int32_t remain = (startPos - endPos) + 1;
1807 1808
    if (remain + numOfExisted > pTsdbReadHandle->outputCapacity) {
      *end = startPos + 1 - (pTsdbReadHandle->outputCapacity - numOfExisted);
H
Haojun Liao 已提交
1809 1810
    } else {
      *end = endPos;
1811 1812 1813 1814 1815 1816 1817
    }

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

dengyihao's avatar
dengyihao 已提交
1818 1819
static void updateInfoAfterMerge(STsdbReadHandle* pTsdbReadHandle, STableCheckInfo* pCheckInfo, int32_t numOfRows,
                                 int32_t endPos) {
1820
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
1821 1822

  pCheckInfo->lastKey = cur->lastKey;
1823
  pTsdbReadHandle->realNumOfRows = numOfRows;
1824 1825 1826 1827
  cur->rows = numOfRows;
  cur->pos = endPos;
}

1828 1829
static void doCheckGeneratedBlockRange(STsdbReadHandle* pTsdbReadHandle) {
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
H
Haojun Liao 已提交
1830 1831

  if (cur->rows > 0) {
1832 1833
    if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
      assert(cur->win.skey >= pTsdbReadHandle->window.skey && cur->win.ekey <= pTsdbReadHandle->window.ekey);
H
Haojun Liao 已提交
1834
    } else {
1835
      assert(cur->win.skey >= pTsdbReadHandle->window.ekey && cur->win.ekey <= pTsdbReadHandle->window.skey);
H
Haojun Liao 已提交
1836 1837
    }

1838
    SColumnInfoData* pColInfoData = taosArrayGet(pTsdbReadHandle->pColumns, 0);
dengyihao's avatar
dengyihao 已提交
1839 1840
    assert(cur->win.skey == ((TSKEY*)pColInfoData->pData)[0] &&
           cur->win.ekey == ((TSKEY*)pColInfoData->pData)[cur->rows - 1]);
H
Haojun Liao 已提交
1841
  } else {
1842
    cur->win = pTsdbReadHandle->window;
H
Haojun Liao 已提交
1843

dengyihao's avatar
dengyihao 已提交
1844
    int32_t step = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 1 : -1;
1845
    cur->lastKey = pTsdbReadHandle->window.ekey + step;
H
Haojun Liao 已提交
1846 1847 1848
  }
}

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

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

dengyihao's avatar
dengyihao 已提交
1856
  int32_t step = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 1 : -1;
1857
  int32_t numOfCols = (int32_t)(QH_GET_NUM_OF_COLS(pTsdbReadHandle));
H
Haojun Liao 已提交
1858 1859 1860 1861 1862 1863

  int32_t pos = cur->pos;

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

1864
  if (!ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
wafwerar's avatar
wafwerar 已提交
1865
    TSWAP(start, end);
H
Haojun Liao 已提交
1866 1867
  }

1868 1869
  assert(pTsdbReadHandle->outputCapacity >= (end - start + 1));
  int32_t numOfRows = doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, 0, start, end);
H
Haojun Liao 已提交
1870 1871

  // the time window should always be ascending order: skey <= ekey
dengyihao's avatar
dengyihao 已提交
1872
  cur->win = (STimeWindow){.skey = tsArray[start], .ekey = tsArray[end]};
H
Haojun Liao 已提交
1873
  cur->mixBlock = (numOfRows != pBlockInfo->rows);
H
Haojun Liao 已提交
1874
  cur->lastKey = tsArray[endPos] + step;
H
Haojun Liao 已提交
1875
  cur->blockCompleted = true;
H
Haojun Liao 已提交
1876

H
Haojun Liao 已提交
1877 1878
  // The value of pos may be -1 or pBlockInfo->rows, and it is invalid in both cases.
  pos = endPos + step;
1879 1880
  updateInfoAfterMerge(pTsdbReadHandle, pCheckInfo, numOfRows, pos);
  doCheckGeneratedBlockRange(pTsdbReadHandle);
H
Haojun Liao 已提交
1881

dengyihao's avatar
dengyihao 已提交
1882 1883 1884
  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 已提交
1885 1886
}

1887
int32_t getEndPosInDataBlock(STsdbReadHandle* pTsdbReadHandle, SDataBlockInfo* pBlockInfo) {
H
Haojun Liao 已提交
1888 1889
  // NOTE: reverse the order to find the end position in data block
  int32_t endPos = -1;
dengyihao's avatar
dengyihao 已提交
1890
  int32_t order = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? TSDB_ORDER_DESC : TSDB_ORDER_ASC;
H
Haojun Liao 已提交
1891

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

1895
  if (ASCENDING_TRAVERSE(pTsdbReadHandle->order) && pTsdbReadHandle->window.ekey >= pBlockInfo->window.ekey) {
H
Haojun Liao 已提交
1896 1897
    endPos = pBlockInfo->rows - 1;
    cur->mixBlock = (cur->pos != 0);
1898
  } else if (!ASCENDING_TRAVERSE(pTsdbReadHandle->order) && pTsdbReadHandle->window.ekey <= pBlockInfo->window.skey) {
H
Haojun Liao 已提交
1899 1900 1901 1902
    endPos = 0;
    cur->mixBlock = (cur->pos != pBlockInfo->rows - 1);
  } else {
    assert(pCols->numOfRows > 0);
1903
    endPos = doBinarySearchKey(pCols->cols[0].pData, pCols->numOfRows, pTsdbReadHandle->window.ekey, order);
H
Haojun Liao 已提交
1904 1905 1906 1907 1908 1909
    cur->mixBlock = true;
  }

  return endPos;
}

H
[td-32]  
hjxilinx 已提交
1910 1911
// 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
1912 1913
static void doMergeTwoLevelData(STsdbReadHandle* pTsdbReadHandle, STableCheckInfo* pCheckInfo, SBlock* pBlock) {
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
1914
  SDataBlockInfo blockInfo = GET_FILE_DATA_BLOCK_INFO(pCheckInfo, pBlock);
H
Hongze Cheng 已提交
1915
  STsdbCfg*      pCfg = REPO_CFG(pTsdbReadHandle->pTsdb);
H
Haojun Liao 已提交
1916

1917
  initTableMemIterator(pTsdbReadHandle, pCheckInfo);
1918

1919 1920
  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 已提交
1921
         cur->pos >= 0 && cur->pos < pBlock->numOfRows);
H
Haojun Liao 已提交
1922

1923
  TSKEY* tsArray = pCols->cols[0].pData;
dengyihao's avatar
dengyihao 已提交
1924 1925
  assert(pCols->numOfRows == pBlock->numOfRows && tsArray[0] == pBlock->keyFirst &&
         tsArray[pBlock->numOfRows - 1] == pBlock->keyLast);
1926

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

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

1933 1934
  int32_t numOfCols = (int32_t)(QH_GET_NUM_OF_COLS(pTsdbReadHandle));
  int32_t endPos = getEndPosInDataBlock(pTsdbReadHandle, &blockInfo);
H
Haojun Liao 已提交
1935

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

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

dengyihao's avatar
dengyihao 已提交
1945 1946
  int16_t   rv1 = -1;
  int16_t   rv2 = -1;
1947 1948
  STSchema* pSchema1 = NULL;
  STSchema* pSchema2 = NULL;
D
fix bug  
dapan1121 已提交
1949

H
Haojun Liao 已提交
1950 1951
  int32_t pos = cur->pos;
  cur->win = TSWINDOW_INITIALIZER;
1952

1953 1954
  // no data in buffer, load data from file directly
  if (pCheckInfo->iiter == NULL && pCheckInfo->iter == NULL) {
1955
    copyAllRemainRowsFromFileBlock(pTsdbReadHandle, pCheckInfo, &blockInfo, endPos);
1956
    return;
1957
  } else if (pCheckInfo->iter != NULL || pCheckInfo->iiter != NULL) {
1958
    SSkipListNode* node = NULL;
C
Cary Xu 已提交
1959
    TSKEY          lastKeyAppend = TSKEY_INITIAL_VAL;
C
Cary Xu 已提交
1960

1961
    do {
H
Haojun Liao 已提交
1962
      STSRow* row2 = NULL;
C
Cary Xu 已提交
1963
      STSRow* row1 = getSRowInTableMem(pCheckInfo, pTsdbReadHandle->order, pCfg->update, &row2, TD_VER_MAX);
1964
      if (row1 == NULL) {
H
[td-32]  
hjxilinx 已提交
1965
        break;
1966
      }
1967

H
Haojun Liao 已提交
1968
      TSKEY key = TD_ROW_KEY(row1);
1969
      if ((key > pTsdbReadHandle->window.ekey && ascScan) || (key < pTsdbReadHandle->window.ekey && !ascScan)) {
1970 1971 1972
        break;
      }

1973 1974
      if (((pos > endPos || tsArray[pos] > pTsdbReadHandle->window.ekey) && ascScan) ||
          ((pos < endPos || tsArray[pos] < pTsdbReadHandle->window.ekey) && !ascScan)) {
1975 1976 1977
        break;
      }

1978
      if ((key < tsArray[pos] && ascScan) || (key > tsArray[pos] && !ascScan)) {
H
Haojun Liao 已提交
1979
        if (rv1 != TD_ROW_SVER(row1)) {
dengyihao's avatar
dengyihao 已提交
1980
          //          pSchema1 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row1));
H
Haojun Liao 已提交
1981
          rv1 = TD_ROW_SVER(row1);
C
Cary Xu 已提交
1982
        }
dengyihao's avatar
dengyihao 已提交
1983 1984
        if (row2 && rv2 != TD_ROW_SVER(row2)) {
          //          pSchema2 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row2));
H
Haojun Liao 已提交
1985
          rv2 = TD_ROW_SVER(row2);
1986
        }
dengyihao's avatar
dengyihao 已提交
1987

C
Cary Xu 已提交
1988 1989 1990
        numOfRows +=
            mergeTwoRowFromMem(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, &curRow, row1, row2, numOfCols,
                               pCheckInfo->tableId, pSchema1, pSchema2, pCfg->update, &lastKeyAppend);
1991 1992 1993
        if (cur->win.skey == TSKEY_INITIAL_VAL) {
          cur->win.skey = key;
        }
1994

1995
        cur->win.ekey = key;
dengyihao's avatar
dengyihao 已提交
1996
        cur->lastKey = key + step;
1997
        cur->mixBlock = true;
1998
        moveToNextRowInMem(pCheckInfo);
1999
      } else if (key == tsArray[pos]) {  // data in buffer has the same timestamp of data in file block, ignore it
C
Cary Xu 已提交
2000
#if 0
H
TD-1439  
Hongze Cheng 已提交
2001
        if (pCfg->update) {
dengyihao's avatar
dengyihao 已提交
2002
          if (pCfg->update == TD_ROW_PARTIAL_UPDATE) {
2003
            doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, numOfRows, pos, pos);
D
fix bug  
dapan1121 已提交
2004
          }
H
Haojun Liao 已提交
2005
          if (rv1 != TD_ROW_SVER(row1)) {
dengyihao's avatar
dengyihao 已提交
2006
            //            pSchema1 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row1));
H
Haojun Liao 已提交
2007
            rv1 = TD_ROW_SVER(row1);
2008
          }
dengyihao's avatar
dengyihao 已提交
2009 2010
          if (row2 && rv2 != TD_ROW_SVER(row2)) {
            //            pSchema2 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row2));
H
Haojun Liao 已提交
2011
            rv2 = TD_ROW_SVER(row2);
2012
          }
dengyihao's avatar
dengyihao 已提交
2013

2014
          bool forceSetNull = pCfg->update != TD_ROW_PARTIAL_UPDATE;
dengyihao's avatar
dengyihao 已提交
2015
          mergeTwoRowFromMem(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, numOfRows, row1, row2, numOfCols,
C
Cary Xu 已提交
2016
                             pCheckInfo->tableId, pSchema1, pSchema2, forceSetNull, &lastRowKey);
H
TD-1439  
Hongze Cheng 已提交
2017 2018 2019 2020 2021 2022 2023 2024 2025
          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 已提交
2026 2027 2028 2029 2030 2031 2032
          moveToNextRowInMem(pCheckInfo);
          pos += step;
        } else {
          moveToNextRowInMem(pCheckInfo);
        }
#endif
        if (TD_SUPPORT_UPDATE(pCfg->update)) {
C
Cary Xu 已提交
2033
          numOfRows = doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, curRow, pos, pos);
2034
          lastKeyAppend = key;
C
Cary Xu 已提交
2035 2036 2037 2038 2039 2040 2041

          if (rv1 != TD_ROW_SVER(row1)) {
            rv1 = TD_ROW_SVER(row1);
          }
          if (row2 && rv2 != TD_ROW_SVER(row2)) {
            rv2 = TD_ROW_SVER(row2);
          }
2042 2043 2044 2045

          // still assign data into current row
          mergeTwoRowFromMem(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, &curRow, row1, row2, numOfCols,
                             pCheckInfo->tableId, pSchema1, pSchema2, pCfg->update, &lastKeyAppend);
C
Cary Xu 已提交
2046 2047 2048 2049 2050 2051 2052 2053 2054

          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 已提交
2055
          moveToNextRowInMem(pCheckInfo);
2056 2057
          ++curRow;

H
TD-1439  
Hongze Cheng 已提交
2058 2059 2060 2061
          pos += step;
        } else {
          moveToNextRowInMem(pCheckInfo);
        }
2062
      } else if ((key > tsArray[pos] && ascScan) || (key < tsArray[pos] && !ascScan)) {
2063 2064 2065
        if (cur->win.skey == TSKEY_INITIAL_VAL) {
          cur->win.skey = tsArray[pos];
        }
2066

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

dengyihao's avatar
dengyihao 已提交
2070
        if (tsArray[end] == key) {  // the value of key in cache equals to the end timestamp value, ignore it
C
Cary Xu 已提交
2071
#if 0
2072
          if (pCfg->update == TD_ROW_DISCARD_UPDATE) {
H
Hongze Cheng 已提交
2073 2074 2075 2076
            moveToNextRowInMem(pCheckInfo);
          } else {
            end -= step;
          }
C
Cary Xu 已提交
2077 2078 2079 2080 2081 2082
#endif
          if (!TD_SUPPORT_UPDATE(pCfg->update)) {
            moveToNextRowInMem(pCheckInfo);
          } else {
            end -= step;
          }
H
Haojun Liao 已提交
2083
        }
2084

2085
        int32_t qstart = 0, qend = 0;
2086
        getQualifiedRowsPos(pTsdbReadHandle, pos, end, numOfRows, &qstart, &qend);
2087

2088
        if ((lastKeyAppend != TSKEY_INITIAL_VAL) && (lastKeyAppend != (ascScan ? tsArray[qstart] : tsArray[qend]))) {
C
Cary Xu 已提交
2089 2090
          ++curRow;
        }
2091

C
Cary Xu 已提交
2092
        numOfRows = doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, curRow, qstart, qend);
2093
        pos += (qend - qstart + 1) * step;
C
Cary Xu 已提交
2094 2095 2096
        if (numOfRows > 0) {
          curRow = numOfRows - 1;
        }
2097

2098
        cur->win.ekey = ascScan ? tsArray[qend] : tsArray[qstart];
dengyihao's avatar
dengyihao 已提交
2099
        cur->lastKey = cur->win.ekey + step;
C
Cary Xu 已提交
2100
        lastKeyAppend = cur->win.ekey;
2101
      }
2102
    } while (numOfRows < pTsdbReadHandle->outputCapacity);
H
Haojun Liao 已提交
2103

2104
    if (numOfRows < pTsdbReadHandle->outputCapacity) {
H
Haojun Liao 已提交
2105 2106 2107 2108
      /**
       * 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.
       */
2109
      if (node == NULL ||
2110 2111
          ((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)) {
2112 2113 2114 2115 2116
        // 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];
        }

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

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

2123
        cur->win.ekey = ascScan ? tsArray[end] : tsArray[start];
dengyihao's avatar
dengyihao 已提交
2124
        cur->lastKey = cur->win.ekey + step;
H
Haojun Liao 已提交
2125
        cur->mixBlock = true;
2126
      }
2127 2128
    }
  }
H
Haojun Liao 已提交
2129

2130 2131
  cur->blockCompleted = (((pos > endPos || cur->lastKey > pTsdbReadHandle->window.ekey) && ascScan) ||
       ((pos < endPos || cur->lastKey < pTsdbReadHandle->window.ekey) && !ascScan));
2132

2133
  if (!ascScan) {
wafwerar's avatar
wafwerar 已提交
2134
    TSWAP(cur->win.skey, cur->win.ekey);
2135
  }
2136

2137 2138
  updateInfoAfterMerge(pTsdbReadHandle, pCheckInfo, numOfRows, pos);
  doCheckGeneratedBlockRange(pTsdbReadHandle);
H
Haojun Liao 已提交
2139

dengyihao's avatar
dengyihao 已提交
2140 2141 2142
  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);
2143 2144
}

2145
int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order) {
H
[td-32]  
hjxilinx 已提交
2146
  int    firstPos, lastPos, midPos = -1;
H
Haojun Liao 已提交
2147
  int    numOfRows;
2148 2149
  TSKEY* keyList;

H
[td-32]  
hjxilinx 已提交
2150
  if (num <= 0) return -1;
2151 2152

  keyList = (TSKEY*)pValue;
H
[td-32]  
hjxilinx 已提交
2153 2154
  firstPos = 0;
  lastPos = num - 1;
2155

2156
  if (order == TSDB_ORDER_DESC) {
H
[td-32]  
hjxilinx 已提交
2157 2158 2159 2160 2161
    // 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;
2162

H
Haojun Liao 已提交
2163 2164
      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1) + firstPos;
2165

H
[td-32]  
hjxilinx 已提交
2166 2167 2168 2169 2170 2171 2172 2173
      if (key < keyList[midPos]) {
        lastPos = midPos - 1;
      } else if (key > keyList[midPos]) {
        firstPos = midPos + 1;
      } else {
        break;
      }
    }
2174

H
[td-32]  
hjxilinx 已提交
2175 2176 2177 2178 2179
  } 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;
2180

H
[td-32]  
hjxilinx 已提交
2181 2182 2183 2184 2185 2186 2187
      if (key > keyList[lastPos]) {
        lastPos = lastPos + 1;
        if (lastPos >= num)
          return -1;
        else
          return lastPos;
      }
2188

H
Haojun Liao 已提交
2189 2190
      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1) + firstPos;
2191

H
[td-32]  
hjxilinx 已提交
2192 2193 2194 2195 2196 2197 2198 2199 2200
      if (key < keyList[midPos]) {
        lastPos = midPos - 1;
      } else if (key > keyList[midPos]) {
        firstPos = midPos + 1;
      } else {
        break;
      }
    }
  }
2201

H
[td-32]  
hjxilinx 已提交
2202 2203 2204
  return midPos;
}

2205
static void cleanBlockOrderSupporter(SBlockOrderSupporter* pSupporter, int32_t numOfTables) {
wafwerar's avatar
wafwerar 已提交
2206 2207
  taosMemoryFreeClear(pSupporter->numOfBlocksPerTable);
  taosMemoryFreeClear(pSupporter->blockIndexArray);
2208 2209

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

wafwerar's avatar
wafwerar 已提交
2214
  taosMemoryFreeClear(pSupporter->pDataBlockInfo);
2215 2216 2217 2218 2219 2220 2221 2222 2223 2224 2225
}

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

2226
  if (leftTableBlockIndex > pSupporter->numOfBlocksPerTable[leftTableIndex]) {
2227 2228
    /* left block is empty */
    return 1;
2229
  } else if (rightTableBlockIndex > pSupporter->numOfBlocksPerTable[rightTableIndex]) {
2230 2231 2232 2233 2234 2235 2236
    /* right block is empty */
    return -1;
  }

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

H
Haojun Liao 已提交
2237
  //    assert(pLeftBlockInfoEx->compBlock->offset != pRightBlockInfoEx->compBlock->offset);
dengyihao's avatar
dengyihao 已提交
2238
#if 0  // TODO: temporarily comment off requested by Dr. Liao
H
Haojun Liao 已提交
2239 2240
  if (pLeftBlockInfoEx->compBlock->offset == pRightBlockInfoEx->compBlock->offset &&
      pLeftBlockInfoEx->compBlock->last == pRightBlockInfoEx->compBlock->last) {
B
Bomin Zhang 已提交
2241
    tsdbError("error in header file, two block with same offset:%" PRId64, (int64_t)pLeftBlockInfoEx->compBlock->offset);
2242
  }
H
Haojun Liao 已提交
2243
#endif
2244

H
Haojun Liao 已提交
2245
  return pLeftBlockInfoEx->compBlock->offset > pRightBlockInfoEx->compBlock->offset ? 1 : -1;
2246 2247
}

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

2251 2252
  if (pTsdbReadHandle->allocSize < size) {
    pTsdbReadHandle->allocSize = (int32_t)size;
wafwerar's avatar
wafwerar 已提交
2253
    char* tmp = taosMemoryRealloc(pTsdbReadHandle->pDataBlockInfo, pTsdbReadHandle->allocSize);
H
Haojun Liao 已提交
2254 2255 2256 2257
    if (tmp == NULL) {
      return TSDB_CODE_TDB_OUT_OF_MEMORY;
    }

dengyihao's avatar
dengyihao 已提交
2258
    pTsdbReadHandle->pDataBlockInfo = (STableBlockInfo*)tmp;
2259 2260
  }

2261
  memset(pTsdbReadHandle->pDataBlockInfo, 0, size);
2262 2263
  *numOfAllocBlocks = numOfBlocks;

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

2267 2268
  SBlockOrderSupporter sup = {0};
  sup.numOfTables = numOfTables;
wafwerar's avatar
wafwerar 已提交
2269 2270 2271
  sup.numOfBlocksPerTable = taosMemoryCalloc(1, sizeof(int32_t) * numOfTables);
  sup.blockIndexArray = taosMemoryCalloc(1, sizeof(int32_t) * numOfTables);
  sup.pDataBlockInfo = taosMemoryCalloc(1, POINTER_BYTES * numOfTables);
2272

2273
  if (sup.numOfBlocksPerTable == NULL || sup.blockIndexArray == NULL || sup.pDataBlockInfo == NULL) {
2274
    cleanBlockOrderSupporter(&sup, 0);
2275
    return TSDB_CODE_TDB_OUT_OF_MEMORY;
2276
  }
H
Haojun Liao 已提交
2277

2278
  int32_t cnt = 0;
2279
  int32_t numOfQualTables = 0;
H
Haojun Liao 已提交
2280

2281
  for (int32_t j = 0; j < numOfTables; ++j) {
2282
    STableCheckInfo* pTableCheck = (STableCheckInfo*)taosArrayGet(pTsdbReadHandle->pTableCheckInfo, j);
2283 2284 2285
    if (pTableCheck->numOfBlocks <= 0) {
      continue;
    }
H
Haojun Liao 已提交
2286

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

wafwerar's avatar
wafwerar 已提交
2290
    char* buf = taosMemoryMalloc(sizeof(STableBlockInfo) * pTableCheck->numOfBlocks);
2291
    if (buf == NULL) {
2292
      cleanBlockOrderSupporter(&sup, numOfQualTables);
2293
      return TSDB_CODE_TDB_OUT_OF_MEMORY;
2294 2295
    }

2296
    sup.pDataBlockInfo[numOfQualTables] = (STableBlockInfo*)buf;
2297 2298

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

H
Haojun Liao 已提交
2301 2302
      pBlockInfo->compBlock = &pBlock[k];
      pBlockInfo->pTableCheckInfo = pTableCheck;
2303 2304 2305
      cnt++;
    }

2306
    numOfQualTables++;
2307 2308
  }

H
Haojun Liao 已提交
2309
  assert(numOfBlocks == cnt);
2310

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

H
Haojun Liao 已提交
2316
    tsdbDebug("%p create data blocks info struct completed for 1 table, %d blocks not sorted %s", pTsdbReadHandle, cnt,
dengyihao's avatar
dengyihao 已提交
2317
              pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
2318 2319
    return TSDB_CODE_SUCCESS;
  }
2320

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

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

2327
  SMultiwayMergeTreeInfo* pTree = NULL;
dengyihao's avatar
dengyihao 已提交
2328
  uint8_t                 ret = tMergeTreeCreate(&pTree, sup.numOfTables, &sup, dataBlockOrderCompar);
2329 2330
  if (ret != TSDB_CODE_SUCCESS) {
    cleanBlockOrderSupporter(&sup, numOfTables);
2331
    return TSDB_CODE_TDB_OUT_OF_MEMORY;
2332 2333 2334 2335 2336
  }

  int32_t numOfTotal = 0;

  while (numOfTotal < cnt) {
2337
    int32_t pos = tMergeTreeGetChosenIndex(pTree);
2338 2339
    int32_t index = sup.blockIndexArray[pos]++;

H
Haojun Liao 已提交
2340
    STableBlockInfo* pBlocksInfo = sup.pDataBlockInfo[pos];
2341
    pTsdbReadHandle->pDataBlockInfo[numOfTotal++] = pBlocksInfo[index];
2342 2343

    // set data block index overflow, in order to disable the offset comparator
2344 2345
    if (sup.blockIndexArray[pos] >= sup.numOfBlocksPerTable[pos]) {
      sup.blockIndexArray[pos] = sup.numOfBlocksPerTable[pos] + 1;
2346
    }
2347

H
Haojun Liao 已提交
2348
    tMergeTreeAdjust(pTree, tMergeTreeGetAdjustIndex(pTree));
2349 2350 2351 2352 2353
  }

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

H
Haojun Liao 已提交
2358
  tsdbDebug("%p %d data blocks sort completed, %s", pTsdbReadHandle, cnt, pTsdbReadHandle->idStr);
2359
  cleanBlockOrderSupporter(&sup, numOfTables);
wafwerar's avatar
wafwerar 已提交
2360
  taosMemoryFree(pTree);
2361 2362 2363 2364

  return TSDB_CODE_SUCCESS;
}

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

dengyihao's avatar
dengyihao 已提交
2367 2368
static int32_t getDataBlockRv(STsdbReadHandle* pTsdbReadHandle, STableBlockInfo* pNext, bool* exists) {
  int32_t        step = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 1 : -1;
2369
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
H
Haojun Liao 已提交
2370

dengyihao's avatar
dengyihao 已提交
2371
  while (1) {
2372
    int32_t code = loadFileDataBlock(pTsdbReadHandle, pNext->compBlock, pNext->pTableCheckInfo, exists);
H
Haojun Liao 已提交
2373 2374 2375 2376
    if (code != TSDB_CODE_SUCCESS || *exists) {
      return code;
    }

2377 2378
    if ((cur->slot == pTsdbReadHandle->numOfBlocks - 1 && ASCENDING_TRAVERSE(pTsdbReadHandle->order)) ||
        (cur->slot == 0 && !ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
H
Haojun Liao 已提交
2379
      // all data blocks in current file has been checked already, try next file if exists
2380
      return getFirstFileDataBlock(pTsdbReadHandle, exists);
H
Haojun Liao 已提交
2381 2382 2383 2384
    } else {  // next block of the same file
      cur->slot += step;
      cur->mixBlock = false;
      cur->blockCompleted = false;
2385
      pNext = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
H
Haojun Liao 已提交
2386 2387 2388 2389
    }
  }
}

2390 2391 2392
static int32_t getFirstFileDataBlock(STsdbReadHandle* pTsdbReadHandle, bool* exists) {
  pTsdbReadHandle->numOfBlocks = 0;
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
H
Haojun Liao 已提交
2393 2394 2395

  int32_t code = TSDB_CODE_SUCCESS;

2396
  int32_t numOfBlocks = 0;
2397
  int32_t numOfTables = (int32_t)taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
2398

C
Cary Xu 已提交
2399
  STsdbKeepCfg* pCfg = REPO_KEEP_CFG(pTsdbReadHandle->pTsdb);
C
Cary Xu 已提交
2400
  STimeWindow   win = TSWINDOW_INITIALIZER;
2401

H
Hongze Cheng 已提交
2402
  while (true) {
2403
    tsdbRLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2404

2405 2406
    if ((pTsdbReadHandle->pFileGroup = tsdbFSIterNext(&pTsdbReadHandle->fileIter)) == NULL) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2407 2408 2409
      break;
    }

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

    // current file are not overlapped with query time window, ignore remain files
2413 2414 2415
    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 已提交
2416 2417
      tsdbDebug("%p remain files are not qualified for qrange:%" PRId64 "-%" PRId64 ", ignore, %s", pTsdbReadHandle,
                pTsdbReadHandle->window.skey, pTsdbReadHandle->window.ekey, pTsdbReadHandle->idStr);
2418 2419
      pTsdbReadHandle->pFileGroup = NULL;
      assert(pTsdbReadHandle->numOfBlocks == 0);
2420 2421 2422
      break;
    }

2423 2424
    if (tsdbSetAndOpenReadFSet(&pTsdbReadHandle->rhelper, pTsdbReadHandle->pFileGroup) < 0) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2425 2426 2427 2428
      code = terrno;
      break;
    }

2429
    tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2430

2431
    if (tsdbLoadBlockIdx(&pTsdbReadHandle->rhelper) < 0) {
H
Hongze Cheng 已提交
2432 2433 2434 2435
      code = terrno;
      break;
    }

2436
    if ((code = getFileCompInfo(pTsdbReadHandle, &numOfBlocks)) != TSDB_CODE_SUCCESS) {
2437 2438
      break;
    }
H
Haojun Liao 已提交
2439

H
Haojun Liao 已提交
2440 2441
    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 已提交
2442

2443 2444 2445 2446
    assert(numOfBlocks >= 0);
    if (numOfBlocks == 0) {
      continue;
    }
H
Haojun Liao 已提交
2447

2448
    // todo return error code to query engine
dengyihao's avatar
dengyihao 已提交
2449 2450
    if ((code = createDataBlocksInfo(pTsdbReadHandle, numOfBlocks, &pTsdbReadHandle->numOfBlocks)) !=
        TSDB_CODE_SUCCESS) {
2451 2452
      break;
    }
H
Haojun Liao 已提交
2453

2454 2455
    assert(numOfBlocks >= pTsdbReadHandle->numOfBlocks);
    if (pTsdbReadHandle->numOfBlocks > 0) {
2456 2457 2458
      break;
    }
  }
H
Haojun Liao 已提交
2459

2460
  // no data in file anymore
2461
  if (pTsdbReadHandle->numOfBlocks <= 0 || code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2462
    if (code == TSDB_CODE_SUCCESS) {
2463
      assert(pTsdbReadHandle->pFileGroup == NULL);
H
Haojun Liao 已提交
2464 2465
    }

D
dapan1121 已提交
2466
    cur->fid = INT32_MIN;  // denote that there are no data in file anymore
H
Haojun Liao 已提交
2467 2468
    *exists = false;
    return code;
2469
  }
H
Haojun Liao 已提交
2470

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

2475 2476
  STableBlockInfo* pBlockInfo = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
  return getDataBlockRv(pTsdbReadHandle, pBlockInfo, exists);
H
Haojun Liao 已提交
2477 2478 2479 2480 2481 2482 2483
}

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

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

2487 2488
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
  assert(cur->slot < pTsdbReadHandle->numOfBlocks && cur->slot >= 0);
H
Haojun Liao 已提交
2489 2490

  cur->slot += step;
dengyihao's avatar
dengyihao 已提交
2491
  cur->mixBlock = false;
H
Haojun Liao 已提交
2492
  cur->blockCompleted = false;
2493
}
H
Haojun Liao 已提交
2494 2495

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

H
Haojun Liao 已提交
2498
  pTableBlockInfo->totalSize = 0;
2499
  pTableBlockInfo->totalRows = 0;
H
Haojun Liao 已提交
2500

2501
  STsdbFS* pFileHandle = REPO_FS(pTsdbReadHandle->pTsdb);
H
Haojun Liao 已提交
2502 2503

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

  tsdbRLockFS(pFileHandle);
2509 2510
  tsdbFSIterInit(&pTsdbReadHandle->fileIter, pFileHandle, pTsdbReadHandle->order);
  tsdbFSIterSeek(&pTsdbReadHandle->fileIter, fid);
H
Haojun Liao 已提交
2511 2512
  tsdbUnLockFS(pFileHandle);

H
Haojun Liao 已提交
2513
  pTableBlockInfo->numOfFiles += 1;
H
Haojun Liao 已提交
2514

H
Haojun Liao 已提交
2515
  int32_t     code = TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
2516
  int32_t     numOfBlocks = 0;
2517
  int32_t     numOfTables = (int32_t)taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
dengyihao's avatar
dengyihao 已提交
2518
  int         defaultRows = 4096;  // TSDB_DEFAULT_BLOCK_ROWS(pCfg->maxRowsPerFileBlock);
H
Haojun Liao 已提交
2519 2520
  STimeWindow win = TSWINDOW_INITIALIZER;

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

H
Haojun Liao 已提交
2523 2524
  while (true) {
    numOfBlocks = 0;
2525
    tsdbRLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2526

2527 2528
    if ((pTsdbReadHandle->pFileGroup = tsdbFSIterNext(&pTsdbReadHandle->fileIter)) == NULL) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2529 2530 2531
      break;
    }

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

    // current file are not overlapped with query time window, ignore remain files
dengyihao's avatar
dengyihao 已提交
2535 2536
    if ((ascTraverse && win.skey > pTsdbReadHandle->window.ekey) ||
        (!ascTraverse && win.ekey < pTsdbReadHandle->window.ekey)) {
2537
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2538 2539
      tsdbDebug("%p remain files are not qualified for qrange:%" PRId64 "-%" PRId64 ", ignore, %s", pTsdbReadHandle,
                pTsdbReadHandle->window.skey, pTsdbReadHandle->window.ekey, pTsdbReadHandle->idStr);
2540
      pTsdbReadHandle->pFileGroup = NULL;
H
Haojun Liao 已提交
2541 2542 2543
      break;
    }

H
Haojun Liao 已提交
2544
    pTableBlockInfo->numOfFiles += 1;
2545 2546
    if (tsdbSetAndOpenReadFSet(&pTsdbReadHandle->rhelper, pTsdbReadHandle->pFileGroup) < 0) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2547 2548 2549 2550
      code = terrno;
      break;
    }

2551
    tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2552

2553
    if (tsdbLoadBlockIdx(&pTsdbReadHandle->rhelper) < 0) {
H
Haojun Liao 已提交
2554 2555 2556 2557
      code = terrno;
      break;
    }

2558
    if ((code = getFileCompInfo(pTsdbReadHandle, &numOfBlocks)) != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2559 2560 2561
      break;
    }

H
Haojun Liao 已提交
2562 2563
    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 已提交
2564 2565 2566 2567 2568 2569

    if (numOfBlocks == 0) {
      continue;
    }

    for (int32_t i = 0; i < numOfTables; ++i) {
2570
      STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, i);
H
Haojun Liao 已提交
2571 2572 2573

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

H
Haojun Liao 已提交
2576
        int32_t numOfRows = pBlock[j].numOfRows;
2577
        pTableBlockInfo->totalRows += numOfRows;
H
Haojun Liao 已提交
2578 2579 2580 2581 2582 2583 2584 2585 2586 2587 2588
        if (numOfRows > pTableBlockInfo->maxRows) {
          pTableBlockInfo->maxRows = numOfRows;
        }

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

        if (numOfRows < defaultRows) {
          pTableBlockInfo->numOfSmallBlocks += 1;
        }
dengyihao's avatar
dengyihao 已提交
2589 2590 2591
        //        int32_t  stepIndex = (numOfRows-1)/TSDB_BLOCK_DIST_STEP_ROWS;
        //        SFileBlockInfo *blockInfo = (SFileBlockInfo*)taosArrayGet(pTableBlockInfo->dataBlockInfos, stepIndex);
        //        blockInfo->numBlocksOfStep++;
H
Haojun Liao 已提交
2592 2593 2594 2595 2596 2597 2598
      }
    }
  }

  return code;
}

2599 2600 2601
static int32_t getDataBlocksInFiles(STsdbReadHandle* pTsdbReadHandle, bool* exists) {
  STsdbFS*       pFileHandle = REPO_FS(pTsdbReadHandle->pTsdb);
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
2602 2603

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

H
Hongze Cheng 已提交
2609
    tsdbRLockFS(pFileHandle);
2610 2611
    tsdbFSIterInit(&pTsdbReadHandle->fileIter, pFileHandle, pTsdbReadHandle->order);
    tsdbFSIterSeek(&pTsdbReadHandle->fileIter, fid);
H
Hongze Cheng 已提交
2612
    tsdbUnLockFS(pFileHandle);
2613

2614
    return getFirstFileDataBlock(pTsdbReadHandle, exists);
2615
  } else {
2616
    // check if current file block is all consumed
2617
    STableBlockInfo* pBlockInfo = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
2618
    STableCheckInfo* pCheckInfo = pBlockInfo->pTableCheckInfo;
H
Haojun Liao 已提交
2619

2620
    // current block is done, try next
H
Haojun Liao 已提交
2621
    if ((!cur->mixBlock) || cur->blockCompleted) {
H
Haojun Liao 已提交
2622
      // all data blocks in current file has been checked already, try next file if exists
2623
    } else {
H
Haojun Liao 已提交
2624 2625
      tsdbDebug("%p continue in current data block, index:%d, pos:%d, %s", pTsdbReadHandle, cur->slot, cur->pos,
                pTsdbReadHandle->idStr);
2626 2627
      int32_t code = handleDataMergeIfNeeded(pTsdbReadHandle, pBlockInfo->compBlock, pCheckInfo);
      *exists = (pTsdbReadHandle->realNumOfRows > 0);
H
Haojun Liao 已提交
2628

H
Haojun Liao 已提交
2629 2630 2631 2632 2633 2634 2635
      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
2636 2637
    if (isEndFileDataBlock(cur, pTsdbReadHandle->numOfBlocks, ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
      return getFirstFileDataBlock(pTsdbReadHandle, exists);
H
Haojun Liao 已提交
2638
    } else {
2639 2640 2641
      moveToNextDataBlockInCurrentFile(pTsdbReadHandle);
      STableBlockInfo* pNext = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
      return getDataBlockRv(pTsdbReadHandle, pNext, exists);
2642 2643
    }
  }
2644 2645
}

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

2649 2650
  while (pTsdbReadHandle->activeIndex < numOfTables) {
    if (hasMoreDataInCache(pTsdbReadHandle)) {
2651 2652
      return true;
    }
H
Haojun Liao 已提交
2653

2654
    pTsdbReadHandle->activeIndex += 1;
2655
  }
H
Haojun Liao 已提交
2656

2657 2658 2659
  return false;
}

dengyihao's avatar
dengyihao 已提交
2660
// todo not unref yet, since it is not support multi-group interpolation query
H
Haojun Liao 已提交
2661
static UNUSED_FUNC void changeQueryHandleForInterpQuery(tsdbReaderT pHandle) {
H
Haojun Liao 已提交
2662
  // filter the queried time stamp in the first place
dengyihao's avatar
dengyihao 已提交
2663
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle;
H
Haojun Liao 已提交
2664 2665

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

  int32_t i = 0;
dengyihao's avatar
dengyihao 已提交
2669
  while (i < numOfTables) {
2670
    STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, i);
H
Haojun Liao 已提交
2671 2672

    // the first qualified table for interpolation query
dengyihao's avatar
dengyihao 已提交
2673 2674 2675 2676
    //    if ((pTsdbReadHandle->window.skey <= pCheckInfo->pTableObj->lastKey) &&
    //        (pCheckInfo->pTableObj->lastKey != TSKEY_INITIAL_VAL)) {
    //      break;
    //    }
H
Haojun Liao 已提交
2677 2678 2679 2680 2681 2682 2683 2684 2685

    i++;
  }

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

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

2689 2690
  info.lastKey = pTsdbReadHandle->window.skey;
  taosArrayPush(pTsdbReadHandle->pTableCheckInfo, &info);
H
Haojun Liao 已提交
2691 2692 2693
}

static int tsdbReadRowsFromCache(STableCheckInfo* pCheckInfo, TSKEY maxKey, int maxRowsToRead, STimeWindow* win,
2694
                                 STsdbReadHandle* pTsdbReadHandle) {
dengyihao's avatar
dengyihao 已提交
2695
  int       numOfRows = 0;
C
Cary Xu 已提交
2696
  int       curRows = 0;
dengyihao's avatar
dengyihao 已提交
2697
  int32_t   numOfCols = (int32_t)taosArrayGetSize(pTsdbReadHandle->pColumns);
H
Hongze Cheng 已提交
2698
  STsdbCfg* pCfg = REPO_CFG(pTsdbReadHandle->pTsdb);
H
Haojun Liao 已提交
2699 2700
  win->skey = TSKEY_INITIAL_VAL;

dengyihao's avatar
dengyihao 已提交
2701 2702
  int64_t   st = taosGetTimestampUs();
  int16_t   rv = -1;
D
fix bug  
dapan1121 已提交
2703
  STSchema* pSchema = NULL;
C
Cary Xu 已提交
2704 2705
  TSKEY     lastRowKey = TSKEY_INITIAL_VAL;

H
Haojun Liao 已提交
2706
  do {
C
Cary Xu 已提交
2707
    STSRow* row = getSRowInTableMem(pCheckInfo, pTsdbReadHandle->order, pCfg->update, NULL, TD_VER_MAX);
H
Haojun Liao 已提交
2708 2709 2710 2711
    if (row == NULL) {
      break;
    }

H
Haojun Liao 已提交
2712
    TSKEY key = TD_ROW_KEY(row);
dengyihao's avatar
dengyihao 已提交
2713 2714 2715 2716
    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 已提交
2717 2718 2719 2720 2721 2722 2723 2724 2725

      break;
    }

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

    win->ekey = key;
H
Haojun Liao 已提交
2726
    if (rv != TD_ROW_SVER(row)) {
H
Hongze Cheng 已提交
2727
      pSchema = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), pCheckInfo->tableId, 1);
H
Haojun Liao 已提交
2728
      rv = TD_ROW_SVER(row);
D
fix bug  
dapan1121 已提交
2729
    }
C
Cary Xu 已提交
2730 2731
    numOfRows += mergeTwoRowFromMem(pTsdbReadHandle, maxRowsToRead, &curRows, row, NULL, numOfCols, pCheckInfo->tableId,
                                    pSchema, NULL, pCfg->update, &lastRowKey);
H
Haojun Liao 已提交
2732

C
Cary Xu 已提交
2733
    if (numOfRows >= maxRowsToRead) {
H
Haojun Liao 已提交
2734 2735 2736 2737
      moveToNextRowInMem(pCheckInfo);
      break;
    }

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

C
Cary Xu 已提交
2740
  taosMemoryFreeClear(pSchema);  // free the STSChema
H
Haojun Liao 已提交
2741 2742 2743
  assert(numOfRows <= maxRowsToRead);

  int64_t elapsedTime = taosGetTimestampUs() - st;
dengyihao's avatar
dengyihao 已提交
2744 2745
  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 已提交
2746 2747 2748 2749

  return numOfRows;
}

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

2753 2754 2755 2756 2757
  while (1) {
    tb_uid_t id = metaCtbCursorNext(pCur);
    if (id == 0) {
      break;
    }
H
Haojun Liao 已提交
2758

2759
    STableKeyInfo info = {.lastKey = TSKEY_INITIAL_VAL, uid = id};
H
Haojun Liao 已提交
2760 2761 2762
    taosArrayPush(list, &info);
  }

C
Cary Xu 已提交
2763
  metaCloseCtbCursor(pCur);
H
Haojun Liao 已提交
2764 2765 2766 2767 2768 2769 2770 2771
  return TSDB_CODE_SUCCESS;
}

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

dengyihao's avatar
dengyihao 已提交
2772 2773 2774 2775 2776 2777
  //  tQueryInfo* pInfo = (tQueryInfo*)param;
  //  if (pInfo->optr != TSDB_RELATION_IN) {
  //    taosMemoryFreeClear(pInfo->q);
  //  } else {
  //    taosHashCleanup((SHashObj *)(pInfo->q));
  //  }
H
Haojun Liao 已提交
2778

wafwerar's avatar
wafwerar 已提交
2779
  taosMemoryFree(param);
H
Haojun Liao 已提交
2780 2781
}

dengyihao's avatar
dengyihao 已提交
2782 2783
#define TSDB_PREV_ROW 0x1
#define TSDB_NEXT_ROW 0x2
2784

dengyihao's avatar
dengyihao 已提交
2785
static bool loadBlockOfActiveTable(STsdbReadHandle* pTsdbReadHandle) {
2786
  if (pTsdbReadHandle->checkFiles) {
H
Haojun Liao 已提交
2787 2788
    // check if the query range overlaps with the file data block
    bool exists = true;
H
Haojun Liao 已提交
2789

2790
    int32_t code = getDataBlocksInFiles(pTsdbReadHandle, &exists);
H
Haojun Liao 已提交
2791
    if (code != TSDB_CODE_SUCCESS) {
2792
      pTsdbReadHandle->checkFiles = false;
H
Haojun Liao 已提交
2793 2794
      return false;
    }
H
Haojun Liao 已提交
2795

H
Haojun Liao 已提交
2796
    if (exists) {
dengyihao's avatar
dengyihao 已提交
2797
      tsdbRetrieveDataBlock((tsdbReaderT*)pTsdbReadHandle, NULL);
2798 2799 2800
      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 已提交
2801 2802
      }

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

2807
    pTsdbReadHandle->checkFiles = false;
H
Haojun Liao 已提交
2808
  }
H
Haojun Liao 已提交
2809

2810 2811
  if (hasMoreDataInCache(pTsdbReadHandle)) {
    pTsdbReadHandle->currentLoadExternalRows = false;
H
Haojun Liao 已提交
2812 2813
    return true;
  }
H
Haojun Liao 已提交
2814

H
Haojun Liao 已提交
2815
  // current result is empty
dengyihao's avatar
dengyihao 已提交
2816 2817 2818
  if (pTsdbReadHandle->currentLoadExternalRows && pTsdbReadHandle->window.skey == pTsdbReadHandle->window.ekey &&
      pTsdbReadHandle->cur.rows == 0) {
    //    STsdbMemTable* pMemRef = pTsdbReadHandle->pMemTable;
H
Haojun Liao 已提交
2819

dengyihao's avatar
dengyihao 已提交
2820 2821
    //    doGetExternalRow(pTsdbReadHandle, TSDB_PREV_ROW, pMemRef);
    //    doGetExternalRow(pTsdbReadHandle, TSDB_NEXT_ROW, pMemRef);
H
Haojun Liao 已提交
2822

2823
    bool result = tsdbGetExternalRow(pTsdbReadHandle);
H
Haojun Liao 已提交
2824

dengyihao's avatar
dengyihao 已提交
2825 2826
    //    pTsdbReadHandle->prev = doFreeColumnInfoData(pTsdbReadHandle->prev);
    //    pTsdbReadHandle->next = doFreeColumnInfoData(pTsdbReadHandle->next);
2827
    pTsdbReadHandle->currentLoadExternalRows = false;
H
Haojun Liao 已提交
2828 2829

    return result;
2830
  }
H
Haojun Liao 已提交
2831

H
Haojun Liao 已提交
2832 2833
  return false;
}
2834

2835
static bool loadCachedLastRow(STsdbReadHandle* pTsdbReadHandle) {
H
Haojun Liao 已提交
2836
  // the last row is cached in buffer, return it directly.
2837
  // here note that the pTsdbReadHandle->window must be the TS_INITIALIZER
dengyihao's avatar
dengyihao 已提交
2838
  int32_t numOfCols = (int32_t)(QH_GET_NUM_OF_COLS(pTsdbReadHandle));
2839
  size_t  numOfTables = taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
H
Haojun Liao 已提交
2840 2841
  assert(numOfTables > 0 && numOfCols > 0);

2842
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
2843

dengyihao's avatar
dengyihao 已提交
2844 2845 2846
  STSRow* pRow = NULL;
  TSKEY   key = TSKEY_INITIAL_VAL;
  int32_t step = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 1 : -1;
C
Cary Xu 已提交
2847 2848
  TSKEY   lastRowKey = TSKEY_INITIAL_VAL;
  int32_t curRow = 0;
2849 2850 2851

  if (++pTsdbReadHandle->activeIndex < numOfTables) {
    STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, pTsdbReadHandle->activeIndex);
dengyihao's avatar
dengyihao 已提交
2852 2853 2854 2855
    //    int32_t ret = tsdbGetCachedLastRow(pCheckInfo->pTableObj, &pRow, &key);
    //    if (ret != TSDB_CODE_SUCCESS) {
    //      return false;
    //    }
C
Cary Xu 已提交
2856 2857
    mergeTwoRowFromMem(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, &curRow, pRow, NULL, numOfCols,
                       pCheckInfo->tableId, NULL, NULL, true, &lastRowKey);
wafwerar's avatar
wafwerar 已提交
2858
    taosMemoryFreeClear(pRow);
H
Haojun Liao 已提交
2859

H
Haojun Liao 已提交
2860 2861 2862
    // update the last key value
    pCheckInfo->lastKey = key + step;

dengyihao's avatar
dengyihao 已提交
2863 2864
    cur->rows = 1;  // only one row
    cur->lastKey = key + step;
H
Haojun Liao 已提交
2865 2866 2867 2868 2869
    cur->mixBlock = true;
    cur->win.skey = key;
    cur->win.ekey = key;

    return true;
2870
  }
H
Haojun Liao 已提交
2871

H
Haojun Liao 已提交
2872 2873 2874
  return false;
}

dengyihao's avatar
dengyihao 已提交
2875
// static bool loadCachedLast(STsdbReadHandle* pTsdbReadHandle) {
2876 2877 2878 2879 2880 2881 2882 2883 2884 2885 2886 2887 2888 2889 2890 2891 2892 2893 2894
//  // 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 已提交
2895 2896
//      tsdbWarn("no last cached for table %s, uid:%" PRIu64 ",tid:%d", pTable->name->data, pTable->uid,
//      pTable->tableId); continue;
2897 2898 2899 2900 2901 2902 2903 2904 2905 2906 2907 2908 2909 2910 2911 2912 2913 2914 2915 2916 2917 2918 2919 2920 2921 2922 2923 2924 2925 2926 2927 2928 2929 2930 2931 2932 2933 2934 2935 2936 2937 2938 2939 2940 2941 2942 2943 2944 2945 2946 2947 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
//    }
//
//    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 已提交
3027 3028 3029
  assert(numOfTables > 0);

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

dengyihao's avatar
dengyihao 已提交
3031
  while (pTsdbReadHandle->activeIndex < numOfTables) {
3032
    if (loadBlockOfActiveTable(pTsdbReadHandle)) {
H
Haojun Liao 已提交
3033 3034 3035
      return true;
    }

3036
    STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, pTsdbReadHandle->activeIndex);
H
Haojun Liao 已提交
3037 3038
    pCheckInfo->numOfBlocks = 0;

3039 3040
    pTsdbReadHandle->activeIndex += 1;
    pTsdbReadHandle->locateStart = false;
dengyihao's avatar
dengyihao 已提交
3041 3042
    pTsdbReadHandle->checkFiles = true;
    pTsdbReadHandle->cur.rows = 0;
3043
    pTsdbReadHandle->currentLoadExternalRows = pTsdbReadHandle->loadExternalRow;
H
Haojun Liao 已提交
3044 3045 3046 3047

    terrno = TSDB_CODE_SUCCESS;

    int64_t elapsedTime = taosGetTimestampUs() - stime;
3048
    pTsdbReadHandle->cost.checkForNextTime += elapsedTime;
H
Haojun Liao 已提交
3049 3050 3051
  }

  return false;
3052 3053
}

H
Haojun Liao 已提交
3054
// handle data in cache situation
H
Haojun Liao 已提交
3055
bool tsdbNextDataBlock(tsdbReaderT pHandle) {
dengyihao's avatar
dengyihao 已提交
3056
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle;
Y
yihaoDeng 已提交
3057

3058 3059
  size_t numOfCols = taosArrayGetSize(pTsdbReadHandle->pColumns);
  for (int32_t i = 0; i < numOfCols; ++i) {
3060 3061 3062 3063
    SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
    colInfoDataCleanup(pColInfo, pTsdbReadHandle->outputCapacity);
  }

3064
  if (emptyQueryTimewindow(pTsdbReadHandle)) {
dengyihao's avatar
dengyihao 已提交
3065 3066
    tsdbDebug("%p query window not overlaps with the data set, no result returned, %s", pTsdbReadHandle,
              pTsdbReadHandle->idStr);
3067 3068 3069
    return false;
  }

Y
yihaoDeng 已提交
3070 3071 3072
  int64_t stime = taosGetTimestampUs();
  int64_t elapsedTime = stime;

3073
  // TODO refactor: remove "type"
3074 3075
  if (pTsdbReadHandle->type == TSDB_QUERY_TYPE_LAST) {
    if (pTsdbReadHandle->cachelastrow == TSDB_CACHED_TYPE_LASTROW) {
dengyihao's avatar
dengyihao 已提交
3076
      //      return loadCachedLastRow(pTsdbReadHandle);
3077
    } else if (pTsdbReadHandle->cachelastrow == TSDB_CACHED_TYPE_LAST) {
dengyihao's avatar
dengyihao 已提交
3078
      //      return loadCachedLast(pTsdbReadHandle);
D
init  
dapan1121 已提交
3079
    }
H
Haojun Liao 已提交
3080
  }
Y
yihaoDeng 已提交
3081

3082 3083
  if (pTsdbReadHandle->loadType == BLOCK_LOAD_TABLE_SEQ_ORDER) {
    return loadDataBlockFromTableSeq(pTsdbReadHandle);
dengyihao's avatar
dengyihao 已提交
3084
  } else {  // loadType == RR and Offset Order
3085
    if (pTsdbReadHandle->checkFiles) {
H
Haojun Liao 已提交
3086 3087 3088
      // check if the query range overlaps with the file data block
      bool exists = true;

3089
      int32_t code = getDataBlocksInFiles(pTsdbReadHandle, &exists);
H
Haojun Liao 已提交
3090
      if (code != TSDB_CODE_SUCCESS) {
3091 3092
        pTsdbReadHandle->activeIndex = 0;
        pTsdbReadHandle->checkFiles = false;
H
Haojun Liao 已提交
3093 3094 3095 3096 3097

        return false;
      }

      if (exists) {
3098
        pTsdbReadHandle->cost.checkForNextTime += (taosGetTimestampUs() - stime);
H
Haojun Liao 已提交
3099 3100
        return exists;
      }
Y
yihaoDeng 已提交
3101

3102 3103
      pTsdbReadHandle->activeIndex = 0;
      pTsdbReadHandle->checkFiles = false;
Y
yihaoDeng 已提交
3104 3105
    }

H
Haojun Liao 已提交
3106
    // TODO: opt by consider the scan order
3107
    bool ret = doHasDataInBuffer(pTsdbReadHandle);
H
Haojun Liao 已提交
3108
    terrno = TSDB_CODE_SUCCESS;
Y
yihaoDeng 已提交
3109

H
Haojun Liao 已提交
3110
    elapsedTime = taosGetTimestampUs() - stime;
3111
    pTsdbReadHandle->cost.checkForNextTime += elapsedTime;
H
Haojun Liao 已提交
3112
    return ret;
Y
yihaoDeng 已提交
3113 3114
  }
}
3115

dengyihao's avatar
dengyihao 已提交
3116
// static int32_t doGetExternalRow(STsdbReadHandle* pTsdbReadHandle, int16_t type, STsdbMemTable* pMemRef) {
3117 3118 3119 3120 3121 3122 3123 3124 3125 3126 3127 3128 3129 3130 3131 3132 3133 3134 3135 3136 3137 3138 3139 3140 3141 3142 3143 3144 3145 3146 3147 3148 3149 3150
//  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 已提交
3151
//    colInfo.pData = taosMemoryCalloc(1, pCol->info.bytes);
3152 3153 3154 3155 3156 3157 3158 3159 3160
//    if (colInfo.pData == NULL) {
//      terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
//      goto out_of_memory;
//    }
//
//    taosArrayPush(row, &colInfo);
//  }
//
//  // load the previous row
3161
//  SQueryTableDataCond cond = {.numOfCols = numOfCols, .loadExternalRows = false, .type = BLOCK_LOAD_OFFSET_SEQ_ORDER};
3162 3163 3164 3165 3166 3167 3168 3169
//  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 已提交
3170
//  cond.colList = taosMemoryCalloc(cond.numOfCols, sizeof(SColumnInfo));
3171 3172 3173 3174 3175 3176 3177 3178 3179 3180
//  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 已提交
3181
//  pSecQueryHandle = tsdbQueryTablesImpl(pTsdbReadHandle->pTsdb, &cond, pTsdbReadHandle->idStr, pMemRef);
wafwerar's avatar
wafwerar 已提交
3182
//  taosMemoryFreeClear(cond.colList);
3183 3184 3185 3186 3187 3188 3189 3190 3191 3192 3193 3194 3195 3196 3197 3198 3199 3200 3201 3202 3203 3204 3205 3206 3207 3208 3209 3210 3211 3212 3213 3214 3215 3216 3217 3218 3219 3220
//
//  // 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 已提交
3221
// out_of_memory:
3222
//  tsdbCleanupReadHandle(pSecQueryHandle);
3223 3224 3225
//  return terrno;
//}

H
Haojun Liao 已提交
3226
bool tsdbGetExternalRow(tsdbReaderT pHandle) {
dengyihao's avatar
dengyihao 已提交
3227 3228
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle;
  SQueryFilePos*   cur = &pTsdbReadHandle->cur;
H
Haojun Liao 已提交
3229

H
Haojun Liao 已提交
3230 3231
  cur->fid = INT32_MIN;
  cur->mixBlock = true;
3232
  if (pTsdbReadHandle->prev == NULL || pTsdbReadHandle->next == NULL) {
H
Haojun Liao 已提交
3233 3234
    cur->rows = 0;
    return false;
H
Haojun Liao 已提交
3235 3236
  }

dengyihao's avatar
dengyihao 已提交
3237
  int32_t numOfCols = (int32_t)QH_GET_NUM_OF_COLS(pTsdbReadHandle);
H
Haojun Liao 已提交
3238
  for (int32_t i = 0; i < numOfCols; ++i) {
3239 3240
    SColumnInfoData* pColInfoData = taosArrayGet(pTsdbReadHandle->pColumns, i);
    SColumnInfoData* first = taosArrayGet(pTsdbReadHandle->prev, i);
H
Haojun Liao 已提交
3241 3242 3243

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

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

    if (i == 0 && pColInfoData->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
H
Haojun Liao 已提交
3248
      cur->win.skey = *(TSKEY*)pColInfoData->pData;
sangshuduo's avatar
sangshuduo 已提交
3249
      cur->win.ekey = *(TSKEY*)(((char*)pColInfoData->pData) + TSDB_KEYSIZE);
H
Haojun Liao 已提交
3250 3251 3252
    }
  }

H
Haojun Liao 已提交
3253 3254
  cur->rows = 2;
  return true;
3255 3256
}

3257
/*
3258
 * if lastRow == NULL, return TSDB_CODE_TDB_NO_CACHE_LAST_ROW
3259
 * else set pRes and return TSDB_CODE_SUCCESS and save lastKey
3260
 */
H
Haojun Liao 已提交
3261
// int32_t tsdbGetCachedLastRow(STable* pTable, STSRow** pRes, TSKEY* lastKey) {
3262 3263 3264 3265 3266 3267 3268 3269 3270 3271 3272 3273 3274 3275 3276 3277
//  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 已提交
3278
// out:
3279 3280 3281 3282
//  TSDB_RUNLOCK_TABLE(pTable);
//  return code;
//}

3283
bool isTsdbCacheLastRow(tsdbReaderT* pReader) {
dengyihao's avatar
dengyihao 已提交
3284
  return ((STsdbReadHandle*)pReader)->cachelastrow > TSDB_CACHED_TYPE_NONE;
D
fix bug  
dapan1121 已提交
3285 3286
}

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

dengyihao's avatar
dengyihao 已提交
3290 3291 3292 3293 3294 3295 3296 3297 3298 3299 3300 3301 3302 3303 3304 3305 3306 3307 3308 3309 3310 3311 3312 3313
  //  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 已提交
3314

3315
  return TSDB_CODE_SUCCESS;
3316 3317
}

3318 3319
int32_t checkForCachedLast(STsdbReadHandle* pTsdbReadHandle) {
  assert(pTsdbReadHandle != NULL);
D
update  
dapan1121 已提交
3320 3321

  int32_t code = 0;
dengyihao's avatar
dengyihao 已提交
3322 3323 3324
  //  if (pTsdbReadHandle->pTsdb && atomic_load_8(&pTsdbReadHandle->pTsdb->hasCachedLastColumn)){
  //    pTsdbReadHandle->cachelastrow = TSDB_CACHED_TYPE_LAST;
  //  }
D
update  
dapan1121 已提交
3325 3326

  // update the tsdb query time range
3327
  if (pTsdbReadHandle->cachelastrow) {
dengyihao's avatar
dengyihao 已提交
3328
    pTsdbReadHandle->checkFiles = false;
3329
    pTsdbReadHandle->activeIndex = -1;  // start from -1
D
update  
dapan1121 已提交
3330 3331 3332 3333 3334
  }

  return code;
}

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

H
Haojun Liao 已提交
3338
  int32_t totalNumOfTable = 0;
3339
  SArray* emptyGroup = taosArrayInit(16, sizeof(int32_t));
H
Haojun Liao 已提交
3340

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

H
Haojun Liao 已提交
3347
    STableKeyInfo keyInfo = {0};
H
Haojun Liao 已提交
3348

H
Haojun Liao 已提交
3349
    size_t numOfTables = taosArrayGetSize(pGroup);
dengyihao's avatar
dengyihao 已提交
3350 3351
    for (int32_t i = 0; i < numOfTables; ++i) {
      STableKeyInfo* pInfo = (STableKeyInfo*)taosArrayGet(pGroup, i);
H
Haojun Liao 已提交
3352

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

dengyihao's avatar
dengyihao 已提交
3358
        //        keyInfo.pTable  = pInfo->pTable;
H
Haojun Liao 已提交
3359
        keyInfo.lastKey = key;
dengyihao's avatar
dengyihao 已提交
3360
        pInfo->lastKey = key;
H
Haojun Liao 已提交
3361

H
Haojun Liao 已提交
3362 3363 3364
        if (key < window.skey) {
          window.skey = key;
        }
3365

H
Haojun Liao 已提交
3366 3367 3368 3369
        if (key > window.ekey) {
          window.ekey = key;
        }
      }
3370
    }
H
Haojun Liao 已提交
3371

H
Haojun Liao 已提交
3372
    // more than one table in each group, only one table left for each group
dengyihao's avatar
dengyihao 已提交
3373 3374 3375 3376 3377 3378 3379 3380 3381 3382 3383 3384
    //    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);
    //    }
3385
  }
H
Haojun Liao 已提交
3386

H
Haojun Liao 已提交
3387 3388 3389
  // window does not being updated, so set the original
  if (window.skey == INT64_MAX && window.ekey == INT64_MIN) {
    window = TSWINDOW_INITIALIZER;
H
Haojun Liao 已提交
3390
    assert(totalNumOfTable == 0 && taosArrayGetSize(groupList->pGroupList) == numOfGroups);
H
Haojun Liao 已提交
3391 3392
  }

dengyihao's avatar
dengyihao 已提交
3393
  taosArrayRemoveBatch(groupList->pGroupList, TARRAY_GET_START(emptyGroup), (int32_t)taosArrayGetSize(emptyGroup));
3394 3395
  taosArrayDestroy(emptyGroup);

H
Haojun Liao 已提交
3396
  groupList->numOfTables = totalNumOfTable;
H
Haojun Liao 已提交
3397
  return window;
H
hjxilinx 已提交
3398 3399
}

H
Haojun Liao 已提交
3400
void tsdbRetrieveDataBlockInfo(tsdbReaderT* pTsdbReadHandle, SDataBlockInfo* pDataBlockInfo) {
3401
  STsdbReadHandle* pHandle = (STsdbReadHandle*)pTsdbReadHandle;
dengyihao's avatar
dengyihao 已提交
3402
  SQueryFilePos*   cur = &pHandle->cur;
3403 3404

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

3406
  // there are data in file
D
dapan1121 已提交
3407
  if (pHandle->cur.fid != INT32_MIN) {
3408
    STableBlockInfo* pBlockInfo = &pHandle->pDataBlockInfo[cur->slot];
3409
    uid = pBlockInfo->pTableCheckInfo->tableId;
H
[td-32]  
hjxilinx 已提交
3410
  } else {
3411
    STableCheckInfo* pCheckInfo = taosArrayGet(pHandle->pTableCheckInfo, pHandle->activeIndex);
3412
    uid = pCheckInfo->tableId;
3413
  }
3414

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

3418
  pDataBlockInfo->uid = uid;
3419 3420 3421 3422 3423 3424

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

dengyihao's avatar
dengyihao 已提交
3425
  pDataBlockInfo->rows = cur->rows;
H
Haojun Liao 已提交
3426
  pDataBlockInfo->window = cur->win;
C
Cary Xu 已提交
3427
  //  ASSERT(pDataBlockInfo->numOfCols >= (int32_t)(QH_GET_NUM_OF_COLS(pHandle));
3428
}
H
hjxilinx 已提交
3429

H
Haojun Liao 已提交
3430 3431 3432
/*
 * return null for mixed data block, if not a complete file data block, the statistics value will always return NULL
 */
3433
int32_t tsdbRetrieveDataBlockStatisInfo(tsdbReaderT* pTsdbReadHandle, SColumnDataAgg*** pBlockStatis, bool* allHave) {
dengyihao's avatar
dengyihao 已提交
3434
  STsdbReadHandle* pHandle = (STsdbReadHandle*)pTsdbReadHandle;
3435
  *allHave = false;
H
Haojun Liao 已提交
3436

H
Haojun Liao 已提交
3437 3438
  SQueryFilePos* c = &pHandle->cur;
  if (c->mixBlock) {
H
Haojun Liao 已提交
3439 3440 3441
    *pBlockStatis = NULL;
    return TSDB_CODE_SUCCESS;
  }
H
Haojun Liao 已提交
3442

H
Haojun Liao 已提交
3443 3444 3445 3446
  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 已提交
3447 3448 3449 3450
  if (pBlockInfo->compBlock->numOfSubBlocks > 1) {
    *pBlockStatis = NULL;
    return TSDB_CODE_SUCCESS;
  }
H
Haojun Liao 已提交
3451 3452

  int64_t stime = taosGetTimestampUs();
3453 3454
  int     statisStatus = tsdbLoadBlockStatis(&pHandle->rhelper, pBlockInfo->compBlock);
  if (statisStatus < TSDB_STATIS_OK) {
H
Hongze Cheng 已提交
3455
    return terrno;
3456 3457 3458
  } else if (statisStatus > TSDB_STATIS_OK) {
    *pBlockStatis = NULL;
    return TSDB_CODE_SUCCESS;
H
Hongze Cheng 已提交
3459
  }
H
Haojun Liao 已提交
3460

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

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

H
Haojun Liao 已提交
3466
  size_t numOfCols = QH_GET_NUM_OF_COLS(pHandle);
3467 3468 3469
  memset(pHandle->suppInfo.plist, 0, numOfCols * POINTER_BYTES);
  memset(pHandle->suppInfo.pstatis, 0, numOfCols * sizeof(SColumnDataAgg));

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

3474 3475
  *allHave = true;
  tsdbGetBlockStatis(&pHandle->rhelper, pHandle->suppInfo.pstatis, (int)numOfCols, pBlockInfo->compBlock);
H
Haojun Liao 已提交
3476 3477

  // always load the first primary timestamp column data
3478
  SColumnDataAgg* pPrimaryColStatis = &pHandle->suppInfo.pstatis[0];
3479
  assert(pPrimaryColStatis->colId == PRIMARYKEY_TIMESTAMP_COL_ID);
H
Haojun Liao 已提交
3480 3481 3482 3483

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

dengyihao's avatar
dengyihao 已提交
3486
  // update the number of NULL data rows
3487
  int32_t* slotIds = pHandle->suppInfo.slotIds;
dengyihao's avatar
dengyihao 已提交
3488
  for (int32_t i = 1; i < numOfCols; ++i) {
3489
    ASSERT(colIds[i] == pHandle->pSchema->columns[slotIds[i]].colId);
C
Cary Xu 已提交
3490
    if (IS_BSMA_ON(&(pHandle->pSchema->columns[slotIds[i]]))) {
3491 3492 3493 3494 3495 3496 3497
      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 已提交
3498 3499
    }
  }
H
Haojun Liao 已提交
3500 3501 3502 3503

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

3504
  *pBlockStatis = pHandle->suppInfo.plist;
3505
  return TSDB_CODE_SUCCESS;
H
hjxilinx 已提交
3506 3507
}

H
Haojun Liao 已提交
3508
SArray* tsdbRetrieveDataBlock(tsdbReaderT* pTsdbReadHandle, SArray* pIdList) {
H
[td-32]  
hjxilinx 已提交
3509
  /**
H
hjxilinx 已提交
3510
   * In the following two cases, the data has been loaded to SColumnInfoData.
H
[td-32]  
hjxilinx 已提交
3511 3512
   * 1. data is from cache, 2. data block is not completed qualified to query time range
   */
3513
  STsdbReadHandle* pHandle = (STsdbReadHandle*)pTsdbReadHandle;
D
dapan1121 已提交
3514
  if (pHandle->cur.fid == INT32_MIN) {
H
[td-32]  
hjxilinx 已提交
3515 3516
    return pHandle->pColumns;
  } else {
H
Haojun Liao 已提交
3517 3518
    STableBlockInfo* pBlockInfo = &pHandle->pDataBlockInfo[pHandle->cur.slot];
    STableCheckInfo* pCheckInfo = pBlockInfo->pTableCheckInfo;
3519

3520
    if (pHandle->cur.mixBlock) {
H
[td-32]  
hjxilinx 已提交
3521 3522
      return pHandle->pColumns;
    } else {
H
Haojun Liao 已提交
3523
      SDataBlockInfo binfo = GET_FILE_DATA_BLOCK_INFO(pCheckInfo, pBlockInfo->compBlock);
3524
      assert(pHandle->realNumOfRows <= binfo.rows);
H
Haojun Liao 已提交
3525

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

H
Hongze Cheng 已提交
3529
      if (pBlockLoadInfo->slot == pHandle->cur.slot && pBlockLoadInfo->fileGroup->fid == pHandle->cur.fid &&
H
Haojun Liao 已提交
3530
          pBlockLoadInfo->uid == pCheckInfo->tableId) {
H
hjxilinx 已提交
3531
        return pHandle->pColumns;
H
Haojun Liao 已提交
3532
      } else {  // only load the file block
H
refact  
Hongze Cheng 已提交
3533
        SBlock* pBlock = pBlockInfo->compBlock;
H
Haojun Liao 已提交
3534
        if (doLoadFileDataBlock(pHandle, pBlock, pCheckInfo, pHandle->cur.slot) != TSDB_CODE_SUCCESS) {
3535 3536
          return NULL;
        }
H
Haojun Liao 已提交
3537

H
Haojun Liao 已提交
3538
        int32_t numOfRows = doCopyRowsFromFileBlock(pHandle, pHandle->outputCapacity, 0, 0, pBlock->numOfRows - 1);
H
hjxilinx 已提交
3539 3540
        return pHandle->pColumns;
      }
H
[td-32]  
hjxilinx 已提交
3541 3542
    }
  }
H
hjxilinx 已提交
3543
}
3544
#if 0
3545
void filterPrepare(void* expr, void* param) {
3546
  tExprNode* pExpr = (tExprNode*)expr;
H
[td-32]  
hjxilinx 已提交
3547
  if (pExpr->_node.info != NULL) {
3548 3549
    return;
  }
3550

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

3553
  STSchema*   pTSSchema = (STSchema*) param;
H
hjxilinx 已提交
3554 3555 3556
  tQueryInfo* pInfo = pExpr->_node.info;
  tVariant*   pCond = pExpr->_node.pRight->pVal;
  SSchema*    pSchema = pExpr->_node.pLeft->pSchema;
3557

3558 3559
  pInfo->sch      = *pSchema;
  pInfo->optr     = pExpr->_node.optr;
Y
yihaoDeng 已提交
3560
  pInfo->compare  = getComparFunc(pInfo->sch.type, pInfo->optr);
H
Haojun Liao 已提交
3561
  pInfo->indexed  = pTSSchema->columns->colId == pInfo->sch.colId;
H
Haojun Liao 已提交
3562

H
hjxilinx 已提交
3563
  if (pInfo->optr == TSDB_RELATION_IN) {
Y
yihaoDeng 已提交
3564
     int dummy = -1;
3565
     SHashObj *pObj = NULL;
Y
yihaoDeng 已提交
3566 3567 3568 3569
     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 已提交
3570
          char* p = taosArrayGetP(arr, i);
3571 3572
          strntolower_s(varDataVal(p), varDataVal(p), varDataLen(p));
          taosHashPut(pObj, varDataVal(p), varDataLen(p), &dummy, sizeof(dummy));
Y
yihaoDeng 已提交
3573 3574 3575 3576
        }
     } else {
       buildFilterSetFromBinary((void **)&pObj, pCond->pz, pCond->nLen);
     }
3577
     pInfo->q = (char *)pObj;
H
Haojun Liao 已提交
3578
  } else if (pCond != NULL) {
3579 3580 3581 3582
    uint32_t size = pCond->nLen * TSDB_NCHAR_SIZE;
    if (size < (uint32_t)pSchema->bytes) {
      size = pSchema->bytes;
    }
wafwerar's avatar
wafwerar 已提交
3583
    // to make sure tonchar does not cause invalid write, since the '\0' needs at least sizeof(TdUcs4) space.
wafwerar's avatar
wafwerar 已提交
3584
    pInfo->q = taosMemoryCalloc(1, size + TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE);
3585
    tVariantDump(pCond, pInfo->q, pSchema->type, true);
weixin_48148422's avatar
weixin_48148422 已提交
3586
  }
3587 3588
}

3589
#endif
3590

dengyihao's avatar
dengyihao 已提交
3591
static int32_t tableGroupComparFn(const void* p1, const void* p2, const void* param) {
3592
#if 0
3593
  STableGroupSupporter* pTableGroupSupp = (STableGroupSupporter*) param;
3594 3595
  STable* pTable1 = ((STableKeyInfo*) p1)->uid;
  STable* pTable2 = ((STableKeyInfo*) p2)->uid;
H
Haojun Liao 已提交
3596

3597 3598 3599
  for (int32_t i = 0; i < pTableGroupSupp->numOfCols; ++i) {
    SColIndex* pColIndex = &pTableGroupSupp->pCols[i];
    int32_t colIndex = pColIndex->colIndex;
H
Haojun Liao 已提交
3600

H
Haojun Liao 已提交
3601
    assert(colIndex >= TSDB_TBNAME_COLUMN_INDEX);
H
Haojun Liao 已提交
3602

3603 3604 3605 3606
    char *  f1 = NULL;
    char *  f2 = NULL;
    int32_t type = 0;
    int32_t bytes = 0;
H
Haojun Liao 已提交
3607

H
Haojun Liao 已提交
3608 3609 3610
    if (colIndex == TSDB_TBNAME_COLUMN_INDEX) {
      f1 = (char*) TABLE_NAME(pTable1);
      f2 = (char*) TABLE_NAME(pTable2);
3611
      type = TSDB_DATA_TYPE_BINARY;
3612
      bytes = tGetTbnameColumnSchema()->bytes;
3613
    } else {
Y
yihaoDeng 已提交
3614 3615 3616 3617 3618 3619 3620
      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);
      } 
3621
    }
H
Haojun Liao 已提交
3622 3623 3624 3625 3626 3627 3628 3629 3630 3631 3632 3633 3634 3635

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

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

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

3636 3637 3638 3639 3640 3641 3642
    int32_t ret = doCompare(f1, f2, type, bytes);
    if (ret == 0) {
      continue;
    } else {
      return ret;
    }
  }
3643
#endif
3644 3645 3646
  return 0;
}

H
Haojun Liao 已提交
3647
static int tsdbCheckInfoCompar(const void* key1, const void* key2) {
3648
  if (((STableCheckInfo*)key1)->tableId < ((STableCheckInfo*)key2)->tableId) {
H
Haojun Liao 已提交
3649
    return -1;
3650
  } else if (((STableCheckInfo*)key1)->tableId > ((STableCheckInfo*)key2)->tableId) {
H
Haojun Liao 已提交
3651 3652 3653 3654 3655 3656 3657 3658 3659
    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) {
3660
  STable* pTable = taosArrayGetP(pTableList, 0);
H
Haojun Liao 已提交
3661 3662
  SArray* g = taosArrayInit(16, sizeof(STableKeyInfo));

3663
  STableKeyInfo info = {.lastKey = skey};
H
Haojun Liao 已提交
3664
  taosArrayPush(g, &info);
3665

3666
  for (int32_t i = 1; i < numOfTables; ++i) {
3667 3668
    STable** prev = taosArrayGet(pTableList, i - 1);
    STable** p = taosArrayGet(pTableList, i);
H
Haojun Liao 已提交
3669

H
hjxilinx 已提交
3670
    int32_t ret = compareFn(prev, p, pSupp);
3671
    assert(ret == 0 || ret == -1);
H
Haojun Liao 已提交
3672

3673
    if (ret == 0) {
3674
      STableKeyInfo info1 = {.lastKey = skey};
H
Haojun Liao 已提交
3675
      taosArrayPush(g, &info1);
3676 3677
    } else {
      taosArrayPush(pGroups, &g);  // current group is ended, start a new group
H
Haojun Liao 已提交
3678 3679
      g = taosArrayInit(16, sizeof(STableKeyInfo));

3680
      STableKeyInfo info1 = {.lastKey = skey};
H
Haojun Liao 已提交
3681
      taosArrayPush(g, &info1);
3682 3683
    }
  }
H
Haojun Liao 已提交
3684

3685
  taosArrayPush(pGroups, &g);
3686 3687
}

dengyihao's avatar
dengyihao 已提交
3688 3689
SArray* createTableGroup(SArray* pTableList, SSchemaWrapper* pTagSchema, SColIndex* pCols, int32_t numOfOrderCols,
                         TSKEY skey) {
3690
  assert(pTableList != NULL);
3691
  SArray* pTableGroup = taosArrayInit(1, POINTER_BYTES);
H
Haojun Liao 已提交
3692

3693 3694
  size_t size = taosArrayGetSize(pTableList);
  if (size == 0) {
S
Shengliang Guan 已提交
3695
    tsdbDebug("no qualified tables");
3696 3697
    return pTableGroup;
  }
H
Haojun Liao 已提交
3698

dengyihao's avatar
dengyihao 已提交
3699
  if (numOfOrderCols == 0 || size == 1) {  // no group by tags clause or only one table
3700
    SArray* sa = taosArrayDup(pTableList);
H
Haojun Liao 已提交
3701 3702 3703 3704
    if (sa == NULL) {
      taosArrayDestroy(pTableGroup);
      return NULL;
    }
H
Haojun Liao 已提交
3705

3706
    taosArrayPush(pTableGroup, &sa);
S
TD-1057  
Shengliang Guan 已提交
3707
    tsdbDebug("all %" PRIzu " tables belong to one group", size);
3708
  } else {
H
Haojun Liao 已提交
3709 3710
    STableGroupSupporter sup = {0};
    sup.numOfCols = numOfOrderCols;
3711
    sup.pTagSchema = pTagSchema->pSchema;
H
Haojun Liao 已提交
3712 3713
    sup.pCols = pCols;

3714 3715
    taosqsort(pTableList->pData, size, sizeof(STableKeyInfo), &sup, tableGroupComparFn);
    createTableGroupImpl(pTableGroup, pTableList, size, skey, &sup, tableGroupComparFn);
3716
  }
H
Haojun Liao 已提交
3717

3718 3719 3720
  return pTableGroup;
}

dengyihao's avatar
dengyihao 已提交
3721
// static bool tableFilterFp(const void* pNode, void* param) {
3722 3723 3724 3725 3726 3727 3728 3729 3730 3731 3732 3733 3734 3735 3736 3737 3738 3739 3740 3741 3742 3743 3744 3745 3746 3747 3748 3749 3750 3751 3752 3753 3754 3755 3756 3757 3758 3759 3760 3761 3762 3763 3764 3765 3766 3767 3768 3769 3770 3771 3772 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
//  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 已提交
3805

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

dengyihao's avatar
dengyihao 已提交
3809 3810
// static int32_t doQueryTableList(STable* pSTable, SArray* pRes, tExprNode* pExpr) {
//  //  // query according to the expression tree
3811
//  SExprTraverseSupp supp = {
dengyihao's avatar
dengyihao 已提交
3812
//      .nodeFilterFn = (__result_filter_fn_t)tableFilterFp,
3813 3814
//      .setupInfoFn = filterPrepare,
//      .pExtInfo = pSTable->tagSchema,
dengyihao's avatar
dengyihao 已提交
3815
//  };
3816 3817 3818 3819 3820
//
//  getTableListfromSkipList(pExpr, pSTable->pIndex, pRes, &supp);
//  tExprTreeDestroy(pExpr, destroyHelper);
//  return TSDB_CODE_SUCCESS;
//}
3821

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

H
Hongze Cheng 已提交
3827
  metaReaderInit(&mr, (SMeta*)pMeta, 0);
H
Hongze Cheng 已提交
3828 3829

  if (metaGetTableEntryByUid(&mr, uid) < 0) {
dengyihao's avatar
dengyihao 已提交
3830
    tsdbError("%p failed to get stable, uid:%" PRIu64 ", TID:0x%" PRIx64 " QID:0x%" PRIx64, pMeta, uid, taskId, reqId);
H
Hongze Cheng 已提交
3831
    metaReaderClear(&mr);
D
dapan1121 已提交
3832
    terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST;
3833
    goto _error;
C
Cary Xu 已提交
3834 3835
  } else {
    tsdbDebug("%p succeed to get stable, uid:%" PRIu64 ", TID:0x%" PRIx64 " QID:0x%" PRIx64, pMeta, uid, taskId, reqId);
3836
  }
H
Haojun Liao 已提交
3837

H
Hongze Cheng 已提交
3838
  if (mr.me.type != TSDB_SUPER_TABLE) {
dengyihao's avatar
dengyihao 已提交
3839 3840 3841
    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 已提交
3842
    metaReaderClear(&mr);
3843
    goto _error;
H
hjxilinx 已提交
3844
  }
3845

H
Hongze Cheng 已提交
3846 3847
  metaReaderClear(&mr);

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

weixin_48148422's avatar
weixin_48148422 已提交
3852 3853
  // no tags and tbname condition, all child tables of this stable are involved
  if (tbnameCond == NULL && (pTagCond == NULL || len == 0)) {
H
Haojun Liao 已提交
3854
    int32_t ret = getAllTableList(pMeta, uid, res);
3855 3856
    if (ret != TSDB_CODE_SUCCESS) {
      goto _error;
3857
    }
3858

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

dengyihao's avatar
dengyihao 已提交
3862 3863 3864
    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);
3865

3866
    taosArrayDestroy(res);
3867 3868
    return ret;
  }
3869

H
hjxilinx 已提交
3870
  int32_t ret = TSDB_CODE_SUCCESS;
3871

dengyihao's avatar
dengyihao 已提交
3872 3873
  SFilterInfo* filterInfo = NULL;
  ret = filterInitFromNode((SNode*)pTagCond, &filterInfo, 0);
dengyihao's avatar
dengyihao 已提交
3874 3875 3876 3877 3878 3879 3880 3881
  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 已提交
3882 3883
  // 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 已提交
3884 3885 3886 3887 3888

  taosArrayDestroy(res);
  return ret;

_error:
3889
  return terrno;
3890
}
3891

dengyihao's avatar
dengyihao 已提交
3892 3893 3894 3895 3896
int32_t tsdbQueryTableList(void* pMeta, SArray* pRes, void* filterInfo) {
  // impl later

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

H
Hongze Cheng 已提交
3900
  metaReaderInit(&mr, (SMeta*)pMeta, 0);
H
Hongze Cheng 已提交
3901 3902

  if (metaGetTableEntryByUid(&mr, uid) < 0) {
D
dapan1121 已提交
3903
    terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST;
3904
    goto _error;
3905
  }
3906

H
Hongze Cheng 已提交
3907 3908
  metaReaderClear(&mr);

3909 3910
  pGroupInfo->numOfTables = 1;
  pGroupInfo->pGroupList = taosArrayInit(1, POINTER_BYTES);
H
Haojun Liao 已提交
3911

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

3914
  STableKeyInfo info = {.lastKey = startKey, .uid = uid};
H
Haojun Liao 已提交
3915
  taosArrayPush(group, &info);
H
Haojun Liao 已提交
3916

3917
  taosArrayPush(pGroupInfo->pGroupList, &group);
3918
  return TSDB_CODE_SUCCESS;
3919

dengyihao's avatar
dengyihao 已提交
3920
_error:
H
Hongze Cheng 已提交
3921
  metaReaderClear(&mr);
3922
  return terrno;
3923
}
3924

3925
#if 0
3926
int32_t tsdbGetTableGroupFromIdListT(STsdb* tsdb, SArray* pTableIdList, STableGroupInfo* pGroupInfo) {
B
Bomin Zhang 已提交
3927 3928 3929
  if (tsdbRLockRepoMeta(tsdb) < 0) {
    return terrno;
  }
3930 3931 3932 3933

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

B
Bomin Zhang 已提交
3936
  for(int32_t i = 0; i < size; ++i) {
3937 3938 3939 3940 3941 3942 3943 3944 3945 3946 3947
    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 已提交
3948 3949 3950
      tsdbUnlockRepoMeta(tsdb);
      taosArrayDestroy(group);
      return terrno;
3951 3952
    }

H
Haojun Liao 已提交
3953 3954
    STableKeyInfo info = {.pTable = pTable, .lastKey = id->key};
    taosArrayPush(group, &info);
3955 3956
  }

B
Bomin Zhang 已提交
3957 3958 3959 3960
  if (tsdbUnlockRepoMeta(tsdb) < 0) {
    taosArrayDestroy(group);
    return terrno;
  }
3961

sangshuduo's avatar
sangshuduo 已提交
3962
  pGroupInfo->numOfTables = (uint32_t) taosArrayGetSize(group);
B
Bomin Zhang 已提交
3963 3964 3965 3966 3967
  if (pGroupInfo->numOfTables > 0) {
    taosArrayPush(pGroupInfo->pGroupList, &group);
  } else {
    taosArrayDestroy(group);
  }
3968 3969 3970

  return TSDB_CODE_SUCCESS;
}
3971
#endif
3972 3973 3974 3975 3976 3977 3978 3979
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 已提交
3980
    taosMemoryFreeClear(pColInfo->pData);
3981 3982 3983 3984 3985 3986
  }

  taosArrayDestroy(pColumnInfoData);
  return NULL;
}

H
Haojun Liao 已提交
3987 3988 3989 3990 3991 3992
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 已提交
3993
    taosMemoryFreeClear(p->pCompInfo);
H
Haojun Liao 已提交
3994 3995 3996 3997 3998 3999
  }

  taosArrayDestroy(pTableCheckInfo);
  return NULL;
}

H
Haojun Liao 已提交
4000
void tsdbCleanupReadHandle(tsdbReaderT queryHandle) {
4001 4002
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)queryHandle;
  if (pTsdbReadHandle == NULL) {
4003 4004
    return;
  }
4005

4006
  pTsdbReadHandle->pColumns = doFreeColumnInfoData(pTsdbReadHandle->pColumns);
4007

4008
  taosArrayDestroy(pTsdbReadHandle->suppInfo.defaultLoadColumn);
wafwerar's avatar
wafwerar 已提交
4009
  taosMemoryFreeClear(pTsdbReadHandle->pDataBlockInfo);
4010 4011
  taosMemoryFreeClear(pTsdbReadHandle->suppInfo.pstatis);
  taosMemoryFreeClear(pTsdbReadHandle->suppInfo.plist);
4012

4013
  if (!emptyQueryTimewindow(pTsdbReadHandle)) {
dengyihao's avatar
dengyihao 已提交
4014
    //    tsdbMayUnTakeMemSnapshot(pTsdbReadHandle);
4015
  } else {
4016
    assert(pTsdbReadHandle->pTableCheckInfo == NULL);
4017 4018
  }

4019 4020
  if (pTsdbReadHandle->pTableCheckInfo != NULL) {
    pTsdbReadHandle->pTableCheckInfo = destroyTableCheckInfo(pTsdbReadHandle->pTableCheckInfo);
4021
  }
4022

4023
  tsdbDestroyReadH(&pTsdbReadHandle->rhelper);
H
Haojun Liao 已提交
4024

4025 4026
  tdFreeDataCols(pTsdbReadHandle->pDataCols);
  pTsdbReadHandle->pDataCols = NULL;
H
Haojun Liao 已提交
4027

4028 4029
  pTsdbReadHandle->prev = doFreeColumnInfoData(pTsdbReadHandle->prev);
  pTsdbReadHandle->next = doFreeColumnInfoData(pTsdbReadHandle->next);
4030

4031
  SIOCostSummary* pCost = &pTsdbReadHandle->cost;
4032

dengyihao's avatar
dengyihao 已提交
4033 4034 4035 4036
  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 已提交
4037

wafwerar's avatar
wafwerar 已提交
4038
  taosMemoryFreeClear(pTsdbReadHandle);
4039
}
4040

4041
#if 0
H
Haojun Liao 已提交
4042
void tsdbDestroyTableGroup(STableGroupInfo *pGroupList) {
4043 4044 4045 4046 4047 4048 4049 4050 4051 4052
  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);
4053 4054 4055 4056
      if (pTable != NULL) { // in case of handling retrieve data from tsdb
        tsdbUnRefTable(pTable);
      }
      //assert(pTable != NULL);
4057 4058 4059 4060 4061
    }

    taosArrayDestroy(p);
  }

4062
  taosHashCleanup(pGroupList->map);
4063
  taosArrayDestroy(pGroupList->pGroupList);
H
Haojun Liao 已提交
4064
  pGroupList->numOfTables = 0;
4065
}
H
Haojun Liao 已提交
4066 4067 4068 4069 4070 4071 4072

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 已提交
4073
    if (exprTreeApplyFilter(pExpr, pNode, param)) {
H
Haojun Liao 已提交
4074 4075 4076 4077 4078 4079 4080 4081 4082 4083 4084 4085 4086 4087 4088 4089 4090 4091 4092 4093 4094 4095 4096
      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 已提交
4097
    pCond->start       = taosMemoryCalloc(1, sizeof(SEndPoint));
H
Haojun Liao 已提交
4098
    pCond->start->optr = queryColInfo->optr;
Y
yihaoDeng 已提交
4099
    pCond->start->v    = queryColInfo->q;
H
Haojun Liao 已提交
4100
  } else if (optr == TSDB_RELATION_LESS || optr == TSDB_RELATION_LESS_EQUAL) {
wafwerar's avatar
wafwerar 已提交
4101
    pCond->end       = taosMemoryCalloc(1, sizeof(SEndPoint));
H
Haojun Liao 已提交
4102
    pCond->end->optr = queryColInfo->optr;
Y
yihaoDeng 已提交
4103 4104
    pCond->end->v    = queryColInfo->q;
  } else if (optr == TSDB_RELATION_IN) {
wafwerar's avatar
wafwerar 已提交
4105
    pCond->start       = taosMemoryCalloc(1, sizeof(SEndPoint));
Y
yihaoDeng 已提交
4106 4107 4108
    pCond->start->optr = queryColInfo->optr;
    pCond->start->v    = queryColInfo->q; 
  } else if (optr == TSDB_RELATION_LIKE) {
H
Haojun Liao 已提交
4109
    assert(0);
4110 4111
  } else if (optr == TSDB_RELATION_MATCH) {
    assert(0);
4112 4113
  } else if (optr == TSDB_RELATION_NMATCH) {
    assert(0);
H
Haojun Liao 已提交
4114 4115 4116 4117 4118 4119 4120 4121 4122 4123 4124 4125 4126 4127 4128 4129 4130 4131 4132 4133 4134 4135 4136 4137 4138 4139 4140 4141 4142 4143 4144 4145 4146 4147 4148 4149 4150 4151 4152 4153 4154 4155 4156 4157 4158 4159 4160 4161 4162 4163 4164 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
  }

  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 已提交
4197 4198 4199 4200 4201 4202 4203 4204 4205 4206 4207 4208 4209
    } 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 已提交
4210 4211 4212 4213 4214 4215 4216 4217 4218 4219 4220 4221 4222 4223 4224 4225 4226 4227 4228 4229 4230 4231 4232 4233 4234 4235 4236 4237 4238 4239 4240 4241 4242 4243 4244 4245 4246 4247 4248 4249 4250
    } 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 已提交
4251 4252
  taosMemoryFree(cond.start);
  taosMemoryFree(cond.end);
H
Haojun Liao 已提交
4253 4254 4255 4256 4257 4258 4259 4260 4261 4262 4263 4264 4265 4266 4267 4268 4269 4270
  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);
4271 4272 4273
      } else if (pQueryInfo->optr == TSDB_RELATION_LIKE ||
                 pQueryInfo->optr == TSDB_RELATION_MATCH ||
                 pQueryInfo->optr == TSDB_RELATION_NMATCH) {
H
Haojun Liao 已提交
4274 4275 4276 4277 4278 4279 4280 4281 4282 4283 4284 4285 4286 4287 4288 4289
        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 已提交
4290 4291 4292 4293 4294 4295 4296 4297 4298 4299 4300 4301 4302 4303 4304 4305 4306 4307 4308 4309 4310 4311 4312 4313 4314 4315 4316 4317 4318 4319 4320 4321 4322
//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 已提交
4323
#endif