From 24bc71568ee948212a4ce15d292a556a795b9077 Mon Sep 17 00:00:00 2001
From: Liu Jicong <liujicong@qq.com>
Date: Tue, 19 Jul 2022 18:29:12 +0800
Subject: [PATCH] refactor(stream): batch optimization for submit msg

---
 source/libs/executor/src/executor.c     | 11 +++++++----
 source/libs/executor/src/scanoperator.c |  2 +-
 source/libs/stream/src/streamData.c     |  6 +++---
 3 files changed, 11 insertions(+), 8 deletions(-)

diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c
index 8fc9ee496b..925e4891fd 100644
--- a/source/libs/executor/src/executor.c
+++ b/source/libs/executor/src/executor.c
@@ -51,10 +51,13 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
       /*qError("submit msg messed up when initing stream block, %s" PRIx64, id);*/
       /*return TSDB_CODE_QRY_APP_ERROR;*/
       /*}*/
-      taosArrayClear(pInfo->pBlockLists);
-      for (int32_t i = 0; i < numOfBlocks; i++) {
-        SSubmitReq* pReq = POINTER_SHIFT(input, i * sizeof(void*));
-        taosArrayPush(pInfo->pBlockLists, &pReq);
+      if (numOfBlocks == 1) {
+        taosArrayPush(pInfo->pBlockLists, &input);
+      } else {
+        for (int32_t i = 0; i < numOfBlocks; i++) {
+          SSubmitReq* pReq = *(void**)POINTER_SHIFT(input, i * sizeof(void*));
+          taosArrayPush(pInfo->pBlockLists, &pReq);
+        }
       }
     } else if (type == STREAM_INPUT__DATA_BLOCK) {
       for (int32_t i = 0; i < numOfBlocks; ++i) {
diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c
index 698bdab71a..213988cb0f 100644
--- a/source/libs/executor/src/scanoperator.c
+++ b/source/libs/executor/src/scanoperator.c
@@ -1433,7 +1433,7 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
         break;
       } else {
         pInfo->tqReader->pMsg = NULL;
-        return NULL;
+        continue;
       }
       /*blockDataCleanup(pInfo->pRes);*/
     }
diff --git a/source/libs/stream/src/streamData.c b/source/libs/stream/src/streamData.c
index b28dba3472..d476980393 100644
--- a/source/libs/stream/src/streamData.c
+++ b/source/libs/stream/src/streamData.c
@@ -81,7 +81,7 @@ SStreamMergedSubmit* streamMergedSubmitNew() {
   SStreamMergedSubmit* pMerged = (SStreamMergedSubmit*)taosAllocateQitem(sizeof(SStreamMergedSubmit), DEF_QITEM);
   if (pMerged == NULL) return NULL;
   pMerged->reqs = taosArrayInit(0, sizeof(void*));
-  pMerged->dataRefs = taosArrayInit(0, sizeof(int32_t*));
+  pMerged->dataRefs = taosArrayInit(0, sizeof(void*));
   if (pMerged->dataRefs == NULL || pMerged->reqs == NULL) goto FAIL;
   pMerged->type = STREAM_INPUT__MERGED_SUBMIT;
   return pMerged;
@@ -93,7 +93,7 @@ FAIL:
 }
 
 int32_t streamMergeSubmit(SStreamMergedSubmit* pMerged, SStreamDataSubmit* pSubmit) {
-  taosArrayPush(pMerged->dataRefs, pSubmit->dataRef);
+  taosArrayPush(pMerged->dataRefs, &pSubmit->dataRef);
   taosArrayPush(pMerged->reqs, &pSubmit->data);
   pMerged->ver = pSubmit->ver;
   return 0;
@@ -165,7 +165,7 @@ void streamFreeQitem(SStreamQueueItem* data) {
     SStreamMergedSubmit* pMerge = (SStreamMergedSubmit*)data;
     int32_t              sz = taosArrayGetSize(pMerge->reqs);
     for (int32_t i = 0; i < sz; i++) {
-      int32_t* ref = taosArrayGet(pMerge->dataRefs, i);
+      int32_t* ref = taosArrayGetP(pMerge->dataRefs, i);
       (*ref)--;
       if (*ref == 0) {
         void* data = taosArrayGetP(pMerge->reqs, i);