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: [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]
