Merge pull request #15817 from taosdata/feature/TD-11274-3.0

fix: tsma query data

include/common/tmsg.h

@@ -2657,6 +2657,34 @@ typedef struct {
   SEpSet  epSet;
 } SVgEpSet;
 
+typedef struct {
+  int64_t refId;
+  int64_t suid;
+  int8_t  level;
+} SRSmaFetchMsg;
+
+static FORCE_INLINE int32_t tEncodeSRSmaFetchMsg(SEncoder* pCoder, const SRSmaFetchMsg* pReq) {
+  if (tStartEncode(pCoder) < 0) return -1;
+
+  if (tEncodeI64(pCoder, pReq->refId) < 0) return -1;
+  if (tEncodeI64(pCoder, pReq->suid) < 0) return -1;
+  if (tEncodeI8(pCoder, pReq->level) < 0) return -1;
+
+  tEndEncode(pCoder);
+  return 0;
+}
+
+static FORCE_INLINE int32_t tDecodeSRSmaFetchMsg(SDecoder* pCoder, SRSmaFetchMsg* pReq) {
+  if (tStartDecode(pCoder) < 0) return -1;
+
+  if (tDecodeI64(pCoder, &pReq->refId) < 0) return -1;
+  if (tDecodeI64(pCoder, &pReq->suid) < 0) return -1;
+  if (tDecodeI8(pCoder, &pReq->level) < 0) return -1;
+
+  tEndDecode(pCoder);
+  return 0;
+}
+
 typedef struct {
   int8_t         version;       // for compatibility(default 0)
   int8_t         intervalUnit;  // MACRO: TIME_UNIT_XXX

include/util/tlockfree.h

@@ -25,9 +25,9 @@ extern "C" {
 // reference counting
 typedef void (*_ref_fn_t)(const void *pObj);
 
-#define T_REF_DECLARE() \
-  struct {              \
-    int32_t val;        \
+#define T_REF_DECLARE()   \
+  struct {                \
+    volatile int32_t val; \
   } _ref;
 
 #define T_REF_REGISTER_FUNC(s, e) \

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

@@ -489,7 +489,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
   smaObj.uid = mndGenerateUid(pCreate->name, TSDB_TABLE_FNAME_LEN);
   ASSERT(smaObj.uid != 0);
   char resultTbName[TSDB_TABLE_FNAME_LEN + 16] = {0};
-  snprintf(resultTbName, TSDB_TABLE_FNAME_LEN + 16, "td.tsma.rst.tb.%s", pCreate->name);
+  snprintf(resultTbName, TSDB_TABLE_FNAME_LEN + 16, "%s_td_tsma_rst_tb",pCreate->name);
   memcpy(smaObj.dstTbName, resultTbName, TSDB_TABLE_FNAME_LEN);
   smaObj.dstTbUid = mndGenerateUid(smaObj.dstTbName, TSDB_TABLE_FNAME_LEN);
   smaObj.stbUid = pStb->uid;
@@ -603,6 +603,9 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
   if (mndPersistStream(pMnode, pTrans, &streamObj) != 0) goto _OVER;
   if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
 
+  mDebug("mndSma: create sma index %s %" PRIi64 " on stb:%" PRIi64 ", dstSuid:%" PRIi64 " dstTb:%s dstVg:%d",
+         pCreate->name, smaObj.uid, smaObj.stbUid, smaObj.dstTbUid, smaObj.dstTbName, smaObj.dstVgId);
+
   code = 0;
 
 _OVER:

source/dnode/vnode/src/inc/sma.h

@@ -115,24 +115,29 @@ struct SSmaStat {
 #define RSMA_FS_LOCK(r)       (&(r)->lock)
 
 struct SRSmaInfoItem {
-  void   *taskInfo;  // qTaskInfo_t
-  int64_t refId;
-  tmr_h   tmrId;
-  int32_t maxDelay;
   int8_t  level;
   int8_t  triggerStat;
+  int32_t maxDelay;
+  tmr_h   tmrId;
 };
 
 struct SRSmaInfo {
   STSchema *pTSchema;
   int64_t   suid;
+  int64_t   refId;  // refId of SRSmaStat
   int8_t    delFlag;
   T_REF_DECLARE()
   SRSmaInfoItem items[TSDB_RETENTION_L2];
+  void         *taskInfo[TSDB_RETENTION_L2];   // qTaskInfo_t
+  void         *iTaskInfo[TSDB_RETENTION_L2];  // immutable
 };
-#define RSMA_INFO_HEAD_LEN   24
-#define RSMA_INFO_IS_DEL(r)  ((r)->delFlag == 1)
-#define RSMA_INFO_SET_DEL(r) ((r)->delFlag = 1)
+
+#define RSMA_INFO_HEAD_LEN     32
+#define RSMA_INFO_IS_DEL(r)    ((r)->delFlag == 1)
+#define RSMA_INFO_SET_DEL(r)   ((r)->delFlag = 1)
+#define RSMA_INFO_QTASK(r, i)  ((r)->taskInfo[i])
+#define RSMA_INFO_IQTASK(r, i) ((r)->iTaskInfo[i])
+#define RSMA_INFO_ITEM(r, i)   (&(r)->items[i])
 
 enum {
   TASK_TRIGGER_STAT_INIT = 0,
@@ -168,8 +173,8 @@ int32_t tdUnRefSmaStat(SSma *pSma, SSmaStat *pStat);
 int32_t tdRefRSmaInfo(SSma *pSma, SRSmaInfo *pRSmaInfo);
 int32_t tdUnRefRSmaInfo(SSma *pSma, SRSmaInfo *pRSmaInfo);
 
-void   *tdAcquireSmaRef(int32_t rsetId, int64_t refId, const char *tags, int32_t ln);
-int32_t tdReleaseSmaRef(int32_t rsetId, int64_t refId, const char *tags, int32_t ln);
+void   *tdAcquireSmaRef(int32_t rsetId, int64_t refId);
+int32_t tdReleaseSmaRef(int32_t rsetId, int64_t refId);
 
 int32_t tdCheckAndInitSmaEnv(SSma *pSma, int8_t smaType);
 
@@ -223,12 +228,11 @@ static FORCE_INLINE void tdSmaStatSetDropped(STSmaStat *pTStat) {
 
 void           tdRSmaQTaskInfoGetFileName(int32_t vid, int64_t version, char *outputName);
 void           tdRSmaQTaskInfoGetFullName(int32_t vid, int64_t version, const char *path, char *outputName);
-int32_t        tdCloneRSmaInfo(SSma *pSma, SRSmaInfo *pDest, SRSmaInfo *pSrc);
+int32_t        tdCloneRSmaInfo(SSma *pSma, SRSmaInfo **pDest, SRSmaInfo *pSrc);
 void           tdFreeQTaskInfo(qTaskInfo_t *taskHandle, int32_t vgId, int32_t level);
 static int32_t tdDestroySmaState(SSmaStat *pSmaStat, int8_t smaType);
 void          *tdFreeSmaState(SSmaStat *pSmaStat, int8_t smaType);
 void          *tdFreeRSmaInfo(SSma *pSma, SRSmaInfo *pInfo, bool isDeepFree);
-void           tdRemoveRSmaInfoBySuid(SSma *pSma, int64_t suid);
 int32_t        tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash);
 
 int32_t tdProcessRSmaCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, const char *tbName);

source/dnode/vnode/src/meta/metaTable.c

@@ -381,6 +381,8 @@ int metaCreateTable(SMeta *pMeta, int64_t version, SVCreateTbReq *pReq) {
     terrno = TSDB_CODE_TDB_TABLE_ALREADY_EXIST;
     metaReaderClear(&mr);
     return -1;
+  } else if (terrno == TSDB_CODE_PAR_TABLE_NOT_EXIST) {
+    terrno = TSDB_CODE_SUCCESS;
   }
   metaReaderClear(&mr);
 

source/dnode/vnode/src/sma/smaCommit.c

@@ -308,12 +308,12 @@ static int32_t tdProcessRSmaSyncPostCommitImpl(SSma *pSma) {
  * @return int32_t
  */
 static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma) {
-  SSmaEnv *pSmaEnv = SMA_RSMA_ENV(pSma);
-  if (!pSmaEnv) {
+  SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
+  if (!pEnv) {
     return TSDB_CODE_SUCCESS;
   }
 
-  SSmaStat  *pStat = SMA_ENV_STAT(pSmaEnv);
+  SSmaStat  *pStat = SMA_ENV_STAT(pEnv);
   SRSmaStat *pRSmaStat = SMA_RSMA_STAT(pStat);
 
   // step 1: set rsma stat
@@ -337,18 +337,26 @@ static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma) {
   }
 
   // step 3:  swap rsmaInfoHash and iRsmaInfoHash
-  ASSERT(!RSMA_IMU_INFO_HASH(pRSmaStat));
+  // lock
+  taosWLockLatch(SMA_ENV_LOCK(pEnv));
+
   ASSERT(RSMA_INFO_HASH(pRSmaStat));
+  ASSERT(!RSMA_IMU_INFO_HASH(pRSmaStat));
 
   RSMA_IMU_INFO_HASH(pRSmaStat) = RSMA_INFO_HASH(pRSmaStat);
   RSMA_INFO_HASH(pRSmaStat) =
       taosHashInit(RSMA_TASK_INFO_HASH_SLOT, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
 
   if (!RSMA_INFO_HASH(pRSmaStat)) {
+    // unlock
+    taosWUnLockLatch(SMA_ENV_LOCK(pEnv));
     smaError("vgId:%d, rsma async commit failed since %s", SMA_VID(pSma), terrstr());
     return TSDB_CODE_FAILED;
   }
 
+  // unlock
+  taosWUnLockLatch(SMA_ENV_LOCK(pEnv));
+
   // step 4: others
   pRSmaStat->commitAppliedVer = pSma->pVnode->state.applied;
 
@@ -383,26 +391,52 @@ static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma) {
  * @return int32_t
  */
 static int32_t tdProcessRSmaAsyncPostCommitImpl(SSma *pSma) {
-  SSmaEnv *pSmaEnv = SMA_RSMA_ENV(pSma);
-  if (!pSmaEnv) {
+  SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
+  if (!pEnv) {
     return TSDB_CODE_SUCCESS;
   }
 
-  SSmaStat  *pStat = SMA_ENV_STAT(pSmaEnv);
+  SSmaStat  *pStat = SMA_ENV_STAT(pEnv);
   SRSmaStat *pRSmaStat = SMA_RSMA_STAT(pStat);
 
   // step 1: merge rsmaInfoHash and iRsmaInfoHash
-  taosWLockLatch(SMA_ENV_LOCK(pSmaEnv));
-
+  // lock
+  taosWLockLatch(SMA_ENV_LOCK(pEnv));
+#if 0
   if (taosHashGetSize(RSMA_INFO_HASH(pRSmaStat)) <= 0) {
-    // TODO: optimization - just switch the hash pointer if rsmaInfoHash is empty
-  }
-
+    // just switch the hash pointer if rsmaInfoHash is empty
+    if (taosHashGetSize(RSMA_IMU_INFO_HASH(pRSmaStat)) > 0) {
+      SHashObj *infoHash = RSMA_INFO_HASH(pRSmaStat);
+      RSMA_INFO_HASH(pRSmaStat) = RSMA_IMU_INFO_HASH(pRSmaStat);
+      RSMA_IMU_INFO_HASH(pRSmaStat) = infoHash;
+    }
+  } else {
+#endif
+#if 1
   void *pIter = taosHashIterate(RSMA_IMU_INFO_HASH(pRSmaStat), NULL);
   while (pIter) {
     tb_uid_t *pSuid = (tb_uid_t *)taosHashGetKey(pIter, NULL);
 
     if (!taosHashGet(RSMA_INFO_HASH(pRSmaStat), pSuid, sizeof(tb_uid_t))) {
+      SRSmaInfo *pRSmaInfo = *(SRSmaInfo **)pIter;
+      if (RSMA_INFO_IS_DEL(pRSmaInfo)) {
+        int32_t refVal = T_REF_VAL_GET(pRSmaInfo);
+        if (refVal == 0) {
+          tdFreeRSmaInfo(pSma, pRSmaInfo, true);
+          smaDebug(
+              "vgId:%d, rsma async post commit, free rsma info since already deleted and ref is 0 for "
+              "table:%" PRIi64,
+              SMA_VID(pSma), *pSuid);
+        } else {
+          smaDebug(
+              "vgId:%d, rsma async post commit, not free rsma info since ref is %d although already deleted for "
+              "table:%" PRIi64,
+              SMA_VID(pSma), refVal, *pSuid);
+        }
+
+        pIter = taosHashIterate(RSMA_IMU_INFO_HASH(pRSmaStat), pIter);
+        continue;
+      }
       taosHashPut(RSMA_INFO_HASH(pRSmaStat), pSuid, sizeof(tb_uid_t), pIter, sizeof(pIter));
       smaDebug("vgId:%d, rsma async post commit, migrated from iRsmaInfoHash for table:%" PRIi64, SMA_VID(pSma),
                *pSuid);
@@ -416,11 +450,14 @@ static int32_t tdProcessRSmaAsyncPostCommitImpl(SSma *pSma) {
 
     pIter = taosHashIterate(RSMA_IMU_INFO_HASH(pRSmaStat), pIter);
   }
+#endif
+  // }
 
   taosHashCleanup(RSMA_IMU_INFO_HASH(pRSmaStat));
   RSMA_IMU_INFO_HASH(pRSmaStat) = NULL;
 
-  taosWUnLockLatch(SMA_ENV_LOCK(pSmaEnv));
+  // unlock
+  taosWUnLockLatch(SMA_ENV_LOCK(pEnv));
 
   // step 2: cleanup outdated qtaskinfo files
   tdCleanupQTaskInfoFiles(pSma, pRSmaStat);

source/dnode/vnode/src/sma/smaEnv.c

@@ -17,7 +17,7 @@
 
 typedef struct SSmaStat SSmaStat;
 
-#define SMA_MGMT_REF_NUM         10240
+#define SMA_MGMT_REF_NUM 10240
 
 extern SSmaMgmt smaMgmt;
 
@@ -171,7 +171,7 @@ int32_t tdUnRefSmaStat(SSma *pSma, SSmaStat *pStat) {
 
 int32_t tdRefRSmaInfo(SSma *pSma, SRSmaInfo *pRSmaInfo) {
   if (!pRSmaInfo) return 0;
-
+  
   int ref = T_REF_INC(pRSmaInfo);
   smaDebug("vgId:%d, ref rsma info:%p, val:%d", SMA_VID(pSma), pRSmaInfo, ref);
   return 0;
@@ -183,9 +183,6 @@ int32_t tdUnRefRSmaInfo(SSma *pSma, SRSmaInfo *pRSmaInfo) {
   int ref = T_REF_DEC(pRSmaInfo);
   smaDebug("vgId:%d, unref rsma info:%p, val:%d", SMA_VID(pSma), pRSmaInfo, ref);
 
-  if (ref == 0) {
-    tdRemoveRSmaInfoBySuid(pSma, pRSmaInfo->suid);
-  }
   return 0;
 }
 

source/dnode/vnode/src/sma/smaTimeRange.c

@@ -116,8 +116,10 @@ int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg) {
     }
 
     // create stable to save tsma result in dstVgId
+    SName stbFullName = {0};
+    tNameFromString(&stbFullName, pCfg->dstTbName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
     SVCreateStbReq pReq = {0};
-    pReq.name = pCfg->dstTbName;
+    pReq.name = (char*)tNameGetTableName(&stbFullName);
     pReq.suid = pCfg->dstTbUid;
     pReq.schemaRow = pCfg->schemaRow;
     pReq.schemaTag = pCfg->schemaTag;
@@ -125,6 +127,12 @@ int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg) {
     if (metaCreateSTable(SMA_META(pSma), version, &pReq) < 0) {
       return -1;
     }
+
+    smaDebug("vgId:%d, success to create sma index %s %" PRIi64 " on stb:%" PRIi64 ", dstSuid:%" PRIi64
+             " dstTb:%s dstVg:%d",
+             SMA_VID(pSma), pCfg->indexName, pCfg->indexUid, pCfg->tableUid, pCfg->dstTbUid, pReq.name, pCfg->dstVgId);
+  } else {
+    ASSERT(0);
   }
 
   return 0;

source/dnode/vnode/src/sma/smaUtil.c

@@ -287,22 +287,22 @@ int32_t tdRemoveTFile(STFile *pTFile) {
 }
 
 // smaXXXUtil ================
-void *tdAcquireSmaRef(int32_t rsetId, int64_t refId, const char *tags, int32_t ln) {
+void *tdAcquireSmaRef(int32_t rsetId, int64_t refId) {
   void *pResult = taosAcquireRef(rsetId, refId);
   if (!pResult) {
-    smaWarn("%s:%d taosAcquireRef for rsetId:%" PRIi64 " refId:%d failed since %s", tags, ln, rsetId, refId, terrstr());
+    smaWarn("rsma acquire ref for rsetId:%" PRIi64 " refId:%d failed since %s", rsetId, refId, terrstr());
   } else {
-    smaDebug("%s:%d taosAcquireRef for rsetId:%" PRIi64 " refId:%d success", tags, ln, rsetId, refId);
+    smaDebug("rsma acquire ref for rsetId:%" PRIi64 " refId:%d success", rsetId, refId);
   }
   return pResult;
 }
 
-int32_t tdReleaseSmaRef(int32_t rsetId, int64_t refId, const char *tags, int32_t ln) {
+int32_t tdReleaseSmaRef(int32_t rsetId, int64_t refId) {
   if (taosReleaseRef(rsetId, refId) < 0) {
-    smaWarn("%s:%d taosReleaseRef for rsetId:%" PRIi64 " refId:%d failed since %s", tags, ln, rsetId, refId, terrstr());
+    smaWarn("rsma release ref for rsetId:%" PRIi64 " refId:%d failed since %s", rsetId, refId, terrstr());
     return TSDB_CODE_FAILED;
   }
-  smaDebug("%s:%d taosReleaseRef for rsetId:%" PRIi64 " refId:%d success", tags, ln, rsetId, refId);
+  smaDebug("rsma release ref for rsetId:%" PRIi64 " refId:%d success", rsetId, refId);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -313,7 +313,7 @@ static int32_t tdCloneQTaskInfo(SSma *pSma, qTaskInfo_t dstTaskInfo, qTaskInfo_t
   char   *pOutput = NULL;
   int32_t len = 0;
 
-  if (qSerializeTaskStatus(srcTaskInfo, &pOutput, &len) < 0) {
+  if ((terrno = qSerializeTaskStatus(srcTaskInfo, &pOutput, &len)) < 0) {
     smaError("vgId:%d, rsma clone, table %" PRIi64 " serialize qTaskInfo failed since %s", TD_VID(pVnode), suid,
              terrstr());
     goto _err;
@@ -337,41 +337,34 @@ static int32_t tdCloneQTaskInfo(SSma *pSma, qTaskInfo_t dstTaskInfo, qTaskInfo_t
     goto _err;
   }
 
-  smaError("vgId:%d, rsma clone, restore rsma task for table:%" PRIi64 " succeed", TD_VID(pVnode), suid);
+  smaDebug("vgId:%d, rsma clone, restore rsma task for table:%" PRIi64 " succeed", TD_VID(pVnode), suid);
 
   taosMemoryFreeClear(pOutput);
   return TSDB_CODE_SUCCESS;
 _err:
   taosMemoryFreeClear(pOutput);
   tdFreeQTaskInfo(dstTaskInfo, TD_VID(pVnode), idx + 1);
+  smaError("vgId:%d, rsma clone, restore rsma task for table:%" PRIi64 " failed since %s", TD_VID(pVnode), suid,
+           terrstr());
   return TSDB_CODE_FAILED;
 }
 
 /**
  * @brief pTSchema is shared
- * 
- * @param pSma 
- * @param pDest 
- * @param pSrc 
- * @return int32_t 
+ *
+ * @param pSma
+ * @param pDest
+ * @param pSrc
+ * @return int32_t
  */
-int32_t tdCloneRSmaInfo(SSma *pSma, SRSmaInfo *pDest, SRSmaInfo *pSrc) {
+int32_t tdCloneRSmaInfo(SSma *pSma, SRSmaInfo **pDest, SRSmaInfo *pSrc) {
   SVnode     *pVnode = pSma->pVnode;
   SRSmaParam *param = NULL;
   if (!pSrc) {
+    *pDest = NULL;
     return TSDB_CODE_SUCCESS;
   }
 
-  if (!pDest) {
-    pDest = taosMemoryCalloc(1, sizeof(SRSmaInfo));
-    if (!pDest) {
-      terrno = TSDB_CODE_OUT_OF_MEMORY;
-      return TSDB_CODE_FAILED;
-    }
-  }
-
-  memcpy(pDest, pSrc, sizeof(SRSmaInfo));
-
   SMetaReader mr = {0};
   metaReaderInit(&mr, SMA_META(pSma), 0);
   smaDebug("vgId:%d, rsma clone, suid is %" PRIi64, TD_VID(pVnode), pSrc->suid);
@@ -384,21 +377,22 @@ int32_t tdCloneRSmaInfo(SSma *pSma, SRSmaInfo *pDest, SRSmaInfo *pSrc) {
   ASSERT(mr.me.uid == pSrc->suid);
   if (TABLE_IS_ROLLUP(mr.me.flags)) {
     param = &mr.me.stbEntry.rsmaParam;
-    for (int i = 0; i < TSDB_RETENTION_L2; ++i) {
-      SRSmaInfoItem *pItem = &pSrc->items[i];
-      if (pItem->taskInfo) {
-        tdCloneQTaskInfo(pSma, pDest->items[i].taskInfo, pItem->taskInfo, param, pSrc->suid, i);
+    for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
+      if (tdCloneQTaskInfo(pSma, pSrc->iTaskInfo[i], pSrc->taskInfo[i], param, pSrc->suid, i) < 0) {
+        goto _err;
       }
     }
     smaDebug("vgId:%d, rsma clone env success for %" PRIi64, TD_VID(pVnode), pSrc->suid);
   }
 
   metaReaderClear(&mr);
+
+  *pDest = pSrc;  // pointer copy
+
   return TSDB_CODE_SUCCESS;
 _err:
+  *pDest = NULL;
   metaReaderClear(&mr);
-  tdFreeRSmaInfo(pSma, pDest, false);
+  smaError("vgId:%d, rsma clone env failed for %" PRIi64 " since %s", TD_VID(pVnode), pSrc->suid, terrstr());
   return TSDB_CODE_FAILED;
-}
-
-// ...
\ No newline at end of file
+}
\ No newline at end of file

source/dnode/vnode/src/vnd/vnodeSvr.c

@@ -351,7 +351,6 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
 // TODO: remove the function
 void smaHandleRes(void *pVnode, int64_t smaId, const SArray *data) {
   // TODO
-
   blockDebugShowDataBlocks(data, __func__);
   tdProcessTSmaInsert(((SVnode *)pVnode)->pSma, smaId, (const char *)data);
 }
@@ -852,6 +851,7 @@ static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq
       if (metaCreateTable(pVnode->pMeta, version, &createTbReq) < 0) {
         if (terrno != TSDB_CODE_TDB_TABLE_ALREADY_EXIST) {
           submitBlkRsp.code = terrno;
+          pRsp->code = terrno;
           tDecoderClear(&decoder);
           taosArrayDestroy(createTbReq.ctb.tagName);
           goto _exit;