This is an automated email from the ASF dual-hosted git repository.
peter-toth pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new aae16282055d [SPARK-58323][SQL] Materialize AliasAware output ordering
and partitioning to avoid StackOverflowError
aae16282055d is described below
commit aae16282055d5574286372925ab40f81370eff89
Author: Peter Toth <[email protected]>
AuthorDate: Sat Jul 25 11:22:36 2026 +0200
[SPARK-58323][SQL] Materialize AliasAware output ordering and partitioning
to avoid StackOverflowError
### What changes were proposed in this pull request?
`AliasAwareQueryOutputOrdering.outputOrdering` and
`PartitioningPreservingUnaryExecNode.outputPartitioning` (and
`BroadcastHashJoinExec`'s output-partitioning expansion) build their results
with `multiTransform`, which returns a `LazyList`, and store the bounded result
unforced. This forces them into strict collections with `.toList`, right after
the existing `.take(...)` so the lazy short-circuit during candidate generation
is preserved.
Note: for `BroadcastHashJoinExec`, using a strict `List` also makes the
existing `case p :: Nil => p` unwrap in `outputPartitioning` reachable — so a
single-element expansion now returns the `HashPartitioning` directly instead of
a one-element `PartitioningCollection` (that branch was previously dead because
the expansion was an unforced `LazyList`, which a `List`-cons pattern never
matches). This is the cleaner, intended form; the corresponding
`BroadcastJoinSuite` assertion is updat [...]
### Why are the changes needed?
Each plan node re-wraps the child ordering/partitioning's unforced
`LazyList`, so across a deep projection chain the deferred nesting grows
unbounded and overflows the stack when the ordering/partitioning is later
serialized or deeply traversed. Concretely, task serialization of a
`SortedMergeCoalescedRDD` (SPARK-55715) that captures the child
`outputOrdering` was observed to fail on the driver with a `StackOverflowError`
at `DAGScheduler.submitMissingTasks` on a large bucketed `MERGE [...]
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
New test in `ProjectedOrderingAndPartitioningSuite` asserting
`sameOrderExpressions` and the projected `PartitioningCollection.partitionings`
are strict (not `LazyList`); it fails before this change (they were `LazyList`)
and passes after. `BroadcastJoinSuite` / `BroadcastJoinSuiteAE` updated for the
single-element output-partitioning form and pass. Existing
`ProjectedOrderingAndPartitioningSuite` / `PlannerSuite` /
`EnsureRequirementsSuite` pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.8
Closes #57502 from
peter-toth/SPARK-58323-aliasaware-strict-ordering-partitioning.
Authored-by: Peter Toth <[email protected]>
Signed-off-by: Peter Toth <[email protected]>
(cherry picked from commit faa1d1ecad4c64dc528ee83be6eb19bb5291c439)
Signed-off-by: Peter Toth <[email protected]>
---
.../plans/AliasAwareOutputExpression.scala | 6 +++++
.../sql/execution/AliasAwareOutputExpression.scala | 7 +++++-
.../execution/joins/BroadcastHashJoinExec.scala | 5 +++++
.../ProjectedOrderingAndPartitioningSuite.scala | 26 ++++++++++++++++++++++
.../sql/execution/joins/BroadcastJoinSuite.scala | 8 ++++++-
5 files changed, 50 insertions(+), 2 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/AliasAwareOutputExpression.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/AliasAwareOutputExpression.scala
index e1a9e8b5ea81..e7d2ab8538a4 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/AliasAwareOutputExpression.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/AliasAwareOutputExpression.scala
@@ -107,6 +107,12 @@ trait AliasAwareQueryOutputOrdering[T <: QueryPlan[T]]
.flatMap(projectExpression)
.filter(e => orderingSet.add(e.canonicalized))
.take(aliasCandidateLimit)
+ // Materialize the bounded result into a strict collection. The
`LazyList` above is only
+ // needed so `take` can short-circuit candidate generation; storing
`sameOrderExpressions`
+ // as an unforced `LazyList` lets each plan node re-wrap the child
ordering's lazy list,
+ // and across a deep projection chain that nesting overflows the
stack when the ordering
+ // is later serialized or deeply traversed.
+ .toList
if (sameOrderings.nonEmpty) {
Some(sortOrder.copy(child = sameOrderings.head,
sameOrderExpressions = sameOrderings.tail))
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala
index b37e1b258e9b..2a3891c321cc 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala
@@ -37,7 +37,12 @@ trait PartitioningPreservingUnaryExecNode extends
UnaryExecNode
projectKeyedPartitionings(keyedPartitionings.map(_.asInstanceOf[KeyedPartitioning]))
val projectedOthers = projectOtherPartitionings(otherPartitionings)
- (projectedKPs ++ projectedOthers).take(aliasCandidateLimit) match {
+ // Materialize into a strict collection. The projected partitionings are
lazy `LazyList`s
+ // (needed so `take` can short-circuit); storing an unforced `LazyList` in
the returned
+ // `PartitioningCollection` lets each plan node re-wrap the child's lazy
list, and across a
+ // deep projection chain that nesting overflows the stack when the
partitioning is later
+ // serialized or deeply traversed.
+ (projectedKPs ++ projectedOthers).take(aliasCandidateLimit).toList match {
case Seq() => UnknownPartitioning(child.outputPartitioning.numPartitions)
case Seq(p) => p
case ps => PartitioningCollection.fromPartitionings(ps)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala
index d8bcd01f256d..e08f24214271 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala
@@ -121,6 +121,11 @@ case class BroadcastHashJoinExec private(
e +: streamedKeyToBuildKeyMapping(e.canonicalized)
}.asInstanceOf[LazyList[Partitioning]]
.take(conf.broadcastHashJoinOutputPartitioningExpandLimit)
+ // Materialize the bounded expansion into a strict `List` -- never
an unforced `LazyList`,
+ // which chained across nodes can overflow the stack when serialized
or deeply traversed.
+ // A `List` also lets the `case p :: Nil` in `outputPartitioning`
unwrap a single-element
+ // expansion to the partitioning itself, not a one-element
`PartitioningCollection`.
+ .toList
case other => Seq(other)
}
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala
index cb5570bfdf2c..42f5e217ff4a 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala
@@ -85,6 +85,32 @@ class ProjectedOrderingAndPartitioningSuite
}
}
+ test("SPARK-58323: AliasAware output ordering and partitioning are strict,
not lazy") {
+ // A multi-alias projection makes `sameOrderExpressions` and the projected
+ // `PartitioningCollection` non-trivial. Both must be strict collections:
the underlying
+ // `multiTransform` produces a `LazyList`, and storing it unforced lets
each plan node re-wrap
+ // the child's lazy list. Across a deep projection chain that nesting
overflows the stack when
+ // the ordering/partitioning is later serialized or deeply traversed.
+ withSQLConf(SQLConf.EXPRESSION_PROJECTION_CANDIDATE_LIMIT.key -> "5") {
+ // Ordering (AliasAwareQueryOutputOrdering): id -> {x, y, z} gives
sameOrderExpressions.
+ val orderDf = spark.range(2).orderBy($"id").selectExpr("id as x", "id as
y", "id as z")
+ val outputOrdering = orderDf.queryExecution.optimizedPlan.outputOrdering
+ assert(outputOrdering.head.sameOrderExpressions.nonEmpty)
+ outputOrdering.foreach { so =>
+ assert(!so.sameOrderExpressions.isInstanceOf[LazyList[_]],
+ s"sameOrderExpressions must be strict, was
${so.sameOrderExpressions.getClass.getName}")
+ }
+
+ // Partitioning (PartitioningPreservingUnaryExecNode): repartition(id)
-> {x, y, z}.
+ val partDf = spark.range(2).repartition($"id").selectExpr("id as x", "id
as y", "id as z")
+ val outputPartitioning =
stripAQEPlan(partDf.queryExecution.executedPlan).outputPartitioning
+ val partitionings =
outputPartitioning.asInstanceOf[PartitioningCollection].partitionings
+ assert(partitionings.size == 3)
+ assert(!partitionings.isInstanceOf[LazyList[_]],
+ s"partitionings must be strict, was ${partitionings.getClass.getName}")
+ }
+ }
+
test("SPARK-42049: Improve AliasAwareOutputExpression - ordering -
multi-references") {
val df = spark.range(2).selectExpr("id as a", "id as b")
.orderBy($"a" + $"b").selectExpr("a as x", "b as y")
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/joins/BroadcastJoinSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/joins/BroadcastJoinSuite.scala
index 52746720eba1..459919be4f80 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/joins/BroadcastJoinSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/joins/BroadcastJoinSuite.scala
@@ -638,7 +638,13 @@ abstract class BroadcastJoinSuiteBase extends QueryTest
None,
left = DummySparkPlan(outputPartitioning = HashPartitioning(Seq(l1,
l2), 1)),
right = DummySparkPlan())
- assert(bhj.outputPartitioning ===
PartitioningCollection(expected.take(limit)))
+ // A single expanded partitioning is returned directly; multiple are
wrapped in a
+ // `PartitioningCollection`.
+ val expectedPartitioning = expected.take(limit) match {
+ case Seq(p) => p
+ case ps => PartitioningCollection(ps)
+ }
+ assert(bhj.outputPartitioning === expectedPartitioning)
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
