kazuyukitanimura commented on code in PR #6835:
URL: https://github.com/apache/datafusion-comet/pull/6835#discussion_r4251243660
##########
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:
Thanks for closing the cache boundary in 2e0fb6346. I think RDDScanExec
needs the same treatment. Dataset.checkpoint and localCheckpoint build their
LogicalRDD through LogicalRDD.fromDataset, which keeps
executedPlan.outputOrdering. So a checkpoint of a Comet NOT IN, of an
undeclared outer join, or of a nested loop join carries the false order the
same way a cache does. With spark.comet.convert.rdd.enabled=true the scan
becomes a CometSparkToColumnar, which reports its child's order, and
mayReportFalseOrder trusts it because RDDScanExec is a leaf. Could the walk
treat an RDDScanExec that reports an order as unknown, since the plan behind it
is gone? Or more generally, could it stop trusting any leaf it doesn't
recognize? A test like the cached NOT IN one, with .localCheckpoint() in place
of .cache(), would pin it down. On the release this case only gets the order
wrong. Here it would split groups.
##########
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:
Treating every outer join of these shapes as suspect, and descending into
all children, keeps this safe. It also leaves a few shapes with #6787's wrong
order. One is a cache written by an outer join whose probe did get the
declaration, then outer-joined again. Another is a NOT IN on an inner join's
build side, which can't affect the streamed order. Could you either narrow the
walk to the child that carries the order, or file an issue listing these shapes
and link it from the PR? Then #6787 isn't closed with cases still open.
##########
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:
Would it work to let RightSemi, RightAnti and RightMark through here as well
as Inner? DataFusion builds their probe indices with get_semi_indices,
get_anti_indices and get_mark_indices, and the first two assert ascending
input. So they keep probe order whether or not the probe declares one. Only
Right depends on right_side_ordered, and null-aware LeftAnti already reports
[false, false]. As it stands, a sorted cache filtered with k IN (SELECT ...)
and then outer-joined still loses its order with default settings, and that's a
common query shape. The "undeclared outer join" test would then need another
way to write an out-of-order cache.
##########
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:
All of the new tests turn AQE off. Since the description points to #5634
making this reachable with default settings, could one case run with AQE on?
From 3.5 a cache written through repartition reports no order under AQE. The
default-settings case seems to be a broadcast left outer join over
spark.table(...).sortWithinPartitions("_1").cache() with no exchange in the
cached 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: [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]
