tsdbRead.c 147.8 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 355
#if 1
int nQUERY = 0;
C
Cary Xu 已提交
356
#endif
C
Cary Xu 已提交
357
static STsdb* getTsdbByRetentions(SVnode* pVnode, STsdbReadHandle* pReadHandle, TSKEY winSKey, SRetention* retentions) {
C
Cary Xu 已提交
358
  if (vnodeIsRollup(pVnode)) {
C
Cary Xu 已提交
359
    int level = 0;
C
Cary Xu 已提交
360
#if 0
C
Cary Xu 已提交
361 362 363 364 365 366
    int64_t now = taosGetTimestamp(pVnode->config.tsdbCfg.precision);
    for (int i = 0; i < TSDB_RETENTION_MAX; ++i) {
      SRetention* pRetention = retentions + i;
      if (pRetention->keep <= 0 || (now - pRetention->keep) >= winSKey) {
        break;
      }
C
Cary Xu 已提交
367
      ++level;
C
Cary Xu 已提交
368 369
    }
#endif
C
Cary Xu 已提交
370 371 372 373 374 375 376 377 378 379 380
#if 1
    switch ((nQUERY++) % 3) {
      case 0:
        level = 0;
        break;
      case 1:
        level = 1;
        break;
      default:
        level = 2;
        break;
C
Cary Xu 已提交
381 382 383
    }
#endif
    if (level == TSDB_RETENTION_L0) {
C
Cary Xu 已提交
384
      tsdbDebug("%p rsma level %d is selected to query\n", pReadHandle, level);
C
Cary Xu 已提交
385 386
      return VND_RSMA0(pVnode);
    } else if (level == TSDB_RETENTION_L1) {
C
Cary Xu 已提交
387
      tsdbDebug("%p rsma level %d is selected to query\n", pReadHandle, level);
C
Cary Xu 已提交
388 389
      return VND_RSMA1(pVnode);
    } else {
C
Cary Xu 已提交
390
      tsdbDebug("%p rsma level %d is selected to query\n", pReadHandle, level);
C
Cary Xu 已提交
391 392
      return VND_RSMA2(pVnode);
    }
C
Cary Xu 已提交
393 394 395 396
  }
  return pVnode->pTsdb;
}

397
static STsdbReadHandle* tsdbQueryTablesImpl(SVnode* pVnode, SQueryTableDataCond* pCond, uint64_t qId, uint64_t taskId) {
wafwerar's avatar
wafwerar 已提交
398
  STsdbReadHandle* pReadHandle = taosMemoryCalloc(1, sizeof(STsdbReadHandle));
399
  if (pReadHandle == NULL) {
400
    goto _end;
401
  }
H
Haojun Liao 已提交
402

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

dengyihao's avatar
dengyihao 已提交
405
  pReadHandle->order = pCond->order;
C
Cary Xu 已提交
406
  pReadHandle->pTsdb = pTsdb;
dengyihao's avatar
dengyihao 已提交
407 408 409 410 411 412
  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;
413
  pReadHandle->locateStart = false;
dengyihao's avatar
dengyihao 已提交
414
  pReadHandle->loadType = pCond->type;
415

dengyihao's avatar
dengyihao 已提交
416
  pReadHandle->outputCapacity = 4096;  //((STsdb*)tsdb)->config.maxRowsPerFileBlock;
417 418 419
  pReadHandle->loadExternalRow = pCond->loadExternalRows;
  pReadHandle->currentLoadExternalRows = pCond->loadExternalRows;

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

C
Cary Xu 已提交
424
  if (tsdbInitReadH(&pReadHandle->rhelper, pReadHandle->pTsdb) != 0) {
425
    goto _end;
B
Bomin Zhang 已提交
426
  }
H
Haojun Liao 已提交
427

428 429
  assert(pCond != NULL);
  setQueryTimewindow(pReadHandle, pCond);
430

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

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

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

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

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

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

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

470 471
  tsdbInitDataBlockLoadInfo(&pReadHandle->dataBlockLoadInfo);
  tsdbInitCompBlockLoadInfo(&pReadHandle->compBlockLoadInfo);
472

H
Haojun Liao 已提交
473
  return (tsdbReaderT)pReadHandle;
474

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

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

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

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

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

502
  pTsdbReadHandle->pSchema = metaGetTbTSchema(pVnode->pMeta, pCheckInfo->tableId, 0);
C
Cary Xu 已提交
503
  int32_t  numOfCols = taosArrayGetSize(pTsdbReadHandle->suppInfo.defaultLoadColumn);
504 505 506 507 508
  int16_t* ids = pTsdbReadHandle->suppInfo.defaultLoadColumn->pData;

  STSchema* pSchema = pTsdbReadHandle->pSchema;

  int32_t i = 0, j = 0;
C
Cary Xu 已提交
509
  while (i < numOfCols && j < pSchema->numOfCols) {
510 511 512 513 514 515 516 517 518 519 520 521
    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;
    }
  }
522

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

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

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

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

    return;
  }

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

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

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

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

565
  resetCheckInfo(pTsdbReadHandle);
H
Haojun Liao 已提交
566 567
}

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

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

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

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

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

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

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

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

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

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

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

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

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

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

635
  return pTsdbReadHandle;
D
init  
dapan1121 已提交
636 637
}

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

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

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

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

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

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

H
Haojun Liao 已提交
669 670 671 672 673
// 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 已提交
674
  STableGroupInfo* pNew = taosMemoryCalloc(1, sizeof(STableGroupInfo));
Y
yihaoDeng 已提交
675
  pNew->pGroupList = taosArrayInit(numOfGroup, POINTER_BYTES);
H
Haojun Liao 已提交
676

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

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

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

  return pNew;
}

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

706 707 708 709 710 711 712 713 714 715 716 717
  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;
    }
  }

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

722
  return pTsdbReadHandle;
723 724
}

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

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

733 734 735
  STbData** pMem = NULL;
  STbData** pIMem = NULL;

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

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

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

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

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

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

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

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

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

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

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

805 806 807
  return true;
}

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

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

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

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

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

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

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

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

C
Cary Xu 已提交
882 883

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

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

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

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

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

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

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

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

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

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

991 992 993
    if (hasNext) {
      return hasNext;
    }
994

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

H
Haojun Liao 已提交
1007 1008 1009
  return hasNext;
}

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

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

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

H
Haojun Liao 已提交
1026
  pCheckInfo->lastKey = TD_ROW_KEY(row);  // first timestamp in buffer
dengyihao's avatar
dengyihao 已提交
1027 1028
  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 已提交
1029

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

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

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

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

1049
  return true;
1050
}
H
hjxilinx 已提交
1051

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

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

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

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

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

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

1078
  int32_t midSlot = firstSlot;
H
Haojun Liao 已提交
1079

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

1084
    if (numOfBlocks == 1) break;
H
Haojun Liao 已提交
1085

1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096
    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 已提交
1097

1098 1099
  return midSlot;
}
1100

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

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

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

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

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

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

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

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

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

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

H
Haojun Liao 已提交
1141
  TSKEY s = TSKEY_INITIAL_VAL, e = TSKEY_INITIAL_VAL;
1142

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

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

H
Haojun Liao 已提交
1154 1155 1156
  // 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 已提交
1157

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

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

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

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

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

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

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

H
Haojun Liao 已提交
1185
  size_t numOfTables = 0;
1186 1187 1188 1189
  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);
1190

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

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

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

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

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

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

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

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

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

1244
  SDataBlockLoadInfo* pBlockLoadInfo = &pTsdbReadHandle->dataBlockLoadInfo;
1245

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

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

1253
  pBlock->numOfRows = pCols->numOfRows;
H
Haojun Liao 已提交
1254

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

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

dengyihao's avatar
dengyihao 已提交
1266 1267 1268 1269
  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 已提交
1270
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1271 1272 1273 1274

_error:
  pBlock->numOfRows = 0;

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

1280
static int32_t getEndPosInDataBlock(STsdbReadHandle* pTsdbReadHandle, SDataBlockInfo* pBlockInfo);
dengyihao's avatar
dengyihao 已提交
1281 1282 1283 1284 1285 1286 1287 1288
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) {
1289
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
H
Hongze Cheng 已提交
1290
  STsdbCfg*      pCfg = REPO_CFG(pTsdbReadHandle->pTsdb);
H
Haojun Liao 已提交
1291
  SDataBlockInfo binfo = GET_FILE_DATA_BLOCK_INFO(pCheckInfo, pBlock);
1292
  TSKEY          key;
dengyihao's avatar
dengyihao 已提交
1293
  int32_t        code = TSDB_CODE_SUCCESS;
1294

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

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

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

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

1308 1309 1310 1311
  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 已提交
1312
      // do not load file block into buffer
1313
      int32_t step = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 1 : -1;
H
Haojun Liao 已提交
1314

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

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

H
Haojun Liao 已提交
1327 1328
      cur->mixBlock = true;
      cur->blockCompleted = false;
H
Haojun Liao 已提交
1329
      return code;
H
Haojun Liao 已提交
1330
    }
H
Haojun Liao 已提交
1331

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

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

1348 1349
    if ((cur->pos == 0 && endPos == binfo.rows - 1 && ascScan) ||
        (cur->pos == (binfo.rows - 1) && endPos == 0 && (!ascScan))) {
1350
      pTsdbReadHandle->realNumOfRows = binfo.rows;
1351 1352

      cur->rows = binfo.rows;
dengyihao's avatar
dengyihao 已提交
1353
      cur->win = binfo.window;
1354
      cur->mixBlock = false;
H
Haojun Liao 已提交
1355 1356
      cur->blockCompleted = true;

1357
      if (ascScan) {
H
Haojun Liao 已提交
1358 1359 1360 1361 1362 1363
        cur->lastKey = binfo.window.ekey + 1;
        cur->pos = binfo.rows;
      } else {
        cur->lastKey = binfo.window.skey - 1;
        cur->pos = -1;
      }
dengyihao's avatar
dengyihao 已提交
1364
    } else {  // partially copy to dest buffer
1365
      copyAllRemainRowsFromFileBlock(pTsdbReadHandle, pCheckInfo, &binfo, endPos);
1366 1367
      cur->mixBlock = true;
    }
1368

H
Haojun Liao 已提交
1369
    assert(cur->blockCompleted);
H
Haojun Liao 已提交
1370
    if (cur->rows == binfo.rows) {
dengyihao's avatar
dengyihao 已提交
1371
      tsdbDebug("%p whole file block qualified, brange:%" PRId64 "-%" PRId64 ", rows:%d, lastKey:%" PRId64 ", %s",
H
Haojun Liao 已提交
1372
                pTsdbReadHandle, cur->win.skey, cur->win.ekey, cur->rows, cur->lastKey, pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
1373
    } else {
dengyihao's avatar
dengyihao 已提交
1374 1375 1376 1377
      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 已提交
1378
    }
1379
  }
H
Haojun Liao 已提交
1380 1381

  return code;
1382 1383
}

1384 1385
static int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order);

dengyihao's avatar
dengyihao 已提交
1386 1387
static int32_t loadFileDataBlock(STsdbReadHandle* pTsdbReadHandle, SBlock* pBlock, STableCheckInfo* pCheckInfo,
                                 bool* exists) {
1388
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
dengyihao's avatar
dengyihao 已提交
1389 1390
  int32_t        code = TSDB_CODE_SUCCESS;
  bool           asc = ASCENDING_TRAVERSE(pTsdbReadHandle->order);
1391

1392
  if (asc) {
H
Haojun Liao 已提交
1393
    // query ended in/started from current block
1394 1395
    if (pTsdbReadHandle->window.ekey < pBlock->keyLast || pCheckInfo->lastKey > pBlock->keyFirst) {
      if ((code = doLoadFileDataBlock(pTsdbReadHandle, pBlock, pCheckInfo, cur->slot)) != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
1396 1397
        *exists = false;
        return code;
1398
      }
1399

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

1403 1404
      if (pCheckInfo->lastKey > pBlock->keyFirst) {
        cur->pos =
1405
            binarySearchForKey(pTSCol->cols[0].pData, pBlock->numOfRows, pCheckInfo->lastKey, pTsdbReadHandle->order);
1406 1407 1408
      } else {
        cur->pos = 0;
      }
H
Haojun Liao 已提交
1409

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

1423
      SDataCols* pTsCol = pTsdbReadHandle->rhelper.pDCols[0];
1424
      if (pCheckInfo->lastKey < pBlock->keyLast) {
dengyihao's avatar
dengyihao 已提交
1425 1426
        cur->pos =
            binarySearchForKey(pTsCol->cols[0].pData, pBlock->numOfRows, pCheckInfo->lastKey, pTsdbReadHandle->order);
1427
      } else {
H
Haojun Liao 已提交
1428
        cur->pos = pBlock->numOfRows - 1;
1429
      }
H
Haojun Liao 已提交
1430

H
Haojun Liao 已提交
1431
      assert(pCheckInfo->lastKey >= pBlock->keyFirst);
1432
      doMergeTwoLevelData(pTsdbReadHandle, pCheckInfo, pBlock);
1433
    } else {
dengyihao's avatar
dengyihao 已提交
1434
      cur->pos = asc ? 0 : (pBlock->numOfRows - 1);
1435
      code = handleDataMergeIfNeeded(pTsdbReadHandle, pBlock, pCheckInfo);
H
[td-32]  
hjxilinx 已提交
1436
    }
1437
  }
1438

1439
  *exists = pTsdbReadHandle->realNumOfRows > 0;
H
Haojun Liao 已提交
1440
  return code;
H
[td-32]  
hjxilinx 已提交
1441 1442
}

1443
static int doBinarySearchKey(char* pValue, int num, TSKEY key, int order) {
1444
  int    firstPos, lastPos, midPos = -1;
H
Haojun Liao 已提交
1445
  int    numOfRows;
1446 1447
  TSKEY* keyList;

1448
  assert(order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC);
H
Haojun Liao 已提交
1449

1450
  if (num <= 0) return -1;
1451 1452

  keyList = (TSKEY*)pValue;
1453 1454
  firstPos = 0;
  lastPos = num - 1;
1455

1456
  if (order == TSDB_ORDER_DESC) {
1457 1458 1459 1460 1461
    // 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;
1462

H
Haojun Liao 已提交
1463 1464
      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1) + firstPos;
1465

1466 1467 1468 1469 1470 1471 1472 1473
      if (key < keyList[midPos]) {
        lastPos = midPos - 1;
      } else if (key > keyList[midPos]) {
        firstPos = midPos + 1;
      } else {
        break;
      }
    }
1474

1475 1476 1477 1478 1479
  } 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;
1480

1481 1482 1483 1484 1485 1486 1487
      if (key > keyList[lastPos]) {
        lastPos = lastPos + 1;
        if (lastPos >= num)
          return -1;
        else
          return lastPos;
      }
1488

H
Haojun Liao 已提交
1489 1490
      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1) + firstPos;
1491

1492 1493 1494 1495 1496 1497 1498 1499 1500
      if (key < keyList[midPos]) {
        lastPos = midPos - 1;
      } else if (key > keyList[midPos]) {
        firstPos = midPos + 1;
      } else {
        break;
      }
    }
  }
1501

1502 1503 1504
  return midPos;
}

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

1510
  int32_t num = end - start + 1;
H
Haojun Liao 已提交
1511 1512 1513 1514 1515 1516
  assert(num >= 0);

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

1517 1518
  bool    ascScan = ASCENDING_TRAVERSE(pTsdbReadHandle->order);
  int32_t trueStart = ascScan ? start : end;
1519
  int32_t trueEnd = ascScan ? end : start;
1520 1521
  int32_t step = ascScan ? 1 : -1;

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

dengyihao's avatar
dengyihao 已提交
1524
  // data in buffer has greater timestamp, copy data in file block
1525
  int32_t i = 0, j = 0;
dengyihao's avatar
dengyihao 已提交
1526
  while (i < requiredNumOfCols && j < pCols->numOfCols) {
1527
    SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
1528 1529 1530 1531 1532 1533 1534

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

L
Liu Jicong 已提交
1535
    if (!isAllRowsNull(src) && pColInfo->info.colId == src->colId) {
1536
      if (!IS_VAR_DATA_TYPE(pColInfo->info.type)) {  // todo opt performance
dengyihao's avatar
dengyihao 已提交
1537
        //        memmove(pData, (char*)src->pData + bytes * start, bytes * num);
1538
        int32_t rowIndex = numOfRows;
1539
        for (int32_t k = trueStart; ((ascScan && k <= trueEnd) || (!ascScan && k >= trueEnd)); k += step, ++rowIndex) {
1540
          SCellVal sVal = {0};
C
Cary Xu 已提交
1541
          if (tdGetColDataOfRow(&sVal, src, k, pCols->bitmapMode) < 0) {
1542 1543
            TASSERT(0);
          }
1544

C
Cary Xu 已提交
1545
          if (sVal.valType == TD_VTYPE_NORM) {
1546
            colDataAppend(pColInfo, rowIndex, sVal.val, false);
C
Cary Xu 已提交
1547 1548
          } else {
            colDataAppendNULL(pColInfo, rowIndex);
1549 1550 1551
          }
        }
      } else {  // handle the var-string
1552 1553
        int32_t rowIndex = numOfRows;

1554
        // todo refactor, only copy one-by-one
1555
        for (int32_t k = trueStart; ((ascScan && k <= trueEnd) || (!ascScan && k >= trueEnd)); k += step, ++rowIndex) {
1556
          SCellVal sVal = {0};
dengyihao's avatar
dengyihao 已提交
1557
          if (tdGetColDataOfRow(&sVal, src, k, pCols->bitmapMode) < 0) {
H
Haojun Liao 已提交
1558 1559
            TASSERT(0);
          }
1560

C
Cary Xu 已提交
1561
          if (sVal.valType == TD_VTYPE_NORM) {
1562
            colDataAppend(pColInfo, rowIndex, sVal.val, false);
C
Cary Xu 已提交
1563 1564
          } else {
            colDataAppendNULL(pColInfo, rowIndex);
1565
          }
1566 1567
        }
      }
1568 1569 1570

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

dengyihao's avatar
dengyihao 已提交
1577
  while (i < requiredNumOfCols) {  // the remain columns are all null data
1578
    SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
1579
    colDataAppendNNULL(pColInfo, numOfRows, num);
1580
    i++;
1581
  }
H
Haojun Liao 已提交
1582

1583 1584
  pTsdbReadHandle->cur.win.ekey = tsArray[trueEnd];
  pTsdbReadHandle->cur.lastKey = tsArray[trueEnd] + step;
1585

1586
  return numOfRows + num;
1587 1588
}

C
Cary Xu 已提交
1589 1590 1591 1592 1593 1594 1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608
/**
 * @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 已提交
1609
#if 1
dengyihao's avatar
dengyihao 已提交
1610 1611 1612 1613 1614 1615 1616 1617 1618
  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 已提交
1619
  SCellVal sVal = {0};
C
Cary Xu 已提交
1620 1621
  TSKEY    rowKey = TSKEY_INITIAL_VAL;
  int32_t  nResult = 0;
C
Cary Xu 已提交
1622
  int32_t  mergeOption = 0;  // 0 discard 1 overwrite 2 merge
1623

H
Haojun Liao 已提交
1624
  // the schema version info is embeded in STSRow
1625 1626 1627
  int32_t numOfColsOfRow1 = 0;

  if (pSchema1 == NULL) {
C
Cary Xu 已提交
1628 1629 1630
    // pSchema1 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, TD_ROW_SVER(row1));
    // TODO: use the real schemaVersion
    pSchema1 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, 0);
1631
  }
1632

C
Cary Xu 已提交
1633 1634 1635 1636
#ifdef TD_DEBUG_PRINT_ROW
  tdSRowPrint(row1, pSchema1, __func__);
#endif

dengyihao's avatar
dengyihao 已提交
1637
  if (isRow1DataRow) {
1638
    numOfColsOfRow1 = schemaNCols(pSchema1);
H
Haojun Liao 已提交
1639
  } else {
H
Haojun Liao 已提交
1640
    numOfColsOfRow1 = tdRowGetNCols(row1);
D
fix bug  
dapan1121 已提交
1641
  }
1642

1643
  int32_t numOfColsOfRow2 = 0;
dengyihao's avatar
dengyihao 已提交
1644
  if (row2) {
H
Haojun Liao 已提交
1645
    isRow2DataRow = TD_IS_TP_ROW(row2);
1646
    if (pSchema2 == NULL) {
C
Cary Xu 已提交
1647 1648 1649
      // pSchema2 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, TD_ROW_SVER(row2));
      // TODO: use the real schemaVersion
      pSchema2 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, 0);
1650
    }
dengyihao's avatar
dengyihao 已提交
1651
    if (isRow2DataRow) {
1652 1653
      numOfColsOfRow2 = schemaNCols(pSchema2);
    } else {
H
Haojun Liao 已提交
1654
      numOfColsOfRow2 = tdRowGetNCols(row2);
1655 1656
    }
  }
C
Cary Xu 已提交
1657

1658
  int32_t i = 0, j = 0, k = 0;
dengyihao's avatar
dengyihao 已提交
1659
  while (i < numOfCols && (j < numOfColsOfRow1 || k < numOfColsOfRow2)) {
1660
    SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
1661 1662

    int32_t colIdOfRow1;
dengyihao's avatar
dengyihao 已提交
1663
    if (j >= numOfColsOfRow1) {
1664
      colIdOfRow1 = INT32_MAX;
dengyihao's avatar
dengyihao 已提交
1665
    } else if (isRow1DataRow) {
1666 1667
      colIdOfRow1 = pSchema1->columns[j].colId;
    } else {
C
Cary Xu 已提交
1668
      colIdOfRow1 = tdKvRowColIdAt(row1, j);
1669 1670 1671
    }

    int32_t colIdOfRow2;
dengyihao's avatar
dengyihao 已提交
1672
    if (k >= numOfColsOfRow2) {
1673
      colIdOfRow2 = INT32_MAX;
dengyihao's avatar
dengyihao 已提交
1674
    } else if (isRow2DataRow) {
1675 1676
      colIdOfRow2 = pSchema2->columns[k].colId;
    } else {
C
Cary Xu 已提交
1677
      colIdOfRow2 = tdKvRowColIdAt(row2, k);
1678 1679
    }

C
Cary Xu 已提交
1680
    if (colIdOfRow1 < colIdOfRow2) { // the most probability
dengyihao's avatar
dengyihao 已提交
1681
      if (colIdOfRow1 < pColInfo->info.colId) {
C
Cary Xu 已提交
1682
        ++j;
C
Cary Xu 已提交
1683 1684
        continue;
      }
1685 1686 1687 1688
      row = row1;
      pSchema = pSchema1;
      isChosenRowDataRow = isRow1DataRow;
      chosen_itr = j;
C
Cary Xu 已提交
1689
    } else if (colIdOfRow1 == colIdOfRow2) {
dengyihao's avatar
dengyihao 已提交
1690
      if (colIdOfRow1 < pColInfo->info.colId) {
C
Cary Xu 已提交
1691 1692
        ++j;
        ++k;
1693
        continue;
C
Cary Xu 已提交
1694
      }
1695 1696 1697 1698 1699
      row = row1;
      pSchema = pSchema1;
      isChosenRowDataRow = isRow1DataRow;
      chosen_itr = j;
    } else {
dengyihao's avatar
dengyihao 已提交
1700
      if (colIdOfRow2 < pColInfo->info.colId) {
C
Cary Xu 已提交
1701
        ++k;
1702 1703 1704 1705 1706 1707 1708
        continue;
      }
      row = row2;
      pSchema = pSchema2;
      chosen_itr = k;
      isChosenRowDataRow = isRow2DataRow;
    }
C
Cary Xu 已提交
1709

dengyihao's avatar
dengyihao 已提交
1710
    if (isChosenRowDataRow) {
1711 1712
      colId = pSchema->columns[chosen_itr].colId;
      offset = pSchema->columns[chosen_itr].offset;
C
Cary Xu 已提交
1713 1714
      // TODO: use STSRowIter
      tdSTpRowGetVal(row, colId, pSchema->columns[chosen_itr].type, pSchema->flen, offset, chosen_itr - 1, &sVal);
C
Cary Xu 已提交
1715 1716 1717
      if (colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
        rowKey = *(TSKEY*)sVal.val;
        if (rowKey != *lastRowKey) {
C
Cary Xu 已提交
1718
          mergeOption = 1;
C
Cary Xu 已提交
1719 1720 1721 1722
          if (*lastRowKey != TSKEY_INITIAL_VAL) {
            ++(*curRow);
          }
          ++nResult;
C
Cary Xu 已提交
1723 1724 1725 1726 1727
        } else if (update){
          mergeOption = 2;
        } else {
          mergeOption = 0;
          break;
C
Cary Xu 已提交
1728
        }
C
Cary Xu 已提交
1729

C
Cary Xu 已提交
1730 1731
        *lastRowKey = rowKey;
      }
1732
    } else {
C
Cary Xu 已提交
1733 1734 1735 1736
      // TODO: use STSRowIter
      if (chosen_itr == 0) {
        colId = PRIMARYKEY_TIMESTAMP_COL_ID;
        tdSKvRowGetVal(row, PRIMARYKEY_TIMESTAMP_COL_ID, -1, -1, &sVal);
C
Cary Xu 已提交
1737 1738
        rowKey = *(TSKEY*)sVal.val;
        if (rowKey != *lastRowKey) {
C
Cary Xu 已提交
1739
          mergeOption = 1;
C
Cary Xu 已提交
1740 1741 1742 1743
          if (*lastRowKey != TSKEY_INITIAL_VAL) {
            ++(*curRow);
          }
          ++nResult;
C
Cary Xu 已提交
1744 1745 1746 1747 1748
        } else if(update) {
          mergeOption = 2;
        } else {
          mergeOption = 0;
          break;
C
Cary Xu 已提交
1749 1750
        }
        *lastRowKey = rowKey;
C
Cary Xu 已提交
1751 1752 1753 1754 1755 1756
      } else {
        SKvRowIdx* pColIdx = tdKvRowColIdxAt(row, chosen_itr - 1);
        colId = pColIdx->colId;
        offset = pColIdx->offset;
        tdSKvRowGetVal(row, colId, offset, chosen_itr - 1, &sVal);
      }
1757 1758
    }

C
Cary Xu 已提交
1759 1760
    ASSERT(rowKey != TSKEY_INITIAL_VAL);

1761
    if (colId == pColInfo->info.colId) {
H
Haojun Liao 已提交
1762
      if (tdValTypeIsNorm(sVal.valType)) {
C
Cary Xu 已提交
1763 1764 1765 1766 1767
        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 已提交
1768
        if (mergeOption == 1) {
C
Cary Xu 已提交
1769 1770 1771 1772
          colDataAppend(pColInfo, *curRow, NULL, true);
        }
      } else {
        ASSERT(0);
1773
      }
H
Haojun Liao 已提交
1774

C
Cary Xu 已提交
1775
      ++i;
C
Cary Xu 已提交
1776

dengyihao's avatar
dengyihao 已提交
1777
      if (row == row1) {
C
Cary Xu 已提交
1778
        ++j;
1779
      } else {
C
Cary Xu 已提交
1780
        ++k;
1781 1782
      }
    } else {
C
Cary Xu 已提交
1783
      if (mergeOption == 1) {
C
Cary Xu 已提交
1784
        colDataAppend(pColInfo, *curRow, NULL, true);
C
Cary Xu 已提交
1785
      }
C
Cary Xu 已提交
1786
      ++i;
1787
    }
1788
  }
1789

C
Cary Xu 已提交
1790
  if (mergeOption == 1) {
dengyihao's avatar
dengyihao 已提交
1791
    while (i < numOfCols) {  // the remain columns are all null data
1792
      SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
C
Cary Xu 已提交
1793 1794
      colDataAppend(pColInfo, *curRow, NULL, true);
      ++i;
1795 1796
    }
  }
C
Cary Xu 已提交
1797 1798

  return nResult;
H
Haojun Liao 已提交
1799
#endif
1800
}
1801

1802 1803
static void moveDataToFront(STsdbReadHandle* pTsdbReadHandle, int32_t numOfRows, int32_t numOfCols) {
  if (numOfRows == 0 || ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
1804 1805 1806 1807
    return;
  }

  // if the buffer is not full in case of descending order query, move the data in the front of the buffer
1808 1809
  if (numOfRows < pTsdbReadHandle->outputCapacity) {
    int32_t emptySize = pTsdbReadHandle->outputCapacity - numOfRows;
dengyihao's avatar
dengyihao 已提交
1810
    for (int32_t i = 0; i < numOfCols; ++i) {
1811
      SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
dengyihao's avatar
dengyihao 已提交
1812 1813
      memmove((char*)pColInfo->pData, (char*)pColInfo->pData + emptySize * pColInfo->info.bytes,
              numOfRows * pColInfo->info.bytes);
1814 1815 1816 1817
    }
  }
}

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

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

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

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

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

  pCheckInfo->lastKey = cur->lastKey;
1849
  pTsdbReadHandle->realNumOfRows = numOfRows;
1850 1851 1852 1853
  cur->rows = numOfRows;
  cur->pos = endPos;
}

1854 1855
static void doCheckGeneratedBlockRange(STsdbReadHandle* pTsdbReadHandle) {
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
H
Haojun Liao 已提交
1856 1857

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

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

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

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

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

dengyihao's avatar
dengyihao 已提交
1882
  int32_t step = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 1 : -1;
1883
  int32_t numOfCols = (int32_t)(QH_GET_NUM_OF_COLS(pTsdbReadHandle));
H
Haojun Liao 已提交
1884 1885 1886 1887 1888 1889

  int32_t pos = cur->pos;

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

1890
  if (!ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
wafwerar's avatar
wafwerar 已提交
1891
    TSWAP(start, end);
H
Haojun Liao 已提交
1892 1893
  }

1894 1895
  assert(pTsdbReadHandle->outputCapacity >= (end - start + 1));
  int32_t numOfRows = doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, 0, start, end);
H
Haojun Liao 已提交
1896 1897

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

  // if the buffer is not full in case of descending order query, move the data in the front of the buffer
1904
  moveDataToFront(pTsdbReadHandle, numOfRows, numOfCols);
H
Haojun Liao 已提交
1905 1906 1907

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

dengyihao's avatar
dengyihao 已提交
1911 1912 1913
  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 已提交
1914 1915
}

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

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

1924
  if (ASCENDING_TRAVERSE(pTsdbReadHandle->order) && pTsdbReadHandle->window.ekey >= pBlockInfo->window.ekey) {
H
Haojun Liao 已提交
1925 1926
    endPos = pBlockInfo->rows - 1;
    cur->mixBlock = (cur->pos != 0);
1927
  } else if (!ASCENDING_TRAVERSE(pTsdbReadHandle->order) && pTsdbReadHandle->window.ekey <= pBlockInfo->window.skey) {
H
Haojun Liao 已提交
1928 1929 1930 1931
    endPos = 0;
    cur->mixBlock = (cur->pos != pBlockInfo->rows - 1);
  } else {
    assert(pCols->numOfRows > 0);
1932
    endPos = doBinarySearchKey(pCols->cols[0].pData, pCols->numOfRows, pTsdbReadHandle->window.ekey, order);
H
Haojun Liao 已提交
1933 1934 1935 1936 1937 1938
    cur->mixBlock = true;
  }

  return endPos;
}

H
[td-32]  
hjxilinx 已提交
1939 1940
// 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
1941 1942
static void doMergeTwoLevelData(STsdbReadHandle* pTsdbReadHandle, STableCheckInfo* pCheckInfo, SBlock* pBlock) {
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
1943
  SDataBlockInfo blockInfo = GET_FILE_DATA_BLOCK_INFO(pCheckInfo, pBlock);
H
Hongze Cheng 已提交
1944
  STsdbCfg*      pCfg = REPO_CFG(pTsdbReadHandle->pTsdb);
H
Haojun Liao 已提交
1945

1946
  initTableMemIterator(pTsdbReadHandle, pCheckInfo);
1947

1948 1949
  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 已提交
1950
         cur->pos >= 0 && cur->pos < pBlock->numOfRows);
H
Haojun Liao 已提交
1951

1952
  TSKEY* tsArray = pCols->cols[0].pData;
dengyihao's avatar
dengyihao 已提交
1953 1954
  assert(pCols->numOfRows == pBlock->numOfRows && tsArray[0] == pBlock->keyFirst &&
         tsArray[pBlock->numOfRows - 1] == pBlock->keyLast);
1955 1956

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

dengyihao's avatar
dengyihao 已提交
1959
  int32_t step = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 1 : -1;
1960
  int32_t numOfCols = (int32_t)(QH_GET_NUM_OF_COLS(pTsdbReadHandle));
1961

H
Haojun Liao 已提交
1962
  STable* pTable = NULL;
1963
  int32_t endPos = getEndPosInDataBlock(pTsdbReadHandle, &blockInfo);
H
Haojun Liao 已提交
1964

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

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

dengyihao's avatar
dengyihao 已提交
1974 1975
  int16_t   rv1 = -1;
  int16_t   rv2 = -1;
1976 1977
  STSchema* pSchema1 = NULL;
  STSchema* pSchema2 = NULL;
D
fix bug  
dapan1121 已提交
1978

H
Haojun Liao 已提交
1979 1980
  int32_t pos = cur->pos;
  cur->win = TSWINDOW_INITIALIZER;
1981

1982 1983
  // no data in buffer, load data from file directly
  if (pCheckInfo->iiter == NULL && pCheckInfo->iter == NULL) {
1984
    copyAllRemainRowsFromFileBlock(pTsdbReadHandle, pCheckInfo, &blockInfo, endPos);
1985
    return;
1986
  } else if (pCheckInfo->iter != NULL || pCheckInfo->iiter != NULL) {
1987
    SSkipListNode* node = NULL;
C
Cary Xu 已提交
1988 1989
    TSKEY          lastRowKey = TSKEY_INITIAL_VAL;

1990
    do {
H
Haojun Liao 已提交
1991
      STSRow* row2 = NULL;
C
Cary Xu 已提交
1992
      STSRow* row1 = getSRowInTableMem(pCheckInfo, pTsdbReadHandle->order, pCfg->update, &row2, TD_VER_MAX);
1993
      if (row1 == NULL) {
H
[td-32]  
hjxilinx 已提交
1994
        break;
1995
      }
1996

H
Haojun Liao 已提交
1997
      TSKEY key = TD_ROW_KEY(row1);
1998 1999
      if ((key > pTsdbReadHandle->window.ekey && ASCENDING_TRAVERSE(pTsdbReadHandle->order)) ||
          (key < pTsdbReadHandle->window.ekey && !ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
2000 2001 2002
        break;
      }

dengyihao's avatar
dengyihao 已提交
2003 2004 2005 2006
      if (((pos > endPos || tsArray[pos] > pTsdbReadHandle->window.ekey) &&
           ASCENDING_TRAVERSE(pTsdbReadHandle->order)) ||
          ((pos < endPos || tsArray[pos] < pTsdbReadHandle->window.ekey) &&
           !ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
2007 2008 2009
        break;
      }

2010 2011
      if ((key < tsArray[pos] && ASCENDING_TRAVERSE(pTsdbReadHandle->order)) ||
          (key > tsArray[pos] && !ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
H
Haojun Liao 已提交
2012
        if (rv1 != TD_ROW_SVER(row1)) {
dengyihao's avatar
dengyihao 已提交
2013
          //          pSchema1 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row1));
H
Haojun Liao 已提交
2014
          rv1 = TD_ROW_SVER(row1);
C
Cary Xu 已提交
2015
        }
dengyihao's avatar
dengyihao 已提交
2016 2017
        if (row2 && rv2 != TD_ROW_SVER(row2)) {
          //          pSchema2 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row2));
H
Haojun Liao 已提交
2018
          rv2 = TD_ROW_SVER(row2);
2019
        }
dengyihao's avatar
dengyihao 已提交
2020

C
Cary Xu 已提交
2021
        numOfRows += mergeTwoRowFromMem(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, &curRow, row1, row2, numOfCols,
C
Cary Xu 已提交
2022
                           pCheckInfo->tableId, pSchema1, pSchema2, pCfg->update, &lastRowKey);
C
Cary Xu 已提交
2023
        // numOfRows += 1;
2024 2025 2026
        if (cur->win.skey == TSKEY_INITIAL_VAL) {
          cur->win.skey = key;
        }
2027

2028
        cur->win.ekey = key;
dengyihao's avatar
dengyihao 已提交
2029
        cur->lastKey = key + step;
2030 2031
        cur->mixBlock = true;

2032
        moveToNextRowInMem(pCheckInfo);
2033
      } else if (key == tsArray[pos]) {  // data in buffer has the same timestamp of data in file block, ignore it
C
Cary Xu 已提交
2034
#if 0
H
TD-1439  
Hongze Cheng 已提交
2035
        if (pCfg->update) {
dengyihao's avatar
dengyihao 已提交
2036
          if (pCfg->update == TD_ROW_PARTIAL_UPDATE) {
2037
            doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, numOfRows, pos, pos);
D
fix bug  
dapan1121 已提交
2038
          }
H
Haojun Liao 已提交
2039
          if (rv1 != TD_ROW_SVER(row1)) {
dengyihao's avatar
dengyihao 已提交
2040
            //            pSchema1 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row1));
H
Haojun Liao 已提交
2041
            rv1 = TD_ROW_SVER(row1);
2042
          }
dengyihao's avatar
dengyihao 已提交
2043 2044
          if (row2 && rv2 != TD_ROW_SVER(row2)) {
            //            pSchema2 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row2));
H
Haojun Liao 已提交
2045
            rv2 = TD_ROW_SVER(row2);
2046
          }
dengyihao's avatar
dengyihao 已提交
2047

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

          if (rv1 != TD_ROW_SVER(row1)) {
            //            pSchema1 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row1));
            rv1 = TD_ROW_SVER(row1);
          }
          if (row2 && rv2 != TD_ROW_SVER(row2)) {
            //            pSchema2 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row2));
            rv2 = TD_ROW_SVER(row2);
          }

          numOfRows += mergeTwoRowFromMem(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, &curRow, row1, row2, numOfCols,
C
Cary Xu 已提交
2079
                             pCheckInfo->tableId, pSchema1, pSchema2, pCfg->update, &lastRowKey);
C
Cary Xu 已提交
2080 2081 2082 2083 2084 2085 2086 2087 2088
          // ++numOfRows;
          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 已提交
2089 2090 2091 2092 2093
          moveToNextRowInMem(pCheckInfo);
          pos += step;
        } else {
          moveToNextRowInMem(pCheckInfo);
        }
2094
      } else if ((key > tsArray[pos] && ASCENDING_TRAVERSE(pTsdbReadHandle->order)) ||
dengyihao's avatar
dengyihao 已提交
2095
                 (key < tsArray[pos] && !ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
2096 2097 2098
        if (cur->win.skey == TSKEY_INITIAL_VAL) {
          cur->win.skey = tsArray[pos];
        }
2099

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

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

2118
        int32_t qstart = 0, qend = 0;
2119
        getQualifiedRowsPos(pTsdbReadHandle, pos, end, numOfRows, &qstart, &qend);
2120

2121
        numOfRows = doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, numOfRows, qstart, qend);
2122 2123
        pos += (qend - qstart + 1) * step;

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

H
[td-32]  
hjxilinx 已提交
2231 2232 2233
  return midPos;
}

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

2307
  int32_t cnt = 0;
2308
  int32_t numOfQualTables = 0;
H
Haojun Liao 已提交
2309

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

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

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

2325
    sup.pDataBlockInfo[numOfQualTables] = (STableBlockInfo*)buf;
2326 2327

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

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

2335
    numOfQualTables++;
2336 2337
  }

H
Haojun Liao 已提交
2338
  assert(numOfBlocks == cnt);
2339

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

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

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

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

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

  int32_t numOfTotal = 0;

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

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

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

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

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

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

  return TSDB_CODE_SUCCESS;
}

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

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

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

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

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

  int32_t code = TSDB_CODE_SUCCESS;

2425
  int32_t numOfBlocks = 0;
2426
  int32_t numOfTables = (int32_t)taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
2427

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

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

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

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

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

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

2458
    tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2459

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

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

H
Haojun Liao 已提交
2469 2470
    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 已提交
2471

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

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

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

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

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

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

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

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

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

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

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

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

H
Haojun Liao 已提交
2527
  pTableBlockInfo->totalSize = 0;
2528
  pTableBlockInfo->totalRows = 0;
H
Haojun Liao 已提交
2529

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

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

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

H
Haojun Liao 已提交
2542
  pTableBlockInfo->numOfFiles += 1;
H
Haojun Liao 已提交
2543

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

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

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

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

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

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

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

2580
    tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2581

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

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

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

    if (numOfBlocks == 0) {
      continue;
    }

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

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

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

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

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

  return code;
}

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

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

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

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

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

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

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

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

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

2686 2687 2688
  return false;
}

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

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

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

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

    i++;
  }

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

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

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

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

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

H
Haojun Liao 已提交
2735 2736

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

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

      break;
    }

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

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

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

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

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

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

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

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

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

  return numOfRows;
}

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

dengyihao's avatar
dengyihao 已提交
2917
// static bool loadCachedLast(STsdbReadHandle* pTsdbReadHandle) {
2918 2919 2920 2921 2922 2923 2924 2925 2926 2927 2928 2929 2930 2931 2932 2933 2934 2935 2936
//  // 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 已提交
2937 2938
//      tsdbWarn("no last cached for table %s, uid:%" PRIu64 ",tid:%d", pTable->name->data, pTable->uid,
//      pTable->tableId); continue;
2939 2940 2941 2942 2943 2944 2945 2946 2947 2948 2949 2950 2951 2952 2953 2954 2955 2956 2957 2958 2959 2960 2961 2962 2963 2964 2965 2966 2967 2968 2969 2970 2971 2972 2973 2974 2975 2976 2977 2978 2979 2980 2981 2982 2983 2984 2985 2986 2987 2988 2989 2990 2991 2992 2993 2994 2995 2996 2997 2998 2999 3000 3001 3002 3003 3004 3005 3006 3007 3008 3009 3010 3011 3012 3013 3014 3015 3016 3017 3018 3019 3020 3021 3022 3023 3024 3025 3026 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
//    }
//
//    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 已提交
3069 3070 3071
  assert(numOfTables > 0);

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

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

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

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

    terrno = TSDB_CODE_SUCCESS;

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

  return false;
3094 3095
}

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

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

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

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

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

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

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

        return false;
      }

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

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

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

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

dengyihao's avatar
dengyihao 已提交
3157
// static int32_t doGetExternalRow(STsdbReadHandle* pTsdbReadHandle, int16_t type, STsdbMemTable* pMemRef) {
3158 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
//  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 已提交
3192
//    colInfo.pData = taosMemoryCalloc(1, pCol->info.bytes);
3193 3194 3195 3196 3197 3198 3199 3200 3201
//    if (colInfo.pData == NULL) {
//      terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
//      goto out_of_memory;
//    }
//
//    taosArrayPush(row, &colInfo);
//  }
//
//  // load the previous row
3202
//  SQueryTableDataCond cond = {.numOfCols = numOfCols, .loadExternalRows = false, .type = BLOCK_LOAD_OFFSET_SEQ_ORDER};
3203 3204 3205 3206 3207 3208 3209 3210
//  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 已提交
3211
//  cond.colList = taosMemoryCalloc(cond.numOfCols, sizeof(SColumnInfo));
3212 3213 3214 3215 3216 3217 3218 3219 3220 3221
//  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 已提交
3222
//  pSecQueryHandle = tsdbQueryTablesImpl(pTsdbReadHandle->pTsdb, &cond, pTsdbReadHandle->idStr, pMemRef);
wafwerar's avatar
wafwerar 已提交
3223
//  taosMemoryFreeClear(cond.colList);
3224 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
//
//  // 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 已提交
3262
// out_of_memory:
3263
//  tsdbCleanupReadHandle(pSecQueryHandle);
3264 3265 3266
//  return terrno;
//}

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

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

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

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

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

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

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

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

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

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

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

3356
  return TSDB_CODE_SUCCESS;
3357 3358
}

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

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

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

  return code;
}

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

3630
#endif
3631

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

3759 3760 3761
  return pTableGroup;
}

dengyihao's avatar
dengyihao 已提交
3762
// static bool tableFilterFp(const void* pNode, void* param) {
3763 3764 3765 3766 3767 3768 3769 3770 3771 3772 3773 3774 3775 3776 3777 3778 3779 3780 3781 3782 3783 3784 3785 3786 3787 3788 3789 3790 3791 3792 3793 3794 3795 3796 3797 3798 3799 3800 3801 3802 3803 3804 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
//  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 已提交
3846

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

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

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

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

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

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

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

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

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

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

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

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

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

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

  taosArrayDestroy(res);
  return ret;

_error:
3928
  return terrno;
3929
}
3930

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

  taosArrayDestroy(pColumnInfoData);
  return NULL;
}

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

  taosArrayDestroy(pTableCheckInfo);
  return NULL;
}

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

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

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

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

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

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

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

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

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

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

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

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

    taosArrayDestroy(p);
  }

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

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 已提交
4112
    if (exprTreeApplyFilter(pExpr, pNode, param)) {
H
Haojun Liao 已提交
4113 4114 4115 4116 4117 4118 4119 4120 4121 4122 4123 4124 4125 4126 4127 4128 4129 4130 4131 4132 4133 4134 4135
      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 已提交
4136
    pCond->start       = taosMemoryCalloc(1, sizeof(SEndPoint));
H
Haojun Liao 已提交
4137
    pCond->start->optr = queryColInfo->optr;
Y
yihaoDeng 已提交
4138
    pCond->start->v    = queryColInfo->q;
H
Haojun Liao 已提交
4139
  } else if (optr == TSDB_RELATION_LESS || optr == TSDB_RELATION_LESS_EQUAL) {
wafwerar's avatar
wafwerar 已提交
4140
    pCond->end       = taosMemoryCalloc(1, sizeof(SEndPoint));
H
Haojun Liao 已提交
4141
    pCond->end->optr = queryColInfo->optr;
Y
yihaoDeng 已提交
4142 4143
    pCond->end->v    = queryColInfo->q;
  } else if (optr == TSDB_RELATION_IN) {
wafwerar's avatar
wafwerar 已提交
4144
    pCond->start       = taosMemoryCalloc(1, sizeof(SEndPoint));
Y
yihaoDeng 已提交
4145 4146 4147
    pCond->start->optr = queryColInfo->optr;
    pCond->start->v    = queryColInfo->q; 
  } else if (optr == TSDB_RELATION_LIKE) {
H
Haojun Liao 已提交
4148
    assert(0);
4149 4150
  } else if (optr == TSDB_RELATION_MATCH) {
    assert(0);
4151 4152
  } else if (optr == TSDB_RELATION_NMATCH) {
    assert(0);
H
Haojun Liao 已提交
4153 4154 4155 4156 4157 4158 4159 4160 4161 4162 4163 4164 4165 4166 4167 4168 4169 4170 4171 4172 4173 4174 4175 4176 4177 4178 4179 4180 4181 4182 4183 4184 4185 4186 4187 4188 4189 4190 4191 4192 4193 4194 4195 4196 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
  }

  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 已提交
4236 4237 4238 4239 4240 4241 4242 4243 4244 4245 4246 4247 4248
    } 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 已提交
4249 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
    } 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 已提交
4290 4291
  taosMemoryFree(cond.start);
  taosMemoryFree(cond.end);
H
Haojun Liao 已提交
4292 4293 4294 4295 4296 4297 4298 4299 4300 4301 4302 4303 4304 4305 4306 4307 4308 4309
  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);
4310 4311 4312
      } else if (pQueryInfo->optr == TSDB_RELATION_LIKE ||
                 pQueryInfo->optr == TSDB_RELATION_MATCH ||
                 pQueryInfo->optr == TSDB_RELATION_NMATCH) {
H
Haojun Liao 已提交
4313 4314 4315 4316 4317 4318 4319 4320 4321 4322 4323 4324 4325 4326 4327 4328
        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 已提交
4329 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
//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 已提交
4362
#endif