forked from apache/pinot
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[multistage] Refactor query planner and dispatcher (apache#10748)
- Loading branch information
Showing
46 changed files
with
1,526 additions
and
711 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
119 changes: 119 additions & 0 deletions
119
...-query-planner/src/main/java/org/apache/pinot/query/planner/DispatchablePlanFragment.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,119 @@ | ||
| /** | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
| package org.apache.pinot.query.planner; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.HashMap; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import org.apache.pinot.core.routing.TimeBoundaryInfo; | ||
| import org.apache.pinot.query.routing.QueryServerInstance; | ||
| import org.apache.pinot.query.routing.StageMetadata; | ||
| import org.apache.pinot.query.routing.WorkerMetadata; | ||
|
|
||
|
|
||
| public class DispatchablePlanFragment { | ||
|
|
||
| public static final String TABLE_NAME_KEY = "tableName"; | ||
| public static final String TIME_BOUNDARY_COLUMN_KEY = "timeBoundaryInfo.timeColumn"; | ||
| public static final String TIME_BOUNDARY_VALUE_KEY = "timeBoundaryInfo.timeValue"; | ||
| private final PlanFragment _planFragment; | ||
| private final List<WorkerMetadata> _workerMetadataList; | ||
|
|
||
| // This is used at broker stage - we don't need to ship it to the server. | ||
| private final Map<QueryServerInstance, List<Integer>> _serverInstanceToWorkerIdMap; | ||
|
|
||
| // used for table scan stage - we use ServerInstance instead of VirtualServer | ||
| // here because all virtual servers that share a server instance will have the | ||
| // same segments on them | ||
| private final Map<Integer, Map<String, List<String>>> _workerIdToSegmentsMap; | ||
|
|
||
| // used for passing custom properties to build StageMetadata on the server. | ||
| private final Map<String, String> _customProperties; | ||
|
|
||
| public DispatchablePlanFragment(PlanFragment planFragment) { | ||
| this(planFragment, new ArrayList<>(), new HashMap<>(), new HashMap<>()); | ||
| } | ||
|
|
||
| public DispatchablePlanFragment(PlanFragment planFragment, List<WorkerMetadata> workerMetadataList, | ||
| Map<QueryServerInstance, List<Integer>> serverInstanceToWorkerIdMap, Map<String, String> customPropertyMap) { | ||
| _planFragment = planFragment; | ||
| _workerMetadataList = workerMetadataList; | ||
| _serverInstanceToWorkerIdMap = serverInstanceToWorkerIdMap; | ||
| _workerIdToSegmentsMap = new HashMap<>(); | ||
| _customProperties = customPropertyMap; | ||
| } | ||
|
|
||
| public PlanFragment getPlanFragment() { | ||
| return _planFragment; | ||
| } | ||
|
|
||
| public List<WorkerMetadata> getWorkerMetadataList() { | ||
| return _workerMetadataList; | ||
| } | ||
|
|
||
| public Map<QueryServerInstance, List<Integer>> getServerInstanceToWorkerIdMap() { | ||
| return _serverInstanceToWorkerIdMap; | ||
| } | ||
|
|
||
| public Map<String, String> getCustomProperties() { | ||
| return _customProperties; | ||
| } | ||
|
|
||
| public String getTableName() { | ||
| return _customProperties.get(TABLE_NAME_KEY); | ||
| } | ||
|
|
||
| public String setTableName(String tableName) { | ||
| return _customProperties.put(TABLE_NAME_KEY, tableName); | ||
| } | ||
|
|
||
| public TimeBoundaryInfo getTimeBoundary() { | ||
| return new TimeBoundaryInfo(_customProperties.get(TIME_BOUNDARY_COLUMN_KEY), | ||
| _customProperties.get(TIME_BOUNDARY_VALUE_KEY)); | ||
| } | ||
|
|
||
| public void setTimeBoundaryInfo(TimeBoundaryInfo timeBoundaryInfo) { | ||
| _customProperties.put(TIME_BOUNDARY_COLUMN_KEY, timeBoundaryInfo.getTimeColumn()); | ||
| _customProperties.put(TIME_BOUNDARY_VALUE_KEY, timeBoundaryInfo.getTimeValue()); | ||
| } | ||
|
|
||
| public Map<Integer, Map<String, List<String>>> getWorkerIdToSegmentsMap() { | ||
| return _workerIdToSegmentsMap; | ||
| } | ||
|
|
||
| public void setWorkerIdToSegmentsMap(Map<Integer, Map<String, List<String>>> workerIdToSegmentsMap) { | ||
| _workerIdToSegmentsMap.clear(); | ||
| _workerIdToSegmentsMap.putAll(workerIdToSegmentsMap); | ||
| } | ||
|
|
||
| public void setWorkerMetadataList(List<WorkerMetadata> workerMetadataList) { | ||
| _workerMetadataList.clear(); | ||
| _workerMetadataList.addAll(workerMetadataList); | ||
| } | ||
|
|
||
| public StageMetadata toStageMetadata() { | ||
| return new StageMetadata(_workerMetadataList, _customProperties); | ||
| } | ||
|
|
||
| public void setServerInstanceToWorkerIdMap(Map<QueryServerInstance, List<Integer>> serverInstanceToWorkerIdMap) { | ||
| _serverInstanceToWorkerIdMap.clear(); | ||
| _serverInstanceToWorkerIdMap.putAll(serverInstanceToWorkerIdMap); | ||
| } | ||
| } |
74 changes: 74 additions & 0 deletions
74
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/DispatchableSubPlan.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,74 @@ | ||
| /** | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
| package org.apache.pinot.query.planner; | ||
|
|
||
| import java.util.List; | ||
| import java.util.Set; | ||
| import org.apache.calcite.util.Pair; | ||
|
|
||
|
|
||
| /** | ||
| * The {@code DispatchableSubPlan} is the dispatchable query execution plan from the result of | ||
| * {@link org.apache.pinot.query.planner.logical.LogicalPlanner} and | ||
| * {@link org.apache.pinot.query.planner.physical.PinotDispatchPlanner}. | ||
| * | ||
| * <p>QueryPlan should contain the necessary stage boundary information and the cross exchange information | ||
| * for: | ||
| * <ul> | ||
| * <li>dispatch individual stages to executor.</li> | ||
| * <li>instruction for stage executor to establish connection channels to other stages.</li> | ||
| * <li>instruction for encoding data blocks & transferring between stages based on partitioning scheme.</li> | ||
| * </ul> | ||
| */ | ||
| public class DispatchableSubPlan { | ||
| private final List<Pair<Integer, String>> _queryResultFields; | ||
| private final List<DispatchablePlanFragment> _queryStageList; | ||
| private final Set<String> _tableNames; | ||
|
|
||
| public DispatchableSubPlan(List<Pair<Integer, String>> fields, List<DispatchablePlanFragment> queryStageList, | ||
| Set<String> tableNames) { | ||
| _queryResultFields = fields; | ||
| _queryStageList = queryStageList; | ||
| _tableNames = tableNames; | ||
| } | ||
|
|
||
| /** | ||
| * Get the list of stage plan root node. | ||
| * @return stage plan map. | ||
| */ | ||
| public List<DispatchablePlanFragment> getQueryStageList() { | ||
| return _queryStageList; | ||
| } | ||
|
|
||
| /** | ||
| * Get the query result field. | ||
| * @return query result field. | ||
| */ | ||
| public List<Pair<Integer, String>> getQueryResultFields() { | ||
| return _queryResultFields; | ||
| } | ||
|
|
||
| /** | ||
| * Get the table names. | ||
| * @return table names. | ||
| */ | ||
| public Set<String> getTableNames() { | ||
| return _tableNames; | ||
| } | ||
| } |
Oops, something went wrong.