tsdbRead.c 148.2 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
        } else if (update) {
C
Cary Xu 已提交
1724 1725 1726 1727
          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
        } else if (update) {
C
Cary Xu 已提交
1745 1746 1747 1748
          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
    TSKEY          lastKeyAppend = TSKEY_INITIAL_VAL;
C
Cary Xu 已提交
1989

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

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

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

C
Cary Xu 已提交
2084 2085 2086 2087 2088 2089 2090 2091
          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 已提交
2092 2093 2094 2095 2096
          moveToNextRowInMem(pCheckInfo);
          pos += step;
        } else {
          moveToNextRowInMem(pCheckInfo);
        }
2097
      } else if ((key > tsArray[pos] && ASCENDING_TRAVERSE(pTsdbReadHandle->order)) ||
dengyihao's avatar
dengyihao 已提交
2098
                 (key < tsArray[pos] && !ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
2099 2100 2101
        if (cur->win.skey == TSKEY_INITIAL_VAL) {
          cur->win.skey = tsArray[pos];
        }
2102

2103
        int32_t end = doBinarySearchKey(pCols->cols[0].pData, pCols->numOfRows, key, order);
2104 2105
        assert(end != -1);

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

2121
        int32_t qstart = 0, qend = 0;
2122
        getQualifiedRowsPos(pTsdbReadHandle, pos, end, numOfRows, &qstart, &qend);
2123

C
Cary Xu 已提交
2124 2125 2126 2127 2128
        if ((lastKeyAppend != TSKEY_INITIAL_VAL) &&
            (lastKeyAppend != (ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? tsArray[qstart] : tsArray[qend]))) {
          ++curRow;
        }
        numOfRows = doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, curRow, qstart, qend);
2129
        pos += (qend - qstart + 1) * step;
C
Cary Xu 已提交
2130 2131 2132
        if (numOfRows > 0) {
          curRow = numOfRows - 1;
        }
2133

dengyihao's avatar
dengyihao 已提交
2134 2135
        cur->win.ekey = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? tsArray[qend] : tsArray[qstart];
        cur->lastKey = cur->win.ekey + step;
C
Cary Xu 已提交
2136
        lastKeyAppend = cur->win.ekey;
2137
      }
2138
    } while (numOfRows < pTsdbReadHandle->outputCapacity);
H
Haojun Liao 已提交
2139

2140
    if (numOfRows < pTsdbReadHandle->outputCapacity) {
H
Haojun Liao 已提交
2141 2142 2143 2144
      /**
       * 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.
       */
2145
      if (node == NULL ||
H
Haojun Liao 已提交
2146
          ((TD_ROW_KEY((STSRow*)SL_GET_NODE_DATA(node)) > pTsdbReadHandle->window.ekey) &&
2147
           ASCENDING_TRAVERSE(pTsdbReadHandle->order)) ||
H
Haojun Liao 已提交
2148
          ((TD_ROW_KEY((STSRow*)SL_GET_NODE_DATA(node)) < pTsdbReadHandle->window.ekey) &&
2149
           !ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
2150 2151 2152 2153 2154
        // 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];
        }

2155
        int32_t start = -1, end = -1;
2156
        getQualifiedRowsPos(pTsdbReadHandle, pos, endPos, numOfRows, &start, &end);
2157

2158
        numOfRows = doCopyRowsFromFileBlock(pTsdbReadHandle, pTsdbReadHandle->outputCapacity, numOfRows, start, end);
2159
        pos += (end - start + 1) * step;
2160

dengyihao's avatar
dengyihao 已提交
2161 2162
        cur->win.ekey = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? tsArray[end] : tsArray[start];
        cur->lastKey = cur->win.ekey + step;
H
Haojun Liao 已提交
2163
        cur->mixBlock = true;
2164
      }
2165 2166
    }
  }
H
Haojun Liao 已提交
2167 2168

  cur->blockCompleted =
2169 2170
      (((pos > endPos || cur->lastKey > pTsdbReadHandle->window.ekey) && ASCENDING_TRAVERSE(pTsdbReadHandle->order)) ||
       ((pos < endPos || cur->lastKey < pTsdbReadHandle->window.ekey) && !ASCENDING_TRAVERSE(pTsdbReadHandle->order)));
2171

2172
  if (!ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
wafwerar's avatar
wafwerar 已提交
2173
    TSWAP(cur->win.skey, cur->win.ekey);
2174
  }
2175

2176 2177 2178
  moveDataToFront(pTsdbReadHandle, numOfRows, numOfCols);
  updateInfoAfterMerge(pTsdbReadHandle, pCheckInfo, numOfRows, pos);
  doCheckGeneratedBlockRange(pTsdbReadHandle);
H
Haojun Liao 已提交
2179

dengyihao's avatar
dengyihao 已提交
2180 2181 2182
  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);
2183 2184
}

2185
int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order) {
H
[td-32]  
hjxilinx 已提交
2186
  int    firstPos, lastPos, midPos = -1;
H
Haojun Liao 已提交
2187
  int    numOfRows;
2188 2189
  TSKEY* keyList;

H
[td-32]  
hjxilinx 已提交
2190
  if (num <= 0) return -1;
2191 2192

  keyList = (TSKEY*)pValue;
H
[td-32]  
hjxilinx 已提交
2193 2194
  firstPos = 0;
  lastPos = num - 1;
2195

2196
  if (order == TSDB_ORDER_DESC) {
H
[td-32]  
hjxilinx 已提交
2197 2198 2199 2200 2201
    // 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;
2202

H
Haojun Liao 已提交
2203 2204
      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1) + firstPos;
2205

H
[td-32]  
hjxilinx 已提交
2206 2207 2208 2209 2210 2211 2212 2213
      if (key < keyList[midPos]) {
        lastPos = midPos - 1;
      } else if (key > keyList[midPos]) {
        firstPos = midPos + 1;
      } else {
        break;
      }
    }
2214

H
[td-32]  
hjxilinx 已提交
2215 2216 2217 2218 2219
  } 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;
2220

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

H
Haojun Liao 已提交
2229 2230
      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1) + firstPos;
2231

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

H
[td-32]  
hjxilinx 已提交
2242 2243 2244
  return midPos;
}

2245
static void cleanBlockOrderSupporter(SBlockOrderSupporter* pSupporter, int32_t numOfTables) {
wafwerar's avatar
wafwerar 已提交
2246 2247
  taosMemoryFreeClear(pSupporter->numOfBlocksPerTable);
  taosMemoryFreeClear(pSupporter->blockIndexArray);
2248 2249

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

wafwerar's avatar
wafwerar 已提交
2254
  taosMemoryFreeClear(pSupporter->pDataBlockInfo);
2255 2256 2257 2258 2259 2260 2261 2262 2263 2264 2265
}

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

2266
  if (leftTableBlockIndex > pSupporter->numOfBlocksPerTable[leftTableIndex]) {
2267 2268
    /* left block is empty */
    return 1;
2269
  } else if (rightTableBlockIndex > pSupporter->numOfBlocksPerTable[rightTableIndex]) {
2270 2271 2272 2273 2274 2275 2276
    /* right block is empty */
    return -1;
  }

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

H
Haojun Liao 已提交
2277
  //    assert(pLeftBlockInfoEx->compBlock->offset != pRightBlockInfoEx->compBlock->offset);
dengyihao's avatar
dengyihao 已提交
2278
#if 0  // TODO: temporarily comment off requested by Dr. Liao
H
Haojun Liao 已提交
2279 2280
  if (pLeftBlockInfoEx->compBlock->offset == pRightBlockInfoEx->compBlock->offset &&
      pLeftBlockInfoEx->compBlock->last == pRightBlockInfoEx->compBlock->last) {
B
Bomin Zhang 已提交
2281
    tsdbError("error in header file, two block with same offset:%" PRId64, (int64_t)pLeftBlockInfoEx->compBlock->offset);
2282
  }
H
Haojun Liao 已提交
2283
#endif
2284

H
Haojun Liao 已提交
2285
  return pLeftBlockInfoEx->compBlock->offset > pRightBlockInfoEx->compBlock->offset ? 1 : -1;
2286 2287
}

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

2291 2292
  if (pTsdbReadHandle->allocSize < size) {
    pTsdbReadHandle->allocSize = (int32_t)size;
wafwerar's avatar
wafwerar 已提交
2293
    char* tmp = taosMemoryRealloc(pTsdbReadHandle->pDataBlockInfo, pTsdbReadHandle->allocSize);
H
Haojun Liao 已提交
2294 2295 2296 2297
    if (tmp == NULL) {
      return TSDB_CODE_TDB_OUT_OF_MEMORY;
    }

dengyihao's avatar
dengyihao 已提交
2298
    pTsdbReadHandle->pDataBlockInfo = (STableBlockInfo*)tmp;
2299 2300
  }

2301
  memset(pTsdbReadHandle->pDataBlockInfo, 0, size);
2302 2303
  *numOfAllocBlocks = numOfBlocks;

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

2307 2308
  SBlockOrderSupporter sup = {0};
  sup.numOfTables = numOfTables;
wafwerar's avatar
wafwerar 已提交
2309 2310 2311
  sup.numOfBlocksPerTable = taosMemoryCalloc(1, sizeof(int32_t) * numOfTables);
  sup.blockIndexArray = taosMemoryCalloc(1, sizeof(int32_t) * numOfTables);
  sup.pDataBlockInfo = taosMemoryCalloc(1, POINTER_BYTES * numOfTables);
2312

2313
  if (sup.numOfBlocksPerTable == NULL || sup.blockIndexArray == NULL || sup.pDataBlockInfo == NULL) {
2314
    cleanBlockOrderSupporter(&sup, 0);
2315
    return TSDB_CODE_TDB_OUT_OF_MEMORY;
2316
  }
H
Haojun Liao 已提交
2317

2318
  int32_t cnt = 0;
2319
  int32_t numOfQualTables = 0;
H
Haojun Liao 已提交
2320

2321
  for (int32_t j = 0; j < numOfTables; ++j) {
2322
    STableCheckInfo* pTableCheck = (STableCheckInfo*)taosArrayGet(pTsdbReadHandle->pTableCheckInfo, j);
2323 2324 2325
    if (pTableCheck->numOfBlocks <= 0) {
      continue;
    }
H
Haojun Liao 已提交
2326

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

wafwerar's avatar
wafwerar 已提交
2330
    char* buf = taosMemoryMalloc(sizeof(STableBlockInfo) * pTableCheck->numOfBlocks);
2331
    if (buf == NULL) {
2332
      cleanBlockOrderSupporter(&sup, numOfQualTables);
2333
      return TSDB_CODE_TDB_OUT_OF_MEMORY;
2334 2335
    }

2336
    sup.pDataBlockInfo[numOfQualTables] = (STableBlockInfo*)buf;
2337 2338

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

H
Haojun Liao 已提交
2341 2342
      pBlockInfo->compBlock = &pBlock[k];
      pBlockInfo->pTableCheckInfo = pTableCheck;
2343 2344 2345
      cnt++;
    }

2346
    numOfQualTables++;
2347 2348
  }

H
Haojun Liao 已提交
2349
  assert(numOfBlocks == cnt);
2350

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

H
Haojun Liao 已提交
2356
    tsdbDebug("%p create data blocks info struct completed for 1 table, %d blocks not sorted %s", pTsdbReadHandle, cnt,
dengyihao's avatar
dengyihao 已提交
2357
              pTsdbReadHandle->idStr);
H
Haojun Liao 已提交
2358 2359
    return TSDB_CODE_SUCCESS;
  }
2360

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

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

2367
  SMultiwayMergeTreeInfo* pTree = NULL;
dengyihao's avatar
dengyihao 已提交
2368
  uint8_t                 ret = tMergeTreeCreate(&pTree, sup.numOfTables, &sup, dataBlockOrderCompar);
2369 2370
  if (ret != TSDB_CODE_SUCCESS) {
    cleanBlockOrderSupporter(&sup, numOfTables);
2371
    return TSDB_CODE_TDB_OUT_OF_MEMORY;
2372 2373 2374 2375 2376
  }

  int32_t numOfTotal = 0;

  while (numOfTotal < cnt) {
2377
    int32_t pos = tMergeTreeGetChosenIndex(pTree);
2378 2379
    int32_t index = sup.blockIndexArray[pos]++;

H
Haojun Liao 已提交
2380
    STableBlockInfo* pBlocksInfo = sup.pDataBlockInfo[pos];
2381
    pTsdbReadHandle->pDataBlockInfo[numOfTotal++] = pBlocksInfo[index];
2382 2383

    // set data block index overflow, in order to disable the offset comparator
2384 2385
    if (sup.blockIndexArray[pos] >= sup.numOfBlocksPerTable[pos]) {
      sup.blockIndexArray[pos] = sup.numOfBlocksPerTable[pos] + 1;
2386
    }
2387

H
Haojun Liao 已提交
2388
    tMergeTreeAdjust(pTree, tMergeTreeGetAdjustIndex(pTree));
2389 2390 2391 2392 2393
  }

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

H
Haojun Liao 已提交
2398
  tsdbDebug("%p %d data blocks sort completed, %s", pTsdbReadHandle, cnt, pTsdbReadHandle->idStr);
2399
  cleanBlockOrderSupporter(&sup, numOfTables);
wafwerar's avatar
wafwerar 已提交
2400
  taosMemoryFree(pTree);
2401 2402 2403 2404

  return TSDB_CODE_SUCCESS;
}

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

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

dengyihao's avatar
dengyihao 已提交
2411
  while (1) {
2412
    int32_t code = loadFileDataBlock(pTsdbReadHandle, pNext->compBlock, pNext->pTableCheckInfo, exists);
H
Haojun Liao 已提交
2413 2414 2415 2416
    if (code != TSDB_CODE_SUCCESS || *exists) {
      return code;
    }

2417 2418
    if ((cur->slot == pTsdbReadHandle->numOfBlocks - 1 && ASCENDING_TRAVERSE(pTsdbReadHandle->order)) ||
        (cur->slot == 0 && !ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
H
Haojun Liao 已提交
2419
      // all data blocks in current file has been checked already, try next file if exists
2420
      return getFirstFileDataBlock(pTsdbReadHandle, exists);
H
Haojun Liao 已提交
2421 2422 2423 2424
    } else {  // next block of the same file
      cur->slot += step;
      cur->mixBlock = false;
      cur->blockCompleted = false;
2425
      pNext = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
H
Haojun Liao 已提交
2426 2427 2428 2429
    }
  }
}

2430 2431 2432
static int32_t getFirstFileDataBlock(STsdbReadHandle* pTsdbReadHandle, bool* exists) {
  pTsdbReadHandle->numOfBlocks = 0;
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
H
Haojun Liao 已提交
2433 2434 2435

  int32_t code = TSDB_CODE_SUCCESS;

2436
  int32_t numOfBlocks = 0;
2437
  int32_t numOfTables = (int32_t)taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
2438

C
Cary Xu 已提交
2439
  STsdbKeepCfg* pCfg = REPO_KEEP_CFG(pTsdbReadHandle->pTsdb);
C
Cary Xu 已提交
2440
  STimeWindow   win = TSWINDOW_INITIALIZER;
2441

H
Hongze Cheng 已提交
2442
  while (true) {
2443
    tsdbRLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2444

2445 2446
    if ((pTsdbReadHandle->pFileGroup = tsdbFSIterNext(&pTsdbReadHandle->fileIter)) == NULL) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2447 2448 2449
      break;
    }

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

    // current file are not overlapped with query time window, ignore remain files
2453 2454 2455
    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 已提交
2456 2457
      tsdbDebug("%p remain files are not qualified for qrange:%" PRId64 "-%" PRId64 ", ignore, %s", pTsdbReadHandle,
                pTsdbReadHandle->window.skey, pTsdbReadHandle->window.ekey, pTsdbReadHandle->idStr);
2458 2459
      pTsdbReadHandle->pFileGroup = NULL;
      assert(pTsdbReadHandle->numOfBlocks == 0);
2460 2461 2462
      break;
    }

2463 2464
    if (tsdbSetAndOpenReadFSet(&pTsdbReadHandle->rhelper, pTsdbReadHandle->pFileGroup) < 0) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2465 2466 2467 2468
      code = terrno;
      break;
    }

2469
    tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Hongze Cheng 已提交
2470

2471
    if (tsdbLoadBlockIdx(&pTsdbReadHandle->rhelper) < 0) {
H
Hongze Cheng 已提交
2472 2473 2474 2475
      code = terrno;
      break;
    }

2476
    if ((code = getFileCompInfo(pTsdbReadHandle, &numOfBlocks)) != TSDB_CODE_SUCCESS) {
2477 2478
      break;
    }
H
Haojun Liao 已提交
2479

H
Haojun Liao 已提交
2480 2481
    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 已提交
2482

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

2488
    // todo return error code to query engine
dengyihao's avatar
dengyihao 已提交
2489 2490
    if ((code = createDataBlocksInfo(pTsdbReadHandle, numOfBlocks, &pTsdbReadHandle->numOfBlocks)) !=
        TSDB_CODE_SUCCESS) {
2491 2492
      break;
    }
H
Haojun Liao 已提交
2493

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

2500
  // no data in file anymore
2501
  if (pTsdbReadHandle->numOfBlocks <= 0 || code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2502
    if (code == TSDB_CODE_SUCCESS) {
2503
      assert(pTsdbReadHandle->pFileGroup == NULL);
H
Haojun Liao 已提交
2504 2505
    }

D
dapan1121 已提交
2506
    cur->fid = INT32_MIN;  // denote that there are no data in file anymore
H
Haojun Liao 已提交
2507 2508
    *exists = false;
    return code;
2509
  }
H
Haojun Liao 已提交
2510

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

2515 2516
  STableBlockInfo* pBlockInfo = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
  return getDataBlockRv(pTsdbReadHandle, pBlockInfo, exists);
H
Haojun Liao 已提交
2517 2518 2519 2520 2521 2522 2523
}

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

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

2527 2528
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
  assert(cur->slot < pTsdbReadHandle->numOfBlocks && cur->slot >= 0);
H
Haojun Liao 已提交
2529 2530

  cur->slot += step;
dengyihao's avatar
dengyihao 已提交
2531
  cur->mixBlock = false;
H
Haojun Liao 已提交
2532
  cur->blockCompleted = false;
2533
}
H
Haojun Liao 已提交
2534 2535

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

H
Haojun Liao 已提交
2538
  pTableBlockInfo->totalSize = 0;
2539
  pTableBlockInfo->totalRows = 0;
H
Haojun Liao 已提交
2540

2541
  STsdbFS* pFileHandle = REPO_FS(pTsdbReadHandle->pTsdb);
H
Haojun Liao 已提交
2542 2543

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

  tsdbRLockFS(pFileHandle);
2549 2550
  tsdbFSIterInit(&pTsdbReadHandle->fileIter, pFileHandle, pTsdbReadHandle->order);
  tsdbFSIterSeek(&pTsdbReadHandle->fileIter, fid);
H
Haojun Liao 已提交
2551 2552
  tsdbUnLockFS(pFileHandle);

H
Haojun Liao 已提交
2553
  pTableBlockInfo->numOfFiles += 1;
H
Haojun Liao 已提交
2554

H
Haojun Liao 已提交
2555
  int32_t     code = TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
2556
  int32_t     numOfBlocks = 0;
2557
  int32_t     numOfTables = (int32_t)taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
dengyihao's avatar
dengyihao 已提交
2558
  int         defaultRows = 4096;  // TSDB_DEFAULT_BLOCK_ROWS(pCfg->maxRowsPerFileBlock);
H
Haojun Liao 已提交
2559 2560
  STimeWindow win = TSWINDOW_INITIALIZER;

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

H
Haojun Liao 已提交
2563 2564
  while (true) {
    numOfBlocks = 0;
2565
    tsdbRLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2566

2567 2568
    if ((pTsdbReadHandle->pFileGroup = tsdbFSIterNext(&pTsdbReadHandle->fileIter)) == NULL) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2569 2570 2571
      break;
    }

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

    // current file are not overlapped with query time window, ignore remain files
dengyihao's avatar
dengyihao 已提交
2575 2576
    if ((ascTraverse && win.skey > pTsdbReadHandle->window.ekey) ||
        (!ascTraverse && win.ekey < pTsdbReadHandle->window.ekey)) {
2577
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2578 2579
      tsdbDebug("%p remain files are not qualified for qrange:%" PRId64 "-%" PRId64 ", ignore, %s", pTsdbReadHandle,
                pTsdbReadHandle->window.skey, pTsdbReadHandle->window.ekey, pTsdbReadHandle->idStr);
2580
      pTsdbReadHandle->pFileGroup = NULL;
H
Haojun Liao 已提交
2581 2582 2583
      break;
    }

H
Haojun Liao 已提交
2584
    pTableBlockInfo->numOfFiles += 1;
2585 2586
    if (tsdbSetAndOpenReadFSet(&pTsdbReadHandle->rhelper, pTsdbReadHandle->pFileGroup) < 0) {
      tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2587 2588 2589 2590
      code = terrno;
      break;
    }

2591
    tsdbUnLockFS(REPO_FS(pTsdbReadHandle->pTsdb));
H
Haojun Liao 已提交
2592

2593
    if (tsdbLoadBlockIdx(&pTsdbReadHandle->rhelper) < 0) {
H
Haojun Liao 已提交
2594 2595 2596 2597
      code = terrno;
      break;
    }

2598
    if ((code = getFileCompInfo(pTsdbReadHandle, &numOfBlocks)) != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2599 2600 2601
      break;
    }

H
Haojun Liao 已提交
2602 2603
    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 已提交
2604 2605 2606 2607 2608 2609

    if (numOfBlocks == 0) {
      continue;
    }

    for (int32_t i = 0; i < numOfTables; ++i) {
2610
      STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, i);
H
Haojun Liao 已提交
2611 2612 2613

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

H
Haojun Liao 已提交
2616
        int32_t numOfRows = pBlock[j].numOfRows;
2617
        pTableBlockInfo->totalRows += numOfRows;
H
Haojun Liao 已提交
2618 2619 2620 2621 2622 2623 2624 2625 2626 2627 2628
        if (numOfRows > pTableBlockInfo->maxRows) {
          pTableBlockInfo->maxRows = numOfRows;
        }

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

        if (numOfRows < defaultRows) {
          pTableBlockInfo->numOfSmallBlocks += 1;
        }
dengyihao's avatar
dengyihao 已提交
2629 2630 2631
        //        int32_t  stepIndex = (numOfRows-1)/TSDB_BLOCK_DIST_STEP_ROWS;
        //        SFileBlockInfo *blockInfo = (SFileBlockInfo*)taosArrayGet(pTableBlockInfo->dataBlockInfos, stepIndex);
        //        blockInfo->numBlocksOfStep++;
H
Haojun Liao 已提交
2632 2633 2634 2635 2636 2637 2638
      }
    }
  }

  return code;
}

2639 2640 2641
static int32_t getDataBlocksInFiles(STsdbReadHandle* pTsdbReadHandle, bool* exists) {
  STsdbFS*       pFileHandle = REPO_FS(pTsdbReadHandle->pTsdb);
  SQueryFilePos* cur = &pTsdbReadHandle->cur;
2642 2643

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

H
Hongze Cheng 已提交
2649
    tsdbRLockFS(pFileHandle);
2650 2651
    tsdbFSIterInit(&pTsdbReadHandle->fileIter, pFileHandle, pTsdbReadHandle->order);
    tsdbFSIterSeek(&pTsdbReadHandle->fileIter, fid);
H
Hongze Cheng 已提交
2652
    tsdbUnLockFS(pFileHandle);
2653

2654
    return getFirstFileDataBlock(pTsdbReadHandle, exists);
2655
  } else {
2656
    // check if current file block is all consumed
2657
    STableBlockInfo* pBlockInfo = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
2658
    STableCheckInfo* pCheckInfo = pBlockInfo->pTableCheckInfo;
H
Haojun Liao 已提交
2659

2660
    // current block is done, try next
H
Haojun Liao 已提交
2661
    if ((!cur->mixBlock) || cur->blockCompleted) {
H
Haojun Liao 已提交
2662
      // all data blocks in current file has been checked already, try next file if exists
2663
    } else {
H
Haojun Liao 已提交
2664 2665
      tsdbDebug("%p continue in current data block, index:%d, pos:%d, %s", pTsdbReadHandle, cur->slot, cur->pos,
                pTsdbReadHandle->idStr);
2666 2667
      int32_t code = handleDataMergeIfNeeded(pTsdbReadHandle, pBlockInfo->compBlock, pCheckInfo);
      *exists = (pTsdbReadHandle->realNumOfRows > 0);
H
Haojun Liao 已提交
2668

H
Haojun Liao 已提交
2669 2670 2671 2672 2673 2674 2675
      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
2676 2677
    if (isEndFileDataBlock(cur, pTsdbReadHandle->numOfBlocks, ASCENDING_TRAVERSE(pTsdbReadHandle->order))) {
      return getFirstFileDataBlock(pTsdbReadHandle, exists);
H
Haojun Liao 已提交
2678
    } else {
2679 2680 2681
      moveToNextDataBlockInCurrentFile(pTsdbReadHandle);
      STableBlockInfo* pNext = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
      return getDataBlockRv(pTsdbReadHandle, pNext, exists);
2682 2683
    }
  }
2684 2685
}

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

2689 2690
  while (pTsdbReadHandle->activeIndex < numOfTables) {
    if (hasMoreDataInCache(pTsdbReadHandle)) {
2691 2692
      return true;
    }
H
Haojun Liao 已提交
2693

2694
    pTsdbReadHandle->activeIndex += 1;
2695
  }
H
Haojun Liao 已提交
2696

2697 2698 2699
  return false;
}

dengyihao's avatar
dengyihao 已提交
2700
// todo not unref yet, since it is not support multi-group interpolation query
H
Haojun Liao 已提交
2701
static UNUSED_FUNC void changeQueryHandleForInterpQuery(tsdbReaderT pHandle) {
H
Haojun Liao 已提交
2702
  // filter the queried time stamp in the first place
dengyihao's avatar
dengyihao 已提交
2703
  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle;
H
Haojun Liao 已提交
2704 2705

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

  int32_t i = 0;
dengyihao's avatar
dengyihao 已提交
2709
  while (i < numOfTables) {
2710
    STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, i);
H
Haojun Liao 已提交
2711 2712

    // the first qualified table for interpolation query
dengyihao's avatar
dengyihao 已提交
2713 2714 2715 2716
    //    if ((pTsdbReadHandle->window.skey <= pCheckInfo->pTableObj->lastKey) &&
    //        (pCheckInfo->pTableObj->lastKey != TSKEY_INITIAL_VAL)) {
    //      break;
    //    }
H
Haojun Liao 已提交
2717 2718 2719 2720 2721 2722 2723 2724 2725

    i++;
  }

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

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

2729 2730
  info.lastKey = pTsdbReadHandle->window.skey;
  taosArrayPush(pTsdbReadHandle->pTableCheckInfo, &info);
H
Haojun Liao 已提交
2731 2732 2733
}

static int tsdbReadRowsFromCache(STableCheckInfo* pCheckInfo, TSKEY maxKey, int maxRowsToRead, STimeWindow* win,
2734
                                 STsdbReadHandle* pTsdbReadHandle) {
dengyihao's avatar
dengyihao 已提交
2735
  int       numOfRows = 0;
C
Cary Xu 已提交
2736
  int       curRows = 0;
dengyihao's avatar
dengyihao 已提交
2737
  int32_t   numOfCols = (int32_t)taosArrayGetSize(pTsdbReadHandle->pColumns);
H
Hongze Cheng 已提交
2738
  STsdbCfg* pCfg = REPO_CFG(pTsdbReadHandle->pTsdb);
H
Haojun Liao 已提交
2739 2740
  win->skey = TSKEY_INITIAL_VAL;

dengyihao's avatar
dengyihao 已提交
2741 2742
  int64_t   st = taosGetTimestampUs();
  int16_t   rv = -1;
D
fix bug  
dapan1121 已提交
2743
  STSchema* pSchema = NULL;
C
Cary Xu 已提交
2744 2745
  TSKEY     lastRowKey = TSKEY_INITIAL_VAL;

H
Haojun Liao 已提交
2746
  do {
C
Cary Xu 已提交
2747
    STSRow* row = getSRowInTableMem(pCheckInfo, pTsdbReadHandle->order, pCfg->update, NULL, TD_VER_MAX);
H
Haojun Liao 已提交
2748 2749 2750 2751
    if (row == NULL) {
      break;
    }

H
Haojun Liao 已提交
2752
    TSKEY key = TD_ROW_KEY(row);
dengyihao's avatar
dengyihao 已提交
2753 2754 2755 2756
    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 已提交
2757 2758 2759 2760 2761 2762 2763 2764 2765

      break;
    }

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

    win->ekey = key;
H
Haojun Liao 已提交
2766
    if (rv != TD_ROW_SVER(row)) {
H
Hongze Cheng 已提交
2767
      pSchema = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), pCheckInfo->tableId, 0);
H
Haojun Liao 已提交
2768
      rv = TD_ROW_SVER(row);
D
fix bug  
dapan1121 已提交
2769
    }
C
Cary Xu 已提交
2770 2771
    numOfRows += mergeTwoRowFromMem(pTsdbReadHandle, maxRowsToRead, &curRows, row, NULL, numOfCols, pCheckInfo->tableId,
                                    pSchema, NULL, pCfg->update, &lastRowKey);
H
Haojun Liao 已提交
2772

C
Cary Xu 已提交
2773
    if (numOfRows >= maxRowsToRead) {
H
Haojun Liao 已提交
2774 2775 2776 2777
      moveToNextRowInMem(pCheckInfo);
      break;
    }

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

C
Cary Xu 已提交
2780
  taosMemoryFreeClear(pSchema);  // free the STSChema
C
Cary Xu 已提交
2781

H
Haojun Liao 已提交
2782 2783 2784
  assert(numOfRows <= maxRowsToRead);

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

dengyihao's avatar
dengyihao 已提交
2788
    for (int32_t i = 0; i < numOfCols; ++i) {
2789
      SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
dengyihao's avatar
dengyihao 已提交
2790 2791
      memmove((char*)pColInfo->pData, (char*)pColInfo->pData + emptySize * pColInfo->info.bytes,
              numOfRows * pColInfo->info.bytes);
H
Haojun Liao 已提交
2792 2793 2794 2795
    }
  }

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

  return numOfRows;
}

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

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

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

2815
  metaCloseCtbCurosr(pCur);
H
Haojun Liao 已提交
2816 2817 2818 2819 2820 2821 2822 2823
  return TSDB_CODE_SUCCESS;
}

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

dengyihao's avatar
dengyihao 已提交
2927
// static bool loadCachedLast(STsdbReadHandle* pTsdbReadHandle) {
2928 2929 2930 2931 2932 2933 2934 2935 2936 2937 2938 2939 2940 2941 2942 2943 2944 2945 2946
//  // 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 已提交
2947 2948
//      tsdbWarn("no last cached for table %s, uid:%" PRIu64 ",tid:%d", pTable->name->data, pTable->uid,
//      pTable->tableId); continue;
2949 2950 2951 2952 2953 2954 2955 2956 2957 2958 2959 2960 2961 2962 2963 2964 2965 2966 2967 2968 2969 2970 2971 2972 2973 2974 2975 2976 2977 2978 2979 2980 2981 2982 2983 2984 2985 2986 2987 2988 2989 2990 2991 2992 2993 2994 2995 2996 2997 2998 2999 3000 3001 3002 3003 3004 3005 3006 3007 3008 3009 3010 3011 3012 3013 3014 3015 3016 3017 3018 3019 3020 3021 3022 3023 3024 3025 3026 3027 3028 3029 3030 3031 3032 3033 3034 3035 3036 3037 3038 3039 3040 3041 3042 3043 3044 3045 3046 3047 3048 3049 3050 3051 3052 3053 3054 3055 3056 3057 3058 3059 3060 3061 3062 3063 3064 3065 3066 3067 3068 3069 3070 3071 3072 3073 3074 3075 3076 3077 3078
//    }
//
//    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 已提交
3079 3080 3081
  assert(numOfTables > 0);

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

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

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

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

    terrno = TSDB_CODE_SUCCESS;

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

  return false;
3104 3105
}

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

dengyihao's avatar
dengyihao 已提交
3110
  for (int32_t i = 0; i < taosArrayGetSize(pTsdbReadHandle->pColumns); ++i) {
3111 3112 3113 3114
    SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
    colInfoDataCleanup(pColInfo, pTsdbReadHandle->outputCapacity);
  }

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

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

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

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

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

        return false;
      }

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

3366
  return TSDB_CODE_SUCCESS;
3367 3368
}

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

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

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

  return code;
}

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

H
Haojun Liao 已提交
3494 3495 3496 3497
  STableBlockInfo* pBlockInfo = &pHandle->pDataBlockInfo[c->slot];
  assert((c->slot >= 0 && c->slot < pHandle->numOfBlocks) || ((c->slot == pHandle->numOfBlocks) && (c->slot == 0)));

  // file block with sub-blocks has no statistics data
H
Haojun Liao 已提交
3498 3499 3500 3501
  if (pBlockInfo->compBlock->numOfSubBlocks > 1) {
    *pBlockStatis = NULL;
    return TSDB_CODE_SUCCESS;
  }
H
Haojun Liao 已提交
3502 3503

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

3640
#endif
3641

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

3769 3770 3771
  return pTableGroup;
}

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

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

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

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

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

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

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

H
Hongze Cheng 已提交
3895 3896
  metaReaderClear(&mr);

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

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

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

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

3915
    taosArrayDestroy(res);
3916 3917
    return ret;
  }
3918

H
hjxilinx 已提交
3919
  int32_t ret = TSDB_CODE_SUCCESS;
3920

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

  taosArrayDestroy(res);
  return ret;

_error:
3938
  return terrno;
3939
}
3940

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

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

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

  if (metaGetTableEntryByUid(&mr, uid) < 0) {
3952 3953
    terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
    goto _error;
3954
  }
3955

H
Hongze Cheng 已提交
3956 3957
  metaReaderClear(&mr);

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

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

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

3966
  taosArrayPush(pGroupInfo->pGroupList, &group);
3967
  return TSDB_CODE_SUCCESS;
3968

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

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

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

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

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

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

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

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

  taosArrayDestroy(pColumnInfoData);
  return NULL;
}

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

  taosArrayDestroy(pTableCheckInfo);
  return NULL;
}

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

4055
  pTsdbReadHandle->pColumns = doFreeColumnInfoData(pTsdbReadHandle->pColumns);
4056

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

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

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

4072
  tsdbDestroyReadH(&pTsdbReadHandle->rhelper);
H
Haojun Liao 已提交
4073

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

4077 4078
  pTsdbReadHandle->prev = doFreeColumnInfoData(pTsdbReadHandle->prev);
  pTsdbReadHandle->next = doFreeColumnInfoData(pTsdbReadHandle->next);
4079

4080
  SIOCostSummary* pCost = &pTsdbReadHandle->cost;
4081

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

wafwerar's avatar
wafwerar 已提交
4087
  taosMemoryFreeClear(pTsdbReadHandle);
4088
}
4089

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

    taosArrayDestroy(p);
  }

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

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 已提交
4122
    if (exprTreeApplyFilter(pExpr, pNode, param)) {
H
Haojun Liao 已提交
4123 4124 4125 4126 4127 4128 4129 4130 4131 4132 4133 4134 4135 4136 4137 4138 4139 4140 4141 4142 4143 4144 4145
      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 已提交
4146
    pCond->start       = taosMemoryCalloc(1, sizeof(SEndPoint));
H
Haojun Liao 已提交
4147
    pCond->start->optr = queryColInfo->optr;
Y
yihaoDeng 已提交
4148
    pCond->start->v    = queryColInfo->q;
H
Haojun Liao 已提交
4149
  } else if (optr == TSDB_RELATION_LESS || optr == TSDB_RELATION_LESS_EQUAL) {
wafwerar's avatar
wafwerar 已提交
4150
    pCond->end       = taosMemoryCalloc(1, sizeof(SEndPoint));
H
Haojun Liao 已提交
4151
    pCond->end->optr = queryColInfo->optr;
Y
yihaoDeng 已提交
4152 4153
    pCond->end->v    = queryColInfo->q;
  } else if (optr == TSDB_RELATION_IN) {
wafwerar's avatar
wafwerar 已提交
4154
    pCond->start       = taosMemoryCalloc(1, sizeof(SEndPoint));
Y
yihaoDeng 已提交
4155 4156 4157
    pCond->start->optr = queryColInfo->optr;
    pCond->start->v    = queryColInfo->q; 
  } else if (optr == TSDB_RELATION_LIKE) {
H
Haojun Liao 已提交
4158
    assert(0);
4159 4160
  } else if (optr == TSDB_RELATION_MATCH) {
    assert(0);
4161 4162
  } else if (optr == TSDB_RELATION_NMATCH) {
    assert(0);
H
Haojun Liao 已提交
4163 4164 4165 4166 4167 4168 4169 4170 4171 4172 4173 4174 4175 4176 4177 4178 4179 4180 4181 4182 4183 4184 4185 4186 4187 4188 4189 4190 4191 4192 4193 4194 4195 4196 4197 4198 4199 4200 4201 4202 4203 4204 4205 4206 4207 4208 4209 4210 4211 4212 4213 4214 4215 4216 4217 4218 4219 4220 4221 4222 4223 4224 4225 4226 4227 4228 4229 4230 4231 4232 4233 4234 4235 4236 4237 4238 4239 4240 4241 4242 4243 4244 4245
  }

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