jayzhan211 opened a new pull request, #24779:
URL: https://github.com/apache/datafusion/pull/24779
## Which issue does this PR close?
- Part of #24768 — the baseline the rest of that epic is measured against.
Nothing to close yet.
## Rationale for this change
A join whose build side does not fit the memory budget fails outright today:
`HashJoinExec` cannot spill, so the query dies at the hash table build
rather than
slowing down. The only way to run such a join is `prefer_hash_join=false`,
which hands
the work to `SortMergeJoinExec`, whose sorts *can* spill — and which then
also applies
to every join that would have fit in memory, at a measurable cost.
That behaviour is well known, but nothing in-tree measures it. There is no
baseline to
hold external hash join work against, no way to see a row of it change, and
no shared
artifact for the numbers quoted in discussions. `SortMergeJoinExec` has
memory-limit
validation coverage; `HashJoinExec` has none.
This PR adds that baseline as a benchmark and as tests. It does not change
any execution
behaviour.
## What changes are included in this PR?
**`dfbench join-mem` / `./bench.sh run join_mem`**
(`benchmarks/src/join_mem.rs`) — one
join workload run through a fixed budget in each configuration a user can
choose today.
Both inputs read the same generated relation, so there is no smaller side
for the planner
to swap in: the failure is not one a better build side avoids.
For each row it reports the outcome next to the recorded baseline, the
allocation that
failed, per-operator spill metrics, and the "SMJ tax" — row 4 divided by row
3a, i.e.
what the workaround costs on a join that would have fit. A row whose outcome
differs from
the recorded baseline is called out by name; rows that fail with `Resources
exhausted`
are the recorded baseline, not a broken run.
**`memory_limit::budgeted_env`** — a `RuntimeEnv` on a budgeted fair pool,
plus
`run_under_budget()`, which reports what a query did under it: completed or
exhausted,
how long it took, and which operators spilled. The existing `TestCase`
covers queries
over the built-in scenario tables and asserts on error text; this covers the
other shape,
where *whether* a query finished, and what it spilled to get there, is the
subject.
**`memory_limit::join_failure_matrix`** — the same six rows as assertions at
test scale
(16MB budget, 2M rows, 2 partitions), one test per row, so a flip names the
row that
flipped. Rows 1 and 3b are the ones external hash join is meant to turn into
`completes`;
when that lands, these are the tests to update.
Docs: a section in `benchmarks/README.md` and an entry in `bench.sh` (help,
data
dispatch, runner). Not added to the `all` group, since a run generates a
20M-row file.
## The baseline
Two queries, run six ways through a **300MB fair pool at 4 partitions over
20M rows**:
```sql
-- the join (rows 1, 2), and with the build side filtered (rows 3a, 3b, 4)
SELECT count(*) FROM t_probe p JOIN t_build b ON p.k = b.k;
SELECT count(*) FROM t_probe p JOIN (SELECT * FROM t_build WHERE k <= N) b
ON p.k = b.k;
-- the control (row 5)
SELECT count(DISTINCT payload) FROM t_build;
```
| # | configuration — what a user can actually set | outcome | mean |
spilled |
| --- | --- | --- | --- | --- |
| 1 | default settings — the planner picks `HashJoinExec` | ❌ `Resources
exhausted` at the hash-table build ¹ | 0.05s | — |
| 2 | the only workaround — `prefer_hash_join=false`, so the sorts (which
spill) carry the join | ✅ completes | 0.483s | `SortExec` 310.6 MB in 28 events
|
| 3a | where the ceiling is — build side filtered to `N = 10M` | ✅ completes
| 0.106s | nothing |
| 3b | just past it — the same filter at `N = 12M` | ❌ `Resources exhausted`
at the hash-table build ² | 0.048s | — |
| 4 | what the workaround costs when it isn't needed — row 3a forced through
`SortMergeJoinExec` | ✅ completes | 0.340s | `SortExec` 233.0 MB in 24 events |
| 5 | control, not a way to run the join — hash *aggregation* through the
same budget | ✅ completes | 1.028s | `AggregateExec` 1079.4 MB in 84 events |
¹ `Failed to allocate additional 95.4 MB for HashJoinInput[1] with 38.2 MB
already allocated for this reservation - 51.9 MB remain available for the total
memory pool: fair(pool_size: 300.0 MB)`
² `Failed to allocate additional 57.2 MB for HashJoinInput[0] with 22.9 MB
already allocated for this reservation - 36.6 MB remain available for the total
memory pool: fair(pool_size: 300.0 MB)`
**SMJ tax (row 4 / row 3a): 3.2x** on a join that would have fit in memory.
Reading the rows: the fatal allocation is the hash map itself, not the
batches. `count(*)`
projects the key alone (~8 B/row), and the map on top of it is sized by
`estimate_memory_size::<(u32, u64)>` at ~19 B/row — asked for as a *single*
`try_grow`
after every build batch has already been admitted. That is the 95.4 MB above
38.2 MB in
row 1, and the 57.2 MB above 22.9 MB in row 3b. Per-batch backpressure alone
cannot
avoid it. Row 5 is a control, not a way to run the join: it shows the budget
itself is
workable, pushing 1079MB of aggregation spill through the same 300MB pool.
## How to reproduce these numbers
```bash
# generates a 20M-row parquet file on the first run (~37MB, cached under
--path)
cd benchmarks && ./bench.sh run join_mem
```
or directly, which is what produced the table above:
```bash
cargo run --release --bin dfbench -- join-mem \
--path /tmp/join_mem -o /tmp/join_mem.json
```
Defaults, all overridable: `--memory-limit 300M`, `--mem-pool-type fair`,
`--partitions 4`,
`--iterations 3`, `--rows 20000000`, `--fit-rows 10000000` (rows 3a/4),
`--over-rows 12000000` (row 3b), and DataFusion's default 10MB sort spill
reservation.
`-q 3b` runs a single row; `-o` writes the usual benchmark JSON, which
carries
`pool_peak_bytes` for failed rows too (rows 1 and 3b peaked at 260MB and
276MB of the
300MB pool).
The pool is a `TrackConsumersPool` over a `FairSpillPool`, the same one
`datafusion-cli --mem-pool-type fair` installs, so the failure dumps are
directly
comparable with a CLI reproduction. Each row gets a fresh runtime, so no row
inherits
another's pool state.
The data is generated once, with no memory limit — building the file under a
300MB pool
is its own fight and not the thing under test:
```sql
COPY (SELECT v AS k, concat('payload-', v, '-', repeat('x', 24)) AS payload
FROM generate_series(1, 20000000) AS t(v))
TO 'join_mem_20000000_rows.parquet' STORED AS PARQUET;
```
Environment for the numbers above: Apple M4 Pro (12 cores), macOS,
`--release`, DataFusion
`main` @ 61bf6b96c.
What is and isn't stable across machines: **the allocation sizes are** —
`95.4 MB` above
`38.2 MB`, `57.2 MB` above `22.9 MB`, and 1079MB of aggregation spill
reproduce exactly,
because they are functions of row count, partition count and key width, not
of hardware.
**Wall-clock is not** — it varies with cores, disk and page cache, so the
SMJ tax (a ratio
of two rows from the same run) is the number to compare, not the individual
times. The
failing partition index and the trailing `... remain available` value vary
run to run with
how many sibling partitions reached their hash-map grow first.
To reproduce without the benchmark, the same matrix runs in `datafusion-cli`:
```bash
datafusion-cli -m 300M --mem-pool-type fair --top-memory-consumers 8
```
`--top-memory-consumers 8` matters: at the default of 3, the consumer that
actually fails
can be cut off from the dump.
## Are these changes tested?
Yes — the tests are part of the change. `memory_limit::join_failure_matrix`
asserts all six
rows at test scale; the full `memory_limit` module passes (42 tests, ~11s).
One note on the test-scale inputs: they are a generated parquet file rather
than
`generate_series` directly. A series is a sorted source with no statistics,
which lets the
sort-merge rows skip their sorts entirely and stops the planner from putting
the smaller
side on the build side — both of which quietly invalidate the matrix. That
cost me a
debugging round; the file says so, so the next person does not repeat it.
## Are there any user-facing changes?
No API or execution changes. New benchmark subcommand (`dfbench join-mem`,
`./bench.sh run join_mem`) and its documentation.
--
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]
