feat(benchmarks): Record the join failure matrix under a fixed memory budget

[jayzhan211]

Which issue does this PR close?

• Part of [EPIC] Spilling Hash Join — run any join in a bounded memory budget #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:

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

# 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:

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:

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 @ 61bf6b9.

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:

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.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 0% with 250 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.47%. Comparing base (61bf6b9) to head (984945b).
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
benchmarks/src/join_mem.rs	0.00%	249 Missing ⚠️
benchmarks/src/bin/dfbench.rs	0.00%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24779      +/-   ##
==========================================
- Coverage   81.52%   81.47%   -0.05%     
==========================================
  Files        1123     1124       +1     
  Lines      405970   406220     +250     
  Branches   405970   406220     +250     
==========================================
+ Hits       330978   330980       +2     
- Misses      55627    55876     +249     
+ Partials    19365    19364       -1     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[getChan]

I ran the benchmark locally using the PR's default configuration. The results are
consistent with the baseline reported in the PR description. The success/failure
outcomes, allocation sizes, and spill metrics matched the recorded baseline; only
wall-clock times varied by environment.

Environment for the numbers below: Apple M5 (10 cores), macOS 26.5.2, --release,
Rust 1.97.0, DataFusion PR #24779 at 984945b.

Command:

cd benchmarks
./bench.sh run join_mem

The benchmark was run with the default configuration: a 300MB fair memory pool, 4
partitions, 20M rows, and 3 iterations.

 #      configuration       outcome                 mean    spilled
━━━━━  ━━━━━━━━━━━━━━━━━━  ━━━━━━━━━━━━━━━━━━━━  ━━━━━━━━  ━━━━━━━━━━━━━━━━━━━━━━━
 1      default settings    ❌ Resources               -    —
        — HashJoinExec      exhausted at the
                            hash-table build
─────  ──────────────────  ────────────────────  ────────  ───────────────────────
 2      workaround —        ✅ completes          0.437s    SortExec 310.6 MB in
        prefer_hash_join                                    28 events
        =false
─────  ──────────────────  ────────────────────  ────────  ───────────────────────
 3a     build side          ✅ completes          0.090s    nothing
        filtered to N =
        10M
─────  ──────────────────  ────────────────────  ────────  ───────────────────────
 3b     build side          ❌ Resources               -    —
        filtered to N =     exhausted at the
        12M                 hash-table build
─────  ──────────────────  ────────────────────  ────────  ───────────────────────
 4      row 3a forced       ✅ completes          0.334s    SortExec 233.0 MB in
        through                                             24 events
        SortMergeJoinExe
        c
─────  ──────────────────  ────────────────────  ────────  ───────────────────────
 5      hash aggregation    ✅ completes          1.037s    AggregateExec 1079.4
        through the same                                    MB in 84 events
        budget

The recorded allocation failures were:

• Row 1: additional allocation of 95.4 MB with 38.2 MB already allocated
• Row 3b: additional allocation of 57.2 MB with 22.9 MB already allocated

SMJ tax (row 4 / row 3a): 3.7x on a join that would have fit in memory.

Every row matched the recorded baseline.