This is an automated email from the ASF dual-hosted git repository.
yashmayya pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new d469cb17507 Support forcing pre-partitioned exchanges for window
functions to avoid data shuffle using query hint (#17395)
d469cb17507 is described below
commit d469cb17507f4e0c35f8bf639272109cb94d9f98
Author: Yash Mayya <[email protected]>
AuthorDate: Thu Jun 18 08:44:55 2026 -0700
Support forcing pre-partitioned exchanges for window functions to avoid
data shuffle using query hint (#17395)
---
.../pinot/calcite/rel/hint/PinotHintOptions.java | 12 +++
.../calcite/rel/hint/PinotHintStrategyTable.java | 1 +
.../rel/logical/PinotLogicalSortExchange.java | 29 +++++--
.../rel/rules/PinotRelDistributionTraitRule.java | 8 ++
.../rules/PinotWindowExchangeNodeInsertRule.java | 8 +-
.../planner/logical/RelToPlanNodeConverter.java | 2 +-
.../apache/pinot/query/QueryCompilationTest.java | 99 ++++++++++++++++++++++
.../resources/queries/ExplainPhysicalPlans.json | 34 ++++++++
.../src/test/resources/queries/QueryHints.json | 12 +++
9 files changed, 196 insertions(+), 9 deletions(-)
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java
index 1966ed3dd04..05258e3b000 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java
@@ -23,6 +23,7 @@ import javax.annotation.Nullable;
import org.apache.calcite.rel.RelDistribution;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.Window;
import org.apache.calcite.rel.hint.RelHint;
@@ -77,6 +78,17 @@ public class PinotHintOptions {
* BREAK: Break window cache build process, continue to perform WINDOW
operation, results might be partial.
*/
public static final String WINDOW_OVERFLOW_MODE = "window_overflow_mode";
+
+ /// Indicates that the input to the window is already partitioned by the
window keys, so we should avoid shuffling
+ /// to repartition.
+ public static final String IS_PARTITIONED_BY_WINDOW_KEYS =
"is_partitioned_by_window_keys";
+
+ @Nullable
+ public static Boolean isPartitionedByWindowKeys(Window window) {
+ String hint =
+ PinotHintStrategyTable.getHintOption(window.getHints(),
WINDOW_HINT_OPTIONS, IS_PARTITIONED_BY_WINDOW_KEYS);
+ return hint != null ? Boolean.parseBoolean(hint) : null;
+ }
}
public static class JoinHintOptions {
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintStrategyTable.java
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintStrategyTable.java
index 503dcfe80f6..8837d1d15ce 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintStrategyTable.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintStrategyTable.java
@@ -40,6 +40,7 @@ public class PinotHintStrategyTable {
.hintStrategy(PinotHintOptions.AGGREGATE_HINT_OPTIONS,
HintPredicates.AGGREGATE)
.hintStrategy(PinotHintOptions.JOIN_HINT_OPTIONS, HintPredicates.JOIN)
.hintStrategy(PinotHintOptions.TABLE_HINT_OPTIONS,
HintPredicates.TABLE_SCAN)
+ .hintStrategy(PinotHintOptions.WINDOW_HINT_OPTIONS,
HintPredicates.WINDOW)
.build();
/**
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/logical/PinotLogicalSortExchange.java
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/logical/PinotLogicalSortExchange.java
index 42bd1243390..01457c5e5d9 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/logical/PinotLogicalSortExchange.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/logical/PinotLogicalSortExchange.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.calcite.rel.logical;
+import javax.annotation.Nullable;
import org.apache.calcite.plan.Convention;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelTraitSet;
@@ -45,14 +46,18 @@ public class PinotLogicalSortExchange extends SortExchange {
protected final boolean _isSortOnSender;
protected final boolean _isSortOnReceiver;
protected final PinotRelExchangeType _exchangeType;
+ // Can be used to override the partitioning info calculated from the
distribution trait.
+ @Nullable
+ private final Boolean _prePartitioned;
private PinotLogicalSortExchange(RelOptCluster cluster, RelTraitSet
traitSet, RelNode input,
RelDistribution distribution, PinotRelExchangeType exchangeType,
RelCollation collation, boolean isSortOnSender,
- boolean isSortOnReceiver) {
+ boolean isSortOnReceiver, @Nullable Boolean prePartitioned) {
super(cluster, traitSet, input, distribution, collation);
_exchangeType = exchangeType;
_isSortOnSender = isSortOnSender;
_isSortOnReceiver = isSortOnReceiver;
+ _prePartitioned = prePartitioned;
}
/**
@@ -63,12 +68,19 @@ public class PinotLogicalSortExchange extends SortExchange {
_exchangeType = PinotRelExchangeType.STREAMING;
_isSortOnSender = false;
_isSortOnReceiver = true;
+ _prePartitioned = null;
}
public static PinotLogicalSortExchange create(RelNode input, RelDistribution
distribution, RelCollation collation,
boolean isSortOnSender, boolean isSortOnReceiver) {
return create(input, distribution,
PinotRelExchangeType.getDefaultExchangeType(), collation, isSortOnSender,
- isSortOnReceiver);
+ isSortOnReceiver, null);
+ }
+
+ public static PinotLogicalSortExchange create(RelNode input, RelDistribution
distribution, RelCollation collation,
+ boolean isSortOnSender, boolean isSortOnReceiver, @Nullable Boolean
prePartitioned) {
+ return create(input, distribution,
PinotRelExchangeType.getDefaultExchangeType(), collation, isSortOnSender,
+ isSortOnReceiver, prePartitioned);
}
/**
@@ -80,15 +92,17 @@ public class PinotLogicalSortExchange extends SortExchange {
* @param collation array of sort specifications
* @param isSortOnSender whether to sort on the sender
* @param isSortOnReceiver whether to sort on receiver
+ * @param prePartitioned whether the exchange is pre-partitioned
*/
public static PinotLogicalSortExchange create(RelNode input, RelDistribution
distribution,
- PinotRelExchangeType exchangeType, RelCollation collation, boolean
isSortOnSender, boolean isSortOnReceiver) {
+ PinotRelExchangeType exchangeType, RelCollation collation, boolean
isSortOnSender, boolean isSortOnReceiver,
+ @Nullable Boolean prePartitioned) {
RelOptCluster cluster = input.getCluster();
collation = RelCollationTraitDef.INSTANCE.canonize(collation);
distribution = RelDistributionTraitDef.INSTANCE.canonize(distribution);
RelTraitSet traitSet =
input.getTraitSet().replace(Convention.NONE).replace(distribution).replace(collation);
return new PinotLogicalSortExchange(cluster, traitSet, input,
distribution, exchangeType, collation, isSortOnSender,
- isSortOnReceiver);
+ isSortOnReceiver, prePartitioned);
}
//~ Methods ----------------------------------------------------------------
@@ -97,7 +111,7 @@ public class PinotLogicalSortExchange extends SortExchange {
public SortExchange copy(RelTraitSet traitSet, RelNode newInput,
RelDistribution newDistribution,
RelCollation newCollation) {
return new PinotLogicalSortExchange(this.getCluster(), traitSet, newInput,
newDistribution, _exchangeType,
- newCollation, _isSortOnSender, _isSortOnReceiver);
+ newCollation, _isSortOnSender, _isSortOnReceiver, _prePartitioned);
}
@Override
@@ -121,4 +135,9 @@ public class PinotLogicalSortExchange extends SortExchange {
public PinotRelExchangeType getExchangeType() {
return _exchangeType;
}
+
+ @Nullable
+ public Boolean getPrePartitioned() {
+ return _prePartitioned;
+ }
}
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotRelDistributionTraitRule.java
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotRelDistributionTraitRule.java
index 978986c5e16..a77780f01ee 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotRelDistributionTraitRule.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotRelDistributionTraitRule.java
@@ -36,6 +36,7 @@ import org.apache.calcite.rel.core.TableScan;
import org.apache.calcite.rel.logical.LogicalFilter;
import org.apache.calcite.rel.logical.LogicalJoin;
import org.apache.calcite.rel.logical.LogicalProject;
+import org.apache.calcite.rel.logical.LogicalWindow;
import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.tools.RelBuilderFactory;
import org.apache.calcite.util.mapping.IntPair;
@@ -158,6 +159,13 @@ public class PinotRelDistributionTraitRule extends
RelOptRule {
// b/c the Join node always puts left relation RowTypes then right
relation RowTypes sequentially.
return inputRelDistribution;
}
+ } else if (node instanceof LogicalWindow) {
+ assert inputs.size() == 1;
+ // Window input should be an exchange node that is hash distributed by
the window's partition keys
+ RelDistribution inputRelDistribution =
inputs.get(0).getTraitSet().getDistribution();
+ if (inputRelDistribution != null) {
+ return inputRelDistribution;
+ }
}
// TODO: add the rest of the nodes.
return computeCurrentDistribution(node);
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotWindowExchangeNodeInsertRule.java
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotWindowExchangeNodeInsertRule.java
index 5eaade4125f..f8f3e0978b9 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotWindowExchangeNodeInsertRule.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotWindowExchangeNodeInsertRule.java
@@ -43,6 +43,7 @@ import org.apache.calcite.rex.RexNode;
import org.apache.calcite.sql.SqlKind;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.calcite.tools.RelBuilderFactory;
+import org.apache.pinot.calcite.rel.hint.PinotHintOptions;
import org.apache.pinot.calcite.rel.logical.PinotLogicalExchange;
import org.apache.pinot.calcite.rel.logical.PinotLogicalSortExchange;
@@ -119,11 +120,12 @@ public class PinotWindowExchangeNodeInsertRule extends
RelOptRule {
// Assess whether this is a PARTITION BY only query or not (includes
queries of the type where PARTITION BY and
// ORDER BY key(s) are the same)
boolean isPartitionByOnly = isPartitionByOnlyQuery(windowGroup);
-
+ // Force pre-partitioned exchange when 'is_partitioned_by_window_keys'
hint is provided
+ Boolean prePartitioned =
PinotHintOptions.WindowHintOptions.isPartitionedByWindowKeys(window);
if (isPartitionByOnly) {
// Only PARTITION BY or PARTITION BY and ORDER BY on the same key(s)
// Add an Exchange hashed on the partition by keys
- exchange = PinotLogicalExchange.create(input,
RelDistributions.hash(windowGroup.keys.toList()));
+ exchange = PinotLogicalExchange.create(input,
RelDistributions.hash(windowGroup.keys.toList()), prePartitioned);
} else {
// PARTITION BY and ORDER BY on different key(s)
// Add a LogicalSortExchange hashed on the partition by keys and
collation based on order by keys
@@ -132,7 +134,7 @@ public class PinotWindowExchangeNodeInsertRule extends
RelOptRule {
// sorting on the receiver side can be a no-op. Add support for
this hint and pass it on. Until sender
// side sorting is implemented, setting this hint will throw an
error on execution.
exchange = PinotLogicalSortExchange.create(input,
RelDistributions.hash(windowGroup.keys.toList()),
- windowGroup.orderKeys, false, true);
+ windowGroup.orderKeys, false, true, prePartitioned);
}
}
// NOTE: Need to create a new LogicalWindow to use the modified window
group.
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/logical/RelToPlanNodeConverter.java
b/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/logical/RelToPlanNodeConverter.java
index cd737896c73..c347e18a79d 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/logical/RelToPlanNodeConverter.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/logical/RelToPlanNodeConverter.java
@@ -581,7 +581,7 @@ public final class RelToPlanNodeConverter {
PinotLogicalSortExchange sortExchange = (PinotLogicalSortExchange) node;
exchangeType = sortExchange.getExchangeType();
keys = distribution.getKeys();
- prePartitioned = null;
+ prePartitioned = sortExchange.getPrePartitioned();
collations = sortExchange.getCollation().getFieldCollations();
sortOnSender = sortExchange.isSortOnSender();
sortOnReceiver = sortExchange.isSortOnReceiver();
diff --git
a/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryCompilationTest.java
b/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryCompilationTest.java
index 142eddcb05b..1bf31abd431 100644
---
a/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryCompilationTest.java
+++
b/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryCompilationTest.java
@@ -38,8 +38,10 @@ import org.apache.pinot.query.planner.plannode.BasePlanNode;
import org.apache.pinot.query.planner.plannode.FilterNode;
import org.apache.pinot.query.planner.plannode.JoinNode;
import org.apache.pinot.query.planner.plannode.MailboxReceiveNode;
+import org.apache.pinot.query.planner.plannode.MailboxSendNode;
import org.apache.pinot.query.planner.plannode.PlanNode;
import org.apache.pinot.query.planner.plannode.ProjectNode;
+import org.apache.pinot.query.planner.plannode.WindowNode;
import org.apache.pinot.query.routing.QueryServerInstance;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Ignore;
@@ -953,6 +955,103 @@ public class QueryCompilationTest extends
QueryEnvironmentTestBase {
"RIGHT side of RIGHT non-equi join should be RANDOM");
}
+ /**
+ * The {@code windowOptions(is_partitioned_by_window_keys='true')} hint
forces a pre-partitioned (direct) exchange
+ * below the window, avoiding a data shuffle. Here the window partitions by
{@code col1}, which is NOT table a's
+ * partition column ({@code col2}), so without the hint the planner would
shuffle. This exercises the
+ * {@link org.apache.pinot.calcite.rel.logical.PinotLogicalExchange} path
(PARTITION BY only).
+ */
+ @Test
+ public void testWindowPartitionByKeysHintForcesPrePartitionedExchange() {
+ String query = "SELECT /*+
windowOptions(is_partitioned_by_window_keys='true') */ "
+ + "col1, SUM(col3) OVER (PARTITION BY col1) FROM a";
+ MailboxSendNode sendNode =
findWindowInputSendNode(_queryEnvironment.planQuery(query));
+ assertEquals(sendNode.getDistributionType(),
RelDistribution.Type.HASH_DISTRIBUTED);
+ assertTrue(sendNode.isPrePartitioned(),
+ "windowOptions(is_partitioned_by_window_keys='true') should force a
pre-partitioned exchange");
+ }
+
+ /**
+ * Without the hint and with a window partition key that does not match the
table's partitioning, the exchange below
+ * the window must be a regular (shuffled) exchange.
+ */
+ @Test
+ public void testWindowWithoutHintIsNotPrePartitioned() {
+ String query = "SELECT col1, SUM(col3) OVER (PARTITION BY col1) FROM a";
+ MailboxSendNode sendNode =
findWindowInputSendNode(_queryEnvironment.planQuery(query));
+ assertFalse(sendNode.isPrePartitioned(),
+ "Without the hint and matching partitioning, the window exchange
should be a full shuffle");
+ }
+
+ /**
+ * The hint must also flow through the {@link
org.apache.pinot.calcite.rel.logical.PinotLogicalSortExchange} path,
+ * used when PARTITION BY and ORDER BY are on different keys. This is the
path the PR fixes: previously
+ * {@code RelToPlanNodeConverter} hardcoded {@code prePartitioned = null}
for sort exchanges, dropping the hint.
+ */
+ @Test
+ public void testWindowPartitionByKeysHintForcesPrePartitionedSortExchange() {
+ String query = "SELECT /*+
windowOptions(is_partitioned_by_window_keys='true') */ "
+ + "col1, SUM(col3) OVER (PARTITION BY col1 ORDER BY col3) FROM a";
+ MailboxSendNode sendNode =
findWindowInputSendNode(_queryEnvironment.planQuery(query));
+ assertEquals(sendNode.getDistributionType(),
RelDistribution.Type.HASH_DISTRIBUTED);
+ assertTrue(sendNode.isPrePartitioned(),
+ "windowOptions hint should force a pre-partitioned sort exchange
(PARTITION BY + ORDER BY on different keys)");
+ }
+
+ /**
+ * Setting the hint to {@code 'false'} overrides the planner's automatic
detection of pre-partitioning. Here table a
+ * is declared partitioned by {@code col2} (via tableOptions) and the window
also partitions by {@code col2}, so the
+ * planner would otherwise auto-detect a pre-partitioned exchange; the hint
disables it.
+ */
+ @Test
+ public void
testWindowPartitionByKeysHintFalseDisablesAutoDetectedPrePartitioning() {
+ String query = "SELECT /*+
windowOptions(is_partitioned_by_window_keys='false') */ "
+ + "col1, SUM(col3) OVER (PARTITION BY col2) "
+ + "FROM a /*+ tableOptions(partition_function='hashcode',
partition_key='col2', partition_size='4') */";
+ MailboxSendNode sendNode =
findWindowInputSendNode(_queryEnvironment.planQuery(query));
+ assertFalse(sendNode.isPrePartitioned(),
+ "windowOptions(is_partitioned_by_window_keys='false') should disable
auto-detected pre-partitioning");
+ }
+
+ /**
+ * With matching tableOptions partitioning and no window hint, the planner
auto-detects pre-partitioning. This must
+ * hold for both window exchange paths: PinotLogicalExchange (PARTITION BY
only) and PinotLogicalSortExchange
+ * (PARTITION BY + ORDER BY). The latter verifies the PR preserves the
{@code null -> auto-detect} behavior.
+ */
+ @Test
+ public void testWindowAutoDetectsPrePartitioningWithoutHint() {
+ String partitionOnly = "SELECT col1, SUM(col3) OVER (PARTITION BY col2) "
+ + "FROM a /*+ tableOptions(partition_function='hashcode',
partition_key='col2', partition_size='4') */";
+
assertTrue(findWindowInputSendNode(_queryEnvironment.planQuery(partitionOnly)).isPrePartitioned(),
+ "PARTITION BY on the table's partition column should auto-detect a
pre-partitioned exchange");
+
+ String partitionAndOrder = "SELECT col1, SUM(col3) OVER (PARTITION BY col2
ORDER BY col3) "
+ + "FROM a /*+ tableOptions(partition_function='hashcode',
partition_key='col2', partition_size='4') */";
+
assertTrue(findWindowInputSendNode(_queryEnvironment.planQuery(partitionAndOrder)).isPrePartitioned(),
+ "Sort exchange should also auto-detect pre-partitioning when the table
is partitioned by the window key");
+ }
+
+ /**
+ * Finds the {@link MailboxSendNode} that feeds the (single) WINDOW stage's
input exchange, i.e. the sender side of
+ * the exchange inserted directly below the window. The {@code
prePartitioned} flag lives on this send node.
+ */
+ private MailboxSendNode findWindowInputSendNode(DispatchableSubPlan
dispatchableSubPlan) {
+ WindowNode window = null;
+ for (DispatchablePlanFragment fragment :
dispatchableSubPlan.getQueryStages()) {
+ window = findNodeOfType(fragment.getPlanFragment().getFragmentRoot(),
WindowNode.class);
+ if (window != null) {
+ break;
+ }
+ }
+ assertNotNull(window, "Expected a WINDOW node in the plan");
+ MailboxReceiveNode receiveNode = findNodeOfType(window,
MailboxReceiveNode.class);
+ assertNotNull(receiveNode, "Expected the WINDOW input to be a mailbox
exchange");
+ PlanNode senderRoot =
+
dispatchableSubPlan.getQueryStageMap().get(receiveNode.getSenderStageId()).getPlanFragment().getFragmentRoot();
+ assertTrue(senderRoot instanceof MailboxSendNode, "Sender fragment root
should be a MailboxSendNode");
+ return (MailboxSendNode) senderRoot;
+ }
+
/**
* Finds the DispatchablePlanFragment containing a JoinNode (non-leaf,
non-root stage).
*/
diff --git
a/pinot-query-planner/src/test/resources/queries/ExplainPhysicalPlans.json
b/pinot-query-planner/src/test/resources/queries/ExplainPhysicalPlans.json
index b64b2da593d..55535941b97 100644
--- a/pinot-query-planner/src/test/resources/queries/ExplainPhysicalPlans.json
+++ b/pinot-query-planner/src/test/resources/queries/ExplainPhysicalPlans.json
@@ -716,6 +716,40 @@
"\n └── [5]@localhost:1|[0] TABLE
SCAN (b) null",
"\n"
]
+ },
+ {
+ "description": "window function (PARTITION BY) without hint shuffles
(hash redistributes) its input to all window-stage workers",
+ "sql": "EXPLAIN IMPLEMENTATION PLAN FOR SELECT col1, SUM(col3) OVER
(PARTITION BY col1) FROM a",
+ "output": [
+ "[0]@localhost:3|[0] MAIL_RECEIVE(BROADCAST_DISTRIBUTED)",
+ "\n├── [1]@localhost:2|[1]
MAIL_SEND(BROADCAST_DISTRIBUTED)->{[0]@localhost:3|[0]} (Subtree Omitted)",
+ "\n└── [1]@localhost:1|[0]
MAIL_SEND(BROADCAST_DISTRIBUTED)->{[0]@localhost:3|[0]}",
+ "\n └── [1]@localhost:1|[0] PROJECT",
+ "\n └── [1]@localhost:1|[0] WINDOW",
+ "\n └── [1]@localhost:1|[0]
MAIL_RECEIVE(HASH_DISTRIBUTED)",
+ "\n ├── [2]@localhost:2|[1]
MAIL_SEND(HASH_DISTRIBUTED)->{[1]@localhost:1|[0],[1]@localhost:2|[1]} (Subtree
Omitted)",
+ "\n └── [2]@localhost:1|[0]
MAIL_SEND(HASH_DISTRIBUTED)->{[1]@localhost:1|[0],[1]@localhost:2|[1]}",
+ "\n └── [2]@localhost:1|[0] PROJECT",
+ "\n └── [2]@localhost:1|[0] TABLE SCAN (a)
null",
+ "\n"
+ ]
+ },
+ {
+ "description": "window function (PARTITION BY) with
is_partitioned_by_window_keys hint uses a pre-partitioned (direct, 1-to-1)
exchange to avoid the shuffle",
+ "sql": "EXPLAIN IMPLEMENTATION PLAN FOR SELECT /*+
windowOptions(is_partitioned_by_window_keys='true') */ col1, SUM(col3) OVER
(PARTITION BY col1) FROM a",
+ "output": [
+ "[0]@localhost:3|[0] MAIL_RECEIVE(BROADCAST_DISTRIBUTED)",
+ "\n├── [1]@localhost:2|[1]
MAIL_SEND(BROADCAST_DISTRIBUTED)->{[0]@localhost:3|[0]} (Subtree Omitted)",
+ "\n└── [1]@localhost:1|[0]
MAIL_SEND(BROADCAST_DISTRIBUTED)->{[0]@localhost:3|[0]}",
+ "\n └── [1]@localhost:1|[0] PROJECT",
+ "\n └── [1]@localhost:1|[0] WINDOW",
+ "\n └── [1]@localhost:1|[0]
MAIL_RECEIVE(HASH_DISTRIBUTED)",
+ "\n ├── [2]@localhost:2|[1]
MAIL_SEND(HASH_DISTRIBUTED)[PARTITIONED]->{[1]@localhost:2|[1]} (Subtree
Omitted)",
+ "\n └── [2]@localhost:1|[0]
MAIL_SEND(HASH_DISTRIBUTED)[PARTITIONED]->{[1]@localhost:1|[0]}",
+ "\n └── [2]@localhost:1|[0] PROJECT",
+ "\n └── [2]@localhost:1|[0] TABLE SCAN (a)
null",
+ "\n"
+ ]
}
]
}
diff --git a/pinot-query-runtime/src/test/resources/queries/QueryHints.json
b/pinot-query-runtime/src/test/resources/queries/QueryHints.json
index 2af747c2704..8b8bcd6f0bb 100644
--- a/pinot-query-runtime/src/test/resources/queries/QueryHints.json
+++ b/pinot-query-runtime/src/test/resources/queries/QueryHints.json
@@ -659,6 +659,18 @@
{
"description": "Local replicated JOIN with partition hint and
parallelism",
"sql": "SELECT /*+ joinOptions(left_distribution_type = 'local',
right_distribution_type = 'local') */ {tbl1}.num, {tbl1}.name, {tbl2}.num,
{tbl2}.val FROM {tbl1} /*+ tableOptions(partition_function='hashcode',
partition_key='num', partition_size='4', partition_parallelism='2') */ JOIN
{tbl2} /*+ tableOptions(is_replicated='true') */ ON {tbl1}.num = {tbl2}.num"
+ },
+ {
+ "description": "Window PARTITION BY the table's partition column with
is_partitioned_by_window_keys hint forces a pre-partitioned (direct) exchange;
tbl1 is partitioned by num so results stay correct",
+ "sql": "SELECT /*+ windowOptions(is_partitioned_by_window_keys='true')
*/ {tbl1}.num, SUM({tbl1}.num) OVER (PARTITION BY {tbl1}.num) FROM {tbl1}"
+ },
+ {
+ "description": "Window PARTITION BY + ORDER BY on different keys with
is_partitioned_by_window_keys hint forces a pre-partitioned sort exchange; tbl1
is partitioned by num so results stay correct",
+ "sql": "SELECT /*+ windowOptions(is_partitioned_by_window_keys='true')
*/ {tbl1}.num, {tbl1}.name, SUM({tbl1}.num) OVER (PARTITION BY {tbl1}.num ORDER
BY {tbl1}.name) FROM {tbl1}"
+ },
+ {
+ "description": "Window with is_partitioned_by_window_keys='false'
disables the auto-detected pre-partitioning (tbl1 is partitioned by num) and
falls back to a shuffle; results stay correct",
+ "sql": "SELECT /*+ windowOptions(is_partitioned_by_window_keys='false')
*/ {tbl1}.num, SUM({tbl1}.num) OVER (PARTITION BY {tbl1}.num) FROM {tbl1} /*+
tableOptions(partition_function='hashcode', partition_key='num',
partition_size='4') */"
}
]}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
