diff --git a/benchmarks/README.md b/benchmarks/README.md index df96b1aeeb140..5a555e15abda2 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -966,6 +966,36 @@ Several queries are included to test hash joins under various workloads. ./bench.sh run hj ``` +## Null-Aware Join + +This benchmark focuses on `NOT IN` subqueries, which plan as null-aware joins: an +outer row that finds no match is TRUE only if neither side has a NULL in scope, +and UNKNOWN otherwise. + +Deciding which outer rows are UNKNOWN is cheap when the `NOT IN` is uncorrelated +(Q01-Q03) and is linear in the table sizes. When the subquery is correlated, the +correlation predicate stays behind as a join filter, and the join has to evaluate +that filter per candidate (build row x probe row) pair to decide which rows the +NULLs actually reach. A non-equality correlation leaves no equality key to narrow +those pairs, so Q04-Q07 measure how that cost grows with the NULL fraction, while +Q08 adds an equality correlation that turns the candidate pairs into a hash +lookup. + +Both table sizes are knobs: `NAJ_ROWS` (default `10000`) sizes the correlated +queries, whose cost grows with its square, and `NAJ_LARGE_ROWS` (default +`1000000`) sizes the uncorrelated ones. + +### Example Run + +```bash +# No need to generate data: this benchmark uses table function `range()` as the data source + +./bench.sh run null_aware_join + +# Or, with more rows for the correlated queries (~4x the per-pair filter work) +NAJ_ROWS=20000 ./bench.sh run null_aware_join +``` + ## Sort Merge Join This benchmark focuses on the performance of queries with sort merge joins, minimizing other overheads such as scanning data sources or evaluating predicates. diff --git a/benchmarks/bench.sh b/benchmarks/bench.sh index 082651656e0f8..485f754a9b2ee 100755 --- a/benchmarks/bench.sh +++ b/benchmarks/bench.sh @@ -112,6 +112,9 @@ parquet_row_filter_skip: Per-RG fully-matched RowFilter skip on Parquet (apache/ range filter + pushdown, so most row groups are fully matched and the per-row RowFilter is skipped on them (subgroups via BENCH_SUBGROUP: skip = clustered key so the skip fires, control = scrambled key so it never fires) (data generated inline by the suite's load SQL; knobs: PRED_ROWS, RG_SIZE) +null_aware_join: Null-aware (NOT IN) hash join micro-benchmarks: uncorrelated, non-equality-correlated and equality-correlated + NOT IN across NULL fractions, to measure the per-pair join-filter work the correlated cases do + (data generated inline by the suite's load SQL from range(); knobs: NAJ_ROWS, NAJ_LARGE_ROWS) # ClickBench Benchmarks clickbench_1: ClickBench queries against a single parquet file @@ -271,6 +274,10 @@ main() { # Data is generated inline by the suite's load SQL (COPY). echo "parquet_row_filter_skip: no external data to generate" ;; + null_aware_join) + # Data is generated inline by the suite's load SQL from range(). + echo "null_aware_join: no external data to generate" + ;; asof_join) data_asof_join ;; @@ -490,6 +497,7 @@ main() { run_tpcds run_smj run_dict + run_null_aware_join ;; tpch) run_tpch "1" "parquet" @@ -518,6 +526,9 @@ main() { parquet_row_filter_skip) run_parquet_row_filter_skip ;; + null_aware_join) + run_null_aware_join + ;; asof_join) run_asof_join ;; @@ -933,6 +944,27 @@ run_parquet_row_filter_skip() { bash -c "$SQL_CARGO_COMMAND" } +# Runs the null_aware_join suite: NOT IN (null-aware) hash joins. The load SQL +# builds every table inline from range(), so there is no data step. +# +# Q01-Q03 are uncorrelated NOT IN and are linear in the table size; they are the +# regression guard for the plain null-aware path. Q04-Q08 are correlated, where +# the correlation predicate stays behind as a join filter that the join applies +# per candidate (build row x probe row) pair while deciding which rows are +# UNKNOWN; with no equality correlation there are no scope keys to narrow those +# pairs, so Q05-Q07 scale with the NULL count times the opposite table's size. +# +# NAJ_ROWS rows per table for the correlated queries (default 10_000) +# NAJ_LARGE_ROWS rows per table for the uncorrelated queries (default 1_000_000) +run_null_aware_join() { + echo "Running null_aware_join benchmark (rows=${NAJ_ROWS:-10000}, large_rows=${NAJ_LARGE_ROWS:-1000000})..." + debug_run env BENCH_NAME=null_aware_join \ + NAJ_ROWS="${NAJ_ROWS:-10000}" \ + NAJ_LARGE_ROWS="${NAJ_LARGE_ROWS:-1000000}" \ + ${QUERY:+BENCH_QUERY="${QUERY}"} \ + bash -c "$SQL_CARGO_COMMAND" +} + # Runs the tpch in memory (needs tpch parquet data) run_tpch_mem() { SCALE_FACTOR=$1 diff --git a/benchmarks/sql_benchmarks/README.md b/benchmarks/sql_benchmarks/README.md index c5a4136e8e5ea..fb09c95daa12f 100644 --- a/benchmarks/sql_benchmarks/README.md +++ b/benchmarks/sql_benchmarks/README.md @@ -37,6 +37,7 @@ in the community: | `hj` | Hash join benchmark | | `imdb` | IMDb benchmark | | `nlj` | Nested‑loop join benchmark | +| `null_aware_join` | Null-aware (`NOT IN`) hash join micro-benchmarks. Q01-Q03 are uncorrelated `NOT IN` across NULL fractions and are linear in the table size (`NAJ_LARGE_ROWS`, default `1000000`). Q04-Q08 are correlated, so the correlation predicate stays behind as a join filter that the join applies per candidate (build row × probe row) pair while deciding which outer rows are UNKNOWN; without an equality correlation there are no scope keys to narrow those pairs, so their cost grows with the square of `NAJ_ROWS` (default `10000`). Q08 adds an equality correlation, which turns those pairs into a hash lookup. All tables are built inline from `range()`, so there is no data step. | | `push_down_topk` | `ORDER BY ... LIMIT` over outer joins (TPC-H data); exercises pushing a TopK through a join | | `smj` | Sort‑merge join benchmark | | `sort tpch` | Sorting benchmarks against the TPC-H lineitem table | diff --git a/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q01.benchmark b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q01.benchmark new file mode 100644 index 0000000000000..b2f9e0ab49c4e --- /dev/null +++ b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q01.benchmark @@ -0,0 +1,26 @@ +name Q01 +group null_aware_join + +load sql_benchmarks/null_aware_join/init/load.sql + +# Correctness canary: the NOT IN result must match a reference count +# that does not use NOT IN. It holds for every NAJ_ROWS / NAJ_LARGE_ROWS. +# Odd ids are never in the subquery, which holds only even ids. +assert I +SELECT count(*) = (SELECT count(*) FROM large_outer WHERE id % 2 = 1) +FROM large_outer o +WHERE o.id NOT IN (SELECT i.id FROM large_inner i); +---- +true + +expect_plan HashJoinExec + +run +-- Q1: uncorrelated NOT IN, non-nullable key on both sides. +-- No null-aware handling is needed at all, so this is the floor for the shape +-- and a guard against the anti-join path getting slower for everyone else. +SELECT count(*) +FROM large_outer o +WHERE o.id NOT IN (SELECT i.id FROM large_inner i); + +cleanup sql_benchmarks/null_aware_join/init/cleanup.sql diff --git a/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q02.benchmark b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q02.benchmark new file mode 100644 index 0000000000000..ba8554b1cb823 --- /dev/null +++ b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q02.benchmark @@ -0,0 +1,27 @@ +name Q02 +group null_aware_join + +load sql_benchmarks/null_aware_join/init/load.sql + +# Correctness canary: the NOT IN result must match a reference count +# that does not use NOT IN. It holds for every NAJ_ROWS / NAJ_LARGE_ROWS. +# The subquery always holds a NULL, so no outer row is TRUE. +assert I +SELECT count(*) = 0 +FROM large_outer o +WHERE o.id NOT IN (SELECT i.id_n1 FROM large_inner i); +---- +true + +expect_plan HashJoinExec +expect_plan null_aware: true + +run +-- Q2: uncorrelated NOT IN, 1% NULL on the subquery side. +-- A single NULL anywhere in the subquery makes every unmatched outer row +-- UNKNOWN, which is the cheap bulk path. Linear in the table sizes. +SELECT count(*) +FROM large_outer o +WHERE o.id NOT IN (SELECT i.id_n1 FROM large_inner i); + +cleanup sql_benchmarks/null_aware_join/init/cleanup.sql diff --git a/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q03.benchmark b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q03.benchmark new file mode 100644 index 0000000000000..28f1ed3410a8a --- /dev/null +++ b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q03.benchmark @@ -0,0 +1,26 @@ +name Q03 +group null_aware_join + +load sql_benchmarks/null_aware_join/init/load.sql + +# Correctness canary: the NOT IN result must match a reference count +# that does not use NOT IN. It holds for every NAJ_ROWS / NAJ_LARGE_ROWS. +# The non-NULL outer ids are odd, and the subquery holds only even ids. +assert I +SELECT count(*) = (SELECT count(*) FROM large_outer WHERE id % 2 = 1) +FROM large_outer o +WHERE o.id_n50 NOT IN (SELECT i.id FROM large_inner i); +---- +true + +expect_plan HashJoinExec +expect_plan null_aware: true + +run +-- Q3: uncorrelated NOT IN, 50% NULL on the outer side. +-- Every NULL outer row is UNKNOWN on its own. Linear in the table sizes. +SELECT count(*) +FROM large_outer o +WHERE o.id_n50 NOT IN (SELECT i.id FROM large_inner i); + +cleanup sql_benchmarks/null_aware_join/init/cleanup.sql diff --git a/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q04.benchmark b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q04.benchmark new file mode 100644 index 0000000000000..c72b5bceed74f --- /dev/null +++ b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q04.benchmark @@ -0,0 +1,34 @@ +name Q04 +group null_aware_join + +load sql_benchmarks/null_aware_join/init/load.sql + +# Correctness canary: the NOT IN result must match a reference count +# that does not use NOT IN. It holds for every NAJ_ROWS / NAJ_LARGE_ROWS. +# A row is TRUE when the subquery is empty, or when its key is not NULL and +# not in the subquery. `id` = 2 * v is in the subquery when v % 1000 < o.z. +assert I +SELECT count(*) = ( + SELECT count(*) FROM small_outer o + WHERE o.z <= (SELECT min(z) FROM small_inner) + OR (o.id_n0 IS NOT NULL AND NOT (o.id % 2 = 0 AND (o.id / 2) % 1000 < o.z)) +) +FROM small_outer o +WHERE o.id_n0 NOT IN (SELECT i.id_n0 FROM small_inner i WHERE i.z < o.z); +---- +true + +expect_plan HashJoinExec +expect_plan null_aware: true + +run +-- Q4: non-equality-correlated NOT IN, nullable keys that hold no NULL. +-- The correlation `i.z < o.z` has no equality, so it stays behind as a join +-- filter and the join has no correlation scope keys. With no NULL on either +-- side there is nothing to mark, so this is the zero-NULL baseline: it shows +-- what the correlated NOT IN shape costs before any per-pair filter work. +SELECT count(*) +FROM small_outer o +WHERE o.id_n0 NOT IN (SELECT i.id_n0 FROM small_inner i WHERE i.z < o.z); + +cleanup sql_benchmarks/null_aware_join/init/cleanup.sql diff --git a/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q05.benchmark b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q05.benchmark new file mode 100644 index 0000000000000..2d0a04077d5cf --- /dev/null +++ b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q05.benchmark @@ -0,0 +1,19 @@ +name Q05 +group null_aware_join + +load sql_benchmarks/null_aware_join/init/load.sql + +expect_plan HashJoinExec +expect_plan null_aware: true + +run +-- Q5: non-equality-correlated NOT IN, 1% NULL on the outer side. +-- With no correlation scope keys the join filter is evaluated over every +-- (NULL outer row x probe row) pair, so the work scales with the NULL count +-- times the subquery size. At 1% NULL that product is still small; Q06 is the +-- same query at 50%. +SELECT count(*) +FROM small_outer o +WHERE o.id_n1 NOT IN (SELECT i.id_n0 FROM small_inner i WHERE i.z < o.z); + +cleanup sql_benchmarks/null_aware_join/init/cleanup.sql diff --git a/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q06.benchmark b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q06.benchmark new file mode 100644 index 0000000000000..668476ded9571 --- /dev/null +++ b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q06.benchmark @@ -0,0 +1,20 @@ +name Q06 +group null_aware_join + +load sql_benchmarks/null_aware_join/init/load.sql + +expect_plan HashJoinExec +expect_plan null_aware: true + +run +-- Q6: non-equality-correlated NOT IN, 50% NULL on the outer side. +-- Worst case for the pairing that has no correlation scope keys: half the +-- outer rows are NULL, and the join filter is evaluated for each of them +-- against every subquery row, including for outer rows already known to be +-- UNKNOWN. Compare against Q05 (same query, 1% NULL) to see how the cost +-- tracks the NULL fraction. +SELECT count(*) +FROM small_outer o +WHERE o.id_n50 NOT IN (SELECT i.id_n0 FROM small_inner i WHERE i.z < o.z); + +cleanup sql_benchmarks/null_aware_join/init/cleanup.sql diff --git a/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q07.benchmark b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q07.benchmark new file mode 100644 index 0000000000000..38bf7e9048bc9 --- /dev/null +++ b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q07.benchmark @@ -0,0 +1,18 @@ +name Q07 +group null_aware_join + +load sql_benchmarks/null_aware_join/init/load.sql + +expect_plan HashJoinExec +expect_plan null_aware: true + +run +-- Q7: non-equality-correlated NOT IN, 50% NULL on the subquery side. +-- The mirror of Q06: a NULL subquery row makes every outer row in its +-- correlation scope UNKNOWN, so the filter is evaluated over every +-- (outer row x NULL probe row) pair. +SELECT count(*) +FROM small_outer o +WHERE o.id_n0 NOT IN (SELECT i.id_n50 FROM small_inner i WHERE i.z < o.z); + +cleanup sql_benchmarks/null_aware_join/init/cleanup.sql diff --git a/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q08.benchmark b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q08.benchmark new file mode 100644 index 0000000000000..39d4fbd4ced22 --- /dev/null +++ b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q08.benchmark @@ -0,0 +1,26 @@ +name Q08 +group null_aware_join + +load sql_benchmarks/null_aware_join/init/load.sql + +expect_plan HashJoinExec + +run +-- Q8: NOT IN correlated by both an equality and a non-equality, 50% NULL on +-- the outer side. The equality correlation becomes a correlation scope key, so +-- the candidate pairs come from a hash lookup on `k` (16 groups here) instead +-- of the full cross product, and only those pairs reach the join filter. Same +-- NULL fraction as Q06, so the gap between the two is what the scope key buys. +-- +-- The `OR` keeps this a mark join: a bare `WHERE ... NOT IN` with an equality +-- correlation is the shape that does not plan at all before this suite's +-- motivating fix, and the point here is to measure the scope-key path, not to +-- pick a shape that only runs on one branch. +SELECT count(*) +FROM small_outer o +WHERE o.z > 900 + OR o.id_n50 NOT IN ( + SELECT i.id_n0 FROM small_inner i WHERE i.k = o.k AND i.z < o.z + ); + +cleanup sql_benchmarks/null_aware_join/init/cleanup.sql diff --git a/benchmarks/sql_benchmarks/null_aware_join/init/cleanup.sql b/benchmarks/sql_benchmarks/null_aware_join/init/cleanup.sql new file mode 100644 index 0000000000000..f92df3fdebc7d --- /dev/null +++ b/benchmarks/sql_benchmarks/null_aware_join/init/cleanup.sql @@ -0,0 +1,4 @@ +DROP TABLE small_outer; +DROP TABLE small_inner; +DROP TABLE large_outer; +DROP TABLE large_inner; diff --git a/benchmarks/sql_benchmarks/null_aware_join/init/load.sql b/benchmarks/sql_benchmarks/null_aware_join/init/load.sql new file mode 100644 index 0000000000000..1225c719951a7 --- /dev/null +++ b/benchmarks/sql_benchmarks/null_aware_join/init/load.sql @@ -0,0 +1,55 @@ +-- Synthetic tables for the null-aware (NOT IN) hash join benchmarks. Built from +-- `range()`, so there is no data generation step. +-- +-- Columns, on both sides: +-- id Int64 NOT NULL -- NOT IN over this never needs null-aware handling +-- id_n0 Int64 NULL, 0% NULL -- nullable, but no NULL is ever present +-- id_n1 Int64 NULL, 1% NULL +-- id_n50 Int64 NULL, 50% NULL +-- z correlation value for a non-equality correlation (`i.z < o.z`) +-- k correlation key for an equality correlation (`i.k = o.k`), 16 groups +-- +-- `id_n0` is nullable so the planner still builds a null-aware join, but holds no +-- NULL, which is what makes it the zero-NULL baseline for the correlated queries. +-- `id` is only ~half covered by the subquery side, so the anti join returns rows +-- rather than degenerating to an empty or full result. + +-- Small tables: used by the correlated queries (Q04-Q08), whose cost grows with +-- the product of the two table sizes. +CREATE TABLE small_outer AS +SELECT + value AS id, + CASE WHEN value < 0 THEN NULL ELSE value END AS id_n0, + CASE WHEN value % 100 = 0 THEN NULL ELSE value END AS id_n1, + CASE WHEN value % 2 = 0 THEN NULL ELSE value END AS id_n50, + value % 1000 AS z, + value % 16 AS k +FROM range(0, ${NAJ_ROWS:-10000}); + +CREATE TABLE small_inner AS +SELECT + value * 2 AS id, + CASE WHEN value < 0 THEN NULL ELSE value * 2 END AS id_n0, + CASE WHEN value % 100 = 0 THEN NULL ELSE value * 2 END AS id_n1, + CASE WHEN value % 2 = 0 THEN NULL ELSE value * 2 END AS id_n50, + value % 1000 AS z, + value % 16 AS k +FROM range(0, ${NAJ_ROWS:-10000}); + +-- Large tables: used by the uncorrelated queries (Q01-Q03), whose cost is linear +-- in the table size. +CREATE TABLE large_outer AS +SELECT + value AS id, + CASE WHEN value < 0 THEN NULL ELSE value END AS id_n0, + CASE WHEN value % 100 = 0 THEN NULL ELSE value END AS id_n1, + CASE WHEN value % 2 = 0 THEN NULL ELSE value END AS id_n50 +FROM range(0, ${NAJ_LARGE_ROWS:-1000000}); + +CREATE TABLE large_inner AS +SELECT + value * 2 AS id, + CASE WHEN value < 0 THEN NULL ELSE value * 2 END AS id_n0, + CASE WHEN value % 100 = 0 THEN NULL ELSE value * 2 END AS id_n1, + CASE WHEN value % 2 = 0 THEN NULL ELSE value * 2 END AS id_n50 +FROM range(0, ${NAJ_LARGE_ROWS:-1000000}); diff --git a/benchmarks/sql_benchmarks/null_aware_join/null_aware_join.suite b/benchmarks/sql_benchmarks/null_aware_join/null_aware_join.suite new file mode 100644 index 0000000000000..559363de0610a --- /dev/null +++ b/benchmarks/sql_benchmarks/null_aware_join/null_aware_join.suite @@ -0,0 +1,31 @@ +description = "Null-aware (NOT IN) hash join benchmarks: uncorrelated, non-equality-correlated, and equality-correlated NOT IN across NULL fractions" + +query_pattern = "q{QUERY_ID_PADDED}.benchmark" + +[[options]] +name = "rows" +short = "r" +env = "NAJ_ROWS" +default = "10000" +values = ["10000", "..."] +help = "Rows per table for the correlated NOT IN queries (Q04-Q08). These evaluate the join filter over candidate build x probe pairs, so their cost grows with the square of this value." + +[[options]] +name = "large-rows" +short = "l" +env = "NAJ_LARGE_ROWS" +default = "1000000" +values = ["1000000", "..."] +help = "Rows per table for the uncorrelated NOT IN queries (Q01-Q03), whose cost is linear in the table size." + +[[examples]] +command = "cargo run --release --bin benchmark_runner -- null_aware_join" +description = "Run all null-aware join queries at the default sizes." + +[[examples]] +command = "cargo run --release --bin benchmark_runner -- null_aware_join --query 6" +description = "Run only Q06, the worst case for filter evaluation over NULL build rows." + +[[examples]] +command = "cargo run --release --bin benchmark_runner -- null_aware_join -r 20000" +description = "Run with 20,000 rows per table for the correlated queries (~4x the filter work of the default)." diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 688ce6f866a9b..7b9e701119ef4 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -920,6 +920,7 @@ impl fmt::Debug for HashJoinExec { .field("left_fut", &self.left_fut) .field("random_state", &self.random_state) .field("mode", &self.mode) + .field("null_aware", &self.null_aware) .field("metrics", &self.metrics) .field("projection", &self.projection) .field("column_indices", &self.column_indices)