This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 5885bed51323 [SPARK-59899][SQL] Restrict DSv2 scan output ordering to
sort orders that hold over the scan output
5885bed51323 is described below
commit 5885bed51323a9dfff3567f796ab1e2f589205bc
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Sep 30 14:25:57 2026 -0700
[SPARK-59899][SQL] Restrict DSv2 scan output ordering to sort orders that
hold over the scan output
### What changes were proposed in this pull request?
This PR aims to make `DataSourceV2ScanExecBase.outputOrdering` keep only
the sort orders that hold over the scan output when a DSv2 source reports an
ordering on a column pruned out of the scan output. It keeps the leading sort
orders over output columns and, for a `KeyedPartitioning`, the partition key
sort orders after them, since each input partition holds a single key.
### Why are the changes needed?
`V2ScanPartitioningAndOrdering` converts the reported ordering against the
relation before column pruning, so the ordering can reference pruned columns. A
row-based scan is safe because its `ProjectExec` truncates such an ordering,
but a columnar scan can reach the k-way merge of `GroupPartitionsExec` without
one, and the merge fails.
```
[INTERNAL_ERROR_ATTRIBUTE_NOT_FOUND] Could not find ts#11 in
[id#9,data#10]. SQLSTATE: XX000
```
### Does this PR introduce _any_ user-facing change?
Yes. This PR fixes a query failure when
`spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled` (default:
`false`) is enabled. In addition, a partition key sort order that follows a
pruned column in the reported ordering now reaches the operators above the
scan, so a query can skip an unnecessary sort, e.g., before a sort-merge join
on the partition key. The query results are the same.
### How was this patch tested?
Pass the CIs with the newly added test case.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59163 from dongjoon-hyun/SPARK-59899.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 3d5202becd166410a7fff78c79084c6e9d65d6b1)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../datasources/v2/DataSourceV2ScanExecBase.scala | 19 ++++--
.../v2/V2ScanPartitioningAndOrdering.scala | 4 ++
.../connector/KeyGroupedPartitioningSuite.scala | 67 ++++++++++++++++++++++
3 files changed, 86 insertions(+), 4 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala
index cde8b1041a95..5c9c38ea7285 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala
@@ -19,7 +19,7 @@ package org.apache.spark.sql.execution.datasources.v2
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.{Ascending, Expression,
RowOrdering, SortOrder}
+import org.apache.spark.sql.catalyst.expressions.{Ascending, Expression,
ExpressionSet, RowOrdering, SortOrder}
import org.apache.spark.sql.catalyst.plans.physical
import org.apache.spark.sql.catalyst.plans.physical.KeyedPartitioning
import org.apache.spark.sql.catalyst.util.truncatedString
@@ -107,15 +107,26 @@ trait DataSourceV2ScanExecBase
/**
* Returns the output ordering for this scan. When the source reports
ordering via
- * `SupportsReportOrdering`, that ordering is returned as-is. Otherwise,
when the output
- * partitioning is a `KeyedPartitioning` and
+ * `SupportsReportOrdering`, that ordering may reference columns pruned out
of the scan output
+ * (see V2ScanPartitioningAndOrdering), while a consumer binds it against
the output, e.g. the
+ * k-way merge of `GroupPartitionsExec`. Ordering is prefix-based, so the
leading run of sort
+ * orders over the output is kept and the rest is dropped, except the sort
orders on a partition
+ * key when the output partitioning is a `KeyedPartitioning`: each partition
holds a single key,
+ * so those still hold. Otherwise, when the output partitioning is a
`KeyedPartitioning` and
* `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` is on, each
partition
* contains rows where the key expressions evaluate to a single constant
value, so the data
* is trivially sorted by those expressions within the partition.
*/
override def outputOrdering: Seq[SortOrder] = {
(ordering, outputPartitioning) match {
- case (Some(o), _) => o
+ case (Some(o), p) =>
+ val (prefix, rest) = o.span(_.references.subsetOf(outputSet))
+ p match {
+ case k: KeyedPartitioning if rest.nonEmpty =>
+ val keyExprs = ExpressionSet(k.expressions)
+ prefix ++ rest.filter(order => keyExprs.contains(order.child))
+ case _ => prefix
+ }
case (_, k: KeyedPartitioning) if
conf.v2BucketingPartitionKeyOrderingEnabled =>
k.expressions.map(SortOrder(_, Ascending))
case _ => Seq.empty
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala
index 479154e7a285..b35d250c37df 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala
@@ -70,6 +70,10 @@ object V2ScanPartitioningAndOrdering extends
Rule[LogicalPlan] with Logging {
private def ordering(plan: LogicalPlan) = plan.transformDown {
case d @ ExtractV2ScanInfo(relation, scan: SupportsReportOrdering, _) =>
+ // The ordering is kept as reported, even where it references columns
pruned out of the scan
+ // output: truncating it here would also drop the sort orders on a
partition key past a
+ // pruned column, which still hold.
`DataSourceV2ScanExecBase.outputOrdering` restricts it
+ // to the scan output instead.
val ordering =
V2ExpressionUtils.toCatalystOrdering(scan.outputOrdering(), relation,
relation.funCatalog)
d.copy(ordering = Some(ordering))
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
index 5f58b74875b1..43a6a9d675ee 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
@@ -2632,6 +2632,73 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
}
}
+ test("SPARK-59899: a scan's reported ordering keeps only partition key sort
orders past a " +
+ "pruned column") {
+ // The source reports its ordering over the full table, so a sort order on
a column pruned out
+ // of the scan output reaches the scan, as a pruned partition key does.
Ordering is
+ // prefix-based, so the scan keeps the leading run of sort orders over its
output: `t1` drops
+ // its last order, `t2` a middle one along with every order after it. Each
split holds a single
+ // key, so a sort order on the partition key still holds past that run:
`t3` keeps `id`.
+ val table1 = "prune_order_t1"
+ val table2 = "prune_order_t2"
+ val table3 = "prune_order_t3"
+ def asc(col: String): SortOrder =
+ sort(FieldReference(col), SortDirection.ASCENDING,
NullOrdering.NULLS_FIRST)
+ createTable(table1, columns, Array(identity("id")), Array(asc("id"),
asc("data"), asc("ts")))
+ createTable(table2, columns, Array(identity("id")), Array(asc("id"),
asc("ts"), asc("data")))
+ createTable(table3, columns, Array(identity("id")), Array(asc("ts"),
asc("id")))
+ Seq(table1, table2, table3).foreach { table =>
+ // Key 1 is stored in two splits, so grouping the scan by key merges
them.
+ sql(s"INSERT INTO testcat.ns.$table VALUES " +
+ "(1, 'bb', cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', cast('2020-01-02' as timestamp)), " +
+ "(2, 'cc', cast('2020-01-03' as timestamp))")
+ }
+
+ // The positions, in the reported ordering, of the sort orders each scan
keeps.
+ Seq(table1 -> Seq(0, 1), table2 -> Seq(0), table3 -> Seq(1)).foreach {
case (table, kept) =>
+ // The scan relation references `ts` through its ordering, which the
plan change validation
+ // flags as dangling without SPARK-40259.
+ val scan = withSQLConf(SQLConf.PLAN_CHANGE_VALIDATION.key -> "false") {
+ collectScans(sql(s"SELECT id, data FROM testcat.ns.$table")
+ .queryExecution.executedPlan).head
+ }
+ val reported = scan.ordering.get
+ assert(reported.exists(!_.references.subsetOf(scan.outputSet)),
+ s"test setup: the ordering of $table is on `ts`, which is pruned out
of the scan")
+ assert(scan.outputOrdering === kept.map(reported))
+
+ // Without the merge, coalescing the two splits of key 1 keeps only the
sort order on the
+ // partition key, which is constant within each group.
+
withSQLConf(SQLConf.V2_BUCKETING_PRESERVE_KEY_ORDERING_ON_COALESCE_ENABLED.key
-> "true") {
+ assert(GroupPartitionsExec(scan).outputOrdering ===
+ reported.filter(_.child.semanticEquals(scan.output.head)))
+ }
+
+ // A k-way merge binds the scan's ordering against the scan output. A
row-based scan like this
+ // one always gets a `ProjectExec`, which truncates the ordering, but a
columnar scan may get
+ // nothing between it and the merge that does. So the merge is built
right over the scan here.
+ val merged = GroupPartitionsExec(scan, enableSortedMerge =
true).execute()
+ assert(merged.isInstanceOf[SortedMergeCoalescedRDD[_]])
+ val rows = merged.map(r => (r.getInt(0), r.getString(1))).collect().toSeq
+ val expected = Seq((1, "aa"), (1, "bb"), (2, "cc"))
+ // Only `t1` orders the two splits of key 1 past the key, by `data`.
+ assert((if (table == table1) rows else rows.sorted) === expected)
+ }
+
+ // A row-based scan's `ProjectExec` passes the kept `(id)` through, so a
sort-merge join on the
+ // partition key needs no sort on either side.
+
withSQLConf(SQLConf.V2_BUCKETING_PRESERVE_KEY_ORDERING_ON_COALESCE_ENABLED.key
-> "true") {
+ val df = sql(s"SELECT a.id, a.data, b.data FROM testcat.ns.$table3 a " +
+ s"JOIN testcat.ns.$table3 b ON a.id = b.id")
+ val smjs = collect(df.queryExecution.executedPlan) { case s:
SortMergeJoinExec => s }
+ assert(smjs.size == 1)
+ assert(smjs.head.children.flatMap(collect(_) { case s: SortExec => s
}).isEmpty)
+ checkAnswer(df, Seq(Row(1, "aa", "aa"), Row(1, "aa", "bb"), Row(1, "bb",
"aa"),
+ Row(1, "bb", "bb"), Row(2, "cc", "cc")))
+ }
+ }
+
test("SPARK-47094: SPJ: Support compatible buckets") {
val table1 = "tab1e1"
val table2 = "table2"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
