From 996f995644fbf8bc2566b99a3c1cd8d979acee4e Mon Sep 17 00:00:00 2001
From: Liu Jicong <liujicong@qq.com>
Date: Sat, 25 Jun 2022 18:14:42 +0800
Subject: [PATCH] fix(sma): drop stream when drop sma

---
 source/dnode/mnode/impl/inc/mndStream.h |  4 ++++
 source/dnode/mnode/impl/src/mndSma.c    | 19 ++++++++++++++++++-
 source/dnode/mnode/impl/src/mndStream.c |  4 ++--
 source/dnode/vnode/src/tq/tqSink.c      |  2 +-
 source/libs/wal/src/walRead.c           |  3 ++-
 5 files changed, 27 insertions(+), 5 deletions(-)

diff --git a/source/dnode/mnode/impl/inc/mndStream.h b/source/dnode/mnode/impl/inc/mndStream.h
index 69385c3a46..0901e77287 100644
--- a/source/dnode/mnode/impl/inc/mndStream.h
+++ b/source/dnode/mnode/impl/inc/mndStream.h
@@ -33,6 +33,10 @@ SSdbRow *mndStreamActionDecode(SSdbRaw *pRaw);
 
 int32_t mndDropStreamByDb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb);
 int32_t mndPersistStream(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream);
+// for sma
+// TODO refactor
+int32_t mndDropStreamTasks(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream);
+int32_t mndPersistDropStreamLog(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream);
 
 #ifdef __cplusplus
 }
diff --git a/source/dnode/mnode/impl/src/mndSma.c b/source/dnode/mnode/impl/src/mndSma.c
index ef24cd0ba4..10eeaba982 100644
--- a/source/dnode/mnode/impl/src/mndSma.c
+++ b/source/dnode/mnode/impl/src/mndSma.c
@@ -15,11 +15,11 @@
 
 #define _DEFAULT_SOURCE
 #include "mndSma.h"
-#include "mndPrivilege.h"
 #include "mndDb.h"
 #include "mndDnode.h"
 #include "mndInfoSchema.h"
 #include "mndMnode.h"
+#include "mndPrivilege.h"
 #include "mndScheduler.h"
 #include "mndShow.h"
 #include "mndStb.h"
@@ -857,6 +857,23 @@ static int32_t mndDropSma(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SSmaObj *p
   mDebug("trans:%d, used to drop sma:%s", pTrans->id, pSma->name);
   mndTransSetDbName(pTrans, pDb->name, NULL);
 
+  SStreamObj *pStream = mndAcquireStream(pMnode, pSma->name);
+  if (pStream == NULL || pStream->smaId != pSma->uid) {
+    sdbRelease(pMnode->pSdb, pStream);
+    goto _OVER;
+  } else {
+    if (mndDropStreamTasks(pMnode, pTrans, pStream) < 0) {
+      mError("stream:%s, failed to drop task since %s", pStream->name, terrstr());
+      sdbRelease(pMnode->pSdb, pStream);
+      goto _OVER;
+    }
+
+    // drop stream
+    if (mndPersistDropStreamLog(pMnode, pTrans, pStream) < 0) {
+      sdbRelease(pMnode->pSdb, pStream);
+      goto _OVER;
+    }
+  }
   if (mndSetDropSmaRedoLogs(pMnode, pTrans, pSma) != 0) goto _OVER;
   if (mndSetDropSmaVgroupRedoLogs(pMnode, pTrans, pVgroup) != 0) goto _OVER;
   if (mndSetDropSmaCommitLogs(pMnode, pTrans, pSma) != 0) goto _OVER;
diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c
index b78756d8b8..cabbac14f1 100644
--- a/source/dnode/mnode/impl/src/mndStream.c
+++ b/source/dnode/mnode/impl/src/mndStream.c
@@ -14,10 +14,10 @@
  */
 
 #include "mndStream.h"
-#include "mndPrivilege.h"
 #include "mndDb.h"
 #include "mndDnode.h"
 #include "mndMnode.h"
+#include "mndPrivilege.h"
 #include "mndScheduler.h"
 #include "mndShow.h"
 #include "mndStb.h"
@@ -490,7 +490,7 @@ static int32_t mndPersistTaskDropReq(STrans *pTrans, SStreamTask *pTask) {
   return 0;
 }
 
-static int32_t mndDropStreamTasks(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream) {
+int32_t mndDropStreamTasks(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream) {
   int32_t lv = taosArrayGetSize(pStream->tasks);
   for (int32_t i = 0; i < lv; i++) {
     SArray *pTasks = taosArrayGetP(pStream->tasks, i);
diff --git a/source/dnode/vnode/src/tq/tqSink.c b/source/dnode/vnode/src/tq/tqSink.c
index 9abc2f639b..0bb9918488 100644
--- a/source/dnode/vnode/src/tq/tqSink.c
+++ b/source/dnode/vnode/src/tq/tqSink.c
@@ -43,7 +43,7 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo
       taosArrayPush(tagArray, &tagVal);
       tTagNew(tagArray, 1, false, &pTag);
       if (pTag == NULL) {
-        taosArrayDestroy(schemaReqs);
+        terrno = TSDB_CODE_OUT_OF_MEMORY;
         taosArrayDestroy(tagArray);
         return NULL;
       }
diff --git a/source/libs/wal/src/walRead.c b/source/libs/wal/src/walRead.c
index 682afbb785..2de0fea9ac 100644
--- a/source/libs/wal/src/walRead.c
+++ b/source/libs/wal/src/walRead.c
@@ -104,6 +104,8 @@ static int32_t walReadChangeFile(SWalReadHandle *pRead, int64_t fileFirstVer) {
     return -1;
   }
 
+  pRead->pReadLogTFile = pLogTFile;
+
   walBuildIdxName(pRead->pWal, fileFirstVer, fnameStr);
   TdFilePtr pIdxTFile = taosOpenFile(fnameStr, TD_FILE_READ);
   if (pIdxTFile == NULL) {
@@ -112,7 +114,6 @@ static int32_t walReadChangeFile(SWalReadHandle *pRead, int64_t fileFirstVer) {
     return -1;
   }
 
-  pRead->pReadLogTFile = pLogTFile;
   pRead->pReadIdxTFile = pIdxTFile;
   return 0;
 }