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

feat(stream): stream task and meta

This commit is contained in:
54liuyao 2022-10-19 09:24:18 +08:00
parent 05bf2bc4fa
commit 7f16081e29
3 changed files with 8 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/libs/stream/src/streamExec.c
View File

@ -137,6 +137,8 @@ int32_t streamPipelineExec(SStreamTask* pTask, int32_t batchNum, bool dispatch)
if (pTask->outputType == TASK_OUTPUT__FIXED_DISPATCH || pTask->outputType == TASK_OUTPUT__SHUFFLE_DISPATCH) {
streamDispatch(pTask);
}
} else {
taosArrayDestroyEx(pRes, (FDelete)blockDataFreeRes);
}
}

2
source/libs/stream/src/streamMeta.c
View File

@ -262,12 +262,14 @@ int32_t streamLoadTasks(SStreamMeta* pMeta) {
if (pMeta->expandFunc(pMeta->ahandle, pTask) < 0) {
tdbFree(pKey);
tdbFree(pVal);
tdbTbcClose(pCur);
return -1;
}
if (taosHashPut(pMeta->pTasks, &pTask->taskId, sizeof(int32_t), &pTask, sizeof(void*)) < 0) {
tdbFree(pKey);
tdbFree(pVal);
tdbTbcClose(pCur);
return -1;
}
}

5
source/libs/stream/src/streamTask.c
View File

@ -119,7 +119,10 @@ int32_t tDecodeSStreamTask(SDecoder* pDecoder, SStreamTask* pTask) {
for (int32_t i = 0; i < epSz; i++) {
SStreamChildEpInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamChildEpInfo));
if (pInfo == NULL) return -1;
if (tDecodeStreamEpInfo(pDecoder, pInfo) < 0) return -1;
if (tDecodeStreamEpInfo(pDecoder, pInfo) < 0) {
taosMemoryFreeClear(pInfo);
return -1;
}
taosArrayPush(pTask->childEpInfo, &pInfo);
}