walterddr commented on code in PR #10748:
URL: https://github.com/apache/pinot/pull/10748#discussion_r1190194915
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java:
##########
@@ -175,7 +175,7 @@ private BrokerResponse handleRequest(long requestId, String
query, @Nullable Sql
return new
BrokerResponseNative(QueryException.getException(QueryException.SQL_PARSING_ERROR,
e));
}
- QueryPlan queryPlan = queryPlanResult.getQueryPlan();
+ DispatchableSubPlan dispatchableQueryPlan = queryPlanResult.getQueryPlan();
Review Comment:
```suggestion
DispatchableSubPlan dispatchableSubPlan = queryPlanResult.getQueryPlan();
```
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/DispatchablePlanFragment.java:
##########
@@ -0,0 +1,121 @@
+/**
+ * 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.PlanFragmentMetadata;
+import org.apache.pinot.query.routing.QueryServerInstance;
+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 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 Map<Integer, Map<String, List<String>>> _workerIdToSegmentsMap;
+ private Map<String, String> _customProperties;
Review Comment:
also these are all final?
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/DispatchablePlanFragment.java:
##########
@@ -0,0 +1,121 @@
+/**
+ * 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.PlanFragmentMetadata;
+import org.apache.pinot.query.routing.QueryServerInstance;
+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 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 Map<Integer, Map<String, List<String>>> _workerIdToSegmentsMap;
+ private Map<String, String> _customProperties;
Review Comment:
can we add a comment indicating that this is the generic property format
that will be used to pass info to stageMetadat on server?
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/SubPlanMetadata.java:
##########
@@ -0,0 +1,59 @@
+/**
+ * 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.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.calcite.util.Pair;
+
+
+public class SubPlanMetadata {
+
+ /**
+ * The set of tables that are scanned in this subplan.
+ */
+ private final Set<String> _tableNames;
+
+ /**
+ * The list of fields that are surfaced by this subplan. Only valid for
SubPlan Id 0.
+ */
+ private List<Pair<Integer, String>> _fields;
+
+ private final Map<String, String> _customProperties;
+
+ public SubPlanMetadata(Set<String> tableNames, List<Pair<Integer, String>>
fields) {
+ _tableNames = tableNames;
+ _fields = fields;
+ _customProperties = new HashMap<>();
+ }
+
+ public List<Pair<Integer, String>> getFields() {
+ return _fields;
+ }
Review Comment:
this and all other metadata format with customProperties map<string, string>
backing them up.
let's make all the getters
```
public T getWithCacheVarName() {
if (_cachedVarName == null) {
_cachedVarName = extractFromCustomProperties(_customProperties);
}
return _cachedVarName;
}
```
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/QueryEnvironment.java:
##########
@@ -155,9 +157,12 @@ public QueryPlannerResult planQuery(String sqlQuery,
SqlNodeAndOptions sqlNodeAn
try (PlannerContext plannerContext = new PlannerContext(_config,
_catalogReader, _typeFactory, _hepProgram)) {
plannerContext.setOptions(sqlNodeAndOptions.getOptions());
RelRoot relRoot = compileQuery(sqlNodeAndOptions.getSqlNode(),
plannerContext);
- Set<String> tableNames = getTableNamesFromRelRoot(relRoot.rel);
- return new QueryPlannerResult(toDispatchablePlan(relRoot,
plannerContext, requestId, tableNames), null,
- tableNames);
+ SubPlan subPlan = toSubPlan(relRoot);
+ // TODO: current code only assume one SubPlan per query, but we should
support multiple SubPlans per query.
+ // Each SubPlan should be able to run independently from Broker then set
the results into the dependent
+ // SubPlan for further processing.
+ DispatchableSubPlan dispatchableSubPlan = toDispatchableSubPlan(subPlan,
plannerContext, requestId);
+ return new QueryPlannerResult(dispatchableSubPlan, null,
dispatchableSubPlan.getTableNames());
Review Comment:
i would all the first argument `rootSubPlan` or `entrypointSubPlan` and
change the QueryPlannerResult member variable accordingly
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/DispatchablePlanFragment.java:
##########
@@ -0,0 +1,121 @@
+/**
+ * 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.PlanFragmentMetadata;
+import org.apache.pinot.query.routing.QueryServerInstance;
+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 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 Map<Integer, Map<String, List<String>>> _workerIdToSegmentsMap;
+ private 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;
+ _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 = workerIdToSegmentsMap;
+ }
+
+ public List<String> getScannedTables() {
+ return _planFragment.getFragmentMetadata().getScannedTables();
+ }
+
+ public void setWorkerMetadataList(List<WorkerMetadata> workerMetadataList) {
+ _workerMetadataList.addAll(workerMetadataList);
+ }
+
+ public void setScannedTables(List<String> scannedTables) {
+ _planFragment.getFragmentMetadata().setScannedTables(scannedTables);
+ }
+
+ public PlanFragmentMetadata toPlanFragmentMetadata() {
+ return new PlanFragmentMetadata(_workerMetadataList, _customProperties);
+ }
Review Comment:
this constructor is confusing to me i don't see a PlanFragmentMetadata
constructor that accepts the 2 arguments?
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/DispatchableSubPlan.java:
##########
@@ -0,0 +1,75 @@
+/**
+ * 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.Map;
+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 Map<Integer, DispatchablePlanFragment> _queryStageMap;
+ private final Set<String> _tableNames;
+
+ public DispatchableSubPlan(List<Pair<Integer, String>> fields, Map<Integer,
DispatchablePlanFragment> queryStageMap,
+ Set<String> tableNames) {
+ _queryResultFields = fields;
+ _queryStageMap = queryStageMap;
+ _tableNames = tableNames;
+ }
+
+ /**
+ * Get the map between stageID and the stage plan root node.
+ * @return stage plan map.
+ */
+ public Map<Integer, DispatchablePlanFragment> getQueryStageMap() {
+ return _queryStageMap;
+ }
Review Comment:
this should be a list as stageId are always 0-indexed
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/PlanFragmentMetadata.java:
##########
@@ -0,0 +1,46 @@
+/**
+ * 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;
+
+
+public class PlanFragmentMetadata {
+ private final Map<String, String> _customProperties = new HashMap<>();
+
+ private List<String> _scannedTables = new ArrayList<>();
Review Comment:
IMO either we keep this unpacked or keep customProperties. we shouldn't kept
both format in member variable. causes confusion (e.g. which is the source of
truth)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
