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 e1a9e8b5ea810..e7d2ab8538a42 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 944ee3b059092..c8212eeb071c4 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 @@ -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] = { 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 ec13d48d45f84..d55e5bd68187b 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,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")