You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "zhuqi-lucas (via GitHub)" <gi...@apache.org> on 2026/07/15 09:20:20 UTC
[I] [EPIC] Memory-efficient dedup on wide payload: fix ROW_NUMBER()=1 and DISTINCT ON patterns [datafusion]
zhuqi-lucas opened a new issue, #23600:
URL: https://github.com/apache/datafusion/issues/23600
## 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`](https://github.com/apache/datafusion/blob/main/datafusion/functions-aggregate/src/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
**#1 — Extend `first_value` / `last_value` `GroupsAccumulator` to nested types** (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 #16620** and removes the memory blowup on wide payload for Scenario B — no SQL rewrite required
**#2 — Coalesce peer `FIRST_VALUE(...) ORDER BY <o>` expressions into a single struct accumulator** (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
**#3 — Logical rewrite `Filter(row_number() = 1) → Aggregate(FIRST_VALUE(... ORDER BY))`** (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 #1 (otherwise emits slow `Accumulator` path); works best with #2
- Semantics: ordering-key ties remain "unspecified", matching existing `ROW_NUMBER` behavior
## Dependencies
```
#1 (GroupsAccumulator nested) — standalone, ships #16620 fix on its own
│
├── #3 (rewrite rule) — auto-fires for users who wrote ROW_NUMBER()=1
│
└── #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 |
## Related
- #16620 — Performance of `distinct on (columns)` — this epic subsumes it
- #12252 — `max_by` aggregate — related aggregate primitive (merged)
- #23263 — Late materialization when LIMIT prunes heavily — different family (scan-level, requires source-side random access + column late decode); complementary but not blocking
- #21479 — `PartitionedTopKExec` (WindowTopN) — incompatible with wide payload on high group cardinality; this epic complements it
--
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: github-unsubscribe@datafusion.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] [EPIC] Memory-efficient dedup on wide payload: fix ROW_NUMBER()=1 and DISTINCT ON patterns [datafusion]
Posted by "g0itb4 (via GitHub) via github" <gi...@datafusion.apache.org>.
g0itb4 commented on issue #23600:
URL: https://github.com/apache/datafusion/issues/23600#issuecomment-6794753368
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`):
```sh
#!/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:**
```sql
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 34eb6dc5bf, 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 #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 #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.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org