xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix(stream): fix shuffle vg id not initialized

This commit is contained in:
Liu Jicong 2022-06-11 15:03:23 +08:00
parent 6f4f6c14fa
commit 767b768e51
2 changed files with 34 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
source/dnode/mnode/impl/src/mndScheduler.c
View File

@ -151,6 +151,9 @@ int32_t mndAddDispatcherToInnerTask(SMnode* pMnode, STrans* pTrans, SStreamObj*
ASSERT(pDb); ASSERT(pDb);
if (mndExtractDbInfo(pMnode, pDb, &pTask->shuffleDispatcher.dbInfo, NULL) < 0) { if (mndExtractDbInfo(pMnode, pDb, &pTask->shuffleDispatcher.dbInfo, NULL) < 0) {
ASSERT(0);
return -1;
}
sdbRelease(pMnode->pSdb, pDb); sdbRelease(pMnode->pSdb, pDb);
SArray* pVgs = pTask->shuffleDispatcher.dbInfo.pVgroupInfos; SArray* pVgs = pTask->shuffleDispatcher.dbInfo.pVgroupInfos;
@ -163,6 +166,7 @@ int32_t mndAddDispatcherToInnerTask(SMnode* pMnode, STrans* pTrans, SStreamObj*
SStreamTask* pLastLevelTask = taosArrayGetP(sinkLv, j); SStreamTask* pLastLevelTask = taosArrayGetP(sinkLv, j);
if (pLastLevelTask->nodeId == pVgInfo->vgId) { if (pLastLevelTask->nodeId == pVgInfo->vgId) {
pVgInfo->taskId = pLastLevelTask->taskId; pVgInfo->taskId = pLastLevelTask->taskId;
ASSERT(pVgInfo->taskId != 0);
break; break;
} }
} }
@ -178,7 +182,6 @@ int32_t mndAddDispatcherToInnerTask(SMnode* pMnode, STrans* pTrans, SStreamObj*
pTask->fixedEpDispatcher.nodeId = lastLevelTask->nodeId; pTask->fixedEpDispatcher.nodeId = lastLevelTask->nodeId;
pTask->fixedEpDispatcher.epSet = lastLevelTask->epSet; pTask->fixedEpDispatcher.epSet = lastLevelTask->epSet;
} }
}
return 0; return 0;
} }
@ -379,7 +382,10 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
pFinalTask->inputType = TASK_INPUT_TYPE__DATA_BLOCK; pFinalTask->inputType = TASK_INPUT_TYPE__DATA_BLOCK;
// dispatch // dispatch
mndAddDispatcherToInnerTask(pMnode, pTrans, pStream, pFinalTask); if (mndAddDispatcherToInnerTask(pMnode, pTrans, pStream, pFinalTask) < 0) {
qDestroyQueryPlan(pPlan);
return -1;
}
// exec // exec
pFinalTask->execType = TASK_EXEC__PIPE; pFinalTask->execType = TASK_EXEC__PIPE;

6
source/libs/stream/src/streamDispatch.c
View File

@ -100,7 +100,6 @@ int32_t streamBuildDispatchMsg(SStreamTask* pTask, SStreamDataBlock* data, SRpcM
.upstreamNodeId = pTask->nodeId, .upstreamNodeId = pTask->nodeId,
.blockNum = blockNum, .blockNum = blockNum,
}; };
qInfo("dispatch from task %d (child id %d)", pTask->taskId, pTask->childId);
req.data = taosArrayInit(blockNum, sizeof(void*)); req.data = taosArrayInit(blockNum, sizeof(void*));
req.dataLen = taosArrayInit(blockNum, sizeof(int32_t)); req.dataLen = taosArrayInit(blockNum, sizeof(int32_t));
@ -142,11 +141,14 @@ int32_t streamBuildDispatchMsg(SStreamTask* pTask, SStreamDataBlock* data, SRpcM
break; break;
} }
} }
ASSERT(vgId != 0);
} }
ASSERT(vgId != 0);
req.taskId = downstreamTaskId; req.taskId = downstreamTaskId;
qInfo("dispatch from task %d (child id %d) to down stream task %d in vnode %d", pTask->taskId, pTask->childId,
downstreamTaskId, vgId);
// serialize // serialize
int32_t tlen; int32_t tlen;
tEncodeSize(tEncodeStreamDispatchReq, &req, tlen, code); tEncodeSize(tEncodeStreamDispatchReq, &req, tlen, code);