From 4f305c74d10bb317b8e96bdad6dac080c946f927 Mon Sep 17 00:00:00 2001
From: wangmm0220 <wangmm0220@gmail.com>
Date: Fri, 24 Jun 2022 11:24:31 +0800
Subject: [PATCH] fix: table group error in stream

---
 source/libs/executor/src/executorimpl.c |  1 +
 source/libs/executor/src/scanoperator.c | 10 ++++++----
 source/libs/planner/src/planOptimizer.c |  2 +-
 3 files changed, 8 insertions(+), 5 deletions(-)

diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c
index b90c769c21..1e5d75655c 100644
--- a/source/libs/executor/src/executorimpl.c
+++ b/source/libs/executor/src/executorimpl.c
@@ -4055,6 +4055,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
       return pOperator;
     } else if (QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN == type) {
       STableMergeScanPhysiNode* pTableScanNode = (STableMergeScanPhysiNode*)pPhyNode;
+      pTableListInfo->needSortTableByGroupId = true;
       int32_t code = createScanTableListInfo(pTableScanNode, pHandle, pTableListInfo, queryId, taskId);
       if(code){
         return NULL;
diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c
index 81e8e19892..ccd7bf04aa 100644
--- a/source/libs/executor/src/scanoperator.c
+++ b/source/libs/executor/src/scanoperator.c
@@ -510,10 +510,11 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
   if(pInfo->currentGroupId == -1){
     pInfo->currentGroupId++;
     if (pInfo->currentGroupId >= taosArrayGetSize(pTaskInfo->tableqinfoList.pGroupList)) {
-      doSetOperatorCompleted(pOperator);
+      setTaskStatus(pTaskInfo, TASK_COMPLETED);
       return NULL;
     }
     SArray *tableList = taosArrayGetP(pTaskInfo->tableqinfoList.pGroupList, pInfo->currentGroupId);
+    tsdbCleanupReadHandle(pInfo->dataReader);
     tsdbReaderT* pReader = tsdbReaderOpen(pInfo->readHandle.vnode, &pInfo->cond, tableList, pInfo->queryId, pInfo->taskId);
     pInfo->dataReader = pReader;
   }
@@ -525,7 +526,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
 
   pInfo->currentGroupId++;
   if (pInfo->currentGroupId >= taosArrayGetSize(pTaskInfo->tableqinfoList.pGroupList)) {
-    doSetOperatorCompleted(pOperator);
+    setTaskStatus(pTaskInfo, TASK_COMPLETED);
     return NULL;
   }
 
@@ -541,7 +542,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
     return result;
   }
 
-  doSetOperatorCompleted(pOperator);
+  setTaskStatus(pTaskInfo, TASK_COMPLETED);
   return NULL;
 }
 
@@ -821,8 +822,9 @@ static bool prepareDataScan(SStreamBlockScanInfo* pInfo) {
   STableScanInfo* pTableScanInfo = pInfo->pSnapshotReadOp->info;
   pTableScanInfo->cond.twindows[0] = win;
   pTableScanInfo->curTWinIdx = 0;
-  tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
+//  tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
   pTableScanInfo->scanTimes = 0;
+  pTableScanInfo->currentGroupId = -1;
   return true;
 }
 
diff --git a/source/libs/planner/src/planOptimizer.c b/source/libs/planner/src/planOptimizer.c
index 5b9dbe8388..c2551392b3 100644
--- a/source/libs/planner/src/planOptimizer.c
+++ b/source/libs/planner/src/planOptimizer.c
@@ -1189,7 +1189,7 @@ static const SOptimizeRule optimizeRuleSet[] = {
   {.pName = "ConditionPushDown", .optimizeFunc = cpdOptimize},
   {.pName = "OrderByPrimaryKey", .optimizeFunc = opkOptimize},
   {.pName = "SmaIndex",          .optimizeFunc = smaOptimize},
-  // {.pName = "PartitionTags",     .optimizeFunc = partTagsOptimize},
+  {.pName = "PartitionTags",     .optimizeFunc = partTagsOptimize},
   {.pName = "EliminateProject",  .optimizeFunc = eliminateProjOptimize}
 };
 // clang-format on