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

benchmarks/README.md

@@ -964,6 +964,40 @@ Several queries are included to test sort merge joins under various workloads.
 
 ./bench.sh run smj
 ```
+
+## Memory-Limited Join
+
+One join workload through a fixed memory budget (300MB by default), in each configuration a user
+can pick today. `HashJoinExec` cannot spill its build side, so some rows are *expected* to fail
+with `Resources exhausted`: that matrix is the recorded baseline for external hash join work, and
+the benchmark prints each row's outcome next to it and calls out any row that flipped.
+
+| row | role | baseline |
+| --- | --- | --- |
+| 1 | default settings — the planner picks `HashJoinExec` | fails |
+| 2 | the only workaround — `prefer_hash_join=false`, so the sorts (which spill) carry the join | completes |
+| 3a | where the ceiling is — the build side filtered to a size that fits | completes |
+| 3b | just past the ceiling — the same filter, slightly larger | fails |
+| 4 | what the workaround costs when it isn't needed — row 3a forced through `SortMergeJoinExec` | completes, ~3.5x slower than 3a |
+| 5 | control, not a way to run the join — hash *aggregation* through the same budget | completes |
+
+Row 4 divided by row 3a is reported as the "SMJ tax". Both inputs are the same generated relation
+(20M rows, written on the first run and cached under `--path`), so there is no smaller side for the
+planner to swap in. Spill metrics are reported per operator, and `-q` runs a single row.
+
+### Example Run
+
+```bash
+# Data is generated on the first run
+./bench.sh run join_mem
+
+# Or directly, e.g. only the ceiling row, under a different budget
+cargo run --release --bin dfbench -- join-mem --query 3b --memory-limit 512M
+```
+
+The same matrix is asserted at test scale in
+`datafusion/core/tests/memory_limit/join_failure_matrix.rs`.
+
 ## Cancellation
 
 Test performance of cancelling queries.

benchmarks/bench.sh

@@ -157,6 +157,8 @@ cancellation:           How long cancelling a query takes
 nlj:                    Benchmark for simple nested loop joins, testing various join scenarios
 hj:                     Benchmark for simple hash joins, testing various join scenarios
 smj:                    Benchmark for simple sort merge joins, testing various join scenarios
+join_mem:               One join workload through a fixed memory budget (default 300M), in each configuration a user can pick today.
+                          Rows failing with 'Resources exhausted' are the recorded baseline (hash join cannot spill), not a broken run
 dict:                   Benchmark for dictionary-encoded group-by scenarios
 compile_profile:        Compile and execute TPC-H across selected Cargo profiles, reporting timing and binary size
 
@@ -374,6 +376,10 @@ main() {
                     # smj uses range() function, no data generation needed
                     echo "SMJ benchmark does not require data generation"
                     ;;
+                join_mem)
+                    # join_mem generates its own parquet file on first run
+                    echo "join_mem benchmark generates its data on first run"
+                    ;;
                 dict)
                     # dict generates in-memory data, no data generation needed
                     echo "DICT benchmark does not require data generation"
@@ -604,6 +610,9 @@ main() {
                 smj)
                     run_smj
                     ;;
+                join_mem)
+                    run_join_mem
+                    ;;
                 dict)
                     run_dict
                     ;;
@@ -1577,6 +1586,15 @@ run_smj() {
     debug_run $CARGO_COMMAND --bin dfbench -- smj --iterations 5 -o "${RESULTS_FILE}" ${QUERY_ARG} ${LATENCY_ARG}
 }
 
+# Runs the memory-limited join benchmark (the join failure matrix)
+run_join_mem() {
+    JOIN_MEM_DIR="${DATA_DIR}/join_mem"
+    RESULTS_FILE="${RESULTS_DIR}/join_mem.json"
+    echo "RESULTS_FILE: ${RESULTS_FILE}"
+    echo "Running join_mem benchmark..."
+    debug_run $CARGO_COMMAND --bin dfbench -- join-mem --iterations 3 --path "${JOIN_MEM_DIR}" -o "${RESULTS_FILE}" ${QUERY_ARG}
+}
+
 # Runs the dict benchmark
 run_dict() {
     RESULTS_FILE="${RESULTS_DIR}/dict.json"

benchmarks/src/bin/dfbench.rs

@@ -32,8 +32,8 @@ static ALLOC: snmalloc_rs::SnMalloc = snmalloc_rs::SnMalloc;
 static ALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc;
 
 use datafusion_benchmarks::{
-    cancellation, clickbench, dict, h2o, hj, imdb, nlj, smj, sort_tpch, statistics,
-    tpcds, tpch,
+    cancellation, clickbench, dict, h2o, hj, imdb, join_mem, nlj, smj, sort_tpch,
+    statistics, tpcds, tpch,
 };
 
 #[derive(Debug, Parser)]
@@ -51,6 +51,7 @@ enum Options {
     H2o(h2o::RunOpt),
     HJ(hj::RunOpt),
     Imdb(imdb::RunOpt),
+    JoinMem(join_mem::RunOpt),
     Nlj(nlj::RunOpt),
     Smj(smj::RunOpt),
     Statistics(statistics::RunOpt),
@@ -73,6 +74,7 @@ pub async fn main() -> Result<()> {
         Options::H2o(opt) => opt.run().await,
         Options::HJ(opt) => opt.run().await,
         Options::Imdb(opt) => Box::pin(opt.run()).await,
+        Options::JoinMem(opt) => opt.run().await,
         Options::Nlj(opt) => opt.run().await,
         Options::Smj(opt) => opt.run().await,
         Options::Statistics(opt) => opt.run().await,