andygrove commented on code in PR #6835:
URL: https://github.com/apache/datafusion-comet/pull/6835#discussion_r4249879841
##########
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>() {
Review Comment:
Confirmed, thanks. I reproduced it in a Comet test with AQE off, 83 groups
for 37 keys. From Spark 3.5 a cached plan with a join runs under AQE and
reports no order, but Spark 3.4 runs every cached plan without AQE, so there it
only takes enabling Comet's cache, which is off by default on 3.4. The NOT IN
isn't the only join that can write a cache like that. One of this PR's own
outer joins can, when the native planner declares no order on its probe input
(an IN subquery underneath is enough), and so can a native nested loop join,
which DataFusion runs build row by build row.
Fixed in 2e0fb63469. Before it sends `streamed_sort_orders`, `doConvert` now
walks down the streamed side, through the `CometScanWrapper` that hides a
native plan's JVM inputs during conversion and into the plan that wrote any
cache it reads, and sends no order when one of those joins produces it with no
sort in between. A sort above such a join still lets the declaration through.
New tests cover the cached NOT IN, a cache written by an undeclared outer join,
one written by a nested loop join, and one sorted above an outer join, and each
fails against a mutant of the case it covers.
--
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]
