sunchao commented on code in PR #6835:
URL: https://github.com/apache/datafusion-comet/pull/6835#discussion_r4249245927
##########
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:
[P1] [P1] Verify ordering across cache boundaries before trusting a scan
A `ScanExec` can read a cache populated by a native null-aware anti join.
Materialize `SELECT k FROM sorted_cached WHERE k NOT IN (...)` into another
cache, then use it as the streamed input of a broadcast left outer join
followed by `GROUP BY k` without an exchange. The anti join reverses build
batches, but `CometBroadcastHashJoinExec.outputOrdering` and Spark’s cache scan
retain `k ASC`. This check sees only the cache scan and returns `true`, so
`SortedInputExec` enables sorted aggregation over unsorted rows. A focused
reproduction returns 49 groups for 37 keys, splitting their counts, whereas the
pre-PR behavior returns 37 correct groups. The existing regression test leaves
the anti join in the same native plan and therefore misses this boundary.
Restore the anti join’s claimed order before caching, or validate the cached
producer’s ordering before applying the declaration, and cover this cached
variant.
Evidence: Recompiled `/tmp/review6835-pinned-8hg9r0yc/cache_boundary.rs`
using the exact-head `sorted_input.rs`. Input: 100,000 sorted rows with `k = (i
% 40) * 1,000,000`, batches of 8,192, excluding keys 1,000,000, 10,000,000,
20,000,000 and an absent 100,000,000 through the native null-aware anti join.
Materialized and replayed its output, outer-joined against the even keys, then
grouped by `k`. Without declaration: 37 groups, count 2,500 each. With the PR’s
declaration: 49 groups containing only 37 distinct keys; key 3,000,000 split
into counts 692 and 1,808. An actual-sort control returned 37 correct groups.
The Spark-only `spark_metadata.py` run confirmed the cached anti result retains
`k ASC` and `SinglePartition`, and the final plan has two aggregates above the
outer join without an exchange. DataFusion 55.1.0 harness dependencies have
byte-identical relevant join and aggregate sources to locked 55.2.0. This was
bounded native execution plus Spark metadata validation, no
t a full Comet JVM run.
--
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]
