You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "dwsmith1983 (via GitHub)" <gi...@apache.org> on 2026/10/08 18:03:28 UTC
[PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
dwsmith1983 opened a new pull request, #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785
## Which issue does this PR close?
Closes #6770.
## Rationale for this change
With `spark.comet.exec.forceShuffledHashJoin` on, `RewriteJoin` turned sort-merge joins into hash joins inside `CometExecRule`, after `EnsureRequirements` and `RemoveRedundantSorts` had run. When a sort-merge join's output ordering already satisfied a sort above it, Spark had removed that sort as redundant, and once the join became a hash join nothing put it back. `sortWithinPartitions` and `SORT BY` on the join key returned unsorted partitions, and a partitioned insert keyed on the join key failed with `FileAlreadyExistsException`, because the planned write's partition sort was the one removed. #6384 restores orderings an operator above still requires, which covers a kept sort-merge join or a window, but here nothing left in the plan requires the ordering.
## What changes are included in this PR?
- `RewriteJoin` is replaced by `CometShuffledHashJoinStrategy`, injected with `injectPlannerStrategy`. Injected strategies run before Spark's `JoinSelection`, so the join is a `ShuffledHashJoinExec` from the start and `EnsureRequirements` and `RemoveRedundantSorts` place and keep the sorts themselves. The strategy also runs on every AQE re-plan, where the join's children carry the materialized shuffle sizes, as the size check did before.
- It plans a hash join only where Spark would otherwise plan a sort-merge join, and leaves the join to Spark when:
- Spark would broadcast it, either by size (`canPlanAsBroadcastHashJoin`, with AQE's threshold for runtime sizes) or because AQE already planned a broadcast stage for one side;
- either side has a join strategy hint (`BROADCAST`, `MERGE`, `SHUFFLE_HASH`, `SHUFFLE_REPLICATE_NL`); AQE's own hints do not count;
- the keys cannot be sorted, so there was no sort-merge join to replace;
- on Spark 4, the keys are not binary stable under their collation (`hashJoinSupported`), or the join is a `LeftSingle`, which Spark never plans as a sort-merge join;
- the build side is over `maxBuildSize`, or it is `LeftSemi` or `ExistenceJoin` building on the right (#2667, #2697). Spark then plans the sort-merge join, and the strategy records the same explain info or fallback reason as before on it.
- The build side choice and the size limit from #6384 move into the strategy unchanged. `restoreRequiredOrdering` is gone, since `EnsureRequirements` now adds those sorts.
- The strategy does nothing when Comet or `spark.comet.exec.enabled` is off, in plan-only mode, or for a streaming join.
Behavior changes:
- A join strategy hint now wins over `forceShuffledHashJoin`. Before, a `MERGE` hint was converted too.
- On Spark 4, joins on collated keys that are not binary stable are no longer forced into Spark's hash join.
- The plan-only report shows the sort-merge joins Spark plans, not the hash joins a real run would use; `understanding-comet-plans.md` says so.
- Joins of static tables inside a streaming query are planned by the session's strategies too, so with the force config on they run as Spark's own `ShuffledHashJoin` (Comet leaves streaming plans alone). The tuning guide says so.
`branch-1.1` has the same rewrite, so 1.1.0 loses these sorts too. A backport needs this strategy rather than a cherry-pick, since `branch-1.1` has no size limit.
## How are these changes tested?
`CometJoinSuite`, with AQE off and on where it matters:
- `sortWithinPartitions` and `SORT BY` on the join key keep every partition sorted, for inner, right outer (building on the smaller left side) and full outer joins, and the rows match Spark.
- A partitioned insert keyed on the join key runs a native hash join, writes all 1000 rows, and writes one file per partition.
- A streamed side already partitioned and sorted on the key keeps its order through the hash join when no exchange is added.
- `MERGE`, `SHUFFLE_HASH` (Spark builds the hinted side) and `BROADCAST` hints are respected.
- A runtime broadcast under AQE, and a broadcast Spark plans from the static size, stay broadcasts.
- A skewed join is split by `OptimizeSkewedJoin`.
- On Spark 4, a join on a computed key with a `UTF8_LCASE` collation keeps the sort-merge join and returns Spark's rows.
- The strategy plans no hash join with Comet off, with `spark.comet.exec.enabled` off, or in plan-only mode.
- The #6384 size limit tests now run without the `MERGE` hint they relied on, and an RDD-backed build side, whose size Spark reports as `spark.sql.defaultSizeInBytes`, keeps the sort-merge join.
- The #6673 tests for a kept sort-merge join, `LeftSemi`, `ExistenceJoin` or window above a forced hash join pass with the sorts `EnsureRequirements` adds.
The new sort and insert tests fail on main. Removing the plan-only check fails its test, and replacing the Spark 4 `hashJoinSupported` check with `true` fails the collation test. `CometJoinSuite` and `CometConfSuite` pass on Spark 3.4, 3.5, 4.0 and 4.1, Spark 4.2 compiles, and the strict warnings build passes.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#discussion_r4321400980
##########
spark/src/main/scala/org/apache/comet/rules/CometShuffledHashJoinStrategy.scala:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.comet.rules
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.RowOrdering
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
+import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{Join, JoinHint, JoinStrategyHint, LogicalPlan}
+import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
+import org.apache.spark.sql.execution.adaptive.{BroadcastQueryStageExec, LogicalQueryStage}
+import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.CometConf
+import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, withFallbackReason, withInfo}
+import org.apache.comet.shims.ShimJoinSelection
+
+/**
+ * Plans an equi-join as a ShuffledHashJoinExec where Spark would plan a SortMergeJoinExec, when
+ * spark.comet.exec.forceShuffledHashJoin is on. Choosing the join here, before EnsureRequirements
+ * and RemoveRedundantSorts run, lets Spark place and keep every sort the plan needs above the
+ * join. Planner strategies also run again on every AQE re-plan, where the join's children carry
+ * the materialized shuffle sizes.
+ *
+ * Adapted from the equivalent rule in Apache Gluten.
+ */
+case class CometShuffledHashJoinStrategy(session: SparkSession)
+ extends SparkStrategy
+ with ShimJoinSelection {
+
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case join: Join if isEnabled(join) => planJoin(join)
+ case _ => Nil
+ }
+
+ private def sqlConf: SQLConf = session.sessionState.conf
+
+ // Planner strategies run whether or not Comet is enabled, and before CometRule, so plan-only
+ // mode needs its own check here.
+ private def isEnabled(join: Join): Boolean =
+ CometConf.COMET_FORCE_SHJ.get(sqlConf) &&
+ CometConf.COMET_EXEC_ENABLED.get(sqlConf) &&
+ !CometConf.COMET_EXPLAIN_PLAN_ONLY_ENABLED.get(sqlConf) &&
+ !join.isStreaming &&
+ isCometLoaded(sqlConf)
+
+ private def planJoin(join: Join): Seq[SparkPlan] = join match {
+ case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, _, left, right, hint)
+ if !isLeftToSpark(join, hint) && canSortMergeJoin(joinType) &&
+ RowOrdering.isOrderable(leftKeys) && hashJoinSupportedShim(leftKeys, rightKeys) =>
+ buildSide(join) match {
+ case Some(BuildRight) if joinType == LeftSemi || joinType.isInstanceOf[ExistenceJoin] =>
+ // LeftSemi https://github.com/apache/datafusion-comet/issues/2667
+ // ExistenceJoin https://github.com/apache/datafusion-comet/issues/2697
+ planBySpark(join) { smj =>
+ withFallbackReason(smj, declineReason(s"BuildRight with $joinType is not supported"))
+ }
+ case Some(side) =>
+ buildSideOverLimit(join, side) match {
+ case Some(reason) =>
+ // The sort-merge join still runs natively, so this is information for extended
+ // explain rather than a fallback reason.
+ planBySpark(join)(smj => withInfo(smj, declineReason(reason)))
+ case None =>
+ Seq(
+ ShuffledHashJoinExec(
Review Comment:
[P1] Preserve the final ordering when a checkpointed probe cannot receive a native ordering declaration. In an enabled Comet session, disable AQE and broadcasts, set two shuffle partitions, and enable `forceShuffledHashJoin`, `spark.comet.convert.rdd.enabled`, and `spark.comet.exec.range.enabled`. Build the probe with `spark.range(2000).selectExpr("id % 100 as k", "id as v").repartition(2, col("k")).sortWithinPartitions("k", "v").localCheckpoint(eager = true)`, left-join `spark.range(10).selectExpr("id * 10 as k")`, then request `sortWithinPartitions("k", "v")`. Selecting the hash join here makes Spark remove the final sort because `RDDScanExec` retains the checkpoint’s ordering. Unlike the sorted-cache case already discussed, this input hits `mayReportFalseOrder`’s unknown-leaf branch. `CometHashJoin.doConvert` therefore sends no `streamedSortOrders` but still converts the join. DataFusion then emits unmatched probe rows after matched rows, violating the requested ordering. T
he exact base retains the final `(k, v)` sort. Please retain that sort or fall back to Spark when an ordered probe cannot safely receive the declaration.
Evidence: The exact-source Spark 4.1.3 differential harness confirms that the base retains one final Sort and the head retains none for this checkpoint query. The checkpoint is an RDDScanExec with two ordering expressions and no children. Source tracing through CometScanWrapper and CometSparkToColumnarExec reaches operators.scala:3036, causing the declaration at lines 2951–2969 to be empty. Replaying the exported Int64 checkpoint partitions through DataFusion 55.2.0’s right outer hash join preserves all 2,000 rows but produces ordering transitions (60,1960) -> (2,2) and (90,1990) -> (1,1). Using the current SortedInputExec removes both violations. Reproduction sources and logs: /tmp/review6785-153a-validation-qhry1dr7/{CachedOrderProbe.scala,run-checkpoint.log,CheckpointProbe.scala,checkpoint-run.log,checkpoint_native.rs,checkpoint-native-run.log}. This is a planning-plus-kernel reproduction, not an integrated Comet execution.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.
dwsmith1983 commented on code in PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#discussion_r4371595549
##########
spark/src/main/scala/org/apache/comet/rules/CometShuffledHashJoinStrategy.scala:
##########
@@ -0,0 +1,175 @@
+/*
+ * 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.comet.rules
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.RowOrdering
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
+import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{Join, JoinHint, JoinStrategyHint, LogicalPlan}
+import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
+import org.apache.spark.sql.execution.adaptive.{BroadcastQueryStageExec, LogicalQueryStage}
+import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.CometConf
+import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, withFallbackReason, withInfo}
+import org.apache.comet.shims.ShimJoinSelection
+
+/**
+ * Plans a ShuffledHashJoinExec in place of a SortMergeJoinExec, before EnsureRequirements and
+ * RemoveRedundantSorts, so the sorts above the join survive. Adapted from Apache Gluten.
+ */
+case class CometShuffledHashJoinStrategy(session: SparkSession)
+ extends SparkStrategy
+ with ShimJoinSelection {
+
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case join: Join if isEnabled(join) => planJoin(join)
+ case _ => Nil
+ }
+
+ private def sqlConf: SQLConf = session.sessionState.conf
+
+ // Planner strategies run with Comet off and before CometRule, which applies plan-only mode.
+ private def isEnabled(join: Join): Boolean =
+ CometConf.COMET_FORCE_SHJ.get(sqlConf) &&
+ CometConf.COMET_EXEC_ENABLED.get(sqlConf) &&
+ !CometConf.COMET_EXPLAIN_PLAN_ONLY_ENABLED.get(sqlConf) &&
+ !join.isStreaming &&
+ isCometLoaded(sqlConf)
+
+ private def planJoin(join: Join): Seq[SparkPlan] = join match {
+ case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, _, left, right, hint)
+ if !isLeftToSpark(join, hint) && canSortMergeJoin(joinType) &&
+ RowOrdering.isOrderable(leftKeys) && hashJoinSupportedShim(leftKeys, rightKeys) =>
+ buildSide(join) match {
+ case Some(BuildRight) if joinType == LeftSemi || joinType.isInstanceOf[ExistenceJoin] =>
+ // LeftSemi https://github.com/apache/datafusion-comet/issues/2667
+ // ExistenceJoin https://github.com/apache/datafusion-comet/issues/2697
+ planBySpark(join) { smj =>
+ withFallbackReason(smj, declineReason(s"BuildRight with $joinType is not supported"))
+ }
+ case Some(side) =>
+ buildSideOverLimit(join, side) match {
+ case Some(reason) =>
+ // Not a fallback: the kept sort-merge join still runs natively.
+ planBySpark(join)(smj => withInfo(smj, declineReason(reason)))
+ case None =>
+ Seq(
+ ShuffledHashJoinExec(
Review Comment:
The nested loop join is the part that loses the order, and main has the same problem: your query with `SHUFFLE_HASH(b)` or `BROADCAST(b)` on the outer join, or the nested loop join alone under `ORDER BY x`, comes back out of order there too (#6898). Spark's `BroadcastNestedLoopJoinExec` keeps its streamed side's order and reports it, while DataFusion's emits its rows build row by build row. `CometBroadcastNestedLoopJoinExec` now stays in Spark whenever Spark reports an output order for the join, so the sort Spark drops is no longer needed. That fixes #6898 as well.
`BroadcastNestedLoopJoin over an ordered streamed side falls back and keeps its order` in `CometJoinSuite` runs your query with forcing on, the two hinted variants, and the nested loop join alone under `ORDER BY x`, and checks that the rows come back sorted.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.
dwsmith1983 commented on PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#issuecomment-6292589203
@andygrove the two label runs on 153a343f4 failed only in [Preflight](https://github.com/apache/datafusion-comet/actions/runs/39416107688/job/118605467360), on an npm fetch: `No matching version found for puppeteer-core@25.15.0`. Every test job passed. Could you rerun it?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.
dwsmith1983 commented on code in PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#discussion_r4371605376
##########
spark/src/main/scala/org/apache/comet/rules/CometShuffledHashJoinStrategy.scala:
##########
@@ -0,0 +1,175 @@
+/*
+ * 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.comet.rules
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.RowOrdering
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
+import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{Join, JoinHint, JoinStrategyHint, LogicalPlan}
+import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
+import org.apache.spark.sql.execution.adaptive.{BroadcastQueryStageExec, LogicalQueryStage}
+import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.CometConf
+import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, withFallbackReason, withInfo}
+import org.apache.comet.shims.ShimJoinSelection
+
+/**
+ * Plans a ShuffledHashJoinExec in place of a SortMergeJoinExec, before EnsureRequirements and
+ * RemoveRedundantSorts, so the sorts above the join survive. Adapted from Apache Gluten.
+ */
+case class CometShuffledHashJoinStrategy(session: SparkSession)
+ extends SparkStrategy
+ with ShimJoinSelection {
+
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case join: Join if isEnabled(join) => planJoin(join)
+ case _ => Nil
+ }
+
+ private def sqlConf: SQLConf = session.sessionState.conf
+
+ // Planner strategies run with Comet off and before CometRule, which applies plan-only mode.
+ private def isEnabled(join: Join): Boolean =
+ CometConf.COMET_FORCE_SHJ.get(sqlConf) &&
+ CometConf.COMET_EXEC_ENABLED.get(sqlConf) &&
+ !CometConf.COMET_EXPLAIN_PLAN_ONLY_ENABLED.get(sqlConf) &&
+ !join.isStreaming &&
+ isCometLoaded(sqlConf)
+
+ private def planJoin(join: Join): Seq[SparkPlan] = join match {
+ case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, _, left, right, hint)
+ if !isLeftToSpark(join, hint) && canSortMergeJoin(joinType) &&
+ RowOrdering.isOrderable(leftKeys) && hashJoinSupportedShim(leftKeys, rightKeys) =>
+ buildSide(join) match {
+ case Some(BuildRight) if joinType == LeftSemi || joinType.isInstanceOf[ExistenceJoin] =>
+ // LeftSemi https://github.com/apache/datafusion-comet/issues/2667
+ // ExistenceJoin https://github.com/apache/datafusion-comet/issues/2697
+ planBySpark(join) { smj =>
+ withFallbackReason(smj, declineReason(s"BuildRight with $joinType is not supported"))
+ }
+ case Some(side) =>
+ buildSideOverLimit(join, side) match {
+ case Some(reason) =>
+ // Not a fallback: the kept sort-merge join still runs natively.
+ planBySpark(join)(smj => withInfo(smj, declineReason(reason)))
+ case None =>
+ Seq(
+ ShuffledHashJoinExec(
+ leftKeys,
+ rightKeys,
+ joinType,
+ side,
+ condition,
+ planLater(left),
Review Comment:
`CometHashJoin` now sends the streamed side's order for every join that reports one, not only the two outer shapes, whenever `mayReportFalseOrder` finds nothing under the streamed side that may have moved its rows. An inner join keeps its probe order either way, so declaring it moves no rows, and the native window above now sees the order. When the order cannot be checked or its first key cannot be expressed, an inner join declares nothing and stays native, as before.
`an inner hash join declares a sorted cache's order for a native window above it` runs your shape with forcing on and with `shuffle_hash` and `broadcast` hints. `an inner hash join declares the part of the streamed order Comet can express` and `an inner hash join over a checkpointed streamed side declares no order` cover the two cases that declare nothing, and `inner_hash_join_declares_streamed_order_on_probe_input` in `planner.rs` covers the native side for both build sides.
A window directly over the sorted cache, with no join in between, fails the same way on main. That needs the window itself to carry the order, so I filed it separately: #7223.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#issuecomment-6085216954
Thanks @dwsmith1983. I added the backport label. I won't review yet since there are existing review comments to be addressed.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.
dwsmith1983 commented on code in PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#discussion_r4322553189
##########
spark/src/main/scala/org/apache/comet/rules/CometShuffledHashJoinStrategy.scala:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.comet.rules
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.RowOrdering
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
+import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{Join, JoinHint, JoinStrategyHint, LogicalPlan}
+import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
+import org.apache.spark.sql.execution.adaptive.{BroadcastQueryStageExec, LogicalQueryStage}
+import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.CometConf
+import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, withFallbackReason, withInfo}
+import org.apache.comet.shims.ShimJoinSelection
+
+/**
+ * Plans an equi-join as a ShuffledHashJoinExec where Spark would plan a SortMergeJoinExec, when
+ * spark.comet.exec.forceShuffledHashJoin is on. Choosing the join here, before EnsureRequirements
+ * and RemoveRedundantSorts run, lets Spark place and keep every sort the plan needs above the
+ * join. Planner strategies also run again on every AQE re-plan, where the join's children carry
+ * the materialized shuffle sizes.
+ *
+ * Adapted from the equivalent rule in Apache Gluten.
+ */
+case class CometShuffledHashJoinStrategy(session: SparkSession)
+ extends SparkStrategy
+ with ShimJoinSelection {
+
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case join: Join if isEnabled(join) => planJoin(join)
+ case _ => Nil
+ }
+
+ private def sqlConf: SQLConf = session.sessionState.conf
+
+ // Planner strategies run whether or not Comet is enabled, and before CometRule, so plan-only
+ // mode needs its own check here.
+ private def isEnabled(join: Join): Boolean =
+ CometConf.COMET_FORCE_SHJ.get(sqlConf) &&
+ CometConf.COMET_EXEC_ENABLED.get(sqlConf) &&
+ !CometConf.COMET_EXPLAIN_PLAN_ONLY_ENABLED.get(sqlConf) &&
+ !join.isStreaming &&
+ isCometLoaded(sqlConf)
+
+ private def planJoin(join: Join): Seq[SparkPlan] = join match {
+ case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, _, left, right, hint)
+ if !isLeftToSpark(join, hint) && canSortMergeJoin(joinType) &&
+ RowOrdering.isOrderable(leftKeys) && hashJoinSupportedShim(leftKeys, rightKeys) =>
+ buildSide(join) match {
+ case Some(BuildRight) if joinType == LeftSemi || joinType.isInstanceOf[ExistenceJoin] =>
+ // LeftSemi https://github.com/apache/datafusion-comet/issues/2667
+ // ExistenceJoin https://github.com/apache/datafusion-comet/issues/2697
+ planBySpark(join) { smj =>
+ withFallbackReason(smj, declineReason(s"BuildRight with $joinType is not supported"))
+ }
+ case Some(side) =>
+ buildSideOverLimit(join, side) match {
+ case Some(reason) =>
+ // The sort-merge join still runs natively, so this is information for extended
+ // explain rather than a fallback reason.
+ planBySpark(join)(smj => withInfo(smj, declineReason(reason)))
+ case None =>
+ Seq(
+ ShuffledHashJoinExec(
Review Comment:
The fix for main is up in #7126.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.
dwsmith1983 commented on code in PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#discussion_r4233303896
##########
spark/src/main/scala/org/apache/comet/rules/CometShuffledHashJoinStrategy.scala:
##########
@@ -0,0 +1,194 @@
+/*
+ * 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.comet.rules
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.RowOrdering
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
+import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{BROADCAST, Join, JoinHint, LogicalPlan, SHUFFLE_HASH, SHUFFLE_MERGE, SHUFFLE_REPLICATE_NL}
+import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
+import org.apache.spark.sql.execution.adaptive.{BroadcastQueryStageExec, LogicalQueryStage}
+import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.CometConf
+import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, withFallbackReason, withInfo}
+import org.apache.comet.shims.ShimJoinSelection
+
+/**
+ * Plans an equi-join as a ShuffledHashJoinExec where Spark would plan a SortMergeJoinExec, when
+ * spark.comet.exec.forceShuffledHashJoin is on. Choosing the join here, before EnsureRequirements
+ * and RemoveRedundantSorts run, lets Spark place and keep every sort the plan needs above the
+ * join. Planner strategies also run again on every AQE re-plan, where the join's children carry
+ * the materialized shuffle sizes.
+ *
+ * Adapted from the equivalent rule in Apache Gluten.
+ */
+case class CometShuffledHashJoinStrategy(session: SparkSession)
+ extends SparkStrategy
+ with ShimJoinSelection {
+
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case join: Join if isEnabled(join) => planJoin(join)
+ case _ => Nil
+ }
+
+ private def sqlConf: SQLConf = session.sessionState.conf
+
+ // Planner strategies run whether or not Comet is enabled, and before CometRule, so plan-only
+ // mode needs its own check here.
+ private def isEnabled(join: Join): Boolean =
+ CometConf.COMET_FORCE_SHJ.get(sqlConf) &&
+ CometConf.COMET_EXEC_ENABLED.get(sqlConf) &&
+ !CometConf.COMET_EXPLAIN_PLAN_ONLY_ENABLED.get(sqlConf) &&
+ !join.isStreaming &&
+ isCometLoaded(sqlConf)
+
+ private def planJoin(join: Join): Seq[SparkPlan] = join match {
+ case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, _, left, right, hint)
+ if !isLeftToSpark(join, hint) && canSortMergeJoin(joinType) &&
+ RowOrdering.isOrderable(leftKeys) && hashJoinSupportedShim(leftKeys, rightKeys) =>
+ buildSide(join) match {
+ case Some(BuildRight) if joinType == LeftSemi || joinType.isInstanceOf[ExistenceJoin] =>
+ // LeftSemi https://github.com/apache/datafusion-comet/issues/2667
+ // ExistenceJoin https://github.com/apache/datafusion-comet/issues/2697
+ planBySpark(join) { smj =>
+ withFallbackReason(smj, declineReason(s"BuildRight with $joinType is not supported"))
+ }
+ case Some(side) =>
+ buildSideOverLimit(join, side) match {
+ case Some(reason) =>
+ // The sort-merge join still runs natively, so this is information for extended
+ // explain rather than a fallback reason.
+ planBySpark(join)(smj => withInfo(smj, declineReason(reason)))
+ case None =>
+ Seq(
+ ShuffledHashJoinExec(
+ leftKeys,
+ rightKeys,
+ joinType,
+ side,
+ condition,
+ planLater(left),
+ planLater(right)))
+ }
+ case None => Nil
+ }
+ case _ => Nil
+ }
+
+ // Broadcasts, including the ones AQE plans once a stage turns out small, and joins with a
+ // strategy hint are planned by Spark as usual.
+ private def isLeftToSpark(join: Join, hint: JoinHint): Boolean =
+ isBroadcastStage(join.left) || isBroadcastStage(join.right) ||
+ canPlanAsBroadcastHashJoin(join, sqlConf) || hasStrategyHint(hint)
+
+ private def isBroadcastStage(plan: LogicalPlan): Boolean = plan match {
+ case LogicalQueryStage(_, _: BroadcastQueryStageExec) => true
+ case _ => false
+ }
+
+ // AQE's own hints, such as NO_BROADCAST_HASH and PREFER_SHUFFLE_HASH, do not count.
+ private def hasStrategyHint(hint: JoinHint): Boolean =
+ Seq(hint.leftHint, hint.rightHint).flatten.flatMap(_.strategy).exists {
+ case BROADCAST | SHUFFLE_MERGE | SHUFFLE_HASH | SHUFFLE_REPLICATE_NL => true
+ case _ => false
Review Comment:
> Could `hasStrategyHint` use it
Done. It now checks `JoinStrategyHint.strategies`, so the four constant imports and the match are gone. I also added a test that a side AQE marks `NO_BROADCAST_HASH` keeps the hash join, since that hint is outside the set and nothing covered it.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.
dwsmith1983 commented on code in PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#discussion_r4233318734
##########
spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala:
##########
@@ -1219,6 +1251,237 @@ class CometJoinSuite extends CometTestBase {
}
}
+ // Spark removes a sort above a sort-merge join whose output ordering satisfies it, so the
+ // forced hash join must be chosen before that happens or the sort is lost (#6770).
+ private def withSortLossConf(adaptive: Boolean)(f: => Unit): Unit = withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptive.toString,
+ SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "2",
+ CometConf.COMET_FORCE_SHJ.key -> "true") {
+ withParquetTable((0 until 10000).map(i => (i % 100, i)), "big") {
+ withParquetTable((0 until 10).map(i => (i * 10, i)), "small") {
Review Comment:
> could the config block become one class-level helper that takes `adaptive` and serves the semi join, window and sort tests?
Done. The semi join, window and sort tests now share one helper that takes `adaptive`, sets the five configs plus `maxBuildSize=-1`, and builds the tables with `withChainedJoinTables(midRows = 3000, midKeys = 100)`.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.
dwsmith1983 commented on code in PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#discussion_r4233338965
##########
spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala:
##########
@@ -1219,6 +1251,237 @@ class CometJoinSuite extends CometTestBase {
}
}
+ // Spark removes a sort above a sort-merge join whose output ordering satisfies it, so the
+ // forced hash join must be chosen before that happens or the sort is lost (#6770).
+ private def withSortLossConf(adaptive: Boolean)(f: => Unit): Unit = withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptive.toString,
+ SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "2",
+ CometConf.COMET_FORCE_SHJ.key -> "true") {
+ withParquetTable((0 until 10000).map(i => (i % 100, i)), "big") {
+ withParquetTable((0 until 10).map(i => (i * 10, i)), "small") {
+ f
+ }
+ }
+ }
+
+ // Checks that a hash join ran, that the rows match Spark's, and that every partition is sorted
+ // on the first column. A local sort leaves rows with equal keys in any order, so the rows are
+ // compared without their order.
+ private def checkPartitionsSortedOverHashJoin(df: => DataFrame): Unit = {
+ // withSQLConf returns Unit on Spark 3.x, so the expected rows are assigned inside it.
+ var expected = Seq.empty[String]
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ expected = df.collect().map(_.toString).sorted.toSeq
+ }
+ val cometDf = df
+ assert(cometDf.collect().map(_.toString).sorted.toSeq == expected)
+ val cometPlan = cometDf.queryExecution.executedPlan
+ assert(collect(cometPlan) { case j: CometHashJoinExec => j }.nonEmpty, cometPlan)
+ val sorted = cometDf.queryExecution.toRdd
+ .mapPartitions { it =>
+ val keys = it.map(_.getInt(0)).toArray
+ Iterator(keys.sameElements(keys.sorted))
+ }
+ .collect()
+ assert(sorted.forall(identity), sorted.mkString(", "))
+ }
+
+ private val sortLossJoin =
+ "SELECT big._1 AS k, big._2 AS v FROM big JOIN small ON big._1 = small._1"
+
+ for (adaptive <- Seq(false, true)) {
+ test(s"forceShuffledHashJoin keeps sortWithinPartitions on the join key, AQE=$adaptive") {
+ withSortLossConf(adaptive) {
+ checkPartitionsSortedOverHashJoin(sql(sortLossJoin).sortWithinPartitions("k"))
+ }
+ }
+
+ test(s"forceShuffledHashJoin keeps SORT BY on the join key, AQE=$adaptive") {
+ withSortLossConf(adaptive) {
+ checkPartitionsSortedOverHashJoin(sql(s"$sortLossJoin SORT BY k"))
Review Comment:
> Would one of them be enough, given that the second adds a run per AQE setting without a different plan?
Yes. Both forms analyze to the same `Sort` with `global = false`, so I dropped the `SORT BY` test and kept the `sortWithinPartitions` one.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.
dwsmith1983 commented on code in PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#discussion_r4361193351
##########
spark/src/main/scala/org/apache/comet/rules/CometShuffledHashJoinStrategy.scala:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.comet.rules
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.RowOrdering
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
+import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{Join, JoinHint, JoinStrategyHint, LogicalPlan}
+import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
+import org.apache.spark.sql.execution.adaptive.{BroadcastQueryStageExec, LogicalQueryStage}
+import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.CometConf
+import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, withFallbackReason, withInfo}
+import org.apache.comet.shims.ShimJoinSelection
+
+/**
+ * Plans an equi-join as a ShuffledHashJoinExec where Spark would plan a SortMergeJoinExec, when
+ * spark.comet.exec.forceShuffledHashJoin is on. Choosing the join here, before EnsureRequirements
+ * and RemoveRedundantSorts run, lets Spark place and keep every sort the plan needs above the
+ * join. Planner strategies also run again on every AQE re-plan, where the join's children carry
+ * the materialized shuffle sizes.
+ *
+ * Adapted from the equivalent rule in Apache Gluten.
+ */
+case class CometShuffledHashJoinStrategy(session: SparkSession)
+ extends SparkStrategy
+ with ShimJoinSelection {
+
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case join: Join if isEnabled(join) => planJoin(join)
+ case _ => Nil
+ }
+
+ private def sqlConf: SQLConf = session.sessionState.conf
+
+ // Planner strategies run whether or not Comet is enabled, and before CometRule, so plan-only
+ // mode needs its own check here.
+ private def isEnabled(join: Join): Boolean =
+ CometConf.COMET_FORCE_SHJ.get(sqlConf) &&
+ CometConf.COMET_EXEC_ENABLED.get(sqlConf) &&
+ !CometConf.COMET_EXPLAIN_PLAN_ONLY_ENABLED.get(sqlConf) &&
+ !join.isStreaming &&
+ isCometLoaded(sqlConf)
+
+ private def planJoin(join: Join): Seq[SparkPlan] = join match {
+ case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, _, left, right, hint)
+ if !isLeftToSpark(join, hint) && canSortMergeJoin(joinType) &&
+ RowOrdering.isOrderable(leftKeys) && hashJoinSupportedShim(leftKeys, rightKeys) =>
+ buildSide(join) match {
+ case Some(BuildRight) if joinType == LeftSemi || joinType.isInstanceOf[ExistenceJoin] =>
+ // LeftSemi https://github.com/apache/datafusion-comet/issues/2667
+ // ExistenceJoin https://github.com/apache/datafusion-comet/issues/2697
+ planBySpark(join) { smj =>
+ withFallbackReason(smj, declineReason(s"BuildRight with $joinType is not supported"))
+ }
+ case Some(side) =>
+ buildSideOverLimit(join, side) match {
+ case Some(reason) =>
+ // The sort-merge join still runs natively, so this is information for extended
+ // explain rather than a fallback reason.
+ planBySpark(join)(smj => withInfo(smj, declineReason(reason)))
+ case None =>
+ Seq(
+ ShuffledHashJoinExec(
Review Comment:
Added in 09c28d2f4, after merging main with #7126: `forceShuffledHashJoin keeps the order of a sorted checkpoint through an outer join` runs your `localCheckpoint` probe for both outer shapes. On the previous head it fails on the order check, with the native hash join directly over `Scan ExistingRDD` and no sort above it. Now the join stays in Spark's `ShuffledHashJoinExec` with #7126's fallback reason, and the `(k, v)` order holds.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#discussion_r4231743130
##########
spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala:
##########
@@ -1219,6 +1251,237 @@ class CometJoinSuite extends CometTestBase {
}
}
+ // Spark removes a sort above a sort-merge join whose output ordering satisfies it, so the
+ // forced hash join must be chosen before that happens or the sort is lost (#6770).
+ private def withSortLossConf(adaptive: Boolean)(f: => Unit): Unit = withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptive.toString,
+ SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "2",
+ CometConf.COMET_FORCE_SHJ.key -> "true") {
+ withParquetTable((0 until 10000).map(i => (i % 100, i)), "big") {
+ withParquetTable((0 until 10).map(i => (i * 10, i)), "small") {
Review Comment:
This repeats setup the suite already has. `big` and `small` are the same data that `withChainedJoinTables` builds, and the five configs here are the ones `withKeptSemiJoinConf` and the window test already set (they also set `maxBuildSize=-1`). Could this helper call `withChainedJoinTables(midRows = 3000, midKeys = 100)` for the tables, and could the config block become one class-level helper that takes `adaptive` and serves the semi join, window and sort tests? I have not run it. The unused `mid` table and `maxBuildSize=-1` should not change these plans, since the build side is always `small`.
##########
spark/src/main/scala/org/apache/comet/rules/CometShuffledHashJoinStrategy.scala:
##########
@@ -0,0 +1,194 @@
+/*
+ * 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.comet.rules
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.RowOrdering
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
+import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{BROADCAST, Join, JoinHint, LogicalPlan, SHUFFLE_HASH, SHUFFLE_MERGE, SHUFFLE_REPLICATE_NL}
+import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
+import org.apache.spark.sql.execution.adaptive.{BroadcastQueryStageExec, LogicalQueryStage}
+import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.CometConf
+import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, withFallbackReason, withInfo}
+import org.apache.comet.shims.ShimJoinSelection
+
+/**
+ * Plans an equi-join as a ShuffledHashJoinExec where Spark would plan a SortMergeJoinExec, when
+ * spark.comet.exec.forceShuffledHashJoin is on. Choosing the join here, before EnsureRequirements
+ * and RemoveRedundantSorts run, lets Spark place and keep every sort the plan needs above the
+ * join. Planner strategies also run again on every AQE re-plan, where the join's children carry
+ * the materialized shuffle sizes.
+ *
+ * Adapted from the equivalent rule in Apache Gluten.
+ */
+case class CometShuffledHashJoinStrategy(session: SparkSession)
+ extends SparkStrategy
+ with ShimJoinSelection {
+
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case join: Join if isEnabled(join) => planJoin(join)
+ case _ => Nil
+ }
+
+ private def sqlConf: SQLConf = session.sessionState.conf
+
+ // Planner strategies run whether or not Comet is enabled, and before CometRule, so plan-only
+ // mode needs its own check here.
+ private def isEnabled(join: Join): Boolean =
+ CometConf.COMET_FORCE_SHJ.get(sqlConf) &&
+ CometConf.COMET_EXEC_ENABLED.get(sqlConf) &&
+ !CometConf.COMET_EXPLAIN_PLAN_ONLY_ENABLED.get(sqlConf) &&
+ !join.isStreaming &&
+ isCometLoaded(sqlConf)
+
+ private def planJoin(join: Join): Seq[SparkPlan] = join match {
+ case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, _, left, right, hint)
+ if !isLeftToSpark(join, hint) && canSortMergeJoin(joinType) &&
+ RowOrdering.isOrderable(leftKeys) && hashJoinSupportedShim(leftKeys, rightKeys) =>
+ buildSide(join) match {
+ case Some(BuildRight) if joinType == LeftSemi || joinType.isInstanceOf[ExistenceJoin] =>
+ // LeftSemi https://github.com/apache/datafusion-comet/issues/2667
+ // ExistenceJoin https://github.com/apache/datafusion-comet/issues/2697
+ planBySpark(join) { smj =>
+ withFallbackReason(smj, declineReason(s"BuildRight with $joinType is not supported"))
+ }
+ case Some(side) =>
+ buildSideOverLimit(join, side) match {
+ case Some(reason) =>
+ // The sort-merge join still runs natively, so this is information for extended
+ // explain rather than a fallback reason.
+ planBySpark(join)(smj => withInfo(smj, declineReason(reason)))
+ case None =>
+ Seq(
+ ShuffledHashJoinExec(
+ leftKeys,
+ rightKeys,
+ joinType,
+ side,
+ condition,
+ planLater(left),
+ planLater(right)))
+ }
+ case None => Nil
+ }
+ case _ => Nil
+ }
+
+ // Broadcasts, including the ones AQE plans once a stage turns out small, and joins with a
+ // strategy hint are planned by Spark as usual.
+ private def isLeftToSpark(join: Join, hint: JoinHint): Boolean =
+ isBroadcastStage(join.left) || isBroadcastStage(join.right) ||
+ canPlanAsBroadcastHashJoin(join, sqlConf) || hasStrategyHint(hint)
+
+ private def isBroadcastStage(plan: LogicalPlan): Boolean = plan match {
+ case LogicalQueryStage(_, _: BroadcastQueryStageExec) => true
+ case _ => false
+ }
+
+ // AQE's own hints, such as NO_BROADCAST_HASH and PREFER_SHUFFLE_HASH, do not count.
+ private def hasStrategyHint(hint: JoinHint): Boolean =
+ Seq(hint.leftHint, hint.rightHint).flatten.flatMap(_.strategy).exists {
+ case BROADCAST | SHUFFLE_MERGE | SHUFFLE_HASH | SHUFFLE_REPLICATE_NL => true
+ case _ => false
Review Comment:
Spark already keeps this exact list as `JoinStrategyHint.strategies` (`BROADCAST`, `SHUFFLE_MERGE`, `SHUFFLE_HASH`, `SHUFFLE_REPLICATE_NL`, without the AQE-only hints). It is public in `hints.scala` on 3.4.3, 3.5.8, 4.0.1 and 4.1.3, and `ResolveHints` uses it to decide which hint names a user can write. Could `hasStrategyHint` use it, for example `Seq(hint.leftHint, hint.rightHint).exists(_.exists(_.strategy.exists(JoinStrategyHint.strategies.contains)))`? That drops the four constant imports and the match, and walks the hints once instead of building two intermediate collections. If Spark adds another user-facing hint later, the join would also keep honoring it without another edit here. I have not compiled the suggested form.
##########
spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala:
##########
@@ -1219,6 +1251,237 @@ class CometJoinSuite extends CometTestBase {
}
}
+ // Spark removes a sort above a sort-merge join whose output ordering satisfies it, so the
+ // forced hash join must be chosen before that happens or the sort is lost (#6770).
+ private def withSortLossConf(adaptive: Boolean)(f: => Unit): Unit = withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptive.toString,
+ SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "2",
+ CometConf.COMET_FORCE_SHJ.key -> "true") {
+ withParquetTable((0 until 10000).map(i => (i % 100, i)), "big") {
+ withParquetTable((0 until 10).map(i => (i * 10, i)), "small") {
+ f
+ }
+ }
+ }
+
+ // Checks that a hash join ran, that the rows match Spark's, and that every partition is sorted
+ // on the first column. A local sort leaves rows with equal keys in any order, so the rows are
+ // compared without their order.
+ private def checkPartitionsSortedOverHashJoin(df: => DataFrame): Unit = {
+ // withSQLConf returns Unit on Spark 3.x, so the expected rows are assigned inside it.
+ var expected = Seq.empty[String]
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ expected = df.collect().map(_.toString).sorted.toSeq
+ }
+ val cometDf = df
+ assert(cometDf.collect().map(_.toString).sorted.toSeq == expected)
+ val cometPlan = cometDf.queryExecution.executedPlan
+ assert(collect(cometPlan) { case j: CometHashJoinExec => j }.nonEmpty, cometPlan)
+ val sorted = cometDf.queryExecution.toRdd
+ .mapPartitions { it =>
+ val keys = it.map(_.getInt(0)).toArray
+ Iterator(keys.sameElements(keys.sorted))
+ }
+ .collect()
+ assert(sorted.forall(identity), sorted.mkString(", "))
+ }
+
+ private val sortLossJoin =
+ "SELECT big._1 AS k, big._2 AS v FROM big JOIN small ON big._1 = small._1"
+
+ for (adaptive <- Seq(false, true)) {
+ test(s"forceShuffledHashJoin keeps sortWithinPartitions on the join key, AQE=$adaptive") {
+ withSortLossConf(adaptive) {
+ checkPartitionsSortedOverHashJoin(sql(sortLossJoin).sortWithinPartitions("k"))
+ }
+ }
+
+ test(s"forceShuffledHashJoin keeps SORT BY on the join key, AQE=$adaptive") {
+ withSortLossConf(adaptive) {
+ checkPartitionsSortedOverHashJoin(sql(s"$sortLossJoin SORT BY k"))
Review Comment:
`sql(sortLossJoin).sortWithinPartitions("k")` and `sql(s"$sortLossJoin SORT BY k")` should analyze to the same `Sort` with `global = false` over the same projection and join. In Spark 3.5.8 `AstBuilder` builds `SORT BY` as `Sort(..., global = false, query)` and `Dataset.sortInternal` builds the same node. From reading the code I expect identical plans and identical assertions here, but I have not compared the plans. The issue lists both forms, so I see why both are here. Would one of them be enough, given that the second adds a run per AQE setting without a different plan?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.
dwsmith1983 commented on code in PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#discussion_r4393052532
##########
spark/src/main/scala/org/apache/comet/rules/CometShuffledHashJoinStrategy.scala:
##########
@@ -0,0 +1,175 @@
+/*
+ * 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.comet.rules
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.RowOrdering
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
+import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{Join, JoinHint, JoinStrategyHint, LogicalPlan}
+import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
+import org.apache.spark.sql.execution.adaptive.{BroadcastQueryStageExec, LogicalQueryStage}
+import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.CometConf
+import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, withFallbackReason, withInfo}
+import org.apache.comet.shims.ShimJoinSelection
+
+/**
+ * Plans a ShuffledHashJoinExec in place of a SortMergeJoinExec, before EnsureRequirements and
+ * RemoveRedundantSorts, so the sorts above the join survive. Adapted from Apache Gluten.
+ */
+case class CometShuffledHashJoinStrategy(session: SparkSession)
+ extends SparkStrategy
+ with ShimJoinSelection {
+
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case join: Join if isEnabled(join) => planJoin(join)
+ case _ => Nil
+ }
+
+ private def sqlConf: SQLConf = session.sessionState.conf
+
+ // Planner strategies run with Comet off and before CometRule, which applies plan-only mode.
+ private def isEnabled(join: Join): Boolean =
+ CometConf.COMET_FORCE_SHJ.get(sqlConf) &&
+ CometConf.COMET_EXEC_ENABLED.get(sqlConf) &&
+ !CometConf.COMET_EXPLAIN_PLAN_ONLY_ENABLED.get(sqlConf) &&
+ !join.isStreaming &&
+ isCometLoaded(sqlConf)
+
+ private def planJoin(join: Join): Seq[SparkPlan] = join match {
+ case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, _, left, right, hint)
+ if !isLeftToSpark(join, hint) && canSortMergeJoin(joinType) &&
+ RowOrdering.isOrderable(leftKeys) && hashJoinSupportedShim(leftKeys, rightKeys) =>
+ buildSide(join) match {
+ case Some(BuildRight) if joinType == LeftSemi || joinType.isInstanceOf[ExistenceJoin] =>
+ // LeftSemi https://github.com/apache/datafusion-comet/issues/2667
+ // ExistenceJoin https://github.com/apache/datafusion-comet/issues/2697
+ planBySpark(join) { smj =>
+ withFallbackReason(smj, declineReason(s"BuildRight with $joinType is not supported"))
+ }
+ case Some(side) =>
+ buildSideOverLimit(join, side) match {
+ case Some(reason) =>
+ // Not a fallback: the kept sort-merge join still runs natively.
+ planBySpark(join)(smj => withInfo(smj, declineReason(reason)))
+ case None =>
+ Seq(
+ ShuffledHashJoinExec(
+ leftKeys,
+ rightKeys,
+ joinType,
+ side,
+ condition,
+ planLater(left),
Review Comment:
> Preserve the window sort for checkpoint probes; forced inner join followed by dense_rank still fails with missing PARTITION BY ordering.
Reproduced, on 3.5 and 4.1. The strategy plans the hash join before Spark places sorts, and the join reports the checkpoint's `(k, v)` order, so Spark drops the window's sort. Comet can't verify a checkpoint's order, so the join declares none and the native window refused its input. On main the sort survived, since the forced join replaced a sort-merge join whose order covers only `k`.
c20eb20f8 restores that sort on the native side: the window now sorts its input by the partition keys and then the order keys whenever its input doesn't show every partition key as ordered. That is the same condition the native window rejects, so plans that work today are unchanged. Restoring Spark's sort instead would mean changing the order the hash join reports to Spark, which the outer join order tests depend on.
`a native window above an inner hash join over a checkpointed streamed side runs` covers the forced, `shuffle_hash` and `broadcast` joins with `dense_rank` and `sum(v) OVER (PARTITION BY k)`, and `a native window over a sorted checkpoint runs without Spark's sort` covers a window directly over the checkpoint; both failed before. `window_input_order.rs` covers when the sort is added on the native side.
This complements #7256, which declares the window's input order when Comet can trust it. With both in, a trusted input gets the declaration and no sort, and only an order Comet can't check is sorted.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "dwsmith1983 (via GitHub) via github" <gi...@datafusion.apache.org>.
dwsmith1983 commented on PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#issuecomment-6707366441
Moved back to draft until #7555 merges. It fixes the same lost order for windows that stay in Spark, and this PR's Spark fallback for windows relies on it.
The latest push makes a native window over an order Comet can't verify restore the whole order its input reports, not only its own keys. Two stacked windows over a sorted checkpoint, the upper one ordered by more keys, numbered rows wrongly before.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.
dwsmith1983 commented on PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#issuecomment-6076350015
@andygrove I'd suggest this for `backport-1.1`. `branch-1.1` has the same `RewriteJoin`, so with `spark.comet.exec.forceShuffledHashJoin` on, 1.1.0 drops the same sorts: `sortWithinPartitions` and `SORT BY` on the join key return unsorted partitions, and a partitioned insert keyed on the join key fails with `FileAlreadyExistsException`. The branch has no #6384 size limit, so the backport would bring the strategy over without the size check rather than cherry-pick this as is.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.
dwsmith1983 commented on code in PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#discussion_r4322322761
##########
spark/src/main/scala/org/apache/comet/rules/CometShuffledHashJoinStrategy.scala:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.comet.rules
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.RowOrdering
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
+import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{Join, JoinHint, JoinStrategyHint, LogicalPlan}
+import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
+import org.apache.spark.sql.execution.adaptive.{BroadcastQueryStageExec, LogicalQueryStage}
+import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.CometConf
+import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, withFallbackReason, withInfo}
+import org.apache.comet.shims.ShimJoinSelection
+
+/**
+ * Plans an equi-join as a ShuffledHashJoinExec where Spark would plan a SortMergeJoinExec, when
+ * spark.comet.exec.forceShuffledHashJoin is on. Choosing the join here, before EnsureRequirements
+ * and RemoveRedundantSorts run, lets Spark place and keep every sort the plan needs above the
+ * join. Planner strategies also run again on every AQE re-plan, where the join's children carry
+ * the materialized shuffle sizes.
+ *
+ * Adapted from the equivalent rule in Apache Gluten.
+ */
+case class CometShuffledHashJoinStrategy(session: SparkSession)
+ extends SparkStrategy
+ with ShimJoinSelection {
+
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case join: Join if isEnabled(join) => planJoin(join)
+ case _ => Nil
+ }
+
+ private def sqlConf: SQLConf = session.sessionState.conf
+
+ // Planner strategies run whether or not Comet is enabled, and before CometRule, so plan-only
+ // mode needs its own check here.
+ private def isEnabled(join: Join): Boolean =
+ CometConf.COMET_FORCE_SHJ.get(sqlConf) &&
+ CometConf.COMET_EXEC_ENABLED.get(sqlConf) &&
+ !CometConf.COMET_EXPLAIN_PLAN_ONLY_ENABLED.get(sqlConf) &&
+ !join.isStreaming &&
+ isCometLoaded(sqlConf)
+
+ private def planJoin(join: Join): Seq[SparkPlan] = join match {
+ case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, _, left, right, hint)
+ if !isLeftToSpark(join, hint) && canSortMergeJoin(joinType) &&
+ RowOrdering.isOrderable(leftKeys) && hashJoinSupportedShim(leftKeys, rightKeys) =>
+ buildSide(join) match {
+ case Some(BuildRight) if joinType == LeftSemi || joinType.isInstanceOf[ExistenceJoin] =>
+ // LeftSemi https://github.com/apache/datafusion-comet/issues/2667
+ // ExistenceJoin https://github.com/apache/datafusion-comet/issues/2697
+ planBySpark(join) { smj =>
+ withFallbackReason(smj, declineReason(s"BuildRight with $joinType is not supported"))
+ }
+ case Some(side) =>
+ buildSideOverLimit(join, side) match {
+ case Some(reason) =>
+ // The sort-merge join still runs natively, so this is information for extended
+ // explain rather than a fallback reason.
+ planBySpark(join)(smj => withInfo(smj, declineReason(reason)))
+ case None =>
+ Seq(
+ ShuffledHashJoinExec(
Review Comment:
Confirmed. It isn't specific to the forced join: on main, a `SHUFFLE_HASH` or `BROADCAST` hint over the same `localCheckpoint` probe loses the order the same way, for both outer shapes and both hash join kinds. It's the second shape of #6922.
The fix goes to main first. When the streamed order can't be checked and reaches the join through an input from the JVM (a checkpoint or a cache) or through a nested loop join, the outer hash join falls back to Spark. Joins over native operators in the same plan stay native, since DataFusion sees the order there. Once that lands I'll merge it here and add the forced checkpoint case.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#discussion_r4268916352
##########
spark/src/main/scala/org/apache/comet/rules/CometShuffledHashJoinStrategy.scala:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.comet.rules
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.RowOrdering
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
+import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{Join, JoinHint, JoinStrategyHint, LogicalPlan}
+import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
+import org.apache.spark.sql.execution.adaptive.{BroadcastQueryStageExec, LogicalQueryStage}
+import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.CometConf
+import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, withFallbackReason, withInfo}
+import org.apache.comet.shims.ShimJoinSelection
+
+/**
+ * Plans an equi-join as a ShuffledHashJoinExec where Spark would plan a SortMergeJoinExec, when
+ * spark.comet.exec.forceShuffledHashJoin is on. Choosing the join here, before EnsureRequirements
+ * and RemoveRedundantSorts run, lets Spark place and keep every sort the plan needs above the
+ * join. Planner strategies also run again on every AQE re-plan, where the join's children carry
+ * the materialized shuffle sizes.
+ *
+ * Adapted from the equivalent rule in Apache Gluten.
+ */
+case class CometShuffledHashJoinStrategy(session: SparkSession)
+ extends SparkStrategy
+ with ShimJoinSelection {
+
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case join: Join if isEnabled(join) => planJoin(join)
+ case _ => Nil
+ }
+
+ private def sqlConf: SQLConf = session.sessionState.conf
+
+ // Planner strategies run whether or not Comet is enabled, and before CometRule, so plan-only
+ // mode needs its own check here.
+ private def isEnabled(join: Join): Boolean =
+ CometConf.COMET_FORCE_SHJ.get(sqlConf) &&
+ CometConf.COMET_EXEC_ENABLED.get(sqlConf) &&
+ !CometConf.COMET_EXPLAIN_PLAN_ONLY_ENABLED.get(sqlConf) &&
+ !join.isStreaming &&
+ isCometLoaded(sqlConf)
+
+ private def planJoin(join: Join): Seq[SparkPlan] = join match {
+ case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, _, left, right, hint)
+ if !isLeftToSpark(join, hint) && canSortMergeJoin(joinType) &&
+ RowOrdering.isOrderable(leftKeys) && hashJoinSupportedShim(leftKeys, rightKeys) =>
+ buildSide(join) match {
+ case Some(BuildRight) if joinType == LeftSemi || joinType.isInstanceOf[ExistenceJoin] =>
+ // LeftSemi https://github.com/apache/datafusion-comet/issues/2667
+ // ExistenceJoin https://github.com/apache/datafusion-comet/issues/2697
+ planBySpark(join) { smj =>
+ withFallbackReason(smj, declineReason(s"BuildRight with $joinType is not supported"))
+ }
+ case Some(side) =>
+ buildSideOverLimit(join, side) match {
+ case Some(reason) =>
+ // The sort-merge join still runs natively, so this is information for extended
+ // explain rather than a fallback reason.
+ planBySpark(join)(smj => withInfo(smj, declineReason(reason)))
+ case None =>
+ Seq(
+ ShuffledHashJoinExec(
Review Comment:
[P1] Preserve cached probe ordering before exposing the hash join’s ordering to Spark. With forcing enabled and AQE/broadcasts disabled, materialize a probe using `.repartition(2, col("k")).sortWithinPartitions("k", "v").cache()`, left-join a smaller table containing only some probe keys, then request `.sortWithinPartitions("k", "v")`. Selecting `ShuffledHashJoinExec` here lets Spark remove the final sort because it advertises the probe’s entire ordering. Comet reads the cache through native `ScanExec`, which carries no ordering metadata. The build-right left outer join becomes DataFusion’s `Right` join, whose unordered-input path appends unmatched probe rows after matched rows. The result therefore violates the requested ordering. This is introduced by the earlier selection: the exact base retains the final `(k, v)` sort because the original sort-merge join advertises only `k`. Please preserve ordering through the native input/outer join, or retain the sort/fallback until t
hat guarantee is available, and add a cached-probe regression test.
Evidence: Spark 4.1.3 differential harness: probe `range(2000).selectExpr("id % 100 as k", "id as v").repartition(2, col("k")).sortWithinPartitions("k", "v").cache()` followed by materialization; build `range(10).selectExpr("id * 10 as k")`; left outer join followed by `sortWithinPartitions("k", "v")`. Applying the exact-base rewrite leaves one final `Sort [k, v]`; the exact-head strategy leaves none. Separately, native DataFusion HashJoinExec with build keys `[1,3]`, ordered probe keys `[0,1,2,3,4]`, and no ordering metadata returned `[1,3,0,2,4]`. Its reproduced 55.1.0 `hash_join/exec.rs`, `hash_join/stream.rs`, and `joins/utils.rs` are byte-identical to pinned 55.2.0. Current Comet `ScanExec::new` initializes `EquivalenceProperties` without ordering. Reproduction files and logs are in `/tmp/review6785-fdde-probe-v6vo_b5b/`. Full Comet native execution was not available.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.
dwsmith1983 commented on code in PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#discussion_r4371614203
##########
spark/src/main/scala/org/apache/comet/rules/CometShuffledHashJoinStrategy.scala:
##########
@@ -0,0 +1,175 @@
+/*
+ * 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.comet.rules
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.RowOrdering
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
+import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{Join, JoinHint, JoinStrategyHint, LogicalPlan}
+import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
+import org.apache.spark.sql.execution.adaptive.{BroadcastQueryStageExec, LogicalQueryStage}
+import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.CometConf
+import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, withFallbackReason, withInfo}
+import org.apache.comet.shims.ShimJoinSelection
+
+/**
+ * Plans a ShuffledHashJoinExec in place of a SortMergeJoinExec, before EnsureRequirements and
+ * RemoveRedundantSorts, so the sorts above the join survive. Adapted from Apache Gluten.
+ */
+case class CometShuffledHashJoinStrategy(session: SparkSession)
+ extends SparkStrategy
+ with ShimJoinSelection {
+
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case join: Join if isEnabled(join) => planJoin(join)
+ case _ => Nil
+ }
+
+ private def sqlConf: SQLConf = session.sessionState.conf
+
+ // Planner strategies run with Comet off and before CometRule, which applies plan-only mode.
+ private def isEnabled(join: Join): Boolean =
+ CometConf.COMET_FORCE_SHJ.get(sqlConf) &&
+ CometConf.COMET_EXEC_ENABLED.get(sqlConf) &&
+ !CometConf.COMET_EXPLAIN_PLAN_ONLY_ENABLED.get(sqlConf) &&
+ !join.isStreaming &&
+ isCometLoaded(sqlConf)
+
+ private def planJoin(join: Join): Seq[SparkPlan] = join match {
+ case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, _, left, right, hint)
+ if !isLeftToSpark(join, hint) && canSortMergeJoin(joinType) &&
+ RowOrdering.isOrderable(leftKeys) && hashJoinSupportedShim(leftKeys, rightKeys) =>
+ buildSide(join) match {
+ case Some(BuildRight) if joinType == LeftSemi || joinType.isInstanceOf[ExistenceJoin] =>
+ // LeftSemi https://github.com/apache/datafusion-comet/issues/2667
+ // ExistenceJoin https://github.com/apache/datafusion-comet/issues/2697
+ planBySpark(join) { smj =>
+ withFallbackReason(smj, declineReason(s"BuildRight with $joinType is not supported"))
+ }
+ case Some(side) =>
+ buildSideOverLimit(join, side) match {
+ case Some(reason) =>
+ // Not a fallback: the kept sort-merge join still runs natively.
+ planBySpark(join)(smj => withInfo(smj, declineReason(reason)))
+ case None =>
+ Seq(
+ ShuffledHashJoinExec(
+ leftKeys,
+ rightKeys,
+ joinType,
+ side,
+ condition,
+ planLater(left),
+ planLater(right)))
+ }
+ case None => Nil
+ }
+ case _ => Nil
+ }
+
+ private def isLeftToSpark(join: Join, hint: JoinHint): Boolean =
+ isBroadcastStage(join.left) || isBroadcastStage(join.right) ||
+ canPlanAsBroadcastHashJoin(join, sqlConf) || hasStrategyHint(hint)
+
+ // canPlanAsBroadcastHashJoin misses a broadcast stage when the adaptive threshold is lower or
+ // off, and AQE discards a re-plan that hash joins one.
+ private def isBroadcastStage(plan: LogicalPlan): Boolean = plan match {
+ case LogicalQueryStage(_, _: BroadcastQueryStageExec) => true
+ case _ => false
+ }
+
+ // AQE's NO_BROADCAST_HASH and PREFER_SHUFFLE_HASH are outside JoinStrategyHint.strategies.
+ private def hasStrategyHint(hint: JoinHint): Boolean =
+ Seq(hint.leftHint, hint.rightHint).exists(
+ _.exists(_.strategy.exists(JoinStrategyHint.strategies.contains)))
Review Comment:
The strategy now builds on the side AQE hints with `PREFER_SHUFFLE_HASH` when only one buildable side carries it, much as Spark's `getShuffleHashJoinBuildSide` does, and keeps the size comparison when both sides or neither do. A lone `NO_BROADCAST_HASH` is handled as before.
`forceShuffledHashJoin builds on the side AQE's PREFER_SHUFFLE_HASH hint names` runs your AQE shape and checks that AQE put the hint on the right side and that the join builds right. `forceShuffledHashJoin builds on the one buildable side PREFER_SHUFFLE_HASH names` calls the strategy with the hint on one side, on both, and on the left side of a LEFT ANTI join, which cannot build there.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.
dwsmith1983 commented on PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#issuecomment-6282703286
> Could the new test also be shown to fail on this branch without the #6835 declaration, so we know it covers the forced join?
Yes. Main, with #6835, is merged in 153a343f4, which also adds `forceShuffledHashJoin keeps the order of a sorted cache through an outer join` to `CometJoinSuite`. It caches a table sorted within its partitions, runs a left outer join built on the right and a right outer join built on the left, once forced with no hint and once with `SHUFFLE_HASH`, and then asks for `sortWithinPartitions(k, v)`. Each case checks that the plan is a Comet hash join on the expected side with no sort above it, that the join declares the streamed order, that the rows match Spark, and that every partition stays in `(k, v)` order.
With only `.addAllStreamedSortOrders(streamedSortOrders.asJava)` removed from `CometHashJoin.doConvert`, the forced left outer case fails. Before the test asserted the declaration, it failed on the order check (`left_outer with hint None is out of order`) over `CometHashJoin LeftOuter, BuildRight` with no sort above it. Now it stops at the declaration check first.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.
dwsmith1983 commented on code in PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#discussion_r4278036358
##########
spark/src/main/scala/org/apache/comet/rules/CometShuffledHashJoinStrategy.scala:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.comet.rules
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.RowOrdering
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
+import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{Join, JoinHint, JoinStrategyHint, LogicalPlan}
+import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
+import org.apache.spark.sql.execution.adaptive.{BroadcastQueryStageExec, LogicalQueryStage}
+import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.CometConf
+import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, withFallbackReason, withInfo}
+import org.apache.comet.shims.ShimJoinSelection
+
+/**
+ * Plans an equi-join as a ShuffledHashJoinExec where Spark would plan a SortMergeJoinExec, when
+ * spark.comet.exec.forceShuffledHashJoin is on. Choosing the join here, before EnsureRequirements
+ * and RemoveRedundantSorts run, lets Spark place and keep every sort the plan needs above the
+ * join. Planner strategies also run again on every AQE re-plan, where the join's children carry
+ * the materialized shuffle sizes.
+ *
+ * Adapted from the equivalent rule in Apache Gluten.
+ */
+case class CometShuffledHashJoinStrategy(session: SparkSession)
+ extends SparkStrategy
+ with ShimJoinSelection {
+
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case join: Join if isEnabled(join) => planJoin(join)
+ case _ => Nil
+ }
+
+ private def sqlConf: SQLConf = session.sessionState.conf
+
+ // Planner strategies run whether or not Comet is enabled, and before CometRule, so plan-only
+ // mode needs its own check here.
+ private def isEnabled(join: Join): Boolean =
+ CometConf.COMET_FORCE_SHJ.get(sqlConf) &&
+ CometConf.COMET_EXEC_ENABLED.get(sqlConf) &&
+ !CometConf.COMET_EXPLAIN_PLAN_ONLY_ENABLED.get(sqlConf) &&
+ !join.isStreaming &&
+ isCometLoaded(sqlConf)
+
+ private def planJoin(join: Join): Seq[SparkPlan] = join match {
+ case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, _, left, right, hint)
+ if !isLeftToSpark(join, hint) && canSortMergeJoin(joinType) &&
+ RowOrdering.isOrderable(leftKeys) && hashJoinSupportedShim(leftKeys, rightKeys) =>
+ buildSide(join) match {
+ case Some(BuildRight) if joinType == LeftSemi || joinType.isInstanceOf[ExistenceJoin] =>
+ // LeftSemi https://github.com/apache/datafusion-comet/issues/2667
+ // ExistenceJoin https://github.com/apache/datafusion-comet/issues/2697
+ planBySpark(join) { smj =>
+ withFallbackReason(smj, declineReason(s"BuildRight with $joinType is not supported"))
+ }
+ case Some(side) =>
+ buildSideOverLimit(join, side) match {
+ case Some(reason) =>
+ // The sort-merge join still runs natively, so this is information for extended
+ // explain rather than a fallback reason.
+ planBySpark(join)(smj => withInfo(smj, declineReason(reason)))
+ case None =>
+ Seq(
+ ShuffledHashJoinExec(
Review Comment:
> Please preserve ordering through the native input/outer join, or retain the sort/fallback until that guarantee is available, and add a cached-probe regression test.
Confirmed. With `forceShuffledHashJoin` the planner now picks `ShuffledHashJoinExec`, which reports the cached probe's `(k, v)` ordering, so Spark drops the final sort. Comet reads the cache through a scan with no ordering, and DataFusion's `Right` join puts the unmatched probe rows after the matched ones in each batch. I reproduced it with a cached `repartition(2, k).sortWithinPartitions(k, v)` probe left-joined to a 5-key table. Every partition comes back out of order, with the right rows.
The same native gap is already reachable on main without this conf, through a `SHUFFLE_HASH` hint or a broadcast hash join. Both fail that test on main. #6835 fixes it at the source: it declares the streamed order on the probe for these two outer shapes and follows the cache scan into the cached plan. With its non-test changes on top of this branch, the forced, hinted and broadcast cases all pass.
So I'd like #6835 to land first. I'll then merge it into this branch and add the cached-probe test for the forced and hinted joins. cc @andygrove, since this puts #6785 after #6835.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#discussion_r4368087541
##########
spark/src/main/scala/org/apache/comet/rules/CometShuffledHashJoinStrategy.scala:
##########
@@ -0,0 +1,175 @@
+/*
+ * 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.comet.rules
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.RowOrdering
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
+import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{Join, JoinHint, JoinStrategyHint, LogicalPlan}
+import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
+import org.apache.spark.sql.execution.adaptive.{BroadcastQueryStageExec, LogicalQueryStage}
+import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.CometConf
+import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, withFallbackReason, withInfo}
+import org.apache.comet.shims.ShimJoinSelection
+
+/**
+ * Plans a ShuffledHashJoinExec in place of a SortMergeJoinExec, before EnsureRequirements and
+ * RemoveRedundantSorts, so the sorts above the join survive. Adapted from Apache Gluten.
+ */
+case class CometShuffledHashJoinStrategy(session: SparkSession)
+ extends SparkStrategy
+ with ShimJoinSelection {
+
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case join: Join if isEnabled(join) => planJoin(join)
+ case _ => Nil
+ }
+
+ private def sqlConf: SQLConf = session.sessionState.conf
+
+ // Planner strategies run with Comet off and before CometRule, which applies plan-only mode.
+ private def isEnabled(join: Join): Boolean =
+ CometConf.COMET_FORCE_SHJ.get(sqlConf) &&
+ CometConf.COMET_EXEC_ENABLED.get(sqlConf) &&
+ !CometConf.COMET_EXPLAIN_PLAN_ONLY_ENABLED.get(sqlConf) &&
+ !join.isStreaming &&
+ isCometLoaded(sqlConf)
+
+ private def planJoin(join: Join): Seq[SparkPlan] = join match {
+ case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, _, left, right, hint)
+ if !isLeftToSpark(join, hint) && canSortMergeJoin(joinType) &&
+ RowOrdering.isOrderable(leftKeys) && hashJoinSupportedShim(leftKeys, rightKeys) =>
+ buildSide(join) match {
+ case Some(BuildRight) if joinType == LeftSemi || joinType.isInstanceOf[ExistenceJoin] =>
+ // LeftSemi https://github.com/apache/datafusion-comet/issues/2667
+ // ExistenceJoin https://github.com/apache/datafusion-comet/issues/2697
+ planBySpark(join) { smj =>
+ withFallbackReason(smj, declineReason(s"BuildRight with $joinType is not supported"))
+ }
+ case Some(side) =>
+ buildSideOverLimit(join, side) match {
+ case Some(reason) =>
+ // Not a fallback: the kept sort-merge join still runs natively.
+ planBySpark(join)(smj => withInfo(smj, declineReason(reason)))
+ case None =>
+ Seq(
+ ShuffledHashJoinExec(
Review Comment:
[P1] Preserve the requested sort when a multirow native broadcast nested-loop join feeds this forced inner hash join. Spark’s `ShuffledHashJoinExec` advertises the probe’s full ordering, so Spark removes `ORDER BY x`. The native nested-loop producer emits matches grouped by broadcast build row, and `x` decreases in the final result. The base retained the sort because its initially planned sort-merge join advertised ordering on `x % 4`. Keep the nested-loop producer on Spark when its streamed order is required, or retain a real sort above it.
Evidence: On Spark 4.1.3, enable native range and `forceShuffledHashJoin`, disable AQE and automatic broadcasts, and use one shuffle partition. Run `SELECT j.x, j.z, b.id AS w FROM (SELECT /*+ BROADCAST(n) */ p.id AS x, n.id AS z FROM range(0,20,1,1) p JOIN range(0,2,1,1) n ON p.id % 2 <= n.id) j JOIN range(0,2,1,1) b ON j.x % 4 = b.id * 2 ORDER BY x`. Spark and the base returned `[0,0,2,2,…,18,18]`. The head returned `[0,2,…,18,0,2,…,18]`. Disabling only `spark.comet.exec.broadcastNestedLoopJoin.enabled` on the head restored the expected order. `SORT BY x` reproduced the same failure.
##########
spark/src/main/scala/org/apache/comet/rules/CometShuffledHashJoinStrategy.scala:
##########
@@ -0,0 +1,175 @@
+/*
+ * 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.comet.rules
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.RowOrdering
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
+import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{Join, JoinHint, JoinStrategyHint, LogicalPlan}
+import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
+import org.apache.spark.sql.execution.adaptive.{BroadcastQueryStageExec, LogicalQueryStage}
+import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.CometConf
+import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, withFallbackReason, withInfo}
+import org.apache.comet.shims.ShimJoinSelection
+
+/**
+ * Plans a ShuffledHashJoinExec in place of a SortMergeJoinExec, before EnsureRequirements and
+ * RemoveRedundantSorts, so the sorts above the join survive. Adapted from Apache Gluten.
+ */
+case class CometShuffledHashJoinStrategy(session: SparkSession)
+ extends SparkStrategy
+ with ShimJoinSelection {
+
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case join: Join if isEnabled(join) => planJoin(join)
+ case _ => Nil
+ }
+
+ private def sqlConf: SQLConf = session.sessionState.conf
+
+ // Planner strategies run with Comet off and before CometRule, which applies plan-only mode.
+ private def isEnabled(join: Join): Boolean =
+ CometConf.COMET_FORCE_SHJ.get(sqlConf) &&
+ CometConf.COMET_EXEC_ENABLED.get(sqlConf) &&
+ !CometConf.COMET_EXPLAIN_PLAN_ONLY_ENABLED.get(sqlConf) &&
+ !join.isStreaming &&
+ isCometLoaded(sqlConf)
+
+ private def planJoin(join: Join): Seq[SparkPlan] = join match {
+ case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, _, left, right, hint)
+ if !isLeftToSpark(join, hint) && canSortMergeJoin(joinType) &&
+ RowOrdering.isOrderable(leftKeys) && hashJoinSupportedShim(leftKeys, rightKeys) =>
+ buildSide(join) match {
+ case Some(BuildRight) if joinType == LeftSemi || joinType.isInstanceOf[ExistenceJoin] =>
+ // LeftSemi https://github.com/apache/datafusion-comet/issues/2667
+ // ExistenceJoin https://github.com/apache/datafusion-comet/issues/2697
+ planBySpark(join) { smj =>
+ withFallbackReason(smj, declineReason(s"BuildRight with $joinType is not supported"))
+ }
+ case Some(side) =>
+ buildSideOverLimit(join, side) match {
+ case Some(reason) =>
+ // Not a fallback: the kept sort-merge join still runs natively.
+ planBySpark(join)(smj => withInfo(smj, declineReason(reason)))
+ case None =>
+ Seq(
+ ShuffledHashJoinExec(
+ leftKeys,
+ rightKeys,
+ joinType,
+ side,
+ condition,
+ planLater(left),
Review Comment:
[P1] A sorted cached left child can make a native window abort above a forced build-right inner join. Spark sees the cache’s `(k, v)` ordering through the hash join and omits the window’s sort. The rows are sorted, but the native cache scan carries no ordering metadata, and `CometHashJoin` sends `streamedSortOrders` only for the relevant outer joins. Propagate verified ordering for this inner probe, retain the sort, or run the window in Spark when its native input cannot establish the required order.
Evidence: On Spark 4.1.3, disable AQE and broadcasts, enable forcing, native range and `spark.comet.convert.inMemoryCache.enabled`, and disable `spark.comet.exec.inMemoryCache.enabled`. Materialize `spark.range(2000).selectExpr("id % 20 AS k", "id AS v").repartition(2, col("k")).sortWithinPartitions("k", "v").cache()`. Inner-join to `range(20)` on `k = id` and collect `dense_rank() OVER (PARTITION BY k ORDER BY v)`. The base retained a `CometSort` below `CometWindowExec` and returned the same 2,000 ranked rows as Spark. The head omitted the sort and raised `CometNativeException: Expects PARTITION BY expression to be ordered`. Disabling only the native window on the head restored all 2,000 correct ranks.
##########
spark/src/main/scala/org/apache/comet/rules/CometShuffledHashJoinStrategy.scala:
##########
@@ -0,0 +1,175 @@
+/*
+ * 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.comet.rules
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.RowOrdering
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
+import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{Join, JoinHint, JoinStrategyHint, LogicalPlan}
+import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
+import org.apache.spark.sql.execution.adaptive.{BroadcastQueryStageExec, LogicalQueryStage}
+import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.CometConf
+import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, withFallbackReason, withInfo}
+import org.apache.comet.shims.ShimJoinSelection
+
+/**
+ * Plans a ShuffledHashJoinExec in place of a SortMergeJoinExec, before EnsureRequirements and
+ * RemoveRedundantSorts, so the sorts above the join survive. Adapted from Apache Gluten.
+ */
+case class CometShuffledHashJoinStrategy(session: SparkSession)
+ extends SparkStrategy
+ with ShimJoinSelection {
+
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case join: Join if isEnabled(join) => planJoin(join)
+ case _ => Nil
+ }
+
+ private def sqlConf: SQLConf = session.sessionState.conf
+
+ // Planner strategies run with Comet off and before CometRule, which applies plan-only mode.
+ private def isEnabled(join: Join): Boolean =
+ CometConf.COMET_FORCE_SHJ.get(sqlConf) &&
+ CometConf.COMET_EXEC_ENABLED.get(sqlConf) &&
+ !CometConf.COMET_EXPLAIN_PLAN_ONLY_ENABLED.get(sqlConf) &&
+ !join.isStreaming &&
+ isCometLoaded(sqlConf)
+
+ private def planJoin(join: Join): Seq[SparkPlan] = join match {
+ case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, _, left, right, hint)
+ if !isLeftToSpark(join, hint) && canSortMergeJoin(joinType) &&
+ RowOrdering.isOrderable(leftKeys) && hashJoinSupportedShim(leftKeys, rightKeys) =>
+ buildSide(join) match {
+ case Some(BuildRight) if joinType == LeftSemi || joinType.isInstanceOf[ExistenceJoin] =>
+ // LeftSemi https://github.com/apache/datafusion-comet/issues/2667
+ // ExistenceJoin https://github.com/apache/datafusion-comet/issues/2697
+ planBySpark(join) { smj =>
+ withFallbackReason(smj, declineReason(s"BuildRight with $joinType is not supported"))
+ }
+ case Some(side) =>
+ buildSideOverLimit(join, side) match {
+ case Some(reason) =>
+ // Not a fallback: the kept sort-merge join still runs natively.
+ planBySpark(join)(smj => withInfo(smj, declineReason(reason)))
+ case None =>
+ Seq(
+ ShuffledHashJoinExec(
+ leftKeys,
+ rightKeys,
+ joinType,
+ side,
+ condition,
+ planLater(left),
+ planLater(right)))
+ }
+ case None => Nil
+ }
+ case _ => Nil
+ }
+
+ private def isLeftToSpark(join: Join, hint: JoinHint): Boolean =
+ isBroadcastStage(join.left) || isBroadcastStage(join.right) ||
+ canPlanAsBroadcastHashJoin(join, sqlConf) || hasStrategyHint(hint)
+
+ // canPlanAsBroadcastHashJoin misses a broadcast stage when the adaptive threshold is lower or
+ // off, and AQE discards a re-plan that hash joins one.
+ private def isBroadcastStage(plan: LogicalPlan): Boolean = plan match {
+ case LogicalQueryStage(_, _: BroadcastQueryStageExec) => true
+ case _ => false
+ }
+
+ // AQE's NO_BROADCAST_HASH and PREFER_SHUFFLE_HASH are outside JoinStrategyHint.strategies.
+ private def hasStrategyHint(hint: JoinHint): Boolean =
+ Seq(hint.leftHint, hint.rightHint).exists(
+ _.exists(_.strategy.exists(JoinStrategyHint.strategies.contains)))
Review Comment:
[P2] Honor AQE’s side-specific `PREFER_SHUFFLE_HASH` when that side is buildable. This hint falls outside `JoinStrategyHint.strategies`, so the new strategy can replace Spark’s preferred build side with the smaller relation by total bytes. One skewed partition on that relation can exhaust the non-spillable hash build. The base preserved Spark’s choice because it rewrote only sort-merge joins. Defer to Spark for this preference while retaining the existing handling of a lone `NO_BROADCAST_HASH`.
Evidence: On Spark 4.1.3, enable forcing, native range and AQE. Disable broadcasts, coalescing and skew splitting. Use 16 shuffle partitions, a 128 KiB advisory size and a 1,536 KiB local-map threshold. Join `range(90000)` keyed by `IF(id < 60000, 0L, id)` to `range(200000)` keyed by `id`, retaining `sha2(cast(id AS string), 256)` from both sides by summing their lengths. The largest left and right shuffle partitions were 4,367,889 and 937,190 bytes. AQE set `rightHint=(strategy=prefer_shuffle_hash)`. The base used `BuildRight`; the head used `BuildLeft`. With `spark.memory.offHeap.size=64m` and `spark.comet.exec.memoryPool.fraction=0.0625`, the base returned `[5760000,5760000]`. The head failed to reserve another 624,704 bytes for `HashJoinInput[0]` against the 4,194,304-byte limit.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts [datafusion-comet]
Posted by "rich7420 (via GitHub)" <gi...@apache.org>.
rich7420 commented on code in PR #6785:
URL: https://github.com/apache/datafusion-comet/pull/6785#discussion_r4391411217
##########
spark/src/main/scala/org/apache/comet/rules/CometShuffledHashJoinStrategy.scala:
##########
@@ -0,0 +1,175 @@
+/*
+ * 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.comet.rules
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.RowOrdering
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
+import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{Join, JoinHint, JoinStrategyHint, LogicalPlan}
+import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
+import org.apache.spark.sql.execution.adaptive.{BroadcastQueryStageExec, LogicalQueryStage}
+import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.CometConf
+import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, withFallbackReason, withInfo}
+import org.apache.comet.shims.ShimJoinSelection
+
+/**
+ * Plans a ShuffledHashJoinExec in place of a SortMergeJoinExec, before EnsureRequirements and
+ * RemoveRedundantSorts, so the sorts above the join survive. Adapted from Apache Gluten.
+ */
+case class CometShuffledHashJoinStrategy(session: SparkSession)
+ extends SparkStrategy
+ with ShimJoinSelection {
+
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case join: Join if isEnabled(join) => planJoin(join)
+ case _ => Nil
+ }
+
+ private def sqlConf: SQLConf = session.sessionState.conf
+
+ // Planner strategies run with Comet off and before CometRule, which applies plan-only mode.
+ private def isEnabled(join: Join): Boolean =
+ CometConf.COMET_FORCE_SHJ.get(sqlConf) &&
+ CometConf.COMET_EXEC_ENABLED.get(sqlConf) &&
+ !CometConf.COMET_EXPLAIN_PLAN_ONLY_ENABLED.get(sqlConf) &&
+ !join.isStreaming &&
+ isCometLoaded(sqlConf)
+
+ private def planJoin(join: Join): Seq[SparkPlan] = join match {
+ case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, _, left, right, hint)
+ if !isLeftToSpark(join, hint) && canSortMergeJoin(joinType) &&
+ RowOrdering.isOrderable(leftKeys) && hashJoinSupportedShim(leftKeys, rightKeys) =>
+ buildSide(join) match {
+ case Some(BuildRight) if joinType == LeftSemi || joinType.isInstanceOf[ExistenceJoin] =>
+ // LeftSemi https://github.com/apache/datafusion-comet/issues/2667
+ // ExistenceJoin https://github.com/apache/datafusion-comet/issues/2697
+ planBySpark(join) { smj =>
+ withFallbackReason(smj, declineReason(s"BuildRight with $joinType is not supported"))
+ }
+ case Some(side) =>
+ buildSideOverLimit(join, side) match {
+ case Some(reason) =>
+ // Not a fallback: the kept sort-merge join still runs natively.
+ planBySpark(join)(smj => withInfo(smj, declineReason(reason)))
+ case None =>
+ Seq(
+ ShuffledHashJoinExec(
+ leftKeys,
+ rightKeys,
+ joinType,
+ side,
+ condition,
+ planLater(left),
Review Comment:
Preserve the window sort for checkpoint probes; forced inner join followed by dense_rank still fails with missing PARTITION BY ordering.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org