fix:error in auto create table for taosX

source/client/src/clientRawBlockWrite.c

@@ -219,8 +219,6 @@ static char* processAlterStb(SMqMetaRsp* metaRsp) {
     goto _err;
   }
   string = buildAlterSTableJson(req.alterOriData, req.alterOriDataLen);
-  tDecoderClear(&coder);
-  return string;
 
 _err:
   tDecoderClear(&coder);
@@ -397,6 +395,10 @@ static char* processAutoCreateTable(STaosxRsp* rsp) {
 _exit:
   for(int i = 0; i < rsp->createTableNum; i++){
     tDecoderClear(&decoder[i]);
+    taosMemoryFreeClear(pCreateReq[i].comment);
+    if (pCreateReq[i].type == TSDB_CHILD_TABLE) {
+      taosArrayDestroy(pCreateReq[i].ctb.tagName);
+    }
   }
   taosMemoryFree(decoder);
   taosMemoryFree(pCreateReq);
@@ -512,6 +514,7 @@ static char* processAlterTable(SMqMetaRsp* metaRsp) {
   string = cJSON_PrintUnformatted(json);
 
 _exit:
+  cJSON_Delete(json);
   tDecoderClear(&decoder);
   return string;
 }
@@ -543,6 +546,7 @@ static char* processDropSTable(SMqMetaRsp* metaRsp) {
   string = cJSON_PrintUnformatted(json);
 
 _exit:
+  cJSON_Delete(json);
   tDecoderClear(&decoder);
   return string;
 }
@@ -583,6 +587,7 @@ static char* processDropTable(SMqMetaRsp* metaRsp) {
   string = cJSON_PrintUnformatted(json);
 
 _exit:
+  cJSON_Delete(json);
   tDecoderClear(&decoder);
   return string;
 }
@@ -688,6 +693,7 @@ static int32_t taosDropStb(TAOS* taos, void* meta, int32_t metaLen) {
     goto end;
   }
 
+  pRequest->syncQuery = true;
   if (!pRequest->pDb) {
     code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
     goto end;
@@ -920,7 +926,7 @@ static int32_t taosDropTable(TAOS* taos, void* meta, int32_t metaLen) {
   if (code != TSDB_CODE_SUCCESS) {
     goto end;
   }
-
+  pRequest->syncQuery = true;
   if (!pRequest->pDb) {
     code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
     goto end;
@@ -1093,6 +1099,7 @@ static int32_t taosAlterTable(TAOS* taos, void* meta, int32_t metaLen) {
     goto end;
   }
 
+  pRequest->syncQuery = true;
   if (!pRequest->pDb) {
     code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
     goto end;
@@ -1212,6 +1219,7 @@ int taos_write_raw_block(TAOS* taos, int rows, char* pData, const char* tbname)
     goto end;
   }
 
+  pRequest->syncQuery = true;
   if (!pRequest->pDb) {
     uError("WriteRaw:not use db");
     code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
@@ -1399,6 +1407,7 @@ static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, int32_t dataLen) {
     return terrno;
   }
 
+  pRequest->syncQuery = true;
   rspObj.resIter = -1;
   rspObj.resType = RES_TYPE__TMQ;
 
@@ -1664,6 +1673,7 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen)
     return terrno;
   }
 
+  pRequest->syncQuery = true;
   rspObj.resIter = -1;
   rspObj.resType = RES_TYPE__TMQ_METADATA;
 
@@ -1739,23 +1749,29 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen)
       int32_t* lenTmp = taosArrayGet(rspObj.rsp.createTableLen, j);
 
       SDecoder           decoderTmp = {0};
-      SVCreateTbReq*     pCreateReq;
+      SVCreateTbReq      pCreateReq = {0};
 
       tDecoderInit(&decoderTmp, *dataTmp, *lenTmp);
-      if (tDecodeSVCreateTbReq(&decoderTmp, pCreateReq) < 0) {
+      if (tDecodeSVCreateTbReq(&decoderTmp, &pCreateReq) < 0) {
         tDecoderClear(&decoderTmp);
+        taosMemoryFreeClear(pCreateReq.comment);
+        taosArrayDestroy(pCreateReq.ctb.tagName);
         goto end;
       }
 
-      ASSERT (pCreateReq->type == TSDB_CHILD_TABLE);
-      if(strcmp(tbName, pCreateReq->name) == 0){
+      ASSERT (pCreateReq.type == TSDB_CHILD_TABLE);
+      if(strcmp(tbName, pCreateReq.name) == 0){
         schemaLen = *lenTmp;
         schemaData = *dataTmp;
-        strcpy(pName.tname, pCreateReq->ctb.name);
+        strcpy(pName.tname, pCreateReq.ctb.name);
         tDecoderClear(&decoderTmp);
+        taosMemoryFreeClear(pCreateReq.comment);
+        taosArrayDestroy(pCreateReq.ctb.tagName);
         break;
       }
       tDecoderClear(&decoderTmp);
+      taosMemoryFreeClear(pCreateReq.comment);
+      taosArrayDestroy(pCreateReq.ctb.tagName);
     }
 
     code = catalogGetTableMeta(pCatalog, &conn, &pName, &pTableMeta);
@@ -1884,6 +1900,8 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen)
     subReq->length += sizeof(SSubmitBlk) + schemaLen + totalLen;
     subReq->numOfBlocks++;
     taosMemoryFreeClear(pTableMeta);
+    rspObj.resInfo.pRspMsg = NULL;
+    doFreeReqResultInfo(&rspObj.resInfo);
   }
 
   pQuery = (SQuery*)nodesMakeNode(QUERY_NODE_QUERY);

source/client/src/clientTmq.c

@@ -1816,7 +1816,7 @@ 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_META(res)) {
+  } else if (TD_RES_TMQ_METADATA(res)) {
     SMqTaosxRspObj* pRspObj = (SMqTaosxRspObj*)res;
     return pRspObj->vgId;
   } else {
@@ -1832,6 +1832,13 @@ 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

@@ -6007,12 +6007,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

@@ -539,7 +539,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

@@ -343,8 +343,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/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;
@@ -382,11 +388,9 @@ int metaTtlSmaller(SMeta *pMeta, uint64_t ttl, SArray *uidList) {
     }
     ttlKey = *(STtlIdxKey *)pKey;
     taosArrayPush(uidList, &ttlKey.uid);
+    tdbFree(pKey);
   }
   tdbTbcClose(pCur);
-
-  tdbFree(pKey);
-
   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

@@ -1191,10 +1191,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;
@@ -1220,17 +1221,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/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);
@@ -520,6 +536,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;
     }
 
@@ -580,14 +597,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/tqOffset.c

@@ -145,8 +145,10 @@ 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);
   }
   // close and rename file
   taosCloseFile(&pFile);

source/libs/executor/src/executorimpl.c

@@ -3645,6 +3645,11 @@ static void cleanupTableSchemaInfo(SSchemaInfo* pSchemaInfo) {
   tDeleteSSchemaWrapper(pSchemaInfo->qsw);
 }
 
+static void cleanupStreamInfo(SStreamTaskInfo* pStreamInfo) {
+  tDeleteSSchemaWrapper(pStreamInfo->schema);
+  cleanupQueryTableDataCond(&pStreamInfo->tableCond);
+}
+
 static int32_t sortTableGroup(STableListInfo* pTableListInfo) {
   taosArrayClear(pTableListInfo->pGroupList);
   SArray* sortSupport = taosArrayInit(16, sizeof(uint64_t));
@@ -4338,6 +4343,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

@@ -1769,6 +1769,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) {

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:

utils/test/c/tmq_taosx_ci.c

@@ -68,6 +68,7 @@ static void msg_process(TAOS_RES* msg) {
 
   tmq_raw_data raw = {0};
   tmq_get_raw(msg, &raw);
+  printf("write raw data type: %d\n", raw.raw_type);
   int32_t ret = tmq_write_raw(pConn, raw);
   printf("write raw data: %s\n", tmq_err2str(ret));
   tmq_free_raw(raw);