Skip to content

Commit 176af6d

Browse files
peter-tothdongjoon-hyun
authored andcommitted
[SPARK-59995][4.3][SQL] Drop sort orders that hold a partition transform from a V2 scan's output ordering
### Differences from #59251 This is the branch-4.3 backport of #59251. The description below is the original one. - The bug is on branch-4.3. Without the fix, both new k-way merge tests fail with the `Cannot generate code for expression` error. - `docs/sql-migration-guide.md` is dropped. Its hunks edit the 4.4 notes on the `partitionKeyOrdering` and `preserveKeyOrderingOnCoalesce` default changes (SPARK-59396), which are not on 4.3. - `docs/sql-performance-tuning.md` is dropped. The config table on 4.3 has no rows for these configs. - `BoundFunction.java` is dropped. The Javadoc list it edits is not on 4.3. - `SupportsReportOrdering.java` only gets the new paragraph about transforms. The paragraph before it on master comes from a later change that is not on 4.3. - In `SQLConf.scala`, the `partitionKeyOrdering` doc gets the new sentence on top of the 4.3 text. The 4.3 text has no clause about an ignored reported ordering. - In `DataSourceV2ScanExecBase.scala`, the code change is the same. The Scaladoc keeps the 4.3 first paragraph and adds the new one. - `GroupPartitionsExec.scala` and `WriteDistributionAndOrderingSuite.scala` apply cleanly. - In `GroupPartitionsExecSuite.scala`, only the imports conflicted. The new test is the same. - `KeyGroupedPartitioningSuite.scala` adds the `SortAggregateExec` import, which 4.3 does not have. - The derived k-way merge test also sets `spark.sql.requireAllClusterKeysForCoPartition` to `false`. On 4.3 a join on a subset of the partition keys needs it. Without it the plan does not k-way merge. - The other new tests already set the configs they need. `partitionKeyOrdering`, `preserveKeyOrderingOnCoalesce` and `preserveOrderingOnCoalesce` are all off by default on 4.3. - Without the fix, the 3 new `KeyGroupedPartitioningSuite` tests and the adjusted SPARK-56321 test fail. For this run `DataSourceV2ScanExecBase.scala` and `GroupPartitionsExec.scala` were restored from branch-4.3. - With the scan change kept and `LazyRowOrdering` built with `GenerateOrdering` again, the new `GroupPartitionsExecSuite` test fails. - Ran `KeyGroupedPartitioningSuite` (188), `GroupPartitionsExecSuite` (23), `EnsureRequirementsSuite` (58), `WriteDistributionAndOrderingSuite` (99), `DataSourceV2Suite` (69) and `ProjectedOrderingAndPartitioningSuite` (36). All 473 tests pass. `TransformExpressionSuite` is not on 4.3. - `dev/lint-scala` passes. ### What changes were proposed in this pull request? `DataSourceV2ScanExecBase.outputOrdering` now drops every sort order that holds a partition transform (`TransformExpression`): - in a reported ordering, a transform ends the leading run of sort orders the scan keeps, like a sort order over a pruned column; - a transform on a partition key is dropped too, while a sort order on another partition key still holds after it, since each partition holds a single key; - in an ordering the scan derives from its partition keys, the transform keys are left out. Dropping them loses nothing today when Spark can call the transform's function, since no operator then requires an ordering over the transform. The write path sorts by the function call instead (`DistributionAndOrderingUtils`). Dropping them is also a safe way to handle a transform Spark cannot evaluate, and it saves comparisons nobody uses. If an ordering over a transform becomes a real requirement, this needs to be revisited. The k-way merge of `GroupPartitionsExec` now also builds its comparator with `RowOrdering.create`, like `SortExec`. It still generates code, and falls back to interpreted evaluation when that fails. It is still built on the executor, at the first comparison. `LazyCodeGenOrdering` is renamed `LazyRowOrdering` to match. The docs of `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled`, its 4.4 migration note and the `SupportsReportOrdering` Javadoc now say that partition transforms are left out. The `BoundFunction` Javadoc now names the scan merge, instead of a retained reported ordering, as what a semantic `equals` is needed for. ### Why are the changes needed? With `spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled` on, `GroupPartitionsExec` can coalesce partitions with a k-way merge over its child's ordering (SPARK-55715). The merge compares rows with generated code (`LazyCodeGenOrdering`). A `TransformExpression` generates no code. So when the ordering contained a partition transform, such as `years(arrive_time)`, the task failed: ``` [INTERNAL_ERROR] Cannot generate code for expression: transformexpression(org.apache.spark.sql.connector.catalog.functions.YearsFunction$..., input[2, timestamp, true], None) SQLSTATE: XX000 ``` The ordering can contain a transform in two ways: - the source reports it via `SupportsReportOrdering`, e.g. `[id, name, years(arrive_time)]`; - the scan derives it from its partition keys (SPARK-56241), e.g. `[id, years(arrive_time)]`. ### Does this PR introduce _any_ user-facing change? Yes. - A query that k-way merges over a transform used to fail with the error above. Now it returns its rows. The merge still runs when the sort orders the scan keeps satisfy the parent. - A sort order on a partition key after a transform can now lead the scan's ordering. For example, with partition keys `[years(ts), id]` the scan reports `[id]` instead of `[years(ts), id]`. So a sort on `id` right above the scan, e.g. from `sortWithinPartitions("id")`, is no longer needed. The new ordering can also turn a hash aggregate into a sort aggregate, since `spark.sql.execution.replaceHashWithSortAgg` keys off the scan's ordering. For example, `SELECT id, max(ts) FROM t GROUP BY id` over the partition keys `[years(ts), id]` now plans its partial aggregate as a `SortAggregateExec`. - The failure exists since 4.2.0, which has both the k-way merge (SPARK-55715) and the key-derived ordering (SPARK-56241). It needs `spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled`, which is off by default on every branch. The derived case also needs `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled`, which defaults to true only on master and branch-4.x. ### How was this patch tested? - New tests in `KeyGroupedPartitioningSuite`: - "a scan's output ordering drops sort orders that hold a partition transform" checks the three cases above, and that a sort on the kept key needs no `SortExec`; - two k-way merge tests, one for each way above. Each checks the answer, that the plan k-way merges, and the ordering it merges over. They fail on master with the error above. The derived one uses `days`, a function Spark cannot call. - New test in `GroupPartitionsExecSuite`: "the k-way merge ordering falls back to interpreted evaluation". It serializes a `LazyRowOrdering` over a transform, which generates no code, and compares rows with it. It fails without the fallback. - The SPARK-56321 test in `WriteDistributionAndOrderingSuite` now checks the reported ordering, and that the output ordering drops its transform. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5.5) Closes #59303 from peter-toth/SPARK-59995-transform-expression-codegen-4.3. Authored-by: Peter Toth <peter.toth@gmail.com> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>
1 parent c465689 commit 176af6d

7 files changed

Lines changed: 192 additions & 31 deletions

File tree

‎sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsReportOrdering.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,11 @@ public interface SupportsReportOrdering extends Scan {
3535

3636
/**
3737
* Returns the order in each partition of this data source scan.
38+
* <p>
39+
* Spark currently does not rely on a sort order over a transform such as {@code days(ts)} to
40+
* avoid a sort, whether or not it is a partition key. Nor does it rely on the sort orders after
41+
* it. The exception is a sort order on a partition key that is not a transform, when Spark uses
42+
* the reported {@code KeyGroupedPartitioning}.
3843
*/
3944
SortOrder[] outputOrdering();
4045
}

‎sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2536,7 +2536,8 @@ object SQLConf {
25362536
.doc("When enabled, Spark derives output ordering from the partition key expressions of " +
25372537
"a V2 data source that reports a KeyedPartitioning but does not report explicit ordering " +
25382538
"via SupportsReportOrdering. Within a single partition all rows share the same key " +
2539-
s"value, so the data is trivially sorted by those expressions. Requires " +
2539+
"value, so the data is trivially sorted by those expressions. Partition transforms such " +
2540+
"as `days(ts)` or `bucket(8, id)` are left out of the ordering. Requires " +
25402541
s"${V2_BUCKETING_ENABLED.key} to be enabled.")
25412542
.version("4.2.0")
25422543
.withBindingPolicy(ConfigBindingPolicy.SESSION)
@@ -2548,9 +2549,10 @@ object SQLConf {
25482549
.doc("When enabled, Spark preserves sort orders over partition key expressions when " +
25492550
"GroupPartitionsExec coalesces multiple input partitions into one output partition. " +
25502551
"Because all merged partitions share the same partition key value, sort orders over " +
2551-
"those key expressions remain valid after the merge. This applies to both key-derived " +
2552-
"ordering (from SupportsReportOrdering) and ordering derived from " +
2553-
s"${V2_BUCKETING_PARTITION_KEY_ORDERING_ENABLED.key}. Requires " +
2552+
"those key expressions remain valid after the merge. This applies to both the ordering " +
2553+
"reported via SupportsReportOrdering and the ordering derived from " +
2554+
s"${V2_BUCKETING_PARTITION_KEY_ORDERING_ENABLED.key}. Sort orders over partition " +
2555+
"transforms such as `days(ts)` or `bucket(8, id)` are left out. Requires " +
25542556
s"${V2_BUCKETING_ENABLED.key} to be enabled.")
25552557
.version("4.2.0")
25562558
.withBindingPolicy(ConfigBindingPolicy.SESSION)

‎sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala‎

Lines changed: 14 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@ package org.apache.spark.sql.execution.datasources.v2
1919

2020
import org.apache.spark.rdd.RDD
2121
import org.apache.spark.sql.catalyst.InternalRow
22-
import org.apache.spark.sql.catalyst.expressions.{Ascending, Expression, ExpressionSet, RowOrdering, SortOrder}
22+
import org.apache.spark.sql.catalyst.expressions.{Ascending, Expression, ExpressionSet, RowOrdering, SortOrder, TransformExpression}
2323
import org.apache.spark.sql.catalyst.plans.physical
2424
import org.apache.spark.sql.catalyst.plans.physical.KeyedPartitioning
2525
import org.apache.spark.sql.catalyst.util.truncatedString
@@ -116,19 +116,29 @@ trait DataSourceV2ScanExecBase
116116
* `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` is on, each partition
117117
* contains rows where the key expressions evaluate to a single constant value, so the data
118118
* is trivially sorted by those expressions within the partition.
119+
*
120+
* Either way, a sort order that holds a partition transform is dropped, even on a partition key.
121+
* In a reported ordering it also ends the leading run, like a sort order over a pruned column.
122+
* Dropping it loses nothing today when Spark can call the transform's function, since no
123+
* operator then requires an ordering over the transform. The write path sorts by the function
124+
* call instead. Keeping such a sort order would add comparisons nobody uses. Dropping it also
125+
* handles a transform Spark cannot evaluate. Revisit this if an ordering over a transform
126+
* becomes a real requirement.
119127
*/
120128
override def outputOrdering: Seq[SortOrder] = {
129+
def holdsTransform(e: Expression): Boolean = e.exists(_.isInstanceOf[TransformExpression])
121130
(ordering, outputPartitioning) match {
122131
case (Some(o), p) =>
123-
val (prefix, rest) = o.span(_.references.subsetOf(outputSet))
132+
val (prefix, rest) =
133+
o.span(order => order.references.subsetOf(outputSet) && !holdsTransform(order.child))
124134
p match {
125135
case k: KeyedPartitioning if rest.nonEmpty =>
126-
val keyExprs = ExpressionSet(k.expressions)
136+
val keyExprs = ExpressionSet(k.expressions.filterNot(holdsTransform))
127137
prefix ++ rest.filter(order => keyExprs.contains(order.child))
128138
case _ => prefix
129139
}
130140
case (_, k: KeyedPartitioning) if conf.v2BucketingPartitionKeyOrderingEnabled =>
131-
k.expressions.map(SortOrder(_, Ascending))
141+
k.expressions.filterNot(holdsTransform).map(SortOrder(_, Ascending))
132142
case _ => Seq.empty
133143
}
134144
}

‎sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala‎

Lines changed: 14 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,6 @@ import org.apache.spark.{Partition, SparkException}
2424
import org.apache.spark.rdd.{CoalescedRDD, PartitionCoalescer, PartitionGroup, RDD, SortedMergeCoalescedRDD}
2525
import org.apache.spark.sql.catalyst.InternalRow
2626
import org.apache.spark.sql.catalyst.expressions._
27-
import org.apache.spark.sql.catalyst.expressions.codegen.GenerateOrdering
2827
import org.apache.spark.sql.catalyst.plans.QueryPlan
2928
import org.apache.spark.sql.catalyst.plans.physical.{IdentityReducer, KeyedPartitioning, KeyReducer, Partitioning, PartitioningCollection, UnknownPartitioning}
3029
import org.apache.spark.sql.catalyst.util.{truncatedString, InternalRowComparableWrapper}
@@ -361,8 +360,8 @@ case class GroupPartitionsExec(
361360
}
362361

363362
/**
364-
* The ordering used by the k-way merge in [[SortedMergeCoalescedRDD]]. The generated comparator
365-
* ([[GenerateOrdering]]) only needs each [[SortOrder]]'s sort key (child, direction, null
363+
* The ordering used by the k-way merge in [[SortedMergeCoalescedRDD]]. The comparator that
364+
* [[RowOrdering]] builds only needs each [[SortOrder]]'s sort key (child, direction, null
366365
* ordering), so `sameOrderExpressions` -- planner-only metadata that would otherwise be
367366
* serialized with the RDD in every task -- is dropped.
368367
*/
@@ -374,7 +373,7 @@ case class GroupPartitionsExec(
374373
sparkContext.emptyRDD
375374
} else if (usesSortedMerge) {
376375
val partitionCoalescer = new GroupedPartitionCoalescer(groupedPartitions.map(_._2))
377-
val rowOrdering = new LazyCodeGenOrdering(kWayMergeOrdering, child.output)
376+
val rowOrdering = new LazyRowOrdering(kWayMergeOrdering, child.output)
378377
new SortedMergeCoalescedRDD[InternalRow](
379378
child.execute(),
380379
groupedPartitions.size,
@@ -426,11 +425,9 @@ case class GroupPartitionsExec(
426425
outputPartitioning match {
427426
case p: Partitioning with Expression
428427
if reducers.isEmpty && conf.v2BucketingPreserveKeyOrderingOnCoalesceEnabled =>
429-
// Without reducers all merged partitions share the same original key value, so the key
430-
// expressions remain constant within the output partition. The child's outputOrdering
431-
// should already be in sync with the partitioning (either reported by the source or
432-
// derived from it in DataSourceV2ScanExecBase), so we only need to keep the sort orders
433-
// whose expression is a partition key expression -- all others are lost by concatenation.
428+
// Without reducers all merged partitions share the same original key value, so the sort
429+
// orders on key expressions still hold. The transform keys match nothing here, since
430+
// `DataSourceV2ScanExecBase.outputOrdering` drops every sort order over a transform.
434431
val keyedPartitionings = p.collect { case k: KeyedPartitioning => k }
435432
val keyExprs = ExpressionSet(keyedPartitionings.flatMap(_.expressions))
436433
child.outputOrdering.filter(order => keyExprs.contains(order.child))
@@ -510,15 +507,15 @@ class GroupedPartitionCoalescer(
510507
}
511508

512509
/**
513-
* A serializable [[Ordering]] for [[InternalRow]] that generates code-compiled comparison logic
514-
* lazily on first use. The [[SortOrder]] expressions and output schema are serialized with the
515-
* RDD; the generated comparator is rebuilt on the executor on first comparison via
516-
* [[GenerateOrdering]].
510+
* A serializable [[Ordering]] for [[InternalRow]]. The [[SortOrder]] expressions and output schema
511+
* are serialized with the RDD. The comparator is built on the executor, at the first comparison.
512+
* [[RowOrdering]] builds it, as for `SortExec`. It tries generated code first and falls back to
513+
* interpreted evaluation when that fails.
517514
*/
518-
private class LazyCodeGenOrdering(
515+
private class LazyRowOrdering(
519516
sortOrders: Seq[SortOrder],
520517
schema: Seq[Attribute]) extends Ordering[InternalRow] with Serializable {
521-
@transient private lazy val generated: Ordering[InternalRow] =
522-
GenerateOrdering.generate(sortOrders, schema)
523-
override def compare(x: InternalRow, y: InternalRow): Int = generated.compare(x, y)
518+
@transient private lazy val ordering: Ordering[InternalRow] =
519+
RowOrdering.create(sortOrders, schema)
520+
override def compare(x: InternalRow, y: InternalRow): Int = ordering.compare(x, y)
524521
}

‎sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala‎

Lines changed: 114 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@ import org.apache.spark.sql.execution.{
4242
SortExec,
4343
SparkPlan,
4444
UnionExec}
45+
import org.apache.spark.sql.execution.aggregate.SortAggregateExec
4546
import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, DataSourceV2ScanRelation, GroupPartitionsExec}
4647
import org.apache.spark.sql.execution.exchange.{ShuffleExchangeExec, ShuffleExchangeLike, ValidateRequirements}
4748
import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, ShuffledHashJoinExec, SortMergeJoinExec}
@@ -5541,6 +5542,119 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase with
55415542
}
55425543
}
55435544

5545+
test("SPARK-59995: a scan's output ordering drops sort orders that hold a partition transform") {
5546+
// `t1` reports a transform in the middle, so the leading run of its ordering stops there.
5547+
// `t2` reports a transform key first. Each split holds a single key, so the sort order on `id`
5548+
// after it still holds. `t3` reports no ordering, so the scan derives one from its keys. In
5549+
// each case a sort on `id` right above the scan needs no `SortExec`. A `GROUP BY id` plans a
5550+
// `SortAggregateExec`. The query selects `ts`, so the transform's column stays in the scan
5551+
// output. Otherwise SPARK-59899's pruned-column handling would give the same orderings even
5552+
// without this fix.
5553+
val table1 = "transform_order_t1"
5554+
val table2 = "transform_order_t2"
5555+
val table3 = "transform_order_t3"
5556+
def asc(expr: Expression): SortOrder = sort(expr, SortDirection.ASCENDING)
5557+
createTable(table1, columns, Array(identity("id")),
5558+
Array(asc(FieldReference("id")), asc(years("ts")), asc(FieldReference("data"))))
5559+
createTable(table2, columns, Array(years("ts"), identity("id")),
5560+
Array(asc(years("ts")), asc(FieldReference("id"))))
5561+
createTable(table3, columns, Array(days("ts"), identity("id")))
5562+
withSQLConf(
5563+
SQLConf.V2_BUCKETING_PARTITION_KEY_ORDERING_ENABLED.key -> "true",
5564+
SQLConf.REPLACE_HASH_WITH_SORT_AGG_ENABLED.key -> "true") {
5565+
Seq(table1, table2, table3).foreach { table =>
5566+
sql(s"INSERT INTO testcat.ns.$table VALUES (1, 'aa', cast('2020-01-01' as timestamp))")
5567+
val df = sql(s"SELECT id, data, ts FROM testcat.ns.$table")
5568+
val scan = collectScans(df.queryExecution.executedPlan).head
5569+
assert(scan.output.exists(_.name == "ts"), s"test setup: ts stays in the output of $table")
5570+
assert(scan.outputOrdering.map(_.child.sql) === Seq("id"), table)
5571+
val sorted = df.sortWithinPartitions("id").queryExecution.executedPlan
5572+
assert(collect(sorted) { case s: SortExec => s }.isEmpty, table)
5573+
val aggregated = sql(s"SELECT id, max(ts) FROM testcat.ns.$table GROUP BY id")
5574+
.queryExecution.executedPlan
5575+
assert(collect(aggregated) { case a: SortAggregateExec => a }.nonEmpty, table)
5576+
}
5577+
}
5578+
}
5579+
5580+
/**
5581+
* Inserts three items, two of them with `id` 1 on their own splits, so grouping the splits by
5582+
* `id` coalesces them. Then joins `items` with `purchases` on `joinCondition` and checks that
5583+
* the plan k-way merges over `expectedOrdering`, the columns the scan's ordering keeps before
5584+
* its transform. The query selects `arrive_time`, so the transform's column stays in the scan
5585+
* output. Otherwise SPARK-59899's pruned-column handling would drop the transform even without
5586+
* this fix.
5587+
*/
5588+
private def checkKWayMergeBeforeTransform(
5589+
joinCondition: String,
5590+
expectedOrdering: Seq[String]): Unit = {
5591+
sql(s"INSERT INTO testcat.ns.$items VALUES " +
5592+
"(1, 'aa', 10.0, cast('2021-01-01' as timestamp)), " +
5593+
"(1, 'ab', 11.0, cast('2022-01-01' as timestamp)), " +
5594+
"(2, 'bb', 20.0, cast('2021-01-01' as timestamp))")
5595+
val df = sql(
5596+
s"""
5597+
|${selectWithMergeJoinHint("i", "p")}
5598+
|i.id, i.name, i.arrive_time
5599+
|FROM testcat.ns.$items i JOIN testcat.ns.$purchases p ON $joinCondition
5600+
|""".stripMargin)
5601+
checkAnswer(df, Seq(
5602+
Row(1, "aa", Timestamp.valueOf("2021-01-01 00:00:00")),
5603+
Row(1, "ab", Timestamp.valueOf("2022-01-01 00:00:00")),
5604+
Row(2, "bb", Timestamp.valueOf("2021-01-01 00:00:00"))))
5605+
val merging = collectAllGroupPartitions(df.queryExecution.executedPlan)
5606+
.filter(_.enableSortedMerge)
5607+
assert(merging.length == 1, "expected one k-way merge")
5608+
assert(merging.head.child.output.exists(_.name == "arrive_time"),
5609+
"test setup: arrive_time stays in the scan output")
5610+
assert(merging.head.child.outputOrdering.map(_.child.sql) == expectedOrdering)
5611+
assert(merging.head.execute().isInstanceOf[SortedMergeCoalescedRDD[_]])
5612+
}
5613+
5614+
test("SPARK-59995: k-way merge over a reported ordering with a partition transform") {
5615+
// The join on (id, name) needs the merge, since the key ordering on id alone is not enough.
5616+
// The scan reports [id, name, years(arrive_time)] and keeps [id, name].
5617+
val itemOrdering = Array(
5618+
sort(FieldReference("id"), SortDirection.ASCENDING),
5619+
sort(FieldReference("name"), SortDirection.ASCENDING),
5620+
sort(years("arrive_time"), SortDirection.ASCENDING))
5621+
createTable(items, itemsColumns, Array(identity("id")), itemOrdering)
5622+
val namedPurchasesColumns = Array(
5623+
Column.create("item_id", LongType),
5624+
Column.create("name", StringType))
5625+
createTable(purchases, namedPurchasesColumns, Array(identity("item_id")))
5626+
sql(s"INSERT INTO testcat.ns.$purchases VALUES (1, 'aa'), (1, 'ab'), (2, 'bb')")
5627+
5628+
withSQLConf(
5629+
SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
5630+
SQLConf.V2_BUCKETING_PRESERVE_ORDERING_ON_COALESCE_ENABLED.key -> "true") {
5631+
checkKWayMergeBeforeTransform("p.item_id = i.id AND p.name = i.name", Seq("id", "name"))
5632+
}
5633+
}
5634+
5635+
test("SPARK-59995: k-way merge over an ordering derived from a partition transform key") {
5636+
// The scan reports no ordering, so it derives [id] from its keys, leaving out
5637+
// days(arrive_time). The merge must not call the transform's function. Spark cannot call this
5638+
// `days` at all, since `DaysFunction` implements neither `invoke` nor `produceResult`. The join
5639+
// on id projects the keys to id. With preserveKeyOrderingOnCoalesce off, the coalesced
5640+
// partitions keep no ordering on id unless they are merged. Before Spark 4.4, a join on a
5641+
// subset of the partition keys also needs requireAllClusterKeysForCoPartition off.
5642+
createTable(items, itemsColumns, Array(identity("id"), days("arrive_time")))
5643+
createTable(purchases, purchasesColumns, Array(identity("item_id")))
5644+
sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
5645+
"(1, 10.0, cast('2021-01-01' as timestamp)), " +
5646+
"(2, 20.0, cast('2021-01-01' as timestamp))")
5647+
5648+
withSQLConf(
5649+
SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
5650+
SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "true",
5651+
SQLConf.V2_BUCKETING_PARTITION_KEY_ORDERING_ENABLED.key -> "true",
5652+
SQLConf.V2_BUCKETING_PRESERVE_ORDERING_ON_COALESCE_ENABLED.key -> "true",
5653+
SQLConf.V2_BUCKETING_PRESERVE_KEY_ORDERING_ON_COALESCE_ENABLED.key -> "false") {
5654+
checkKWayMergeBeforeTransform("p.item_id = i.id", Seq("id"))
5655+
}
5656+
}
5657+
55445658
test("SPARK-56549: k-way merge enabled only when parent requires ordering") {
55455659
// Dynamic gate: with the config enabled, k-way merge must be activated only when the parent
55465660
// actually requires ordering (SMJ), and must stay off when the parent does not (hash join).

‎sql/core/src/test/scala/org/apache/spark/sql/connector/WriteDistributionAndOrderingSuite.scala‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1545,10 +1545,12 @@ class WriteDistributionAndOrderingSuite extends DistributionAndOrderingSuiteBase
15451545
val df = sql("SELECT id, data FROM testcat.ns1.test_table")
15461546
val scans = collect(df.queryExecution.executedPlan) { case s: BatchScanExec => s }
15471547
assert(scans.size === 1)
1548-
val ordering = scans.head.outputOrdering
1548+
val ordering = scans.head.ordering.getOrElse(Seq.empty)
15491549
assert(ordering.nonEmpty,
1550-
"scan should report non-empty outputOrdering via SupportsReportOrdering")
1550+
"scan should keep the ordering reported via SupportsReportOrdering")
15511551
assert(ordering.head.child.isInstanceOf[TransformExpression],
15521552
"bucket-based sort order should resolve to a TransformExpression")
1553+
// SPARK-59995: the output ordering drops the sort order over the transform.
1554+
assert(scans.head.outputOrdering.isEmpty)
15531555
}
15541556
}

0 commit comments

Comments
 (0)