diff --git a/include/libs/planner/planner.h b/include/libs/planner/planner.h index f86ce77803..3f4d21a746 100644 --- a/include/libs/planner/planner.h +++ b/include/libs/planner/planner.h @@ -108,7 +108,7 @@ typedef struct SProjectPhyNode { typedef struct SExchangePhyNode { SPhyNode node; uint64_t srcTemplateId; // template id of datasource suplans - SArray *pSourceEpSet; // SEpSet, scheduler fill by calling qSetSuplanExecutionNode + SArray *pSrcEndPoints; // SEpAddrMsg, scheduler fill by calling qSetSuplanExecutionNode } SExchangePhyNode; typedef struct SSubplanId { @@ -129,6 +129,7 @@ typedef struct SSubplan { typedef struct SQueryDag { uint64_t queryId; + int32_t numOfSubplans; SArray *pSubplans; // Element is SArray*, and nested element is SSubplan. The execution level of subplan, starting from 0. } SQueryDag; @@ -137,7 +138,11 @@ typedef struct SQueryDag { */ int32_t qCreateQueryDag(const struct SQueryStmtInfo* pQueryInfo, struct SEpSet* pQnode, struct SQueryDag** pDag); -int32_t qSetSubplanExecutionNode(SSubplan* subplan, SArray* eps); +// Set datasource of this subplan, multiple calls may be made to a subplan. +// @subplan subplan to be schedule +// @templateId templateId of a group of datasource subplans of this @subplan +// @eps Execution location of this group of datasource subplans, is an array of SEpAddr structures +int32_t qSetSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SArray* eps); int32_t qExplainQuery(const struct SQueryStmtInfo* pQueryInfo, struct SEpSet* pQnode, char** str); diff --git a/source/libs/planner/inc/plannerInt.h b/source/libs/planner/inc/plannerInt.h index 68a9518680..19563a8a0c 100644 --- a/source/libs/planner/inc/plannerInt.h +++ b/source/libs/planner/inc/plannerInt.h @@ -100,8 +100,9 @@ int32_t queryPlanToString(struct SQueryPlanNode* pQueryNode, char** str); int32_t queryPlanToSql(struct SQueryPlanNode* pQueryNode, char** sql); int32_t createDag(SQueryPlanNode* pQueryNode, struct SCatalog* pCatalog, SQueryDag** pDag); - +int32_t setSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SArray* eps); int32_t subPlanToString(const SSubplan *pPhyNode, char** str); +int32_t stringToSubplan(const char* str, SSubplan** subplan); /** * Destroy the query plan object. diff --git a/source/libs/planner/src/physicalPlan.c b/source/libs/planner/src/physicalPlan.c index 5a891d4b28..f187ec0ec9 100644 --- a/source/libs/planner/src/physicalPlan.c +++ b/source/libs/planner/src/physicalPlan.c @@ -19,6 +19,13 @@ #define STORE_CURRENT_SUBPLAN(cxt) SSubplan* _ = cxt->pCurrentSubplan #define RECOVERY_CURRENT_SUBPLAN(cxt) cxt->pCurrentSubplan = _ +typedef struct SPlanContext { + struct SCatalog* pCatalog; + struct SQueryDag* pDag; + SSubplan* pCurrentSubplan; + SSubplanId nextId; +} SPlanContext; + static const char* gOpName[] = { "Unknown", #define INCLUDE_AS_NAME @@ -26,12 +33,14 @@ static const char* gOpName[] = { #undef INCLUDE_AS_NAME }; -typedef struct SPlanContext { - struct SCatalog* pCatalog; - struct SQueryDag* pDag; - SSubplan* pCurrentSubplan; - SSubplanId nextId; -} SPlanContext; +int32_t opNameToOpType(const char* name) { + for (int32_t i = 1; i < sizeof(gOpName) / sizeof(gOpName[0]); ++i) { + if (strcmp(name, gOpName[i])) { + return i; + } + } + return OP_Unknown; +} static void toDataBlockSchema(SQueryPlanNode* pPlanNode, SDataBlockSchema* dataBlockSchema) { SWAP(dataBlockSchema->pSchema, pPlanNode->pSchema, SSchema*); @@ -216,11 +225,6 @@ int32_t createDag(SQueryPlanNode* pQueryNode, struct SCatalog* pCatalog, SQueryD return TSDB_CODE_SUCCESS; } -int32_t opNameToOpType(const char* name) { - for (int32_t i = 1; i < sizeof(gOpName) / sizeof(gOpName[0]); ++i) { - if (strcmp(name, gOpName[i])) { - return i; - } - } - return OP_Unknown; +int32_t setSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SArray* eps) { + //todo } diff --git a/source/libs/planner/src/physicalPlanJson.c b/source/libs/planner/src/physicalPlanJson.c index 7c0dc25627..2510797158 100644 --- a/source/libs/planner/src/physicalPlanJson.c +++ b/source/libs/planner/src/physicalPlanJson.c @@ -463,6 +463,191 @@ static bool exprInfoFromJson(const cJSON* json, void* obj) { return res; } +static const char* jkTimeWindowStartKey = "StartKey"; +static const char* jkTimeWindowEndKey = "EndKey"; + +static bool timeWindowToJson(const void* obj, cJSON* json) { + const STimeWindow* win = (const STimeWindow*)obj; + bool res = cJSON_AddNumberToObject(json, jkTimeWindowStartKey, win->skey); + if (res) { + res = cJSON_AddNumberToObject(json, jkTimeWindowEndKey, win->ekey); + } + return res; +} + +static bool timeWindowFromJson(const cJSON* json, void* obj) { + STimeWindow* win = (STimeWindow*)obj; + win->skey = getNumber(json, jkTimeWindowStartKey); + win->ekey = getNumber(json, jkTimeWindowEndKey); + return true; +} + +static const char* jkScanNodeTableId = "TableId"; +static const char* jkScanNodeTableType = "TableType"; + +static bool scanNodeToJson(const void* obj, cJSON* json) { + const SScanPhyNode* scan = (const SScanPhyNode*)obj; + bool res = cJSON_AddNumberToObject(json, jkScanNodeTableId, scan->uid); + if (res) { + res = cJSON_AddNumberToObject(json, jkScanNodeTableType, scan->tableType); + } + return res; +} + +static bool scanNodeFromJson(const cJSON* json, void* obj) { + SScanPhyNode* scan = (SScanPhyNode*)obj; + scan->uid = getNumber(json, jkScanNodeTableId); + scan->tableType = getNumber(json, jkScanNodeTableType); + return true; +} + +static const char* jkTableScanNodeFlag = "Flag"; +static const char* jkTableScanNodeWindow = "Window"; +static const char* jkTableScanNodeTagsConditions = "TagsConditions"; + +static bool tableScanNodeToJson(const void* obj, cJSON* json) { + const STableScanPhyNode* scan = (const STableScanPhyNode*)obj; + bool res = scanNodeToJson(obj, json); + if (res) { + res = cJSON_AddNumberToObject(json, jkTableScanNodeFlag, scan->scanFlag); + } + if (res) { + res = addObject(json, jkTableScanNodeWindow, timeWindowToJson, &scan->window); + } + if (res) { + res = addArray(json, jkTableScanNodeTagsConditions, exprInfoToJson, scan->pTagsConditions); + } + return res; +} + +static bool tableScanNodeFromJson(const cJSON* json, void* obj) { + STableScanPhyNode* scan = (STableScanPhyNode*)obj; + bool res = scanNodeFromJson(json, obj); + if (res) { + scan->scanFlag = getNumber(json, jkTableScanNodeFlag); + } + if (res) { + res = fromObject(json, jkTableScanNodeWindow, timeWindowFromJson, &scan->window, true); + } + if (res) { + res = fromArray(json, jkTableScanNodeTagsConditions, exprInfoFromJson, &scan->pTagsConditions, sizeof(SExprInfo)); + } + return res; +} + +static const char* jkEpAddrFqdn = "Fqdn"; +static const char* jkEpAddrPort = "Port"; + +static bool epAddrToJson(const void* obj, cJSON* json) { + const SEpAddrMsg* ep = (const SEpAddrMsg*)obj; + bool res = cJSON_AddStringToObject(json, jkEpAddrFqdn, ep->fqdn); + if (res) { + res = cJSON_AddNumberToObject(json, jkEpAddrPort, ep->port); + } + return res; +} + +static bool epAddrFromJson(const cJSON* json, void* obj) { + SEpAddrMsg* ep = (SEpAddrMsg*)obj; + copyString(json, jkEpAddrFqdn, ep->fqdn); + ep->port = getNumber(json, jkEpAddrPort); + return true; +} + +static const char* jkExchangeNodeSrcTemplateId = "SrcTemplateId"; +static const char* jkExchangeNodeSrcEndPoints = "SrcEndPoints"; + +static bool exchangeNodeToJson(const void* obj, cJSON* json) { + const SExchangePhyNode* exchange = (const SExchangePhyNode*)obj; + bool res = cJSON_AddNumberToObject(json, jkExchangeNodeSrcTemplateId, exchange->srcTemplateId); + if (res) { + res = addArray(json, jkExchangeNodeSrcEndPoints, epAddrToJson, exchange->pSrcEndPoints); + } + return res; +} + +static bool exchangeNodeFromJson(const cJSON* json, void* obj) { + SExchangePhyNode* exchange = (SExchangePhyNode*)obj; + exchange->srcTemplateId = getNumber(json, jkExchangeNodeSrcTemplateId); + return fromArray(json, jkExchangeNodeSrcEndPoints, epAddrFromJson, &exchange->pSrcEndPoints, sizeof(SEpAddrMsg)); +} + +static bool specificPhyNodeToJson(const void* obj, cJSON* json) { + const SPhyNode* phyNode = (const SPhyNode*)obj; + switch (phyNode->info.type) { + case OP_TableScan: + case OP_DataBlocksOptScan: + case OP_TableSeqScan: + return tableScanNodeToJson(obj, json); + case OP_TagScan: + case OP_SystemTableScan: + return scanNodeToJson(obj, json); + case OP_Aggregate: + break; // todo + case OP_Project: + return true; + case OP_Groupby: + case OP_Limit: + case OP_SLimit: + case OP_TimeWindow: + case OP_SessionWindow: + case OP_StateWindow: + case OP_Fill: + case OP_MultiTableAggregate: + case OP_MultiTableTimeInterval: + case OP_Filter: + case OP_Distinct: + case OP_Join: + case OP_AllTimeWindow: + case OP_AllMultiTableTimeInterval: + case OP_Order: + break; // todo + case OP_Exchange: + return exchangeNodeToJson(obj, json); + default: + break; + } + return false; +} + +static bool specificPhyNodeFromJson(const cJSON* json, void* obj) { + SPhyNode* phyNode = (SPhyNode*)obj; + switch (phyNode->info.type) { + case OP_TableScan: + case OP_DataBlocksOptScan: + case OP_TableSeqScan: + return tableScanNodeFromJson(json, obj); + case OP_TagScan: + case OP_SystemTableScan: + return scanNodeFromJson(json, obj); + case OP_Aggregate: + break; // todo + case OP_Project: + return true; + case OP_Groupby: + case OP_Limit: + case OP_SLimit: + case OP_TimeWindow: + case OP_SessionWindow: + case OP_StateWindow: + case OP_Fill: + case OP_MultiTableAggregate: + case OP_MultiTableTimeInterval: + case OP_Filter: + case OP_Distinct: + case OP_Join: + case OP_AllTimeWindow: + case OP_AllMultiTableTimeInterval: + case OP_Order: + break; // todo + case OP_Exchange: + return exchangeNodeFromJson(json, obj); + default: + break; + } + return false; +} + static const char* jkPnodeName = "Name"; static const char* jkPnodeTargets = "Targets"; static const char* jkPnodeConditions = "Conditions"; @@ -484,6 +669,9 @@ static bool phyNodeToJson(const void* obj, cJSON* jNode) { if (res) { res = addArray(jNode, jkPnodeChildren, phyNodeToJson, phyNode->pChildren); } + if (res) { + res = addObject(jNode, phyNode->info.name, specificPhyNodeToJson, phyNode); + } return res; } @@ -501,6 +689,9 @@ static bool phyNodeFromJson(const cJSON* json, void* obj) { if (res) { res = fromArray(json, jkPnodeChildren, phyNodeFromJson, &node->pChildren, sizeof(SSlotSchema)); } + if (res) { + res = fromObject(json, node->info.name, specificPhyNodeFromJson, node, true); + } return res; } diff --git a/source/libs/planner/src/planner.c b/source/libs/planner/src/planner.c index 3d2bb895d5..3a90acb5fd 100644 --- a/source/libs/planner/src/planner.c +++ b/source/libs/planner/src/planner.c @@ -46,6 +46,10 @@ int32_t qCreateQueryDag(const struct SQueryStmtInfo* pQueryInfo, struct SEpSet* return TSDB_CODE_SUCCESS; } +int32_t qSetSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SArray* eps) { + return setSubplanExecutionNode(subplan, templateId, eps); +} + int32_t qSubPlanToString(const SSubplan *subplan, char** str) { return subPlanToString(subplan, str); }