[SPARK-58420][SQL] Normalize map values in distinct aggregates

[vladimirg-db]

What changes were proposed in this pull request?

Normalize map-typed arguments of distinct aggregate expressions with MapSort. Complex distinct
arguments are projected to attributes before normalization, and the existing grouping aliases are
reused when applicable. The rule retains its plan-wide fast path and has a default-on internal
killswitch.

Why are the changes needed?

Map equality is independent of entry order, but distinct aggregation currently compares the
physical map representation. Equivalent maps with different insertion orders are therefore treated
as different values.

InsertMapSortInGroupingExpressions was introduced by
SPARK-47430
to normalize MapType grouping keys. This PR extends the same normalization to distinct aggregate
arguments.

Does this PR introduce any user-facing change?

Yes. COUNT(DISTINCT map_column) now treats maps containing the same entries in different orders as
the same value. For example:

SELECT count(DISTINCT m)
FROM VALUES
  (map('a', 1, 'b', 2)),
  (map('b', 2, 'a', 1)),
  (map('a', 3)) AS t(m);

Before this change, the query returns 3. After this change, it returns 2. Distinct aggregates
that return their argument, such as COLLECT_LIST(DISTINCT m), now return maps whose entries are
sorted by key.

Normalization is scoped to each use of the map. In a query containing both
MAX(map_values(m)[0]) and COUNT(DISTINCT m), the non-distinct aggregate sees the original map
while the distinct aggregate sees the sorted map. Existing grouping-key behavior is unchanged: a
normalized map grouping key is also the grouped value exposed to the select list.

Setting spark.sql.optimizer.insertMapSortInDistinctAggregates.enabled to false disables the
distinct-specific normalization and restores the previous behavior; grouping-key normalization is
unchanged.

How was this patch tested?

Added a dedicated SQL golden test covering COUNT(DISTINCT map), returned
COLLECT_LIST(DISTINCT map) values, multiple and multi-argument distinct aggregates, grouped
aggregation, aggregate filters, ORDER BY, HAVING, nested maps, null and empty maps, grouping
sets, and representative cases with the killswitch disabled.

build/sbt 'sql/testOnly org.apache.spark.sql.SQLQueryTestSuite -- -z distinct-map-aggregates.sql'

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

Generated-by: OpenAI Codex (GPT-5)

[uros-b]

The user-facing change is a bit broader than described. Normalization doesn't only fix counts, it changes returned values: collect_list(DISTINCT m), first(DISTINCT m), and friends now return map-sorted maps. The golden test wraps it in SIZE(...), which hides exactly that. Worth stating in the description and adding a case that shows the real collect_list(DISTINCT m) output. Relatedly, the tests do capture a deliberate asymmetry worth calling out explicitly: MAX(map_values(m)[0]) still sees the raw map while COUNT(DISTINCT m) sees the sorted one, so m means two different things within one query. That's fine (you don't want to perturb non-distinct aggregates), but it should be a stated decision rather than a side effect, especially since for grouping keys the sorted map does leak into the rest of the select list. Please mention this in the PR description.

[uros-b]

A few test gaps:

• null / empty maps as distinct arguments
• ROLLUP/GROUPING SETS combined with COUNT(DISTINCT map)

[vladimirg-db]

Thank you for the review, addressing now.

[vladimirg-db]

@uros-b all done!

[uros-b]

We also need an entry for this in the migration guide.

Please add this in docs/sql-migration-guide.md under "Upgrading from Spark SQL 4.2 to 4.3", covering both halves of the user-visible change: COUNT(DISTINCT map) now collapses maps that differ only in entry order, and aggregates that return their argument (COLLECT_LIST(DISTINCT m) and friends) now return key-sorted maps. Worth naming spark.sql.optimizer.insertMapSortInDistinctAggregates.enabled there as the way back to the old behavior.

[uros-b]

Thank you @vladimirg-db! Left a few more comments, PTAL ^^

[vladimirg-db]

Entry in docs/sql-migration-guide.md added.

[uros-b]

Left a few more comments - please address, otherwise looks good

[vladimirg-db]

@uros-b tests passed, the only failure is unrelated. Ready to merge.

[HyukjinKwon]

0 blocking, 2 non-blocking, 0 nits.
Core logic is sound and thoroughly reviewed; two test-quality items from the reviewer remain open.

Already raised in existing discussion (2)

• The flag-off case in distinct-map-aggregates.sql can't demonstrate the legacy behavior it's meant to: HiveResult.toHiveString sorts map entries when formatting output, so the recorded result looks identically sorted whether the flag is on or off -- a reader can't tell the legacy case from the enabled one. It also asserts on a full multi-element COLLECT_LIST array, whose ordering isn't deterministic across shuffle-partition counts, so it's a flakiness risk. Applying the map_entries + explode/ORDER BY treatment you already use elsewhere in the file would make the legacy behavior visible and remove the flakiness. (Raised by @uros-b, still open.) -- existing discussion
• Coverage gap worth closing while the suite is new: the rule now early-returns and leaves a map-free Aggregate untouched once the plan-wide shouldRewrite passes (a deliberate improvement), but nothing pins that down. A plan with a map-free Aggregate alongside a map-carrying one, asserting the map-free one comes back unchanged, would cover it cheaply. (Raised by @uros-b, still open.) Non-blocking. -- existing discussion

Verification

Verified the distinct path is normalized in this rule rather than RewriteDistinctAggregates (a single distinct group without a filter bypasses that rule), the shouldRewrite pre-scan guard is restored, the transformDown order is load-bearing and commented, and grouping aliases are reused. The two open items are test-quality, not logic: the flag-off case is non-demonstrative (HiveResult sorts map output so on/off look identical) and flaky (asserts a non-deterministic COLLECT_LIST array), and the map-free-aggregate early-return path is untested.

[uros-b]

yeah @vladimirg-db PTAL at the minor comments, but otherwise looks good so should be ready to merge soon

[vladimirg-db]

@uros-b all addressed, just needed to reply.

[uros-b]

Merge Summary:

• merged into master f649423
• merged into branch-4.x b724fd8

Posted by merge_spark_pr.py

[uros-b]

Merge Summary:

• merged into branch-4.3 6f05301

Posted by merge_spark_pr.py