[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 order of the window's partition keys, the only order the native window checks, when the child reports an order that satisfies the window's requirement, so Spark dropped the sort. Over a native sort the order is sent too, and the planner leaves the sort's own declaration as it is. Nothing is sent when mayReportFalseOrder doubts that order. A new mayReportFalseInputOrder in the CometHashJoin trait exposes that check. A window whose partition or order key holds a collated string falls back to Spark when that order is trusted, as WindowGroupLimit does, since the native window compares keys byte by byte. Over a doubted order it stays native, as on main, since Spark's window would trust that order too.
• 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 whose declared order already satisfies the one sent is left as it is. When the input declares only part of it, as a window over another window does, the rest is declared next to it. So a window stacked over another one gets the longer order it needs, including the order of another alias of the same sort key. A hash join's probe input that declares part of its order is extended the same way. Both window operators find the partition keys' order through get_ordered_partition_by_indices on that shared input, so the declaration covers both.
• keeps_source_order also accepts an EmptyExec, which is how the native scan plans a bucket with no file. Its rows are in any order, and the window checks its input's order before it polls it, so one empty bucket of a sorted bucketed table failed the query. The hash join's probe side gets the same declaration, which changes nothing there. It also accepts a right outer hash join whose probe declares an order, since that is when DataFusion keeps the join's unmatched probe rows in place. A window over an outer join then gets an alias's order declared next to the join's.

When the order the input reports may be false, nothing is sent, and a native window directly over it fails with the ordering error as on main. Falling back would hand the same rows to Spark's window, which trusts the order too and would group them wrongly. #6785 covers that case: it keeps a nested loop join over an ordered streamed side in Spark (#6898), and it sorts a window's input natively when the partition keys' order is not declared. DataFusion's InputOrderMode::Linear needs no declared order, but it hashes every partition and WindowAggExec has no such mode, so Comet keeps Sorted and declares the order instead.

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 whose order covers the partition key, and that without the orders it is the bare scan, whose order does not. declared_order_extends_an_order_the_input_declares checks the declaration over an EmptyExec, that an input whose order already satisfies the one sent is left alone, that a trusted input gets the missing order next to its own, and that a sort gets nothing. hash_join_declares_streamed_order_on_probe_input also checks that a right outer join over a declared probe keeps the source order. 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, so the declared order has to match the cache's null placement, and the order key is descending. 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 k ASC NULLS FIRST and not the order key, and that no sort is planned.
• CometInMemoryCacheSuite: a native window declares its order next to the order its input declares runs a window by k under one by (k, x) over a cache sorted by (k, x), a window by k under one by a, an alias of k, and a window by k over a left outer join whose streamed order Spark reports as the alias a, with a residual condition that leaves unmatched rows among the matched ones of a key. Each checks the rows, the keys each window sends, and that no sort is planned.
• CometInMemoryCacheSuite: a window over a cache sorted by a collated string key stays in Spark (Spark 4.x) runs a collated partition key and a collated order key, and checks that the window falls back and returns Spark's rows.
• CometWindowExecSuite: window over a sorted bucketed table with empty buckets runs natively writes a table with eight buckets sorted by (k, v) and two keys, reads it with spark.sql.legacy.bucketedTableScan.outputOrdering on, and checks the window against Spark and that it sits directly on the bucketed native scan.
• 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. The SortedInputExec doc and "Reporting an Output Ordering" in the operator guide now cover the window.

Run locally with JDK 17, rebased on 383ec5627c: TODO (re-run after the second review round: the native tests, both cache window tests and the bucketed test on -Pspark-3.5 and -Pspark-4.1, the collated test 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.

[dwsmith1983]

#6785 adds a sort to WindowBuilder for the case this PR leaves as is: when the window's input doesn't show every partition key as ordered, the window sorts it by the partition keys and then the order keys. Both PRs change the try_new calls in window.rs. Whichever lands second can apply with_declared_order first and run the check on its result, so a trusted input keeps the declaration and gets no sort, and the "no orders" half of window_declares_input_order_on_a_scan then expects a SortExec over the scan.

[rich7420]

LGTM. Window regressions and Spark SQL passed CI. No local tests.

[sunchao]

The direct sorted-cache tests now exercise both native window operators. I found two remaining cases of #7223: a second window needs a longer ordering than the first one declares, and a sorted bucketed scan has an empty task. I’ve left a reproducer for each inline.

I verified both Spark 4.1.3 results and the exact-head JVM plans, including the serialized key counts and the fileless partitions. I traced the execution failures through the head’s Rust planner and DataFusion 55.2.0. The local Spark 4.1 JVM and test compilation passed. I did not run these queries against the head’s native binary locally. The head’s Rust and Spark 4.1 SQL CI checks passed.

[sunchao]

Could we preserve the usable part of the cache's ordering for the next window? On a single-partition cache sorted by (k, v), compute s = sum(v) OVER (PARTITION BY k), then sum(s) OVER (PARTITION BY k, v). Spark 4.1.3 returns all 24 rows in my example with no intervening sort or exchange. On this head, both CometWindowExecs advertise (k, v) to Spark, but the first serializes only k. The second sends both keys and with_declared_order returns early because its input already has an ordering. DataFusion therefore has no ordering for v when the second window starts. Keeping the trusted, expressible suffix at the first window, or extending a partial declaration safely at the second, would cover this. Could we add a stacked-window regression?

[comphead]

Thanks, good catch. The lower window now sends the required keys and then the rest of the order its child reports, up to the first key that cannot be expressed. In your example it declares (k, v) on the cache scan, and the upper window's input carries that order through DataFusion's window equivalence properties, so with_declared_order can return early there. Fixed in 16b1add.

I extended the order at the lower window, not at the upper one. A key left out at the lower window is a collated string or cannot be expressed. The upper window would be checking the same attribute, so I expect it to fall back to Spark rather than fail natively.

stacked native windows over a sorted cache keep the cache's whole order runs your shape with the native cache on and off: sum(v) OVER (PARTITION BY k) under sum(s) OVER (PARTITION BY k, v), over a cache sorted by (k, v) with a nullable k. It checks the rows, that both windows send k ASC NULLS FIRST, v ASC NULLS FIRST, and that no sort is planned.

[sunchao]

Could this also declare the order for an empty bucket? I tried a Parquet table written by one writer with bucketBy(8, "k").sortBy("k", "v"), two distinct k values, and legacy bucketed scan output ordering enabled. The head plans CometWindowExec directly over CometNativeScan for dense_rank() OVER (PARTITION BY k ORDER BY v) and sends both sort keys. Six of the eight tasks have no files. build_parquet_scan_plan gives those tasks an EmptyExec, which keeps_source_order rejects. DataFusion's bounded window checks its partition ordering before polling the empty stream, so one empty bucket is enough to fail the query. Vanilla Spark returns all 16 rows. An empty input satisfies any ordering. Could we handle that case and cover a table with unpopulated buckets?

[comphead]

Agreed, an empty input satisfies any order. keeps_source_order now accepts EmptyExec, so an empty bucket gets the same declaration as a bucket with a file. Fixed in 16b1add.

declared_order_reaches_an_empty_bucket in planner.rs checks the declaration over an EmptyExec. window over a sorted bucketed table with empty buckets runs natively in CometWindowExecSuite writes your table shape: one writer, bucketBy(8, "k").sortBy("k", "v") and two keys. It reads the table with the legacy bucketed scan ordering on, then checks dense_rank() OVER (PARTITION BY k ORDER BY v) against Spark. It also checks that the window sits directly on the bucketed CometNativeScanExec and that no sort is planned. The hash join's probe side uses the same walk. An empty probe input gets the declaration there too, which changes nothing for it.

[sunchao]

Could we assert the columns, directions and null placements in input_sort_orders here? If only the declaration for v changed from DESC NULLS LAST to ASC NULLS FIRST, the cache would still deliver the same rows and the window would still use its separate order_by_list to calculate both results. The count would also stay at two. Checking k ASC NULLS FIRST, v DESC NULLS LAST directly would catch that regression.

[comphead]

Done in 16b1add. The test now compares the declared keys with 0 Ascending NullsFirst, 1 Descending NullsLast. It uses a small windowInputOrder helper, which the stacked-window test uses as well.

[sunchao]

I re-ran the two earlier reproductions on Spark 4.1.3 with the native library from this head's CI run. The stacked (k) then (k, v) windows matched Spark on all 24 rows. The eight-bucket scan matched on all 16 rows, including six empty tasks. The cache test also now asserts the bound columns, directions and null placements. All three of my earlier examples are addressed.

One case of #7223 remains when Spark knows two keys have the same ordering. I sorted a single-partition input by k, projected k AS a, k AS b, cached the result, and ran sum(v) OVER (PARTITION BY a) followed by sum(s) OVER (PARTITION BY b), where s is the first sum. Spark recognized the aliases and returned all 12 rows without an intervening sort. Comet declared a for the lower window. The upper window sent b, but with_declared_order kept the existing ordering on a, and execution failed with All partition by columns should have an ordering. In a matched control, using a for both windows succeeded.

[sunchao]

There is also a projection-sensitive version of the alias case. On a cache sorted by (k, x), expose x twice as a and b, then run a window partitioned by k under one partitioned by (k, b). With both aliases in the scan output, the lower window declares (k, b) and Comet returns all 24 rows. When the query drops unused a, Spark still reports b as the suffix key and lists a in sameOrderExpressions. order.references includes both names, although only b is serialized. This check therefore drops the suffix, the lower window declares just k, and the upper window fails with two partition keys required and only one ordered. Spark 4.1.3 returned all 24 rows in both projections.

[comphead]

Thanks. The check now looks at order.child.references, since only the child is serialized. Fixed in dc94ccf. With a pruned, the lower window declares (k, b) again.

The second query in stacked native windows over a sorted cache with aliases of a sort key is this shape. A cache sorted by (k, x) exposes x as a and b, the query drops a, and sum(v) OVER (PARTITION BY k) runs under sum(s) OVER (PARTITION BY k, b). It checks the rows, that both windows send k, b, and that no sort is planned.

[comphead]

Thanks @sunchao, both alias cases are fixed in dc94ccf.

with_declared_order used to return early whenever its input declared an order. Now it returns early only when the declared order already satisfies the one Spark sent. Otherwise, if keeps_source_order trusts the input, it adds the order next to what the input declares. keeps_source_order walks through both window operators and SortedInputExec, since all of them keep their input's order. In your first example the upper window's input declares a, and b is now added. This also extends a partial declaration at the upper window, your earlier suggestion. So a suffix key the lower window leaves out is no longer the only way for the upper window to get its order. An operator that may move rows, such as a native sort, still blocks the declaration.

declared_order_extends_an_order_the_input_declares in planner.rs covers the three outcomes: an order the input already satisfies is left alone, a trusted input gets the second order, and a sort gets nothing. The first query in stacked native windows over a sorted cache with aliases of a sort key is your 12-row example, extended to 24 rows with NULL keys: k AS a, k AS b over a cache sorted by k, then sum(v) OVER (PARTITION BY a) under sum(s) OVER (PARTITION BY b). It checks the rows, that the lower window sends a and the upper b, and that no sort is planned.

[comphead]

A follow-up in 9c5dcd0 simplifies the last two rounds.

The window serde no longer sends the keys its child reports past the ones the window requires. That also removes the order.child.references check mentioned above. Since dc94ccf3d3, with_declared_order extends a partial order at the upper window, so the upper window declares what it needs itself. That covers both the (k) then (k, v) stack and the alias case. Nothing else in a native plan relies on a declared order it did not send itself. A sort-based aggregate adds its own sort, and the others use a declared order only to run faster. So each operator now declares only the order it needs.

The two stacked-window tests are now one, stacked native windows over a sorted cache. It runs a window by k under one by (k, x), and a window by k under one by a, an alias of k. The lower windows now send only k. I also dropped the separate EmptyExec unit test, which declared_order_extends_an_order_the_input_declares now covers, and trimmed some comments.

[andygrove]

Thanks @comphead. The root cause matches what I see in DataFusion 55.2.0, and declaring the order on the window's shared input is the right place. The follow-up rounds closed the cases sunchao found, and the stacked and empty-bucket tests cover them well.

I compared against branch-1.2. A window over a child whose reported order satisfies the requirement used to fail natively with Expects PARTITION BY expression to be ordered. Now it runs, or falls back to Spark with a reason. The hash join probe change only adds an order Spark already relies on. I found no change that makes Comet diverge from Spark where the release did not.

A few things I'd like addressed.

1. The doc on SortedInputExec in sorted_input.rs still says the planner puts it only on a hash join's probe input. A window's input gets one now too. The "Reporting an Output Ordering" section of adding_a_new_operator.md also describes only the hash join and streamedSortOrders. Could both say that CometWindowExec sends the order its child reports? They could also say that an operator which reports an order it cannot keep natively now breaks a window as well as a join.

2. In CometWindowExec.scala, when mayReportFalseInputOrder is true, Spark has already dropped the sort. The native window then fails with Expects PARTITION BY expression to be ordered, which the description says is unchanged from main. This block already falls back with a reason when a key cannot be expressed. Would it make sense to fall back here too, so the query runs in Spark instead of failing the task? If you would rather keep it out of this PR, could you file an issue and link it?

3. Both window operators read only the partition keys' order, as far as I can tell. The declaration also covers the order keys. For a float order key, Spark treats -0.0 and 0.0 as equal and DataFusion's order does not, so the declared order could claim more than the data has. The window itself would not notice, but anything above it that trusts a declared order might. Does declaring only the partition-key prefix work for every case in the stacked tests? If you keep the full order, could you add a test with a double order key holding -0.0, 0.0 and NaN over a sorted cache?

The Spark 3.5 [scans] job is still pending. I'll approve once it is green and these are answered.

[comphead]

Thanks @andygrove, all three are addressed in 8e0b8e0.

1. The SortedInputExec doc now says the planner puts it on a window's input too. The "Reporting an Output Ordering" section says that a native window declares the order of its partition keys. An operator that reports an order it cannot keep now breaks a window as well as an outer hash join, and both serdes fall back over it.
2. The window now falls back the way the outer hash join does. When mayReportFalseOrder with countInPlan = false flags the child's order, nothing native can declare it, so the window stays in Spark with the reason "the order of the window's input, which Spark relies on, cannot be checked, so the window stays in Spark". When only a Comet join in the same native plan raises the doubt, nothing is sent, and DataFusion uses the order it sees there. a window over a cache whose order cannot be checked stays in Spark covers the fallback. It reuses the setup of the outer-join test, with a filter that hides the first join's declaration, and it fails natively without the change.
3. Agreed, the window now declares only its partition keys. get_partition_by_sort_exprs is the only check either operator makes. required_input_ordering lists the order keys too, but only DataFusion's optimizer reads it, and Comet does not run that. Every case in the stacked tests still runs, since each window sends its own partition keys. A float order key is never declared now. A float partition key never reaches the declaration either. NormalizeFloatingNumbers rewrites the window's partition keys, so a cache sorted by the raw column does not satisfy the required order, and Spark plans a sort.

[sunchao]

I rechecked 8e0b8e0da1 on Spark 4.1.3. The stacked windows over a sorted cache and the empty-bucket scan now match Spark, as do the signed-zero and NaN window cases I tried in both sort directions.

I reproduced two wrong results on this revision: a Spark window counts separate runs of the same key after a native nested-loop join, and a native window assigns separate ranks to case-insensitive peers. There is also a remaining execution failure when a window partitions by an equivalent alias above an ordered outer hash join. Each inline comment includes the inputs and a control.

I used JVM classes from this revision and the prior CI native library after verifying that the production Rust implementation is unchanged between the two commits.

[sunchao]

Falling back to Spark here leaves the window above a join whose output is out of order. I cached 12 rows in one partition, sorted by k = v % 3, then broadcast two build rows (z = 0, 1) and joined on v % 2 <= z. The native nested-loop join emits keys 0, 1, 2 twice, once for each build row. Spark has already omitted the sort, so count(*) OVER (PARTITION BY k) counts the separate runs. Six rows get 2 and twelve get 4. Vanilla Spark and a control with a physical sort after the same native join both return 6 on all 18 rows. Caching the join output also reproduces it. The previous revision raised an ordering error on this query. The fallback needs to restore the required order before the Spark window consumes those rows.

[comphead]

Thanks, agreed. Falling back hands the same out-of-order rows to Spark's window, which trusts the order as much as the native one does. 136e5cf removes the fallback. When the order may be false, the window sends no order, so the native window rejects its input with the ordering error, as on main and the previous revision, instead of returning wrong rows.

Restoring the order needs either a nested loop join that keeps it or a sort below the window. #6785 does both: it keeps a nested loop join over an ordered streamed side in Spark (#6898), and it sorts a window's input natively when the partition keys' order is not declared.

From reading the code, a native WindowGroupLimit between the join and the window gives wrong rows on main already, since PartitionedRankLimitExec declares the partition and order keys' order itself. Keeping the join in Spark covers that shape too.

[sunchao]

The partition keys are sufficient for the ordering declaration, but a window still compares its ORDER BY keys to find peers. With the native cache reader, I cached (k, s) sorted by both columns, where k is an integer and s uses UTF8_LCASE. For A, a, B, b within each k, Spark 4.1.3 gives dense_rank() OVER (PARTITION BY k ORDER BY s) ranks 1, 1, 2, 2. This commit runs CometWindow directly over CometInMemoryTableScan and gives 1, 2, 3, 4. The previous revision fell back and matched Spark. Checking collation on the order keys before conversion would still allow the declaration itself to contain only partition keys.

[comphead]

Thanks. 136e5cf checks every partition and order key for a collated string before anything is converted, so such a window stays in Spark whatever its input is. The native window finds partitions and peers by comparing keys byte by byte. The check now also covers a window with no PARTITION BY, which never reached the old one. The declaration still holds only the partition keys.

a window over a cache sorted by a collated string key stays in Spark now also runs your shape: dense_rank() OVER (PARTITION BY n ORDER BY k) over a cache sorted by an integer n and a UTF8_LCASE k. It checks Spark's ranks and the fallback reason.

[sunchao]

This early return still drops a valid ordering above a left outer broadcast hash join. I sorted range(12) by k = id % 3, projected k AS a, k AS b, id AS v, and cached the result in one partition. I then left-joined a broadcast one-row table on a, kept both aliases in the output, and ran sum(v) OVER (PARTITION BY a). Spark knows a and b have the same order and returns all 12 rows (group sums 18, 22, 26). The native join declares its probe ordered by b, and the window sends a, but keeps_source_order rejects the DataFusion right outer join even with that ordered probe. The task fails with All partition by columns should have an ordering. Changing only the window to PARTITION BY b returns all 12 expected rows with the same native join and no sort.

[comphead]

Thanks. keeps_source_order now accepts a right outer join whose probe declares an order. That is exactly when DataFusion keeps the join's unmatched probe rows in place, since HashJoinExec::execute hands right.output_ordering().is_some() to the stream. So the window's order is declared next to the join's, and your query runs. A right outer join whose probe declares nothing is still rejected. Fixed in 136e5cf.

hash_join_declares_streamed_order_on_probe_input in planner.rs now also checks the rule for both outer join shapes. The cache test a native window declares its order next to the order its input declares now runs your shape: a left outer join with a one-row broadcast table on k, and sum(v) OVER (PARTITION BY a) above it, where a is an alias of k. A residual condition leaves unmatched rows among the matched ones of key 0, so the query would split that key if the join moved them. It checks the rows, that the window sends a, and that no sort is planned.

[comphead]

Thanks @sunchao, all three are fixed in 136e5cf. Details are inline.

@andygrove, this reverts the fallback from your point 2. sunchao showed that it returns wrong rows when the order really is false, since Spark's window trusts that order too. A window whose input order may be false fails natively with the ordering error again, as on main. #6785 covers that case: it keeps a nested loop join over an ordered streamed side in Spark (#6898), and it sorts a window's input natively when the partition keys' order is not declared.

[sunchao]

I rechecked the three previous findings on Spark 4.1.3. On a sorted cache, case-insensitive peers now receive the same rank as in Spark. The ordered outer hash join also returns all 12 expected rows with either alias. For the numeric-only window above a genuinely reordered nested-loop join, Comet again raises the missing-order error.

One wrong-result case remains when that reordered join feeds a window with a collated key. I left the reproduction inline, along with a small change that makes the outer-join regression test distinguish the old and new native planners.

[sunchao]

This fallback can still give Spark a window input whose rows are out of order. I cached 12 rows in one partition, sorted by integer k = v % 3 and s COLLATE UTF8_LCASE, with s cycling through A, a, B, b. I then broadcast two rows (z = 0, 1) and joined on v % 2 <= z. The native nested-loop join emits each k in two separate runs. The collated order key sends count(*) OVER (PARTITION BY k ORDER BY s ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) to Spark, which has already omitted the sort.

On Spark 4.1.3 the fallback returned 2 on six rows and 4 on twelve. Vanilla Spark and a physical sort above the same native join both returned 6 on all 18. Caching the join output also reproduced it. The base window serializer rejects this unordered input in the same fixture. Restoring the full (k, s) ordering before the fallback window, or preventing the join from advertising it, would let Spark apply the collation correctly.

[comphead]

Thanks. 1bb942a applies the collation fallback only when the window's input order can be trusted. That is the same check that decides whether the order is declared. Over a doubted order the window stays native, as on main. Your fixture fails with the ordering error again instead of handing the reordered rows to Spark's window. When the order is trusted, Spark's window gets the rows in the order it relies on and applies the collation.

#6785 also makes Spark's window correct in your fixture, since it keeps the nested loop join in Spark.

[sunchao]

This last query passes with the pre-fix native library. On Spark 4.1.3 the join declares a (ordinal 1), and PARTITION BY a asks for the same column, so the window never needs the new right-outer-join branch. I ran the repository test with that library and it passed.

Changing only this final window to PARTITION BY k and its expected declaration below to 0 Ascending NullsFirst preserves the expected rows. That version fails with the old library (All partition by columns should have an ordering) and passes with this commit’s library. It covers the end-to-end case where the window needs an equivalent key that the join did not declare.

[comphead]

Thanks, applied in 1bb942a. The last window now partitions by k and expects 0 Ascending NullsFirst. It needs its order declared next to the join's a, and the test comment says why.

[sunchao]

Reviewed 3e991c7 across five independent scopes, including an executed Spark 4.1.3 probe using the head native artifact and updated window serializer. The revised numeric/partitioned cases and outer-join regression are addressed. The narrowed collation fallback still has a wrong-result case for an unpartitioned window. The underlying nested-loop ordering defect is pre-existing; the new guard relies on a native ordering rejection that does not occur for this shape.

[sunchao]

The native window does not necessarily reject an untrusted order: with no PARTITION BY, there are no partition-order requirements to fail. This now also skips the collation fallback. I cached one sorted partition of 12 rows with v = 0..11 and s cycling through A, a, B, b COLLATE UTF8_LCASE, broadcast z = 0,1, joined on v % 2 <= z, and selected dense_rank() OVER (ORDER BY s) from the 18 joined rows. Spark gives ranks 1 and 2. This head runs CometWindowExec and gives ranks 1 through 14, including ranks 1 and 3 for identical A values. A control with a physical sort above the same native join matches Spark.

The join-order problem is already known; simply restoring the unconditional Spark fallback also leaves the rows unsorted (that control gives four ranks). Could this guard handle the unpartitioned case explicitly, preserving both ordering and collation or rejecting the unsupported shape, and cover it in the collated-window regression? The current partitioned/count fixture cannot expose it.