zml1206 opened a new pull request, #57815:
URL: https://github.com/apache/spark/pull/57815
### 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:
```sql
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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
