[SPARK-42746][SQL][FIXUP] Fix optimizer failure for SortOrder in the LISTAGG function #51117

uros-db · 2025-06-08T21:58:01Z

What changes were proposed in this pull request?

Update the RewriteDistinctAggregates optimizer rule to properly rewrite aggregates with: DISTINCT listagg / string_agg. This fixes failures with multiple distinct list aggregations in the same query. Also, to enable this behaviour, this PR refactors the ListAgg catalyst expression to use separate children in the case class definition, instead of using SortOrder for order expressions (note that SortOrder is an unevaluable expression).

Why are the changes needed?

Queries such as:

SELECT
  LISTAGG(DISTINCT col1) WITHIN GROUP (ORDER BY col1),
  LISTAGG(DISTINCT col2) WITHIN GROUP (ORDER BY col2)
FROM df;

are failing with java.lang.ClassCastException, because the RewriteDistinctAggregates replaces SortOrders with AttributeReferences. This PR fixes the issue, and SortOrders are no longer processed like this.

Does this PR introduce any user-facing change?

Yes, queries with multiple distinct list aggregations now work properly.

How was this patch tested?

Added appropriate end-to-end SQL tests to verify the behaviour.

Was this patch authored or co-authored using generative AI tooling?

No.

uros-db

@cloud-fan Please review this PR, which is a fixup for ListAgg - originally introduced in: #48748.

mihailom-db · 2025-06-09T10:17:51Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala

    mutableAggBufferOffset: Int = 0,
    inputAggBufferOffset: Int = 0)
  extends Collect[mutable.ArrayBuffer[Any]]
  with SupportsOrderingWithinGroup
  with ImplicitCastInputTypes {

+  val orderExpressions: Seq[SortOrder] = orderChildExpressions.zipWithIndex.map {


Should this be a lazy val?

It can be, but not sure if it would bring any special value to this particular expression. No strong opinion, I can update if needed.

it should be lazy val, to avoid repeatedly triggerring it during plan transformation when plan nodes are copied frequently.

Makes sense, I'll update to lazy.

mihailom-db · 2025-06-09T10:20:43Z

...alyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregates.scala

        af.withNewChildren(newChildren).asInstanceOf[AggregateFunction]
      }

      // Setup unique distinct aggregate children.
-      val distinctAggChildren = distinctAggGroups.keySet.flatten.toSeq.distinct
+      val distinctAggChildren = distinctAggGroups.keySet.flatten.toSeq


Is this safe to do? Does any other aggregate expression use this? Could you please confirm with a test?

What exactly are you referring to? If you mean filtering out SortOrder here, as far as I'm aware - the only catalyst expression that uses it is ListAgg, and tests are added for this expression (including some tests that combine listagg with other aggregates to verify the correct behaviour).

Do we have a way of adding a test for this rule, so that if new expression was added with SortOrder, we actually alert with a proper error? Just want to make sure we do not get into the same state in future, where we have bugs because of optimization.

mihailom-db · 2025-06-09T10:22:47Z

...alyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregates.scala

@@ -258,12 +258,18 @@ object RewriteDistinctAggregates extends Rule[LogicalPlan] {
      def patchAggregateFunctionChildren(
          af: AggregateFunction)(
          attrs: Expression => Option[Expression]): AggregateFunction = {
-        val newChildren = af.children.map(c => attrs(c).getOrElse(c))
+        val newChildren = af.children.map {
+          case so: SortOrder =>


So is SortOrder the only expression that should propagate into child's child?

Yes, this is the only one that I'm aware of, with respect to aggregate patching. Note that the issue originally came up with ListAgg - which is a bit of a specific aggregate expression. I don't think that we're aware of any other similar issues at this time, but please feel free to share any additional info.

Sorry I'm confused. SortOrder is no longer child of ListAgg, how can we hit it here?

SortOrder is no longer child of ListAgg, how can we hit it here?

That's because of this:

override def children: Seq[Expression] = child +: delimiter +: orderExpressions

Otherwise, it would be more difficult to pack up all of the optional / dynamic order expressions (that themselves include Seq[Expression] and even Seq[Seq[Expression]]).

why do we generate SortOrder and then put it in the children? Can we just put orderChildExpressions in the children?

ideally children should be selected from the constructor parameters, not something generated on the fly.

ideally children should be selected from the constructor parameters, not something generated on the fly.

Agreed. However, I don't think that's viable with this particular expression (please see below for more details).

Can we just put orderChildExpressions in the children?

I don't think that orderChildExpressions is enough, we need full sortorder information in order to be able to reconstruct the original aggregate - this can be either in a SortOrder object (as is currently implemented) or a sort of struct containing all of the relevant parameters (which would essentially be SortOrder, but with extra steps). Otherwise, packing up all optional / dynamic child expressions expressions in a children: Seq[Expression] would be very difficult.

Can you point me to the code where we collect the full sort order information from children? Why can't we match the ListAgg expression and collect whatever info we need?

cloud-fan · 2025-06-10T22:19:46Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala

+    orderChildExpressions: Seq[Expression] = Nil,
+    orderDirections: Seq[SortDirection] = Nil,
+    orderNullOrderings: Seq[NullOrdering] = Nil,
+    orderSameOrderExpressions: Seq[Seq[Expression]] = Nil,


do we need this? There is no SortOrder anymore, and we don't need it to make the planner smart to remove duplicated sorts.

SortOrder is used because of SupportsOrderingWithinGroup. Note that withOrderingWithinGroup uses all of the relevant children for ListAgg in s copy call, and then later we rely on orderExpressions - which is no longer a child (important because SortOrder is unevaluable). So to make this approach, I think we need this. We could of course explore alternative approaches if you have any suggestions?

cloud-fan · 2025-06-10T22:21:57Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala

-  override protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): Expression =
+  override protected def withNewChildrenInternal(
+      newChildren: IndexedSeq[Expression]): Expression = {
+    val sortOrderExpressions: Seq[SortOrder] = newChildren.drop(2).map(_.asInstanceOf[SortOrder])


do we really call ListAgg#withChildren this way? passing SortOrder as the new children?

Yes, because children is overriden (please see #51117 (comment)).

Initial commit

49be3fc

github-actions bot added the SQL label Jun 8, 2025

Fix analyzer results

004f577

uros-db changed the title ~~[WIP][SPARK-42746][SQL] Fix optimizer failure for SortOrder in the LISTAGG function~~ [SPARK-42746][SQL][FIXUP] Fix optimizer failure for SortOrder in the LISTAGG function Jun 9, 2025

uros-db added 2 commits June 9, 2025 09:05

Update tests

41012dc

Cleanup

6ced813

uros-db commented Jun 9, 2025

View reviewed changes

mihailom-db reviewed Jun 9, 2025

View reviewed changes

cloud-fan reviewed Jun 10, 2025

View reviewed changes

[SPARK-42746][SQL][FIXUP] Fix optimizer failure for SortOrder in the LISTAGG function #51117

Are you sure you want to change the base?

[SPARK-42746][SQL][FIXUP] Fix optimizer failure for SortOrder in the LISTAGG function #51117

Uh oh!

Conversation

uros-db commented Jun 8, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

Uh oh!

uros-db left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan Jun 10, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

uros-db Jun 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

uros-db commented Jun 8, 2025 •

edited

Loading

cloud-fan Jun 10, 2025 •

edited

Loading

uros-db Jun 11, 2025 •

edited

Loading