From a98bf9d1b57270c070b9bf0dce777647a1cf36a8 Mon Sep 17 00:00:00 2001
From: Liu Jicong <liujicong@qq.com>
Date: Thu, 21 Apr 2022 14:19:58 +0800
Subject: [PATCH] add convert for rSma

---
 include/libs/planner/planner.h             |  1 +
 source/dnode/mnode/impl/inc/mndScheduler.h |  2 +
 source/dnode/mnode/impl/src/mndScheduler.c | 48 ++++++++++++++++++++++
 source/dnode/vnode/src/tq/tqRead.c         |  2 +-
 4 files changed, 52 insertions(+), 1 deletion(-)

diff --git a/include/libs/planner/planner.h b/include/libs/planner/planner.h
index f343295c56..0ebbfd015a 100644
--- a/include/libs/planner/planner.h
+++ b/include/libs/planner/planner.h
@@ -30,6 +30,7 @@ typedef struct SPlanContext {
   SNode* pAstRoot;
   bool topicQuery;
   bool streamQuery;
+  bool rSmaQuery;
   bool showRewrite;
   int8_t triggerType;
   int64_t watermark;
diff --git a/source/dnode/mnode/impl/inc/mndScheduler.h b/source/dnode/mnode/impl/inc/mndScheduler.h
index 42951beca2..33af040915 100644
--- a/source/dnode/mnode/impl/inc/mndScheduler.h
+++ b/source/dnode/mnode/impl/inc/mndScheduler.h
@@ -29,6 +29,8 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
 
 int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream);
 
+int32_t mndConvertRSmaTask(const char* ast, int8_t triggerType, int64_t watermark, char** pStr, int32_t* pLen);
+
 #ifdef __cplusplus
 }
 #endif
diff --git a/source/dnode/mnode/impl/src/mndScheduler.c b/source/dnode/mnode/impl/src/mndScheduler.c
index b3583af1dc..73583058f1 100644
--- a/source/dnode/mnode/impl/src/mndScheduler.c
+++ b/source/dnode/mnode/impl/src/mndScheduler.c
@@ -34,6 +34,54 @@
 
 extern bool tsStreamSchedV;
 
+int32_t mndConvertRSmaTask(const char* ast, int8_t triggerType, int64_t watermark, char** pStr, int32_t* pLen) {
+  SNode*      pAst = NULL;
+  SQueryPlan* pPlan = NULL;
+  terrno = TSDB_CODE_SUCCESS;
+
+  if (nodesStringToNode(ast, &pAst) < 0) {
+    terrno = TSDB_CODE_QRY_INVALID_INPUT;
+    goto END;
+  }
+
+  SPlanContext cxt = {
+      .pAstRoot = pAst,
+      .topicQuery = false,
+      .streamQuery = true,
+      .rSmaQuery = true,
+      .triggerType = triggerType,
+      .watermark = watermark,
+  };
+  if (qCreateQueryPlan(&cxt, &pPlan, NULL) < 0) {
+    terrno = TSDB_CODE_QRY_INVALID_INPUT;
+    goto END;
+  }
+
+  int32_t levelNum = LIST_LENGTH(pPlan->pSubplans);
+  if (levelNum != 1) {
+    terrno = TSDB_CODE_QRY_INVALID_INPUT;
+    goto END;
+  }
+  SNodeListNode* inner = nodesListGetNode(pPlan->pSubplans, 0);
+
+  int32_t opNum = LIST_LENGTH(inner->pNodeList);
+  if (opNum != 1) {
+    terrno = TSDB_CODE_QRY_INVALID_INPUT;
+    goto END;
+  }
+
+  SSubplan* plan = nodesListGetNode(inner->pNodeList, 0);
+  if (qSubPlanToString(plan, pStr, pLen) < 0) {
+    terrno = TSDB_CODE_QRY_INVALID_INPUT;
+    goto END;
+  }
+
+END:
+  if (pAst) nodesDestroyNode(pAst);
+  if (pPlan) nodesDestroyNode(pPlan);
+  return terrno;
+}
+
 int32_t mndPersistTaskDeployReq(STrans* pTrans, SStreamTask* pTask, const SEpSet* pEpSet, tmsg_t type, int32_t nodeId) {
   SCoder encoder;
   tCoderInit(&encoder, TD_LITTLE_ENDIAN, NULL, 0, TD_ENCODER);
diff --git a/source/dnode/vnode/src/tq/tqRead.c b/source/dnode/vnode/src/tq/tqRead.c
index c94fbeec98..2994aabd02 100644
--- a/source/dnode/vnode/src/tq/tqRead.c
+++ b/source/dnode/vnode/src/tq/tqRead.c
@@ -141,7 +141,7 @@ int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, int16_t* pG
       colInfo.info.colId = pColSchema->colId;
       colInfo.info.type = pColSchema->type;
 
-      if (colInfoDataEnsureCapacity(&colInfo, 0, numOfRows) < 0) {
+      if (colInfoDataEnsureCapacity(&colInfo, 0, *pNumOfRows) < 0) {
         goto FAIL;
       }
       taosArrayPush(*ppCols, &colInfo);
-- 
GitLab