Merge pull request #18383 from taosdata/feature/stream

fix(stream): memory leak

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

@@ -1275,6 +1275,7 @@ int32_t tqProcessDelReq(STQ* pTq, void* pReq, int32_t len, int64_t ver) {
       if (streamTaskInput(pTask, (SStreamQueueItem*)pRefBlock) < 0) {
         qError("stream task input del failed, task id %d", pTask->taskId);
 
+        atomic_sub_fetch_32(pRef, 1);
         taosFreeQitem(pRefBlock);
         continue;
       }
@@ -1292,7 +1293,7 @@ int32_t tqProcessDelReq(STQ* pTq, void* pReq, int32_t len, int64_t ver) {
   int32_t ref = atomic_sub_fetch_32(pRef, 1);
   ASSERT(ref >= 0);
   if (ref == 0) {
-    taosMemoryFree(pDelBlock);
+    blockDataDestroy(pDelBlock);
     taosMemoryFree(pRef);
   }
 

source/libs/executor/src/groupoperator.c

@@ -936,6 +936,9 @@ static SSDataBlock* buildStreamPartitionResult(SOperatorInfo* pOperator) {
       } else {
         pDest->info.parTbName[0] = 0;
       }
+      if (pParInfo->groupId && pDest->info.parTbName[0]) {
+        streamStatePutParName(pOperator->pTaskInfo->streamInfo.pState, pParInfo->groupId, pDest->info.parTbName);
+      }
       /*printf("\n\n set name %s\n\n", pDest->info.parTbName);*/
       blockDataDestroy(pTmpBlock);
       blockDataDestroy(pResBlock);