peter-toth commented on code in PR #59306:
URL: https://github.com/apache/spark/pull/59306#discussion_r4246414728
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -253,23 +256,33 @@ case class GroupPartitionsExec(
case _ => true
}
+ /** Whether the k-way merge was planned for this node. See
`sortedMergeOrdering`. */
+ def enableSortedMerge: Boolean = sortedMergeOrdering.isDefined
+
+ /**
+ * The ordering the k-way merge merges by. Before the merge is planned it is
the child's. After
+ * that it is the one the merge was planned with, see `sortedMergeOrdering`.
+ */
+ private def mergeOrdering: Seq[SortOrder] =
sortedMergeOrdering.getOrElse(child.outputOrdering)
+
/** Whether a k-way merge would work at all, leaving aside whether it is
switched on. */
@transient private lazy val kWayMergeIsFeasible: Boolean =
- child.outputOrdering.nonEmpty && childIsSafeForKWayMerge
+ mergeOrdering.nonEmpty && childIsSafeForKWayMerge
- /** Whether this node performs the k-way merge. No config term, see
`enableSortedMerge`. */
+ /** Whether this node performs the k-way merge. No config term, see
`sortedMergeOrdering`. */
@transient private lazy val usesSortedMerge: Boolean =
Review Comment:
Thanks. A planned merge that cannot run now fails at execution
(`checkPlannedMergeRuns`), as `checkChildStillMatches` does for the
partitioning. Planning keeps the fallback, since `EnsureRequirements` then adds
a `SortExec`.
The check also showed that, without AQE, the `debug()` helper wraps every
node in `DebugExec`, which was not whitelisted. So `debug()` on a planned merge
would now fail. `DebugExec` keeps its per-partition state in its evaluator and
its accumulators only add, so it is now `SafeForKWayMerge`, with a test.
Fixed in
[`24bb37cf81b`](https://github.com/apache/spark/pull/59306/commits/24bb37cf81b6fd665b34a0ab3458797b1392e76b).
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -253,23 +256,33 @@ case class GroupPartitionsExec(
case _ => true
}
+ /** Whether the k-way merge was planned for this node. See
`sortedMergeOrdering`. */
+ def enableSortedMerge: Boolean = sortedMergeOrdering.isDefined
+
+ /**
+ * The ordering the k-way merge merges by. Before the merge is planned it is
the child's. After
+ * that it is the one the merge was planned with, see `sortedMergeOrdering`.
+ */
+ private def mergeOrdering: Seq[SortOrder] =
sortedMergeOrdering.getOrElse(child.outputOrdering)
+
/** Whether a k-way merge would work at all, leaving aside whether it is
switched on. */
@transient private lazy val kWayMergeIsFeasible: Boolean =
- child.outputOrdering.nonEmpty && childIsSafeForKWayMerge
+ mergeOrdering.nonEmpty && childIsSafeForKWayMerge
- /** Whether this node performs the k-way merge. No config term, see
`enableSortedMerge`. */
+ /** Whether this node performs the k-way merge. No config term, see
`sortedMergeOrdering`. */
@transient private lazy val usesSortedMerge: Boolean =
enableSortedMerge && hasCoalescing && kWayMergeIsFeasible
/**
* Returns a copy of this node with k-way merge enabled, when the config is
on, this node
* coalesces partitions and the merge is feasible. The only read of
- * `preserveOrderingOnCoalesce`.
+ * `preserveOrderingOnCoalesce`. The copy keeps the ordering checked here,
see
+ * `sortedMergeOrdering`.
*/
def tryEnableSortedMerge(): Option[GroupPartitionsExec] = {
Option.when(conf.v2BucketingPreserveOrderingOnCoalesceEnabled &&
hasCoalescing &&
kWayMergeIsFeasible) {
- val newGroupPartitions = copy(enableSortedMerge = true)
+ val newGroupPartitions = copy(sortedMergeOrdering = Some(mergeOrdering))
Review Comment:
Done in
[`24bb37cf81b`](https://github.com/apache/spark/pull/59306/commits/24bb37cf81b6fd665b34a0ab3458797b1392e76b),
`tryEnableSortedMerge` reads the ordering once. It also leaves a node that
already merges as it is, so the merge is decided once.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -75,14 +75,17 @@ import org.apache.spark.sql.vectorized.ColumnarBatch
* [[REPLICATED_FOR_JOIN]] groups the splits first and
repeats each group
* into its expected slots, and `None` groups them,
one partition per key.
* See [[UngroupingOrigin]].
- * @param enableSortedMerge When true, uses [[SortedMergeCoalescedRDD]] to
perform a k-way merge
- * of the coalesced partitions, preserving the
child's output ordering
- * end-to-end. Set by [[EnsureRequirements]] when a
parent operator
- * requires the ordering that this node can satisfy
via sorted merge.
- * This flag is the decision, not a hint. Nothing
below re-reads the
- * config that produced it. A node planned with the
merge would otherwise
- * concatenate instead, and a sort-merge join above
it would lose rows
- * (SPARK-59279).
+ * @param sortedMergeOrdering When set, uses [[SortedMergeCoalescedRDD]] to
perform a k-way merge
+ * of the coalesced partitions by this ordering,
preserving it
+ * end-to-end. Set by [[EnsureRequirements]] when a
parent operator
+ * requires the ordering that this node can satisfy
via sorted merge.
+ * This is the decision, not a hint. Nothing below
re-reads the config
+ * that produced it. A node planned with the merge
would otherwise
+ * concatenate instead, and a sort-merge join above
it would lose rows
+ * (SPARK-59279). The ordering is captured from the
child when the merge
+ * is decided. That is the ordering the parent's
requirement was checked
+ * against. The child's ordering can follow a
config too, so a later
Review Comment:
The scan's data does not depend on the config, only what it reports. So the
other readers only do extra work after such a change, e.g. a reverted AQE
rewrite. A per-instance freeze like the one for `reportedKeyedPartitioning`
would not cover every case either. It is a `lazy val`, so a new instance reads
the config again. `PlanAdaptiveDynamicPruningFilters` makes one at stage
creation for a scan with a dynamic pruning filter. So I'd keep the freeze in
the node whose execution depends on the ordering, and leave the scan as it is.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -253,23 +256,33 @@ case class GroupPartitionsExec(
case _ => true
}
+ /** Whether the k-way merge was planned for this node. See
`sortedMergeOrdering`. */
+ def enableSortedMerge: Boolean = sortedMergeOrdering.isDefined
+
+ /**
+ * The ordering the k-way merge merges by. Before the merge is planned it is
the child's. After
+ * that it is the one the merge was planned with, see `sortedMergeOrdering`.
+ */
+ private def mergeOrdering: Seq[SortOrder] =
sortedMergeOrdering.getOrElse(child.outputOrdering)
Review Comment:
Done in
[`24bb37cf81b`](https://github.com/apache/spark/pull/59306/commits/24bb37cf81b6fd665b34a0ab3458797b1392e76b).
The constructor requires a non-empty ordering. `tryEnableSortedMerge` reads
the child's ordering directly, and the merge path reads only the frozen one.
`mergeOrdering` and `kWayMergeIsFeasible` are gone.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -490,8 +512,8 @@ case class GroupPartitionsExec(
s"Reducers: ${truncatedString(names, "[", ", ", "]", joinKeyMaxFields)}"
}
val distributeStr = Iterator(s"DistributePartitions:
$distributePartitions")
- // Rendered from the constructor field, as `DistributePartitions` above
is, and not from
- // `usesSortedMerge`, which reads the child's ordering. This method feeds
`simpleString`, which
+ // Rendered from the constructor field `sortedMergeOrdering`, as
`DistributePartitions` above
+ // is. `usesSortedMerge` would walk the child. This method feeds
`simpleString`, which
// `treeString` calls on error paths, so it stays on what the node was
built with.
val sortedMergeStr = Iterator(s"SortedMerge: $enableSortedMerge")
Review Comment:
Done in
[`24bb37cf81b`](https://github.com/apache/spark/pull/59306/commits/24bb37cf81b6fd665b34a0ab3458797b1392e76b),
e.g. `SortedMerge: [id#1 ASC NULLS FIRST, name#2 ASC NULLS FIRST]`. A node
without a merge still shows `SortedMerge: false`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -428,16 +443,24 @@ case class GroupPartitionsExec(
* `plannedPartitioning` are decided for one child's partitioning, so a node
handed another has to
* be decided again over it.
*
- * Rebuilding is all this does: what the operator above reads, the ordering
it was planned
- * against included, is the caller's to hold, since nothing here knows what
that operator
- * requires.
+ * A merging node is turned away. Its frozen `sortedMergeOrdering` was
checked against this node's
+ * child, not against `newChild`. No merging node reaches this path today.
The only caller,
+ * `CombineAdjacentAggregation`, rebuilds a node over a subtree that holds
an aggregate, which is
+ * not `SafeForKWayMerge`.
+ *
+ * Rebuilding is all this does. Nothing here knows what the operator above
requires, so checking
+ * that the rebuilt node still meets it is the caller's job.
*/
def withKeyPositionsFor(newChild: SparkPlan): Option[GroupPartitionsExec] = {
// An aligned node's slot order comes from the parent distribution rather
than the child, so the
// factory cannot re-decide it for `newChild`.
if (expectedKeyCount.isDefined) {
return None
}
+ // The frozen merge ordering was checked against this node's child, not
against `newChild`.
+ if (enableSortedMerge) {
Review Comment:
I'd keep refusing it:
- Rebuilding without the frozen ordering would drop a merge the parent
relies on, since its sort was not added.
- Deciding again over `newChild` needs the parent's required ordering, which
this method does not see.
- An assertion would fail a query, while refusing only skips the rebuild in
`CombineAdjacentAggregation`, an optimization.
No query sees the change, since no merging node reaches this method today.
The method already refuses an aligned node (`expectedKeyCount`) for a similar
reason.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExecSuite.scala:
##########
@@ -493,7 +505,7 @@ class GroupPartitionsExecSuite extends SharedSparkSession {
withSQLConf(
SQLConf.V2_BUCKETING_PRESERVE_ORDERING_ON_COALESCE_ENABLED.key ->
configEnabled.toString) {
- val flagged = GroupPartitionsExec(child, enableSortedMerge = true)
+ val flagged = GroupPartitionsExec(child, sortedMergeOrdering =
Some(child.outputOrdering))
Review Comment:
Added "SPARK-60044: a planned merge keeps the ordering it was planned with"
in
[`24bb37cf81b`](https://github.com/apache/spark/pull/59306/commits/24bb37cf81b6fd665b34a0ab3458797b1392e76b).
Its child's ordering follows `partitionKeyOrdering.enabled`, like the scan's.
It checks that `tryEnableSortedMerge` captures the ordering, and that the node,
a `copy` and a `withNewChildren` over an `InputAdapter` keep it after the
config is turned off. It fails if `withNewChildInternal` drops the ordering or
`outputOrdering` reads the child's.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExecSuite.scala:
##########
@@ -450,7 +460,8 @@ class GroupPartitionsExecSuite extends SharedSparkSession {
withSQLConf(SQLConf.V2_BUCKETING_PRESERVE_KEY_ORDERING_ON_COALESCE_ENABLED.key
-> "true") {
// Even though enableSortedMerge = true, the child is not safe for k-way
merge,
Review Comment:
Done in
[`24bb37cf81b`](https://github.com/apache/spark/pull/59306/commits/24bb37cf81b6fd665b34a0ab3458797b1392e76b).
--
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]
