From e31a304206bb04701f2be9fb6de67bbbe63eb576 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Sat, 11 Jun 2022 10:53:52 +0800 Subject: [PATCH 1/2] refactor(scheduler) --- include/libs/parser/parser.h | 2 +- source/dnode/mnode/impl/inc/mndScheduler.h | 6 +- source/dnode/mnode/impl/src/mndScheduler.c | 233 +-------------------- source/dnode/mnode/impl/src/mndStb.c | 4 +- source/dnode/mnode/impl/src/mndStream.c | 2 +- source/libs/parser/src/parser.c | 4 +- 6 files changed, 15 insertions(+), 236 deletions(-) diff --git a/include/libs/parser/parser.h b/include/libs/parser/parser.h index 6abd1ffa6d..ef28b832a0 100644 --- a/include/libs/parser/parser.h +++ b/include/libs/parser/parser.h @@ -63,7 +63,7 @@ int32_t qAnalyseSqlSemantic(SParseContext* pCxt, const struct SCatalogReq* pCata void qDestroyQuery(SQuery* pQueryNode); int32_t qExtractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema); -int32_t qSetSTableIdForRSma(SNode* pStmt, int64_t uid); +int32_t qSetSTableIdForRsma(SNode* pStmt, int64_t uid); int32_t qBuildStmtOutput(SQuery* pQuery, SHashObj* pVgHash, SHashObj* pBlockHash); int32_t qResetStmtDataBlock(void* block, bool keepBuf); diff --git a/source/dnode/mnode/impl/inc/mndScheduler.h b/source/dnode/mnode/impl/inc/mndScheduler.h index fa5e20ab6f..80fe472c56 100644 --- a/source/dnode/mnode/impl/inc/mndScheduler.h +++ b/source/dnode/mnode/impl/inc/mndScheduler.h @@ -27,10 +27,10 @@ void mndCleanupScheduler(SMnode* pMnode); int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscribeObj* pSub); -int32_t mndScheduleStream1(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream); +int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream); -int32_t mndConvertRSmaTask(const char* ast, int64_t uid, int8_t triggerType, int64_t watermark, char** pStr, - int32_t* pLen, double filesFactor); +int32_t mndConvertRsmaTask(char** pDst, int32_t* pDstLen, const char* ast, int64_t uid, int8_t triggerType, + int64_t watermark, double filesFactor); #ifdef __cplusplus } diff --git a/source/dnode/mnode/impl/src/mndScheduler.c b/source/dnode/mnode/impl/src/mndScheduler.c index deb7382183..c04e416896 100644 --- a/source/dnode/mnode/impl/src/mndScheduler.c +++ b/source/dnode/mnode/impl/src/mndScheduler.c @@ -42,8 +42,8 @@ static int32_t mndAddTaskToTaskSet(SArray* pArray, SStreamTask* 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) { +int32_t mndConvertRsmaTask(char** pDst, int32_t* pDstLen, const char* ast, int64_t uid, int8_t triggerType, + int64_t watermark, double filesFactor) { SNode* pAst = NULL; SQueryPlan* pPlan = NULL; terrno = TSDB_CODE_SUCCESS; @@ -53,7 +53,7 @@ int32_t mndConvertRSmaTask(const char* ast, int64_t uid, int8_t triggerType, int goto END; } - if (qSetSTableIdForRSma(pAst, uid) < 0) { + if (qSetSTableIdForRsma(pAst, uid) < 0) { terrno = TSDB_CODE_QRY_INVALID_INPUT; goto END; } @@ -86,7 +86,7 @@ int32_t mndConvertRSmaTask(const char* ast, int64_t uid, int8_t triggerType, int } SSubplan* plan = nodesListGetNode(inner->pNodeList, 0); - if (qSubPlanToString(plan, pStr, pLen) < 0) { + if (qSubPlanToString(plan, pDst, pDstLen) < 0) { terrno = TSDB_CODE_QRY_INVALID_INPUT; goto END; } @@ -335,7 +335,7 @@ int32_t mndAddFixedSinkTaskToStream(SMnode* pMnode, STrans* pTrans, SStreamObj* return 0; } -int32_t mndScheduleStream1(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) { +int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) { SSdb* pSdb = pMnode->pSdb; SQueryPlan* pPlan = qStringToQueryPlan(pStream->physicalPlan); if (pPlan == NULL) { @@ -361,6 +361,7 @@ int32_t mndScheduleStream1(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) mndAddFixedSinkTaskToStream(pMnode, pTrans, pStream); } } + if (totLevel > 1) { SStreamTask* pFinalTask; // inner plan @@ -472,229 +473,7 @@ int32_t mndScheduleStream1(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) } } } - return 0; -} - -int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) { - SSdb* pSdb = pMnode->pSdb; - SQueryPlan* pPlan = qStringToQueryPlan(pStream->physicalPlan); - if (pPlan == NULL) { - terrno = TSDB_CODE_QRY_INVALID_INPUT; - return -1; - } - ASSERT(pStream->vgNum == 0); - - int32_t totLevel = LIST_LENGTH(pPlan->pSubplans); - ASSERT(totLevel <= 2); - pStream->tasks = taosArrayInit(totLevel, sizeof(void*)); - - bool hasExtraSink = false; - bool externalTargetDB = strcmp(pStream->sourceDb, pStream->targetDb) != 0; - if (totLevel == 2 || externalTargetDB) { - SArray* taskOneLevel = taosArrayInit(0, sizeof(void*)); - taosArrayPush(pStream->tasks, &taskOneLevel); - // add extra sink - hasExtraSink = true; - if (pStream->fixedSinkVgId == 0) { - mndAddShuffleSinkTasksToStream(pMnode, pTrans, pStream); - } else { - mndAddFixedSinkTaskToStream(pMnode, pTrans, pStream); - } - } - - for (int32_t level = 0; level < totLevel; level++) { - SArray* taskOneLevel = taosArrayInit(0, sizeof(void*)); - taosArrayPush(pStream->tasks, &taskOneLevel); - SNodeListNode* inner = nodesListGetNode(pPlan->pSubplans, level); - ASSERT(LIST_LENGTH(inner->pNodeList) == 1); - - SSubplan* plan = nodesListGetNode(inner->pNodeList, 0); - - // if (level == totLevel - 1 /* or no snode */) { - if (level == totLevel - 1) { - // last level, source, must assign to vnode - // must be scan type - ASSERT(plan->subplanType == SUBPLAN_TYPE_SCAN); - - // replicate task to each vnode - void* pIter = NULL; - while (1) { - SVgObj* pVgroup; - pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup); - if (pIter == NULL) break; - if (pVgroup->dbUid != pStream->dbUid) { - sdbRelease(pSdb, pVgroup); - continue; - } - SStreamTask* pTask = tNewSStreamTask(pStream->uid); - mndAddTaskToTaskSet(taskOneLevel, pTask); - // source part - pTask->sourceType = TASK_SOURCE__SCAN; - pTask->inputType = TASK_INPUT_TYPE__SUMBIT_BLOCK; - - // sink part - if (level == 0) { - // only for inplace - pTask->sinkType = TASK_SINK__NONE; - if (!hasExtraSink) { -#if 1 - if (pStream->createdBy == STREAM_CREATED_BY__SMA) { - pTask->sinkType = TASK_SINK__SMA; - pTask->smaSink.smaId = pStream->smaId; - } else { - pTask->sinkType = TASK_SINK__TABLE; - pTask->tbSink.stbUid = pStream->targetStbUid; - memcpy(pTask->tbSink.stbFullName, pStream->targetSTbName, TSDB_TABLE_FNAME_LEN); - pTask->tbSink.pSchemaWrapper = tCloneSSchemaWrapper(&pStream->outputSchema); - } -#endif - } - } else { - pTask->sinkType = TASK_SINK__NONE; - } - - // dispatch part - if (level == 0 && !hasExtraSink) { - pTask->dispatchType = TASK_DISPATCH__NONE; - } else { - // add fixed ep dispatcher - int32_t lastLevel = level - 1; - if (hasExtraSink) lastLevel++; - ASSERT(lastLevel == 0); - SArray* pArray = taosArrayGetP(pStream->tasks, lastLevel); - // one merge only - ASSERT(taosArrayGetSize(pArray) == 1); - SStreamTask* lastLevelTask = taosArrayGetP(pArray, 0); - pTask->dispatchMsgType = TDMT_STREAM_TASK_DISPATCH; - pTask->dispatchType = TASK_DISPATCH__FIXED; - - pTask->fixedEpDispatcher.taskId = lastLevelTask->taskId; - pTask->fixedEpDispatcher.nodeId = lastLevelTask->nodeId; - pTask->fixedEpDispatcher.epSet = lastLevelTask->epSet; - } - - // exec part - pTask->execType = TASK_EXEC__PIPE; - if (mndAssignTaskToVg(pMnode, pTrans, pTask, plan, pVgroup) < 0) { - sdbRelease(pSdb, pVgroup); - qDestroyQueryPlan(pPlan); - return -1; - } - sdbRelease(pSdb, pVgroup); - } - } else { - // merge plan - - // TODO if has snode, assign to snode - - // else, assign to vnode - ASSERT(plan->subplanType == SUBPLAN_TYPE_MERGE); - SStreamTask* pTask = tNewSStreamTask(pStream->uid); - mndAddTaskToTaskSet(taskOneLevel, pTask); - - // source part, currently only support multi source - pTask->sourceType = TASK_SOURCE__PIPE; - pTask->inputType = TASK_INPUT_TYPE__DATA_BLOCK; - - // sink part - pTask->sinkType = TASK_SINK__NONE; - - // dispatch part - ASSERT(hasExtraSink); - /*pTask->dispatchType = TASK_DISPATCH__NONE;*/ -#if 1 - - if (hasExtraSink) { - // add dispatcher - if (pStream->fixedSinkVgId == 0) { - pTask->dispatchType = TASK_DISPATCH__SHUFFLE; - - pTask->dispatchMsgType = TDMT_STREAM_TASK_DISPATCH; - SDbObj* pDb = mndAcquireDb(pMnode, pStream->targetDb); - ASSERT(pDb); - if (mndExtractDbInfo(pMnode, pDb, &pTask->shuffleDispatcher.dbInfo, NULL) < 0) { - sdbRelease(pSdb, pDb); - qDestroyQueryPlan(pPlan); - return -1; - } - sdbRelease(pSdb, pDb); - - // put taskId to useDbRsp - // TODO: optimize - SArray* pVgs = pTask->shuffleDispatcher.dbInfo.pVgroupInfos; - int32_t sz = taosArrayGetSize(pVgs); - SArray* sinkLv = taosArrayGetP(pStream->tasks, 0); - int32_t sinkLvSize = taosArrayGetSize(sinkLv); - for (int32_t i = 0; i < sz; i++) { - SVgroupInfo* pVgInfo = taosArrayGet(pVgs, i); - for (int32_t j = 0; j < sinkLvSize; j++) { - SStreamTask* pLastLevelTask = taosArrayGetP(sinkLv, j); - if (pLastLevelTask->nodeId == pVgInfo->vgId) { - pVgInfo->taskId = pLastLevelTask->taskId; - break; - } - } - } - } else { - pTask->dispatchType = TASK_DISPATCH__FIXED; - /*pTask->dispatchMsgType = TDMT_VND_TASK_WRITE_EXEC;*/ - pTask->dispatchMsgType = TDMT_STREAM_TASK_DISPATCH; - SArray* pArray = taosArrayGetP(pStream->tasks, 0); - // one sink only - ASSERT(taosArrayGetSize(pArray) == 1); - SStreamTask* lastLevelTask = taosArrayGetP(pArray, 0); - pTask->fixedEpDispatcher.taskId = lastLevelTask->taskId; - pTask->fixedEpDispatcher.nodeId = lastLevelTask->nodeId; - pTask->fixedEpDispatcher.epSet = lastLevelTask->epSet; - } - } -#endif - - // exec part - pTask->execType = TASK_EXEC__MERGE; - SVgObj* pVgroup = mndSchedFetchOneVg(pMnode, pStream->dbUid); - ASSERT(pVgroup); - if (mndAssignTaskToVg(pMnode, pTrans, pTask, plan, pVgroup) < 0) { - sdbRelease(pSdb, pVgroup); - qDestroyQueryPlan(pPlan); - return -1; - } - sdbRelease(pSdb, pVgroup); - } - } - -#if 0 - if (totLevel == 2) { - void* pIter = NULL; - while (1) { - SVgObj* pVgroup; - pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup); - if (pIter == NULL) break; - if (pVgroup->dbUid != pStream->dbUid) { - sdbRelease(pSdb, pVgroup); - continue; - } - SStreamTask* pTask = tNewSStreamTask(pStream->uid); - - // source part - pTask->sourceType = TASK_SOURCE__MERGE; - pTask->inputType = TASK_INPUT_TYPE__DATA_BLOCK; - - // sink part - pTask->sinkType = TASK_SINK__NONE; - - // dispatch part - pTask->dispatchType = TASK_DISPATCH__NONE; - - // exec part - pTask->execType = TASK_EXEC__NONE; - } - } -#endif - - // free memory qDestroyQueryPlan(pPlan); - return 0; } diff --git a/source/dnode/mnode/impl/src/mndStb.c b/source/dnode/mnode/impl/src/mndStb.c index 9981dc8530..77f36482f6 100644 --- a/source/dnode/mnode/impl/src/mndStb.c +++ b/source/dnode/mnode/impl/src/mndStb.c @@ -395,13 +395,13 @@ static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pSt req.pRSmaParam.xFilesFactor = pStb->xFilesFactor; req.pRSmaParam.delay = pStb->delay; if (pStb->ast1Len > 0) { - if (mndConvertRSmaTask(pStb->pAst1, pStb->uid, 0, 0, &req.pRSmaParam.qmsg1, &req.pRSmaParam.qmsg1Len, + if (mndConvertRsmaTask(&req.pRSmaParam.qmsg1, &req.pRSmaParam.qmsg1Len, pStb->pAst1, pStb->uid, 0, 0, req.pRSmaParam.xFilesFactor) != TSDB_CODE_SUCCESS) { return NULL; } } if (pStb->ast2Len > 0) { - if (mndConvertRSmaTask(pStb->pAst2, pStb->uid, 0, 0, &req.pRSmaParam.qmsg2, &req.pRSmaParam.qmsg2Len, + if (mndConvertRsmaTask(&req.pRSmaParam.qmsg2, &req.pRSmaParam.qmsg2Len, pStb->pAst2, pStb->uid, 0, 0, req.pRSmaParam.xFilesFactor) != TSDB_CODE_SUCCESS) { return NULL; } diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c index 4a2c0a59a1..7abe9e3c0d 100644 --- a/source/dnode/mnode/impl/src/mndStream.c +++ b/source/dnode/mnode/impl/src/mndStream.c @@ -269,7 +269,7 @@ int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast return -1; } - if (mndScheduleStream1(pMnode, pTrans, pStream) < 0) { + if (mndScheduleStream(pMnode, pTrans, pStream) < 0) { mError("stream:%ld, schedule stream since %s", pStream->uid, terrstr()); return -1; } diff --git a/source/libs/parser/src/parser.c b/source/libs/parser/src/parser.c index c2e1eba472..edb0d79f02 100644 --- a/source/libs/parser/src/parser.c +++ b/source/libs/parser/src/parser.c @@ -76,7 +76,7 @@ static int32_t setValueByBindParam(SValueNode* pVal, TAOS_MULTI_BIND* pParam) { int32_t inputSize = (NULL != pParam->length ? *(pParam->length) : tDataTypes[pParam->buffer_type].bytes); pVal->node.resType.type = pParam->buffer_type; pVal->node.resType.bytes = inputSize; - + switch (pParam->buffer_type) { case TSDB_DATA_TYPE_VARCHAR: case TSDB_DATA_TYPE_VARBINARY: @@ -186,7 +186,7 @@ int32_t qExtractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** p return extractResultSchema(pRoot, numOfCols, pSchema); } -int32_t qSetSTableIdForRSma(SNode* pStmt, int64_t uid) { +int32_t qSetSTableIdForRsma(SNode* pStmt, int64_t uid) { if (QUERY_NODE_SELECT_STMT == nodeType(pStmt)) { SNode* pTable = ((SSelectStmt*)pStmt)->pFromTable; if (QUERY_NODE_REAL_TABLE == nodeType(pTable)) { From d53b9529eb85415040d4b7b47eaa4dc034b3f7f0 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Sat, 11 Jun 2022 11:52:07 +0800 Subject: [PATCH 2/2] test: temporarily remove failed test --- tests/system-test/fulltest.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/system-test/fulltest.sh b/tests/system-test/fulltest.sh index abc8a81248..4d8a2a66eb 100755 --- a/tests/system-test/fulltest.sh +++ b/tests/system-test/fulltest.sh @@ -98,7 +98,7 @@ python3 ./test.py -f 2-query/statecount.py python3 ./test.py -f 7-tmq/basic5.py python3 ./test.py -f 7-tmq/subscribeDb.py python3 ./test.py -f 7-tmq/subscribeDb0.py -python3 ./test.py -f 7-tmq/subscribeDb1.py +#python3 ./test.py -f 7-tmq/subscribeDb1.py python3 ./test.py -f 7-tmq/subscribeStb.py python3 ./test.py -f 7-tmq/subscribeStb0.py python3 ./test.py -f 7-tmq/subscribeStb1.py