comphead opened a new pull request, #7256: URL: https://github.com/apache/datafusion-comet/pull/7256
## Which issue does this PR close? Closes #7223. ## Rationale for this change Spark puts no sort below a window when its input reports the order the window needs, a relation sorted before it was cached for instance. When those rows enter the native plan from the JVM, the `ScanExec` they arrive through declares no order to DataFusion. The native window runs in `InputOrderMode::Sorted`, checks its input's declared order, and fails the task with `Expects PARTITION BY expression to be ordered`. The rows are sorted. Only the declaration is missing. DataFusion has no change pending here. `get_partition_by_sort_exprs` raises the error on DataFusion `main` as it does in 55.2.0, and it is right to: it trusts the order its input declares, and Comet's scan declares none. #6835 fixed the same gap for outer hash joins with `streamed_sort_orders`. This PR does the same for the window, which is the first option in the issue. ## What changes are included in this PR? - `Window` gets `repeated Expr input_sort_orders = 5`. - `CometWindowExec` sends the window's required ordering (the partition keys ascending, then the order keys) when the child reports an order that satisfies it, so Spark dropped the sort, and `mayReportFalseOrder` trusts that order. A new `mayReportFalseInputOrder` in the `CometHashJoin` trait forwards to it. If the order is trusted but a key cannot be expressed or holds a collated string, the window falls back to Spark with a reason instead of failing natively. - `WindowBuilder` declares the order on the window's input with `with_declared_order`, now `pub(super)`. It is the same walk the hash join uses, so the order is declared only above a `ScanExec`, `ShuffleScanExec` or a scan of one file group, and an input that already declares an order is left as it is. `BoundedWindowAggExec` and `WindowAggExec` both run the check, and the declaration on their shared input covers both. When the order the input reports may be false, nothing is declared and the native window fails as it does on `main`. This PR does not change that case. ## How are these changes tested? - `planner.rs`: `window_declares_input_order_on_a_scan` checks that the window's input is a `SortedInputExec` over the scan and that `partition_by_sort_keys()` succeeds, and that without the orders it returns the original error. `window_keeps_input_that_declares_an_order` checks that a native sort below the window is left alone. The `row_number` window expression moved into a helper that `window_cast_projection_reports_window_time` now uses as well. - `CometInMemoryCacheSuite`: `a native window over a sorted cache declares the cache's order` runs the repro from the issue with the native cache scan on and off. It checks the ranks, that the serde sent both orders, and that no sort is planned. Run locally on `-Pspark-3.5` with JDK 17: the three native tests and the new Scala test pass. With only the native change in `window.rs` reverted, the Scala test fails with `CometNativeException: Expects PARTITION BY expression to be ordered`. Spark's SQL suite and the other Spark profiles have not been 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]
