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

H
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);
H
Hongze Cheng 已提交
326
  return now - (tsTickPerDay[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("%p rsma level %d is selected to query\n", pReadHandle, level);
C
Cary Xu 已提交
376 377
      return VND_RSMA0(pVnode);
    } else if (level == TSDB_RETENTION_L1) {
C
Cary Xu 已提交
378
      tsdbDebug("%p rsma level %d is selected to query\n", pReadHandle, level);
C
Cary Xu 已提交
379 380
      return VND_RSMA1(pVnode);
    } else {
C
Cary Xu 已提交
381
      tsdbDebug("%p rsma level %d is selected to query\n", pReadHandle, level);
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

493
  pTsdbReadHandle->pSchema = metaGetTbTSchema(pVnode->pMeta, pCheckInfo->tableId, 0);
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 * tsTickPerDay[precision]);
D
dapan1121 已提交
1051
  }
dengyihao's avatar
dengyihao 已提交
1052

1053
  int64_t fid = (int64_t)(key / (daysPerFile * tsTickPerDay[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 1279
static int32_t doCopyRowsFromFileBlock(STsdbReadHandle* pTsdbReadHandle, int32_t capacity, int32_t numOfRows,
                                       int32_t start, int32_t end);
static void    moveDataToFront(STsdbReadHandle* pTsdbReadHandle, int32_t numOfRows, int32_t numOfCols);
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) {
1280
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
H
Hongze Cheng 已提交
1281
  STsdbCfg*      pCfg = REPO_CFG(pTsdbReadHandle->pTsdb);
H
Haojun Liao 已提交
1282
  SDataBlockInfo binfo = GET_FILE_DATA_BLOCK_INFO(pCheckInfo, pBlock);
1283
  TSKEY          key;
dengyihao's avatar
dengyihao 已提交
1284
  int32_t        code = TSDB_CODE_SUCCESS;
1285

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

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

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

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

1299 1300 1301 1302
  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 已提交
1303
      // do not load file block into buffer
1304
      int32_t step = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 1 : -1;
H
Haojun Liao 已提交
1305

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

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

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

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

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

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

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

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

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

  return code;
1373 1374
}

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

1493 1494 1495
  return midPos;
}

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

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

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

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

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

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

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

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

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

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

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

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

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

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

1577
  return numOfRows + num;
1578 1579
}

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

1793 1794
static void moveDataToFront(STsdbReadHandle* pTsdbReadHandle, int32_t numOfRows, int32_t numOfCols) {
  if (numOfRows == 0 || ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
1795 1796 1797 1798
    return;
  }

  // if the buffer is not full in case of descending order query, move the data in the front of the buffer
1799 1800
  if (numOfRows < pTsdbReadHandle->outputCapacity) {
    int32_t emptySize = pTsdbReadHandle->outputCapacity - numOfRows;
dengyihao's avatar
dengyihao 已提交
1801
    for (int32_t i = 0; i < numOfCols; ++i) {
1802
      SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
dengyihao's avatar
dengyihao 已提交
1803 1804
      memmove((char*)pColInfo->pData, (char*)pColInfo->pData + emptySize * pColInfo->info.bytes,
              numOfRows * pColInfo->info.bytes);
1805 1806 1807 1808
    }
  }
}

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

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

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

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

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

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

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

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

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

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

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

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

dengyihao's avatar
dengyihao 已提交
1873
  int32_t step = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 1 : -1;
1874
  int32_t numOfCols = (int32_t)(QH_GET_NUM_OF_COLS(pTsdbReadHandle));
H
Haojun Liao 已提交
1875 1876 1877 1878 1879 1880

  int32_t pos = cur->pos;

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

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

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

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

  // if the buffer is not full in case of descending order query, move the data in the front of the buffer
1895
  moveDataToFront(pTsdbReadHandle, numOfRows, numOfCols);
H
Haojun Liao 已提交
1896 1897 1898

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

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

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

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

1915
  if (ASCENDING_TRAVERSE(pTsdbReadHandle->order) && pTsdbReadHandle->window.ekey >= pBlockInfo->window.ekey) {
H
Haojun Liao 已提交
1916 1917
    endPos = pBlockInfo->rows - 1;
    cur->mixBlock = (cur->pos != 0);
1918
  } else if (!ASCENDING_TRAVERSE(pTsdbReadHandle->order) && pTsdbReadHandle->window.ekey <= pBlockInfo->window.skey) {
H
Haojun Liao 已提交
1919 1920 1921 1922
    endPos = 0;
    cur->mixBlock = (cur->pos != pBlockInfo->rows - 1);
  } else {
    assert(pCols->numOfRows > 0);
1923
    endPos = doBinarySearchKey(pCols->cols[0].pData, pCols->numOfRows, pTsdbReadHandle->window.ekey, order);
H
Haojun Liao 已提交
1924 1925 1926 1927 1928 1929
    cur->mixBlock = true;
  }

  return endPos;
}

H
[td-32]  
hjxilinx 已提交
1930 1931
// 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
1932 1933
static void doMergeTwoLevelData(STsdbReadHandle* pTsdbReadHandle, STableCheckInfo* pCheckInfo, SBlock* pBlock) {
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
1934
  SDataBlockInfo blockInfo = GET_FILE_DATA_BLOCK_INFO(pCheckInfo, pBlock);
H
Hongze Cheng 已提交
1935
  STsdbCfg*      pCfg = REPO_CFG(pTsdbReadHandle->pTsdb);
H
Haojun Liao 已提交
1936

1937
  initTableMemIterator(pTsdbReadHandle, pCheckInfo);
1938

1939 1940
  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 已提交
1941
         cur->pos >= 0 && cur->pos < pBlock->numOfRows);
H
Haojun Liao 已提交
1942

1943
  TSKEY* tsArray = pCols->cols[0].pData;
dengyihao's avatar
dengyihao 已提交
1944 1945
  assert(pCols->numOfRows == pBlock->numOfRows && tsArray[0] == pBlock->keyFirst &&
         tsArray[pBlock->numOfRows - 1] == pBlock->keyLast);
1946 1947

  // for search the endPos, so the order needs to reverse
dengyihao's avatar
dengyihao 已提交
1948
  int32_t order = (pTsdbReadHandle->order == TSDB_ORDER_ASC) ? TSDB_ORDER_DESC : TSDB_ORDER_ASC;
1949

dengyihao's avatar
dengyihao 已提交
1950
  int32_t step = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 1 : -1;
1951
  int32_t numOfCols = (int32_t)(QH_GET_NUM_OF_COLS(pTsdbReadHandle));
1952

H
Haojun Liao 已提交
1953
  STable* pTable = NULL;
1954
  int32_t endPos = getEndPosInDataBlock(pTsdbReadHandle, &blockInfo);
H
Haojun Liao 已提交
1955

H
Hongze Cheng 已提交
1956 1957
  tsdbDebug("%p uid:%" PRIu64 " start merge data block, file block range:%" PRIu64 "-%" PRIu64
            " rows:%d, start:%d, end:%d, %s",
dengyihao's avatar
dengyihao 已提交
1958 1959
            pTsdbReadHandle, pCheckInfo->tableId, blockInfo.window.skey, blockInfo.window.ekey, blockInfo.rows,
            cur->pos, endPos, pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
1960

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

dengyihao's avatar
dengyihao 已提交
1965 1966
  int16_t   rv1 = -1;
  int16_t   rv2 = -1;
1967 1968
  STSchema* pSchema1 = NULL;
  STSchema* pSchema2 = NULL;
D
fix bug  
dapan1121 已提交
1969

H
Haojun Liao 已提交
1970 1971
  int32_t pos = cur->pos;
  cur->win = TSWINDOW_INITIALIZER;
1972

1973 1974
  // no data in buffer, load data from file directly
  if (pCheckInfo->iiter == NULL && pCheckInfo->iter == NULL) {
1975
    copyAllRemainRowsFromFileBlock(pTsdbReadHandle, pCheckInfo, &blockInfo, endPos);
1976
    return;
1977
  } else if (pCheckInfo->iter != NULL || pCheckInfo->iiter != NULL) {
1978
    SSkipListNode* node = NULL;
C
Cary Xu 已提交
1979
    TSKEY          lastKeyAppend = TSKEY_INITIAL_VAL;
C
Cary Xu 已提交
1980

1981
    do {
H
Haojun Liao 已提交
1982
      STSRow* row2 = NULL;
C
Cary Xu 已提交
1983
      STSRow* row1 = getSRowInTableMem(pCheckInfo, pTsdbReadHandle->order, pCfg->update, &row2, TD_VER_MAX);
1984
      if (row1 == NULL) {
H
[td-32]  
hjxilinx 已提交
1985
        break;
1986
      }
1987

H
Haojun Liao 已提交
1988
      TSKEY key = TD_ROW_KEY(row1);
1989 1990
      if ((key > pTsdbReadHandle->window.ekey && ASCENDING_TRAVERSE(pTsdbReadHandle->order)) ||
          (key < pTsdbReadHandle->window.ekey && !ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
1991 1992 1993
        break;
      }

dengyihao's avatar
dengyihao 已提交
1994 1995 1996 1997
      if (((pos > endPos || tsArray[pos] > pTsdbReadHandle->window.ekey) &&
           ASCENDING_TRAVERSE(pTsdbReadHandle->order)) ||
          ((pos < endPos || tsArray[pos] < pTsdbReadHandle->window.ekey) &&
           !ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
1998 1999 2000
        break;
      }

2001 2002
      if ((key < tsArray[pos] && ASCENDING_TRAVERSE(pTsdbReadHandle->order)) ||
          (key > tsArray[pos] && !ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
H
Haojun Liao 已提交
2003
        if (rv1 != TD_ROW_SVER(row1)) {
dengyihao's avatar
dengyihao 已提交
2004
          //          pSchema1 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row1));
H
Haojun Liao 已提交
2005
          rv1 = TD_ROW_SVER(row1);
C
Cary Xu 已提交
2006
        }
dengyihao's avatar
dengyihao 已提交
2007 2008
        if (row2 && rv2 != TD_ROW_SVER(row2)) {
          //          pSchema2 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row2));
H
Haojun Liao 已提交
2009
          rv2 = TD_ROW_SVER(row2);
2010
        }
dengyihao's avatar
dengyihao 已提交
2011

C
Cary Xu 已提交
2012 2013 2014
        numOfRows +=
            mergeTwoRowFromMem(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, &curRow, row1, row2, numOfCols,
                               pCheckInfo->tableId, pSchema1, pSchema2, pCfg->update, &lastKeyAppend);
2015 2016 2017
        if (cur->win.skey == TSKEY_INITIAL_VAL) {
          cur->win.skey = key;
        }
2018

2019
        cur->win.ekey = key;
dengyihao's avatar
dengyihao 已提交
2020
        cur->lastKey = key + step;
2021
        cur->mixBlock = true;
2022
        moveToNextRowInMem(pCheckInfo);
2023
      } else if (key == tsArray[pos]) {  // data in buffer has the same timestamp of data in file block, ignore it
C
Cary Xu 已提交
2024
#if 0
H
TD-1439  
Hongze Cheng 已提交
2025
        if (pCfg->update) {
dengyihao's avatar
dengyihao 已提交
2026
          if (pCfg->update == TD_ROW_PARTIAL_UPDATE) {
2027
            doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, numOfRows, pos, pos);
D
fix bug  
dapan1121 已提交
2028
          }
H
Haojun Liao 已提交
2029
          if (rv1 != TD_ROW_SVER(row1)) {
dengyihao's avatar
dengyihao 已提交
2030
            //            pSchema1 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row1));
H
Haojun Liao 已提交
2031
            rv1 = TD_ROW_SVER(row1);
2032
          }
dengyihao's avatar
dengyihao 已提交
2033 2034
          if (row2 && rv2 != TD_ROW_SVER(row2)) {
            //            pSchema2 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row2));
H
Haojun Liao 已提交
2035
            rv2 = TD_ROW_SVER(row2);
2036
          }
dengyihao's avatar
dengyihao 已提交
2037

2038
          bool forceSetNull = pCfg->update != TD_ROW_PARTIAL_UPDATE;
dengyihao's avatar
dengyihao 已提交
2039
          mergeTwoRowFromMem(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, numOfRows, row1, row2, numOfCols,
C
Cary Xu 已提交
2040
                             pCheckInfo->tableId, pSchema1, pSchema2, forceSetNull, &lastRowKey);
H
TD-1439  
Hongze Cheng 已提交
2041 2042 2043 2044 2045 2046 2047 2048 2049
          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 已提交
2050 2051 2052 2053 2054 2055 2056
          moveToNextRowInMem(pCheckInfo);
          pos += step;
        } else {
          moveToNextRowInMem(pCheckInfo);
        }
#endif
        if (TD_SUPPORT_UPDATE(pCfg->update)) {
C
Cary Xu 已提交
2057
          if (lastKeyAppend != key) {
C
Cary Xu 已提交
2058 2059 2060 2061
            lastKeyAppend = key;
            ++curRow;
          }
          numOfRows = doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, curRow, pos, pos);
C
Cary Xu 已提交
2062 2063 2064 2065 2066 2067 2068 2069 2070

          if (rv1 != TD_ROW_SVER(row1)) {
            //            pSchema1 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row1));
            rv1 = TD_ROW_SVER(row1);
          }
          if (row2 && rv2 != TD_ROW_SVER(row2)) {
            //            pSchema2 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row2));
            rv2 = TD_ROW_SVER(row2);
          }
C
Cary Xu 已提交
2071 2072 2073
          numOfRows +=
              mergeTwoRowFromMem(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, &curRow, row1, row2, numOfCols,
                                 pCheckInfo->tableId, pSchema1, pSchema2, pCfg->update, &lastKeyAppend);
C
Cary Xu 已提交
2074

C
Cary Xu 已提交
2075 2076 2077 2078 2079 2080 2081 2082
          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 已提交
2083 2084 2085 2086 2087
          moveToNextRowInMem(pCheckInfo);
          pos += step;
        } else {
          moveToNextRowInMem(pCheckInfo);
        }
2088
      } else if ((key > tsArray[pos] && ASCENDING_TRAVERSE(pTsdbReadHandle->order)) ||
dengyihao's avatar
dengyihao 已提交
2089
                 (key < tsArray[pos] && !ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
2090 2091 2092
        if (cur->win.skey == TSKEY_INITIAL_VAL) {
          cur->win.skey = tsArray[pos];
        }
2093

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

dengyihao's avatar
dengyihao 已提交
2097
        if (tsArray[end] == key) {  // the value of key in cache equals to the end timestamp value, ignore it
C
Cary Xu 已提交
2098
#if 0
2099
          if (pCfg->update == TD_ROW_DISCARD_UPDATE) {
H
Hongze Cheng 已提交
2100 2101 2102 2103
            moveToNextRowInMem(pCheckInfo);
          } else {
            end -= step;
          }
C
Cary Xu 已提交
2104 2105 2106 2107 2108 2109
#endif
          if (!TD_SUPPORT_UPDATE(pCfg->update)) {
            moveToNextRowInMem(pCheckInfo);
          } else {
            end -= step;
          }
H
Haojun Liao 已提交
2110
        }
2111

2112
        int32_t qstart = 0, qend = 0;
2113
        getQualifiedRowsPos(pTsdbReadHandle, pos, end, numOfRows, &qstart, &qend);
2114

C
Cary Xu 已提交
2115 2116 2117 2118 2119
        if ((lastKeyAppend != TSKEY_INITIAL_VAL) &&
            (lastKeyAppend != (ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? tsArray[qstart] : tsArray[qend]))) {
          ++curRow;
        }
        numOfRows = doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, curRow, qstart, qend);
2120
        pos += (qend - qstart + 1) * step;
C
Cary Xu 已提交
2121 2122 2123
        if (numOfRows > 0) {
          curRow = numOfRows - 1;
        }
2124

dengyihao's avatar
dengyihao 已提交
2125 2126
        cur->win.ekey = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? tsArray[qend] : tsArray[qstart];
        cur->lastKey = cur->win.ekey + step;
C
Cary Xu 已提交
2127
        lastKeyAppend = cur->win.ekey;
2128
      }
2129
    } while (numOfRows < pTsdbReadHandle->outputCapacity);
H
Haojun Liao 已提交
2130

2131
    if (numOfRows < pTsdbReadHandle->outputCapacity) {
H
Haojun Liao 已提交
2132 2133 2134 2135
      /**
       * 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.
       */
2136
      if (node == NULL ||
H
Haojun Liao 已提交
2137
          ((TD_ROW_KEY((STSRow*)SL_GET_NODE_DATA(node)) > pTsdbReadHandle->window.ekey) &&
2138
           ASCENDING_TRAVERSE(pTsdbReadHandle->order)) ||
H
Haojun Liao 已提交
2139
          ((TD_ROW_KEY((STSRow*)SL_GET_NODE_DATA(node)) < pTsdbReadHandle->window.ekey) &&
2140
           !ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
2141 2142 2143 2144 2145
        // 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];
        }

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

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

dengyihao's avatar
dengyihao 已提交
2152 2153
        cur->win.ekey = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? tsArray[end] : tsArray[start];
        cur->lastKey = cur->win.ekey + step;
H
Haojun Liao 已提交
2154
        cur->mixBlock = true;
2155
      }
2156 2157
    }
  }
H
Haojun Liao 已提交
2158 2159

  cur->blockCompleted =
2160 2161
      (((pos > endPos || cur->lastKey > pTsdbReadHandle->window.ekey) && ASCENDING_TRAVERSE(pTsdbReadHandle->order)) ||
       ((pos < endPos || cur->lastKey < pTsdbReadHandle->window.ekey) && !ASCENDING_TRAVERSE(pTsdbReadHandle->order)));
2162

2163
  if (!ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
wafwerar's avatar
wafwerar 已提交
2164
    TSWAP(cur->win.skey, cur->win.ekey);
2165
  }
2166

2167 2168 2169
  moveDataToFront(pTsdbReadHandle, numOfRows, numOfCols);
  updateInfoAfterMerge(pTsdbReadHandle, pCheckInfo, numOfRows, pos);
  doCheckGeneratedBlockRange(pTsdbReadHandle);
H
Haojun Liao 已提交
2170

dengyihao's avatar
dengyihao 已提交
2171 2172 2173
  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);
2174 2175
}

2176
int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order) {
H
[td-32]  
hjxilinx 已提交
2177
  int    firstPos, lastPos, midPos = -1;
H
Haojun Liao 已提交
2178
  int    numOfRows;
2179 2180
  TSKEY* keyList;

H
[td-32]  
hjxilinx 已提交
2181
  if (num <= 0) return -1;
2182 2183

  keyList = (TSKEY*)pValue;
H
[td-32]  
hjxilinx 已提交
2184 2185
  firstPos = 0;
  lastPos = num - 1;
2186

2187
  if (order == TSDB_ORDER_DESC) {
H
[td-32]  
hjxilinx 已提交
2188 2189 2190 2191 2192
    // 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;
2193

H
Haojun Liao 已提交
2194 2195
      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1) + firstPos;
2196

H
[td-32]  
hjxilinx 已提交
2197 2198 2199 2200 2201 2202 2203 2204
      if (key < keyList[midPos]) {
        lastPos = midPos - 1;
      } else if (key > keyList[midPos]) {
        firstPos = midPos + 1;
      } else {
        break;
      }
    }
2205

H
[td-32]  
hjxilinx 已提交
2206 2207 2208 2209 2210
  } 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;
2211

H
[td-32]  
hjxilinx 已提交
2212 2213 2214 2215 2216 2217 2218
      if (key > keyList[lastPos]) {
        lastPos = lastPos + 1;
        if (lastPos >= num)
          return -1;
        else
          return lastPos;
      }
2219

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

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

H
[td-32]  
hjxilinx 已提交
2233 2234 2235
  return midPos;
}

2236
static void cleanBlockOrderSupporter(SBlockOrderSupporter* pSupporter, int32_t numOfTables) {
wafwerar's avatar
wafwerar 已提交
2237 2238
  taosMemoryFreeClear(pSupporter->numOfBlocksPerTable);
  taosMemoryFreeClear(pSupporter->blockIndexArray);
2239 2240

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

wafwerar's avatar
wafwerar 已提交
2245
  taosMemoryFreeClear(pSupporter->pDataBlockInfo);
2246 2247 2248 2249 2250 2251 2252 2253 2254 2255 2256
}

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

2257
  if (leftTableBlockIndex > pSupporter->numOfBlocksPerTable[leftTableIndex]) {
2258 2259
    /* left block is empty */
    return 1;
2260
  } else if (rightTableBlockIndex > pSupporter->numOfBlocksPerTable[rightTableIndex]) {
2261 2262 2263 2264 2265 2266 2267
    /* right block is empty */
    return -1;
  }

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

H
Haojun Liao 已提交
2268
  //    assert(pLeftBlockInfoEx->compBlock->offset != pRightBlockInfoEx->compBlock->offset);
dengyihao's avatar
dengyihao 已提交
2269
#if 0  // TODO: temporarily comment off requested by Dr. Liao
H
Haojun Liao 已提交
2270 2271
  if (pLeftBlockInfoEx->compBlock->offset == pRightBlockInfoEx->compBlock->offset &&
      pLeftBlockInfoEx->compBlock->last == pRightBlockInfoEx->compBlock->last) {
B
Bomin Zhang 已提交
2272
    tsdbError("error in header file, two block with same offset:%" PRId64, (int64_t)pLeftBlockInfoEx->compBlock->offset);
2273
  }
H
Haojun Liao 已提交
2274
#endif
2275

H
Haojun Liao 已提交
2276
  return pLeftBlockInfoEx->compBlock->offset > pRightBlockInfoEx->compBlock->offset ? 1 : -1;
2277 2278
}

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

2282 2283
  if (pTsdbReadHandle->allocSize < size) {
    pTsdbReadHandle->allocSize = (int32_t)size;
wafwerar's avatar
wafwerar 已提交
2284
    char* tmp = taosMemoryRealloc(pTsdbReadHandle->pDataBlockInfo, pTsdbReadHandle->allocSize);
H
Haojun Liao 已提交
2285 2286 2287 2288
    if (tmp == NULL) {
      return TSDB_CODE_TDB_OUT_OF_MEMORY;
    }

dengyihao's avatar
dengyihao 已提交
2289
    pTsdbReadHandle->pDataBlockInfo = (STableBlockInfo*)tmp;
2290 2291
  }

2292
  memset(pTsdbReadHandle->pDataBlockInfo, 0, size);
2293 2294
  *numOfAllocBlocks = numOfBlocks;

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

2298 2299
  SBlockOrderSupporter sup = {0};
  sup.numOfTables = numOfTables;
wafwerar's avatar
wafwerar 已提交
2300 2301 2302
  sup.numOfBlocksPerTable = taosMemoryCalloc(1, sizeof(int32_t) * numOfTables);
  sup.blockIndexArray = taosMemoryCalloc(1, sizeof(int32_t) * numOfTables);
  sup.pDataBlockInfo = taosMemoryCalloc(1, POINTER_BYTES * numOfTables);
2303

2304
  if (sup.numOfBlocksPerTable == NULL || sup.blockIndexArray == NULL || sup.pDataBlockInfo == NULL) {
2305
    cleanBlockOrderSupporter(&sup, 0);
2306
    return TSDB_CODE_TDB_OUT_OF_MEMORY;
2307
  }
H
Haojun Liao 已提交
2308

2309
  int32_t cnt = 0;
2310
  int32_t numOfQualTables = 0;
H
Haojun Liao 已提交
2311

2312
  for (int32_t j = 0; j < numOfTables; ++j) {
2313
    STableCheckInfo* pTableCheck = (STableCheckInfo*)taosArrayGet(pTsdbReadHandle->pTableCheckInfo, j);
2314 2315 2316
    if (pTableCheck->numOfBlocks <= 0) {
      continue;
    }
H
Haojun Liao 已提交
2317

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

wafwerar's avatar
wafwerar 已提交
2321
    char* buf = taosMemoryMalloc(sizeof(STableBlockInfo) * pTableCheck->numOfBlocks);
2322
    if (buf == NULL) {
2323
      cleanBlockOrderSupporter(&sup, numOfQualTables);
2324
      return TSDB_CODE_TDB_OUT_OF_MEMORY;
2325 2326
    }

2327
    sup.pDataBlockInfo[numOfQualTables] = (STableBlockInfo*)buf;
2328 2329

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

H
Haojun Liao 已提交
2332 2333
      pBlockInfo->compBlock = &pBlock[k];
      pBlockInfo->pTableCheckInfo = pTableCheck;
2334 2335 2336
      cnt++;
    }

2337
    numOfQualTables++;
2338 2339
  }

H
Haojun Liao 已提交
2340
  assert(numOfBlocks == cnt);
2341

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

H
Haojun Liao 已提交
2347
    tsdbDebug("%p create data blocks info struct completed for 1 table, %d blocks not sorted %s", pTsdbReadHandle, cnt,
dengyihao's avatar
dengyihao 已提交
2348
              pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
2349 2350
    return TSDB_CODE_SUCCESS;
  }
2351

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

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

2358
  SMultiwayMergeTreeInfo* pTree = NULL;
dengyihao's avatar
dengyihao 已提交
2359
  uint8_t                 ret = tMergeTreeCreate(&pTree, sup.numOfTables, &sup, dataBlockOrderCompar);
2360 2361
  if (ret != TSDB_CODE_SUCCESS) {
    cleanBlockOrderSupporter(&sup, numOfTables);
2362
    return TSDB_CODE_TDB_OUT_OF_MEMORY;
2363 2364 2365 2366 2367
  }

  int32_t numOfTotal = 0;

  while (numOfTotal < cnt) {
2368
    int32_t pos = tMergeTreeGetChosenIndex(pTree);
2369 2370
    int32_t index = sup.blockIndexArray[pos]++;

H
Haojun Liao 已提交
2371
    STableBlockInfo* pBlocksInfo = sup.pDataBlockInfo[pos];
2372
    pTsdbReadHandle->pDataBlockInfo[numOfTotal++] = pBlocksInfo[index];
2373 2374

    // set data block index overflow, in order to disable the offset comparator
2375 2376
    if (sup.blockIndexArray[pos] >= sup.numOfBlocksPerTable[pos]) {
      sup.blockIndexArray[pos] = sup.numOfBlocksPerTable[pos] + 1;
2377
    }
2378

H
Haojun Liao 已提交
2379
    tMergeTreeAdjust(pTree, tMergeTreeGetAdjustIndex(pTree));
2380 2381 2382 2383 2384
  }

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

H
Haojun Liao 已提交
2389
  tsdbDebug("%p %d data blocks sort completed, %s", pTsdbReadHandle, cnt, pTsdbReadHandle->idStr);
2390
  cleanBlockOrderSupporter(&sup, numOfTables);
wafwerar's avatar
wafwerar 已提交
2391
  taosMemoryFree(pTree);
2392 2393 2394 2395

  return TSDB_CODE_SUCCESS;
}

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

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

dengyihao's avatar
dengyihao 已提交
2402
  while (1) {
2403
    int32_t code = loadFileDataBlock(pTsdbReadHandle, pNext->compBlock, pNext->pTableCheckInfo, exists);
H
Haojun Liao 已提交
2404 2405 2406 2407
    if (code != TSDB_CODE_SUCCESS || *exists) {
      return code;
    }

2408 2409
    if ((cur->slot == pTsdbReadHandle->numOfBlocks - 1 && ASCENDING_TRAVERSE(pTsdbReadHandle->order)) ||
        (cur->slot == 0 && !ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
H
Haojun Liao 已提交
2410
      // all data blocks in current file has been checked already, try next file if exists
2411
      return getFirstFileDataBlock(pTsdbReadHandle, exists);
H
Haojun Liao 已提交
2412 2413 2414 2415
    } else {  // next block of the same file
      cur->slot += step;
      cur->mixBlock = false;
      cur->blockCompleted = false;
2416
      pNext = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
H
Haojun Liao 已提交
2417 2418 2419 2420
    }
  }
}

2421 2422 2423
static int32_t getFirstFileDataBlock(STsdbReadHandle* pTsdbReadHandle, bool* exists) {
  pTsdbReadHandle->numOfBlocks = 0;
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
H
Haojun Liao 已提交
2424 2425 2426

  int32_t code = TSDB_CODE_SUCCESS;

2427
  int32_t numOfBlocks = 0;
2428
  int32_t numOfTables = (int32_t)taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
2429

C
Cary Xu 已提交
2430
  STsdbKeepCfg* pCfg = REPO_KEEP_CFG(pTsdbReadHandle->pTsdb);
C
Cary Xu 已提交
2431
  STimeWindow   win = TSWINDOW_INITIALIZER;
2432

H
Hongze Cheng 已提交
2433
  while (true) {
2434
    tsdbRLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2435

2436 2437
    if ((pTsdbReadHandle->pFileGroup = tsdbFSIterNext(&pTsdbReadHandle->fileIter)) == NULL) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2438 2439 2440
      break;
    }

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

    // current file are not overlapped with query time window, ignore remain files
2444 2445 2446
    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 已提交
2447 2448
      tsdbDebug("%p remain files are not qualified for qrange:%" PRId64 "-%" PRId64 ", ignore, %s", pTsdbReadHandle,
                pTsdbReadHandle->window.skey, pTsdbReadHandle->window.ekey, pTsdbReadHandle->idStr);
2449 2450
      pTsdbReadHandle->pFileGroup = NULL;
      assert(pTsdbReadHandle->numOfBlocks == 0);
2451 2452 2453
      break;
    }

2454 2455
    if (tsdbSetAndOpenReadFSet(&pTsdbReadHandle->rhelper, pTsdbReadHandle->pFileGroup) < 0) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2456 2457 2458 2459
      code = terrno;
      break;
    }

2460
    tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2461

2462
    if (tsdbLoadBlockIdx(&pTsdbReadHandle->rhelper) < 0) {
H
Hongze Cheng 已提交
2463 2464 2465 2466
      code = terrno;
      break;
    }

2467
    if ((code = getFileCompInfo(pTsdbReadHandle, &numOfBlocks)) != TSDB_CODE_SUCCESS) {
2468 2469
      break;
    }
H
Haojun Liao 已提交
2470

H
Haojun Liao 已提交
2471 2472
    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 已提交
2473

2474 2475 2476 2477
    assert(numOfBlocks >= 0);
    if (numOfBlocks == 0) {
      continue;
    }
H
Haojun Liao 已提交
2478

2479
    // todo return error code to query engine
dengyihao's avatar
dengyihao 已提交
2480 2481
    if ((code = createDataBlocksInfo(pTsdbReadHandle, numOfBlocks, &pTsdbReadHandle->numOfBlocks)) !=
        TSDB_CODE_SUCCESS) {
2482 2483
      break;
    }
H
Haojun Liao 已提交
2484

2485 2486
    assert(numOfBlocks >= pTsdbReadHandle->numOfBlocks);
    if (pTsdbReadHandle->numOfBlocks > 0) {
2487 2488 2489
      break;
    }
  }
H
Haojun Liao 已提交
2490

2491
  // no data in file anymore
2492
  if (pTsdbReadHandle->numOfBlocks <= 0 || code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2493
    if (code == TSDB_CODE_SUCCESS) {
2494
      assert(pTsdbReadHandle->pFileGroup == NULL);
H
Haojun Liao 已提交
2495 2496
    }

D
dapan1121 已提交
2497
    cur->fid = INT32_MIN;  // denote that there are no data in file anymore
H
Haojun Liao 已提交
2498 2499
    *exists = false;
    return code;
2500
  }
H
Haojun Liao 已提交
2501

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

2506 2507
  STableBlockInfo* pBlockInfo = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
  return getDataBlockRv(pTsdbReadHandle, pBlockInfo, exists);
H
Haojun Liao 已提交
2508 2509 2510 2511 2512 2513 2514
}

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

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

2518 2519
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
  assert(cur->slot < pTsdbReadHandle->numOfBlocks && cur->slot >= 0);
H
Haojun Liao 已提交
2520 2521

  cur->slot += step;
dengyihao's avatar
dengyihao 已提交
2522
  cur->mixBlock = false;
H
Haojun Liao 已提交
2523
  cur->blockCompleted = false;
2524
}
H
Haojun Liao 已提交
2525 2526

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

H
Haojun Liao 已提交
2529
  pTableBlockInfo->totalSize = 0;
2530
  pTableBlockInfo->totalRows = 0;
H
Haojun Liao 已提交
2531

2532
  STsdbFS* pFileHandle = REPO_FS(pTsdbReadHandle->pTsdb);
H
Haojun Liao 已提交
2533 2534

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

  tsdbRLockFS(pFileHandle);
2540 2541
  tsdbFSIterInit(&pTsdbReadHandle->fileIter, pFileHandle, pTsdbReadHandle->order);
  tsdbFSIterSeek(&pTsdbReadHandle->fileIter, fid);
H
Haojun Liao 已提交
2542 2543
  tsdbUnLockFS(pFileHandle);

H
Haojun Liao 已提交
2544
  pTableBlockInfo->numOfFiles += 1;
H
Haojun Liao 已提交
2545

H
Haojun Liao 已提交
2546
  int32_t     code = TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
2547
  int32_t     numOfBlocks = 0;
2548
  int32_t     numOfTables = (int32_t)taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
dengyihao's avatar
dengyihao 已提交
2549
  int         defaultRows = 4096;  // TSDB_DEFAULT_BLOCK_ROWS(pCfg->maxRowsPerFileBlock);
H
Haojun Liao 已提交
2550 2551
  STimeWindow win = TSWINDOW_INITIALIZER;

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

H
Haojun Liao 已提交
2554 2555
  while (true) {
    numOfBlocks = 0;
2556
    tsdbRLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2557

2558 2559
    if ((pTsdbReadHandle->pFileGroup = tsdbFSIterNext(&pTsdbReadHandle->fileIter)) == NULL) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2560 2561 2562
      break;
    }

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

    // current file are not overlapped with query time window, ignore remain files
dengyihao's avatar
dengyihao 已提交
2566 2567
    if ((ascTraverse && win.skey > pTsdbReadHandle->window.ekey) ||
        (!ascTraverse && win.ekey < pTsdbReadHandle->window.ekey)) {
2568
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2569 2570
      tsdbDebug("%p remain files are not qualified for qrange:%" PRId64 "-%" PRId64 ", ignore, %s", pTsdbReadHandle,
                pTsdbReadHandle->window.skey, pTsdbReadHandle->window.ekey, pTsdbReadHandle->idStr);
2571
      pTsdbReadHandle->pFileGroup = NULL;
H
Haojun Liao 已提交
2572 2573 2574
      break;
    }

H
Haojun Liao 已提交
2575
    pTableBlockInfo->numOfFiles += 1;
2576 2577
    if (tsdbSetAndOpenReadFSet(&pTsdbReadHandle->rhelper, pTsdbReadHandle->pFileGroup) < 0) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2578 2579 2580 2581
      code = terrno;
      break;
    }

2582
    tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2583

2584
    if (tsdbLoadBlockIdx(&pTsdbReadHandle->rhelper) < 0) {
H
Haojun Liao 已提交
2585 2586 2587 2588
      code = terrno;
      break;
    }

2589
    if ((code = getFileCompInfo(pTsdbReadHandle, &numOfBlocks)) != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2590 2591 2592
      break;
    }

H
Haojun Liao 已提交
2593 2594
    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 已提交
2595 2596 2597 2598 2599 2600

    if (numOfBlocks == 0) {
      continue;
    }

    for (int32_t i = 0; i < numOfTables; ++i) {
2601
      STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, i);
H
Haojun Liao 已提交
2602 2603 2604

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

H
Haojun Liao 已提交
2607
        int32_t numOfRows = pBlock[j].numOfRows;
2608
        pTableBlockInfo->totalRows += numOfRows;
H
Haojun Liao 已提交
2609 2610 2611 2612 2613 2614 2615 2616 2617 2618 2619
        if (numOfRows > pTableBlockInfo->maxRows) {
          pTableBlockInfo->maxRows = numOfRows;
        }

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

        if (numOfRows < defaultRows) {
          pTableBlockInfo->numOfSmallBlocks += 1;
        }
dengyihao's avatar
dengyihao 已提交
2620 2621 2622
        //        int32_t  stepIndex = (numOfRows-1)/TSDB_BLOCK_DIST_STEP_ROWS;
        //        SFileBlockInfo *blockInfo = (SFileBlockInfo*)taosArrayGet(pTableBlockInfo->dataBlockInfos, stepIndex);
        //        blockInfo->numBlocksOfStep++;
H
Haojun Liao 已提交
2623 2624 2625 2626 2627 2628 2629
      }
    }
  }

  return code;
}

2630 2631 2632
static int32_t getDataBlocksInFiles(STsdbReadHandle* pTsdbReadHandle, bool* exists) {
  STsdbFS*       pFileHandle = REPO_FS(pTsdbReadHandle->pTsdb);
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
2633 2634

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

H
Hongze Cheng 已提交
2640
    tsdbRLockFS(pFileHandle);
2641 2642
    tsdbFSIterInit(&pTsdbReadHandle->fileIter, pFileHandle, pTsdbReadHandle->order);
    tsdbFSIterSeek(&pTsdbReadHandle->fileIter, fid);
H
Hongze Cheng 已提交
2643
    tsdbUnLockFS(pFileHandle);
2644

2645
    return getFirstFileDataBlock(pTsdbReadHandle, exists);
2646
  } else {
2647
    // check if current file block is all consumed
2648
    STableBlockInfo* pBlockInfo = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
2649
    STableCheckInfo* pCheckInfo = pBlockInfo->pTableCheckInfo;
H
Haojun Liao 已提交
2650

2651
    // current block is done, try next
H
Haojun Liao 已提交
2652
    if ((!cur->mixBlock) || cur->blockCompleted) {
H
Haojun Liao 已提交
2653
      // all data blocks in current file has been checked already, try next file if exists
2654
    } else {
H
Haojun Liao 已提交
2655 2656
      tsdbDebug("%p continue in current data block, index:%d, pos:%d, %s", pTsdbReadHandle, cur->slot, cur->pos,
                pTsdbReadHandle->idStr);
2657 2658
      int32_t code = handleDataMergeIfNeeded(pTsdbReadHandle, pBlockInfo->compBlock, pCheckInfo);
      *exists = (pTsdbReadHandle->realNumOfRows > 0);
H
Haojun Liao 已提交
2659

H
Haojun Liao 已提交
2660 2661 2662 2663 2664 2665 2666
      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
2667 2668
    if (isEndFileDataBlock(cur, pTsdbReadHandle->numOfBlocks, ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
      return getFirstFileDataBlock(pTsdbReadHandle, exists);
H
Haojun Liao 已提交
2669
    } else {
2670 2671 2672
      moveToNextDataBlockInCurrentFile(pTsdbReadHandle);
      STableBlockInfo* pNext = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
      return getDataBlockRv(pTsdbReadHandle, pNext, exists);
2673 2674
    }
  }
2675 2676
}

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

2680 2681
  while (pTsdbReadHandle->activeIndex < numOfTables) {
    if (hasMoreDataInCache(pTsdbReadHandle)) {
2682 2683
      return true;
    }
H
Haojun Liao 已提交
2684

2685
    pTsdbReadHandle->activeIndex += 1;
2686
  }
H
Haojun Liao 已提交
2687

2688 2689 2690
  return false;
}

dengyihao's avatar
dengyihao 已提交
2691
// todo not unref yet, since it is not support multi-group interpolation query
H
Haojun Liao 已提交
2692
static UNUSED_FUNC void changeQueryHandleForInterpQuery(tsdbReaderT pHandle) {
H
Haojun Liao 已提交
2693
  // filter the queried time stamp in the first place
dengyihao's avatar
dengyihao 已提交
2694
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle;
H
Haojun Liao 已提交
2695 2696

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

  int32_t i = 0;
dengyihao's avatar
dengyihao 已提交
2700
  while (i < numOfTables) {
2701
    STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, i);
H
Haojun Liao 已提交
2702 2703

    // the first qualified table for interpolation query
dengyihao's avatar
dengyihao 已提交
2704 2705 2706 2707
    //    if ((pTsdbReadHandle->window.skey <= pCheckInfo->pTableObj->lastKey) &&
    //        (pCheckInfo->pTableObj->lastKey != TSKEY_INITIAL_VAL)) {
    //      break;
    //    }
H
Haojun Liao 已提交
2708 2709 2710 2711 2712 2713 2714 2715 2716

    i++;
  }

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

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

2720 2721
  info.lastKey = pTsdbReadHandle->window.skey;
  taosArrayPush(pTsdbReadHandle->pTableCheckInfo, &info);
H
Haojun Liao 已提交
2722 2723 2724
}

static int tsdbReadRowsFromCache(STableCheckInfo* pCheckInfo, TSKEY maxKey, int maxRowsToRead, STimeWindow* win,
2725
                                 STsdbReadHandle* pTsdbReadHandle) {
dengyihao's avatar
dengyihao 已提交
2726
  int       numOfRows = 0;
C
Cary Xu 已提交
2727
  int       curRows = 0;
dengyihao's avatar
dengyihao 已提交
2728
  int32_t   numOfCols = (int32_t)taosArrayGetSize(pTsdbReadHandle->pColumns);
H
Hongze Cheng 已提交
2729
  STsdbCfg* pCfg = REPO_CFG(pTsdbReadHandle->pTsdb);
H
Haojun Liao 已提交
2730 2731
  win->skey = TSKEY_INITIAL_VAL;

dengyihao's avatar
dengyihao 已提交
2732 2733
  int64_t   st = taosGetTimestampUs();
  int16_t   rv = -1;
D
fix bug  
dapan1121 已提交
2734
  STSchema* pSchema = NULL;
C
Cary Xu 已提交
2735 2736
  TSKEY     lastRowKey = TSKEY_INITIAL_VAL;

H
Haojun Liao 已提交
2737
  do {
C
Cary Xu 已提交
2738
    STSRow* row = getSRowInTableMem(pCheckInfo, pTsdbReadHandle->order, pCfg->update, NULL, TD_VER_MAX);
H
Haojun Liao 已提交
2739 2740 2741 2742
    if (row == NULL) {
      break;
    }

H
Haojun Liao 已提交
2743
    TSKEY key = TD_ROW_KEY(row);
dengyihao's avatar
dengyihao 已提交
2744 2745 2746 2747
    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 已提交
2748 2749 2750 2751 2752 2753 2754 2755 2756

      break;
    }

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

    win->ekey = key;
H
Haojun Liao 已提交
2757
    if (rv != TD_ROW_SVER(row)) {
H
Hongze Cheng 已提交
2758
      pSchema = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), pCheckInfo->tableId, 0);
H
Haojun Liao 已提交
2759
      rv = TD_ROW_SVER(row);
D
fix bug  
dapan1121 已提交
2760
    }
C
Cary Xu 已提交
2761 2762
    numOfRows += mergeTwoRowFromMem(pTsdbReadHandle, maxRowsToRead, &curRows, row, NULL, numOfCols, pCheckInfo->tableId,
                                    pSchema, NULL, pCfg->update, &lastRowKey);
H
Haojun Liao 已提交
2763

C
Cary Xu 已提交
2764
    if (numOfRows >= maxRowsToRead) {
H
Haojun Liao 已提交
2765 2766 2767 2768
      moveToNextRowInMem(pCheckInfo);
      break;
    }

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

C
Cary Xu 已提交
2771
  taosMemoryFreeClear(pSchema);  // free the STSChema
C
Cary Xu 已提交
2772

H
Haojun Liao 已提交
2773 2774 2775
  assert(numOfRows <= maxRowsToRead);

  // if the buffer is not full in case of descending order query, move the data in the front of the buffer
2776
  if (!ASCENDING_TRAVERSE(pTsdbReadHandle->order) && numOfRows < maxRowsToRead) {
H
Haojun Liao 已提交
2777 2778
    int32_t emptySize = maxRowsToRead - numOfRows;

dengyihao's avatar
dengyihao 已提交
2779
    for (int32_t i = 0; i < numOfCols; ++i) {
2780
      SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
dengyihao's avatar
dengyihao 已提交
2781 2782
      memmove((char*)pColInfo->pData, (char*)pColInfo->pData + emptySize * pColInfo->info.bytes,
              numOfRows * pColInfo->info.bytes);
H
Haojun Liao 已提交
2783 2784 2785 2786
    }
  }

  int64_t elapsedTime = taosGetTimestampUs() - st;
dengyihao's avatar
dengyihao 已提交
2787 2788
  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 已提交
2789 2790 2791 2792

  return numOfRows;
}

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

2796 2797 2798 2799 2800
  while (1) {
    tb_uid_t id = metaCtbCursorNext(pCur);
    if (id == 0) {
      break;
    }
H
Haojun Liao 已提交
2801

2802
    STableKeyInfo info = {.lastKey = TSKEY_INITIAL_VAL, uid = id};
H
Haojun Liao 已提交
2803 2804 2805
    taosArrayPush(list, &info);
  }

2806
  metaCloseCtbCurosr(pCur);
H
Haojun Liao 已提交
2807 2808 2809 2810 2811 2812 2813 2814
  return TSDB_CODE_SUCCESS;
}

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

dengyihao's avatar
dengyihao 已提交
2815 2816 2817 2818 2819 2820
  //  tQueryInfo* pInfo = (tQueryInfo*)param;
  //  if (pInfo->optr != TSDB_RELATION_IN) {
  //    taosMemoryFreeClear(pInfo->q);
  //  } else {
  //    taosHashCleanup((SHashObj *)(pInfo->q));
  //  }
H
Haojun Liao 已提交
2821

wafwerar's avatar
wafwerar 已提交
2822
  taosMemoryFree(param);
H
Haojun Liao 已提交
2823 2824
}

dengyihao's avatar
dengyihao 已提交
2825 2826
#define TSDB_PREV_ROW 0x1
#define TSDB_NEXT_ROW 0x2
2827

dengyihao's avatar
dengyihao 已提交
2828
static bool loadBlockOfActiveTable(STsdbReadHandle* pTsdbReadHandle) {
2829
  if (pTsdbReadHandle->checkFiles) {
H
Haojun Liao 已提交
2830 2831
    // check if the query range overlaps with the file data block
    bool exists = true;
H
Haojun Liao 已提交
2832

2833
    int32_t code = getDataBlocksInFiles(pTsdbReadHandle, &exists);
H
Haojun Liao 已提交
2834
    if (code != TSDB_CODE_SUCCESS) {
2835
      pTsdbReadHandle->checkFiles = false;
H
Haojun Liao 已提交
2836 2837
      return false;
    }
H
Haojun Liao 已提交
2838

H
Haojun Liao 已提交
2839
    if (exists) {
dengyihao's avatar
dengyihao 已提交
2840
      tsdbRetrieveDataBlock((tsdbReaderT*)pTsdbReadHandle, NULL);
2841 2842 2843
      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 已提交
2844 2845
      }

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

2850
    pTsdbReadHandle->checkFiles = false;
H
Haojun Liao 已提交
2851
  }
H
Haojun Liao 已提交
2852

2853 2854
  if (hasMoreDataInCache(pTsdbReadHandle)) {
    pTsdbReadHandle->currentLoadExternalRows = false;
H
Haojun Liao 已提交
2855 2856
    return true;
  }
H
Haojun Liao 已提交
2857

H
Haojun Liao 已提交
2858
  // current result is empty
dengyihao's avatar
dengyihao 已提交
2859 2860 2861
  if (pTsdbReadHandle->currentLoadExternalRows && pTsdbReadHandle->window.skey == pTsdbReadHandle->window.ekey &&
      pTsdbReadHandle->cur.rows == 0) {
    //    STsdbMemTable* pMemRef = pTsdbReadHandle->pMemTable;
H
Haojun Liao 已提交
2862

dengyihao's avatar
dengyihao 已提交
2863 2864
    //    doGetExternalRow(pTsdbReadHandle, TSDB_PREV_ROW, pMemRef);
    //    doGetExternalRow(pTsdbReadHandle, TSDB_NEXT_ROW, pMemRef);
H
Haojun Liao 已提交
2865

2866
    bool result = tsdbGetExternalRow(pTsdbReadHandle);
H
Haojun Liao 已提交
2867

dengyihao's avatar
dengyihao 已提交
2868 2869
    //    pTsdbReadHandle->prev = doFreeColumnInfoData(pTsdbReadHandle->prev);
    //    pTsdbReadHandle->next = doFreeColumnInfoData(pTsdbReadHandle->next);
2870
    pTsdbReadHandle->currentLoadExternalRows = false;
H
Haojun Liao 已提交
2871 2872

    return result;
2873
  }
H
Haojun Liao 已提交
2874

H
Haojun Liao 已提交
2875 2876
  return false;
}
2877

2878
static bool loadCachedLastRow(STsdbReadHandle* pTsdbReadHandle) {
H
Haojun Liao 已提交
2879
  // the last row is cached in buffer, return it directly.
2880
  // here note that the pTsdbReadHandle->window must be the TS_INITIALIZER
dengyihao's avatar
dengyihao 已提交
2881
  int32_t numOfCols = (int32_t)(QH_GET_NUM_OF_COLS(pTsdbReadHandle));
2882
  size_t  numOfTables = taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
H
Haojun Liao 已提交
2883 2884
  assert(numOfTables > 0 && numOfCols > 0);

2885
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
2886

dengyihao's avatar
dengyihao 已提交
2887 2888 2889
  STSRow* pRow = NULL;
  TSKEY   key = TSKEY_INITIAL_VAL;
  int32_t step = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 1 : -1;
C
Cary Xu 已提交
2890 2891
  TSKEY   lastRowKey = TSKEY_INITIAL_VAL;
  int32_t curRow = 0;
2892 2893 2894

  if (++pTsdbReadHandle->activeIndex < numOfTables) {
    STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, pTsdbReadHandle->activeIndex);
dengyihao's avatar
dengyihao 已提交
2895 2896 2897 2898
    //    int32_t ret = tsdbGetCachedLastRow(pCheckInfo->pTableObj, &pRow, &key);
    //    if (ret != TSDB_CODE_SUCCESS) {
    //      return false;
    //    }
C
Cary Xu 已提交
2899 2900
    mergeTwoRowFromMem(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, &curRow, pRow, NULL, numOfCols,
                       pCheckInfo->tableId, NULL, NULL, true, &lastRowKey);
wafwerar's avatar
wafwerar 已提交
2901
    taosMemoryFreeClear(pRow);
H
Haojun Liao 已提交
2902

H
Haojun Liao 已提交
2903 2904 2905
    // update the last key value
    pCheckInfo->lastKey = key + step;

dengyihao's avatar
dengyihao 已提交
2906 2907
    cur->rows = 1;  // only one row
    cur->lastKey = key + step;
H
Haojun Liao 已提交
2908 2909 2910 2911 2912
    cur->mixBlock = true;
    cur->win.skey = key;
    cur->win.ekey = key;

    return true;
2913
  }
H
Haojun Liao 已提交
2914

H
Haojun Liao 已提交
2915 2916 2917
  return false;
}

dengyihao's avatar
dengyihao 已提交
2918
// static bool loadCachedLast(STsdbReadHandle* pTsdbReadHandle) {
2919 2920 2921 2922 2923 2924 2925 2926 2927 2928 2929 2930 2931 2932 2933 2934 2935 2936 2937
//  // 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 已提交
2938 2939
//      tsdbWarn("no last cached for table %s, uid:%" PRIu64 ",tid:%d", pTable->name->data, pTable->uid,
//      pTable->tableId); continue;
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 3027 3028 3029 3030 3031 3032 3033 3034 3035 3036 3037 3038 3039 3040 3041 3042 3043 3044 3045 3046 3047 3048 3049 3050 3051 3052 3053 3054 3055 3056 3057 3058 3059 3060 3061 3062 3063 3064 3065 3066 3067 3068 3069
//    }
//
//    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 已提交
3070 3071 3072
  assert(numOfTables > 0);

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

dengyihao's avatar
dengyihao 已提交
3074
  while (pTsdbReadHandle->activeIndex < numOfTables) {
3075
    if (loadBlockOfActiveTable(pTsdbReadHandle)) {
H
Haojun Liao 已提交
3076 3077 3078
      return true;
    }

3079
    STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, pTsdbReadHandle->activeIndex);
H
Haojun Liao 已提交
3080 3081
    pCheckInfo->numOfBlocks = 0;

3082 3083
    pTsdbReadHandle->activeIndex += 1;
    pTsdbReadHandle->locateStart = false;
dengyihao's avatar
dengyihao 已提交
3084 3085
    pTsdbReadHandle->checkFiles = true;
    pTsdbReadHandle->cur.rows = 0;
3086
    pTsdbReadHandle->currentLoadExternalRows = pTsdbReadHandle->loadExternalRow;
H
Haojun Liao 已提交
3087 3088 3089 3090

    terrno = TSDB_CODE_SUCCESS;

    int64_t elapsedTime = taosGetTimestampUs() - stime;
3091
    pTsdbReadHandle->cost.checkForNextTime += elapsedTime;
H
Haojun Liao 已提交
3092 3093 3094
  }

  return false;
3095 3096
}

H
Haojun Liao 已提交
3097
// handle data in cache situation
H
Haojun Liao 已提交
3098
bool tsdbNextDataBlock(tsdbReaderT pHandle) {
dengyihao's avatar
dengyihao 已提交
3099
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle;
Y
yihaoDeng 已提交
3100

dengyihao's avatar
dengyihao 已提交
3101
  for (int32_t i = 0; i < taosArrayGetSize(pTsdbReadHandle->pColumns); ++i) {
3102 3103 3104 3105
    SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
    colInfoDataCleanup(pColInfo, pTsdbReadHandle->outputCapacity);
  }

3106
  if (emptyQueryTimewindow(pTsdbReadHandle)) {
dengyihao's avatar
dengyihao 已提交
3107 3108
    tsdbDebug("%p query window not overlaps with the data set, no result returned, %s", pTsdbReadHandle,
              pTsdbReadHandle->idStr);
3109 3110 3111
    return false;
  }

Y
yihaoDeng 已提交
3112 3113 3114
  int64_t stime = taosGetTimestampUs();
  int64_t elapsedTime = stime;

3115
  // TODO refactor: remove "type"
3116 3117
  if (pTsdbReadHandle->type == TSDB_QUERY_TYPE_LAST) {
    if (pTsdbReadHandle->cachelastrow == TSDB_CACHED_TYPE_LASTROW) {
dengyihao's avatar
dengyihao 已提交
3118
      //      return loadCachedLastRow(pTsdbReadHandle);
3119
    } else if (pTsdbReadHandle->cachelastrow == TSDB_CACHED_TYPE_LAST) {
dengyihao's avatar
dengyihao 已提交
3120
      //      return loadCachedLast(pTsdbReadHandle);
D
init  
dapan1121 已提交
3121
    }
H
Haojun Liao 已提交
3122
  }
Y
yihaoDeng 已提交
3123

3124 3125
  if (pTsdbReadHandle->loadType == BLOCK_LOAD_TABLE_SEQ_ORDER) {
    return loadDataBlockFromTableSeq(pTsdbReadHandle);
dengyihao's avatar
dengyihao 已提交
3126
  } else {  // loadType == RR and Offset Order
3127
    if (pTsdbReadHandle->checkFiles) {
H
Haojun Liao 已提交
3128 3129 3130
      // check if the query range overlaps with the file data block
      bool exists = true;

3131
      int32_t code = getDataBlocksInFiles(pTsdbReadHandle, &exists);
H
Haojun Liao 已提交
3132
      if (code != TSDB_CODE_SUCCESS) {
3133 3134
        pTsdbReadHandle->activeIndex = 0;
        pTsdbReadHandle->checkFiles = false;
H
Haojun Liao 已提交
3135 3136 3137 3138 3139

        return false;
      }

      if (exists) {
3140
        pTsdbReadHandle->cost.checkForNextTime += (taosGetTimestampUs() - stime);
H
Haojun Liao 已提交
3141 3142
        return exists;
      }
Y
yihaoDeng 已提交
3143

3144 3145
      pTsdbReadHandle->activeIndex = 0;
      pTsdbReadHandle->checkFiles = false;
Y
yihaoDeng 已提交
3146 3147
    }

H
Haojun Liao 已提交
3148
    // TODO: opt by consider the scan order
3149
    bool ret = doHasDataInBuffer(pTsdbReadHandle);
H
Haojun Liao 已提交
3150
    terrno = TSDB_CODE_SUCCESS;
Y
yihaoDeng 已提交
3151

H
Haojun Liao 已提交
3152
    elapsedTime = taosGetTimestampUs() - stime;
3153
    pTsdbReadHandle->cost.checkForNextTime += elapsedTime;
H
Haojun Liao 已提交
3154
    return ret;
Y
yihaoDeng 已提交
3155 3156
  }
}
3157

dengyihao's avatar
dengyihao 已提交
3158
// static int32_t doGetExternalRow(STsdbReadHandle* pTsdbReadHandle, int16_t type, STsdbMemTable* pMemRef) {
3159 3160 3161 3162 3163 3164 3165 3166 3167 3168 3169 3170 3171 3172 3173 3174 3175 3176 3177 3178 3179 3180 3181 3182 3183 3184 3185 3186 3187 3188 3189 3190 3191 3192
//  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 已提交
3193
//    colInfo.pData = taosMemoryCalloc(1, pCol->info.bytes);
3194 3195 3196 3197 3198 3199 3200 3201 3202
//    if (colInfo.pData == NULL) {
//      terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
//      goto out_of_memory;
//    }
//
//    taosArrayPush(row, &colInfo);
//  }
//
//  // load the previous row
3203
//  SQueryTableDataCond cond = {.numOfCols = numOfCols, .loadExternalRows = false, .type = BLOCK_LOAD_OFFSET_SEQ_ORDER};
3204 3205 3206 3207 3208 3209 3210 3211
//  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 已提交
3212
//  cond.colList = taosMemoryCalloc(cond.numOfCols, sizeof(SColumnInfo));
3213 3214 3215 3216 3217 3218 3219 3220 3221 3222
//  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 已提交
3223
//  pSecQueryHandle = tsdbQueryTablesImpl(pTsdbReadHandle->pTsdb, &cond, pTsdbReadHandle->idStr, pMemRef);
wafwerar's avatar
wafwerar 已提交
3224
//  taosMemoryFreeClear(cond.colList);
3225 3226 3227 3228 3229 3230 3231 3232 3233 3234 3235 3236 3237 3238 3239 3240 3241 3242 3243 3244 3245 3246 3247 3248 3249 3250 3251 3252 3253 3254 3255 3256 3257 3258 3259 3260 3261 3262
//
//  // 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 已提交
3263
// out_of_memory:
3264
//  tsdbCleanupReadHandle(pSecQueryHandle);
3265 3266 3267
//  return terrno;
//}

H
Haojun Liao 已提交
3268
bool tsdbGetExternalRow(tsdbReaderT pHandle) {
dengyihao's avatar
dengyihao 已提交
3269 3270
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle;
  SQueryFilePos*   cur = &pTsdbReadHandle->cur;
H
Haojun Liao 已提交
3271

H
Haojun Liao 已提交
3272 3273
  cur->fid = INT32_MIN;
  cur->mixBlock = true;
3274
  if (pTsdbReadHandle->prev == NULL || pTsdbReadHandle->next == NULL) {
H
Haojun Liao 已提交
3275 3276
    cur->rows = 0;
    return false;
H
Haojun Liao 已提交
3277 3278
  }

dengyihao's avatar
dengyihao 已提交
3279
  int32_t numOfCols = (int32_t)QH_GET_NUM_OF_COLS(pTsdbReadHandle);
H
Haojun Liao 已提交
3280
  for (int32_t i = 0; i < numOfCols; ++i) {
3281 3282
    SColumnInfoData* pColInfoData = taosArrayGet(pTsdbReadHandle->pColumns, i);
    SColumnInfoData* first = taosArrayGet(pTsdbReadHandle->prev, i);
H
Haojun Liao 已提交
3283 3284 3285

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

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

    if (i == 0 && pColInfoData->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
H
Haojun Liao 已提交
3290
      cur->win.skey = *(TSKEY*)pColInfoData->pData;
sangshuduo's avatar
sangshuduo 已提交
3291
      cur->win.ekey = *(TSKEY*)(((char*)pColInfoData->pData) + TSDB_KEYSIZE);
H
Haojun Liao 已提交
3292 3293 3294
    }
  }

H
Haojun Liao 已提交
3295 3296
  cur->rows = 2;
  return true;
3297 3298
}

3299
/*
3300
 * if lastRow == NULL, return TSDB_CODE_TDB_NO_CACHE_LAST_ROW
3301
 * else set pRes and return TSDB_CODE_SUCCESS and save lastKey
3302
 */
H
Haojun Liao 已提交
3303
// int32_t tsdbGetCachedLastRow(STable* pTable, STSRow** pRes, TSKEY* lastKey) {
3304 3305 3306 3307 3308 3309 3310 3311 3312 3313 3314 3315 3316 3317 3318 3319
//  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 已提交
3320
// out:
3321 3322 3323 3324
//  TSDB_RUNLOCK_TABLE(pTable);
//  return code;
//}

3325
bool isTsdbCacheLastRow(tsdbReaderT* pReader) {
dengyihao's avatar
dengyihao 已提交
3326
  return ((STsdbReadHandle*)pReader)->cachelastrow > TSDB_CACHED_TYPE_NONE;
D
fix bug  
dapan1121 已提交
3327 3328
}

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

dengyihao's avatar
dengyihao 已提交
3332 3333 3334 3335 3336 3337 3338 3339 3340 3341 3342 3343 3344 3345 3346 3347 3348 3349 3350 3351 3352 3353 3354 3355
  //  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 已提交
3356

3357
  return TSDB_CODE_SUCCESS;
3358 3359
}

3360 3361
int32_t checkForCachedLast(STsdbReadHandle* pTsdbReadHandle) {
  assert(pTsdbReadHandle != NULL);
D
update  
dapan1121 已提交
3362 3363

  int32_t code = 0;
dengyihao's avatar
dengyihao 已提交
3364 3365 3366
  //  if (pTsdbReadHandle->pTsdb && atomic_load_8(&pTsdbReadHandle->pTsdb->hasCachedLastColumn)){
  //    pTsdbReadHandle->cachelastrow = TSDB_CACHED_TYPE_LAST;
  //  }
D
update  
dapan1121 已提交
3367 3368

  // update the tsdb query time range
3369
  if (pTsdbReadHandle->cachelastrow) {
dengyihao's avatar
dengyihao 已提交
3370
    pTsdbReadHandle->checkFiles = false;
3371
    pTsdbReadHandle->activeIndex = -1;  // start from -1
D
update  
dapan1121 已提交
3372 3373 3374 3375 3376
  }

  return code;
}

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

H
Haojun Liao 已提交
3380
  int32_t totalNumOfTable = 0;
3381
  SArray* emptyGroup = taosArrayInit(16, sizeof(int32_t));
H
Haojun Liao 已提交
3382

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

H
Haojun Liao 已提交
3389
    STableKeyInfo keyInfo = {0};
H
Haojun Liao 已提交
3390

H
Haojun Liao 已提交
3391
    size_t numOfTables = taosArrayGetSize(pGroup);
dengyihao's avatar
dengyihao 已提交
3392 3393
    for (int32_t i = 0; i < numOfTables; ++i) {
      STableKeyInfo* pInfo = (STableKeyInfo*)taosArrayGet(pGroup, i);
H
Haojun Liao 已提交
3394

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

dengyihao's avatar
dengyihao 已提交
3400
        //        keyInfo.pTable  = pInfo->pTable;
H
Haojun Liao 已提交
3401
        keyInfo.lastKey = key;
dengyihao's avatar
dengyihao 已提交
3402
        pInfo->lastKey = key;
H
Haojun Liao 已提交
3403

H
Haojun Liao 已提交
3404 3405 3406
        if (key < window.skey) {
          window.skey = key;
        }
3407

H
Haojun Liao 已提交
3408 3409 3410 3411
        if (key > window.ekey) {
          window.ekey = key;
        }
      }
3412
    }
H
Haojun Liao 已提交
3413

H
Haojun Liao 已提交
3414
    // more than one table in each group, only one table left for each group
dengyihao's avatar
dengyihao 已提交
3415 3416 3417 3418 3419 3420 3421 3422 3423 3424 3425 3426
    //    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);
    //    }
3427
  }
H
Haojun Liao 已提交
3428

H
Haojun Liao 已提交
3429 3430 3431
  // window does not being updated, so set the original
  if (window.skey == INT64_MAX && window.ekey == INT64_MIN) {
    window = TSWINDOW_INITIALIZER;
H
Haojun Liao 已提交
3432
    assert(totalNumOfTable == 0 && taosArrayGetSize(groupList->pGroupList) == numOfGroups);
H
Haojun Liao 已提交
3433 3434
  }

dengyihao's avatar
dengyihao 已提交
3435
  taosArrayRemoveBatch(groupList->pGroupList, TARRAY_GET_START(emptyGroup), (int32_t)taosArrayGetSize(emptyGroup));
3436 3437
  taosArrayDestroy(emptyGroup);

H
Haojun Liao 已提交
3438
  groupList->numOfTables = totalNumOfTable;
H
Haojun Liao 已提交
3439
  return window;
H
hjxilinx 已提交
3440 3441
}

H
Haojun Liao 已提交
3442
void tsdbRetrieveDataBlockInfo(tsdbReaderT* pTsdbReadHandle, SDataBlockInfo* pDataBlockInfo) {
3443
  STsdbReadHandle* pHandle = (STsdbReadHandle*)pTsdbReadHandle;
dengyihao's avatar
dengyihao 已提交
3444
  SQueryFilePos*   cur = &pHandle->cur;
3445 3446

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

3448
  // there are data in file
D
dapan1121 已提交
3449
  if (pHandle->cur.fid != INT32_MIN) {
3450
    STableBlockInfo* pBlockInfo = &pHandle->pDataBlockInfo[cur->slot];
3451
    uid = pBlockInfo->pTableCheckInfo->tableId;
H
[td-32]  
hjxilinx 已提交
3452
  } else {
3453
    STableCheckInfo* pCheckInfo = taosArrayGet(pHandle->pTableCheckInfo, pHandle->activeIndex);
3454
    uid = pCheckInfo->tableId;
3455
  }
3456

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

3460
  pDataBlockInfo->uid = uid;
3461 3462 3463 3464 3465 3466

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

dengyihao's avatar
dengyihao 已提交
3467
  pDataBlockInfo->rows = cur->rows;
H
Haojun Liao 已提交
3468
  pDataBlockInfo->window = cur->win;
C
Cary Xu 已提交
3469
  //  ASSERT(pDataBlockInfo->numOfCols >= (int32_t)(QH_GET_NUM_OF_COLS(pHandle));
3470
}
H
hjxilinx 已提交
3471

H
Haojun Liao 已提交
3472 3473 3474
/*
 * return null for mixed data block, if not a complete file data block, the statistics value will always return NULL
 */
3475
int32_t tsdbRetrieveDataBlockStatisInfo(tsdbReaderT* pTsdbReadHandle, SColumnDataAgg*** pBlockStatis, bool* allHave) {
dengyihao's avatar
dengyihao 已提交
3476
  STsdbReadHandle* pHandle = (STsdbReadHandle*)pTsdbReadHandle;
3477
  *allHave = false;
H
Haojun Liao 已提交
3478

H
Haojun Liao 已提交
3479 3480
  SQueryFilePos* c = &pHandle->cur;
  if (c->mixBlock) {
H
Haojun Liao 已提交
3481 3482 3483
    *pBlockStatis = NULL;
    return TSDB_CODE_SUCCESS;
  }
H
Haojun Liao 已提交
3484

H
Haojun Liao 已提交
3485 3486 3487 3488
  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 已提交
3489 3490 3491 3492
  if (pBlockInfo->compBlock->numOfSubBlocks > 1) {
    *pBlockStatis = NULL;
    return TSDB_CODE_SUCCESS;
  }
H
Haojun Liao 已提交
3493 3494

  int64_t stime = taosGetTimestampUs();
3495 3496
  int     statisStatus = tsdbLoadBlockStatis(&pHandle->rhelper, pBlockInfo->compBlock);
  if (statisStatus < TSDB_STATIS_OK) {
H
Hongze Cheng 已提交
3497
    return terrno;
3498 3499 3500
  } else if (statisStatus > TSDB_STATIS_OK) {
    *pBlockStatis = NULL;
    return TSDB_CODE_SUCCESS;
H
Hongze Cheng 已提交
3501
  }
H
Haojun Liao 已提交
3502

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

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

H
Haojun Liao 已提交
3508
  size_t numOfCols = QH_GET_NUM_OF_COLS(pHandle);
3509 3510 3511
  memset(pHandle->suppInfo.plist, 0, numOfCols * POINTER_BYTES);
  memset(pHandle->suppInfo.pstatis, 0, numOfCols * sizeof(SColumnDataAgg));

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

3516 3517
  *allHave = true;
  tsdbGetBlockStatis(&pHandle->rhelper, pHandle->suppInfo.pstatis, (int)numOfCols, pBlockInfo->compBlock);
H
Haojun Liao 已提交
3518 3519

  // always load the first primary timestamp column data
3520
  SColumnDataAgg* pPrimaryColStatis = &pHandle->suppInfo.pstatis[0];
3521
  assert(pPrimaryColStatis->colId == PRIMARYKEY_TIMESTAMP_COL_ID);
H
Haojun Liao 已提交
3522 3523 3524 3525

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

dengyihao's avatar
dengyihao 已提交
3528
  // update the number of NULL data rows
3529
  int32_t* slotIds = pHandle->suppInfo.slotIds;
dengyihao's avatar
dengyihao 已提交
3530
  for (int32_t i = 1; i < numOfCols; ++i) {
3531
    ASSERT(colIds[i] == pHandle->pSchema->columns[slotIds[i]].colId);
C
Cary Xu 已提交
3532
    if (IS_BSMA_ON(&(pHandle->pSchema->columns[slotIds[i]]))) {
3533 3534 3535 3536 3537 3538 3539
      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 已提交
3540 3541
    }
  }
H
Haojun Liao 已提交
3542 3543 3544 3545

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

3546
  *pBlockStatis = pHandle->suppInfo.plist;
3547
  return TSDB_CODE_SUCCESS;
H
hjxilinx 已提交
3548 3549
}

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

3562
    if (pHandle->cur.mixBlock) {
H
[td-32]  
hjxilinx 已提交
3563 3564
      return pHandle->pColumns;
    } else {
H
Haojun Liao 已提交
3565
      SDataBlockInfo binfo = GET_FILE_DATA_BLOCK_INFO(pCheckInfo, pBlockInfo->compBlock);
3566
      assert(pHandle->realNumOfRows <= binfo.rows);
H
Haojun Liao 已提交
3567

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

H
Hongze Cheng 已提交
3571
      if (pBlockLoadInfo->slot == pHandle->cur.slot && pBlockLoadInfo->fileGroup->fid == pHandle->cur.fid &&
H
Haojun Liao 已提交
3572
          pBlockLoadInfo->uid == pCheckInfo->tableId) {
H
hjxilinx 已提交
3573
        return pHandle->pColumns;
H
Haojun Liao 已提交
3574
      } else {  // only load the file block
H
refact  
Hongze Cheng 已提交
3575
        SBlock* pBlock = pBlockInfo->compBlock;
H
Haojun Liao 已提交
3576
        if (doLoadFileDataBlock(pHandle, pBlock, pCheckInfo, pHandle->cur.slot) != TSDB_CODE_SUCCESS) {
3577 3578
          return NULL;
        }
H
Haojun Liao 已提交
3579

H
Haojun Liao 已提交
3580
        int32_t numOfRows = doCopyRowsFromFileBlock(pHandle, pHandle->outputCapacity, 0, 0, pBlock->numOfRows - 1);
H
hjxilinx 已提交
3581 3582
        return pHandle->pColumns;
      }
H
[td-32]  
hjxilinx 已提交
3583 3584
    }
  }
H
hjxilinx 已提交
3585
}
3586
#if 0
3587
void filterPrepare(void* expr, void* param) {
3588
  tExprNode* pExpr = (tExprNode*)expr;
H
[td-32]  
hjxilinx 已提交
3589
  if (pExpr->_node.info != NULL) {
3590 3591
    return;
  }
3592

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

3595
  STSchema*   pTSSchema = (STSchema*) param;
H
hjxilinx 已提交
3596 3597 3598
  tQueryInfo* pInfo = pExpr->_node.info;
  tVariant*   pCond = pExpr->_node.pRight->pVal;
  SSchema*    pSchema = pExpr->_node.pLeft->pSchema;
3599

3600 3601
  pInfo->sch      = *pSchema;
  pInfo->optr     = pExpr->_node.optr;
Y
yihaoDeng 已提交
3602
  pInfo->compare  = getComparFunc(pInfo->sch.type, pInfo->optr);
H
Haojun Liao 已提交
3603
  pInfo->indexed  = pTSSchema->columns->colId == pInfo->sch.colId;
H
Haojun Liao 已提交
3604

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

3631
#endif
3632

dengyihao's avatar
dengyihao 已提交
3633
static int32_t tableGroupComparFn(const void* p1, const void* p2, const void* param) {
3634
#if 0
3635
  STableGroupSupporter* pTableGroupSupp = (STableGroupSupporter*) param;
3636 3637
  STable* pTable1 = ((STableKeyInfo*) p1)->uid;
  STable* pTable2 = ((STableKeyInfo*) p2)->uid;
H
Haojun Liao 已提交
3638

3639 3640 3641
  for (int32_t i = 0; i < pTableGroupSupp->numOfCols; ++i) {
    SColIndex* pColIndex = &pTableGroupSupp->pCols[i];
    int32_t colIndex = pColIndex->colIndex;
H
Haojun Liao 已提交
3642

H
Haojun Liao 已提交
3643
    assert(colIndex >= TSDB_TBNAME_COLUMN_INDEX);
H
Haojun Liao 已提交
3644

3645 3646 3647 3648
    char *  f1 = NULL;
    char *  f2 = NULL;
    int32_t type = 0;
    int32_t bytes = 0;
H
Haojun Liao 已提交
3649

H
Haojun Liao 已提交
3650 3651 3652
    if (colIndex == TSDB_TBNAME_COLUMN_INDEX) {
      f1 = (char*) TABLE_NAME(pTable1);
      f2 = (char*) TABLE_NAME(pTable2);
3653
      type = TSDB_DATA_TYPE_BINARY;
3654
      bytes = tGetTbnameColumnSchema()->bytes;
3655
    } else {
Y
yihaoDeng 已提交
3656 3657 3658 3659 3660 3661 3662
      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);
      } 
3663
    }
H
Haojun Liao 已提交
3664 3665 3666 3667 3668 3669 3670 3671 3672 3673 3674 3675 3676 3677

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

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

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

3678 3679 3680 3681 3682 3683 3684
    int32_t ret = doCompare(f1, f2, type, bytes);
    if (ret == 0) {
      continue;
    } else {
      return ret;
    }
  }
3685
#endif
3686 3687 3688
  return 0;
}

H
Haojun Liao 已提交
3689
static int tsdbCheckInfoCompar(const void* key1, const void* key2) {
3690
  if (((STableCheckInfo*)key1)->tableId < ((STableCheckInfo*)key2)->tableId) {
H
Haojun Liao 已提交
3691
    return -1;
3692
  } else if (((STableCheckInfo*)key1)->tableId > ((STableCheckInfo*)key2)->tableId) {
H
Haojun Liao 已提交
3693 3694 3695 3696 3697 3698 3699 3700 3701
    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) {
3702
  STable* pTable = taosArrayGetP(pTableList, 0);
H
Haojun Liao 已提交
3703 3704
  SArray* g = taosArrayInit(16, sizeof(STableKeyInfo));

3705
  STableKeyInfo info = {.lastKey = skey};
H
Haojun Liao 已提交
3706
  taosArrayPush(g, &info);
3707

3708
  for (int32_t i = 1; i < numOfTables; ++i) {
3709 3710
    STable** prev = taosArrayGet(pTableList, i - 1);
    STable** p = taosArrayGet(pTableList, i);
H
Haojun Liao 已提交
3711

H
hjxilinx 已提交
3712
    int32_t ret = compareFn(prev, p, pSupp);
3713
    assert(ret == 0 || ret == -1);
H
Haojun Liao 已提交
3714

3715
    if (ret == 0) {
3716
      STableKeyInfo info1 = {.lastKey = skey};
H
Haojun Liao 已提交
3717
      taosArrayPush(g, &info1);
3718 3719
    } else {
      taosArrayPush(pGroups, &g);  // current group is ended, start a new group
H
Haojun Liao 已提交
3720 3721
      g = taosArrayInit(16, sizeof(STableKeyInfo));

3722
      STableKeyInfo info1 = {.lastKey = skey};
H
Haojun Liao 已提交
3723
      taosArrayPush(g, &info1);
3724 3725
    }
  }
H
Haojun Liao 已提交
3726

3727
  taosArrayPush(pGroups, &g);
3728 3729
}

dengyihao's avatar
dengyihao 已提交
3730 3731
SArray* createTableGroup(SArray* pTableList, SSchemaWrapper* pTagSchema, SColIndex* pCols, int32_t numOfOrderCols,
                         TSKEY skey) {
3732
  assert(pTableList != NULL);
3733
  SArray* pTableGroup = taosArrayInit(1, POINTER_BYTES);
H
Haojun Liao 已提交
3734

3735 3736
  size_t size = taosArrayGetSize(pTableList);
  if (size == 0) {
S
Shengliang Guan 已提交
3737
    tsdbDebug("no qualified tables");
3738 3739
    return pTableGroup;
  }
H
Haojun Liao 已提交
3740

dengyihao's avatar
dengyihao 已提交
3741
  if (numOfOrderCols == 0 || size == 1) {  // no group by tags clause or only one table
3742
    SArray* sa = taosArrayDup(pTableList);
H
Haojun Liao 已提交
3743 3744 3745 3746
    if (sa == NULL) {
      taosArrayDestroy(pTableGroup);
      return NULL;
    }
H
Haojun Liao 已提交
3747

3748
    taosArrayPush(pTableGroup, &sa);
S
TD-1057  
Shengliang Guan 已提交
3749
    tsdbDebug("all %" PRIzu " tables belong to one group", size);
3750
  } else {
H
Haojun Liao 已提交
3751 3752
    STableGroupSupporter sup = {0};
    sup.numOfCols = numOfOrderCols;
3753
    sup.pTagSchema = pTagSchema->pSchema;
H
Haojun Liao 已提交
3754 3755
    sup.pCols = pCols;

3756 3757
    taosqsort(pTableList->pData, size, sizeof(STableKeyInfo), &sup, tableGroupComparFn);
    createTableGroupImpl(pTableGroup, pTableList, size, skey, &sup, tableGroupComparFn);
3758
  }
H
Haojun Liao 已提交
3759

3760 3761 3762
  return pTableGroup;
}

dengyihao's avatar
dengyihao 已提交
3763
// static bool tableFilterFp(const void* pNode, void* param) {
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 3805 3806 3807 3808 3809 3810 3811 3812 3813 3814 3815 3816 3817 3818 3819 3820 3821 3822 3823 3824 3825 3826 3827 3828 3829 3830 3831 3832 3833 3834 3835 3836 3837 3838 3839 3840 3841 3842 3843 3844 3845 3846
//  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 已提交
3847

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

dengyihao's avatar
dengyihao 已提交
3851 3852
// static int32_t doQueryTableList(STable* pSTable, SArray* pRes, tExprNode* pExpr) {
//  //  // query according to the expression tree
3853
//  SExprTraverseSupp supp = {
dengyihao's avatar
dengyihao 已提交
3854
//      .nodeFilterFn = (__result_filter_fn_t)tableFilterFp,
3855 3856
//      .setupInfoFn = filterPrepare,
//      .pExtInfo = pSTable->tagSchema,
dengyihao's avatar
dengyihao 已提交
3857
//  };
3858 3859 3860 3861 3862
//
//  getTableListfromSkipList(pExpr, pSTable->pIndex, pRes, &supp);
//  tExprTreeDestroy(pExpr, destroyHelper);
//  return TSDB_CODE_SUCCESS;
//}
3863

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

H
Hongze Cheng 已提交
3869
  metaReaderInit(&mr, (SMeta*)pMeta, 0);
H
Hongze Cheng 已提交
3870 3871

  if (metaGetTableEntryByUid(&mr, uid) < 0) {
dengyihao's avatar
dengyihao 已提交
3872
    tsdbError("%p failed to get stable, uid:%" PRIu64 ", TID:0x%" PRIx64 " QID:0x%" PRIx64, pMeta, uid, taskId, reqId);
3873 3874
    terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
    goto _error;
C
Cary Xu 已提交
3875 3876
  } else {
    tsdbDebug("%p succeed to get stable, uid:%" PRIu64 ", TID:0x%" PRIx64 " QID:0x%" PRIx64, pMeta, uid, taskId, reqId);
3877
  }
H
Haojun Liao 已提交
3878

H
Hongze Cheng 已提交
3879
  if (mr.me.type != TSDB_SUPER_TABLE) {
dengyihao's avatar
dengyihao 已提交
3880 3881 3882
    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
3883
    goto _error;
H
hjxilinx 已提交
3884
  }
3885

H
Hongze Cheng 已提交
3886 3887
  metaReaderClear(&mr);

dengyihao's avatar
dengyihao 已提交
3888 3889
  // NOTE: not add ref count for super table
  SArray*         res = taosArrayInit(8, sizeof(STableKeyInfo));
H
Haojun Liao 已提交
3890
  SSchemaWrapper* pTagSchema = metaGetTableSchema(pMeta, uid, 0, true);
H
Haojun Liao 已提交
3891

weixin_48148422's avatar
weixin_48148422 已提交
3892 3893
  // no tags and tbname condition, all child tables of this stable are involved
  if (tbnameCond == NULL && (pTagCond == NULL || len == 0)) {
H
Haojun Liao 已提交
3894
    int32_t ret = getAllTableList(pMeta, uid, res);
3895 3896
    if (ret != TSDB_CODE_SUCCESS) {
      goto _error;
3897
    }
3898

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

dengyihao's avatar
dengyihao 已提交
3902 3903 3904
    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);
3905

3906
    taosArrayDestroy(res);
3907 3908
    return ret;
  }
3909

H
hjxilinx 已提交
3910
  int32_t ret = TSDB_CODE_SUCCESS;
3911

dengyihao's avatar
dengyihao 已提交
3912 3913
  SFilterInfo* filterInfo = NULL;
  ret = filterInitFromNode((SNode*)pTagCond, &filterInfo, 0);
dengyihao's avatar
dengyihao 已提交
3914 3915 3916 3917 3918 3919 3920 3921
  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 已提交
3922 3923
  // 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 已提交
3924 3925 3926 3927 3928

  taosArrayDestroy(res);
  return ret;

_error:
3929
  return terrno;
3930
}
3931

dengyihao's avatar
dengyihao 已提交
3932 3933 3934 3935 3936
int32_t tsdbQueryTableList(void* pMeta, SArray* pRes, void* filterInfo) {
  // impl later

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

H
Hongze Cheng 已提交
3940
  metaReaderInit(&mr, (SMeta*)pMeta, 0);
H
Hongze Cheng 已提交
3941 3942

  if (metaGetTableEntryByUid(&mr, uid) < 0) {
3943 3944
    terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
    goto _error;
3945
  }
3946

H
Hongze Cheng 已提交
3947 3948
  metaReaderClear(&mr);

3949 3950
  pGroupInfo->numOfTables = 1;
  pGroupInfo->pGroupList = taosArrayInit(1, POINTER_BYTES);
H
Haojun Liao 已提交
3951

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

3954
  STableKeyInfo info = {.lastKey = startKey, .uid = uid};
H
Haojun Liao 已提交
3955
  taosArrayPush(group, &info);
H
Haojun Liao 已提交
3956

3957
  taosArrayPush(pGroupInfo->pGroupList, &group);
3958
  return TSDB_CODE_SUCCESS;
3959

dengyihao's avatar
dengyihao 已提交
3960
_error:
H
Hongze Cheng 已提交
3961
  metaReaderClear(&mr);
3962
  return terrno;
3963
}
3964

3965
#if 0
3966
int32_t tsdbGetTableGroupFromIdListT(STsdb* tsdb, SArray* pTableIdList, STableGroupInfo* pGroupInfo) {
B
Bomin Zhang 已提交
3967 3968 3969
  if (tsdbRLockRepoMeta(tsdb) < 0) {
    return terrno;
  }
3970 3971 3972 3973

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

B
Bomin Zhang 已提交
3976
  for(int32_t i = 0; i < size; ++i) {
3977 3978 3979 3980 3981 3982 3983 3984 3985 3986 3987
    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 已提交
3988 3989 3990
      tsdbUnlockRepoMeta(tsdb);
      taosArrayDestroy(group);
      return terrno;
3991 3992
    }

H
Haojun Liao 已提交
3993 3994
    STableKeyInfo info = {.pTable = pTable, .lastKey = id->key};
    taosArrayPush(group, &info);
3995 3996
  }

B
Bomin Zhang 已提交
3997 3998 3999 4000
  if (tsdbUnlockRepoMeta(tsdb) < 0) {
    taosArrayDestroy(group);
    return terrno;
  }
4001

sangshuduo's avatar
sangshuduo 已提交
4002
  pGroupInfo->numOfTables = (uint32_t) taosArrayGetSize(group);
B
Bomin Zhang 已提交
4003 4004 4005 4006 4007
  if (pGroupInfo->numOfTables > 0) {
    taosArrayPush(pGroupInfo->pGroupList, &group);
  } else {
    taosArrayDestroy(group);
  }
4008 4009 4010

  return TSDB_CODE_SUCCESS;
}
4011
#endif
4012 4013 4014 4015 4016 4017 4018 4019
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 已提交
4020
    taosMemoryFreeClear(pColInfo->pData);
4021 4022 4023 4024 4025 4026
  }

  taosArrayDestroy(pColumnInfoData);
  return NULL;
}

H
Haojun Liao 已提交
4027 4028 4029 4030 4031 4032
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 已提交
4033
    taosMemoryFreeClear(p->pCompInfo);
H
Haojun Liao 已提交
4034 4035 4036 4037 4038 4039
  }

  taosArrayDestroy(pTableCheckInfo);
  return NULL;
}

H
Haojun Liao 已提交
4040
void tsdbCleanupReadHandle(tsdbReaderT queryHandle) {
4041 4042
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)queryHandle;
  if (pTsdbReadHandle == NULL) {
4043 4044
    return;
  }
4045

4046
  pTsdbReadHandle->pColumns = doFreeColumnInfoData(pTsdbReadHandle->pColumns);
4047

4048
  taosArrayDestroy(pTsdbReadHandle->suppInfo.defaultLoadColumn);
wafwerar's avatar
wafwerar 已提交
4049
  taosMemoryFreeClear(pTsdbReadHandle->pDataBlockInfo);
4050 4051
  taosMemoryFreeClear(pTsdbReadHandle->suppInfo.pstatis);
  taosMemoryFreeClear(pTsdbReadHandle->suppInfo.plist);
4052

4053
  if (!emptyQueryTimewindow(pTsdbReadHandle)) {
dengyihao's avatar
dengyihao 已提交
4054
    //    tsdbMayUnTakeMemSnapshot(pTsdbReadHandle);
4055
  } else {
4056
    assert(pTsdbReadHandle->pTableCheckInfo == NULL);
4057 4058
  }

4059 4060
  if (pTsdbReadHandle->pTableCheckInfo != NULL) {
    pTsdbReadHandle->pTableCheckInfo = destroyTableCheckInfo(pTsdbReadHandle->pTableCheckInfo);
4061
  }
4062

4063
  tsdbDestroyReadH(&pTsdbReadHandle->rhelper);
H
Haojun Liao 已提交
4064

4065 4066
  tdFreeDataCols(pTsdbReadHandle->pDataCols);
  pTsdbReadHandle->pDataCols = NULL;
H
Haojun Liao 已提交
4067

4068 4069
  pTsdbReadHandle->prev = doFreeColumnInfoData(pTsdbReadHandle->prev);
  pTsdbReadHandle->next = doFreeColumnInfoData(pTsdbReadHandle->next);
4070

4071
  SIOCostSummary* pCost = &pTsdbReadHandle->cost;
4072

dengyihao's avatar
dengyihao 已提交
4073 4074 4075 4076
  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 已提交
4077

wafwerar's avatar
wafwerar 已提交
4078
  taosMemoryFreeClear(pTsdbReadHandle);
4079
}
4080

4081
#if 0
H
Haojun Liao 已提交
4082
void tsdbDestroyTableGroup(STableGroupInfo *pGroupList) {
4083 4084 4085 4086 4087 4088 4089 4090 4091 4092
  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);
4093 4094 4095 4096
      if (pTable != NULL) { // in case of handling retrieve data from tsdb
        tsdbUnRefTable(pTable);
      }
      //assert(pTable != NULL);
4097 4098 4099 4100 4101
    }

    taosArrayDestroy(p);
  }

4102
  taosHashCleanup(pGroupList->map);
4103
  taosArrayDestroy(pGroupList->pGroupList);
H
Haojun Liao 已提交
4104
  pGroupList->numOfTables = 0;
4105
}
H
Haojun Liao 已提交
4106 4107 4108 4109 4110 4111 4112

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 已提交
4113
    if (exprTreeApplyFilter(pExpr, pNode, param)) {
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
      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 已提交
4137
    pCond->start       = taosMemoryCalloc(1, sizeof(SEndPoint));
H
Haojun Liao 已提交
4138
    pCond->start->optr = queryColInfo->optr;
Y
yihaoDeng 已提交
4139
    pCond->start->v    = queryColInfo->q;
H
Haojun Liao 已提交
4140
  } else if (optr == TSDB_RELATION_LESS || optr == TSDB_RELATION_LESS_EQUAL) {
wafwerar's avatar
wafwerar 已提交
4141
    pCond->end       = taosMemoryCalloc(1, sizeof(SEndPoint));
H
Haojun Liao 已提交
4142
    pCond->end->optr = queryColInfo->optr;
Y
yihaoDeng 已提交
4143 4144
    pCond->end->v    = queryColInfo->q;
  } else if (optr == TSDB_RELATION_IN) {
wafwerar's avatar
wafwerar 已提交
4145
    pCond->start       = taosMemoryCalloc(1, sizeof(SEndPoint));
Y
yihaoDeng 已提交
4146 4147 4148
    pCond->start->optr = queryColInfo->optr;
    pCond->start->v    = queryColInfo->q; 
  } else if (optr == TSDB_RELATION_LIKE) {
H
Haojun Liao 已提交
4149
    assert(0);
4150 4151
  } else if (optr == TSDB_RELATION_MATCH) {
    assert(0);
4152 4153
  } else if (optr == TSDB_RELATION_NMATCH) {
    assert(0);
H
Haojun Liao 已提交
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 4197 4198 4199 4200 4201 4202 4203 4204 4205 4206 4207 4208 4209 4210 4211 4212 4213 4214 4215 4216 4217 4218 4219 4220 4221 4222 4223 4224 4225 4226 4227 4228 4229 4230 4231 4232 4233 4234 4235 4236
  }

  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 已提交
4237 4238 4239 4240 4241 4242 4243 4244 4245 4246 4247 4248 4249
    } 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 已提交
4250 4251 4252 4253 4254 4255 4256 4257 4258 4259 4260 4261 4262 4263 4264 4265 4266 4267 4268 4269 4270 4271 4272 4273 4274 4275 4276 4277 4278 4279 4280 4281 4282 4283 4284 4285 4286 4287 4288 4289 4290
    } 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 已提交
4291 4292
  taosMemoryFree(cond.start);
  taosMemoryFree(cond.end);
H
Haojun Liao 已提交
4293 4294 4295 4296 4297 4298 4299 4300 4301 4302 4303 4304 4305 4306 4307 4308 4309 4310
  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);
4311 4312 4313
      } else if (pQueryInfo->optr == TSDB_RELATION_LIKE ||
                 pQueryInfo->optr == TSDB_RELATION_MATCH ||
                 pQueryInfo->optr == TSDB_RELATION_NMATCH) {
H
Haojun Liao 已提交
4314 4315 4316 4317 4318 4319 4320 4321 4322 4323 4324 4325 4326 4327 4328 4329
        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 已提交
4330 4331 4332 4333 4334 4335 4336 4337 4338 4339 4340 4341 4342 4343 4344 4345 4346 4347 4348 4349 4350 4351 4352 4353 4354 4355 4356 4357 4358 4359 4360 4361 4362
//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 已提交
4363
#endif