[comphead]

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 needs its input to declare the partition keys' order. BoundedWindowAggExec, which runs in InputOrderMode::Sorted, fails the task with Expects PARTITION BY expression to be ordered. WindowAggExec fails when its stream starts, with All partition by columns should have an ordering. 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 CometSortExec child declares its own order, so nothing is sent for it. 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. Both window operators find the partition keys' order through get_ordered_partition_by_indices on that shared input, so the declaration 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 covers row_number (BoundedWindowAggExec) and percent_rank (WindowAggExec). It checks that the window's input is a SortedInputExec over the scan whose order covers the partition key, and that without the orders it is the bare scan, whose order does not. Leaving an input that already declares an order alone is the same early return in with_declared_order that hash_join_keeps_probe_input_that_declares_an_order covers. The 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. The partition key is nullable and the order key descending, so the declared order has to match the cache's direction and null placement. It runs dense_rank, which lands on BoundedWindowAggExec, and a sum with a frame up to UNBOUNDED FOLLOWING, which lands on WindowAggExec. Each checks the rows, that the serde sent both orders, and that no sort is planned.
• CometInMemoryCacheSuite: a window over a cache sorted by a collated string key stays in Spark (Spark 4.x) checks that the window falls back with the new reason and returns Spark's rows.
• A Spark SortExec directly under a native window does not come up. Comet converts only leaf operators into a native operator's JVM input, and with spark.comet.exec.sort.enabled=false the window stays in Spark.

The window snippet in the debugging guide now builds both operators over input.

Run locally with JDK 17, rebased on 7d251b3619: the native tests above pass, the cache test passes on -Pspark-3.5 and -Pspark-4.1, and the collated test passes on -Pspark-4.1. With only the native change in window.rs reverted, the cache test fails with CometNativeException: Expects PARTITION BY expression to be ordered. The Spark SQL suites run through the run-all-spark-profiles and run-spark-4.1-tests labels.

[andygrove]

Thanks @comphead. This follows the streamed_sort_orders approach from #6835 and the root cause in the description matches what I see in DataFusion 55.2.0. Both BoundedWindowAggExec and WindowAggExec call get_partition_by_sort_exprs, so declaring the order on their shared input is the right place. Declaring it only above a ScanExec, ShuffleScanExec or single-file-group scan, and only when the input declares nothing, looks right to me.

I compared against branch-1.2. A window over a child whose reported order satisfies it used to fail natively with Expects PARTITION BY expression to be ordered. Now it runs, or falls back to Spark when a key can't be expressed or is a collated string. I found no change that makes Comet diverge from Spark where the release did not. When the reported order may be false, nothing is declared and the native window still fails. I think that is the safer outcome, since Spark dropped its sort on the same false order.

I have a few things I'd like addressed before merge.

1. Could you add a test with an unbounded frame, for example sum(v) OVER (PARTITION BY k ORDER BY v ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING) over the same cache? Both new tests land on BoundedWindowAggExec (row_number in Rust, dense_rank in Scala), so the WindowAggExec branch in WindowBuilder isn't exercised. The description says the declaration covers both.

2. Would it be worth adding a nullable partition key and a descending order key to the cache test data? That would show that the direction and null ordering taken from requiredOrdering match what the cache reports. If they ever disagreed, the window would split partitions silently instead of failing.

3. Could you add a Spark 4.x test for the new collated string key fallback? It should check that the window stays in Spark and the result is correct. Is the case where the child is a Spark SortExec that Comet could not convert covered anywhere? The serde now sends the order there too, since the child is not a CometSortExec.

4. The WindowBuilder::build snippet in docs/source/contributor-guide/debugging.md (around line 270) still passes Arc::clone(&child.native_plan) to both window operators. Could it be updated to input?

5. This touches the serde and the planner on a default path, and the PR tier only ran Spark 4.1. Could we get a verdict from the Spark SQL suites before this is queued? Applying run-all-spark-profiles and run-spark-4.1-tests, or running dev/local-ci.sh spark sql_core-1, would do it.

[comphead]

Thanks @andygrove, addressed in 94aaa19.

1. The cache test now also runs sum(v) OVER (PARTITION BY k ORDER BY v DESC ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING), which lands on WindowAggExec. It runs as a separate query from dense_rank, since Spark would merge the two into one window. The planner test covers row_number and percent_rank too. This turned up a mistake in my description: WindowAggExec doesn't fail in partition_by_sort_keys(). It fails when its stream starts, with "All partition by columns should have an ordering". So the test now checks the input's order through get_ordered_partition_by_indices, which both operators use. I've corrected the description.
2. Done. The partition key is NULL for a twentieth of the rows, and the cache and both windows order v descending. The rows are checked against computed values.
3. Added a Spark 4.x test over a cache sorted by a UTF8_LCASE key. It checks that the window stays in Spark with the new fallback reason and returns Spark's rows. A Spark SortExec directly under a native window doesn't come up. Comet converts only leaf operators (cache, range, scans) into a native operator's JVM input. With spark.comet.exec.sort.enabled=false the window stays in Spark, so the order is never sent there.
4. Updated debugging.md to pass input.
5. Applied run-all-spark-profiles and run-spark-4.1-tests.