viirya commented on code in PR #6835:
URL: https://github.com/apache/datafusion-comet/pull/6835#discussion_r4267168974
##########
native/core/src/execution/planner.rs:
##########
@@ -2937,6 +2951,63 @@ 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) {
Review Comment:
There is a case in this direction, and it's one of the PR's own tests. `an
outer hash join over a cache of an outer hash join with no declared order`
reads a `TABLESAMPLE` under the outer join and asserts that the serde sent the
order (`getStreamedSortOrdersCount > 0`). The native `SampleExec` doesn't
override `maintains_input_order`, so `keeps_jvm_input_order` stops there and
nothing is declared. A `debug_assert` would fire on that test, and a comment
stating the invariant wouldn't hold today.
Within that query the mismatch doesn't change the result, because Spark's
`SampleExec` reports no order and Spark keeps the sort above the join. The
serde sends the order only because `CometSampleExec` reports its child's order
where Spark's doesn't. That report is what does the damage outside the query.
The outer join reports `k ASC` over rows it moved, and a cache of it carries
that claim to whatever reads it next, which is @kazuyukitanimura's point in the
sample thread.
Either of the fixes there would remove this mismatch. If `CometSampleExec`
reported no order, as Spark's does, the serde would send nothing for this join.
If the native `SampleExec` reported `maintains_input_order`, the planner would
declare the order the serde sent. That only covers the sample, though. Any
Comet operator that reports an order but whose native side doesn't report
`maintains_input_order` would split the two checks the same way.
--
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]
