Merge pull request #19178 from taosdata/fix/set_scan_limit

enh: set scan limit

include/libs/executor/executor.h

@@ -213,6 +213,7 @@ int32_t qStreamSourceRecoverStep1(qTaskInfo_t tinfo, int64_t ver);
 int32_t qStreamSourceRecoverStep2(qTaskInfo_t tinfo, int64_t ver);
 int32_t qStreamRecoverFinish(qTaskInfo_t tinfo);
 int32_t qStreamRestoreParam(qTaskInfo_t tinfo);
+bool    qStreamRecoverScanFinished(qTaskInfo_t tinfo);
 
 #ifdef __cplusplus
 }

source/libs/executor/inc/executorimpl.h

@@ -136,6 +136,7 @@ typedef struct {
   SSchemaWrapper*     schema;
   char                tbName[TSDB_TABLE_NAME_LEN];
   int8_t              recoverStep;
+  int8_t              recoverScanFinished;
   SQueryTableDataCond tableCond;
   int64_t             fillHistoryVer1;
   int64_t             fillHistoryVer2;
@@ -182,7 +183,7 @@ struct SExecTaskInfo {
   SSubplan*             pSubplan;
   struct SOperatorInfo* pRoot;
   SLocalFetch           localFetch;
-  SArray*               pResultBlockList;// result block list
+  SArray*               pResultBlockList;  // result block list
   STaskStopInfo         stopInfo;
 };
 
@@ -199,7 +200,7 @@ typedef struct SOperatorFpSet {
   __optr_fn_t         getNextFn;
   __optr_fn_t         cleanupFn;  // call this function to release the allocated resources ASAP
   __optr_close_fn_t   closeFn;
-  __optr_reqBuf_fn_t  reqBufFn;   // total used buffer for blocking operator
+  __optr_reqBuf_fn_t  reqBufFn;  // total used buffer for blocking operator
   __optr_encode_fn_t  encodeResultRow;
   __optr_decode_fn_t  decodeResultRow;
   __optr_explain_fn_t getExplainFn;
@@ -255,22 +256,22 @@ typedef struct SLimitInfo {
 } SLimitInfo;
 
 typedef struct SExchangeInfo {
-  SArray*             pSources;
-  SArray*             pSourceDataInfo;
-  tsem_t              ready;
-  void*               pTransporter;
+  SArray* pSources;
+  SArray* pSourceDataInfo;
+  tsem_t  ready;
+  void*   pTransporter;
 
   // SArray<SSDataBlock*>, result block list, used to keep the multi-block that
   // passed by downstream operator
-  SArray*             pResultBlockList;
-  SArray*             pRecycledBlocks;// build a pool for small data block to avoid to repeatly create and then destroy.
-  SSDataBlock*        pDummyBlock;    // dummy block, not keep data
-  bool                seqLoadData;    // sequential load data or not, false by default
-  int32_t             current;
+  SArray*      pResultBlockList;
+  SArray*      pRecycledBlocks;  // build a pool for small data block to avoid to repeatly create and then destroy.
+  SSDataBlock* pDummyBlock;      // dummy block, not keep data
+  bool         seqLoadData;      // sequential load data or not, false by default
+  int32_t      current;
   SLoadRemoteDataInfo loadInfo;
   uint64_t            self;
   SLimitInfo          limitInfo;
-  int64_t             openedTs;       // start exec time stamp, todo: move to SLoadRemoteDataInfo
+  int64_t             openedTs;  // start exec time stamp, todo: move to SLoadRemoteDataInfo
 } SExchangeInfo;
 
 typedef struct SScanInfo {
@@ -305,9 +306,9 @@ typedef struct {
 } SAggOptrPushDownInfo;
 
 typedef struct STableMetaCacheInfo {
-  SLRUCache*             pTableMetaEntryCache; // 100 by default
-  uint64_t               metaFetch;
-  uint64_t               cacheHit;
+  SLRUCache* pTableMetaEntryCache;  // 100 by default
+  uint64_t   metaFetch;
+  uint64_t   cacheHit;
 } STableMetaCacheInfo;
 
 typedef struct STableScanBase {
@@ -325,46 +326,46 @@ typedef struct STableScanBase {
 } STableScanBase;
 
 typedef struct STableScanInfo {
-  STableScanBase         base;
-  SScanInfo              scanInfo;
-  int32_t                scanTimes;
-  SSDataBlock*           pResBlock;
-  SSampleExecInfo        sample;  // sample execution info
-  int32_t                currentGroupId;
-  int32_t                currentTable;
-  int8_t                 scanMode;
-  int8_t                 assignBlockUid;
-  bool                   hasGroupByTag;
+  STableScanBase  base;
+  SScanInfo       scanInfo;
+  int32_t         scanTimes;
+  SSDataBlock*    pResBlock;
+  SSampleExecInfo sample;  // sample execution info
+  int32_t         currentGroupId;
+  int32_t         currentTable;
+  int8_t          scanMode;
+  int8_t          assignBlockUid;
+  bool            hasGroupByTag;
 } STableScanInfo;
 
 typedef struct STableMergeScanInfo {
-  int32_t                tableStartIndex;
-  int32_t                tableEndIndex;
-  bool                   hasGroupId;
-  uint64_t               groupId;
-  SArray*                queryConds;  // array of queryTableDataCond
-  STableScanBase         base;
-  int32_t                bufPageSize;
-  uint32_t               sortBufSize;  // max buffer size for in-memory sort
-  SArray*                pSortInfo;
-  SSortHandle*           pSortHandle;
-  SSDataBlock*           pSortInputBlock;
-  int64_t                startTs;  // sort start time
-  SArray*                sortSourceParams;
-  SLimitInfo             limitInfo;
-  int64_t                numOfRows;
-  SScanInfo              scanInfo;
-  SSDataBlock*           pResBlock;
-  SSampleExecInfo        sample;  // sample execution info
-  SSortExecInfo          sortExecInfo;
+  int32_t         tableStartIndex;
+  int32_t         tableEndIndex;
+  bool            hasGroupId;
+  uint64_t        groupId;
+  SArray*         queryConds;  // array of queryTableDataCond
+  STableScanBase  base;
+  int32_t         bufPageSize;
+  uint32_t        sortBufSize;  // max buffer size for in-memory sort
+  SArray*         pSortInfo;
+  SSortHandle*    pSortHandle;
+  SSDataBlock*    pSortInputBlock;
+  int64_t         startTs;  // sort start time
+  SArray*         sortSourceParams;
+  SLimitInfo      limitInfo;
+  int64_t         numOfRows;
+  SScanInfo       scanInfo;
+  SSDataBlock*    pResBlock;
+  SSampleExecInfo sample;  // sample execution info
+  SSortExecInfo   sortExecInfo;
 } STableMergeScanInfo;
 
 typedef struct STagScanInfo {
-  SColumnInfo*    pCols;
-  SSDataBlock*    pRes;
-  SColMatchInfo   matchInfo;
-  int32_t         curPos;
-  SReadHandle     readHandle;
+  SColumnInfo*  pCols;
+  SSDataBlock*  pRes;
+  SColMatchInfo matchInfo;
+  int32_t       curPos;
+  SReadHandle   readHandle;
 } STagScanInfo;
 
 typedef enum EStreamScanMode {
@@ -468,6 +469,11 @@ typedef struct SStreamScanInfo {
   SNodeList* pGroupTags;
   SNode*     pTagCond;
   SNode*     pTagIndexCond;
+
+  // recover
+  int32_t blockRecoverContiCnt;
+  int32_t blockRecoverTotCnt;
+
 } SStreamScanInfo;
 
 typedef struct {
@@ -499,8 +505,8 @@ typedef struct STableCountScanOperatorInfo {
 
   STableCountScanSupp supp;
 
-  int32_t   currGrpIdx;
-  SArray*   stbUidList; // when group by db_name and/or stable_name
+  int32_t currGrpIdx;
+  SArray* stbUidList;  // when group by db_name and/or stable_name
 } STableCountScanOperatorInfo;
 
 typedef struct SOptrBasicInfo {
@@ -678,19 +684,19 @@ void           setOperatorInfo(SOperatorInfo* pOperator, const char* name, int32
 void           destroyOperatorInfo(SOperatorInfo* pOperator);
 int32_t        optrDefaultBufFn(SOperatorInfo* pOperator);
 
-void    initBasicInfo(SOptrBasicInfo* pInfo, SSDataBlock* pBlock);
-void    cleanupBasicInfo(SOptrBasicInfo* pInfo);
+void initBasicInfo(SOptrBasicInfo* pInfo, SSDataBlock* pBlock);
+void cleanupBasicInfo(SOptrBasicInfo* pInfo);
 
 int32_t initExprSupp(SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfExpr);
 void    cleanupExprSupp(SExprSupp* pSup);
 
-void    destroyExprInfo(SExprInfo* pExpr, int32_t numOfExprs);
+void destroyExprInfo(SExprInfo* pExpr, int32_t numOfExprs);
 
 int32_t initAggSup(SExprSupp* pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, size_t keyBufSize,
                    const char* pkey);
 void    cleanupAggSup(SAggSupporter* pAggSup);
 
-void    initResultSizeInfo(SResultInfo* pResultInfo, int32_t numOfRows);
+void initResultSizeInfo(SResultInfo* pResultInfo, int32_t numOfRows);
 
 void doBuildStreamResBlock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
                            SDiskbasedBuf* pBuf);
@@ -803,10 +809,10 @@ void setInputDataBlock(SExprSupp* pExprSupp, SSDataBlock* pBlock, int32_t order,
 
 int32_t checkForQueryBuf(size_t numOfTables);
 
-bool    isTaskKilled(SExecTaskInfo* pTaskInfo);
-void    setTaskKilled(SExecTaskInfo* pTaskInfo, int32_t rspCode);
-void    doDestroyTask(SExecTaskInfo* pTaskInfo);
-void    setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status);
+bool isTaskKilled(SExecTaskInfo* pTaskInfo);
+void setTaskKilled(SExecTaskInfo* pTaskInfo, int32_t rspCode);
+void doDestroyTask(SExecTaskInfo* pTaskInfo);
+void setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status);
 
 int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId,
                                char* sql, EOPTR_EXEC_MODEL model);
@@ -828,8 +834,8 @@ bool isDeletedWindow(STimeWindow* pWin, uint64_t groupId, SAggSupporter* pSup);
 bool isDeletedStreamWindow(STimeWindow* pWin, uint64_t groupId, SStreamState* pState, STimeWindowAggSupp* pTwSup);
 void appendOneRowToStreamSpecialBlock(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t* pUid,
                                       uint64_t* pGp, void* pTbName);
-uint64_t    calGroupIdByData(SPartitionBySupporter* pParSup, SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t rowId);
-void        calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock);
+uint64_t calGroupIdByData(SPartitionBySupporter* pParSup, SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t rowId);
+void     calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock);
 
 int32_t finalizeResultRows(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition, SExprSupp* pSup,
                            SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);

source/libs/executor/src/executor.c

@@ -936,6 +936,10 @@ int32_t qStreamRestoreParam(qTaskInfo_t tinfo) {
   }
   return 0;
 }
+bool qStreamRecoverScanFinished(qTaskInfo_t tinfo) {
+  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
+  return pTaskInfo->streamInfo.recoverScanFinished;
+}
 
 void* qExtractReaderFromStreamScanner(void* scanner) {
   SStreamScanInfo* pInfo = scanner;

source/libs/executor/src/scanoperator.c

@@ -744,8 +744,8 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
       tableListGetGroupList(pTaskInfo->pTableInfoList, pInfo->currentGroupId, &pList, &num);
       ASSERT(pInfo->base.dataReader == NULL);
 
-      int32_t code = tsdbReaderOpen(pInfo->base.readHandle.vnode, &pInfo->base.cond, pList, num,
-                                    pInfo->pResBlock, (STsdbReader**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo));
+      int32_t code = tsdbReaderOpen(pInfo->base.readHandle.vnode, &pInfo->base.cond, pList, num, pInfo->pResBlock,
+                                    (STsdbReader**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo));
       if (code != TSDB_CODE_SUCCESS) {
         T_LONG_JMP(pTaskInfo->env, code);
       }
@@ -962,8 +962,8 @@ static SSDataBlock* readPreVersionData(SOperatorInfo* pTableScanOp, uint64_t tbU
 
   SSDataBlock* pBlock = pTableScanInfo->pResBlock;
   STsdbReader* pReader = NULL;
-  int32_t      code = tsdbReaderOpen(pTableScanInfo->base.readHandle.vnode, &cond, &tblInfo, 1, pBlock, (STsdbReader**)&pReader,
-                                     GET_TASKID(pTaskInfo));
+  int32_t      code = tsdbReaderOpen(pTableScanInfo->base.readHandle.vnode, &cond, &tblInfo, 1, pBlock,
+                                     (STsdbReader**)&pReader, GET_TASKID(pTaskInfo));
   if (code != TSDB_CODE_SUCCESS) {
     terrno = code;
     T_LONG_JMP(pTaskInfo->env, code);
@@ -971,7 +971,7 @@ static SSDataBlock* readPreVersionData(SOperatorInfo* pTableScanOp, uint64_t tbU
   }
 
   if (tsdbNextDataBlock(pReader)) {
-    /*SSDataBlock* p = */tsdbRetrieveDataBlock(pReader, NULL);
+    /*SSDataBlock* p = */ tsdbRetrieveDataBlock(pReader, NULL);
     doSetTagColumnData(&pTableScanInfo->base, pBlock, pTaskInfo, pBlock->info.rows);
     pBlock->info.id.groupId = getTableGroupId(pTaskInfo->pTableInfoList, pBlock->info.id.uid);
   }
@@ -1200,7 +1200,7 @@ static int32_t generateIntervalScanRange(SStreamScanInfo* pInfo, SSDataBlock* pS
   SColumnInfoData* pSrcUidCol = taosArrayGet(pSrcBlock->pDataBlock, UID_COLUMN_INDEX);
   SColumnInfoData* pSrcGpCol = taosArrayGet(pSrcBlock->pDataBlock, GROUPID_COLUMN_INDEX);
 
-  uint64_t*        srcUidData = (uint64_t*)pSrcUidCol->pData;
+  uint64_t* srcUidData = (uint64_t*)pSrcUidCol->pData;
   ASSERT(pSrcStartTsCol->info.type == TSDB_DATA_TYPE_TIMESTAMP);
   TSKEY*  srcStartTsCol = (TSKEY*)pSrcStartTsCol->pData;
   TSKEY*  srcEndTsCol = (TSKEY*)pSrcEndTsCol->pData;
@@ -1729,11 +1729,18 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
     pTSInfo->scanTimes = 0;
     pTSInfo->currentGroupId = -1;
     pTaskInfo->streamInfo.recoverStep = STREAM_RECOVER_STEP__SCAN;
+    pTaskInfo->streamInfo.recoverScanFinished = false;
   }
 
   if (pTaskInfo->streamInfo.recoverStep == STREAM_RECOVER_STEP__SCAN) {
+    if (pInfo->blockRecoverContiCnt > 100) {
+      pInfo->blockRecoverTotCnt += pInfo->blockRecoverContiCnt;
+      pInfo->blockRecoverContiCnt = 0;
+      return NULL;
+    }
     SSDataBlock* pBlock = doTableScan(pInfo->pTableScanOp);
     if (pBlock != NULL) {
+      pInfo->blockRecoverContiCnt++;
       calBlockTbName(pInfo, pBlock);
       if (pInfo->pUpdateInfo) {
         TSKEY maxTs = updateInfoFillBlockData(pInfo->pUpdateInfo, pBlock, pInfo->primaryTsIndex);
@@ -1751,6 +1758,7 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
     pTSInfo->base.cond.startVersion = -1;
     pTSInfo->base.cond.endVersion = -1;
 
+    pTaskInfo->streamInfo.recoverScanFinished = true;
     return NULL;
   }
 
@@ -2262,7 +2270,8 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
     if (pHandle->initTableReader) {
       pTSInfo->scanMode = TABLE_SCAN__TABLE_ORDER;
       pTSInfo->base.dataReader = NULL;
-      code = tsdbReaderOpen(pHandle->vnode, &pTSInfo->base.cond, pList, num, pTSInfo->pResBlock, &pTSInfo->base.dataReader, NULL);
+      code = tsdbReaderOpen(pHandle->vnode, &pTSInfo->base.cond, pList, num, pTSInfo->pResBlock,
+                            &pTSInfo->base.dataReader, NULL);
       if (code != 0) {
         terrno = code;
         destroyTableScanOperatorInfo(pTableScanOp);
@@ -2333,7 +2342,8 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
   pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pRes->pDataBlock);
 
   __optr_fn_t nextFn = pTaskInfo->execModel == OPTR_EXEC_MODEL_STREAM ? doStreamScan : doQueueScan;
-  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, nextFn, NULL, destroyStreamScanOperatorInfo, optrDefaultBufFn, NULL);
+  pOperator->fpSet =
+      createOperatorFpSet(optrDummyOpenFn, nextFn, NULL, destroyStreamScanOperatorInfo, optrDefaultBufFn, NULL);
 
   return pOperator;
 
@@ -2470,7 +2480,8 @@ SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysi
   initResultSizeInfo(&pOperator->resultInfo, 4096);
   blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);
 
-  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTagScan, NULL, destroyTagScanOperatorInfo, optrDefaultBufFn, NULL);
+  pOperator->fpSet =
+      createOperatorFpSet(optrDummyOpenFn, doTagScan, NULL, destroyTagScanOperatorInfo, optrDefaultBufFn, NULL);
 
   return pOperator;
 
@@ -2491,11 +2502,12 @@ static SSDataBlock* getTableDataBlockImpl(void* param) {
 
   SQueryTableDataCond* pQueryCond = taosArrayGet(pInfo->queryConds, readIdx);
 
-  int64_t st = taosGetTimestampUs();
-  void*   p = tableListGetInfo(pTaskInfo->pTableInfoList, readIdx + pInfo->tableStartIndex);
+  int64_t      st = taosGetTimestampUs();
+  void*        p = tableListGetInfo(pTaskInfo->pTableInfoList, readIdx + pInfo->tableStartIndex);
   SReadHandle* pHandle = &pInfo->base.readHandle;
 
-  int32_t code = tsdbReaderOpen(pHandle->vnode, pQueryCond, p, 1, pBlock, &pInfo->base.dataReader, GET_TASKID(pTaskInfo));
+  int32_t code =
+      tsdbReaderOpen(pHandle->vnode, pQueryCond, p, 1, pBlock, &pInfo->base.dataReader, GET_TASKID(pTaskInfo));
   if (code != 0) {
     T_LONG_JMP(pTaskInfo->env, code);
   }
@@ -2893,8 +2905,8 @@ static void         buildVnodeGroupedNtbTableCount(STableCountScanOperatorInfo*
                                                    SSDataBlock* pRes, char* dbName);
 static void         buildVnodeFilteredTbCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,
                                               STableCountScanSupp* pSupp, SSDataBlock* pRes, char* dbName);
-static void buildVnodeGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,
-                                        STableCountScanSupp* pSupp, SSDataBlock* pRes, int32_t vgId, char* dbName);
+static void         buildVnodeGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,
+                                                STableCountScanSupp* pSupp, SSDataBlock* pRes, int32_t vgId, char* dbName);
 static SSDataBlock* buildVnodeDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,
                                            STableCountScanSupp* pSupp, SSDataBlock* pRes);
 static void         buildSysDbGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,
@@ -3019,8 +3031,8 @@ SOperatorInfo* createTableCountScanOperatorInfo(SReadHandle* readHandle, STableC
 
   setOperatorInfo(pOperator, "TableCountScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN, false, OP_NOT_OPENED,
                   pInfo, pTaskInfo);
-  pOperator->fpSet =
-      createOperatorFpSet(optrDummyOpenFn, doTableCountScan, NULL, destoryTableCountScanOperator, optrDefaultBufFn, NULL);
+  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTableCountScan, NULL, destoryTableCountScanOperator,
+                                         optrDefaultBufFn, NULL);
   return pOperator;
 
 _error:

source/libs/stream/src/streamExec.c

@@ -116,7 +116,11 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) {
         ASSERT(0);
       }
       if (output == NULL) {
-        finished = true;
+        if (qStreamRecoverScanFinished(exec)) {
+          finished = true;
+        } else {
+          qSetStreamOpOpen(exec);
+        }
         break;
       }