peter-toth commented on code in PR #58543:
URL: https://github.com/apache/spark/pull/58543#discussion_r3955632894
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExecSuite.scala:
##########
@@ -411,22 +409,37 @@ class GroupPartitionsExecSuite extends SharedSparkSession
{
assert(!GroupPartitionsExec(child).groupedPartitions.forall(_._2.size <=
1),
"expected coalescing")
- withSQLConf(SQLConf.V2_BUCKETING_PRESERVE_ORDERING_ON_COALESCE_ENABLED.key
-> "true") {
- assert(GroupPartitionsExec(child).outputOrdering !== childOrdering,
- "config alone should not enable k-way merge; enableSortedMerge must be
set by planner")
- assert(GroupPartitionsExec(child, enableSortedMerge =
true).outputOrdering === childOrdering)
- }
- withSQLConf(
- SQLConf.V2_BUCKETING_PRESERVE_ORDERING_ON_COALESCE_ENABLED.key ->
"false",
- SQLConf.V2_BUCKETING_PRESERVE_KEY_ORDERING_ON_COALESCE_ENABLED.key ->
"true") {
- // Sorted-merge config disabled, key-ordering config enabled: only
key-expression orders
- // survive simple concatenation (non-key exprC is dropped).
- val ordering = GroupPartitionsExec(child, enableSortedMerge =
true).outputOrdering
+ assert(GroupPartitionsExec(child, enableSortedMerge = true).outputOrdering
=== childOrdering)
+
withSQLConf(SQLConf.V2_BUCKETING_PRESERVE_KEY_ORDERING_ON_COALESCE_ENABLED.key
-> "true") {
+ // Without the flag there is no k-way merge, so only key-expression
orders survive simple
+ // concatenation and the non-key exprC is dropped.
+ val ordering = GroupPartitionsExec(child).outputOrdering
assert(ordering.length === 1)
assert(ordering.head.child === exprA)
}
}
+ test("SPARK-59279: enableSortedMerge decides the k-way merge, not the
config") {
+ // The config is the planner's input, and `enableSortedMerge` records what
the planner decided
+ // under it. Once the flag is set the config no longer matters, because
the plan above was
+ // built on the ordering the merge delivers.
+ val partitionKeys = Seq(row(1), row(2), row(1))
+ val childOrdering = Seq(SortOrder(exprA, Ascending), SortOrder(exprC,
Ascending))
+ val child = DummyLeafSparkPlan(
+ outputPartitioning = KeyedPartitioning(Seq(exprA), partitionKeys),
+ outputOrdering = childOrdering)
+
+ Seq(true, false).foreach { configEnabled =>
Review Comment:
You are right, the grid was half a grid. Fixed in
[`6a138ec`](https://github.com/apache/spark/commit/6a138ec0fef3298ceaa5c1fff0308bafc5d7496f).
The unflagged case now runs under both config values, so the assertion this
PR removed from the test above is back and covers more than it did: it used to
check `config = true` only.
val unflagged = GroupPartitionsExec(child)
assert(unflagged.outputOrdering !== childOrdering,
s"config=$configEnabled: without the flag there is no k-way merge to
report, and the " +
"config cannot supply one")
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -440,8 +451,11 @@ case class GroupPartitionsExec(
s"Reducers: ${truncatedString(names, "[", ", ", "]", joinKeyMaxFields)}"
}
val distributeStr = Iterator(s"DistributePartitions:
$distributePartitions")
- joinKeyStr ++ expectedStr ++ reducersStr ++ distributeStr
-
+ // Rendered from the constructor field, as `DistributePartitions` above
is. Not from
+ // `usesSortedMerge`, because that forces `grouping`, which can throw, and
this method feeds
+ // `simpleString`, which `treeString` calls on error paths.
+ val sortedMergeStr = Iterator(s"SortedMerge: $enableSortedMerge")
Review Comment:
Conscious, and thank you for pinning it down rather than letting it pass.
Two reasons I would keep it. The flag is now the only thing in the plan that
says whether the node k-way merges, since the config no longer decides at
execution, so a reader of an EXPLAIN has nowhere else to look. And that is
exactly the failure this PR is about: the plan above the node was built on an
ordering the node then did not deliver, and the EXPLAIN gave no sign of it.
On the maintenance branches, `GroupPartitions` itself is recent, no golden
file contains it, and the line already carries `DistributePartitions`, so the
shape of the string is not new, only one more field on it.
If you would rather not change the string in 4.3 and 4.2 at all, the
alternative I would take is printing `SortedMerge` only when it is true. That
keeps every existing EXPLAIN byte-identical and still surfaces the case a
reader needs to see. Say the word and I will do that instead, here or in the
backports only.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
