From 2a1ec3c304d889f1629b4bf4e7f2fee8b0a53380 Mon Sep 17 00:00:00 2001 From: Xiaoyu Wang Date: Mon, 6 Jun 2022 18:39:23 +0800 Subject: [PATCH] feat: order by distributed split --- include/libs/nodes/plannodes.h | 3 ++- source/libs/nodes/src/nodesCloneFuncs.c | 2 +- source/libs/nodes/src/nodesCodeFuncs.c | 13 ++++++++++--- source/libs/planner/src/planLogicCreater.c | 4 ++-- source/libs/planner/src/planPhysiCreater.c | 18 +++++++++++++----- 5 files changed, 28 insertions(+), 12 deletions(-) diff --git a/include/libs/nodes/plannodes.h b/include/libs/nodes/plannodes.h index d646effa58..5811b76588 100644 --- a/include/libs/nodes/plannodes.h +++ b/include/libs/nodes/plannodes.h @@ -95,7 +95,7 @@ typedef struct SVnodeModifyLogicNode { int32_t msgType; SArray* pDataBlocks; SVgDataBlocks* pVgDataBlocks; - SNode* pModifyRows; // SColumnNode + SNode* pAffectedRows; // SColumnNode uint64_t tableId; int8_t tableType; // table type char tableFName[TSDB_TABLE_FNAME_LEN]; @@ -392,6 +392,7 @@ typedef struct SDataDeleterNode { int8_t tableType; // table type char tableFName[TSDB_TABLE_FNAME_LEN]; STimeWindow deleteTimeRange; + SNode* pAffectedRows; } SDataDeleterNode; typedef struct SSubplan { diff --git a/source/libs/nodes/src/nodesCloneFuncs.c b/source/libs/nodes/src/nodesCloneFuncs.c index 657ffd32a5..7d58b999dd 100644 --- a/source/libs/nodes/src/nodesCloneFuncs.c +++ b/source/libs/nodes/src/nodesCloneFuncs.c @@ -371,7 +371,7 @@ static SNode* logicVnodeModifCopy(const SVnodeModifyLogicNode* pSrc, SVnodeModif COPY_BASE_OBJECT_FIELD(node, logicNodeCopy); COPY_SCALAR_FIELD(modifyType); COPY_SCALAR_FIELD(msgType); - CLONE_NODE_FIELD(pModifyRows); + CLONE_NODE_FIELD(pAffectedRows); COPY_SCALAR_FIELD(tableId); COPY_SCALAR_FIELD(tableType); COPY_CHAR_ARRAY_FIELD(tableFName); diff --git a/source/libs/nodes/src/nodesCodeFuncs.c b/source/libs/nodes/src/nodesCodeFuncs.c index 246ee2bfed..8583f72c26 100644 --- a/source/libs/nodes/src/nodesCodeFuncs.c +++ b/source/libs/nodes/src/nodesCodeFuncs.c @@ -613,7 +613,7 @@ static int32_t jsonToLogicProjectNode(const SJson* pJson, void* pObj) { static const char* jkVnodeModifyLogicPlanModifyType = "ModifyType"; static const char* jkVnodeModifyLogicPlanMsgType = "MsgType"; -static const char* jkVnodeModifyLogicPlanModifyRows = "ModifyRows"; +static const char* jkVnodeModifyLogicPlanAffectedRows = "AffectedRows"; static int32_t logicVnodeModifyNodeToJson(const void* pObj, SJson* pJson) { const SVnodeModifyLogicNode* pNode = (const SVnodeModifyLogicNode*)pObj; @@ -626,7 +626,7 @@ static int32_t logicVnodeModifyNodeToJson(const void* pObj, SJson* pJson) { code = tjsonAddIntegerToObject(pJson, jkVnodeModifyLogicPlanMsgType, pNode->msgType); } if (TSDB_CODE_SUCCESS == code) { - code = tjsonAddObject(pJson, jkVnodeModifyLogicPlanModifyRows, nodeToJson, pNode->pModifyRows); + code = tjsonAddObject(pJson, jkVnodeModifyLogicPlanAffectedRows, nodeToJson, pNode->pAffectedRows); } return code; @@ -643,7 +643,7 @@ static int32_t jsonToLogicVnodeModifyNode(const SJson* pJson, void* pObj) { code = tjsonGetIntValue(pJson, jkVnodeModifyLogicPlanMsgType, &pNode->msgType); } if (TSDB_CODE_SUCCESS == code) { - code = jsonToNodeObject(pJson, jkVnodeModifyLogicPlanModifyRows, &pNode->pModifyRows); + code = jsonToNodeObject(pJson, jkVnodeModifyLogicPlanAffectedRows, &pNode->pAffectedRows); } return code; @@ -2016,6 +2016,7 @@ static const char* jkDeletePhysiPlanTableType = "TableType"; static const char* jkDeletePhysiPlanTableFName = "TableFName"; static const char* jkDeletePhysiPlanDeleteTimeRangeStartKey = "DeleteTimeRangeStartKey"; static const char* jkDeletePhysiPlanDeleteTimeRangeEndKey = "DeleteTimeRangeEndKey"; +static const char* jkDeletePhysiPlanAffectedRows = "AffectedRows"; static int32_t physiDeleteNodeToJson(const void* pObj, SJson* pJson) { const SDataDeleterNode* pNode = (const SDataDeleterNode*)pObj; @@ -2036,6 +2037,9 @@ static int32_t physiDeleteNodeToJson(const void* pObj, SJson* pJson) { if (TSDB_CODE_SUCCESS == code) { code = tjsonAddIntegerToObject(pJson, jkDeletePhysiPlanDeleteTimeRangeEndKey, pNode->deleteTimeRange.ekey); } + if (TSDB_CODE_SUCCESS == code) { + code = tjsonAddObject(pJson, jkDeletePhysiPlanAffectedRows, nodeToJson, pNode->pAffectedRows); + } return code; } @@ -2059,6 +2063,9 @@ static int32_t jsonToPhysiDeleteNode(const SJson* pJson, void* pObj) { if (TSDB_CODE_SUCCESS == code) { code = tjsonGetBigIntValue(pJson, jkDeletePhysiPlanDeleteTimeRangeEndKey, &pNode->deleteTimeRange.ekey); } + if (TSDB_CODE_SUCCESS == code) { + code = jsonToNodeObject(pJson, jkDeletePhysiPlanAffectedRows, &pNode->pAffectedRows); + } return code; } diff --git a/source/libs/planner/src/planLogicCreater.c b/source/libs/planner/src/planLogicCreater.c index d37e22b316..ac46531fad 100644 --- a/source/libs/planner/src/planLogicCreater.c +++ b/source/libs/planner/src/planLogicCreater.c @@ -1057,8 +1057,8 @@ static int32_t createVnodeModifLogicNodeByDelete(SLogicPlanContext* pCxt, SDelet snprintf(pModify->tableFName, sizeof(pModify->tableFName), "%d.%s.%s", pCxt->pPlanCxt->acctId, pRealTable->table.dbName, pRealTable->table.tableName); pModify->deleteTimeRange = pDelete->timeRange; - pModify->pModifyRows = nodesCloneNode(pDelete->pCountFunc); - if (NULL == pModify->pModifyRows) { + pModify->pAffectedRows = nodesCloneNode(pDelete->pCountFunc); + if (NULL == pModify->pAffectedRows) { nodesDestroyNode(pModify); return TSDB_CODE_OUT_OF_MEMORY; } diff --git a/source/libs/planner/src/planPhysiCreater.c b/source/libs/planner/src/planPhysiCreater.c index 991d6d66b8..bf5ff57949 100644 --- a/source/libs/planner/src/planPhysiCreater.c +++ b/source/libs/planner/src/planPhysiCreater.c @@ -1332,13 +1332,21 @@ static int32_t createDataDeleter(SPhysiPlanContext* pCxt, SVnodeModifyLogicNode* strcpy(pDeleter->tableFName, pModify->tableFName); pDeleter->deleteTimeRange = pModify->deleteTimeRange; - pDeleter->sink.pInputDataBlockDesc = nodesCloneNode(pRoot->pOutputDataBlockDesc); - if (NULL == pDeleter->sink.pInputDataBlockDesc) { - nodesDestroyNode(pDeleter); - return TSDB_CODE_OUT_OF_MEMORY; + int32_t code = setNodeSlotId(pCxt, pRoot->pOutputDataBlockDesc->dataBlockId, -1, pModify->pAffectedRows, + &pDeleter->pAffectedRows); + if (TSDB_CODE_SUCCESS == code) { + pDeleter->sink.pInputDataBlockDesc = nodesCloneNode(pRoot->pOutputDataBlockDesc); + if (NULL == pDeleter->sink.pInputDataBlockDesc) { + code = TSDB_CODE_OUT_OF_MEMORY; + } + } + + if (TSDB_CODE_SUCCESS == code) { + *pSink = (SDataSinkNode*)pDeleter; + } else { + nodesDestroyNode(pDeleter); } - *pSink = (SDataSinkNode*)pDeleter; return TSDB_CODE_SUCCESS; }