Merge pull request #1592 from taosdata/feature/query

[td-98] fix bug in descending order query

src/kit/shell/src/shellEngine.c

@@ -284,7 +284,7 @@ void shellRunCommandOnServer(TAOS *con, char command[]) {
 /* Function to do regular expression check */
 int regex_match(const char *s, const char *reg, int cflags) {
   regex_t regex;
-  char    msgbuf[100];
+  char    msgbuf[100] = {0};
 
   /* Compile regular expression */
   if (regcomp(&regex, reg, cflags) != 0) {

src/kit/shell/src/shellMain.c

@@ -97,6 +97,8 @@ int main(int argc, char* argv[]) {
 
   /* Interupt handler. */
   struct sigaction act;
+  memset(&act, 0, sizeof(struct sigaction));
+  
   act.sa_handler = interruptHandler;
   sigaction(SIGTERM, &act, NULL);
   sigaction(SIGINT, &act, NULL);

src/vnode/tsdb/inc/tsdbMain.h

@@ -222,8 +222,9 @@ int         tsdbOpenFile(SFile *pFile, int oflag);
 int         tsdbCloseFile(SFile *pFile); SFileGroup *tsdbOpenFilesForCommit(STsdbFileH *pFileH, int fid);
 int         tsdbRemoveFileGroup(STsdbFileH *pFile, int fid);
 
-#define TSDB_FGROUP_ITER_FORWARD 0
-#define TSDB_FGROUP_ITER_BACKWARD 1
+#define TSDB_FGROUP_ITER_FORWARD  TSDB_ORDER_ASC
+#define TSDB_FGROUP_ITER_BACKWARD TSDB_ORDER_DESC
+
 typedef struct {
   int         numOfFGroups;
   SFileGroup *base;

src/vnode/tsdb/src/tsdbFile.c

@@ -154,7 +154,7 @@ void tsdbSeekFileGroupIter(SFileGroupIter *pIter, int fid) {
   }
   
   int flags = (pIter->direction == TSDB_FGROUP_ITER_FORWARD) ? TD_GE : TD_LE;
-  void *ptr = taosbsearch(&fid, pIter->base, sizeof(SFileGroup), pIter->numOfFGroups, compFGroupKey, flags);
+  void *ptr = taosbsearch(&fid, pIter->base, pIter->numOfFGroups, sizeof(SFileGroup), compFGroupKey, flags);
   if (ptr == NULL) {
     pIter->pFileGroup = NULL;
   } else {
@@ -173,7 +173,7 @@ SFileGroup *tsdbGetFileGroupNext(SFileGroupIter *pIter) {
       pIter->pFileGroup += 1;
     }
   } else {
-    if (pIter->pFileGroup - 1 == pIter->base) {
+    if (pIter->pFileGroup == pIter->base) {
       pIter->pFileGroup = NULL;
     } else {
       pIter->pFileGroup -= 1;

src/vnode/tsdb/src/tsdbRead.c

@@ -27,7 +27,7 @@
 #define EXTRA_BYTES 2
 #define PRIMARY_TSCOL_REQUIRED(c) (((SColumnInfoData*)taosArrayGet(c, 0))->info.colId == PRIMARYKEY_TIMESTAMP_COL_INDEX)
 #define QUERY_IS_ASC_QUERY(o) (o == TSDB_ORDER_ASC)
-#define QH_GET_NUM_OF_COLS(handle) (taosArrayGetSize((handle)->pColumns))
+#define QH_GET_NUM_OF_COLS(handle) ((size_t)(taosArrayGetSize((handle)->pColumns)))
 
 enum {
   QUERY_RANGE_LESS_EQUAL = 0,
@@ -87,14 +87,8 @@ typedef struct STableBlockInfo {
   int32_t          groupIdx; /* number of group is less than the total number of tables */
 } STableBlockInfo;
 
-enum {
-  SINGLE_TABLE_MODEL = 1,
-  MULTI_TABLE_MODEL = 2,
-};
-
 typedef struct STsdbQueryHandle {
   STsdbRepo*    pTsdb;
-  int8_t        model;  // access model, single table model or multi-table model
   SQueryFilePos cur;    // current position
   SQueryFilePos start;  // the start position, used for secondary/third iteration
 
@@ -128,21 +122,6 @@ typedef struct STsdbQueryHandle {
   SCompIdx*      compIndex;
 } STsdbQueryHandle;
 
-int32_t doAllocateBuf(STsdbQueryHandle* pQueryHandle, int32_t rowsPerFileBlock) {
-  // record the maximum column width among columns of this meter/metric
-  SColumnInfoData* pColumn = taosArrayGet(pQueryHandle->pColumns, 0);
-
-  int32_t maxColWidth = pColumn->info.bytes;
-  for (int32_t i = 1; i < QH_GET_NUM_OF_COLS(pQueryHandle); ++i) {
-    int32_t bytes = pColumn[i].info.bytes;
-    if (bytes > maxColWidth) {
-      maxColWidth = bytes;
-    }
-  }
-
-  return TSDB_CODE_SUCCESS;
-}
-
 static void tsdbInitDataBlockLoadInfo(SDataBlockLoadInfo* pBlockLoadInfo) {
   pBlockLoadInfo->slot = -1;
   pBlockLoadInfo->sid = -1;
@@ -176,23 +155,31 @@ tsdb_query_handle_t* tsdbQueryByTableId(tsdb_repo_t* tsdb, STsdbQueryCond* pCond
   for (int32_t i = 0; i < size; ++i) {
     STableId id = *(STableId*) taosArrayGet(idList, i);
     
+    STable* pTable = tsdbGetTableByUid(tsdbGetMeta(tsdb), id.uid);
+    if (pTable == NULL) {
+      dError("%p failed to get table, error uid:%" PRIu64, pQueryHandle, id.uid);
+      continue;
+    }
+    
     STableCheckInfo info = {
       .lastKey = pQueryHandle->window.skey,
       .tableId = id,
-        .pTableObj = tsdbGetTableByUid(tsdbGetMeta(tsdb), id.uid),  // todo this may be failed
-        .pCompInfo = NULL,
+      .pTableObj = pTable,
     };
 
-    assert(info.pTableObj != NULL);
     taosArrayPush(pQueryHandle->pTableCheckInfo, &info);
   }
 
-  pQueryHandle->model = (size > 1) ? MULTI_TABLE_MODEL : SINGLE_TABLE_MODEL;
-  pQueryHandle->checkFiles = 1;
+  dTrace("%p total numOfTable:%d in query", pQueryHandle, taosArrayGetSize(pQueryHandle->pTableCheckInfo));
   
+  /*
+   * For ascending timestamp order query, query starts from data files. In contrast, buffer will be checked in the first place
+   * in case of descending timestamp order query.
+   */
+  pQueryHandle->checkFiles  = QUERY_IS_ASC_QUERY(pQueryHandle->order);
   pQueryHandle->activeIndex = 0;
 
-  // malloc buffer in order to load data from file
+  // allocate buffer in order to load data blocks from file
   int32_t numOfCols = taosArrayGetSize(pColumnInfo);
   size_t  bufferCapacity = 4096;
 
@@ -206,10 +193,6 @@ tsdb_query_handle_t* tsdbQueryByTableId(tsdb_repo_t* tsdb, STsdbQueryCond* pCond
     taosArrayPush(pQueryHandle->pColumns, &pDest);
   }
 
-  if (doAllocateBuf(pQueryHandle, bufferCapacity) != TSDB_CODE_SUCCESS) {
-    return NULL;
-  }
-
   tsdbInitDataBlockLoadInfo(&pQueryHandle->dataBlockLoadInfo);
   tsdbInitCompBlockLoadInfo(&pQueryHandle->compBlockLoadInfo);
 
@@ -239,7 +222,12 @@ static bool hasMoreDataInCache(STsdbQueryHandle* pHandle) {
 
 // todo dynamic get the daysperfile
 static int32_t getFileIdFromKey(TSKEY key) {
-  return (int32_t)(key / 10);  // set the starting fileId
+  int64_t fid = (int64_t)(key / 10);  // set the starting fileId
+  if (fid > INT32_MAX) {
+    fid = INT32_MAX;
+  }
+  
+  return fid;
 }
 
 static int32_t binarySearchForBlockImpl(SCompBlock* pBlock, int32_t numOfBlocks, TSKEY skey, int32_t order);
@@ -247,8 +235,12 @@ static int32_t binarySearchForBlockImpl(SCompBlock* pBlock, int32_t numOfBlocks,
 static int32_t getFileCompInfo(STsdbQueryHandle* pQueryHandle, int32_t* numOfBlocks, int32_t type) {
   // todo check open file failed
   SFileGroup* fileGroup = pQueryHandle->pFileGroup;
+  
+  assert(fileGroup->files[TSDB_FILE_TYPE_HEAD].fname > 0);
   if (fileGroup->files[TSDB_FILE_TYPE_HEAD].fd == FD_INITIALIZER) {
     fileGroup->files[TSDB_FILE_TYPE_HEAD].fd = open(fileGroup->files[TSDB_FILE_TYPE_HEAD].fname, O_RDONLY);
+  } else {
+    assert(FD_VALID(fileGroup->files[TSDB_FILE_TYPE_HEAD].fd));
   }
 
   // load all the comp offset value for all tables in this file
@@ -262,7 +254,7 @@ static int32_t getFileCompInfo(STsdbQueryHandle* pQueryHandle, int32_t* numOfBlo
 
     SCompIdx* compIndex = &pQueryHandle->compIndex[pCheckInfo->tableId.tid];
     if (compIndex->len == 0 || compIndex->numOfSuperBlocks == 0) {  // no data block in this file, try next file
-
+      assert(0);
     } else {
       if (pCheckInfo->compSize < compIndex->len) {
         assert(compIndex->len > 0);
@@ -271,46 +263,35 @@ static int32_t getFileCompInfo(STsdbQueryHandle* pQueryHandle, int32_t* numOfBlo
         assert(t != NULL);
         
         pCheckInfo->pCompInfo = (SCompInfo*) t;
+        pCheckInfo->compSize = compIndex->len;
       }
       
       tsdbLoadCompBlocks(fileGroup, compIndex, pCheckInfo->pCompInfo);
   
-      int32_t index = 0;
+      SCompInfo* pCompInfo = pCheckInfo->pCompInfo;
+      
+      TSKEY s = MIN(pCheckInfo->lastKey, pQueryHandle->window.ekey);
+      TSKEY e = MAX(pCheckInfo->lastKey, pQueryHandle->window.ekey);
+      
       // discard the unqualified data block based on the query time window
-      int32_t start = binarySearchForBlockImpl(pCheckInfo->pCompInfo->blocks, compIndex->numOfSuperBlocks,
-                                               pQueryHandle->order, pCheckInfo->lastKey);
+      int32_t start = binarySearchForBlockImpl(pCompInfo->blocks, compIndex->numOfSuperBlocks, s, TSDB_ORDER_ASC);
+      int32_t end = start;
       
-      if (type == QUERY_RANGE_GREATER_EQUAL) {
-        if (pCheckInfo->lastKey <= pCheckInfo->pCompInfo->blocks[start].keyLast) {
-          //          break;
-        } else {
-          index = -1;
-        }
-      } else {
-        if (pCheckInfo->lastKey >= pCheckInfo->pCompInfo->blocks[start].keyFirst) {
-          //          break;
-        } else {
-          index = -1;
-        }
-      }
-
-      // not found in data blocks in current file
-      if (index == -1) {
+      if (s > pCompInfo->blocks[start].keyLast) {
         continue;
       }
 
-      // todo speedup the procedure of locating end block
-      int32_t e = start;
-      while (e < compIndex->numOfSuperBlocks &&
-             (pCheckInfo->pCompInfo->blocks[e].keyFirst <= pQueryHandle->window.ekey)) {
-        e += 1;
+      // todo speedup the procedure of located end block
+      while (end < compIndex->numOfSuperBlocks && (pCompInfo->blocks[end].keyFirst <= e)) {
+        end += 1;
       }
 
+      pCheckInfo->numOfBlocks = (end - start);
+      
       if (start > 0) {
-        memmove(pCheckInfo->pCompInfo->blocks, &pCheckInfo->pCompInfo->blocks[start], (e - start) * sizeof(SCompBlock));
+        memmove(pCompInfo->blocks, &pCompInfo->blocks[start], pCheckInfo->numOfBlocks * sizeof(SCompBlock));
       }
 
-      pCheckInfo->numOfBlocks = (e - start);
       (*numOfBlocks) += pCheckInfo->numOfBlocks;
     }
   }
@@ -413,7 +394,6 @@ static bool loadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* pBlock
       }
 
       SDataCols* pDataCols = pCheckInfo->pDataCols;
-
       if (pCheckInfo->lastKey > pBlock->keyFirst) {
         cur->pos =
             binarySearchForKey(pDataCols->cols[0].pData, pBlock->numOfPoints, pCheckInfo->lastKey, pQueryHandle->order);
@@ -425,8 +405,24 @@ static bool loadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* pBlock
     } else {  // the whole block is loaded in to buffer
       pQueryHandle->realNumOfRows = pBlock->numOfPoints;
     }
-  } else {  // todo desc query
+  } else {
+    // query ended in current block
     if (pQueryHandle->window.ekey > pBlock->keyFirst) {
+      if (!doLoadFileDataBlock(pQueryHandle, pBlock, pCheckInfo)) {
+        return false;
+      }
+      
+      SDataCols* pDataCols = pCheckInfo->pDataCols;
+      if (pCheckInfo->lastKey < pBlock->keyLast) {
+        cur->pos =
+            binarySearchForKey(pDataCols->cols[0].pData, pBlock->numOfPoints, pCheckInfo->lastKey, pQueryHandle->order);
+      } else {
+        cur->pos = pBlock->numOfPoints - 1;
+      }
+      
+      filterDataInDataBlock(pQueryHandle, pCheckInfo, pBlock, sa);
+    } else {
+      pQueryHandle->realNumOfRows = pBlock->numOfPoints;
     }
   }
 
@@ -619,7 +615,9 @@ static void filterDataInDataBlock(STsdbQueryHandle* pQueryHandle, STableCheckInf
 }
 
 static SArray* getColumnIdList(STsdbQueryHandle* pQueryHandle) {
-  int32_t numOfCols = QH_GET_NUM_OF_COLS(pQueryHandle);
+  size_t numOfCols = QH_GET_NUM_OF_COLS(pQueryHandle);
+  assert(numOfCols <= TSDB_MAX_COLUMNS);
+  
   SArray* pIdList = taosArrayInit(numOfCols, sizeof(int16_t));
   for (int32_t i = 0; i < numOfCols; ++i) {
     SColumnInfoData* pCol = taosArrayGet(pQueryHandle->pColumns, i);
@@ -924,7 +922,7 @@ static bool getDataBlocksInFiles(STsdbQueryHandle* pQueryHandle) {
 
     int32_t fid = getFileIdFromKey(pQueryHandle->window.skey);
     
-    tsdbInitFileGroupIter(pFileHandle, &pQueryHandle->fileIter, TSDB_FGROUP_ITER_FORWARD);
+    tsdbInitFileGroupIter(pFileHandle, &pQueryHandle->fileIter, pQueryHandle->order);
     tsdbSeekFileGroupIter(&pQueryHandle->fileIter, fid);
 
     int32_t numOfBlocks = -1;
@@ -934,13 +932,14 @@ static bool getDataBlocksInFiles(STsdbQueryHandle* pQueryHandle) {
     int32_t numOfTables = taosArrayGetSize(pQueryHandle->pTableCheckInfo);
 
     while ((pQueryHandle->pFileGroup = tsdbGetFileGroupNext(&pQueryHandle->fileIter)) != NULL) {
-      if (getFileCompInfo(pQueryHandle, &numOfBlocks, 1) != TSDB_CODE_SUCCESS) {
+      int32_t type = QUERY_IS_ASC_QUERY(pQueryHandle->order)? QUERY_RANGE_GREATER_EQUAL:QUERY_RANGE_LESS_EQUAL;
+      if (getFileCompInfo(pQueryHandle, &numOfBlocks, type) != TSDB_CODE_SUCCESS) {
         break;
       }
 
       assert(numOfBlocks >= 0);
       dTrace("%p %d blocks found in file for %d table(s), fid:%d", pQueryHandle, numOfBlocks,
-             pQueryHandle->pFileGroup->fileId, numOfTables);
+             numOfTables, pQueryHandle->pFileGroup->fileId);
       
       // todo return error code to query engine
       if (createDataBlocksInfo(pQueryHandle, numOfBlocks, &pQueryHandle->numOfBlocks) != TSDB_CODE_SUCCESS) {
@@ -961,7 +960,7 @@ static bool getDataBlocksInFiles(STsdbQueryHandle* pQueryHandle) {
       return false;
     }
 
-    cur->slot = 0;
+    cur->slot = QUERY_IS_ASC_QUERY(pQueryHandle->order)? 0:pQueryHandle->numOfBlocks-1;
     cur->fid = pQueryHandle->pFileGroup->fileId;
 
     STableBlockInfo* pBlockInfo = &pQueryHandle->pDataBlockInfo[cur->slot];
@@ -970,8 +969,10 @@ static bool getDataBlocksInFiles(STsdbQueryHandle* pQueryHandle) {
     
     return loadFileDataBlock(pQueryHandle, pBlock, pCheckInfo);
   } else {
-    if (cur->slot == pQueryHandle->numOfBlocks - 1) { // all blocks
+    if ((cur->slot == pQueryHandle->numOfBlocks - 1 && QUERY_IS_ASC_QUERY(pQueryHandle->order)) ||
+        (cur->slot == 0 && !QUERY_IS_ASC_QUERY(pQueryHandle->order))) { // all blocks
       int32_t numOfBlocks = -1;
+      
       int32_t numOfTables = taosArrayGetSize(pQueryHandle->pTableCheckInfo);
       pQueryHandle->numOfBlocks = 0;
       
@@ -1002,66 +1003,83 @@ static bool getDataBlocksInFiles(STsdbQueryHandle* pQueryHandle) {
         return false;
       }
   
-      cur->slot = 0;
+      cur->slot = QUERY_IS_ASC_QUERY(pQueryHandle->order)? 0:pQueryHandle->numOfBlocks-1;
       cur->fid = pQueryHandle->pFileGroup->fileId;
       
       STableBlockInfo* pBlockInfo = &pQueryHandle->pDataBlockInfo[cur->slot];
-
       STableCheckInfo* pCheckInfo = pBlockInfo->pTableCheckInfo;
       SCompBlock*      pBlock = pBlockInfo->pBlock.compBlock;
       
       return loadFileDataBlock(pQueryHandle, pBlock, pCheckInfo);
     } else {  // next block of the same file
-      cur->slot += 1;
-      cur->pos = 0;
+      int32_t step = QUERY_IS_ASC_QUERY(pQueryHandle->order)? 1:-1;
+      cur->slot += step;
       
       STableBlockInfo* pBlockInfo = &pQueryHandle->pDataBlockInfo[cur->slot];
+      if (QUERY_IS_ASC_QUERY(pQueryHandle->order)) {
+        cur->pos = 0;
+      } else {
+        cur->pos = pBlockInfo->pBlock.compBlock->numOfPoints - 1;
+      }
+
       return loadFileDataBlock(pQueryHandle, pBlockInfo->pBlock.compBlock, pBlockInfo->pTableCheckInfo);
     }
   }
 }
 
-// handle data in cache situation
-bool tsdbNextDataBlock(tsdb_query_handle_t* pQueryHandle) {
-  STsdbQueryHandle* pHandle = (STsdbQueryHandle*)pQueryHandle;
 
-  size_t numOfTables = taosArrayGetSize(pHandle->pTableCheckInfo);
+static bool doHasDataInBuffer(STsdbQueryHandle* pQueryHandle) {
+  size_t numOfTables = taosArrayGetSize(pQueryHandle->pTableCheckInfo);
+  // todo add assert, the value of numOfTables should be less than the maximum value for each vnode capacity
   assert(numOfTables > 0);
   
-  if (pHandle->checkFiles) {
-    if (getDataBlocksInFiles(pHandle)) {
+  while (pQueryHandle->activeIndex < numOfTables) {
+    if (hasMoreDataInCache(pQueryHandle)) {
       return true;
     }
     
-    pHandle->activeIndex = 0;
-    pHandle->checkFiles = 0;
-    
-    while (pHandle->activeIndex < numOfTables) {
-      if (hasMoreDataInCache(pHandle)) {
-        return true;
-      }
-      
-      pHandle->activeIndex += 1;
-    }
-    
-    return false;
-  } else {
-    while (pHandle->activeIndex < numOfTables) {
-      if (hasMoreDataInCache(pHandle)) {
-        return true;
-      }
-      
-      pHandle->activeIndex += 1;
+    pQueryHandle->activeIndex += 1;
   }
   
   return false;
 }
+
+// handle data in cache situation
+bool tsdbNextDataBlock(tsdb_query_handle_t* pqHandle) {
+  STsdbQueryHandle* pQueryHandle = (STsdbQueryHandle*) pqHandle;
+  
+  size_t numOfTables = taosArrayGetSize(pQueryHandle->pTableCheckInfo);
+  assert(numOfTables > 0);
+  
+  if (QUERY_IS_ASC_QUERY(pQueryHandle->order)) {
+    if (pQueryHandle->checkFiles) {
+      if (getDataBlocksInFiles(pQueryHandle)) {
+        return true;
+      }
+
+      pQueryHandle->activeIndex = 0;
+      pQueryHandle->checkFiles  = false;
+    }
+    
+    return doHasDataInBuffer(pQueryHandle);
+  } else {  // starts from the buffer in case of descending timestamp order check data blocks
+    if (!pQueryHandle->checkFiles) {
+      if (doHasDataInBuffer(pQueryHandle)) {
+        return true;
+      }
+      
+      pQueryHandle->checkFiles = true;
+    }
+
+    return getDataBlocksInFiles(pQueryHandle);
+  }
+  
 }
 
 static int tsdbReadRowsFromCache(SSkipListIterator* pIter, TSKEY maxKey, int maxRowsToRead, TSKEY* skey, TSKEY* ekey,
-                                 STsdbQueryHandle* pHandle) {
+                                 STsdbQueryHandle* pQueryHandle) {
   int     numOfRows = 0;
-  int32_t numOfCols = taosArrayGetSize(pHandle->pColumns);
+  int32_t numOfCols = taosArrayGetSize(pQueryHandle->pColumns);
   *skey = INT64_MIN;
 
   while (tSkipListIterNext(pIter)) {
@@ -1079,7 +1097,7 @@ static int tsdbReadRowsFromCache(SSkipListIterator* pIter, TSKEY maxKey, int max
 
     int32_t offset = 0;
     for (int32_t i = 0; i < numOfCols; ++i) {
-      SColumnInfoData* pColInfo = taosArrayGet(pHandle->pColumns, i);
+      SColumnInfoData* pColInfo = taosArrayGet(pQueryHandle->pColumns, i);
       memcpy(pColInfo->pData + numOfRows * pColInfo->info.bytes, dataRowTuple(row) + offset, pColInfo->info.bytes);
       offset += pColInfo->info.bytes;
     }