Merge pull request #16617 from taosdata/feature/stream

feat(stream): support delete data

include/common/tdatablock.h

@@ -184,7 +184,8 @@ static FORCE_INLINE void colDataAppendDouble(SColumnInfoData* pColumnInfoData, u
 int32_t getJsonValueLen(const char* data);
 
 int32_t colDataAppend(SColumnInfoData* pColumnInfoData, uint32_t currentRow, const char* pData, bool isNull);
-int32_t colDataAppendNItems(SColumnInfoData* pColumnInfoData, uint32_t currentRow, const char* pData, uint32_t numOfRows);
+int32_t colDataAppendNItems(SColumnInfoData* pColumnInfoData, uint32_t currentRow, const char* pData,
+                            uint32_t numOfRows);
 int32_t colDataMergeCol(SColumnInfoData* pColumnInfoData, int32_t numOfRow1, int32_t* capacity,
                         const SColumnInfoData* pSource, int32_t numOfRow2);
 int32_t colDataAssign(SColumnInfoData* pColumnInfoData, const SColumnInfoData* pSource, int32_t numOfRows,
@@ -225,15 +226,16 @@ size_t blockDataGetCapacityInRow(const SSDataBlock* pBlock, size_t pageSize);
 int32_t blockDataTrimFirstNRows(SSDataBlock* pBlock, size_t n);
 int32_t blockDataKeepFirstNRows(SSDataBlock* pBlock, size_t n);
 
-int32_t      assignOneDataBlock(SSDataBlock* dst, const SSDataBlock* src);
-int32_t      copyDataBlock(SSDataBlock* dst, const SSDataBlock* src);
+int32_t assignOneDataBlock(SSDataBlock* dst, const SSDataBlock* src);
+int32_t copyDataBlock(SSDataBlock* dst, const SSDataBlock* src);
 
 SSDataBlock* createDataBlock();
 void*        blockDataDestroy(SSDataBlock* pBlock);
 void         blockDataFreeRes(SSDataBlock* pBlock);
 SSDataBlock* createOneDataBlock(const SSDataBlock* pDataBlock, bool copyData);
+SSDataBlock* createSpecialDataBlock(EStreamType type);
 
-int32_t      blockDataAppendColInfo(SSDataBlock* pBlock, SColumnInfoData* pColInfoData);
+int32_t blockDataAppendColInfo(SSDataBlock* pBlock, SColumnInfoData* pColInfoData);
 
 SColumnInfoData  createColumnInfoData(int16_t type, int32_t bytes, int16_t colId);
 SColumnInfoData* bdGetColumnInfoData(const SSDataBlock* pBlock, int32_t index);
@@ -249,7 +251,6 @@ char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** dumpBuf);
 int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SSDataBlock* pDataBlocks, STSchema* pTSchema, int32_t vgId,
                                     tb_uid_t suid);
 
-
 char* buildCtbNameByGroupId(const char* stbName, uint64_t groupId);
 
 static FORCE_INLINE int32_t blockGetEncodeSize(const SSDataBlock* pBlock) {

source/common/src/tdatablock.c

@@ -140,7 +140,8 @@ int32_t colDataReserve(SColumnInfoData* pColumnInfoData, size_t newSize) {
   return TSDB_CODE_SUCCESS;
 }
 
-static void doCopyNItems(struct SColumnInfoData* pColumnInfoData, int32_t currentRow, const char* pData, int32_t itemLen, int32_t numOfRows) {
+static void doCopyNItems(struct SColumnInfoData* pColumnInfoData, int32_t currentRow, const char* pData,
+                         int32_t itemLen, int32_t numOfRows) {
   ASSERT(pColumnInfoData->info.bytes >= itemLen);
   size_t start = 1;
 
@@ -148,21 +149,23 @@ static void doCopyNItems(struct SColumnInfoData* pColumnInfoData, int32_t curren
   memcpy(pColumnInfoData->pData, pData, itemLen);
 
   int32_t t = 0;
-  int32_t count = log(numOfRows)/log(2);
-  while(t < count) {
+  int32_t count = log(numOfRows) / log(2);
+  while (t < count) {
     int32_t xlen = 1 << t;
-    memcpy(pColumnInfoData->pData + start * itemLen + pColumnInfoData->varmeta.length, pColumnInfoData->pData, xlen * itemLen);
+    memcpy(pColumnInfoData->pData + start * itemLen + pColumnInfoData->varmeta.length, pColumnInfoData->pData,
+           xlen * itemLen);
     t += 1;
     start += xlen;
   }
 
   // the tail part
   if (numOfRows > start) {
-    memcpy(pColumnInfoData->pData + start * itemLen + currentRow * itemLen, pColumnInfoData->pData, (numOfRows - start) * itemLen);
+    memcpy(pColumnInfoData->pData + start * itemLen + currentRow * itemLen, pColumnInfoData->pData,
+           (numOfRows - start) * itemLen);
   }
 
   if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
-    for(int32_t i = 0; i < numOfRows; ++i) {
+    for (int32_t i = 0; i < numOfRows; ++i) {
       pColumnInfoData->varmeta.offset[i + currentRow] = pColumnInfoData->varmeta.length + i * itemLen;
     }
 
@@ -170,7 +173,8 @@ static void doCopyNItems(struct SColumnInfoData* pColumnInfoData, int32_t curren
   }
 }
 
-int32_t colDataAppendNItems(SColumnInfoData* pColumnInfoData, uint32_t currentRow, const char* pData, uint32_t numOfRows) {
+int32_t colDataAppendNItems(SColumnInfoData* pColumnInfoData, uint32_t currentRow, const char* pData,
+                            uint32_t numOfRows) {
   ASSERT(pData != NULL && pColumnInfoData != NULL);
 
   int32_t len = pColumnInfoData->info.bytes;
@@ -278,7 +282,7 @@ int32_t colDataMergeCol(SColumnInfoData* pColumnInfoData, int32_t numOfRow1, int
   } else {
     if (finalNumOfRows > *capacity || (numOfRow1 == 0 && pColumnInfoData->info.bytes != 0)) {
       // all data may be null, when the pColumnInfoData->info.type == 0, bytes == 0;
-//      ASSERT(finalNumOfRows * pColumnInfoData->info.bytes);
+      //      ASSERT(finalNumOfRows * pColumnInfoData->info.bytes);
       char* tmp = taosMemoryRealloc(pColumnInfoData->pData, finalNumOfRows * pColumnInfoData->info.bytes);
       if (tmp == NULL) {
         return TSDB_CODE_VND_OUT_OF_MEMORY;
@@ -557,7 +561,7 @@ int32_t blockDataToBuf(char* buf, const SSDataBlock* pBlock) {
 }
 
 int32_t blockDataFromBuf(SSDataBlock* pBlock, const char* buf) {
-  int32_t numOfRows = *(int32_t*) buf;
+  int32_t numOfRows = *(int32_t*)buf;
   blockDataEnsureCapacity(pBlock, numOfRows);
 
   pBlock->info.rows = numOfRows;
@@ -676,7 +680,8 @@ size_t blockDataGetRowSize(SSDataBlock* pBlock) {
  * @return
  */
 size_t blockDataGetSerialMetaSize(uint32_t numOfCols) {
-  // | version | total length | total rows | total columns | flag seg| block group id | column schema | each column length |
+  // | version | total length | total rows | total columns | flag seg| block group id | column schema | each column
+  // length |
   return sizeof(int32_t) + sizeof(int32_t) + sizeof(int32_t) + sizeof(int32_t) + sizeof(int32_t) + sizeof(uint64_t) +
          numOfCols * (sizeof(int8_t) + sizeof(int32_t)) + numOfCols * sizeof(int32_t);
 }
@@ -1302,6 +1307,40 @@ int32_t copyDataBlock(SSDataBlock* dst, const SSDataBlock* src) {
   return TSDB_CODE_SUCCESS;
 }
 
+SSDataBlock* createSpecialDataBlock(EStreamType type) {
+  SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
+  pBlock->info.hasVarCol = false;
+  pBlock->info.groupId = 0;
+  pBlock->info.rows = 0;
+  pBlock->info.type = type;
+  pBlock->info.rowSize =
+      sizeof(TSKEY) + sizeof(TSKEY) + sizeof(uint64_t) + sizeof(uint64_t) + sizeof(TSKEY) + sizeof(TSKEY);
+  pBlock->info.watermark = INT64_MIN;
+
+  pBlock->pDataBlock = taosArrayInit(6, sizeof(SColumnInfoData));
+  SColumnInfoData infoData = {0};
+  infoData.info.type = TSDB_DATA_TYPE_TIMESTAMP;
+  infoData.info.bytes = sizeof(TSKEY);
+  // window start ts
+  taosArrayPush(pBlock->pDataBlock, &infoData);
+  // window end ts
+  taosArrayPush(pBlock->pDataBlock, &infoData);
+
+  infoData.info.type = TSDB_DATA_TYPE_UBIGINT;
+  infoData.info.bytes = sizeof(uint64_t);
+  // uid
+  taosArrayPush(pBlock->pDataBlock, &infoData);
+  // group id
+  taosArrayPush(pBlock->pDataBlock, &infoData);
+
+  // calculate start ts
+  taosArrayPush(pBlock->pDataBlock, &infoData);
+  // calculate end ts
+  taosArrayPush(pBlock->pDataBlock, &infoData);
+
+  return pBlock;
+}
+
 SSDataBlock* createOneDataBlock(const SSDataBlock* pDataBlock, bool copyData) {
   if (pDataBlock == NULL) {
     return NULL;
@@ -1426,7 +1465,7 @@ size_t blockDataGetCapacityInRow(const SSDataBlock* pBlock, size_t pageSize) {
 }
 
 void colDataDestroy(SColumnInfoData* pColData) {
-  if(!pColData) return;
+  if (!pColData) return;
   if (IS_VAR_DATA_TYPE(pColData->info.type)) {
     taosMemoryFreeClear(pColData->varmeta.offset);
   } else {
@@ -1693,7 +1732,7 @@ static char* formatTimestamp(char* buf, int64_t val, int precision) {
   }
   struct tm ptm = {0};
   taosLocalTime(&tt, &ptm);
-  size_t     pos = strftime(buf, 35, "%Y-%m-%d %H:%M:%S", &ptm);
+  size_t pos = strftime(buf, 35, "%Y-%m-%d %H:%M:%S", &ptm);
 
   if (precision == TSDB_TIME_PRECISION_NANO) {
     sprintf(buf + pos, ".%09d", ms);
@@ -1847,20 +1886,20 @@ char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf)
           break;
         case TSDB_DATA_TYPE_VARCHAR: {
           memset(pBuf, 0, sizeof(pBuf));
-          char* pData = colDataGetVarData(pColInfoData, j);
+          char*   pData = colDataGetVarData(pColInfoData, j);
           int32_t dataSize = TMIN(sizeof(pBuf), varDataLen(pData));
           memcpy(pBuf, varDataVal(pData), dataSize);
           len += snprintf(dumpBuf + len, size - len, " %15s |", pBuf);
           if (len >= size - 1) return dumpBuf;
-          } break;
+        } break;
         case TSDB_DATA_TYPE_NCHAR: {
-          char* pData = colDataGetVarData(pColInfoData, j);
+          char*   pData = colDataGetVarData(pColInfoData, j);
           int32_t dataSize = TMIN(sizeof(pBuf), varDataLen(pData));
           memset(pBuf, 0, sizeof(pBuf));
-          taosUcs4ToMbs((TdUcs4 *)varDataVal(pData), dataSize, pBuf);
+          taosUcs4ToMbs((TdUcs4*)varDataVal(pData), dataSize, pBuf);
           len += snprintf(dumpBuf + len, size - len, " %15s |", pBuf);
           if (len >= size - 1) return dumpBuf;
-          } break;
+        } break;
       }
     }
     len += snprintf(dumpBuf + len, size - len, "\n");
@@ -1877,7 +1916,7 @@ char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf)
  * @param pDataBlocks
  * @param vgId
  * @param suid
- * 
+ *
  */
 int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SSDataBlock* pDataBlock, STSchema* pTSchema, int32_t vgId,
                                     tb_uid_t suid) {
@@ -1904,8 +1943,8 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SSDataBlock* pDataB
   tdSRowInit(&rb, pTSchema->version);
 
   for (int32_t i = 0; i < sz; ++i) {
-    int32_t      colNum = taosArrayGetSize(pDataBlock->pDataBlock);
-    int32_t      rows = pDataBlock->info.rows;
+    int32_t colNum = taosArrayGetSize(pDataBlock->pDataBlock);
+    int32_t rows = pDataBlock->info.rows;
     //    int32_t      rowSize = pDataBlock->info.rowSize;
     //    int64_t      groupId = pDataBlock->info.groupId;
 
@@ -1926,7 +1965,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SSDataBlock* pDataB
 
     msgLen += sizeof(SSubmitBlk);
     int32_t dataLen = 0;
-    for (int32_t j = 0; j < rows; ++j) {                     // iterate by row
+    for (int32_t j = 0; j < rows; ++j) {                               // iterate by row
       tdSRowResetBuf(&rb, POINTER_SHIFT(pDataBuf, msgLen + dataLen));  // set row buf
       bool    isStartKey = false;
       int32_t offset = 0;
@@ -2089,7 +2128,7 @@ void blockEncode(const SSDataBlock* pBlock, char* data, int32_t* dataLen, int32_
   // flag segment.
   // the inital bit is for column info
   int32_t* flagSegment = (int32_t*)data;
-  *flagSegment = (1<<31);
+  *flagSegment = (1 << 31);
 
   data += sizeof(int32_t);
 
@@ -2149,7 +2188,7 @@ void blockEncode(const SSDataBlock* pBlock, char* data, int32_t* dataLen, int32_
 const char* blockDecode(SSDataBlock* pBlock, const char* pData) {
   const char* pStart = pData;
 
-  int32_t version = *(int32_t*) pStart;
+  int32_t version = *(int32_t*)pStart;
   pStart += sizeof(int32_t);
   ASSERT(version == 1);
 
@@ -2158,7 +2197,7 @@ const char* blockDecode(SSDataBlock* pBlock, const char* pData) {
   pStart += sizeof(int32_t);
 
   // total rows sizeof(int32_t)
-  int32_t numOfRows =  *(int32_t*)pStart;
+  int32_t numOfRows = *(int32_t*)pStart;
   pStart += sizeof(int32_t);
 
   // total columns sizeof(int32_t)

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

@@ -102,7 +102,7 @@ int             metaCommit(SMeta* pMeta);
 int             metaCreateSTable(SMeta* pMeta, int64_t version, SVCreateStbReq* pReq);
 int             metaAlterSTable(SMeta* pMeta, int64_t version, SVCreateStbReq* pReq);
 int             metaDropSTable(SMeta* pMeta, int64_t verison, SVDropStbReq* pReq, SArray* tbUidList);
-int             metaCreateTable(SMeta* pMeta, int64_t version, SVCreateTbReq* pReq, STableMetaRsp **pMetaRsp);
+int             metaCreateTable(SMeta* pMeta, int64_t version, SVCreateTbReq* pReq, STableMetaRsp** pMetaRsp);
 int             metaDropTable(SMeta* pMeta, int64_t version, SVDropTbReq* pReq, SArray* tbUids);
 int             metaTtlDropTable(SMeta* pMeta, int64_t ttl, SArray* tbUids);
 int             metaAlterTable(SMeta* pMeta, int64_t version, SVAlterTbReq* pReq, STableMetaRsp* pMetaRsp);
@@ -173,7 +173,8 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg);
 // tq-stream
 int32_t tqProcessTaskDeployReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
 int32_t tqProcessTaskDropReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
-int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* data, int64_t ver);
+int32_t tqProcessSubmitReq(STQ* pTq, SSubmitReq* data, int64_t ver);
+int32_t tqProcessDelReq(STQ* pTq, void* pReq, int32_t len, int64_t ver);
 int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskDispatchReq(STQ* pTq, SRpcMsg* pMsg, bool exec);
 int32_t tqProcessTaskRecoverReq(STQ* pTq, SRpcMsg* pMsg);

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

@@ -816,7 +816,86 @@ int32_t tqProcessTaskDeployReq(STQ* pTq, int64_t version, char* msg, int32_t msg
   return streamMetaAddSerializedTask(pTq->pStreamMeta, version, msg, msgLen);
 }
 
-int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* pReq, int64_t ver) {
+int32_t tqProcessDelReq(STQ* pTq, void* pReq, int32_t len, int64_t ver) {
+  bool        failed = false;
+  SDecoder*   pCoder = &(SDecoder){0};
+  SDeleteRes* pRes = &(SDeleteRes){0};
+
+  pRes->uidList = taosArrayInit(0, sizeof(tb_uid_t));
+  if (pRes->uidList == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    failed = true;
+  }
+
+  tDecoderInit(pCoder, pReq, len);
+  tDecodeDeleteRes(pCoder, pRes);
+  tDecoderClear(pCoder);
+
+  int32_t sz = taosArrayGetSize(pRes->uidList);
+  if (sz == 0) {
+    taosArrayDestroy(pRes->uidList);
+    return 0;
+  }
+  SSDataBlock* pDelBlock = createSpecialDataBlock(STREAM_DELETE_DATA);
+  blockDataEnsureCapacity(pDelBlock, sz);
+  pDelBlock->info.rows = sz;
+  pDelBlock->info.version = ver;
+
+  for (int32_t i = 0; i < sz; i++) {
+    // start key column
+    SColumnInfoData* pStartCol = taosArrayGet(pDelBlock->pDataBlock, START_TS_COLUMN_INDEX);
+    colDataAppend(pStartCol, i, (const char*)&pRes->skey, false);  // end key column
+    SColumnInfoData* pEndCol = taosArrayGet(pDelBlock->pDataBlock, END_TS_COLUMN_INDEX);
+    colDataAppend(pEndCol, i, (const char*)&pRes->ekey, false);
+    // uid column
+    SColumnInfoData* pUidCol = taosArrayGet(pDelBlock->pDataBlock, UID_COLUMN_INDEX);
+    int64_t*         pUid = taosArrayGet(pRes->uidList, i);
+    colDataAppend(pUidCol, i, (const char*)pUid, false);
+
+    colDataAppendNULL(taosArrayGet(pDelBlock->pDataBlock, GROUPID_COLUMN_INDEX), i);
+    colDataAppendNULL(taosArrayGet(pDelBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX), i);
+    colDataAppendNULL(taosArrayGet(pDelBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX), i);
+  }
+
+  taosArrayDestroy(pRes->uidList);
+
+  void* pIter = NULL;
+  while (1) {
+    pIter = taosHashIterate(pTq->pStreamMeta->pTasks, pIter);
+    if (pIter == NULL) break;
+    SStreamTask* pTask = *(SStreamTask**)pIter;
+    if (pTask->taskLevel != TASK_LEVEL__SOURCE) continue;
+
+    qDebug("delete req enqueue stream task: %d, ver: %" PRId64, pTask->taskId, ver);
+
+    SStreamDataBlock* pStreamBlock = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM);
+    pStreamBlock->type = STREAM_INPUT__DATA_BLOCK;
+    pStreamBlock->blocks = taosArrayInit(0, sizeof(SSDataBlock));
+    SSDataBlock block = {0};
+    assignOneDataBlock(&block, pDelBlock);
+    block.info.type = STREAM_DELETE_DATA;
+    taosArrayPush(pStreamBlock->blocks, &block);
+
+    if (!failed) {
+      if (streamTaskInput(pTask, (SStreamQueueItem*)pStreamBlock) < 0) {
+        qError("stream task input del failed, task id %d", pTask->taskId);
+        continue;
+      }
+
+      if (streamSchedExec(pTask) < 0) {
+        qError("stream task launch failed, task id %d", pTask->taskId);
+        continue;
+      }
+    } else {
+      streamTaskInputFail(pTask);
+    }
+  }
+  blockDataDestroy(pDelBlock);
+
+  return 0;
+}
+
+int32_t tqProcessSubmitReq(STQ* pTq, SSubmitReq* pReq, int64_t ver) {
   void*              pIter = NULL;
   bool               failed = false;
   SStreamDataSubmit* pSubmit = NULL;

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

@@ -213,20 +213,25 @@ int32_t tqPushMsgNew(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_
 #endif
 
 int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) {
-  if (vnodeIsRoleLeader(pTq->pVnode) && msgType == TDMT_VND_SUBMIT) {
-    if (taosHashGetSize(pTq->pStreamMeta->pTasks) == 0) return 0;
-
-    void* data = taosMemoryMalloc(msgLen);
-    if (data == NULL) {
-      terrno = TSDB_CODE_OUT_OF_MEMORY;
-      tqError("failed to copy data for stream since out of memory");
-      return -1;
+  if (vnodeIsRoleLeader(pTq->pVnode)) {
+    if (msgType == TDMT_VND_SUBMIT) {
+      if (taosHashGetSize(pTq->pStreamMeta->pTasks) == 0) return 0;
+
+      void* data = taosMemoryMalloc(msgLen);
+      if (data == NULL) {
+        terrno = TSDB_CODE_OUT_OF_MEMORY;
+        tqError("failed to copy data for stream since out of memory");
+        return -1;
+      }
+      memcpy(data, msg, msgLen);
+      SSubmitReq* pReq = (SSubmitReq*)data;
+      pReq->version = ver;
+
+      tqProcessSubmitReq(pTq, data, ver);
+    }
+    if (msgType == TDMT_VND_DELETE) {
+      tqProcessDelReq(pTq, POINTER_SHIFT(msg, sizeof(SMsgHead)), msgLen - sizeof(SMsgHead), ver);
     }
-    memcpy(data, msg, msgLen);
-    SSubmitReq* pReq = (SSubmitReq*)data;
-    pReq->version = ver;
-
-    tqProcessStreamTrigger(pTq, data, ver);
   }
 
   return 0;

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

@@ -371,7 +371,7 @@ void vnodeUpdateMetaRsp(SVnode *pVnode, STableMetaRsp *pMetaRsp) {
   if (NULL == pMetaRsp) {
     return;
   }
-  
+
   strcpy(pMetaRsp->dbFName, pVnode->config.dbname);
   pMetaRsp->dbId = pVnode->config.dbId;
   pMetaRsp->vgId = TD_VID(pVnode);
@@ -527,7 +527,7 @@ static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pR
       cRsp.code = TSDB_CODE_SUCCESS;
       tdFetchTbUidList(pVnode->pSma, &pStore, pCreateReq->ctb.suid, pCreateReq->uid);
       taosArrayPush(tbUids, &pCreateReq->uid);
-      vnodeUpdateMetaRsp(pVnode, cRsp.pMeta); 
+      vnodeUpdateMetaRsp(pVnode, cRsp.pMeta);
     }
 
     taosArrayPush(rsp.pArray, &cRsp);
@@ -1107,6 +1107,7 @@ static int32_t vnodeProcessDeleteReq(SVnode *pVnode, int64_t version, void *pReq
 
   tDecoderInit(pCoder, pReq, len);
   tDecodeDeleteRes(pCoder, pRes);
+  ASSERT(taosArrayGetSize(pRes->uidList) == 0 || (pRes->skey != 0 && pRes->ekey != 0));
 
   for (int32_t iUid = 0; iUid < taosArrayGetSize(pRes->uidList); iUid++) {
     code = tsdbDeleteTableData(pVnode->pTsdb, version, pRes->suid, *(uint64_t *)taosArrayGet(pRes->uidList, iUid),

source/libs/executor/src/executor.c

@@ -52,7 +52,11 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
     // TODO: if a block was set but not consumed,
     // prevent setting a different type of block
     pInfo->validBlockIndex = 0;
-    taosArrayClear(pInfo->pBlockLists);
+    if (pInfo->blockType == STREAM_INPUT__DATA_BLOCK) {
+      taosArrayClearP(pInfo->pBlockLists, taosMemoryFree);
+    } else {
+      taosArrayClear(pInfo->pBlockLists);
+    }
 
     if (type == STREAM_INPUT__MERGED_SUBMIT) {
       // ASSERT(numOfBlocks > 1);

source/libs/executor/src/scanoperator.c

@@ -955,9 +955,7 @@ void resetTableScanInfo(STableScanInfo* pTableScanInfo, STimeWindow* pWin) {
   pTableScanInfo->currentGroupId = -1;
 }
 
-static void freeArray(void* array) {
-  taosArrayDestroy(array);
-}
+static void freeArray(void* array) { taosArrayDestroy(array); }
 
 static void resetTableScanOperator(SOperatorInfo* pTableScanOp) {
   STableScanInfo* pTableScanInfo = pTableScanOp->info;
@@ -971,15 +969,16 @@ static void resetTableScanOperator(SOperatorInfo* pTableScanOp) {
   resetTableScanInfo(pTableScanOp->info, &win);
 }
 
-static SSDataBlock* readPreVersionData(SOperatorInfo* pTableScanOp, uint64_t tbUid, TSKEY startTs, TSKEY endTs, int64_t maxVersion) {
+static SSDataBlock* readPreVersionData(SOperatorInfo* pTableScanOp, uint64_t tbUid, TSKEY startTs, TSKEY endTs,
+                                       int64_t maxVersion) {
   SArray* gpTbls = pTableScanOp->pTaskInfo->tableqinfoList.pGroupList;
   taosArrayClear(gpTbls);
   STableKeyInfo tblInfo = {.uid = tbUid, .groupId = 0};
-  SArray* tbls = taosArrayInit(1, sizeof(STableKeyInfo));
+  SArray*       tbls = taosArrayInit(1, sizeof(STableKeyInfo));
   taosArrayPush(tbls, &tblInfo);
   taosArrayPush(gpTbls, &tbls);
 
-  STimeWindow win = {.skey = startTs, .ekey = endTs};
+  STimeWindow     win = {.skey = startTs, .ekey = endTs};
   STableScanInfo* pTableScanInfo = pTableScanOp->info;
   pTableScanInfo->cond.startVersion = -1;
   pTableScanInfo->cond.endVersion = maxVersion;
@@ -1068,8 +1067,8 @@ static STimeWindow getSlidingWindow(TSKEY* tsCol, SInterval* pInterval, SDataBlo
     if (hasGroup) {
       (*pRowIndex) += 1;
     } else {
-      (*pRowIndex) += getNumOfRowsInTimeWindow(pDataBlockInfo, tsCol, *pRowIndex, endWin.ekey,
-          binarySearchForKey, NULL, TSDB_ORDER_ASC);
+      (*pRowIndex) += getNumOfRowsInTimeWindow(pDataBlockInfo, tsCol, *pRowIndex, endWin.ekey, binarySearchForKey, NULL,
+                                               TSDB_ORDER_ASC);
     }
     do {
       preWin = endWin;
@@ -1110,8 +1109,8 @@ static SSDataBlock* doRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32
           for (int32_t j = 0; j < pInfo->pTableScanOp->exprSupp.numOfExprs; j++) {
             SColumnInfoData* pSrcCol = taosArrayGet(tmpBlock->pDataBlock, j);
             SColumnInfoData* pDestCol = taosArrayGet(pResult->pDataBlock, j);
-            bool isNull = colDataIsNull(pSrcCol, tmpBlock->info.rows, i, NULL);
-            char* pSrcData = colDataGetData(pSrcCol, i);
+            bool             isNull = colDataIsNull(pSrcCol, tmpBlock->info.rows, i, NULL);
+            char*            pSrcData = colDataGetData(pSrcCol, i);
             colDataAppend(pDestCol, pResult->info.rows, pSrcData, isNull);
           }
           pResult->info.rows++;
@@ -1152,7 +1151,7 @@ static int32_t generateSessionScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSr
   SColumnInfoData* pDestCalStartTsCol = taosArrayGet(pDestBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);
   SColumnInfoData* pDestCalEndTsCol = taosArrayGet(pDestBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);
   int32_t          dummy = 0;
-  int64_t version = pSrcBlock->info.version - 1;
+  int64_t          version = pSrcBlock->info.version - 1;
   for (int32_t i = 0; i < pSrcBlock->info.rows; i++) {
     uint64_t groupId = getGroupIdByData(pInfo, uidCol[i], startData[i], version);
     // gap must be 0.
@@ -1201,15 +1200,15 @@ static int32_t generateIntervalScanRange(SStreamScanInfo* pInfo, SSDataBlock* pS
   SColumnInfoData* pGpCol = taosArrayGet(pDestBlock->pDataBlock, GROUPID_COLUMN_INDEX);
   SColumnInfoData* pCalStartTsCol = taosArrayGet(pDestBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);
   SColumnInfoData* pCalEndTsCol = taosArrayGet(pDestBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);
-  int64_t version = pSrcBlock->info.version - 1;
+  int64_t          version = pSrcBlock->info.version - 1;
   for (int32_t i = 0; i < rows;) {
     uint64_t srcUid = srcUidData[i];
     uint64_t groupId = getGroupIdByData(pInfo, srcUid, tsCol[i], version);
     uint64_t srcGpId = srcGp[i];
-    TSKEY calStartTs = tsCol[i];
+    TSKEY    calStartTs = tsCol[i];
     colDataAppend(pCalStartTsCol, pDestBlock->info.rows, (const char*)(&calStartTs), false);
     STimeWindow win = getSlidingWindow(tsCol, &pInfo->interval, &pSrcBlock->info, &i, pInfo->partitionSup.needCalc);
-    TSKEY calEndTs = tsCol[i - 1];
+    TSKEY       calEndTs = tsCol[i - 1];
     colDataAppend(pCalEndTsCol, pDestBlock->info.rows, (const char*)(&calEndTs), false);
     colDataAppend(pDeUidCol, pDestBlock->info.rows, (const char*)(&srcUid), false);
     colDataAppend(pStartTsCol, pDestBlock->info.rows, (const char*)(&win.skey), false);
@@ -1277,10 +1276,10 @@ static void checkUpdateData(SStreamScanInfo* pInfo, bool invertible, SSDataBlock
     bool closedWin = isClosed && isSignleIntervalWindow(pInfo) &&
                      isDeletedWindow(&win, pBlock->info.groupId, pInfo->windowSup.pIntervalAggSup);
     if ((update || closedWin) && out) {
-      uint64_t gpId = closedWin&&pInfo->partitionSup.needCalc ?
-          calGroupIdByData(&pInfo->partitionSup, pInfo->pPartScalarSup, pBlock, rowId) : 0;
-      appendOneRow(pInfo->pUpdateDataRes, tsCol + rowId, tsCol + rowId, &pBlock->info.uid,
-          &gpId);
+      uint64_t gpId = closedWin && pInfo->partitionSup.needCalc
+                          ? calGroupIdByData(&pInfo->partitionSup, pInfo->pPartScalarSup, pBlock, rowId)
+                          : 0;
+      appendOneRow(pInfo->pUpdateDataRes, tsCol + rowId, tsCol + rowId, &pBlock->info.uid, &gpId);
     }
   }
   if (out && pInfo->pUpdateDataRes->info.rows > 0) {
@@ -1515,6 +1514,7 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
         generateScanRange(pInfo, pBlock, pInfo->pUpdateRes);
         prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex);
         copyDataBlock(pInfo->pDeleteDataRes, pInfo->pUpdateRes);
+        pInfo->pDeleteDataRes->info.type = STREAM_DELETE_DATA;
         pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE;
         return pInfo->pDeleteDataRes;
       } break;
@@ -1915,7 +1915,7 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
       ASSERT(pHandle->tqReader);
       pInfo->tqReader = pHandle->tqReader;
     }
-    
+
     pInfo->pUpdateInfo = NULL;
     pInfo->pTableScanOp = pTableScanOp;
     pInfo->interval = pTSInfo->pdInfo.interval;
@@ -1947,8 +1947,7 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
   pInfo->pUpdateRes = createSpecialDataBlock(STREAM_CLEAR);
   pInfo->pCondition = pScanPhyNode->node.pConditions;
   pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
-  pInfo->windowSup =
-      (SWindowSupporter){.pStreamAggSup = NULL, .gap = -1, .parentType = QUERY_NODE_PHYSICAL_PLAN};
+  pInfo->windowSup = (SWindowSupporter){.pStreamAggSup = NULL, .gap = -1, .parentType = QUERY_NODE_PHYSICAL_PLAN};
   pInfo->groupId = 0;
   pInfo->pPullDataRes = createSpecialDataBlock(STREAM_RETRIEVE);
   pInfo->pStreamScanOp = pOperator;

source/libs/function/src/builtinsimpl.c

@@ -76,11 +76,11 @@ typedef struct STopBotResItem {
 } STopBotResItem;
 
 typedef struct STopBotRes {
-  int32_t         maxSize;
-  int16_t         type;
+  int32_t maxSize;
+  int16_t type;
 
-  STuplePos       nullTuplePos;
-  bool            nullTupleSaved;
+  STuplePos nullTuplePos;
+  bool      nullTupleSaved;
 
   STopBotResItem* pItems;
 } STopBotRes;
@@ -223,14 +223,14 @@ typedef struct SMavgInfo {
 } SMavgInfo;
 
 typedef struct SSampleInfo {
-  int32_t    samples;
-  int32_t    totalPoints;
-  int32_t    numSampled;
-  uint8_t    colType;
-  int16_t    colBytes;
+  int32_t samples;
+  int32_t totalPoints;
+  int32_t numSampled;
+  uint8_t colType;
+  int16_t colBytes;
 
-  STuplePos  nullTuplePos;
-  bool       nullTupleSaved;
+  STuplePos nullTuplePos;
+  bool      nullTupleSaved;
 
   char*      data;
   STuplePos* tuplePos;
@@ -1147,7 +1147,7 @@ bool getMinmaxFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
 }
 
 static STuplePos saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock);
-static int32_t updateTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STuplePos* pPos);
+static int32_t   updateTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STuplePos* pPos);
 static const char* loadTupleData(SqlFunctionCtx* pCtx, const STuplePos* pPos);
 
 static int32_t findRowIndex(int32_t start, int32_t num, SColumnInfoData* pCol, const char* tval) {
@@ -1357,8 +1357,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
 
         numOfElems += 1;
       }
-    } else if (type == TSDB_DATA_TYPE_BIGINT ||
-               type == TSDB_DATA_TYPE_TIMESTAMP) {
+    } else if (type == TSDB_DATA_TYPE_BIGINT || type == TSDB_DATA_TYPE_TIMESTAMP) {
       int64_t* pData = (int64_t*)pCol->pData;
       int64_t* val = (int64_t*)&pBuf->v;
 
@@ -1581,7 +1580,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
   }
 
 _min_max_over:
-  if (numOfElems == 0 && pCtx->subsidiaries.num > 0 && !pBuf->nullTupleSaved ) {
+  if (numOfElems == 0 && pCtx->subsidiaries.num > 0 && !pBuf->nullTupleSaved) {
     pBuf->nullTuplePos = saveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock);
     pBuf->nullTupleSaved = true;
   }
@@ -1601,7 +1600,8 @@ int32_t maxFunction(SqlFunctionCtx* pCtx) {
 }
 
 static void setNullSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t rowIndex);
-static void setSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, const STuplePos* pTuplePos, int32_t rowIndex);
+static void setSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, const STuplePos* pTuplePos,
+                                int32_t rowIndex);
 
 int32_t minmaxFunctionFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(pCtx);
@@ -1651,7 +1651,7 @@ void setSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, const STuple
 
   if (pCtx->saveHandle.pBuf != NULL) {
     if (pTuplePos->pageId != -1) {
-      int32_t    numOfCols = pCtx->subsidiaries.num;
+      int32_t     numOfCols = pCtx->subsidiaries.num;
       const char* p = loadTupleData(pCtx, pTuplePos);
 
       bool* nullList = (bool*)p;
@@ -1660,7 +1660,7 @@ void setSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, const STuple
       // todo set the offset value to optimize the performance.
       for (int32_t j = 0; j < numOfCols; ++j) {
         SqlFunctionCtx* pc = pCtx->subsidiaries.pCtx[j];
-        int32_t      dstSlotId = pc->pExpr->base.resSchema.slotId;
+        int32_t         dstSlotId = pc->pExpr->base.resSchema.slotId;
 
         SColumnInfoData* pDstCol = taosArrayGet(pBlock->pDataBlock, dstSlotId);
         ASSERT(pc->pExpr->base.resSchema.bytes == pDstCol->info.bytes);
@@ -1701,7 +1701,7 @@ void appendSelectivityValue(SqlFunctionCtx* pCtx, int32_t rowIndex, int32_t pos)
     char* pData = colDataGetData(pSrcCol, rowIndex);
 
     // append to dest col
-    int32_t  dstSlotId = pc->pExpr->base.resSchema.slotId;
+    int32_t dstSlotId = pc->pExpr->base.resSchema.slotId;
 
     SColumnInfoData* pDstCol = taosArrayGet(pCtx->pDstBlock->pDataBlock, dstSlotId);
     ASSERT(pc->pExpr->base.resSchema.bytes == pDstCol->info.bytes);
@@ -1712,7 +1712,6 @@ void appendSelectivityValue(SqlFunctionCtx* pCtx, int32_t rowIndex, int32_t pos)
       colDataAppend(pDstCol, pos, pData, false);
     }
   }
-
 }
 
 void replaceTupleData(STuplePos* pDestPos, STuplePos* pSourcePos) {
@@ -2590,8 +2589,8 @@ static void apercentileTransferInfo(SAPercentileInfo* pInput, SAPercentileInfo*
       memcpy(pHisto, pInput->pHisto, sizeof(SHistogramInfo) + sizeof(SHistBin) * (MAX_HISTOGRAM_BIN + 1));
       pHisto->elems = (SHistBin*)((char*)pHisto + sizeof(SHistogramInfo));
 
-      qDebug("%s merge histo, total:%" PRId64 ", entry:%d, %p", __FUNCTION__, pHisto->numOfElems,
-              pHisto->numOfEntries, pHisto);
+      qDebug("%s merge histo, total:%" PRId64 ", entry:%d, %p", __FUNCTION__, pHisto->numOfElems, pHisto->numOfEntries,
+             pHisto);
     } else {
       pHisto->elems = (SHistBin*)((char*)pHisto + sizeof(SHistogramInfo));
       qDebug("%s input histogram, elem:%" PRId64 ", entry:%d, %p", __FUNCTION__, pHisto->numOfElems,
@@ -2601,8 +2600,8 @@ static void apercentileTransferInfo(SAPercentileInfo* pInput, SAPercentileInfo*
       memcpy(pHisto, pRes, sizeof(SHistogramInfo) + sizeof(SHistBin) * MAX_HISTOGRAM_BIN);
       pHisto->elems = (SHistBin*)((char*)pHisto + sizeof(SHistogramInfo));
 
-      qDebug("%s merge histo, total:%" PRId64 ", entry:%d, %p", __FUNCTION__, pHisto->numOfElems,
-             pHisto->numOfEntries, pHisto);
+      qDebug("%s merge histo, total:%" PRId64 ", entry:%d, %p", __FUNCTION__, pHisto->numOfElems, pHisto->numOfEntries,
+             pHisto);
       tHistogramDestroy(&pRes);
     }
   }
@@ -2629,8 +2628,8 @@ int32_t apercentileFunctionMerge(SqlFunctionCtx* pCtx) {
   }
 
   if (pInfo->algo != APERCT_ALGO_TDIGEST) {
-    qDebug("%s after merge, total:%d, numOfEntry:%d, %p", __FUNCTION__, pInfo->pHisto->numOfElems, pInfo->pHisto->numOfEntries,
-           pInfo->pHisto);
+    qDebug("%s after merge, total:%d, numOfEntry:%d, %p", __FUNCTION__, pInfo->pHisto->numOfElems,
+           pInfo->pHisto->numOfEntries, pInfo->pHisto);
   }
 
   SET_VAL(pResInfo, 1, 1);
@@ -2709,7 +2708,7 @@ int32_t apercentileCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx)
 }
 
 EFuncDataRequired lastDynDataReq(void* pRes, STimeWindow* pTimeWindow) {
-  SResultRowEntryInfo* pEntry = (SResultRowEntryInfo*) pRes;
+  SResultRowEntryInfo* pEntry = (SResultRowEntryInfo*)pRes;
 
   // not initialized yet, data is required
   if (pEntry == NULL) {
@@ -2752,7 +2751,8 @@ static FORCE_INLINE TSKEY getRowPTs(SColumnInfoData* pTsColInfo, int32_t rowInde
   return *(TSKEY*)colDataGetData(pTsColInfo, rowIndex);
 }
 
-static void firstlastSaveTupleData(const SSDataBlock* pSrcBlock, int32_t rowIndex, SqlFunctionCtx* pCtx, SFirstLastRes* pInfo) {
+static void firstlastSaveTupleData(const SSDataBlock* pSrcBlock, int32_t rowIndex, SqlFunctionCtx* pCtx,
+                                   SFirstLastRes* pInfo) {
   if (pCtx->subsidiaries.num <= 0) {
     return;
   }
@@ -3176,7 +3176,7 @@ bool diffFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResInfo) {
 static void doSetPrevVal(SDiffInfo* pDiffInfo, int32_t type, const char* pv) {
   switch (type) {
     case TSDB_DATA_TYPE_BOOL:
-      pDiffInfo->prev.i64 = *(bool*)pv? 1:0;
+      pDiffInfo->prev.i64 = *(bool*)pv ? 1 : 0;
       break;
     case TSDB_DATA_TYPE_TINYINT:
       pDiffInfo->prev.i64 = *(int8_t*)pv;
@@ -3537,7 +3537,8 @@ void doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSData
  * |(n columns, one bit for each column)| src column #1| src column #2|
  * +------------------------------------+--------------+--------------+
  */
-void* serializeTupleData(const SSDataBlock* pSrcBlock, int32_t rowIndex, SSubsidiaryResInfo* pSubsidiaryies, char* buf) {
+void* serializeTupleData(const SSDataBlock* pSrcBlock, int32_t rowIndex, SSubsidiaryResInfo* pSubsidiaryies,
+                         char* buf) {
   char* nullList = buf;
   char* pStart = (char*)(nullList + sizeof(bool) * pSubsidiaryies->num);
 
@@ -3585,7 +3586,7 @@ static STuplePos doSaveTupleData(SSerializeDataHandle* pHandle, const void* pBuf
       }
     }
 
-    p = (STuplePos) {.pageId = pHandle->currentPage, .offset = pPage->num};
+    p = (STuplePos){.pageId = pHandle->currentPage, .offset = pPage->num};
     memcpy(pPage->data + pPage->num, pBuf, length);
 
     pPage->num += length;
@@ -3621,7 +3622,6 @@ static int32_t doUpdateTupleData(SSerializeDataHandle* pHandle, const void* pBuf
     setBufPageDirty(pPage, true);
     releaseBufPage(pHandle->pBuf, pPage);
   } else {
-
   }
 
   return TSDB_CODE_SUCCESS;
@@ -3636,7 +3636,7 @@ static int32_t updateTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSD
 static char* doLoadTupleData(SSerializeDataHandle* pHandle, const STuplePos* pPos) {
   if (pHandle->pBuf != NULL) {
     SFilePage* pPage = getBufPage(pHandle->pBuf, pPos->pageId);
-    char* p = pPage->data + pPos->offset;
+    char*      p = pPage->data + pPos->offset;
     releaseBufPage(pHandle->pBuf, pPage);
     return p;
   } else {
@@ -3980,8 +3980,8 @@ int32_t elapsedFunction(SqlFunctionCtx* pCtx) {
       }
 
       if (pCtx->end.key == INT64_MIN) {
-        pInfo->min = (pInfo->min > ptsList[start + pInput->numOfRows - 1]) ?
-                     ptsList[start + pInput->numOfRows - 1] : pInfo->min;
+        pInfo->min =
+            (pInfo->min > ptsList[start + pInput->numOfRows - 1]) ? ptsList[start + pInput->numOfRows - 1] : pInfo->min;
       } else {
         pInfo->min = pCtx->end.key;
       }
@@ -3993,8 +3993,8 @@ int32_t elapsedFunction(SqlFunctionCtx* pCtx) {
       }
 
       if (pCtx->end.key == INT64_MIN) {
-        pInfo->max = (pInfo->max < ptsList[start + pInput->numOfRows - 1]) ?
-                     ptsList[start + pInput->numOfRows - 1] : pInfo->max;
+        pInfo->max =
+            (pInfo->max < ptsList[start + pInput->numOfRows - 1]) ? ptsList[start + pInput->numOfRows - 1] : pInfo->max;
       } else {
         pInfo->max = pCtx->end.key + 1;
       }

source/libs/stream/src/streamDispatch.c

@@ -243,6 +243,36 @@ FAIL:
   return 0;
 }
 
+int32_t streamSearchAndAddBlock(SStreamTask* pTask, SStreamDispatchReq* pReqs, SSDataBlock* pDataBlock, int32_t vgSz,
+                                int64_t groupId) {
+  char*   ctbName = buildCtbNameByGroupId(pTask->shuffleDispatcher.stbFullName, groupId);
+  SArray* vgInfo = pTask->shuffleDispatcher.dbInfo.pVgroupInfos;
+
+  // TODO: get hash function by hashMethod
+  uint32_t hashValue = MurmurHash3_32(ctbName, strlen(ctbName));
+  taosMemoryFree(ctbName);
+  bool found = false;
+  // TODO: optimize search
+  int32_t j;
+  for (j = 0; j < vgSz; j++) {
+    SVgroupInfo* pVgInfo = taosArrayGet(vgInfo, j);
+    ASSERT(pVgInfo->vgId > 0);
+    if (hashValue >= pVgInfo->hashBegin && hashValue <= pVgInfo->hashEnd) {
+      if (streamAddBlockToDispatchMsg(pDataBlock, &pReqs[j]) < 0) {
+        return -1;
+      }
+      if (pReqs[j].blockNum == 0) {
+        atomic_add_fetch_32(&pTask->shuffleDispatcher.waitingRspCnt, 1);
+      }
+      pReqs[j].blockNum++;
+      found = true;
+      break;
+    }
+  }
+  ASSERT(found);
+  return 0;
+}
+
 int32_t streamDispatchAllBlocks(SStreamTask* pTask, const SStreamDataBlock* pData) {
   int32_t code = -1;
   int32_t blockNum = taosArrayGetSize(pData->blocks);
@@ -317,20 +347,10 @@ int32_t streamDispatchAllBlocks(SStreamTask* pTask, const SStreamDataBlock* pDat
 
     for (int32_t i = 0; i < blockNum; i++) {
       SSDataBlock* pDataBlock = taosArrayGet(pData->blocks, i);
-      char*        ctbName = buildCtbNameByGroupId(pTask->shuffleDispatcher.stbFullName, pDataBlock->info.groupId);
-
-      // TODO: get hash function by hashMethod
-      uint32_t hashValue = MurmurHash3_32(ctbName, strlen(ctbName));
-
-      taosMemoryFree(ctbName);
 
-      bool found = false;
-      // TODO: optimize search
-      int32_t j;
-      for (j = 0; j < vgSz; j++) {
-        SVgroupInfo* pVgInfo = taosArrayGet(vgInfo, j);
-        ASSERT(pVgInfo->vgId > 0);
-        if (hashValue >= pVgInfo->hashBegin && hashValue <= pVgInfo->hashEnd) {
+      // TODO: do not use broadcast
+      if (pDataBlock->info.type == STREAM_DELETE_RESULT) {
+        for (int32_t j = 0; j < vgSz; j++) {
           if (streamAddBlockToDispatchMsg(pDataBlock, &pReqs[j]) < 0) {
             goto FAIL_SHUFFLE_DISPATCH;
           }
@@ -338,11 +358,13 @@ int32_t streamDispatchAllBlocks(SStreamTask* pTask, const SStreamDataBlock* pDat
             atomic_add_fetch_32(&pTask->shuffleDispatcher.waitingRspCnt, 1);
           }
           pReqs[j].blockNum++;
-          found = true;
-          break;
         }
+        continue;
+      }
+
+      if (streamSearchAndAddBlock(pTask, pReqs, pDataBlock, vgSz, pDataBlock->info.groupId) < 0) {
+        goto FAIL_SHUFFLE_DISPATCH;
       }
-      ASSERT(found);
     }
 
     for (int32_t i = 0; i < vgSz; i++) {