andygrove commented on code in PR #6835:
URL: https://github.com/apache/datafusion-comet/pull/6835#discussion_r4252061051
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -2876,12 +2921,44 @@ trait CometHashJoin {
else OperatorOuterClass.BuildSide.BuildRight)
.setNullAwareAntiJoin(isNullAwareAntiJoin)
.setDynamicFilterEnabled(CometConf.COMET_EXEC_JOIN_DYNAMIC_FILTER_ENABLED.get(join.conf))
+ .addAllStreamedSortOrders(streamedSortOrders.asJava)
condition.foreach(joinBuilder.setCondition)
Some(builder.setHashJoin(joinBuilder).build())
} else {
None
}
}
+
+ // Whether a Comet join that may move its streamed rows, while reporting
Spark's order for
+ // them, feeds the order that `plan` reports with no sort in between.
Whether an outer join
+ // declared the order on its probe input is up to the native planner, so
every one built that
+ // way counts. A cache reports the order of the plan that wrote it, so the
walk goes on into
+ // that plan. While CometExecRule converts a plan, a native plan's JVM
inputs sit behind a
+ // CometScanWrapper, a leaf, so the walk looks through that too.
+ private def mayReportFalseOrder(plan: SparkPlan): Boolean = plan match {
+ case _ if plan.outputOrdering.isEmpty => false
+ case _: SortExec | _: CometSortExec => false
+ case join: CometBroadcastHashJoinExec
+ if join.isNullAwareAntiJoin || runsAsRightOuterJoin(join.joinType,
join.buildSide) =>
+ true
+ case join: CometHashJoinExec if runsAsRightOuterJoin(join.joinType,
join.buildSide) => true
+ case _: CometBroadcastNestedLoopJoinExec => true
+ case wrapper: CometScanWrapper => mayReportFalseOrder(wrapper.originalPlan)
+ case scan: CometInMemoryTableScanExec =>
+ mayReportFalseOrder(scan.originalPlan.relation.cachedPlan)
+ case scan: InMemoryTableScanExec =>
mayReportFalseOrder(scan.relation.cachedPlan)
+ case stage: QueryStageExec => mayReportFalseOrder(stage.plan)
+ case _ => plan.children.exists(mayReportFalseOrder)
Review Comment:
Right, a checkpoint keeps `executedPlan.outputOrdering` in its `LogicalRDD`,
so the plan that wrote it is gone and the walk could not see a NOT IN or an
undeclared outer join behind it. I took your second suggestion: the walk now
trusts a leaf that reports an order only if it is a file or DSv2 scan, a range
or a Comet native leaf, and treats any other as unknown. The NOT IN test with
`localCheckpoint()` in place of `.cache()` returns 83 groups for 37 keys
without that. Fixed in 79b4d336d0.
##########
native/core/src/execution/planner.rs:
##########
@@ -2937,6 +2951,54 @@ impl PhysicalPlanner {
Ok(Arc::new(prepared))
}
+ /// `plan`, declaring `sort_orders` as the order its rows arrive in,
unless it already
+ /// declares one. Spark's plan knows the order of an input that enters the
native plan from
+ /// the JVM, but the `ScanExec` it arrives through declares none. Only
that order is declared:
+ /// when an operator in between may have moved rows, `plan` is left as it
is.
+ fn with_declared_order(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ sort_orders: &[Expr],
+ ) -> Result<Arc<dyn ExecutionPlan>, ExecutionError> {
+ if plan.output_ordering().is_some() ||
!Self::keeps_jvm_input_order(&plan) {
+ return Ok(plan);
+ }
+ let exprs = sort_orders
+ .iter()
+ .map(|expr| self.create_sort_expr(expr, plan.schema()))
+ .collect::<Result<Vec<_>, _>>()?;
+ match LexOrdering::new(exprs) {
+ Some(ordering) => Ok(Arc::new(SortedInputExec::new(plan,
ordering))),
+ None => Ok(plan),
+ }
+ }
+
+ /// Whether `plan`'s rows come from an input outside the native plan
through operators that
+ /// keep their input's order, so that the order Spark reports for them
still holds. Any other
+ /// operator may have moved rows, such as the null-aware anti join, whose
streamed side Comet
+ /// builds on, and declaring Spark's order over it would be wrong.
+ fn keeps_jvm_input_order(plan: &Arc<dyn ExecutionPlan>) -> bool {
+ if plan.is::<ScanExec>() || plan.is::<ShuffleScanExec>() {
+ return true;
+ }
+ // DataFusion reports every right hash join as keeping its probe
order, but keeps it only
+ // when the probe declares one, and no input on this path declares an
order.
+ if let Some(join) = plan.downcast_ref::<HashJoinExec>() {
+ if join.join_type() != &DFJoinType::Inner {
Review Comment:
Yes. In DataFusion 55.2 `adjust_indices_by_join_type` passes
`preserve_order_for_right` only for `Right`, `get_semi_indices` and
`get_anti_indices` assert ascending input, and `get_mark_indices` keeps the
whole range, so `keeps_jvm_input_order` now walks through Inner, RightSemi,
RightAnti and RightMark and refuses the rest. The planner test covers each
type, and a Comet test puts `k IN (SELECT ...)` under the outer join: it
declares the order and the partitions stay sorted. The undeclared outer join
test wrote its out-of-order cache through the IN subquery, so it now writes it
through a native sample, which the planner declares nothing through. Fixed in
79b4d336d0.
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -2876,12 +2921,44 @@ trait CometHashJoin {
else OperatorOuterClass.BuildSide.BuildRight)
.setNullAwareAntiJoin(isNullAwareAntiJoin)
.setDynamicFilterEnabled(CometConf.COMET_EXEC_JOIN_DYNAMIC_FILTER_ENABLED.get(join.conf))
+ .addAllStreamedSortOrders(streamedSortOrders.asJava)
condition.foreach(joinBuilder.setCondition)
Some(builder.setHashJoin(joinBuilder).build())
} else {
None
}
}
+
+ // Whether a Comet join that may move its streamed rows, while reporting
Spark's order for
+ // them, feeds the order that `plan` reports with no sort in between.
Whether an outer join
+ // declared the order on its probe input is up to the native planner, so
every one built that
+ // way counts. A cache reports the order of the plan that wrote it, so the
walk goes on into
+ // that plan. While CometExecRule converts a plan, a native plan's JVM
inputs sit behind a
+ // CometScanWrapper, a leaf, so the walk looks through that too.
+ private def mayReportFalseOrder(plan: SparkPlan): Boolean = plan match {
Review Comment:
Narrowed. The walk follows only the streamed side of a hash join, and an
outer join in a cache now counts as declared when it sent its order and its
probe input is a scan in its own plan, which is the one case where the planner
is sure to have declared it. A new test caches a sorted outer join and the join
over that cache declares the order. A build side behind an exchange reports no
order, so your NOT IN case only shows when the inner join's build side is a
cache with an order of its own. The new test uses two caches partitioned on the
key and fails against a walk of both children. Two shapes stay conservative, an
outer join in a cache whose probe is a native operator over the scan and an RDD
checkpoint, and I listed them in #6922 and linked it from the description.
Fixed in 79b4d336d0.
##########
spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala:
##########
@@ -3288,4 +3289,154 @@ class CometInMemoryCacheSuite extends CometTestBase {
}.nonEmpty)
}
}
+
+ test("hash joins over a sorted cache keep the streamed side's order") {
Review Comment:
Added `a broadcast outer hash join over a sorted cache keeps its order with
default settings` to `CometInMemoryCacheSuite`: AQE on, only the native cache
flag set, a cache with no exchange in its plan, and the broadcast chosen by
size. The cached NOT IN tests run with AQE off and on now too, in
`CometJoinSuite` for Spark's scan and in `CometInMemoryCacheSuite` for the
native one. Fixed in 79b4d336d0.
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -2876,12 +2921,44 @@ trait CometHashJoin {
else OperatorOuterClass.BuildSide.BuildRight)
.setNullAwareAntiJoin(isNullAwareAntiJoin)
.setDynamicFilterEnabled(CometConf.COMET_EXEC_JOIN_DYNAMIC_FILTER_ENABLED.get(join.conf))
+ .addAllStreamedSortOrders(streamedSortOrders.asJava)
condition.foreach(joinBuilder.setCondition)
Some(builder.setHashJoin(joinBuilder).build())
} else {
None
}
}
+
+ // Whether a Comet join that may move its streamed rows, while reporting
Spark's order for
+ // them, feeds the order that `plan` reports with no sort in between.
Whether an outer join
+ // declared the order on its probe input is up to the native planner, so
every one built that
+ // way counts. A cache reports the order of the plan that wrote it, so the
walk goes on into
+ // that plan. While CometExecRule converts a plan, a native plan's JVM
inputs sit behind a
+ // CometScanWrapper, a leaf, so the walk looks through that too.
+ private def mayReportFalseOrder(plan: SparkPlan): Boolean = plan match {
+ case _ if plan.outputOrdering.isEmpty => false
+ case _: SortExec | _: CometSortExec => false
+ case join: CometBroadcastHashJoinExec
+ if join.isNullAwareAntiJoin || runsAsRightOuterJoin(join.joinType,
join.buildSide) =>
+ true
+ case join: CometHashJoinExec if runsAsRightOuterJoin(join.joinType,
join.buildSide) => true
+ case _: CometBroadcastNestedLoopJoinExec => true
+ case wrapper: CometScanWrapper => mayReportFalseOrder(wrapper.originalPlan)
+ case scan: CometInMemoryTableScanExec =>
+ mayReportFalseOrder(scan.originalPlan.relation.cachedPlan)
Review Comment:
Confirmed, thanks. `InMemoryTableScanExec` reads `adaptive.executedPlan`
once the plan is final, and the adaptive plan reports no order itself, so the
walk stopped at it. `mayReportFalseOrder` now reads a final adaptive plan
before it looks at the order, as the scan does. The cached NOT IN tests run
with AQE off and on, and without the change the AQE-on run returns 83 groups
for 37 keys. Fixed in 79b4d336d0.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
