[rkrishn7]

Which issue does this PR close?

• Closes Support GroupsAccumulator for unordered first_value #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:

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 MAXs 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.

[zhuqi-lucas]

LGTM, thanks @rkrishn7!

[xudong963]

How about adding a regression test through FirstValue::groups_accumulator_supported/create_groups_accumulator with no ORDER BY.

The current tests instantiate the new private accumulator directly, so restoring the old ordering gate would make the optimization unreachable while every new test remained green.

[rkrishn7]

How about adding a regression test through FirstValue::groups_accumulator_supported/create_groups_accumulator with no ORDER BY.

Added in c36da6e