Merge pull request #13095 from taosdata/szhou/feature/sma

enhance: table scan operator can accept more than one time range

include/common/tcommon.h

@@ -105,12 +105,14 @@ typedef struct SColumnInfoData {
 } SColumnInfoData;
 
 typedef struct SQueryTableDataCond {
-  STimeWindow  twindow;
+  //STimeWindow  twindow;
   int32_t      order;  // desc|asc order to iterate the data block
   int32_t      numOfCols;
   SColumnInfo *colList;
   bool         loadExternalRows;  // load external rows or not
   int32_t      type;              // data block load type:
+  int32_t      numOfTWindows;
+  STimeWindow *twindows;
 } SQueryTableDataCond;
 
 void*   blockDataDestroy(SSDataBlock* pBlock);

source/dnode/vnode/inc/vnode.h

@@ -111,7 +111,7 @@ bool         tsdbNextDataBlock(tsdbReaderT pTsdbReadHandle);
 void         tsdbRetrieveDataBlockInfo(tsdbReaderT *pTsdbReadHandle, SDataBlockInfo *pBlockInfo);
 int32_t tsdbRetrieveDataBlockStatisInfo(tsdbReaderT *pTsdbReadHandle, SColumnDataAgg ***pBlockStatis, bool *allHave);
 SArray *tsdbRetrieveDataBlock(tsdbReaderT *pTsdbReadHandle, SArray *pColumnIdList);
-void    tsdbResetReadHandle(tsdbReaderT queryHandle, SQueryTableDataCond *pCond);
+void    tsdbResetReadHandle(tsdbReaderT queryHandle, SQueryTableDataCond *pCond, int32_t tWinIdx);
 void    tsdbCleanupReadHandle(tsdbReaderT queryHandle);
 
 // tq

source/dnode/vnode/src/tsdb/tsdbRead.c

@@ -317,28 +317,28 @@ static int64_t getEarliestValidTimestamp(STsdb* pTsdb) {
   return now - (tsTickPerMin[pCfg->precision] * pCfg->keep2) + 1;  // needs to add one tick
 }
 
-static void setQueryTimewindow(STsdbReadHandle* pTsdbReadHandle, SQueryTableDataCond* pCond) {
-  pTsdbReadHandle->window = pCond->twindow;
+static void setQueryTimewindow(STsdbReadHandle* pTsdbReadHandle, SQueryTableDataCond* pCond, int32_t tWinIdx) {
+  pTsdbReadHandle->window = pCond->twindows[tWinIdx];
 
   bool    updateTs = false;
   int64_t startTs = getEarliestValidTimestamp(pTsdbReadHandle->pTsdb);
   if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
     if (startTs > pTsdbReadHandle->window.skey) {
       pTsdbReadHandle->window.skey = startTs;
-      pCond->twindow.skey = startTs;
+      pCond->twindows[tWinIdx].skey = startTs;
       updateTs = true;
     }
   } else {
     if (startTs > pTsdbReadHandle->window.ekey) {
       pTsdbReadHandle->window.ekey = startTs;
-      pCond->twindow.ekey = startTs;
+      pCond->twindows[tWinIdx].ekey = startTs;
       updateTs = true;
     }
   }
 
   if (updateTs) {
     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, pCond->twindows[tWinIdx].skey, pCond->twindows[tWinIdx].ekey, pTsdbReadHandle->window.skey,
               pTsdbReadHandle->window.ekey, pTsdbReadHandle->idStr);
   }
 }
@@ -382,7 +382,7 @@ static STsdbReadHandle* tsdbQueryTablesImpl(SVnode* pVnode, SQueryTableDataCond*
     goto _end;
   }
 
-  STsdb* pTsdb = getTsdbByRetentions(pVnode, pReadHandle, pCond->twindow.skey, pVnode->config.tsdbCfg.retentions);
+  STsdb* pTsdb = getTsdbByRetentions(pVnode, pReadHandle, pCond->twindows[0].skey, pVnode->config.tsdbCfg.retentions);
 
   pReadHandle->order = pCond->order;
   pReadHandle->pTsdb = pTsdb;
@@ -408,7 +408,7 @@ static STsdbReadHandle* tsdbQueryTablesImpl(SVnode* pVnode, SQueryTableDataCond*
   }
 
   assert(pCond != NULL);
-  setQueryTimewindow(pReadHandle, pCond);
+  setQueryTimewindow(pReadHandle, pCond, 0);
 
   if (pCond->numOfCols > 0) {
     int32_t rowLen = 0;
@@ -520,7 +520,7 @@ tsdbReaderT* tsdbQueryTables(SVnode* pVnode, SQueryTableDataCond* pCond, STableL
   return (tsdbReaderT)pTsdbReadHandle;
 }
 
-void tsdbResetReadHandle(tsdbReaderT queryHandle, SQueryTableDataCond* pCond) {
+void tsdbResetReadHandle(tsdbReaderT queryHandle, SQueryTableDataCond* pCond, int32_t tWinIdx) {
   STsdbReadHandle* pTsdbReadHandle = queryHandle;
 
   if (emptyQueryTimewindow(pTsdbReadHandle)) {
@@ -533,7 +533,7 @@ void tsdbResetReadHandle(tsdbReaderT queryHandle, SQueryTableDataCond* pCond) {
   }
 
   pTsdbReadHandle->order = pCond->order;
-  pTsdbReadHandle->window = pCond->twindow;
+  setQueryTimewindow(pTsdbReadHandle, pCond, tWinIdx);
   pTsdbReadHandle->type = TSDB_QUERY_TYPE_ALL;
   pTsdbReadHandle->cur.fid = -1;
   pTsdbReadHandle->cur.win = TSWINDOW_INITIALIZER;
@@ -558,11 +558,11 @@ void tsdbResetReadHandle(tsdbReaderT queryHandle, SQueryTableDataCond* pCond) {
   resetCheckInfo(pTsdbReadHandle);
 }
 
-void tsdbResetQueryHandleForNewTable(tsdbReaderT queryHandle, SQueryTableDataCond* pCond, STableListInfo* tableList) {
+void tsdbResetQueryHandleForNewTable(tsdbReaderT queryHandle, SQueryTableDataCond* pCond, STableListInfo* tableList, int32_t tWinIdx) {
   STsdbReadHandle* pTsdbReadHandle = queryHandle;
 
   pTsdbReadHandle->order = pCond->order;
-  pTsdbReadHandle->window = pCond->twindow;
+  pTsdbReadHandle->window = pCond->twindows[tWinIdx];
   pTsdbReadHandle->type = TSDB_QUERY_TYPE_ALL;
   pTsdbReadHandle->cur.fid = -1;
   pTsdbReadHandle->cur.win = TSWINDOW_INITIALIZER;
@@ -602,7 +602,7 @@ void tsdbResetQueryHandleForNewTable(tsdbReaderT queryHandle, SQueryTableDataCon
 
 tsdbReaderT tsdbQueryLastRow(SVnode* pVnode, SQueryTableDataCond* pCond, STableListInfo* pList, uint64_t qId,
                              uint64_t taskId) {
-  pCond->twindow = updateLastrowForEachGroup(pList);
+  pCond->twindows[0] = updateLastrowForEachGroup(pList);
 
   // no qualified table
   if (taosArrayGetSize(pList->pTableList) == 0) {
@@ -620,7 +620,7 @@ tsdbReaderT tsdbQueryLastRow(SVnode* pVnode, SQueryTableDataCond* pCond, STableL
     return NULL;
   }
 
-  assert(pCond->order == TSDB_ORDER_ASC && pCond->twindow.skey <= pCond->twindow.ekey);
+  assert(pCond->order == TSDB_ORDER_ASC && pCond->twindows[0].skey <= pCond->twindows[0].ekey);
   if (pTsdbReadHandle->cachelastrow) {
     pTsdbReadHandle->type = TSDB_QUERY_TYPE_LAST;
   }

source/libs/executor/inc/executorimpl.h

@@ -334,6 +334,8 @@ typedef struct STableScanInfo {
   int32_t         dataBlockLoadFlag;
   double          sampleRatio;  // data block sample ratio, 1 by default
   SInterval       interval;     // if the upstream is an interval operator, the interval info is also kept here to get the time window to check if current data block needs to be loaded.
+
+  int32_t         curTWinIdx;
 } STableScanInfo;
 
 typedef struct STagScanInfo {
@@ -803,6 +805,8 @@ SResultWindowInfo* getSessionTimeWindow(SArray* pWinInfos, TSKEY ts, int64_t gap
 int32_t updateSessionWindowInfo(SResultWindowInfo* pWinInfo, TSKEY* pTs, int32_t rows,
     int32_t start, int64_t gap, SHashObj* pStDeleted);
 bool functionNeedToExecute(SqlFunctionCtx* pCtx);
+
+int32_t compareTimeWindow(const void* p1, const void* p2, const void* param);
 #ifdef __cplusplus
 }
 #endif

source/libs/executor/src/executorimpl.c

@@ -4706,6 +4706,18 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
   return pOptr;
 }
 
+int32_t compareTimeWindow(const void* p1, const void* p2, const void* param) {
+  const SQueryTableDataCond *pCond = param;
+  const STimeWindow *pWin1 = p1;
+  const STimeWindow *pWin2 = p2;
+  if (pCond->order == TSDB_ORDER_ASC) {
+    return pWin1->skey - pWin2->skey;
+  } else if (pCond->order == TSDB_ORDER_DESC) {
+    return pWin2->skey - pWin1->skey;
+  }
+  return 0;
+}
+
 int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode) {
   pCond->loadExternalRows = false;
 
@@ -4717,16 +4729,34 @@ int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysi
     return terrno;
   }
 
-  pCond->twindow = pTableScanNode->scanRange;
+  //pCond->twindow = pTableScanNode->scanRange;
+  //TODO: get it from stable scan node
+  pCond->numOfTWindows = 1;
+  pCond->twindows = taosMemoryCalloc(pCond->numOfTWindows, sizeof(STimeWindow));
+  pCond->twindows[0] = pTableScanNode->scanRange;
 
 #if 1
   // todo work around a problem, remove it later
-  if ((pCond->order == TSDB_ORDER_ASC && pCond->twindow.skey > pCond->twindow.ekey) ||
-      (pCond->order == TSDB_ORDER_DESC && pCond->twindow.skey < pCond->twindow.ekey)) {
-    TSWAP(pCond->twindow.skey, pCond->twindow.ekey);
+  for (int32_t i = 0; i < pCond->numOfTWindows; ++i) {
+    if ((pCond->order == TSDB_ORDER_ASC && pCond->twindows[i].skey > pCond->twindows[i].ekey) ||
+        (pCond->order == TSDB_ORDER_DESC && pCond->twindows[i].skey < pCond->twindows[i].ekey)) {
+      TSWAP(pCond->twindows[i].skey, pCond->twindows[i].ekey);
+    }
   }
 #endif
 
+  for (int32_t i = 0; i < pCond->numOfTWindows; ++i) {
+    if ((pCond->order == TSDB_ORDER_ASC && pCond->twindows[i].skey > pCond->twindows[i].ekey) ||
+        (pCond->order == TSDB_ORDER_DESC && pCond->twindows[i].skey < pCond->twindows[i].ekey)) {
+      TSWAP(pCond->twindows[i].skey, pCond->twindows[i].ekey);
+    }
+  }
+  taosqsort(pCond->twindows,
+            pCond->numOfTWindows,
+            sizeof(STimeWindow),
+            pCond,
+            compareTimeWindow);
+
   pCond->type = BLOCK_LOAD_OFFSET_SEQ_ORDER;
   //  pCond->type = pTableScanNode->scanFlag;
 

source/libs/executor/src/scanoperator.c

@@ -274,9 +274,17 @@ static void prepareForDescendingScan(STableScanInfo* pTableScanInfo, SqlFunction
   switchCtxOrder(pCtx, numOfOutput);
   //  setupQueryRangeForReverseScan(pTableScanInfo);
 
-  STimeWindow* pTWindow = &pTableScanInfo->cond.twindow;
-  TSWAP(pTWindow->skey, pTWindow->ekey);
   pTableScanInfo->cond.order = TSDB_ORDER_DESC;
+  for (int32_t i = 0; i < pTableScanInfo->cond.numOfTWindows; ++i) {
+    STimeWindow* pTWindow = &pTableScanInfo->cond.twindows[i];
+    TSWAP(pTWindow->skey, pTWindow->ekey);
+  }
+  SQueryTableDataCond *pCond = &pTableScanInfo->cond;
+  taosqsort(pCond->twindows,
+            pCond->numOfTWindows,
+            sizeof(STimeWindow),
+            pCond,
+            compareTimeWindow);
 }
 
 void addTagPseudoColumnData(STableScanInfo* pTableScanInfo, SSDataBlock* pBlock) {
@@ -380,7 +388,6 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
     pOperator->cost.totalCost = pTableScanInfo->readRecorder.elapsedTime;
     return pBlock;
   }
-
   return NULL;
 }
 
@@ -395,24 +402,30 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
 
   // do the ascending order traverse in the first place.
   while (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) {
+    while (pTableScanInfo->curTWinIdx < pTableScanInfo->cond.numOfTWindows) {
       SSDataBlock* p = doTableScanImpl(pOperator);
       if (p != NULL) {
         return p;
       }
+      pTableScanInfo->curTWinIdx += 1;
+      if (pTableScanInfo->curTWinIdx < pTableScanInfo->cond.numOfTWindows) {
+        tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, pTableScanInfo->curTWinIdx);
+      }
+    }
 
     pTableScanInfo->scanTimes += 1;
 
     if (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) {
       setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);
       pTableScanInfo->scanFlag = REPEAT_SCAN;
-
-      STimeWindow* pWin = &pTableScanInfo->cond.twindow;
-      qDebug("%s start to repeat ascending order scan data blocks due to query func required, qrange:%" PRId64
-             "-%" PRId64,
-             GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
-
+      qDebug("%s start to repeat ascending order scan data blocks due to query func required", GET_TASKID(pTaskInfo));
+      for (int32_t i = 0; i < pTableScanInfo->cond.numOfTWindows; ++i) {
+        STimeWindow* pWin = &pTableScanInfo->cond.twindows[i];
+        qDebug("%s\t qrange:%" PRId64 "-%" PRId64, GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
+      }
       // do prepare for the next round table scan operation
-      tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond);
+      tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
+      pTableScanInfo->curTWinIdx = 0;
     }
   }
 
@@ -420,31 +433,40 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
   if (pTableScanInfo->scanTimes < total) {
     if (pTableScanInfo->cond.order == TSDB_ORDER_ASC) {
       prepareForDescendingScan(pTableScanInfo, pTableScanInfo->pCtx, pTableScanInfo->numOfOutput);
-      tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond);
+      tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
+      pTableScanInfo->curTWinIdx = 0;
     }
 
-    STimeWindow* pWin = &pTableScanInfo->cond.twindow;
-    qDebug("%s start to descending order scan data blocks due to query func required, qrange:%" PRId64 "-%" PRId64,
-           GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
-
+    qDebug("%s start to descending order scan data blocks due to query func required", GET_TASKID(pTaskInfo));
+    for (int32_t i = 0; i < pTableScanInfo->cond.numOfTWindows; ++i) {
+      STimeWindow* pWin = &pTableScanInfo->cond.twindows[i];
+      qDebug("%s\t qrange:%" PRId64 "-%" PRId64, GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
+    }
     while (pTableScanInfo->scanTimes < total) {
+      while (pTableScanInfo->curTWinIdx < pTableScanInfo->cond.numOfTWindows) {
         SSDataBlock* p = doTableScanImpl(pOperator);
         if (p != NULL) {
           return p;
         }
+        pTableScanInfo->curTWinIdx += 1;
+        if (pTableScanInfo->curTWinIdx < pTableScanInfo->cond.numOfTWindows) {
+          tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, pTableScanInfo->curTWinIdx);
+        }
+      }
 
       pTableScanInfo->scanTimes += 1;
 
-      if (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) {
+      if (pTableScanInfo->scanTimes < total) {
         setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);
         pTableScanInfo->scanFlag = REPEAT_SCAN;
 
-        qDebug("%s start to repeat descending order scan data blocks due to query func required, qrange:%" PRId64
-               "-%" PRId64,
-               GET_TASKID(pTaskInfo), pTaskInfo->window.skey, pTaskInfo->window.ekey);
-
-        // do prepare for the next round table scan operation
-        tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond);
+        qDebug("%s start to repeat descending order scan data blocks due to query func required", GET_TASKID(pTaskInfo));
+        for (int32_t i = 0; i < pTableScanInfo->cond.numOfTWindows; ++i) {
+          STimeWindow* pWin = &pTableScanInfo->cond.twindows[i];
+          qDebug("%s\t qrange:%" PRId64 "-%" PRId64, GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
+        }
+        tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
+        pTableScanInfo->curTWinIdx = 0;
       }
     }
   }
@@ -524,6 +546,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
   pInfo->dataReader = pDataReader;
   pInfo->scanFlag = MAIN_SCAN;
   pInfo->pColMatchInfo = pColList;
+  pInfo->curTWinIdx = 0;
 
   pOperator->name = "TableScanOperator";  // for debug purpose
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN;
@@ -678,8 +701,9 @@ static bool prepareDataScan(SStreamBlockScanInfo* pInfo) {
                                                         binarySearchForKey, NULL, TSDB_ORDER_ASC);
     }
     STableScanInfo* pTableScanInfo = pInfo->pOperatorDumy->info;
-    pTableScanInfo->cond.twindow = win;
-    tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond);
+    pTableScanInfo->cond.twindows[0] = win;
+    pTableScanInfo->curTWinIdx = 0;
+    tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
     pTableScanInfo->scanTimes = 0;
     return true;
   } else {