Merge pull request #14448 from taosdata/fix/tsim

fix: drop sma

source/dnode/mgmt/mgmt_vnode/src/vmInt.c

@@ -21,7 +21,7 @@ SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId) {
 
   taosThreadRwlockRdlock(&pMgmt->lock);
   taosHashGetDup(pMgmt->hash, &vgId, sizeof(int32_t), (void *)&pVnode);
-  if (pVnode == NULL) {
+  if (pVnode == NULL || pVnode->dropped) {
     terrno = TSDB_CODE_VND_INVALID_VGROUP_ID;
   } else {
     int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1);
@@ -81,16 +81,18 @@ void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
 
   vmReleaseVnode(pMgmt, pVnode);
   while (pVnode->refCount > 0) taosMsleep(10);
+  dTrace("vgId:%d, wait for vnode queue is empty", pVnode->vgId);
+
   while (!taosQueueEmpty(pVnode->pWriteQ)) taosMsleep(10);
   while (!taosQueueEmpty(pVnode->pSyncQ)) taosMsleep(10);
   while (!taosQueueEmpty(pVnode->pApplyQ)) taosMsleep(10);
   while (!taosQueueEmpty(pVnode->pQueryQ)) taosMsleep(10);
   while (!taosQueueEmpty(pVnode->pFetchQ)) taosMsleep(10);
+  dTrace("vgId:%d, vnode-fetch queue is empty", pVnode->vgId);
 
   vmFreeQueue(pMgmt, pVnode);
   vnodeClose(pVnode->pImpl);
   pVnode->pImpl = NULL;
-
   dDebug("vgId:%d, vnode is closed", pVnode->vgId);
 
   if (pVnode->dropped) {

source/dnode/mgmt/mgmt_vnode/src/vmWorker.c

@@ -107,7 +107,7 @@ static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOf
     const STraceId *trace = &pMsg->info.traceId;
     dGTrace("vgId:%d, msg:%p get from vnode-sync queue", pVnode->vgId, pMsg);
 
-    int32_t code = vnodeProcessSyncReq(pVnode->pImpl, pMsg, NULL); // no response here
+    int32_t code = vnodeProcessSyncReq(pVnode->pImpl, pMsg, NULL);  // no response here
     dGTrace("vgId:%d, msg:%p is freed, code:0x%x", pVnode->vgId, pMsg, code);
     rpcFreeCont(pMsg->pCont);
     taosFreeQitem(pMsg);
@@ -146,8 +146,8 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp
 
   SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
   if (pVnode == NULL) {
-    dGError("vgId:%d, msg:%p failed to put into vnode queue since %s, type:%s", pHead->vgId, pMsg, terrstr(),
-            TMSG_INFO(pMsg->msgType));
+    dGError("vgId:%d, msg:%p failed to put into vnode queue since %s, msgtype:%s qtype:%d", pHead->vgId, pMsg,
+            terrstr(), TMSG_INFO(pMsg->msgType), qtype);
     return terrno != 0 ? terrno : -1;
   }
 

source/dnode/mnode/impl/inc/mndStream.h

@@ -38,6 +38,8 @@ int32_t mndPersistStream(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream);
 int32_t mndDropStreamTasks(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream);
 int32_t mndPersistDropStreamLog(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream);
 
+int32_t mndDropStreamByDb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb);
+
 #ifdef __cplusplus
 }
 #endif

source/dnode/mnode/impl/src/mndDb.c

@@ -21,6 +21,7 @@
 #include "mndShow.h"
 #include "mndSma.h"
 #include "mndStb.h"
+#include "mndStream.h"
 #include "mndSubscribe.h"
 #include "mndTopic.h"
 #include "mndTrans.h"
@@ -927,6 +928,7 @@ static int32_t mndDropDb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb) {
   if (mndDropOffsetByDB(pMnode, pTrans, pDb) != 0) goto _OVER;
   if (mndDropSubByDB(pMnode, pTrans, pDb) != 0) goto _OVER;
   if (mndDropTopicByDB(pMnode, pTrans, pDb) != 0) goto _OVER;
+  if (mndDropStreamByDb(pMnode, pTrans, pDb) != 0) goto _OVER;
   if (mndDropSmasByDb(pMnode, pTrans, pDb) != 0) goto _OVER;
   if (mndSetDropDbRedoActions(pMnode, pTrans, pDb) != 0) goto _OVER;
 
@@ -947,7 +949,6 @@ static int32_t mndDropDb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb) {
   mndTransSetRpcRsp(pTrans, pRsp, rspLen);
 
   if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
-
   code = 0;
 
 _OVER:

source/dnode/mnode/impl/src/mndSma.c

@@ -523,6 +523,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
   streamObj.updateTime = streamObj.createTime;
   streamObj.uid = mndGenerateUid(pCreate->name, strlen(pCreate->name));
   streamObj.sourceDbUid = pDb->uid;
+  streamObj.targetDbUid = pDb->uid;
   streamObj.version = 1;
   streamObj.sql = pCreate->sql;
   streamObj.smaId = smaObj.uid;
@@ -853,36 +854,26 @@ _OVER:
 }
 
 int32_t mndDropSmasByDb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) {
-  SSdb    *pSdb = pMnode->pSdb;
-  SSmaObj *pSma = NULL;
-  void    *pIter = NULL;
-  SVgObj  *pVgroup = NULL;
-  int32_t  code = -1;
+  SSdb *pSdb = pMnode->pSdb;
+  void *pIter = NULL;
 
   while (1) {
+    SSmaObj *pSma = NULL;
     pIter = sdbFetch(pSdb, SDB_SMA, pIter, (void **)&pSma);
     if (pIter == NULL) break;
 
     if (pSma->dbUid == pDb->uid) {
-      pVgroup = mndAcquireVgroup(pMnode, pSma->dstVgId);
-      if (pVgroup == NULL) goto _OVER;
-      if (mndSetDropSmaVgroupCommitLogs(pMnode, pTrans, pVgroup) != 0) goto _OVER;
-      if (mndSetDropSmaVgroupRedoActions(pMnode, pTrans, pDb, pVgroup) != 0) goto _OVER;
-      if (mndSetDropSmaCommitLogs(pMnode, pTrans, pSma) != 0) goto _OVER;
-      mndReleaseVgroup(pMnode, pVgroup);
-      pVgroup = NULL;
+      if (mndSetDropSmaCommitLogs(pMnode, pTrans, pSma) != 0) {
+        sdbRelease(pSdb, pSma);
+        sdbCancelFetch(pSdb, pSma);
+        return -1;
+      }
     }
 
     sdbRelease(pSdb, pSma);
   }
 
-  code = 0;
-
-_OVER:
-  sdbCancelFetch(pSdb, pIter);
-  sdbRelease(pSdb, pSma);
-  mndReleaseVgroup(pMnode, pVgroup);
-  return code;
+  return 0;
 }
 
 static int32_t mndProcessDropSmaReq(SRpcMsg *pReq) {

source/dnode/mnode/impl/src/mndStream.c

@@ -674,27 +674,29 @@ static int32_t mndProcessDropStreamReq(SRpcMsg *pReq) {
 
 int32_t mndDropStreamByDb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) {
   SSdb *pSdb = pMnode->pSdb;
+  void *pIter = NULL;
 
-  void       *pIter = NULL;
-  SStreamObj *pStream = NULL;
   while (1) {
+    SStreamObj *pStream = NULL;
     pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&pStream);
     if (pIter == NULL) break;
 
     if (pStream->sourceDbUid == pDb->uid || pStream->targetDbUid == pDb->uid) {
       if (pStream->sourceDbUid != pStream->targetDbUid) {
         sdbRelease(pSdb, pStream);
+        sdbCancelFetch(pSdb, pIter);
+        mError("db:%s, failed to drop stream:%s since sourceDbUid:%" PRId64 " not match with targetDbUid:%" PRId64,
+               pDb->name, pStream->name, pStream->sourceDbUid, pStream->targetDbUid);
+        terrno = TSDB_CODE_MND_STREAM_ALREADY_EXIST;
         return -1;
       } else {
         // TODO drop all task on snode
         if (mndPersistDropStreamLog(pMnode, pTrans, pStream) < 0) {
           sdbRelease(pSdb, pStream);
+          sdbCancelFetch(pSdb, pIter);
           return -1;
         }
       }
-    } else {
-      sdbRelease(pSdb, pStream);
-      continue;
     }
 
 #if 0

source/dnode/mnode/impl/src/mndTrans.c

@@ -90,7 +90,7 @@ static int32_t mndTransGetActionsSize(SArray *pArray) {
   for (int32_t i = 0; i < actionNum; ++i) {
     STransAction *pAction = taosArrayGet(pArray, i);
     if (pAction->actionType == TRANS_ACTION_RAW) {
-      rawDataLen += (sdbGetRawTotalSize(pAction->pRaw) + sizeof(int32_t));
+      rawDataLen += (sizeof(STransAction) + sdbGetRawTotalSize(pAction->pRaw));
     } else if (pAction->actionType == TRANS_ACTION_MSG) {
       rawDataLen += (sizeof(STransAction) + pAction->contLen);
     } else {
@@ -105,7 +105,7 @@ static int32_t mndTransGetActionsSize(SArray *pArray) {
 static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
   terrno = TSDB_CODE_OUT_OF_MEMORY;
 
-  int32_t rawDataLen = sizeof(STrans) + TRANS_RESERVE_SIZE;
+  int32_t rawDataLen = sizeof(STrans) + TRANS_RESERVE_SIZE + pTrans->paramLen;
   rawDataLen += mndTransGetActionsSize(pTrans->redoActions);
   rawDataLen += mndTransGetActionsSize(pTrans->undoActions);
   rawDataLen += mndTransGetActionsSize(pTrans->commitActions);
@@ -226,7 +226,8 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
 
 _OVER:
   if (terrno != 0) {
-    mError("trans:%d, failed to encode to raw:%p len:%d since %s", pTrans->id, pRaw, dataPos, terrstr());
+    mError("trans:%d, failed to encode to raw:%p maxlen:%d len:%d since %s", pTrans->id, pRaw, sdbGetRawTotalSize(pRaw),
+           dataPos, terrstr());
     sdbFreeRaw(pRaw);
     return NULL;
   }

source/dnode/vnode/src/tq/tqSink.c

@@ -185,5 +185,7 @@ void tqTableSink(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
       .contLen = ntohl(pReq->length),
   };
 
-  ASSERT(tmsgPutToQueue(&pVnode->msgCb, WRITE_QUEUE, &msg) == 0);
+  if (tmsgPutToQueue(&pVnode->msgCb, WRITE_QUEUE, &msg) != 0) {
+    tqDebug("failed to put into write-queue since %s", terrstr());
+  }
 }

source/libs/index/CMakeLists.txt

@@ -35,7 +35,7 @@ if (${BUILD_WITH_INVERTEDINDEX})
 endif(${BUILD_WITH_INVERTEDINDEX})
 
 
-if (${BUILD_TEST})
-  add_subdirectory(test)
-endif(${BUILD_TEST})
+# if (${BUILD_TEST})
+#   add_subdirectory(test)
+# endif(${BUILD_TEST})
 

tests/script/jenkins/basic.txt

@@ -93,6 +93,7 @@
 ./test.sh -f tsim/stream/basic0.sim
 ./test.sh -f tsim/stream/basic1.sim
 ./test.sh -f tsim/stream/basic2.sim
+./test.sh -f tsim/stream/drop_stream.sim
 ./test.sh -f tsim/stream/distributeInterval0.sim
 # ./test.sh -f tsim/stream/distributeIntervalRetrive0.sim
 # ./test.sh -f tsim/stream/distributesession0.sim
@@ -159,6 +160,7 @@
 #./test.sh -f tsim/mnode/basic1.sim -m
 
 # --- sma
+./test.sh -f tsim/sma/drop_sma.sim
 ./test.sh -f tsim/sma/tsmaCreateInsertQuery.sim
 ./test.sh -f tsim/sma/rsmaCreateInsertQuery.sim
 

tests/script/sh/deploy.sh

@@ -135,7 +135,7 @@ echo "jniDebugFlag           143"                >> $TAOS_CFG
 echo "qDebugFlag             143"                >> $TAOS_CFG
 echo "rpcDebugFlag           143"                >> $TAOS_CFG
 echo "tmrDebugFlag           131"                >> $TAOS_CFG
-echo "uDebugFlag             143"                >> $TAOS_CFG
+echo "uDebugFlag             131"                >> $TAOS_CFG
 echo "sDebugFlag             143"                >> $TAOS_CFG
 echo "wDebugFlag             143"                >> $TAOS_CFG
 echo "numOfLogLines          20000000"           >> $TAOS_CFG

tests/script/tsim/sma/drop_sma.sim

@@ -64,12 +64,59 @@ sql create sma index sma_index_name1 on stb function(max(c1),max(c2),min(c1)) in
 print --> drop stb
 sql drop table stb; 
 
+print ========== step5 
+sql drop database if exists db;
+sql create database db duration 300;
+sql use db;
+sql create table stb1(ts timestamp, c_int int, c_bint bigint, c_sint smallint, c_tint tinyint, c_float float, c_double double, c_bool bool, c_binary binary(16), c_nchar nchar(32), c_ts timestamp, c_tint_un tinyint unsigned, c_sint_un smallint unsigned, c_int_un int unsigned, c_bint_un bigint unsigned) tags (t_int int);
+sql CREATE SMA INDEX sma_index_1 ON stb1 function(min(c_int), max(c_int)) interval(6m, 10s) sliding(6m) watermark 5s;
+
+print ========== step6 repeat
+sql drop database if exists db;
+sql create database db duration 300;
+sql use db;
+sql create table stb1(ts timestamp, c_int int, c_bint bigint ) tags (t_int int);
+sql CREATE SMA INDEX sma_index_1 ON stb1 function(min(c_int), max(c_int)) interval(6m, 10s) sliding(6m) watermark 5s;
+
+print ========== step7
+sql drop database if exists db;
+sql create database db duration 300;
+sql use db;
+sql create table stb1(ts timestamp, c_int int, c_bint bigint, c_sint smallint, c_tint tinyint,c_float float, c_double double, c_bool bool,c_binary binary(16), c_nchar nchar(32), c_ts timestamp,c_tint_un tinyint unsigned, c_sint_un smallint unsigned,c_int_un int unsigned, c_bint_un bigint unsigned) tags (t_int int);
+
+sql create table ct1 using stb1 tags ( 1 );
+sql create table ct2 using stb1 tags ( 2 );
+sql create table ct3 using stb1 tags ( 3 );
+sql create table ct4 using stb1 tags ( 4 );
+
+sql CREATE SMA INDEX sma_index_1 ON stb1 function(min(c_int), max(c_int)) interval(6m, 10s) sliding(6m) watermark 5s;
+sql CREATE SMA INDEX sma_index_2 ON stb1 function(min(c_int), max(c_int)) interval(6m, 10s) sliding(6m) max_delay 6m;
+sql CREATE SMA INDEX sma_index_3 ON stb1 function(min(c_int), max(c_int)) interval(6m, 10s) watermark 5s max_delay 6m;
+
+sql DROP INDEX sma_index_1 ;
+sql DROP INDEX sma_index_2 ;
+sql DROP INDEX sma_index_3 ;
+
+print ========== step8
+sql drop database if exists db;
+sql create database db duration 300;
+sql use db;
+sql create table stb1(ts timestamp, c_int int, c_bint bigint, c_sint smallint, c_tint tinyint,c_float float, c_double double, c_bool bool,c_binary binary(16), c_nchar nchar(32), c_ts timestamp,c_tint_un tinyint unsigned, c_sint_un smallint unsigned,c_int_un int unsigned, c_bint_un bigint unsigned) tags (t_int int);
+
+sql create table ct1 using stb1 tags ( 1 );
+sql create table ct2 using stb1 tags ( 2 );
+sql create table ct3 using stb1 tags ( 3 );
+sql create table ct4 using stb1 tags ( 4 );
+
+sql CREATE SMA INDEX sma_index_1 ON stb1 function(min(c_int), max(c_int)) interval(6m, 10s) sliding(6m) watermark 5s;
+sql CREATE SMA INDEX sma_index_2 ON stb1 function(min(c_int), max(c_int)) interval(6m, 10s) sliding(6m) max_delay 6m;
+sql CREATE SMA INDEX sma_index_3 ON stb1 function(min(c_int), max(c_int)) interval(6m, 10s) watermark 5s max_delay 6m;
+
+sql DROP INDEX sma_index_1 ;
+sql DROP INDEX sma_index_2 ;
+sql DROP INDEX sma_index_3 ;
 
 system sh/exec.sh -n dnode1 -s stop  -x SIGINT
 system sh/exec.sh -n dnode2 -s stop  -x SIGINT
 system sh/exec.sh -n dnode3 -s stop  -x SIGINT
-system sh/exec.sh -n dnode4 -s stop  -x SIGINT
-system sh/exec.sh -n dnode5 -s stop  -x SIGINT
+system sh/exec.sh -n dnode4 -s stop  -x SIGINT
\ No newline at end of file
-system sh/exec.sh -n dnode6 -s stop  -x SIGINT
-system sh/exec.sh -n dnode7 -s stop  -x SIGINT
-system sh/exec.sh -n dnode8 -s stop  -x SIGINT
\ No newline at end of file