[ykisana]

What changes were proposed in this pull request?

V2ScanPartitioningAndOrdering resolves a scan's SupportsReportOrdering ordering against the full DataSourceV2Relation. For a nested sort key such as s.y, the result is GetStructField(s, 1) over the unpruned type s: struct<x, y>. When nested schema pruning narrows the scan output to s: struct<y>, the attribute keeps its ExprId but changes type, and y moves to ordinal 0. DataSourceV2ScanExecBase.outputOrdering checks the ordering against the output only by ExprId, so the stale expression is kept and never matches a required ordering on the pruned s.

This PR binds the reported ordering to the scan output in DataSourceV2ScanExecBase before outputOrdering uses it. It uses ProjectionOverSchema, which rebinds the attribute to its pruned type and recomputes each field ordinal by name, the same way DataSourceV2ScanRelation.outputBoundPushedFilters remaps pushed filters. Attribute names are first normalized to the output's by ExprId, since ProjectionOverSchema matches attributes by name and a source may report a key such as S.y for the column s. A field of an array of structs, such as arr.Y, takes its name from the output too, since that name takes part in semanticEquals. A sort order on a nested field that the scan no longer reads (FIELD_NOT_FOUND) is treated like a sort order on a pruned column and ends the kept prefix. So is a sort order whose own type pruning narrows, such as one on the struct s itself or on s.inner: rows sorted by a struct are not necessarily sorted by a narrower one. Since the ordering is only an optimization, any other failure to bind, such as the internal error ProjectionOverSchema raises for a source whose readSchema() changes a struct column's type, drops the sort order as well instead of failing planning. The ordering derived from the partition keys is bound the same way, so a narrowed struct partition key no longer yields a sort order over the unpruned type. The docs of spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled and the migration guide now list both cases. Both orderings are bound once, inside the keptOrdering and derivedOrdering lazy vals, like outputPartitioning, because the planner asks for outputOrdering many times.

The logical DataSourceV2ScanRelation.ordering is unchanged and stays in relation space, which PlanMerger relies on to compare two scans of the same table.

Why are the changes needed?

Follow-up to SPARK-59948, which removed the redundant sort only when the scan reads the whole struct. With s struct<x: bigint, y: bigint> and a reported ordering on s.y:

SELECT id, s.y FROM t SORT BY s.y

plans a redundant sort above a scan that already reports that ordering:

Sort [_extract_y#23L ASC NULLS FIRST], false, 0
+- Project [id#13L, s#14.y AS y#15L, s#14.y AS _extract_y#23L]
   +- BatchScan testcat.ns.nested_items[id#13L, s#14] ...

Found by @peter-toth while reviewing #59199.

Does this PR introduce any user-facing change?

No. Query results are unchanged. For DSv2 sources that report an ordering on a nested field, the physical plan no longer contains a redundant SortExec when nested schema pruning narrows the struct.

How was this patch tested?

Added tests to KeyGroupedPartitioningSuite. A table with s: struct<x, y> reports an ordering on s.y, or on S.y, with all rows in one split.

• SELECT id, s.y ... SORT BY s.y has no SortExec and returns the rows in order. Without the fix, the plan contains the SortExec above.
• SELECT id, s.x ... SORT BY s.x, where pruning drops y, reports no scan ordering, still plans a sort and returns the rows in order of x.
• With a reported ordering on the struct s or on s.inner (also in upper case), with and without identity(s) partitioning, the scan reports no ordering when pruning narrows that struct (SELECT id, s.inner.b), and keeps it when the scan reads the whole struct (SELECT id, s).
• With identity(s) partitioning and no reported ordering, the scan derives [s] when it reads s whole, and nothing when pruning narrows s.
• A kept two-level key s.inner.b and an arr.y or arr.Y key over arr: array<struct<x, y>> remove the sort, and the scan reports no ordering when pruning drops the field.
• A source whose readSchema() returns the struct s as a string and that reports an ordering on s.x plans SELECT count(s) with no scan ordering instead of failing with INTERNAL_ERROR.
• With identity(id) partitioning, a reported [s.y] or [s.y, id] gives [id] when pruning drops y.
• A storage-partitioned sort-merge join on (id, s.y) over two splits of id 1 written out of s.y order plans one k-way merge and no SortExec, and returns every match.

Also ran KeyGroupedPartitioningSuite, MergeSubplansSuite, WriteDistributionAndOrderingSuite and DataSourceV2Suite.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Opus 5.5

Fix written by and all changes reviewed by @ykisana. Tests, comments, pr desc by Claude Opus.

[peter-toth]

Thanks for the PR, @ykisana!

The PR binds the scan's reported ordering to the scan output in DataSourceV2ScanExecBase. It uses ProjectionOverSchema, the same way outputBoundPushedFilters remaps pushed filters. So a sort order on s.y keeps matching a required ordering after nested schema pruning narrows s. The logical ordering stays in relation space, so PlanMerger is unaffected. I checked that the new test fails without the fix (the SortExec stays) and passes with it. One case still reports an ordering that does not hold, see finding 1.

Blocking

• 1. A narrowed struct sort key is still reported: a sort order on a struct that pruning narrows (s itself, or s.inner) is rebound to the narrowed type and kept. That claims an order the rows don't have, and the new scaladoc says it should give None. Nothing uses the claim today, but a scan must not report an ordering that doesn't hold. The fix is to drop a sort order whose type changed, plus a test that pins it. inline: DataSourceV2ScanExecBase.scala:186

Minor

• 2. Stale comment: V2ScanPartitioningAndOrdering.scala:107-110 still limits the match to "when the scan reads the whole struct" and points at SPARK-60011 as the open gap. inline: DataSourceV2ScanExecBase.scala:181

[ykisana]

Thanks for the PR, @ykisana!

The PR binds the scan's reported ordering to the scan output in DataSourceV2ScanExecBase. It uses ProjectionOverSchema, the same way outputBoundPushedFilters remaps pushed filters. So a sort order on s.y keeps matching a required ordering after nested schema pruning narrows s. The logical ordering stays in relation space, so PlanMerger is unaffected. I checked that the new test fails without the fix (the SortExec stays) and passes with it. One case still reports an ordering that does not hold, see finding 1.

Blocking

* **1.** _A narrowed struct sort key is still reported:_ a sort order on a struct that pruning narrows (`s` itself, or `s.inner`) is rebound to the narrowed type and kept. That claims an order the rows don't have, and the new scaladoc says it should give `None`. Nothing uses the claim today, but a scan must not report an ordering that doesn't hold. The fix is to drop a sort order whose type changed, plus a test that pins it. [inline: `DataSourceV2ScanExecBase.scala:186`](https://github.com/apache/spark/pull/59387#discussion_r4294284334)

Minor

* **2.** _Stale comment:_ [`V2ScanPartitioningAndOrdering.scala:107-110`](https://github.com/apache/spark/blob/28d5adb4e62621af1b27504600cebd665a4a932f/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala#L107-L110) still limits the match to "when the scan reads the whole struct" and points at [SPARK-60011](https://issues.apache.org/jira/browse/SPARK-60011) as the open gap. [inline: `DataSourceV2ScanExecBase.scala:181`](https://github.com/apache/spark/pull/59387#discussion_r4294284360)

@peter-toth addressed both of these

[peter-toth]

Re-checked through 1c26a916817. Findings 1 and 2 are resolved. A sort order whose type pruning narrows now gives None, and the comment in V2ScanPartitioningAndOrdering is updated. The new test fails on the base, and also with either the sameType check or the FIELD_NOT_FOUND catch removed. Nothing regressed, and the PR merges cleanly onto master.

Minor

• 3. The outputOrdering scaladoc still says [s] derives the key ordering only when s is pruned (new): a narrowed s derives it too now. inline: DataSourceV2ScanExecBase.scala:180
• 4. The description calls the bound ordering a lazy val (new): outputBoundOrdering is a def, computed once inside the keptOrdering lazy val.

[dongjoon-hyun]

Thank you for the fix, @ykisana. I reviewed the latest commit, 1c26a91, and left 10 inline comments. Rebinding the reported ordering through ProjectionOverSchema fixes the s.y case in the description, and I found no case where the scan now claims an ordering that the rows do not have. Here is a summary, ordered by priority.

Correctness

1. DataSourceV2ScanExecBase L196: A reported key whose root name differs in case from the column, such as S.y for s, is not rebound. The stale sort order is kept, so the redundant SortExec stays.
2. DataSourceV2ScanExecBase L190: This is an existing bug outside this diff. outputBoundPushedFilters, which this code mirrors, has no narrowing check, so a fully pushed predicate on a whole struct is re-typed after nested schema pruning.
3. DataSourceV2ScanExecBase L202: Only FIELD_NOT_FOUND is caught. An internal error from ProjectionOverSchema now fails planning when a connector reads a struct column as another type.
4. DataSourceV2ScanExecBase L181: derivedOrdering is not rebound, so a narrowed struct partition key still falls back to a sort order over the unpruned type.

Tests

5. KeyGroupedPartitioningSuite L4193: No test with a partitioned table pins the plan changes for key-grouped scans.
6. KeyGroupedPartitioningSuite L4206: The prunedDf check also passes on the base, and with one row per split neither checkAnswer can see the ordering.
7. KeyGroupedPartitioningSuite L4219: A kept two-level key and the GetArrayStructFields path are not covered.

Docs and comments

8. DataSourceV2ScanExecBase L187-188: "rows sorted by a struct are not sorted by a narrower one" is too strong. The PR description has the same sentence.

Cleanup

9. DataSourceV2ScanExecBase L192: This is the third copy of the ProjectionOverSchema and FIELD_NOT_FOUND idiom.
10. DataSourceV2ScanExecBase L193: The projection is built for every scan, even when no ordering is reported.

Item 1 is the most important one for this PR, and items 5 and 6 would pin the new behavior. Item 2 predates this PR, so a separate JIRA may suit it better. The rest are minor.

[ykisana]

Thank you for the fix, @ykisana. I reviewed the latest commit, 1c26a91, and left 10 inline comments. Rebinding the reported ordering through ProjectionOverSchema fixes the s.y case in the description, and I found no case where the scan now claims an ordering that the rows do not have. Here is a summary, ordered by priority.

Correctness

1. `DataSourceV2ScanExecBase` L196: A reported key whose root name differs in case from the column, such as `S.y` for `s`, is not rebound. The stale sort order is kept, so the redundant `SortExec` stays.

2. `DataSourceV2ScanExecBase` L190: This is an existing bug outside this diff. `outputBoundPushedFilters`, which this code mirrors, has no narrowing check, so a fully pushed predicate on a whole struct is re-typed after nested schema pruning.

3. `DataSourceV2ScanExecBase` L202: Only `FIELD_NOT_FOUND` is caught. An internal error from `ProjectionOverSchema` now fails planning when a connector reads a struct column as another type.

4. `DataSourceV2ScanExecBase` L181: `derivedOrdering` is not rebound, so a narrowed struct partition key still falls back to a sort order over the unpruned type.

Tests

5. `KeyGroupedPartitioningSuite` L4193: No test with a partitioned table pins the plan changes for key-grouped scans.

6. `KeyGroupedPartitioningSuite` L4206: The `prunedDf` check also passes on the base, and with one row per split neither `checkAnswer` can see the ordering.

7. `KeyGroupedPartitioningSuite` L4219: A kept two-level key and the `GetArrayStructFields` path are not covered.

Docs and comments

8. `DataSourceV2ScanExecBase` L187-188: "rows sorted by a struct are not sorted by a narrower one" is too strong. The PR description has the same sentence.

Cleanup

9. `DataSourceV2ScanExecBase` L192: This is the third copy of the `ProjectionOverSchema` and `FIELD_NOT_FOUND` idiom.

10. `DataSourceV2ScanExecBase` L193: The projection is built for every scan, even when no ordering is reported.

Item 1 is the most important one for this PR, and items 5 and 6 would pin the new behavior. Item 2 predates this PR, so a separate JIRA may suit it better. The rest are minor.

Updated, 2 and 9 can be follow ups. For 9 might be a more involved refactor.

[ykisana]

Follow up tickets from @dongjoon-hyun comments:
https://issues.apache.org/jira/browse/SPARK-60410
https://issues.apache.org/jira/browse/SPARK-60411

[dongjoon-hyun]

Thank you for addressing items 1-10, @ykisana. I reviewed the latest commit, 7683fa2, and left 7 inline comments, continuing the numbering from items 1-10. Items 1, 3-8 and 10 are resolved, and items 2 and 9 are tracked in SPARK-60410 and SPARK-60411. I found no case where the scan now claims an ordering that the rows do not have. Here is a summary, ordered by priority.

Correctness

11. DataSourceV2ScanExecBase L210: The normalizeExprs call that I suggested in item 1 renames only attributes, so a reported array field in another case, such as arr.Y, still leaves the redundant SortExec.

Tests

12. DataSourceV2ScanExecBase L218: No test reaches the NonFatal catch through anything but FIELD_NOT_FOUND.
13. KeyGroupedPartitioningSuite L4234: The kept ordering already meets the nonEmpty check, so no test pins the derived ordering on a struct key that the scan reads whole.

Docs and comments

14. DataSourceV2ScanExecBase L206: "Mirrors outputBoundPushedFilters" no longer holds.
15. DataSourceV2ScanExecBase L159: The docs of spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled and the migration guide do not cover the two cases that nested schema pruning adds.

Cleanup

16. DataSourceV2ScanExecBase L223: The normalizeExprs call rebuilds an AttributeMap over the whole output on every bindToOutput call.
17. KeyGroupedPartitioningSuite L4200: The new tests define the same columns three times, shadow the class-level columns, and have two different local scan helpers.

All of these are minor. Item 11 is a gap in my own suggestion for item 1, and items 12 and 13 would pin the new NonFatal catch and the rebinding of the derived ordering.

[dongjoon-hyun]

Minor: The normalizeExprs call that I suggested in item 1 renames only attributes. So it fixes a mixed-case root such as S.y, but not a mixed-case field of an array of structs. For a reported arr.Y, ExtractValue keeps the reported spelling in the field (fields(ordinal).copy(name = fieldName), complexTypeExtractors.scala:104), and ProjectionOverSchema copies it back (prunedField.copy(name = a.field.name), ProjectionOverSchema.scala:56). GetArrayStructFields has no canonicalized override, so the field name takes part in semanticEquals, and the bound arr.Y never matches a required arr.y. A struct field is fine, because GetStructField.canonicalized drops name (complexTypeExtractors.scala:222-223), which is why S.y passes. So with ascending("arr.Y") on the arrays table, SELECT id, arr.y ... SORT BY arr.y should still plan a SortExec. I traced this by reading the code and did not run it. It is only a missed optimization, and the same gap exists when the scan reads arr whole. Could you add an arr.Y case, and either take the field name from the output schema after the projection or leave a comment for a follow-up?

[dongjoon-hyun]

Minor: No test reaches this catch through anything but FIELD_NOT_FOUND. Changing it back to the round-1 catch, case e: SparkIllegalArgumentException if e.getCondition == "FIELD_NOT_FOUND" => None, still passes the suite. So a later narrowing would bring back the INTERNAL_ERROR from item 3 (ProjectionOverSchema.scala:61 and :109) without a test failing. CustomReportingScan already delegates readSchema() (KeyGroupedPartitioningSuite.scala:10654). Could you let CustomReportingCatalog return s as a string there, report an ordering on s.x, and check that the scan of SELECT count(s) reports no ordering instead of throwing?

[dongjoon-hyun]

Minor: For the identity(s) tables (j = 1), the kept [s] or [s.inner] already meets this nonEmpty check, since a kept sort order wins over the derived one. So no test pins that the new flatMap(bindToOutput) in derivedOrdering still derives [s ASC] for a struct key that the scan reads whole. A bug that drops every struct-typed key there would still pass: test 2 and the SPARK-59981 tests use only id keys, and the struct-key SPJ tests from L3105 check shuffles and answers, not the ordering. The comment at L4219, "The partitioned tables derive an ordering on s", also describes nothing that these assertions check. Could you add an identity(s) table with no reported ordering and assert orderingNames(scan(table, "s")) == Seq((Seq("s"), Ascending))?

[dongjoon-hyun]

Nit: "Mirrors DataSourceV2ScanRelation.outputBoundPushedFilters" no longer holds. This method normalizes the names by exprId, drops a narrowed value with sameType, and drops any NonFatal failure. outputBoundPushedFilters (DataSourceV2Relation.scala:298-310) does none of these, and its missing narrowing check is item 2. A reader who trusts the word may take that method as safe for narrowed structs, or copy its FIELD_NOT_FOUND-only catch back here. Could you say something like "Like outputBoundPushedFilters, but it also normalizes the names, drops a narrowed value and drops any failure"?

[dongjoon-hyun]

Minor: This sentence covers the two new cases, but the user-facing docs of spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled do not. SQLConf.scala:2682-2689 and docs/sql-migration-guide.md:49 list when the key ordering is derived ("starts with a sort order on a column pruned from the scan output or on a partition transform") and which keys are left out ("Partition transforms such as days(ts) or bucket(8, id)"). With this PR, a report that starts with a sort order on a nested field that pruning drops, or on a struct that pruning narrows, also falls back to the key ordering. A key such as identity(s) whose struct pruning narrows is now left out of it. So a reader cannot tell why SELECT id, s.inner.b on an identity(s) table reports no ordering while SELECT id, s does. Could you mention both cases there?

[dongjoon-hyun]

Nit: outputProjection is built once, but the normalizeExprs call that I suggested in item 1 builds a new AttributeMap over the whole output on every call (DataSourceStrategy.scala:578), so once per reported sort order and once per partition key. The cost is tiny, as in item 10. Could you build the map once next to this lazy val, e.g. @transient private lazy val outputNames = AttributeMap(output.map(a => a -> a.name)), and rename with e.transform { case a: AttributeReference => a.withName(outputNames.getOrElse(a, a.name)) } at L210?

[dongjoon-hyun]

Nit: The same id and s: struct<x, y> columns are defined three times (here, L4253 and nestedItemsColumns in the k-way merge test), and elements at L4244 repeats the struct. Both columns vals shadow the class-level columns at L425, and the columns parameter of scan at L4223 shadows this one too. There are also two local scan helpers with different signatures (L4223 and L4255), and collect(...) { case s: SortExec => s }.isEmpty appears four times. Could you hoist one columns val and one scan helper next to ascending?