You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "ankitsultana (via GitHub)" <gi...@apache.org> on 2025/05/08 15:34:26 UTC
[PR] [multistage] Multistage Engine Lite Mode (prototype) [pinot]
ankitsultana opened a new pull request, #15743:
URL: https://github.com/apache/pinot/pull/15743
# Summary
**Goal:** Prototype the MSE Lite Mode to enable testing in our clusters so we can make informed decisions about its semantics and overall guarantees. (e.g. what is the most intuitive way to add limit to the leaf stage that is least confusing and maximally transparent to users)
Prototypes the Multistage Engine Lite Mode as described in #14640. The feature is hidden behind a flag and can only be enabled if one has also set the `usePhysicalOptimizer=true` query option.
This builds on top of the existing Optimizer changes and leverages many of the optimizations that are provided by that optimizer.
# Specifics
## Worker Assignment
Worker and Exchange are assigned using a new rule. This is because the generic Worker/Exchange assignment rule handles the general case, and for the Lite Mode we have some custom logic like picking a random server instance out of the ones assigned to the leaf stage.
The assignment is quite simple: the leaf stage assignment is done by the existing `LeafStageWorkerAssignmentRule`, and the new Lite Mode assignment rule simply samples a server instance from the leaf stage workers, and uses it for the all plan nodes except the leaf stage.
## Integration with Sort/Aggregate Pushdown
Both of these rules are still run since there are scenarios where we would like to push down the Sort or Aggregate to the leaf. e.g. if I have a query like `SELECT col1, COUNT(*) FROM tbl GROUP BY col1`, then the aggregate should be pushed down to the leaf stage in most cases for obvious reasons.
## Sort Insert Rule
If there doesn't exist a limit already in the leaf stage, we add it. Currently I am using a hardcoded value but in the future we'll make it configurable per se.
# Semantics
None of the semantics in this PR are final and are subject to broader community review. Based on testing from this PR, I'll file a PEP to describe the semantics in detail.
# Test Plan
Added Unit Tests. We are also testing this out in our clusters.
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [multistage] Multistage Engine Lite Mode (prototype) [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #15743:
URL: https://github.com/apache/pinot/pull/15743#discussion_r2085627131
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/opt/rules/LiteModeWorkerAssignmentRule.java:
##########
@@ -0,0 +1,118 @@
+/**
+ * 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.physical.v2.opt.rules;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Random;
+import java.util.Set;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.pinot.calcite.rel.traits.PinotExecStrategyTrait;
+import org.apache.pinot.query.context.PhysicalPlannerContext;
+import org.apache.pinot.query.planner.physical.v2.ExchangeStrategy;
+import org.apache.pinot.query.planner.physical.v2.PRelNode;
+import org.apache.pinot.query.planner.physical.v2.PinotDataDistribution;
+import org.apache.pinot.query.planner.physical.v2.nodes.PhysicalExchange;
+import org.apache.pinot.query.planner.physical.v2.opt.PRelNodeTransformer;
+
+
+/**
+ * Lite mode uses a single worker for all stages except the leaf stage. Since leaf stage assignment is done before this
+ * rule is called, we simply need to sample a random worker from the leaf stage, and assign it to all the non-leaf
+ * plan nodes.
+ */
+public class LiteModeWorkerAssignmentRule implements PRelNodeTransformer {
+ private static final Random RANDOM = new Random();
+ private final PhysicalPlannerContext _context;
+
+ public LiteModeWorkerAssignmentRule(PhysicalPlannerContext context) {
+ _context = context;
+ }
+
+ @Override
+ public PRelNode execute(PRelNode currentNode) {
+ Set<String> workerSet = new HashSet<>();
+ accumulateWorkers(currentNode, workerSet);
+ List<String> workers = List.of(sampleWorker(workerSet));
+ PinotDataDistribution pdd = new PinotDataDistribution(RelDistribution.Type.SINGLETON, workers, workers.hashCode(),
+ null, null);
+ return addExchangeAndWorkers(currentNode, null, pdd);
+ }
+
+ public PRelNode addExchangeAndWorkers(PRelNode currentNode, @Nullable PRelNode parent, PinotDataDistribution pdd) {
+ if (currentNode.isLeafStage()) {
+ if (parent == null) {
+ return currentNode;
+ }
+ return new PhysicalExchange(nodeId(), currentNode, pdd, Collections.emptyList(),
+ ExchangeStrategy.SINGLETON_EXCHANGE, null, PinotExecStrategyTrait.getDefaultExecStrategy());
+ }
+ List<PRelNode> newInputs = new ArrayList<>();
+ for (PRelNode input : currentNode.getPRelInputs()) {
+ newInputs.add(addExchangeAndWorkers(input, currentNode, pdd));
+ }
+ return currentNode.with(newInputs, pdd);
+ }
+
+ /**
+ * Stores workers assigned to the leaf stage nodes into the provided Set. Note that each worker has an integer prefix
+ * which denotes the "workerId". We remove that prefix before storing them in the set.
+ */
+ @VisibleForTesting
+ static void accumulateWorkers(PRelNode currentNode, Set<String> workerSink) {
+ if (currentNode.isLeafStage()) {
+ workerSink.addAll(currentNode.getPinotDataDistributionOrThrow().getWorkers().stream()
+ .map(LiteModeWorkerAssignmentRule::stripIdPrefixFromWorker).collect(Collectors.toSet()));
Review Comment:
the workSink itself is a Set. The addAll itself would do dedup. So I think collect(Collectors.toList()) can be more efficient?
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [multistage] Multistage Engine Lite Mode (prototype) [pinot]
Posted by "ankitsultana (via GitHub)" <gi...@apache.org>.
ankitsultana commented on code in PR #15743:
URL: https://github.com/apache/pinot/pull/15743#discussion_r2079995355
##########
pinot-query-planner/src/test/resources/queries/PhysicalOptimizerPlans.json:
##########
@@ -421,5 +421,104 @@
]
}
]
+ },
+ "physical_opt_lite_mode_single_rel_queries": {
+ "queries": [
+ {
+ "description": "Simple SELECT with WHERE query.",
+ "sql": "SET usePhysicalOptimizer=true; SET useLiteMode=true; EXPLAIN PLAN FOR SELECT col2, col3 FROM a WHERE col1 = 'foo'",
+ "output": [
+ "Execution Plan",
+ "\nPhysicalSort(fetch=[100000])",
+ "\n PhysicalProject(col2=[$1], col3=[$2])",
+ "\n PhysicalFilter(condition=[=($0, _UTF-8'foo')])",
+ "\n PhysicalTableScan(table=[[default, a]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "Auto elimination of partial aggregate when group-by on partitioning column. There's no sort because the limit is added to Agg.",
Review Comment:
I'll raise a small change to update the `explainTerms` of PhysicalAggregate to match that of `PinotLogicalAggregate`. Didn't do it in this PR because it will change quite a lot of tests in this json file.
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [multistage] Multistage Engine Lite Mode (prototype) [pinot]
Posted by "ankitsultana (via GitHub)" <gi...@apache.org>.
ankitsultana commented on code in PR #15743:
URL: https://github.com/apache/pinot/pull/15743#discussion_r2080001777
##########
pinot-query-planner/src/test/resources/queries/PhysicalOptimizerPlans.json:
##########
@@ -421,5 +421,104 @@
]
}
]
+ },
+ "physical_opt_lite_mode_single_rel_queries": {
+ "queries": [
+ {
+ "description": "Simple SELECT with WHERE query.",
+ "sql": "SET usePhysicalOptimizer=true; SET useLiteMode=true; EXPLAIN PLAN FOR SELECT col2, col3 FROM a WHERE col1 = 'foo'",
+ "output": [
+ "Execution Plan",
+ "\nPhysicalSort(fetch=[100000])",
+ "\n PhysicalProject(col2=[$1], col3=[$2])",
+ "\n PhysicalFilter(condition=[=($0, _UTF-8'foo')])",
+ "\n PhysicalTableScan(table=[[default, a]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "Auto elimination of partial aggregate when group-by on partitioning column. There's no sort because the limit is added to Agg.",
+ "sql": "SET usePhysicalOptimizer=true; SET useLiteMode=true; EXPLAIN PLAN FOR SELECT col2, COUNT(*) FROM a WHERE col1 = 'foo' GROUP BY col2",
+ "output": [
+ "Execution Plan",
+ "\nPhysicalAggregate(group=[{1}], agg#0=[COUNT()])",
+ "\n PhysicalFilter(condition=[=($0, _UTF-8'foo')])",
+ "\n PhysicalTableScan(table=[[default, a]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "Sub-queries with chained transformations",
+ "sql": "SET usePhysicalOptimizer=true; SET useLiteMode=true; EXPLAIN PLAN FOR WITH tmp AS (SELECT col1, col2, col3, COUNT(*) FROM a WHERE col1 = 'foo' GROUP BY col1, col2, col3) SELECT * FROM (SELECT ROW_NUMBER() OVER (PARTITION BY col2 ORDER BY col3) as rnk, col1 FROM tmp) WHERE rnk = 1",
+ "output": [
+ "Execution Plan",
+ "\nPhysicalProject(rnk=[$3], col1=[$0])",
+ "\n PhysicalFilter(condition=[=($3, 1)])",
+ "\n PhysicalWindow(window#0=[window(partition {1} order by [2] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])",
+ "\n PhysicalAggregate(group=[{0, 1, 2}])",
+ "\n PhysicalExchange(exchangeStrategy=[SINGLETON_EXCHANGE], distKeys=[[]], execStrategy=[STREAMING], collation=[[]])",
+ "\n PhysicalAggregate(group=[{0, 1, 2}])",
+ "\n PhysicalFilter(condition=[=($0, _UTF-8'foo')])",
+ "\n PhysicalTableScan(table=[[default, a]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "Pagination on group-by results",
+ "sql": "SET usePhysicalOptimizer=true; SET useLiteMode=true; EXPLAIN PLAN FOR WITH tmp AS (SELECT col1, col2, col3, COUNT(*) FROM a WHERE col1 = 'foo' GROUP BY col1, col2, col3 ORDER BY col2) SELECT * FROM tmp LIMIT 100,400",
+ "output": [
+ "Execution Plan",
+ "\nPhysicalSort(offset=[100], fetch=[400])",
+ "\n PhysicalAggregate(group=[{0, 1, 2}], agg#0=[COUNT($3)])",
+ "\n PhysicalExchange(exchangeStrategy=[SINGLETON_EXCHANGE], distKeys=[[]], execStrategy=[STREAMING], collation=[[]])",
+ "\n PhysicalAggregate(group=[{0, 1, 2}], agg#0=[COUNT()])",
+ "\n PhysicalFilter(condition=[=($0, _UTF-8'foo')])",
+ "\n PhysicalTableScan(table=[[default, a]])",
+ "\n"
+ ]
+ }
+ ]
+ },
+ "physical_opt_lite_mode_bi_rel_queries": {
+ "queries": [
+ {
+ "description": "Query with single semi join",
+ "sql": "SET usePhysicalOptimizer=true; SET useLiteMode=true; EXPLAIN PLAN FOR SELECT col2, col3 FROM a WHERE col1 = 'foo' AND col2 IN (SELECT col1 FROM b)",
+ "output": [
+ "Execution Plan",
+ "\nPhysicalJoin(condition=[=($0, $2)], joinType=[semi])",
+ "\n PhysicalExchange(exchangeStrategy=[SINGLETON_EXCHANGE], distKeys=[[]], execStrategy=[STREAMING], collation=[[]])",
+ "\n PhysicalSort(fetch=[100000])",
+ "\n PhysicalProject(col2=[$1], col3=[$2])",
+ "\n PhysicalFilter(condition=[=($0, _UTF-8'foo')])",
+ "\n PhysicalTableScan(table=[[default, a]])",
+ "\n PhysicalExchange(exchangeStrategy=[SINGLETON_EXCHANGE], distKeys=[[]], execStrategy=[STREAMING], collation=[[]])",
+ "\n PhysicalSort(fetch=[100000])",
+ "\n PhysicalProject(col1=[$0])",
+ "\n PhysicalTableScan(table=[[default, b]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "Query with single semi join and aggregation",
+ "sql": "SET usePhysicalOptimizer=true; SET useLiteMode=true; EXPLAIN PLAN FOR SELECT COUNT(*), col2 FROM a WHERE col1 = 'foo' AND col2 IN (SELECT col1 FROM b) GROUP BY col2",
Review Comment:
This query demonstrates one challenge. Most of the users would expect that the Semi Join filter will be evaluated first and then the aggregation will take place. But the aggregation can only be done after the filter.
This can be fixed with semi-join dynamic filtering, but the same issue also holds true for anti semi-joins too.
One way to address this is to use the automated Sub-Plan fragmenter or dynamically switching to pipeline breaker like how the MSE does it.
I think before we think about optimizations, we'll iron out the semantics so they remain as intuitive as possible for the users.
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [multistage] Multistage Engine Lite Mode (prototype) [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #15743:
URL: https://github.com/apache/pinot/pull/15743#issuecomment-2863640133
:x: We are unable to process any of the uploaded JUnit XML files. Please ensure your files are in the right format.
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [multistage] Multistage Engine Lite Mode (prototype) [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #15743:
URL: https://github.com/apache/pinot/pull/15743#discussion_r2085627131
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/opt/rules/LiteModeWorkerAssignmentRule.java:
##########
@@ -0,0 +1,118 @@
+/**
+ * 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.physical.v2.opt.rules;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Random;
+import java.util.Set;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.pinot.calcite.rel.traits.PinotExecStrategyTrait;
+import org.apache.pinot.query.context.PhysicalPlannerContext;
+import org.apache.pinot.query.planner.physical.v2.ExchangeStrategy;
+import org.apache.pinot.query.planner.physical.v2.PRelNode;
+import org.apache.pinot.query.planner.physical.v2.PinotDataDistribution;
+import org.apache.pinot.query.planner.physical.v2.nodes.PhysicalExchange;
+import org.apache.pinot.query.planner.physical.v2.opt.PRelNodeTransformer;
+
+
+/**
+ * Lite mode uses a single worker for all stages except the leaf stage. Since leaf stage assignment is done before this
+ * rule is called, we simply need to sample a random worker from the leaf stage, and assign it to all the non-leaf
+ * plan nodes.
+ */
+public class LiteModeWorkerAssignmentRule implements PRelNodeTransformer {
+ private static final Random RANDOM = new Random();
+ private final PhysicalPlannerContext _context;
+
+ public LiteModeWorkerAssignmentRule(PhysicalPlannerContext context) {
+ _context = context;
+ }
+
+ @Override
+ public PRelNode execute(PRelNode currentNode) {
+ Set<String> workerSet = new HashSet<>();
+ accumulateWorkers(currentNode, workerSet);
+ List<String> workers = List.of(sampleWorker(workerSet));
+ PinotDataDistribution pdd = new PinotDataDistribution(RelDistribution.Type.SINGLETON, workers, workers.hashCode(),
+ null, null);
+ return addExchangeAndWorkers(currentNode, null, pdd);
+ }
+
+ public PRelNode addExchangeAndWorkers(PRelNode currentNode, @Nullable PRelNode parent, PinotDataDistribution pdd) {
+ if (currentNode.isLeafStage()) {
+ if (parent == null) {
+ return currentNode;
+ }
+ return new PhysicalExchange(nodeId(), currentNode, pdd, Collections.emptyList(),
+ ExchangeStrategy.SINGLETON_EXCHANGE, null, PinotExecStrategyTrait.getDefaultExecStrategy());
+ }
+ List<PRelNode> newInputs = new ArrayList<>();
+ for (PRelNode input : currentNode.getPRelInputs()) {
+ newInputs.add(addExchangeAndWorkers(input, currentNode, pdd));
+ }
+ return currentNode.with(newInputs, pdd);
+ }
+
+ /**
+ * Stores workers assigned to the leaf stage nodes into the provided Set. Note that each worker has an integer prefix
+ * which denotes the "workerId". We remove that prefix before storing them in the set.
+ */
+ @VisibleForTesting
+ static void accumulateWorkers(PRelNode currentNode, Set<String> workerSink) {
+ if (currentNode.isLeafStage()) {
+ workerSink.addAll(currentNode.getPinotDataDistributionOrThrow().getWorkers().stream()
+ .map(LiteModeWorkerAssignmentRule::stripIdPrefixFromWorker).collect(Collectors.toSet()));
Review Comment:
the workSink itself is a Set. The addAll itself would do dedup. So I think collect to a List can be more efficient?
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [multistage] Multistage Engine Lite Mode (prototype) [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #15743:
URL: https://github.com/apache/pinot/pull/15743#discussion_r2085623134
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/opt/rules/LiteModeWorkerAssignmentRule.java:
##########
@@ -0,0 +1,118 @@
+/**
+ * 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.physical.v2.opt.rules;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Random;
+import java.util.Set;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.pinot.calcite.rel.traits.PinotExecStrategyTrait;
+import org.apache.pinot.query.context.PhysicalPlannerContext;
+import org.apache.pinot.query.planner.physical.v2.ExchangeStrategy;
+import org.apache.pinot.query.planner.physical.v2.PRelNode;
+import org.apache.pinot.query.planner.physical.v2.PinotDataDistribution;
+import org.apache.pinot.query.planner.physical.v2.nodes.PhysicalExchange;
+import org.apache.pinot.query.planner.physical.v2.opt.PRelNodeTransformer;
+
+
+/**
+ * Lite mode uses a single worker for all stages except the leaf stage. Since leaf stage assignment is done before this
+ * rule is called, we simply need to sample a random worker from the leaf stage, and assign it to all the non-leaf
+ * plan nodes.
+ */
+public class LiteModeWorkerAssignmentRule implements PRelNodeTransformer {
+ private static final Random RANDOM = new Random();
Review Comment:
might consider to abstract it as a functional interface. One of selection strategy that I feel we can reduce the network cost is to select the server has most of segments
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [multistage] Multistage Engine Lite Mode (prototype) [pinot]
Posted by "ankitsultana (via GitHub)" <gi...@apache.org>.
ankitsultana commented on code in PR #15743:
URL: https://github.com/apache/pinot/pull/15743#discussion_r2085677019
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/opt/rules/LiteModeWorkerAssignmentRule.java:
##########
@@ -0,0 +1,118 @@
+/**
+ * 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.physical.v2.opt.rules;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Random;
+import java.util.Set;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.pinot.calcite.rel.traits.PinotExecStrategyTrait;
+import org.apache.pinot.query.context.PhysicalPlannerContext;
+import org.apache.pinot.query.planner.physical.v2.ExchangeStrategy;
+import org.apache.pinot.query.planner.physical.v2.PRelNode;
+import org.apache.pinot.query.planner.physical.v2.PinotDataDistribution;
+import org.apache.pinot.query.planner.physical.v2.nodes.PhysicalExchange;
+import org.apache.pinot.query.planner.physical.v2.opt.PRelNodeTransformer;
+
+
+/**
+ * Lite mode uses a single worker for all stages except the leaf stage. Since leaf stage assignment is done before this
+ * rule is called, we simply need to sample a random worker from the leaf stage, and assign it to all the non-leaf
+ * plan nodes.
+ */
+public class LiteModeWorkerAssignmentRule implements PRelNodeTransformer {
+ private static final Random RANDOM = new Random();
Review Comment:
Abstraction can definitely be considered as we evolve this. I think right now it's a bit too early. I am planning to add support for parallelism, server pruning, etc. soon so that will also increase the case handling of this Rule a bit.
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [multistage] Multistage Engine Lite Mode (prototype) [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #15743:
URL: https://github.com/apache/pinot/pull/15743#discussion_r2085623134
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/opt/rules/LiteModeWorkerAssignmentRule.java:
##########
@@ -0,0 +1,118 @@
+/**
+ * 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.physical.v2.opt.rules;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Random;
+import java.util.Set;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.pinot.calcite.rel.traits.PinotExecStrategyTrait;
+import org.apache.pinot.query.context.PhysicalPlannerContext;
+import org.apache.pinot.query.planner.physical.v2.ExchangeStrategy;
+import org.apache.pinot.query.planner.physical.v2.PRelNode;
+import org.apache.pinot.query.planner.physical.v2.PinotDataDistribution;
+import org.apache.pinot.query.planner.physical.v2.nodes.PhysicalExchange;
+import org.apache.pinot.query.planner.physical.v2.opt.PRelNodeTransformer;
+
+
+/**
+ * Lite mode uses a single worker for all stages except the leaf stage. Since leaf stage assignment is done before this
+ * rule is called, we simply need to sample a random worker from the leaf stage, and assign it to all the non-leaf
+ * plan nodes.
+ */
+public class LiteModeWorkerAssignmentRule implements PRelNodeTransformer {
+ private static final Random RANDOM = new Random();
Review Comment:
might consider to abstract it as a functional interface. One selection strategy that can reduce the network cost is to select the server has most of segments.
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org