This is an automated email from the ASF dual-hosted git repository.
peter-toth 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 5915191b4dd6 [SPARK-58323][SQL] Materialize AliasAware output ordering
and partitioning to avoid StackOverflowError
5915191b4dd6 is described below
commit 5915191b4dd6405a4c846e742b2b17cdab41bcbb
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
`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 [...]
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 [...]
No.
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.
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 +++++
.../execution/joins/BroadcastHashJoinExec.scala | 5 +++++
.../ProjectedOrderingAndPartitioningSuite.scala | 26 ++++++++++++++++++++++
.../sql/execution/joins/BroadcastJoinSuite.scala | 8 ++++++-
4 files changed, 44 insertions(+), 1 deletion(-)
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/joins/BroadcastHashJoinExec.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala
index e4f18c9144dd..70b5efa3e82f 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 ec13d48d45f8..e67ce5f973ca 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
@@ -84,6 +84,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]
