From 14c6dbd2af6f7a5b440abc25c0c99ea0b7e1c496 Mon Sep 17 00:00:00 2001
From: 54liuyao <54liuyao>
Date: Tue, 12 Mar 2024 11:45:54 +0800
Subject: [PATCH] use non sort merge

---
 source/libs/executor/src/mergeoperator.c   |  14 ++-
 source/libs/executor/src/projectoperator.c |  10 ++
 source/libs/planner/src/planSpliter.c      |   6 +-
 tests/script/tsim/query/interval.sim       | 131 +++++++++++++++++++++
 4 files changed, 158 insertions(+), 3 deletions(-)

diff --git a/source/libs/executor/src/mergeoperator.c b/source/libs/executor/src/mergeoperator.c
index 093b6ab11e..caef9e1de6 100755
--- a/source/libs/executor/src/mergeoperator.c
+++ b/source/libs/executor/src/mergeoperator.c
@@ -269,6 +269,7 @@ SSDataBlock* doNonSortMerge(SOperatorInfo* pOperator) {
   SMultiwayMergeOperatorInfo* pInfo = pOperator->info;
   SNonSortMergeInfo*          pNonSortMerge = &pInfo->nsortMergeInfo;
   SSDataBlock*                pBlock = NULL;
+  SSDataBlock*                pRes = pInfo->binfo.pRes;
 
   qDebug("start to merge no sorted rows, %s", GET_TASKID(pTaskInfo));
 
@@ -278,13 +279,19 @@ SSDataBlock* doNonSortMerge(SOperatorInfo* pOperator) {
     if (NULL == pBlock) {
       TSWAP(pNonSortMerge->pSourceStatus[pNonSortMerge->sourceWorkIdx], pNonSortMerge->pSourceStatus[idx]);
       pNonSortMerge->sourceWorkIdx++;
-      idx = NON_SORT_NEXT_SRC(pNonSortMerge, idx);
+      idx = NON_SORT_NEXT_SRC(pNonSortMerge, pNonSortMerge->lastSourceIdx);
       continue;
     }
     break;
   }
 
-  return pBlock;
+  if (!pBlock) {
+    return NULL;
+  }
+  blockDataCleanup(pRes);
+  copyDataBlock(pRes, pBlock);
+
+  return pRes;
 }
 
 void destroyNonSortMergeOperatorInfo(void* param) {
@@ -491,6 +498,9 @@ SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** downStreams, size
     }
     case MERGE_TYPE_NON_SORT: {
       SNonSortMergeInfo* pNonSortMerge = &pInfo->nsortMergeInfo;
+      pInfo->binfo.pRes = createDataBlockFromDescNode(pDescNode);
+      initResultSizeInfo(&pOperator->resultInfo, 1024);
+      blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
       break;
     }
     case MERGE_TYPE_COLUMNS: {
diff --git a/source/libs/executor/src/projectoperator.c b/source/libs/executor/src/projectoperator.c
index 691210e5ba..c6ebb04446 100644
--- a/source/libs/executor/src/projectoperator.c
+++ b/source/libs/executor/src/projectoperator.c
@@ -28,6 +28,7 @@ typedef struct SProjectOperatorInfo {
   bool           mergeDataBlocks;
   SSDataBlock*   pFinalRes;
   bool           inputIgnoreGroup;
+  bool           outputIgnoreGroup;
 } SProjectOperatorInfo;
 
 typedef struct SIndefOperatorInfo {
@@ -111,6 +112,7 @@ SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhys
   pInfo->binfo.inputTsOrder = pProjPhyNode->node.inputTsOrder;
   pInfo->binfo.outputTsOrder = pProjPhyNode->node.outputTsOrder;
   pInfo->inputIgnoreGroup = pProjPhyNode->inputIgnoreGroup;
+  pInfo->outputIgnoreGroup = pProjPhyNode->ignoreGroupId;
   
   if (pTaskInfo->execModel == OPTR_EXEC_MODEL_STREAM || pTaskInfo->execModel == OPTR_EXEC_MODEL_QUEUE) {
     pInfo->mergeDataBlocks = false;
@@ -276,6 +278,10 @@ SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
       T_LONG_JMP(pTaskInfo->env, code);
     }
 
+    if (pProjectInfo->outputIgnoreGroup) {
+      pRes->info.id.groupId = 0;
+    }
+
     return (pRes->info.rows > 0) ? pRes : NULL;
   }
 
@@ -385,6 +391,10 @@ SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
     printDataBlock(p, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
   }
 
+  if (pProjectInfo->outputIgnoreGroup) {
+    p->info.id.groupId = 0;
+  }
+
   return (p->info.rows > 0) ? p : NULL;
 }
 
diff --git a/source/libs/planner/src/planSpliter.c b/source/libs/planner/src/planSpliter.c
index 9664be1da2..49bde5ae4c 100644
--- a/source/libs/planner/src/planSpliter.c
+++ b/source/libs/planner/src/planSpliter.c
@@ -1320,7 +1320,11 @@ static int32_t stbSplSplitScanNodeWithPartTags(SSplitContext* pCxt, SStableSplit
   SLogicNode* pSplitNode = NULL;
   int32_t     code = stbSplGetSplitNodeForScan(pInfo, &pSplitNode);
   if (TSDB_CODE_SUCCESS == code) {
-    code = stbSplCreateMergeNode(pCxt, pInfo->pSubplan, pSplitNode, NULL, pSplitNode, true, true);
+    bool needSort = true;
+    if (QUERY_NODE_LOGIC_PLAN_PROJECT == nodeType(pSplitNode)) {
+      needSort = !((SProjectLogicNode*)pSplitNode)->ignoreGroupId;
+    }
+    code = stbSplCreateMergeNode(pCxt, pInfo->pSubplan, pSplitNode, NULL, pSplitNode, needSort, needSort);
   }
   if (TSDB_CODE_SUCCESS == code) {
     code = nodesListMakeStrictAppend(&pInfo->pSubplan->pChildren,
diff --git a/tests/script/tsim/query/interval.sim b/tests/script/tsim/query/interval.sim
index 135fcc8591..7f950ea69c 100644
--- a/tests/script/tsim/query/interval.sim
+++ b/tests/script/tsim/query/interval.sim
@@ -226,4 +226,135 @@ print =============== clear
 #  return -1
 #endi
 
+print ================= step12
+
+sql create database test2 vgroups 4;
+sql use test2;
+sql create stable stb (ts timestamp, c1 int) tags (t1 int);
+sql create table t1 using stb tags (1);
+sql create table t2 using stb tags (2);
+sql create table t3 using stb tags (3);
+sql create table t4 using stb tags (4);
+sql create table t5 using stb tags (4);
+sql create table t6 using stb tags (4);
+sql insert into t1 values ("2024-03-01 14:29:07.051", 11);
+sql insert into t2 values ("2024-03-01 14:29:07.051", 21);
+sql insert into t3 values ("2024-03-01 14:29:07.051", 31);
+sql insert into t4 values ("2024-03-01 14:29:07.051", 41);
+sql insert into t5 values ("2024-03-01 14:29:07.051", 51);
+sql insert into t6 values ("2024-03-01 14:29:07.051", 61);
+sql insert into t1 values ("2024-03-01 14:30:07.051", 12);
+sql insert into t2 values ("2024-03-01 14:30:07.051", 22);
+sql insert into t3 values ("2024-03-01 14:30:07.051", 32);
+sql insert into t4 values ("2024-03-01 14:30:07.051", 42);
+sql insert into t5 values ("2024-03-01 14:30:07.051", 52);
+sql insert into t6 values ("2024-03-01 14:30:07.051", 62);
+sql insert into t1 values ("2024-03-01 14:31:07.051", 13);
+sql insert into t2 values ("2024-03-01 14:31:07.051", 23);
+sql insert into t3 values ("2024-03-01 14:31:07.051", 33);
+sql insert into t4 values ("2024-03-01 14:31:07.051", 43);
+sql insert into t5 values ("2024-03-01 14:31:07.051", 53);
+sql insert into t6 values ("2024-03-01 14:31:07.051", 63);
+sql insert into t1 values ("2024-03-01 14:32:07.051", 14);
+sql insert into t2 values ("2024-03-01 14:32:07.051", 24);
+sql insert into t3 values ("2024-03-01 14:32:07.051", 34);
+sql insert into t4 values ("2024-03-01 14:32:07.051", 44);
+sql insert into t5 values ("2024-03-01 14:32:07.051", 54);
+sql insert into t6 values ("2024-03-01 14:32:07.051", 64);
+sql insert into t1 values ("2024-03-01 14:33:07.051", 15);
+sql insert into t2 values ("2024-03-01 14:33:07.051", 25);
+sql insert into t3 values ("2024-03-01 14:33:07.051", 35);
+sql insert into t4 values ("2024-03-01 14:33:07.051", 45);
+sql insert into t5 values ("2024-03-01 14:33:07.051", 55);
+sql insert into t6 values ("2024-03-01 14:33:07.051", 65);
+sql insert into t1 values ("2024-03-01 14:34:07.051", 16);
+sql insert into t2 values ("2024-03-01 14:34:07.051", 26);
+sql insert into t3 values ("2024-03-01 14:34:07.051", 36);
+sql insert into t4 values ("2024-03-01 14:34:07.051", 46);
+sql insert into t5 values ("2024-03-01 14:34:07.051", 56);
+sql insert into t6 values ("2024-03-01 14:34:07.051", 66);
+
+sleep 300
+
+sql select _wstart, count(*) from (select * from stb partition by tbname) interval(2s);
+
+print $data00,$data01
+print $data10,$data11
+print $data20,$data21
+print $data30,$data31
+print $data40,$data41
+print $data50,$data51
+print $data60,$data61
+print $data70,$data71
+
+if $rows != 6 then
+  print $rows
+  return -1
+endi
+
+if $data01 != 6 then
+  print $data01
+endi
+
+if $data11 != 6 then
+  print $data11
+endi
+
+if $data21 != 6 then
+  print $data21
+endi
+
+if $data31 != 6 then
+  print $data31
+endi
+
+if $data41 != 6 then
+  print $data41
+endi
+
+if $data51 != 6 then
+  print $data51
+endi
+
+
+sql select _wstart, count(*) from (select * from stb partition by tbname slimit 2) interval(2s);
+
+print $data00,$data01
+print $data10,$data11
+print $data20,$data21
+print $data30,$data31
+print $data40,$data41
+print $data50,$data51
+print $data60,$data61
+print $data70,$data71
+
+if $rows != 6 then
+  print $rows then
+  return -1
+endi
+
+if $data01 != 2 then
+  print $data01
+endi
+
+if $data11 != 2 then
+  print $data11
+endi
+
+if $data21 != 2 then
+  print $data21
+endi
+
+if $data31 != 2 then
+  print $data31
+endi
+
+if $data41 != 2 then
+  print $data41
+endi
+
+if $data51 != 2 then
+  print $data51
+endi
+
 system sh/exec.sh -n dnode1 -s stop -x SIGINT