From b2dd2935683d025d20114c1aba6c5562ee728081 Mon Sep 17 00:00:00 2001
From: 54liuyao <54liuyao@163.com>
Date: Fri, 29 Apr 2022 13:42:55 +0800
Subject: [PATCH] stream supports filter

---
 source/libs/executor/inc/executorimpl.h |  3 ++-
 source/libs/executor/src/executorimpl.c |  2 +-
 source/libs/executor/src/scanoperator.c | 10 +++++++---
 3 files changed, 10 insertions(+), 5 deletions(-)

diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h
index c7fdb61d16..ee8cdc6b1b 100644
--- a/source/libs/executor/inc/executorimpl.h
+++ b/source/libs/executor/inc/executorimpl.h
@@ -375,6 +375,7 @@ typedef struct SStreamBlockScanInfo {
   uint64_t     numOfExec;        // execution times
   void*        readerHandle;     // stream block reader handle
   SArray*      pColMatchInfo;    //
+  SNode*       pCondition;
 } SStreamBlockScanInfo;
 
 typedef struct SSysTableScanInfo {
@@ -672,7 +673,7 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pEx
                                        const STableGroupInfo* pTableGroupInfo);
 SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SExecTaskInfo* pTaskInfo);
 SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, SSDataBlock* pResBlock, SArray* pColList,
-                                            SArray* pTableIdList, SExecTaskInfo* pTaskInfo);
+                                            SArray* pTableIdList, SExecTaskInfo* pTaskInfo, SNode* pConditions);
 
 SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols,
                                       SInterval* pInterval, SSDataBlock* pResBlock, int32_t fillType, char* fillVal,
diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c
index 882c3c3252..9ff02369c5 100644
--- a/source/libs/executor/src/executorimpl.c
+++ b/source/libs/executor/src/executorimpl.c
@@ -6538,7 +6538,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
       int32_t numOfCols = 0;
       SArray* pCols = extractColMatchInfo(pScanPhyNode->pScanCols, pScanPhyNode->node.pOutputDataBlockDesc, &numOfCols);
       SOperatorInfo* pOperator =
-          createStreamScanOperatorInfo(pHandle->reader, pResBlock, pCols, tableIdList, pTaskInfo);
+          createStreamScanOperatorInfo(pHandle->reader, pResBlock, pCols, tableIdList, pTaskInfo, pScanPhyNode->node.pConditions);
       taosArrayDestroy(tableIdList);
       return pOperator;
     } else if (QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN == type) {
diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c
index dc87b864f1..8977e6e7c0 100644
--- a/source/libs/executor/src/scanoperator.c
+++ b/source/libs/executor/src/scanoperator.c
@@ -514,6 +514,7 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator, bool* newgroup)
   // NOTE: this operator does never check if current status is done or not
   SExecTaskInfo*        pTaskInfo = pOperator->pTaskInfo;
   SStreamBlockScanInfo* pInfo = pOperator->info;
+  int32_t rows = 0;
 
   pTaskInfo->code = pOperator->fpSet._openFn(pOperator);
   if (pTaskInfo->code != TSDB_CODE_SUCCESS || pOperator->status == OP_EXEC_DONE) {
@@ -580,6 +581,8 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator, bool* newgroup)
         pTaskInfo->code = terrno;
         return NULL;
       }
+      rows = pBlockInfo->rows;
+      doFilter(pInfo->pCondition, pInfo->pRes);
 
       break;
     }
@@ -588,16 +591,16 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator, bool* newgroup)
     pInfo->numOfExec++;
     pInfo->numOfRows += pBlockInfo->rows;
 
-    if (pBlockInfo->rows == 0) {
+    if (rows == 0) {
       pOperator->status = OP_EXEC_DONE;
     }
 
-    return (pBlockInfo->rows == 0) ? NULL : pInfo->pRes;
+    return (rows == 0) ? NULL : pInfo->pRes;
   }
 }
 
 SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, SSDataBlock* pResBlock, SArray* pColList,
-                                            SArray* pTableIdList, SExecTaskInfo* pTaskInfo) {
+                                            SArray* pTableIdList, SExecTaskInfo* pTaskInfo, SNode* pCondition) {
   SStreamBlockScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamBlockScanInfo));
   SOperatorInfo*        pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   if (pInfo == NULL || pOperator == NULL) {
@@ -635,6 +638,7 @@ SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, SSDataBlock*
 
   pInfo->readerHandle = streamReadHandle;
   pInfo->pRes = pResBlock;
+  pInfo->pCondition = pCondition;
 
   pOperator->name = "StreamBlockScanOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN;