fix mem leak

include/common/common.h

@@ -174,6 +174,24 @@ static FORCE_INLINE void* tDecodeSMqConsumeRsp(void* buf, SMqConsumeRsp* pRsp) {
   return buf;
 }
 
+static FORCE_INLINE void destroySSDataBlock(SSDataBlock* pBlock) {
+  if (pBlock == NULL) {
+    return;
+  }
+
+  //int32_t numOfOutput = pBlock->info.numOfCols;
+  int32_t sz = taosArrayGetSize(pBlock->pDataBlock);
+  for(int32_t i = 0; i < sz; ++i) {
+    SColumnInfoData* pColInfoData = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, i);
+    tfree(pColInfoData->pData);
+  }
+
+  taosArrayDestroy(pBlock->pDataBlock);
+  tfree(pBlock->pBlockAgg);
+  tfree(pBlock);
+}
+
+
 //======================================================================================================================
 // the following structure shared by parser and executor
 typedef struct SColumn {

source/client/test/clientTests.cpp

@@ -606,7 +606,7 @@ TEST(testCase, create_topic_stb_Test) {
 
   taos_free_result(pRes);
 
-  char* sql = "select ts, k from st1";
+  char* sql = "select * from st1";
   pRes = taos_create_topic(pConn, "test_stb_topic_1", sql, strlen(sql));
   taos_free_result(pRes);
   taos_close(pConn);

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

@@ -785,6 +785,13 @@ int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) {
   }
   void* abuf = buf;
   tEncodeSMqConsumeRsp(&abuf, &rsp);
+  if (rsp.pBlockData) {
+    for (int i = 0; i < taosArrayGetSize(rsp.pBlockData); i++) {
+      SSDataBlock* pBlock = taosArrayGet(rsp.pBlockData, i);
+      destroySSDataBlock(pBlock);
+    }
+    free(rsp.pBlockData);
+  }
   pMsg->pCont = buf;
   pMsg->contLen = tlen;
   pMsg->code = 0;