[EPIC] Memory-efficient dedup on wide payload: fix ROW_NUMBER()=1 and DISTINCT ON patterns

Motivation

Two real-world scenarios hit unnecessary OOM / severe memory blowup for the same class of query — "keep one row per group, pick winner by an ordering column":

Scenario A — SELECT DISTINCT ON (id) * ORDER BY ts LIMIT 10 (already reported as #16620)

Peak memory 1.4 GB for ~500K rows × 14 columns
~4× slower than SELECT DISTINCT * on the same data

Scenario B — production batch pipeline running SELECT * FROM (SELECT *, ROW_NUMBER() OVER (PARTITION BY p ORDER BY o) rn FROM t) WHERE rn = 1 over nested JSON with wide List<Struct<...>> payload

11M rows after UNNEST, ~2.5 KB per row
OOMs at ~90 GB peak on 16 CPU (dominated by ExternalSorterMerge)
Manual SQL rewrite into a two-step MAX + JOIN + FIRST_VALUE-without-ORDER-BY variant reduces peak to 3.6 GB and wall time to ~40s
The naive rewrite FIRST_VALUE(wide_col ORDER BY o) GROUP BY p blows up to ~132 GB — measured by an OS memory kill — due to per-group Accumulator fallback on nested types

Root cause

first_value / last_value with ORDER BY falls back to the per-group Accumulator path for nested types (List, Struct, Map, ...) because groups_accumulator_supported() in first_last.rs whitelists only scalar primitives (Int, Float, Decimal, Timestamp, Utf8, Binary).

Per-group Accumulator stores state as Vec<ScalarValue>. For List<Struct<...>> a ScalarValue::List is a heap-allocated deep clone of the list value. update_batch clones on every candidate row and compares. Cost scales as #rows × #groups × sizeof(wide_ScalarValue), catastrophic on wide payload.

The columnar GroupsAccumulator path — one ArrayRef per expression, updated via arrow::compute::take in batches — would scale as #groups × wide_row × #expressions. For Scenario B: ~125 MB instead of 132 GB.

Additionally, ROW_NUMBER() OVER (...) WHERE rn = 1 has no logical rewrite today. Currently the only physical optimization is PartitionedTopKExec (via the opt-in enable_window_topn), but that operator has its own #groups × K × wide_row scaling problem — for Scenario B we measured OOM against a 16 GB pool even with K = 1.

Sub-issues

Part 1 — Extend first_value / last_value GroupsAccumulator to nested types (#23601, blocker for the rest)

Support List, LargeList, ListView, Struct, Map in groups_accumulator_supported() for first_value / last_value with ORDER BY
Implement columnar accumulator state: ArrayRef per expression, updated via arrow::compute::take when a candidate wins per group
Tests covering List<Utf8>, Struct<int, utf8>, nested List<Struct<...>>
Immediately fixes Performance of distinct on (columns) #16620 and removes the memory blowup on wide payload for Scenario B — no SQL rewrite required

Part 2 — Coalesce peer FIRST_VALUE(...) ORDER BY <o> expressions into a single struct accumulator (#23602, optional, but recommended)

Detect SELECT ..., FIRST_VALUE(a ORDER BY o), FIRST_VALUE(b ORDER BY o), ... GROUP BY p where all peer FIRST_VALUE expressions share the same ORDER BY
Rewrite as FIRST_VALUE(NAMED_STRUCT(a, b, ...) ORDER BY o) GROUP BY p internally
Cuts N argmax scans to 1 pass over the input; particularly beneficial for wide payload

Part 3 — Logical rewrite Filter(row_number() = 1) → Aggregate(FIRST_VALUE(... ORDER BY)) (#23603, user-visible auto-optimization)

Match Filter(Column(rn) = Literal(1)) → Projection? → WindowAggr(row_number() PARTITION BY p ORDER BY o)
- Also cover rn <= 1 and rn < 2 as equivalents
Preconditions: single ROW_NUMBER window function; filter is Column op Literal(1); no other consumers of rn
Rewrite to Aggregate(GROUP BY p, aggr=[first_value(cols ORDER BY o)])
Depends on Part 1 (otherwise emits slow Accumulator path); works best with Part 2
Semantics: ordering-key ties remain "unspecified", matching existing ROW_NUMBER behavior

Dependencies

Part 1 (GroupsAccumulator nested) — standalone, ships #16620 fix on its own
    │
    ├── Part 3 (rewrite rule) — auto-fires for users who wrote ROW_NUMBER()=1
    │
    └── Part 2 (coalesce peer FIRST_VALUE) — optional but recommended

Success criteria

Case	Before	Target
#16620 `DISTINCT ON` (~500K rows × 14 cols)	peak 1.4 GB, ~2 s	peak < 100 MB, sub-second
Wide-payload dedup (11M rows × ~2.5 KB)	OOM ~90 GB	peak < 500 MB, no SQL change

Performance of distinct on (columns) #16620 — Performance of distinct on (columns) — this epic subsumes it
Support max_by in Aggregation function #12252 — max_by aggregate — related aggregate primitive (merged)
Late materialization when LIMIT prunes heavily. #23263 — Late materialization when LIMIT prunes heavily — different family (scan-level, requires source-side random access + column late decode); complementary but not blocking
Perf: Window topn optimisation #21479 — PartitionedTopKExec (WindowTopN) — incompatible with wide payload on high group cardinality; this epic complements it

feat(optimizer): coalesce peer first_value / last_value into a single…

feat(functions-aggregate): support nested types (List, Struct, Map) i…

bench(functions-aggregate): add coalesce-peers first_value comparison

bench(functions-aggregate): benchmark coalescing peer first_value int…

A reproduction and measurements for the ROW_NUMBER() = 1 case on a wide payload, hoping it helps open PRs (#27197, #26160) and #26381.

I ran into this deduplicating wide revision data: the default window plan used several times the time and memory of a hand-written equivalent, so I'm sharing a self-contained reproduction.

Shape
each key (entity, flag, valid_time) has 4 revisions, and the newest one is kept, ranked by four columns. The payload is wide, 55 columns: 20 DECIMAL(15,5), 10 DOUBLE, 10 timestamps with a time zone, and 8 strings. In total there are 32 M rows, 8 M keys and 4.3 GB of Parquet in 4 files. Revisions arrive in time order, so the input is not sorted by key.

Generator (DataFusion SQL, run with datafusion-cli):

#!/bin/sh
# Writes gen.sql: N rows, R revisions per key, written in arrival order.
N=${N:-32000000}; R=${R:-4}; OUT=${OUT:-data}
G=$((N / R))
{
echo "COPY (SELECT"
echo "  'E' || lpad(CAST(g % 500 AS VARCHAR), 4, '0') AS entity,"
echo "  CAST(g % 997 = 0 AS BIGINT) AS flag,"
echo "  arrow_cast(to_timestamp_micros(1704067200000000 + (g / 500) * 300000000), 'Timestamp(Microsecond, Some(\"+10:00\"))') AS valid_time,"
echo "  CAST(rev / 2 AS BIGINT) AS revision,"
echo "  arrow_cast(to_timestamp_micros(1704067200000000 + (g / 500) * 300000000 + rev * 60000000), 'Timestamp(Microsecond, Some(\"+10:00\"))') AS revised_at,"
echo "  arrow_cast(to_timestamp_micros(1704067200000000 + (g / 500) * 300000000 + rev * 61000000), 'Timestamp(Microsecond, Some(\"+10:00\"))') AS received_at,"
echo "  'feed_' || CAST(rev AS VARCHAR) AS source,"
for k in $(seq 1 20); do echo "  CAST(((i * $((k * 31 + 7))) % 1000003) / 7.0 AS DECIMAL(15,5)) AS d$k,"; done
for k in $(seq 1 10); do echo "  CAST((i * $((k * 17 + 3))) % 100003 AS DOUBLE) / 13.0 AS f$k,"; done
for k in $(seq 1 10); do echo "  arrow_cast(to_timestamp_micros(1704067200000000 + ((i * $k) % 31536000) * 1000000), 'Timestamp(Microsecond, Some(\"+10:00\"))') AS t$k,"; done
for k in $(seq 1 7); do echo "  'code_' || CAST((i * $k) % $((k * 50)) AS VARCHAR) AS s$k,"; done
echo "  md5(CAST(i AS VARCHAR)) AS s8"
echo "FROM (SELECT i, i % $G AS g, i / $G AS rev FROM generate_series(0, $((N - 1))) AS t(i)))"
echo "TO '$OUT/' STORED AS PARQUET;"
} > gen.sql

Query:

CREATE EXTERNAL TABLE t STORED AS PARQUET LOCATION 'data/';
SELECT * FROM (
  SELECT *, row_number() OVER (
    PARTITION BY entity, flag, valid_time
    ORDER BY revision DESC NULLS LAST, revised_at DESC NULLS LAST,
             received_at DESC NULLS LAST, source DESC) AS rn
  FROM t) WHERE rn = 1;   -- and the same with rn <= 1

Results
These were run on main at 34eb6dc, with datafusion-cli built using --profile release-nonlto, on an 18-core Apple M-series machine with 48 GB of RAM. There was no memory limit. Each figure is the wall time and peak RSS of one run, measured with /usr/bin/time -l.

setting	predicate	time	peak RSS	plan
default	`rn = 1`	7.33 s	30.3 GB	`RepartitionExec` → `SortExec` (all 32 M rows × 55 columns) → `BoundedWindowAggExec` → `FilterExec`
default	`rn <= 1`	7.02 s	30.5 GB	same
`enable_window_topn = true`	`rn <= 1`	2.99 s	17.9 GB	`RepartitionExec` → `PartitionedTopKExec` (fetch=1)
`enable_window_topn = true`	`rn = 1`	7.05 s	30.5 GB	the rule does not fire for `= 1` (#26160)
`enable_row_number_to_aggregate = true`	`rn = 1`	not finished after 10 min	~14 GB when stopped	ran at about 170 % CPU, far slower than the window plan with this many columns

Observations

PartitionedTopKExec cuts the time by 2.4×. Its peak (17.9 GB) is still several times the ~5 GB of rows it keeps (8 M × 55 columns), which looks like PartitionedTopKExec keeps whole Utf8View input buffers alive for the rows it retains #26381 (input buffers kept alive). I have not yet confirmed this with schema_force_view_types = false.
PartitionedTopKExec has no spill path. I have not yet run these cases with --memory-limit.
In this shape, the first_value(... ORDER BY) rewrite was much slower than the window plan. That may be worth checking before chore(optimizer): enabling config for ReplaceFilterTop1 by default #27305 turns it on by default for wide payloads.
For comparison: on a real dataset of a similar shape (70 columns, 9.4 M rows across 5 schema versions), a hand-written equivalent took 4.7 s and 0.9 GB. It encodes the ranking columns with arrow-row, sorts a u32 index vector, and calls take once per month of data.

Navigation Menu

Repository navigation

[EPIC] Memory-efficient dedup on wide payload: fix ROW_NUMBER()=1 and DISTINCT ON patterns #23600

Motivation

Root cause

Sub-issues

Dependencies

Success criteria

Related

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

[EPIC] Memory-efficient dedup on wide payload: fix ROW_NUMBER()=1 and DISTINCT ON patterns #23600

Description

Motivation

Root cause

Sub-issues

Dependencies

Success criteria

Related

Activity

g0itb4 commented on Nov 11, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions