rkrishn7 opened a new pull request, #27150:
URL: https://github.com/apache/datafusion/pull/27150
## Which issue does this PR close?
- Closes #27149
## Rationale for this change
`first_value` only has a `GroupsAccumulator` when it has an `ORDER BY`.
Without one, grouped aggregation uses `GroupsAccumulatorAdapter`, which keeps
one boxed `TrivialFirstValueAccumulator`, holding a `ScalarValue`, per group.
The planner produces unordered `first_value` for every `DISTINCT ON` without
`ORDER BY`: `ReplaceDistinctWithAggregate` turns `SELECT DISTINCT ON (keys)
cols` into an aggregate grouped by `keys` with one `first_value(col)` per
selected column. With many distinct keys, the per-group accumulators dominate
both time and memory.
## What changes are included in this PR?
- `UnorderedFirstValueGroupsAccumulator`: a `GroupsAccumulator` for
`first_value` without `ORDER BY`. It stores values in the existing `ValueState`
implementations (`PrimitiveValueState`, `BytesValueState`, `GenericValueState`)
and keeps one `is_set` bit per group.
- `FirstValue::groups_accumulator_supported` no longer requires an ordering.
`LastValue` and ordered `FirstValue` are unchanged.
A group takes the first row that passes the `FILTER` clause (and, with
`IGNORE NULLS`, is not null), and is never written again. That is the row
`TrivialFirstValueAccumulator` keeps: rows within a batch in input order, and
batches in arrival order. `merge_batch` applies the same rule to rows whose
`is_set` flag is true.
The intermediate state is `[first_value, is_set]`, the same layout as
`TrivialFirstValueAccumulator`, so partial and final aggregation see the same
schema as before. `convert_to_state` sets `is_set` from the filter and `IGNORE
NULLS`, so skipped partial aggregation keeps the same semantics.
A batch costs one bit test per row plus one value copy per newly decided
group. Nothing visits groups the batch has no rows for.
## What is the testing strategy for this PR?
Yes. New unit tests in `first_last.rs` cover:
- `RESPECT NULLS` keeping a leading null, and later batches not overwriting
a group's value
- `IGNORE NULLS`
- a `FILTER` predicate that is null (excluded)
- `merge_batch` keeping existing values, skipping rows with `is_set =
false`, and filling groups out of order
- `convert_to_state` folding `FILTER` and `IGNORE NULLS` into `is_set`
- `EmitTo::First`
- Utf8 and List values
### Benchmarks
ClickBench `hits.parquet` (`./benchmarks/bench.sh data clickbench_1`),
`datafusion-cli` release builds of `main` (b6b2e0e) and this PR, Apple M4 Pro
(14 cores, 24 GB). Mean of 5 runs after 1 warmup (`hyperfine`); peak memory is
the process peak footprint from `/usr/bin/time -l`, median of 3 runs.
| Query | Groups | `main` | This PR | Speedup | Peak memory `main` → PR |
|---|---|---|---|---|---|
| Q1 `DISTINCT ON ("UserID")`, 5 columns | 17.6M | 73.3 s ± 12.7 | 6.1 s ±
0.9 | 12.0x | 21.2 GB → 11.3 GB |
| Q2 `first_value("URL")` by `UserID` | 17.6M | 5.78 s | 0.95 s | 6.1x | 8.3
GB → 5.7 GB |
| Q3 `first_value("WatchID")` by `UserID` | 17.6M | 6.24 s | 0.34 s | 18.3x
| 6.6 GB → 2.4 GB |
| Q4 `DISTINCT ON ("OS")`, 5 columns | 91 | 1.21 s | 1.13 s | 1.07x | 1.6 GB
→ 1.6 GB |
On `main`, Q1 peaks close to the machine's 24 GB, so part of its time and
variance is likely memory pressure. Q4 has few groups and shows no regression.
**Queries**:
```sql
CREATE EXTERNAL TABLE hits STORED AS PARQUET LOCATION
'benchmarks/data/hits.parquet';
-- Q1
SELECT MAX(LENGTH("URL")), MAX(LENGTH("Title")),
MAX(LENGTH("SearchPhrase")), MAX("EventTime"), MAX("RegionID")
FROM (
SELECT DISTINCT ON ("UserID") "UserID", "URL", "Title", "SearchPhrase",
"EventTime", "RegionID"
FROM hits
);
-- Q2
SELECT MAX(LENGTH(u)) FROM (
SELECT "UserID", first_value("URL") AS u FROM hits GROUP BY "UserID"
);
-- Q3
SELECT MAX(w) FROM (
SELECT "UserID", first_value("WatchID") AS w FROM hits GROUP BY "UserID"
);
-- Q4
SELECT MAX(LENGTH("URL")), MAX(LENGTH("Title")),
MAX(LENGTH("SearchPhrase")), MAX("EventTime"), MAX("RegionID")
FROM (
SELECT DISTINCT ON ("OS") "OS", "URL", "Title", "SearchPhrase",
"EventTime", "RegionID"
FROM hits
);
```
The outer `MAX`s keep every `first_value` in the plan while printing one
row. Results of unordered `first_value` depend on partition scheduling on both
`main` and this PR; with `target_partitions = 1`, both return identical results.
## Are there any user-facing changes?
No API changes. Queries using `first_value` without `ORDER BY`, including
`DISTINCT ON` without `ORDER BY`, use less time and memory. Which row is
"first" is unchanged: as before, it depends on input order.
--
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]
