diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 4a3c4b0c3f236557ffe4e28ff7cedce3e82093fa..d9c39cefe6215d31b850eb08f9a71e9eb03a9fc9 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -1205,6 +1205,7 @@ typedef struct { int8_t completed; // all results are returned to client int8_t precision; int8_t compressed; + int8_t streamBlockType; int32_t compLen; int32_t numOfRows; int32_t numOfCols; @@ -2493,14 +2494,14 @@ int32_t tSerializeSTableIndexReq(void* buf, int32_t bufLen, STableIndexReq* pReq int32_t tDeserializeSTableIndexReq(void* buf, int32_t bufLen, STableIndexReq* pReq); typedef struct { - int8_t intervalUnit; - int8_t slidingUnit; - int64_t interval; - int64_t offset; - int64_t sliding; - int64_t dstTbUid; - int32_t dstVgId; // for stream - char* expr; + int8_t intervalUnit; + int8_t slidingUnit; + int64_t interval; + int64_t offset; + int64_t sliding; + int64_t dstTbUid; + int32_t dstVgId; // for stream + char* expr; } STableIndexInfo; typedef struct { @@ -2510,7 +2511,6 @@ typedef struct { int32_t tSerializeSTableIndexRsp(void* buf, int32_t bufLen, const STableIndexRsp* pRsp); int32_t tDeserializeSTableIndexRsp(void* buf, int32_t bufLen, STableIndexRsp* pRsp); - typedef struct { int8_t mqMsgType; int32_t code; @@ -2751,8 +2751,8 @@ typedef struct { char* msg; } SVDeleteReq; -int32_t tSerializeSVDeleteReq(void *buf, int32_t bufLen, SVDeleteReq *pReq); -int32_t tDeserializeSVDeleteReq(void *buf, int32_t bufLen, SVDeleteReq *pReq); +int32_t tSerializeSVDeleteReq(void* buf, int32_t bufLen, SVDeleteReq* pReq); +int32_t tDeserializeSVDeleteReq(void* buf, int32_t bufLen, SVDeleteReq* pReq); typedef struct { int64_t affectedRows; diff --git a/source/dnode/mnode/impl/src/mndScheduler.c b/source/dnode/mnode/impl/src/mndScheduler.c index bac76a19d3154e51afb51b3572328843f455c5b3..55ff34f900f9b4921bc1406a04a6cd75e9d8a0a3 100644 --- a/source/dnode/mnode/impl/src/mndScheduler.c +++ b/source/dnode/mnode/impl/src/mndScheduler.c @@ -35,6 +35,13 @@ extern bool tsStreamSchedV; +static int32_t mndAddTaskToTaskSet(SArray* pArray, SStreamTask* pTask) { + int32_t childId = taosArrayGetSize(pArray); + pTask->childId = childId; + taosArrayPush(pArray, &pTask); + return 0; +} + int32_t mndConvertRSmaTask(const char* ast, int64_t uid, int8_t triggerType, int64_t watermark, char** pStr, int32_t* pLen, double filesFactor) { SNode* pAst = NULL; @@ -195,7 +202,7 @@ int32_t mndAddShuffledSinkToStream(SMnode* pMnode, STrans* pTrans, SStreamObj* p terrno = TSDB_CODE_OUT_OF_MEMORY; return -1; } - taosArrayPush(tasks, &pTask); + mndAddTaskToTaskSet(tasks, pTask); pTask->nodeId = pVgroup->vgId; pTask->epSet = mndGetVgroupEpset(pMnode, pVgroup); @@ -235,7 +242,7 @@ int32_t mndAddFixedSinkToStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStr terrno = TSDB_CODE_OUT_OF_MEMORY; return -1; } - taosArrayPush(tasks, &pTask); + mndAddTaskToTaskSet(tasks, pTask); pTask->nodeId = pStream->fixedSinkVgId; #if 0 @@ -378,7 +385,7 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) { return -1; } sdbRelease(pSdb, pVgroup); - taosArrayPush(taskOneLevel, &pTask); + mndAddTaskToTaskSet(taskOneLevel, pTask); } } else { // merge plan diff --git a/source/libs/stream/src/streamData.c b/source/libs/stream/src/streamData.c index 473ad3275228c9cc9098366942ce6a4011f82a9d..6699e86b1e6137cd9f7a5f873d207c23a0cc09f9 100644 --- a/source/libs/stream/src/streamData.c +++ b/source/libs/stream/src/streamData.c @@ -53,6 +53,7 @@ int32_t streamDispatchReqToData(const SStreamDispatchReq* pReq, SStreamDataBlock SSDataBlock* pDataBlock = taosArrayGet(pArray, i); blockCompressDecode(pDataBlock, htonl(pRetrieve->numOfCols), htonl(pRetrieve->numOfRows), pRetrieve->data); // TODO: refactor + pDataBlock->info.type = pRetrieve->streamBlockType; pDataBlock->info.childId = pReq->sourceChildId; } pData->blocks = pArray; diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c index 16da418677dad58bbfac70add0c625635a30da6f..38a3509e7b57b0c48a91786629f7fb3c2156da33 100644 --- a/source/libs/stream/src/streamDispatch.c +++ b/source/libs/stream/src/streamDispatch.c @@ -72,6 +72,7 @@ static int32_t streamAddBlockToDispatchMsg(const SSDataBlock* pBlock, SStreamDis pRetrieve->precision = TSDB_DEFAULT_PRECISION; pRetrieve->compressed = 0; pRetrieve->completed = 1; + pRetrieve->streamBlockType = pBlock->info.type; pRetrieve->numOfRows = htonl(pBlock->info.rows); pRetrieve->numOfCols = htonl(pBlock->info.numOfCols);