Skip to content

[SPARK-58614][SQL] Support DISTINCT aggregates in unbounded preceding window frames - #57815

Open
zml1206 wants to merge 9 commits into
apache:masterfrom
zml1206:distinct_window
Open

[SPARK-58614][SQL] Support DISTINCT aggregates in unbounded preceding window frames#57815
zml1206 wants to merge 9 commits into
apache:masterfrom
zml1206:distinct_window

Conversation

@zml1206

@zml1206 zml1206 commented Aug 6, 2026

Copy link
Copy Markdown
Contributor

What changes were proposed in this pull request?

This PR adds support for distinct aggregate window functions whose lower frame bound is
UNBOUNDED PRECEDING, including:

  • Full-partition frames ending at UNBOUNDED FOLLOWING.
  • Growing ROWS and RANGE frames.
  • General distinct aggregate functions, not only COUNT(DISTINCT ...).
  • Multiple distinct arguments, aggregate filters, and order-sensitive aggregates such as
    LISTAGG(DISTINCT ...).

The implementation:

  • Uses BytesToBytesMap to deduplicate binary-stable keys in memory.
  • Permanently falls back to UnsafeKVExternalSorter when the hash threshold is reached or
    memory allocation fails.
  • Uses first-occurrence events to incrementally update growing window frames.
  • Evaluates full-partition frames once without retaining or restoring the input row order.
  • Adds caller-owned sorter lifecycle support and spill metrics.
  • Introduces the internal
    spark.sql.windowExec.distinct.hash.fallbackThreshold configuration.

Bounded and sliding distinct window frames remain unsupported.

Why are the changes needed?

Spark currently rejects all distinct aggregate window functions, including common expressions
such as:

COUNT(DISTINCT value) OVER (
  PARTITION BY key
  ORDER BY id
  ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)

A simple in-memory hash set can consume unbounded memory for partitions with many distinct
values. This implementation combines hash-based deduplication with an external-sort fallback,
providing fast execution for normal workloads while allowing large partitions to spill to disk.

Does this PR introduce any user-facing change?

Yes.
Users can now execute distinct aggregate window functions when the lower frame bound is
UNBOUNDED PRECEDING.
Distinct window functions with bounded lower frames, such as sliding windows, continue to
produce the existing DISTINCT_WINDOW_FUNCTION_UNSUPPORTED analysis error.
Python UDAFs and distinct inputs without a supported ordering remain unsupported.

How was this patch tested?

The test coverage includes:
Growing ROWS and RANGE frames.
Full-partition frames.
COUNT, SUM, AVG, COLLECT_LIST, and LISTAGG with DISTINCT.
Multiple distinct arguments, filters, and null values.
Hash fallback triggered by entry-count and memory thresholds.
External sorter spills and spill metric reporting.
Binary-unstable collations, normalized floating-point values, and empty distinct keys.
Negative tests for unsupported bounded frames.
The relevant targeted SQL window tests and scalastyle checks pass locally. The LISTAGG
SQL golden results were regenerated to cover the newly supported behavior.

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

Generated-by: OpenAI Codex (GPT-5)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant