Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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))
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -117,7 +117,10 @@ case class BroadcastHashJoinExec private(
case e: Expression if streamedKeyToBuildKeyMapping.contains(e.canonicalized) =>
e +: streamedKeyToBuildKeyMapping(e.canonicalized)
}.asInstanceOf[LazyList[HashPartitioningLike]]
.take(conf.broadcastHashJoinOutputPartitioningExpandLimit))
.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.
.toList)
}

protected override def doExecute(): RDD[InternalRow] = {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,23 @@ class ProjectedOrderingAndPartitioningSuite
}
}

test("SPARK-58323: AliasAware output ordering is strict, not lazy") {
// A multi-alias projection makes `sameOrderExpressions` non-trivial. It must be a strict
// collection: the underlying `multiTransform` produces a `LazyList`, and storing it unforced
// lets each plan node re-wrap the child ordering's lazy list. Across a deep projection chain
// that nesting overflows the stack when the ordering is later serialized or deeply traversed.
withSQLConf(SQLConf.EXPRESSION_PROJECTION_CANDIDATE_LIMIT.key -> "5") {
// id -> {x, y, z} gives sameOrderExpressions.
val df = spark.range(2).orderBy($"id").selectExpr("id as x", "id as y", "id as z")
val outputOrdering = df.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}")
}
}
}

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")
Expand Down