Merge branch '3.0' of https://github.com/taosdata/TDengine into feat/stream_compression

include/client/taos.h

@@ -254,7 +254,7 @@ enum tmq_res_t {
   TMQ_RES_INVALID = -1,
   TMQ_RES_DATA = 1,
   TMQ_RES_TABLE_META = 2,
-  TMQ_RES_TAOSX = 3,
+  TMQ_RES_METADATA = 3,
 };
 
 typedef struct tmq_raw_data {

source/client/inc/clientInt.h

@@ -52,7 +52,7 @@ enum {
   RES_TYPE__QUERY = 1,
   RES_TYPE__TMQ,
   RES_TYPE__TMQ_META,
-  RES_TYPE__TAOSX,
+  RES_TYPE__TMQ_METADATA,
 };
 
 #define SHOW_VARIABLES_RESULT_COLS       2
@@ -60,9 +60,9 @@ enum {
 #define SHOW_VARIABLES_RESULT_FIELD2_LEN (TSDB_CONFIG_VALUE_LEN + VARSTR_HEADER_SIZE)
 
 #define TD_RES_QUERY(res)     (*(int8_t*)res == RES_TYPE__QUERY)
-#define TD_RES_TMQ(res)       (*(int8_t*)res == RES_TYPE__TMQ || *(int8_t*)res == RES_TYPE__TAOSX)
+#define TD_RES_TMQ(res)       (*(int8_t*)res == RES_TYPE__TMQ)
 #define TD_RES_TMQ_META(res)  (*(int8_t*)res == RES_TYPE__TMQ_META)
-#define TD_RES_TMQ_TAOSX(res) (*(int8_t*)res == RES_TYPE__TAOSX)
+#define TD_RES_TMQ_METADATA(res) (*(int8_t*)res == RES_TYPE__TMQ_METADATA)
 
 typedef struct SAppInstInfo SAppInstInfo;
 

source/client/src/clientImpl.c

@@ -189,6 +189,7 @@ int32_t buildRequest(uint64_t connId, const char* sql, int sqlLen, void* param,
     tscError("%d failed to add to request container, reqId:0x%" PRIx64 ", conn:%d, %s", (*pRequest)->self,
              (*pRequest)->requestId, pTscObj->id, sql);
 
+    taosMemoryFree(param);
     destroyRequest(*pRequest);
     *pRequest = NULL;
     return TSDB_CODE_TSC_OUT_OF_MEMORY;

source/client/src/clientMain.c

@@ -148,7 +148,7 @@ int taos_errno(TAOS_RES *res) {
     return terrno;
   }
 
-  if (TD_RES_TMQ(res)) {
+  if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
     return 0;
   }
 
@@ -162,7 +162,7 @@ const char *taos_errstr(TAOS_RES *res) {
     return (const char *)tstrerror(terrno);
   }
 
-  if (TD_RES_TMQ(res)) {
+  if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
     return "success";
   }
 
@@ -184,7 +184,7 @@ void taos_free_result(TAOS_RES *res) {
     SRequestObj *pRequest = (SRequestObj *)res;
     tscDebug("0x%" PRIx64 " taos_free_result start to free query", pRequest->requestId);
     destroyRequest(pRequest);
-  } else if (TD_RES_TMQ_TAOSX(res)) {
+  } else if (TD_RES_TMQ_METADATA(res)) {
     SMqTaosxRspObj *pRsp = (SMqTaosxRspObj *)res;
     if (pRsp->rsp.blockData) taosArrayDestroyP(pRsp->rsp.blockData, taosMemoryFree);
     if (pRsp->rsp.blockDataLen) taosArrayDestroy(pRsp->rsp.blockDataLen);
@@ -264,7 +264,7 @@ TAOS_ROW taos_fetch_row(TAOS_RES *res) {
     return doFetchRows(pRequest, true, true);
 #endif
 
-  } else if (TD_RES_TMQ(res)) {
+  } else if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
     SMqRspObj      *msg = ((SMqRspObj *)res);
     SReqResultInfo *pResultInfo;
     if (msg->resIter == -1) {
@@ -437,7 +437,7 @@ const char *taos_data_type(int type) {
 const char *taos_get_client_info() { return version; }
 
 int taos_affected_rows(TAOS_RES *res) {
-  if (res == NULL || TD_RES_TMQ(res) || TD_RES_TMQ_META(res)) {
+  if (res == NULL || TD_RES_TMQ(res) || TD_RES_TMQ_META(res) || TD_RES_TMQ_METADATA(res)) {
     return 0;
   }
 
@@ -454,7 +454,7 @@ int taos_result_precision(TAOS_RES *res) {
   if (TD_RES_QUERY(res)) {
     SRequestObj *pRequest = (SRequestObj *)res;
     return pRequest->body.resInfo.precision;
-  } else if (TD_RES_TMQ(res)) {
+  } else if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
     SReqResultInfo *info = tmqGetCurResInfo(res);
     return info->precision;
   }
@@ -487,7 +487,7 @@ int taos_select_db(TAOS *taos, const char *db) {
 }
 
 void taos_stop_query(TAOS_RES *res) {
-  if (res == NULL || TD_RES_TMQ(res) || TD_RES_TMQ_META(res)) {
+  if (res == NULL || TD_RES_TMQ(res) || TD_RES_TMQ_META(res) || TD_RES_TMQ_METADATA(res)) {
     return;
   }
 
@@ -559,7 +559,7 @@ int taos_fetch_block_s(TAOS_RES *res, int *numOfRows, TAOS_ROW *rows) {
     (*rows) = pResultInfo->row;
     (*numOfRows) = pResultInfo->numOfRows;
     return pRequest->code;
-  } else if (TD_RES_TMQ(res)) {
+  } else if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
     SReqResultInfo *pResultInfo = tmqGetNextResInfo(res, true);
     if (pResultInfo == NULL) return -1;
 
@@ -578,7 +578,7 @@ int taos_fetch_raw_block(TAOS_RES *res, int *numOfRows, void **pData) {
     return 0;
   }
 
-  if (TD_RES_TMQ(res)) {
+  if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
     SReqResultInfo *pResultInfo = tmqGetNextResInfo(res, false);
     if (pResultInfo == NULL) {
       (*numOfRows) = 0;

source/client/src/clientTmq.c

@@ -515,6 +515,10 @@ int32_t tmqCommitMsgImpl(tmq_t* tmq, const TAOS_RES* msg, int8_t async, tmq_comm
     SMqMetaRspObj* pMetaRspObj = (SMqMetaRspObj*)msg;
     topic = pMetaRspObj->topic;
     vgId = pMetaRspObj->vgId;
+  } else if(TD_RES_TMQ_METADATA(msg)) {
+    SMqTaosxRspObj* pRspObj = (SMqTaosxRspObj*)msg;
+    topic = pRspObj->topic;
+    vgId = pRspObj->vgId;
   } else {
     return TSDB_CODE_TMQ_INVALID_MSG;
   }
@@ -1471,16 +1475,16 @@ SMqRspObj* tmqBuildRspFromWrapper(SMqPollRspWrapper* pWrapper) {
 
 SMqTaosxRspObj* tmqBuildTaosxRspFromWrapper(SMqPollRspWrapper* pWrapper) {
   SMqTaosxRspObj* pRspObj = taosMemoryCalloc(1, sizeof(SMqTaosxRspObj));
-  pRspObj->resType = RES_TYPE__TAOSX;
+  pRspObj->resType = RES_TYPE__TMQ_METADATA;
   tstrncpy(pRspObj->topic, pWrapper->topicHandle->topicName, TSDB_TOPIC_FNAME_LEN);
   tstrncpy(pRspObj->db, pWrapper->topicHandle->db, TSDB_DB_FNAME_LEN);
   pRspObj->vgId = pWrapper->vgHandle->vgId;
   pRspObj->resIter = -1;
-  memcpy(&pRspObj->rsp, &pWrapper->dataRsp, sizeof(SMqTaosxRspObj));
+  memcpy(&pRspObj->rsp, &pWrapper->taosxRsp, sizeof(STaosxRsp));
 
   pRspObj->resInfo.totalRows = 0;
   pRspObj->resInfo.precision = TSDB_TIME_PRECISION_MILLI;
-  if (!pWrapper->dataRsp.withSchema) {
+  if (!pWrapper->taosxRsp.withSchema) {
     setResSchemaInfo(&pRspObj->resInfo, pWrapper->topicHandle->schema.pSchema, pWrapper->topicHandle->schema.nCols);
   }
 
@@ -1654,8 +1658,14 @@ void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
           rspWrapper = NULL;
           continue;
         }
+
         // build rsp
-        SMqRspObj* pRsp = tmqBuildRspFromWrapper(pollRspWrapper);
+        void* pRsp = NULL;
+        if(pollRspWrapper->taosxRsp.createTableNum == 0){
+          pRsp = tmqBuildRspFromWrapper(pollRspWrapper);
+        }else{
+          pRsp = tmqBuildTaosxRspFromWrapper(pollRspWrapper);
+        }
         taosFreeQitem(pollRspWrapper);
         return pRsp;
       } else {
@@ -1775,11 +1785,11 @@ tmq_res_t tmq_get_res_type(TAOS_RES* res) {
   } else if (TD_RES_TMQ_META(res)) {
     SMqMetaRspObj* pMetaRspObj = (SMqMetaRspObj*)res;
     if (pMetaRspObj->metaRsp.resMsgType == TDMT_VND_DELETE) {
-      return TMQ_RES_TAOSX;
+      return TMQ_RES_DATA;
     }
     return TMQ_RES_TABLE_META;
-  } else if (TD_RES_TMQ_TAOSX(res)) {
-    return TMQ_RES_DATA;
+  } else if (TD_RES_TMQ_METADATA(res)) {
+    return TMQ_RES_METADATA;
   } else {
     return TMQ_RES_INVALID;
   }
@@ -1792,6 +1802,9 @@ const char* tmq_get_topic_name(TAOS_RES* res) {
   } else if (TD_RES_TMQ_META(res)) {
     SMqMetaRspObj* pMetaRspObj = (SMqMetaRspObj*)res;
     return strchr(pMetaRspObj->topic, '.') + 1;
+  } else if (TD_RES_TMQ_METADATA(res)) {
+    SMqTaosxRspObj* pRspObj = (SMqTaosxRspObj*)res;
+    return strchr(pRspObj->topic, '.') + 1;
   } else {
     return NULL;
   }
@@ -1804,6 +1817,9 @@ const char* tmq_get_db_name(TAOS_RES* res) {
   } else if (TD_RES_TMQ_META(res)) {
     SMqMetaRspObj* pMetaRspObj = (SMqMetaRspObj*)res;
     return strchr(pMetaRspObj->db, '.') + 1;
+  } else if (TD_RES_TMQ_METADATA(res)) {
+    SMqTaosxRspObj* pRspObj = (SMqTaosxRspObj*)res;
+    return strchr(pRspObj->db, '.') + 1;
   } else {
     return NULL;
   }
@@ -1816,6 +1832,9 @@ int32_t tmq_get_vgroup_id(TAOS_RES* res) {
   } else if (TD_RES_TMQ_META(res)) {
     SMqMetaRspObj* pMetaRspObj = (SMqMetaRspObj*)res;
     return pMetaRspObj->vgId;
+  } else if (TD_RES_TMQ_METADATA(res)) {
+    SMqTaosxRspObj* pRspObj = (SMqTaosxRspObj*)res;
+    return pRspObj->vgId;
   } else {
     return -1;
   }
@@ -1829,7 +1848,14 @@ const char* tmq_get_table_name(TAOS_RES* res) {
       return NULL;
     }
     return (const char*)taosArrayGetP(pRspObj->rsp.blockTbName, pRspObj->resIter);
-  }
+  } else if (TD_RES_TMQ_METADATA(res)) {
+    SMqTaosxRspObj* pRspObj = (SMqTaosxRspObj*)res;
+      if (!pRspObj->rsp.withTbName || pRspObj->rsp.blockTbName == NULL || pRspObj->resIter < 0 ||
+          pRspObj->resIter >= pRspObj->rsp.blockNum) {
+        return NULL;
+      }
+      return (const char*)taosArrayGetP(pRspObj->rsp.blockTbName, pRspObj->resIter);
+    }
   return NULL;
 }
 

source/common/src/tmsg.c

@@ -6018,12 +6018,18 @@ int32_t tDecodeSTaosxRsp(SDecoder *pDecoder, STaosxRsp *pRsp) {
 
 void tDeleteSTaosxRsp(STaosxRsp *pRsp) {
   taosArrayDestroy(pRsp->blockDataLen);
+  pRsp->blockDataLen = NULL;
   taosArrayDestroyP(pRsp->blockData, (FDelete)taosMemoryFree);
+  pRsp->blockData = NULL;
   taosArrayDestroyP(pRsp->blockSchema, (FDelete)tDeleteSSchemaWrapper);
+  pRsp->blockSchema = NULL;
   taosArrayDestroyP(pRsp->blockTbName, (FDelete)taosMemoryFree);
+  pRsp->blockTbName = NULL;
 
   taosArrayDestroy(pRsp->createTableLen);
+  pRsp->createTableLen = NULL;
   taosArrayDestroyP(pRsp->createTableReq, (FDelete)taosMemoryFree);
+  pRsp->createTableReq = NULL;
 }
 
 int32_t tEncodeSSingleDeleteReq(SEncoder *pEncoder, const SSingleDeleteReq *pReq) {

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

@@ -540,7 +540,7 @@ typedef struct {
 } SMqConsumerEp;
 
 SMqConsumerEp* tCloneSMqConsumerEp(const SMqConsumerEp* pEp);
-void           tDeleteSMqConsumerEp(SMqConsumerEp* pEp);
+void           tDeleteSMqConsumerEp(void* pEp);
 int32_t        tEncodeSMqConsumerEp(void** buf, const SMqConsumerEp* pEp);
 void*          tDecodeSMqConsumerEp(const void* buf, SMqConsumerEp* pEp);
 

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

@@ -197,11 +197,12 @@ static int32_t mndProcessMqTimerMsg(SRpcMsg *pMsg) {
       SMqConsumerLostMsg *pLostMsg = rpcMallocCont(sizeof(SMqConsumerLostMsg));
 
       pLostMsg->consumerId = pConsumer->consumerId;
-      SRpcMsg *pRpcMsg = taosMemoryCalloc(1, sizeof(SRpcMsg));
-      pRpcMsg->msgType = TDMT_MND_MQ_CONSUMER_LOST;
-      pRpcMsg->pCont = pLostMsg;
-      pRpcMsg->contLen = sizeof(SMqConsumerLostMsg);
-      tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, pRpcMsg);
+      SRpcMsg pRpcMsg = {
+          .msgType = TDMT_MND_MQ_CONSUMER_LOST,
+          .pCont = pLostMsg,
+          .contLen = sizeof(SMqConsumerLostMsg),
+      };
+      tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &pRpcMsg);
     }
     if (status == MQ_CONSUMER_STATUS__LOST_REBD || status == MQ_CONSUMER_STATUS__READY) {
       // do nothing
@@ -280,11 +281,12 @@ static int32_t mndProcessMqHbReq(SRpcMsg *pMsg) {
     SMqConsumerRecoverMsg *pRecoverMsg = rpcMallocCont(sizeof(SMqConsumerRecoverMsg));
 
     pRecoverMsg->consumerId = consumerId;
-    SRpcMsg *pRpcMsg = taosMemoryCalloc(1, sizeof(SRpcMsg));
-    pRpcMsg->msgType = TDMT_MND_MQ_CONSUMER_RECOVER;
-    pRpcMsg->pCont = pRecoverMsg;
-    pRpcMsg->contLen = sizeof(SMqConsumerRecoverMsg);
-    tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, pRpcMsg);
+    SRpcMsg pRpcMsg = {
+        .msgType = TDMT_MND_MQ_CONSUMER_RECOVER,
+        .pCont = pRecoverMsg,
+        .contLen = sizeof(SMqConsumerRecoverMsg),
+    };
+    tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &pRpcMsg);
   }
 
   mndReleaseConsumer(pMnode, pConsumer);
@@ -318,11 +320,12 @@ static int32_t mndProcessAskEpReq(SRpcMsg *pMsg) {
     SMqConsumerRecoverMsg *pRecoverMsg = rpcMallocCont(sizeof(SMqConsumerRecoverMsg));
 
     pRecoverMsg->consumerId = consumerId;
-    SRpcMsg *pRpcMsg = taosMemoryCalloc(1, sizeof(SRpcMsg));
-    pRpcMsg->msgType = TDMT_MND_MQ_CONSUMER_RECOVER;
-    pRpcMsg->pCont = pRecoverMsg;
-    pRpcMsg->contLen = sizeof(SMqConsumerRecoverMsg);
-    tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, pRpcMsg);
+    SRpcMsg pRpcMsg = {
+        .msgType = TDMT_MND_MQ_CONSUMER_RECOVER,
+        .pCont = pRecoverMsg,
+        .contLen = sizeof(SMqConsumerRecoverMsg),
+    };
+    tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &pRpcMsg);
   }
 #endif
 

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

@@ -338,8 +338,8 @@ SMqConsumerEp *tCloneSMqConsumerEp(const SMqConsumerEp *pConsumerEpOld) {
   return pConsumerEpNew;
 }
 
-void tDeleteSMqConsumerEp(SMqConsumerEp *pConsumerEp) {
-  //
+void tDeleteSMqConsumerEp(void *data) {
+  SMqConsumerEp *pConsumerEp = (SMqConsumerEp*)data;
   taosArrayDestroyP(pConsumerEp->vgs, (FDelete)tDeleteSMqVgEp);
 }
 

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

@@ -184,7 +184,7 @@ int32_t tqProcessTaskRecoverRsp(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskRetrieveReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskRetrieveRsp(STQ* pTq, SRpcMsg* pMsg);
 
-SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchema* pSchema, bool createTb, int64_t suid,
+SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchema* pSchema, SSchemaWrapper* pTagSchemaWrapper, bool createTb, int64_t suid,
                             const char* stbFullName, SBatchDeleteReq* pDeleteReq);
 
 // sma

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

@@ -319,8 +319,12 @@ _query:
         pSchema = tCloneSSchemaWrapper(&meNew.stbEntry.schemaRow);
         tDecoderClear(&dcNew);
         tdbTbcClose(pCur);
+        tdbFree(pKey);
+        tdbFree(pVal);
         goto _exit;
       }
+      tdbFree(pKey);
+      tdbFree(pVal);
       tdbTbcClose(pCur);
     }
   } else if (me.type == TSDB_CHILD_TABLE) {
@@ -347,11 +351,13 @@ _query:
   tDecoderClear(&dc);
 
 _exit:
+  tDecoderClear(&dc);
   metaULock(pMeta);
   tdbFree(pData);
   return pSchema;
 
 _err:
+  tDecoderClear(&dc);
   metaULock(pMeta);
   tdbFree(pData);
   return NULL;
@@ -383,10 +389,8 @@ int metaTtlSmaller(SMeta *pMeta, uint64_t ttl, SArray *uidList) {
     ttlKey = *(STtlIdxKey *)pKey;
     taosArrayPush(uidList, &ttlKey.uid);
   }
-  tdbTbcClose(pCur);
-
   tdbFree(pKey);
-
+  tdbTbcClose(pCur);
   return 0;
 }
 

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

@@ -353,6 +353,8 @@ int32_t buildSnapContext(SMeta* pMeta, int64_t snapVersion, int64_t suid, int8_t
     metaDebug("tmqsnap init idVersion uid:%" PRIi64 " version:%" PRIi64 " index:%d", *uid, idData->version, idData->index);
   }
 
+  tdbFree(pKey);
+  tdbFree(pVal);
   return TDB_CODE_SUCCESS;
 }
 
@@ -528,6 +530,7 @@ int32_t getMetafromSnapShot(SSnapContext* ctx, void **pBuf, int32_t *contLen, in
           }
         }
       }
+      taosArrayDestroy(pTagVals);
     }
 //    SIdInfo* sidInfo = (SIdInfo*)taosHashGet(ctx->idVersion, &me.ctbEntry.suid, sizeof(tb_uid_t));
 //    if(sidInfo->version >= idInfo->version){

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

@@ -1192,10 +1192,11 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) {
   const void    *pTagData = NULL;  //
   int32_t        nTagData = 0;
   SDecoder       dc = {0};
-
+  int32_t        ret = 0;
   // get super table
   if (tdbTbGet(pMeta->pUidIdx, &pCtbEntry->ctbEntry.suid, sizeof(tb_uid_t), &pData, &nData) != 0) {
-    return -1;
+    ret = -1;
+    goto end;
   }
   tbDbKey.uid = pCtbEntry->ctbEntry.suid;
   tbDbKey.version = ((SUidIdxVal *)pData)[0].version;
@@ -1221,17 +1222,20 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) {
     // nTagData = ((const STag *)pCtbEntry->ctbEntry.pTags)->len;
     pTagData = pCtbEntry->ctbEntry.pTags;
     nTagData = ((const STag *)pCtbEntry->ctbEntry.pTags)->len;
-    return metaSaveJsonVarToIdx(pMeta, pCtbEntry, pTagColumn);
+    ret = metaSaveJsonVarToIdx(pMeta, pCtbEntry, pTagColumn);
+    goto end;
   }
   if (metaCreateTagIdxKey(pCtbEntry->ctbEntry.suid, pTagColumn->colId, pTagData, nTagData, pTagColumn->type,
                           pCtbEntry->uid, &pTagIdxKey, &nTagIdxKey) < 0) {
-    return -1;
+    ret = -1;
+    goto end;
   }
   tdbTbUpsert(pMeta->pTagIdx, pTagIdxKey, nTagIdxKey, NULL, 0, &pMeta->txn);
+end:
   metaDestroyTagIdxKey(pTagIdxKey);
   tDecoderClear(&dc);
   tdbFree(pData);
-  return 0;
+  return ret;
 }
 
 static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) {

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

@@ -204,7 +204,7 @@ static int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char
   }
 
   SBatchDeleteReq deleteReq;
-  SSubmitReq     *pSubmitReq = tqBlockToSubmit(pSma->pVnode, (const SArray *)msg, pTsmaStat->pTSchema, true,
+  SSubmitReq     *pSubmitReq = tqBlockToSubmit(pSma->pVnode, (const SArray *)msg, pTsmaStat->pTSchema, &pTsmaStat->pTSma->schemaTag, true,
                                                pTsmaStat->pTSma->dstTbUid, pTsmaStat->pTSma->dstTbName, &deleteReq);
 
   if (!pSubmitReq) {

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

@@ -51,6 +51,20 @@ void tqCleanUp() {
   }
 }
 
+static void destroySTqHandle(void* data) {
+  STqHandle* pData = (STqHandle*)data;
+  qDestroyTask(pData->execHandle.task);
+  if (pData->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
+  } else if (pData->execHandle.subType == TOPIC_SUB_TYPE__DB) {
+    tqCloseReader(pData->execHandle.pExecReader);
+    walCloseReader(pData->pWalReader);
+    taosHashCleanup(pData->execHandle.execDb.pFilterOutTbUid);
+  } else if (pData->execHandle.subType == TOPIC_SUB_TYPE__TABLE){
+    walCloseReader(pData->pWalReader);
+    tqCloseReader(pData->execHandle.pExecReader);
+  }
+}
+
 STQ* tqOpen(const char* path, SVnode* pVnode) {
   STQ* pTq = taosMemoryCalloc(1, sizeof(STQ));
   if (pTq == NULL) {
@@ -62,6 +76,8 @@ STQ* tqOpen(const char* path, SVnode* pVnode) {
 
   pTq->pHandle = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK);
 
+  taosHashSetFreeFp(pTq->pHandle, destroySTqHandle);
+
   pTq->pPushMgr = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
 
   pTq->pCheckInfo = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK);
@@ -517,6 +533,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
     int64_t fetchVer = fetchOffsetNew.version + 1;
     pCkHead = taosMemoryMalloc(sizeof(SWalCkHead) + 2048);
     if (pCkHead == NULL) {
+      tDeleteSTaosxRsp(&taosxRsp);
       return -1;
     }
 
@@ -577,14 +594,17 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
         if (tqSendMetaPollRsp(pTq, pMsg, pReq, &metaRsp) < 0) {
           code = -1;
           taosMemoryFree(pCkHead);
+          tDeleteSTaosxRsp(&taosxRsp);
           return code;
         }
         code = 0;
         if (pCkHead) taosMemoryFree(pCkHead);
+        tDeleteSTaosxRsp(&taosxRsp);
         return code;
       }
     }
   }
+  tDeleteSTaosxRsp(&taosxRsp);
   return 0;
 }
 

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

@@ -243,14 +243,15 @@ int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, STaosxRsp
       }
       if (pHandle->fetchMeta) {
         SSubmitBlk* pBlk = pReader->pBlock;
-        if (pBlk->schemaLen > 0) {
+        int32_t schemaLen = htonl(pBlk->schemaLen);
+        if (schemaLen > 0) {
           if (pRsp->createTableNum == 0) {
             pRsp->createTableLen = taosArrayInit(0, sizeof(int32_t));
             pRsp->createTableReq = taosArrayInit(0, sizeof(void*));
           }
-          void* createReq = taosMemoryCalloc(1, pBlk->schemaLen);
-          memcpy(createReq, pBlk->data, pBlk->schemaLen);
-          taosArrayPush(pRsp->createTableLen, &pBlk->schemaLen);
+          void* createReq = taosMemoryCalloc(1, schemaLen);
+          memcpy(createReq, pBlk->data, schemaLen);
+          taosArrayPush(pRsp->createTableLen, &schemaLen);
           taosArrayPush(pRsp->createTableReq, &createReq);
           pRsp->createTableNum++;
         }
@@ -277,14 +278,15 @@ int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, STaosxRsp
       }
       if (pHandle->fetchMeta) {
         SSubmitBlk* pBlk = pReader->pBlock;
-        if (pBlk->schemaLen > 0) {
+        int32_t schemaLen = htonl(pBlk->schemaLen);
+        if (schemaLen > 0) {
           if (pRsp->createTableNum == 0) {
             pRsp->createTableLen = taosArrayInit(0, sizeof(int32_t));
             pRsp->createTableReq = taosArrayInit(0, sizeof(void*));
           }
-          void* createReq = taosMemoryCalloc(1, pBlk->schemaLen);
-          memcpy(createReq, pBlk->data, pBlk->schemaLen);
-          taosArrayPush(pRsp->createTableLen, &pBlk->schemaLen);
+          void* createReq = taosMemoryCalloc(1, schemaLen);
+          memcpy(createReq, pBlk->data, schemaLen);
+          taosArrayPush(pRsp->createTableLen, &schemaLen);
           taosArrayPush(pRsp->createTableReq, &createReq);
           pRsp->createTableNum++;
         }

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

@@ -145,6 +145,7 @@ int32_t tqOffsetCommitFile(STqOffsetStore* pStore) {
       ASSERT(0);
       tqError("write offset incomplete, len %d, write len %" PRId64, bodyLen, writeLen);
       taosHashCancelIterate(pStore->pHash, pIter);
+      taosMemoryFree(buf);
       return -1;
     }
     taosMemoryFree(buf);

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

@@ -80,6 +80,13 @@ bool isValValidForTable(STqHandle* pHandle, SWalCont *pHead){
       void* buf = taosMemoryMalloc(tlen);
       if (NULL == buf) {
         taosArrayDestroy(reqNew.pArray);
+        for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
+          pCreateReq = req.pReqs + iReq;
+          taosMemoryFreeClear(pCreateReq->comment);
+          if (pCreateReq->type == TSDB_CHILD_TABLE) {
+            taosArrayDestroy(pCreateReq->ctb.tagName);
+          }
+        }
         goto end;
       }
       SEncoder coderNew = {0};
@@ -91,6 +98,14 @@ bool isValValidForTable(STqHandle* pHandle, SWalCont *pHead){
       taosMemoryFree(buf);
       taosArrayDestroy(reqNew.pArray);
     }
+
+    for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
+      pCreateReq = req.pReqs + iReq;
+      taosMemoryFreeClear(pCreateReq->comment);
+      if (pCreateReq->type == TSDB_CHILD_TABLE) {
+        taosArrayDestroy(pCreateReq->ctb.tagName);
+      }
+    }
   } else if (msgType == TDMT_VND_ALTER_TABLE) {
     SVAlterTbReq   req = {0};
 

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

@@ -48,7 +48,7 @@ int32_t tqBuildDeleteReq(SVnode* pVnode, const char* stbFullName, const SSDataBl
   return 0;
 }
 
-SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchema* pTSchema, bool createTb,
+SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchema* pTSchema, SSchemaWrapper* pTagSchemaWrapper, bool createTb,
                             int64_t suid, const char* stbFullName, SBatchDeleteReq* pDeleteReq) {
   SSubmitReq* ret = NULL;
   SArray*     schemaReqs = NULL;
@@ -89,6 +89,32 @@ SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchem
         return NULL;
       }
 
+      SArray *tagName = taosArrayInit(1, TSDB_COL_NAME_LEN);
+      char     tagNameStr[TSDB_COL_NAME_LEN] = {0};
+      strcpy(tagNameStr, "group_id");
+      taosArrayPush(tagName, tagNameStr);
+
+//      STag* pTag = NULL;
+//      taosArrayClear(tagArray);
+//      SArray *tagName = taosArrayInit(1, TSDB_COL_NAME_LEN);
+//      for(int j = 0; j < pTagSchemaWrapper->nCols; j++){
+//        STagVal tagVal = {
+//            .cid = pTagSchemaWrapper->pSchema[j].colId,
+//            .type = pTagSchemaWrapper->pSchema[j].type,
+//            .i64 = (int64_t)pDataBlock->info.groupId,
+//        };
+//        taosArrayPush(tagArray, &tagVal);
+//        taosArrayPush(tagName, pTagSchemaWrapper->pSchema[j].name);
+//      }
+//
+//      tTagNew(tagArray, 1, false, &pTag);
+//      if (pTag == NULL) {
+//        terrno = TSDB_CODE_OUT_OF_MEMORY;
+//        taosArrayDestroy(tagArray);
+//        taosArrayDestroy(tagName);
+//        return NULL;
+//      }
+
       SVCreateTbReq createTbReq = {0};
       SName         name = {0};
       tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
@@ -99,6 +125,8 @@ SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchem
       createTbReq.type = TSDB_CHILD_TABLE;
       createTbReq.ctb.suid = suid;
       createTbReq.ctb.pTag = (uint8_t*)pTag;
+      createTbReq.ctb.tagNum = taosArrayGetSize(tagArray);
+      createTbReq.ctb.tagName = tagName;
 
       int32_t code;
       int32_t schemaLen;
@@ -113,6 +141,7 @@ SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchem
       void* schemaStr = taosMemoryMalloc(schemaLen);
       if (schemaStr == NULL) {
         terrno = TSDB_CODE_OUT_OF_MEMORY;
+        tdDestroySVCreateTbReq(&createTbReq);
         return NULL;
       }
       taosArrayPush(schemaReqs, &schemaStr);
@@ -123,6 +152,7 @@ SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchem
       code = tEncodeSVCreateTbReq(&encoder, &createTbReq);
       if (code < 0) {
         terrno = TSDB_CODE_OUT_OF_MEMORY;
+        tdDestroySVCreateTbReq(&createTbReq);
         return NULL;
       }
       tEncoderClear(&encoder);
@@ -231,7 +261,7 @@ void tqTableSink(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
 
   ASSERT(pTask->tbSink.pTSchema);
   deleteReq.deleteReqs = taosArrayInit(0, sizeof(SSingleDeleteReq));
-  SSubmitReq* submitReq = tqBlockToSubmit(pVnode, pRes, pTask->tbSink.pTSchema, true, pTask->tbSink.stbUid,
+  SSubmitReq* submitReq = tqBlockToSubmit(pVnode, pRes, pTask->tbSink.pTSchema, pTask->tbSink.pSchemaWrapper, true, pTask->tbSink.stbUid,
                                           pTask->tbSink.stbFullName, &deleteReq);
 
   tqDebug("vgId:%d, task %d convert blocks over, put into write-queue", TD_VID(pVnode), pTask->taskId);

source/libs/executor/inc/executorimpl.h

@@ -1064,6 +1064,7 @@ bool               functionNeedToExecute(SqlFunctionCtx* pCtx);
 bool               isOverdue(TSKEY ts, STimeWindowAggSupp* pSup);
 bool               isCloseWindow(STimeWindow* pWin, STimeWindowAggSupp* pSup);
 bool               isDeletedWindow(STimeWindow* pWin, uint64_t groupId, SAggSupporter* pSup);
+bool               isDeletedStreamWindow(STimeWindow* pWin, uint64_t groupId, SOperatorInfo* pOperator, STimeWindowAggSupp* pTwSup);
 void               appendOneRow(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t* pUid, uint64_t* pGp);
 void               printDataBlock(SSDataBlock* pBlock, const char* flag);
 uint64_t calGroupIdByData(SPartitionBySupporter* pParSup, SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t rowId);

source/libs/executor/src/executil.c

@@ -1342,7 +1342,7 @@ int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysi
   return TSDB_CODE_SUCCESS;
 }
 
-void cleanupQueryTableDataCond(SQueryTableDataCond* pCond) { taosMemoryFree(pCond->colList); }
+void cleanupQueryTableDataCond(SQueryTableDataCond* pCond) { taosMemoryFreeClear(pCond->colList); }
 
 int32_t convertFillType(int32_t mode) {
   int32_t type = TSDB_FILL_NONE;

source/libs/executor/src/executor.c

@@ -876,6 +876,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
     tsdbReaderOpen(pInfo->vnode, &pTaskInfo->streamInfo.tableCond, pTaskInfo->tableqinfoList.pTableList,
                    &pInfo->dataReader, NULL);
 
+    cleanupQueryTableDataCond(&pTaskInfo->streamInfo.tableCond);
     strcpy(pTaskInfo->streamInfo.tbName, mtInfo.tbName);
     tDeleteSSchemaWrapper(pTaskInfo->streamInfo.schema);
     pTaskInfo->streamInfo.schema = mtInfo.schema;

source/libs/executor/src/executorimpl.c

@@ -3350,6 +3350,10 @@ static void cleanupTableSchemaInfo(SSchemaInfo* pSchemaInfo) {
   tDeleteSSchemaWrapper(pSchemaInfo->qsw);
 }
 
+static void cleanupStreamInfo(SStreamTaskInfo* pStreamInfo) {
+  tDeleteSSchemaWrapper(pStreamInfo->schema);
+}
+
 static int32_t sortTableGroup(STableListInfo* pTableListInfo) {
   taosArrayClear(pTableListInfo->pGroupList);
   SArray* sortSupport = taosArrayInit(16, sizeof(uint64_t));
@@ -4043,6 +4047,7 @@ void doDestroyTask(SExecTaskInfo* pTaskInfo) {
   doDestroyTableList(&pTaskInfo->tableqinfoList);
   destroyOperatorInfo(pTaskInfo->pRoot);
   cleanupTableSchemaInfo(&pTaskInfo->schemaInfo);
+  cleanupStreamInfo(&pTaskInfo->streamInfo);
 
   nodesDestroyNode((SNode*)pTaskInfo->pSubplan);
 

source/libs/executor/src/scanoperator.c

@@ -1331,7 +1331,7 @@ static void checkUpdateData(SStreamScanInfo* pInfo, bool invertible, SSDataBlock
     // must check update info first.
     bool update = updateInfoIsUpdated(pInfo->pUpdateInfo, pBlock->info.uid, tsCol[rowId]);
     bool closedWin = isClosed && isSignleIntervalWindow(pInfo) &&
-                     isDeletedWindow(&win, pBlock->info.groupId, pInfo->windowSup.pIntervalAggSup);
+                     isDeletedStreamWindow(&win, pBlock->info.groupId, pInfo->pTableScanOp, &pInfo->twAggSup);
     if ((update || closedWin) && out) {
       qDebug("stream update check not pass, update %d, closedWin %d", update, closedWin);
       uint64_t gpId = closedWin && pInfo->partitionSup.needCalc
@@ -1780,6 +1780,7 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
       qDebug("tmqsnap change get data uid:%ld", mtInfo.uid);
       qStreamPrepareScan(pTaskInfo, &pTaskInfo->streamInfo.prepareStatus, pInfo->sContext->subType);
     }
+    tDeleteSSchemaWrapper(mtInfo.schema);
     qDebug("tmqsnap stream scan tsdb return null");
     return NULL;
   } else if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_META) {
@@ -1931,11 +1932,6 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
 
   pInfo->pTagCond = pTagCond;
   pInfo->pGroupTags = pTableScanNode->pGroupTags;
-  pInfo->twAggSup = (STimeWindowAggSupp){
-      .waterMark = pTableScanNode->watermark,
-      .calTrigger = pTableScanNode->triggerType,
-      .maxTs = INT64_MIN,
-  };
 
   int32_t numOfCols = 0;
   pInfo->pColMatchInfo = extractColMatchInfo(pScanPhyNode->pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID);
@@ -1985,7 +1981,6 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
 
     pInfo->pUpdateInfo = NULL;
     pInfo->pTableScanOp = pTableScanOp;
-    pInfo->interval = pTSInfo->pdInfo.interval;
 
     pInfo->readHandle = *pHandle;
     pInfo->tableUid = pScanPhyNode->uid;

source/libs/executor/src/timewindowoperator.c

@@ -1753,16 +1753,17 @@ static bool timeWindowinterpNeeded(SqlFunctionCtx* pCtx, int32_t numOfCols, SInt
 }
 
 void initIntervalDownStream(SOperatorInfo* downstream, uint16_t type, SAggSupporter* pSup, SInterval* pInterval,
-                            int64_t waterMark) {
+                            STimeWindowAggSupp* pTwSup) {
   if (downstream->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
-    initIntervalDownStream(downstream->pDownstream[0], type, pSup, pInterval, waterMark);
+    initIntervalDownStream(downstream->pDownstream[0], type, pSup, pInterval, pTwSup);
     return;
   }
   SStreamScanInfo* pScanInfo = downstream->info;
   pScanInfo->windowSup.parentType = type;
   pScanInfo->windowSup.pIntervalAggSup = pSup;
-  pScanInfo->pUpdateInfo = updateInfoInitP(pInterval, waterMark);
+  pScanInfo->pUpdateInfo = updateInfoInitP(pInterval, pTwSup->waterMark);
   pScanInfo->interval = *pInterval;
+  pScanInfo->twAggSup = *pTwSup;
 }
 
 void initStreamFunciton(SqlFunctionCtx* pCtx, int32_t numOfExpr) {
@@ -1847,11 +1848,6 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo*
   pOperator->fpSet = createOperatorFpSet(doOpenIntervalAgg, doBuildIntervalResult, NULL, NULL,
                                          destroyIntervalOperatorInfo, aggEncodeResultRow, aggDecodeResultRow, NULL);
 
-  if (nodeType(pPhyNode) == QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL) {
-    initIntervalDownStream(downstream, QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL, &pInfo->aggSup, &pInfo->interval,
-                           pInfo->twAggSup.waterMark);
-  }
-
   code = appendDownstream(pOperator, &downstream, 1);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
@@ -2868,6 +2864,19 @@ bool isDeletedWindow(STimeWindow* pWin, uint64_t groupId, SAggSupporter* pSup) {
   return p1 == NULL;
 }
 
+bool isDeletedStreamWindow(STimeWindow* pWin, uint64_t groupId, SOperatorInfo* pOperator, STimeWindowAggSupp* pTwSup) {
+  if (pWin->ekey < pTwSup->maxTs - pTwSup->deleteMark) {
+    SWinKey key = {.ts = pWin->skey, .groupId = groupId};
+    void*   pVal = NULL;
+    int32_t size = 0;
+    if (streamStateGet(pOperator->pTaskInfo->streamInfo.pState, &key, &pVal, &size) < 0) {
+      return false;
+    }
+    streamStateReleaseBuf(pOperator->pTaskInfo->streamInfo.pState, &key, pVal);
+  }
+  return false;
+}
+
 int32_t getNexWindowPos(SInterval* pInterval, SDataBlockInfo* pBlockInfo, TSKEY* tsCols, int32_t startPos, TSKEY eKey,
                         STimeWindow* pNextWin) {
   int32_t forwardRows =
@@ -3425,7 +3434,7 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
       createOperatorFpSet(NULL, doStreamFinalIntervalAgg, NULL, NULL, destroyStreamFinalIntervalOperatorInfo,
                           aggEncodeResultRow, aggDecodeResultRow, NULL);
   if (pPhyNode->type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL) {
-    initIntervalDownStream(downstream, pPhyNode->type, &pInfo->aggSup, &pInfo->interval, pInfo->twAggSup.waterMark);
+    initIntervalDownStream(downstream, pPhyNode->type, &pInfo->aggSup, &pInfo->interval, &pInfo->twAggSup);
   }
   code = appendDownstream(pOperator, &downstream, 1);
   if (code != TSDB_CODE_SUCCESS) {
@@ -5944,7 +5953,7 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhys
       createOperatorFpSet(operatorDummyOpenFn, doStreamIntervalAgg, NULL, NULL, destroyStreamIntervalOperatorInfo,
                           aggEncodeResultRow, aggDecodeResultRow, NULL);
 
-  initIntervalDownStream(downstream, pPhyNode->type, &pInfo->aggSup, &pInfo->interval, pInfo->twAggSup.waterMark);
+  initIntervalDownStream(downstream, pPhyNode->type, &pInfo->aggSup, &pInfo->interval, &pInfo->twAggSup);
   code = appendDownstream(pOperator, &downstream, 1);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;

source/libs/parser/src/parInsert.c

@@ -1059,7 +1059,7 @@ end:
   for (int i = 0; i < taosArrayGetSize(pTagVals); ++i) {
     STagVal* p = (STagVal*)taosArrayGet(pTagVals, i);
     if (IS_VAR_DATA_TYPE(p->type)) {
-      taosMemoryFree(p->pData);
+      taosMemoryFreeClear(p->pData);
     }
   }
   taosArrayDestroy(pTagVals);
@@ -2040,7 +2040,7 @@ end:
   for (int i = 0; i < taosArrayGetSize(pTagArray); ++i) {
     STagVal* p = (STagVal*)taosArrayGet(pTagArray, i);
     if (p->type == TSDB_DATA_TYPE_NCHAR) {
-      taosMemoryFree(p->pData);
+      taosMemoryFreeClear(p->pData);
     }
   }
   taosArrayDestroy(pTagArray);

source/libs/parser/src/parTranslater.c

@@ -6664,12 +6664,7 @@ static int32_t buildUpdateTagValReq(STranslateContext* pCxt, SAlterTableStmt* pS
         break;
       }
     } while (0);
-    for (int i = 0; i < taosArrayGetSize(pTagVals); ++i) {
-      STagVal* p = (STagVal*)taosArrayGet(pTagVals, i);
-      if (IS_VAR_DATA_TYPE(p->type)) {
-        taosMemoryFree(p->pData);
-      }
-    }
+
     taosArrayDestroy(pTagVals);
     if (code != TSDB_CODE_SUCCESS) {
       return code;

source/libs/parser/src/parUtil.c

@@ -410,6 +410,12 @@ end:
   if (retCode == TSDB_CODE_SUCCESS) {
     tTagNew(pTagVals, 1, true, ppTag);
   }
+  for (int i = 0; i < taosArrayGetSize(pTagVals); ++i) {
+    STagVal* p = (STagVal*)taosArrayGet(pTagVals, i);
+    if (IS_VAR_DATA_TYPE(p->type)) {
+      taosMemoryFreeClear(p->pData);
+    }
+  }
   cJSON_Delete(root);
   return retCode;
 }

source/libs/qworker/src/qwMsg.c

@@ -632,6 +632,7 @@ int32_t qWorkerProcessDeleteMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, SD
 
   QW_ERR_JRET(qwProcessDelete(QW_FPARAMS(), &qwMsg, pRes));
 
+  taosMemoryFreeClear(req.msg);
   QW_SCH_TASK_DLOG("processDelete end, node:%p", node);
 
 _return:

tests/system-test/7-tmq/tmq_taosx.py

@@ -31,18 +31,20 @@ class TDTestCase:
         while True:
             dst = queryFile.readline()
             src = consumeFile.readline()
-
-            if dst:
+            if src:
                 if dst != src:
-                    tdLog.exit("compare error: %s != %s"%src, dst)
+                    tdLog.exit("compare error: %s != %s"%(src, dst))
             else:
                 break
         return
 
-    def checkDropData(self):
+    def checkDropData(self, drop):
         tdSql.execute('use db_taosx')
         tdSql.query("show tables")
-        tdSql.checkRows(2)
+        if drop:
+            tdSql.checkRows(10)
+        else:
+            tdSql.checkRows(15)
         tdSql.query("select * from jt order by i")
         tdSql.checkRows(2)
         tdSql.checkData(0, 1, 1)
@@ -50,15 +52,72 @@ class TDTestCase:
         tdSql.checkData(0, 2, '{"k1":1,"k2":"hello"}')
         tdSql.checkData(1, 2, None)
 
+        tdSql.query("select * from sttb order by ts")
+        tdSql.checkRows(2)
+        tdSql.checkData(0, 1, 13)
+        tdSql.checkData(1, 1, 16)
+        tdSql.checkData(0, 2, 22)
+        tdSql.checkData(1, 2, 25)
+        tdSql.checkData(0, 5, "sttb3")
+        tdSql.checkData(1, 5, "sttb4")
+
+        tdSql.query("select * from stt order by ts")
+        tdSql.checkRows(2)
+        tdSql.checkData(0, 1, 1)
+        tdSql.checkData(1, 1, 21)
+        tdSql.checkData(0, 2, 2)
+        tdSql.checkData(1, 2, 21)
+        tdSql.checkData(0, 5, "stt3")
+        tdSql.checkData(1, 5, "stt4")
+
         tdSql.execute('use abc1')
         tdSql.query("show tables")
-        tdSql.checkRows(2)
+        if drop:
+            tdSql.checkRows(10)
+        else:
+            tdSql.checkRows(15)
         tdSql.query("select * from jt order by i")
         tdSql.checkRows(2)
         tdSql.checkData(0, 1, 1)
         tdSql.checkData(1, 1, 11)
         tdSql.checkData(0, 2, '{"k1":1,"k2":"hello"}')
         tdSql.checkData(1, 2, None)
+
+        tdSql.query("select * from sttb order by ts")
+        tdSql.checkRows(2)
+        tdSql.checkData(0, 1, 13)
+        tdSql.checkData(1, 1, 16)
+        tdSql.checkData(0, 2, 22)
+        tdSql.checkData(1, 2, 25)
+        tdSql.checkData(0, 5, "sttb3")
+        tdSql.checkData(1, 5, "sttb4")
+
+        tdSql.query("select * from stt order by ts")
+        tdSql.checkRows(2)
+        tdSql.checkData(0, 1, 1)
+        tdSql.checkData(1, 1, 21)
+        tdSql.checkData(0, 2, 2)
+        tdSql.checkData(1, 2, 21)
+        tdSql.checkData(0, 5, "stt3")
+        tdSql.checkData(1, 5, "stt4")
+
+        return
+
+    def checkDataTable(self):
+        tdSql.execute('use db_taosx')
+        tdSql.query("select * from meters_summary")
+        tdSql.checkRows(1)
+        tdSql.checkData(0, 1, 120)
+        tdSql.checkData(0, 2, 1)
+        tdSql.checkData(0, 3, "San Francisco")
+
+        tdSql.execute('use abc1')
+        tdSql.query("select * from meters_summary")
+        tdSql.checkRows(1)
+        tdSql.checkData(0, 1, 120)
+        tdSql.checkData(0, 2, 1)
+        tdSql.checkData(0, 3, "San Francisco")
+
         return
 
     def checkData(self):
@@ -145,6 +204,19 @@ class TDTestCase:
 
         self.checkJson(cfgPath, "tmq_taosx_tmp")
         self.checkData()
+        self.checkDropData(False)
+
+        return
+
+    def checkWal1VgroupTable(self):
+        buildPath = tdCom.getBuildPath()
+        cfgPath = tdCom.getClientCfgPath()
+        cmdStr = '%s/build/bin/tmq_taosx_ci -c %s -sv 1 -dv 1 -t'%(buildPath, cfgPath)
+        tdLog.info(cmdStr)
+        os.system(cmdStr)
+
+        self.checkJson(cfgPath, "tmq_taosx_tmp")
+        self.checkDataTable()
 
         return
 
@@ -155,6 +227,7 @@ class TDTestCase:
         os.system(cmdStr)
 
         self.checkData()
+        self.checkDropData(False)
 
         return
 
@@ -164,7 +237,7 @@ class TDTestCase:
         tdLog.info(cmdStr)
         os.system(cmdStr)
 
-        self.checkDropData()
+        self.checkDropData(True)
 
         return
 
@@ -177,6 +250,19 @@ class TDTestCase:
 
         self.checkJson(cfgPath, "tmq_taosx_tmp_snapshot")
         self.checkData()
+        self.checkDropData(False)
+
+        return
+
+    def checkSnapshot1VgroupTable(self):
+        buildPath = tdCom.getBuildPath()
+        cfgPath = tdCom.getClientCfgPath()
+        cmdStr = '%s/build/bin/tmq_taosx_ci -c %s -sv 1 -dv 1 -s -t'%(buildPath, cfgPath)
+        tdLog.info(cmdStr)
+        os.system(cmdStr)
+
+        self.checkJson(cfgPath, "tmq_taosx_tmp_snapshot")
+        self.checkDataTable()
 
         return
 
@@ -187,6 +273,7 @@ class TDTestCase:
         os.system(cmdStr)
 
         self.checkData()
+        self.checkDropData(False)
 
         return
 
@@ -196,7 +283,7 @@ class TDTestCase:
         tdLog.info(cmdStr)
         os.system(cmdStr)
 
-        self.checkDropData()
+        self.checkDropData(True)
 
         return
 
@@ -205,6 +292,9 @@ class TDTestCase:
         self.checkWal1Vgroup()
         self.checkSnapshot1Vgroup()
 
+        self.checkWal1VgroupTable()
+        self.checkSnapshot1VgroupTable()
+
         self.checkWalMultiVgroups()
         self.checkSnapshotMultiVgroups()
 

utils/test/c/tmqSim.c

@@ -653,23 +653,23 @@ static int32_t meta_msg_process(TAOS_RES* msg, SThreadInfo* pInfo, int32_t msgIn
     int32_t code = tmq_get_raw(msg, &raw);
 	
     if(code == TSDB_CODE_SUCCESS){
-	  int retCode = queryDB(pInfo->taos, "use metadb");
-	  if (retCode != 0) {
-		taosFprintfFile(g_fp, "error when use metadb\n");
-		taosCloseFile(&g_fp);
-		exit(-1);
-	  }	  
-	  taosFprintfFile(g_fp, "raw:%p\n", &raw);
-	
-      tmq_write_raw(pInfo->taos, raw);
+//	  int retCode = queryDB(pInfo->taos, "use metadb");
+//	  if (retCode != 0) {
+//		taosFprintfFile(g_fp, "error when use metadb\n");
+//		taosCloseFile(&g_fp);
+//		exit(-1);
+//	  }
+//	  taosFprintfFile(g_fp, "raw:%p\n", &raw);
+//
+//      tmq_write_raw(pInfo->taos, raw);
     }
 	
     char* result = tmq_get_json_meta(msg);
-    if(result){
+    if(result && strcmp(result, "") != 0){
   	  //printf("meta result: %s\n", result);
   	  taosFprintfFile(pInfo->pConsumeMetaFile, "%s\n", result);
-  	  taosMemoryFree(result);
     }
+    tmq_free_json_meta(result);
   }
 
   totalRows++;
@@ -818,8 +818,12 @@ void loop_consume(SThreadInfo* pInfo) {
 	  	tmq_res_t msgType = tmq_get_res_type(tmqMsg);
 		if (msgType == TMQ_RES_TABLE_META) {
  		  totalRows += meta_msg_process(tmqMsg, pInfo, totalMsgs);
-		} else if (msgType == TMQ_RES_DATA)
-          totalRows += data_msg_process(tmqMsg, pInfo, totalMsgs);
+		} else if (msgType == TMQ_RES_DATA){
+                  totalRows += data_msg_process(tmqMsg, pInfo, totalMsgs);
+                } else if (msgType == TMQ_RES_METADATA){
+                  meta_msg_process(tmqMsg, pInfo, totalMsgs);
+                  totalRows += data_msg_process(tmqMsg, pInfo, totalMsgs);
+                }
       }
 
       taos_free_result(tmqMsg);