diff --git a/benchmarks/sql_benchmarks/README.md b/benchmarks/sql_benchmarks/README.md index 8b68311531c9c..7866f9ebd6a78 100644 --- a/benchmarks/sql_benchmarks/README.md +++ b/benchmarks/sql_benchmarks/README.md @@ -39,6 +39,7 @@ in the community: | `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 | +| `pwmj` | Piecewise merge join: classic `INNER`/`LEFT`/`RIGHT`/`FULL` joins over `lhs.key < rhs.key`. Subgroups (`--subgroup`) are match regimes: `no_match`, `null_heavy`, `half_match`, `selective`, `all_match`. Generated with `range()`, no data needed. | | `smj` | Sort‑merge join benchmark | | `sort tpch` | Sorting benchmarks against the TPC-H lineitem table | | `taxi` | NYC taxi dataset benchmark | diff --git a/benchmarks/sql_benchmarks/pwmj/benchmarks/q01.benchmark b/benchmarks/sql_benchmarks/pwmj/benchmarks/q01.benchmark new file mode 100644 index 0000000000000..0d0ae3eb9547a --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/benchmarks/q01.benchmark @@ -0,0 +1,11 @@ +# INNER JOIN, no_match: every streamed key is below every buffered key, so no row matches. + +subgroup no_match + +template sql_benchmarks/pwmj/pwmj.benchmark.template +QPAD=01 +JOIN=INNER +JOIN_KIND=Inner +RIGHT_ROWS=2000000 +RIGHT_KEY=-(h % 1000) +EXPECTED_ROWS=0 diff --git a/benchmarks/sql_benchmarks/pwmj/benchmarks/q02.benchmark b/benchmarks/sql_benchmarks/pwmj/benchmarks/q02.benchmark new file mode 100644 index 0000000000000..c4b4e590760c1 --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/benchmarks/q02.benchmark @@ -0,0 +1,11 @@ +# LEFT JOIN, no_match: every streamed key is below every buffered key, so no row matches. + +subgroup no_match + +template sql_benchmarks/pwmj/pwmj.benchmark.template +QPAD=02 +JOIN=LEFT +JOIN_KIND=Left +RIGHT_ROWS=2000000 +RIGHT_KEY=-(h % 1000) +EXPECTED_ROWS=100000 diff --git a/benchmarks/sql_benchmarks/pwmj/benchmarks/q03.benchmark b/benchmarks/sql_benchmarks/pwmj/benchmarks/q03.benchmark new file mode 100644 index 0000000000000..6467d66afb362 --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/benchmarks/q03.benchmark @@ -0,0 +1,11 @@ +# RIGHT JOIN, no_match: every streamed key is below every buffered key, so no row matches. + +subgroup no_match + +template sql_benchmarks/pwmj/pwmj.benchmark.template +QPAD=03 +JOIN=RIGHT +JOIN_KIND=Right +RIGHT_ROWS=2000000 +RIGHT_KEY=-(h % 1000) +EXPECTED_ROWS=2000000 diff --git a/benchmarks/sql_benchmarks/pwmj/benchmarks/q04.benchmark b/benchmarks/sql_benchmarks/pwmj/benchmarks/q04.benchmark new file mode 100644 index 0000000000000..163bcd9572c76 --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/benchmarks/q04.benchmark @@ -0,0 +1,11 @@ +# FULL JOIN, no_match: every streamed key is below every buffered key, so no row matches. + +subgroup no_match + +template sql_benchmarks/pwmj/pwmj.benchmark.template +QPAD=04 +JOIN=FULL +JOIN_KIND=Full +RIGHT_ROWS=2000000 +RIGHT_KEY=-(h % 1000) +EXPECTED_ROWS=2100000 diff --git a/benchmarks/sql_benchmarks/pwmj/benchmarks/q05.benchmark b/benchmarks/sql_benchmarks/pwmj/benchmarks/q05.benchmark new file mode 100644 index 0000000000000..f4d9175ed48da --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/benchmarks/q05.benchmark @@ -0,0 +1,11 @@ +# INNER JOIN, null_heavy: half the streamed keys are NULL (never match), the rest as no_match. + +subgroup null_heavy + +template sql_benchmarks/pwmj/pwmj.benchmark.template +QPAD=05 +JOIN=INNER +JOIN_KIND=Inner +RIGHT_ROWS=2000000 +RIGHT_KEY=CASE WHEN value % 2 = 0 THEN NULL ELSE -(h % 1000) END +EXPECTED_ROWS=0 diff --git a/benchmarks/sql_benchmarks/pwmj/benchmarks/q06.benchmark b/benchmarks/sql_benchmarks/pwmj/benchmarks/q06.benchmark new file mode 100644 index 0000000000000..6361a0656241e --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/benchmarks/q06.benchmark @@ -0,0 +1,11 @@ +# LEFT JOIN, null_heavy: half the streamed keys are NULL (never match), the rest as no_match. + +subgroup null_heavy + +template sql_benchmarks/pwmj/pwmj.benchmark.template +QPAD=06 +JOIN=LEFT +JOIN_KIND=Left +RIGHT_ROWS=2000000 +RIGHT_KEY=CASE WHEN value % 2 = 0 THEN NULL ELSE -(h % 1000) END +EXPECTED_ROWS=100000 diff --git a/benchmarks/sql_benchmarks/pwmj/benchmarks/q07.benchmark b/benchmarks/sql_benchmarks/pwmj/benchmarks/q07.benchmark new file mode 100644 index 0000000000000..06128915dfdf3 --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/benchmarks/q07.benchmark @@ -0,0 +1,11 @@ +# RIGHT JOIN, null_heavy: half the streamed keys are NULL (never match), the rest as no_match. + +subgroup null_heavy + +template sql_benchmarks/pwmj/pwmj.benchmark.template +QPAD=07 +JOIN=RIGHT +JOIN_KIND=Right +RIGHT_ROWS=2000000 +RIGHT_KEY=CASE WHEN value % 2 = 0 THEN NULL ELSE -(h % 1000) END +EXPECTED_ROWS=2000000 diff --git a/benchmarks/sql_benchmarks/pwmj/benchmarks/q08.benchmark b/benchmarks/sql_benchmarks/pwmj/benchmarks/q08.benchmark new file mode 100644 index 0000000000000..c4f6bdfc9469a --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/benchmarks/q08.benchmark @@ -0,0 +1,11 @@ +# FULL JOIN, null_heavy: half the streamed keys are NULL (never match), the rest as no_match. + +subgroup null_heavy + +template sql_benchmarks/pwmj/pwmj.benchmark.template +QPAD=08 +JOIN=FULL +JOIN_KIND=Full +RIGHT_ROWS=2000000 +RIGHT_KEY=CASE WHEN value % 2 = 0 THEN NULL ELSE -(h % 1000) END +EXPECTED_ROWS=2100000 diff --git a/benchmarks/sql_benchmarks/pwmj/benchmarks/q09.benchmark b/benchmarks/sql_benchmarks/pwmj/benchmarks/q09.benchmark new file mode 100644 index 0000000000000..14a6ebf3738c6 --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/benchmarks/q09.benchmark @@ -0,0 +1,11 @@ +# INNER JOIN, half_match: half the streamed rows as no_match, half as selective. + +subgroup half_match + +template sql_benchmarks/pwmj/pwmj.benchmark.template +QPAD=09 +JOIN=INNER +JOIN_KIND=Inner +RIGHT_ROWS=2000000 +RIGHT_KEY=CASE WHEN value % 2 = 0 THEN -(h % 1000) ELSE 2 + h % 4 END +EXPECTED_ROWS=3000000 diff --git a/benchmarks/sql_benchmarks/pwmj/benchmarks/q10.benchmark b/benchmarks/sql_benchmarks/pwmj/benchmarks/q10.benchmark new file mode 100644 index 0000000000000..6fb5be1b35da3 --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/benchmarks/q10.benchmark @@ -0,0 +1,11 @@ +# LEFT JOIN, half_match: half the streamed rows as no_match, half as selective. + +subgroup half_match + +template sql_benchmarks/pwmj/pwmj.benchmark.template +QPAD=10 +JOIN=LEFT +JOIN_KIND=Left +RIGHT_ROWS=2000000 +RIGHT_KEY=CASE WHEN value % 2 = 0 THEN -(h % 1000) ELSE 2 + h % 4 END +EXPECTED_ROWS=3099996 diff --git a/benchmarks/sql_benchmarks/pwmj/benchmarks/q11.benchmark b/benchmarks/sql_benchmarks/pwmj/benchmarks/q11.benchmark new file mode 100644 index 0000000000000..618e1c059599c --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/benchmarks/q11.benchmark @@ -0,0 +1,11 @@ +# RIGHT JOIN, half_match: half the streamed rows as no_match, half as selective. + +subgroup half_match + +template sql_benchmarks/pwmj/pwmj.benchmark.template +QPAD=11 +JOIN=RIGHT +JOIN_KIND=Right +RIGHT_ROWS=2000000 +RIGHT_KEY=CASE WHEN value % 2 = 0 THEN -(h % 1000) ELSE 2 + h % 4 END +EXPECTED_ROWS=4000000 diff --git a/benchmarks/sql_benchmarks/pwmj/benchmarks/q12.benchmark b/benchmarks/sql_benchmarks/pwmj/benchmarks/q12.benchmark new file mode 100644 index 0000000000000..da826fa7f5e8d --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/benchmarks/q12.benchmark @@ -0,0 +1,11 @@ +# FULL JOIN, half_match: half the streamed rows as no_match, half as selective. + +subgroup half_match + +template sql_benchmarks/pwmj/pwmj.benchmark.template +QPAD=12 +JOIN=FULL +JOIN_KIND=Full +RIGHT_ROWS=2000000 +RIGHT_KEY=CASE WHEN value % 2 = 0 THEN -(h % 1000) ELSE 2 + h % 4 END +EXPECTED_ROWS=4099996 diff --git a/benchmarks/sql_benchmarks/pwmj/benchmarks/q13.benchmark b/benchmarks/sql_benchmarks/pwmj/benchmarks/q13.benchmark new file mode 100644 index 0000000000000..37d32a1fea8aa --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/benchmarks/q13.benchmark @@ -0,0 +1,12 @@ +# INNER JOIN, selective: every streamed row matches, but only the 1-4 smallest buffered keys, so its +# first match sits at the far end of the buffered side while the output stays small. + +subgroup selective + +template sql_benchmarks/pwmj/pwmj.benchmark.template +QPAD=13 +JOIN=INNER +JOIN_KIND=Inner +RIGHT_ROWS=2000000 +RIGHT_KEY=2 + h % 4 +EXPECTED_ROWS=5000000 diff --git a/benchmarks/sql_benchmarks/pwmj/benchmarks/q14.benchmark b/benchmarks/sql_benchmarks/pwmj/benchmarks/q14.benchmark new file mode 100644 index 0000000000000..dea39a0fa8c2c --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/benchmarks/q14.benchmark @@ -0,0 +1,12 @@ +# LEFT JOIN, selective: every streamed row matches, but only the 1-4 smallest buffered keys, so its +# first match sits at the far end of the buffered side while the output stays small. + +subgroup selective + +template sql_benchmarks/pwmj/pwmj.benchmark.template +QPAD=14 +JOIN=LEFT +JOIN_KIND=Left +RIGHT_ROWS=2000000 +RIGHT_KEY=2 + h % 4 +EXPECTED_ROWS=5099996 diff --git a/benchmarks/sql_benchmarks/pwmj/benchmarks/q15.benchmark b/benchmarks/sql_benchmarks/pwmj/benchmarks/q15.benchmark new file mode 100644 index 0000000000000..164418be03553 --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/benchmarks/q15.benchmark @@ -0,0 +1,12 @@ +# RIGHT JOIN, selective: every streamed row matches, but only the 1-4 smallest buffered keys, so its +# first match sits at the far end of the buffered side while the output stays small. + +subgroup selective + +template sql_benchmarks/pwmj/pwmj.benchmark.template +QPAD=15 +JOIN=RIGHT +JOIN_KIND=Right +RIGHT_ROWS=2000000 +RIGHT_KEY=2 + h % 4 +EXPECTED_ROWS=5000000 diff --git a/benchmarks/sql_benchmarks/pwmj/benchmarks/q16.benchmark b/benchmarks/sql_benchmarks/pwmj/benchmarks/q16.benchmark new file mode 100644 index 0000000000000..3e6165d715e16 --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/benchmarks/q16.benchmark @@ -0,0 +1,12 @@ +# FULL JOIN, selective: every streamed row matches, but only the 1-4 smallest buffered keys, so its +# first match sits at the far end of the buffered side while the output stays small. + +subgroup selective + +template sql_benchmarks/pwmj/pwmj.benchmark.template +QPAD=16 +JOIN=FULL +JOIN_KIND=Full +RIGHT_ROWS=2000000 +RIGHT_KEY=2 + h % 4 +EXPECTED_ROWS=5099996 diff --git a/benchmarks/sql_benchmarks/pwmj/benchmarks/q17.benchmark b/benchmarks/sql_benchmarks/pwmj/benchmarks/q17.benchmark new file mode 100644 index 0000000000000..7ed1229517ab4 --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/benchmarks/q17.benchmark @@ -0,0 +1,13 @@ +# INNER JOIN, all_match: every streamed key is above every buffered key, so each streamed row matches +# the whole buffered side and the join is bound by the output it builds. The output +# is lhs * rhs rows, so this subgroup streams only 2000 rows. + +subgroup all_match + +template sql_benchmarks/pwmj/pwmj.benchmark.template +QPAD=17 +JOIN=INNER +JOIN_KIND=Inner +RIGHT_ROWS=2000 +RIGHT_KEY=100001 + h % 1000 +EXPECTED_ROWS=200000000 diff --git a/benchmarks/sql_benchmarks/pwmj/benchmarks/q18.benchmark b/benchmarks/sql_benchmarks/pwmj/benchmarks/q18.benchmark new file mode 100644 index 0000000000000..4d3ff07346a1f --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/benchmarks/q18.benchmark @@ -0,0 +1,13 @@ +# LEFT JOIN, all_match: every streamed key is above every buffered key, so each streamed row matches +# the whole buffered side and the join is bound by the output it builds. The output +# is lhs * rhs rows, so this subgroup streams only 2000 rows. + +subgroup all_match + +template sql_benchmarks/pwmj/pwmj.benchmark.template +QPAD=18 +JOIN=LEFT +JOIN_KIND=Left +RIGHT_ROWS=2000 +RIGHT_KEY=100001 + h % 1000 +EXPECTED_ROWS=200000000 diff --git a/benchmarks/sql_benchmarks/pwmj/benchmarks/q19.benchmark b/benchmarks/sql_benchmarks/pwmj/benchmarks/q19.benchmark new file mode 100644 index 0000000000000..fb8df3043257a --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/benchmarks/q19.benchmark @@ -0,0 +1,13 @@ +# RIGHT JOIN, all_match: every streamed key is above every buffered key, so each streamed row matches +# the whole buffered side and the join is bound by the output it builds. The output +# is lhs * rhs rows, so this subgroup streams only 2000 rows. + +subgroup all_match + +template sql_benchmarks/pwmj/pwmj.benchmark.template +QPAD=19 +JOIN=RIGHT +JOIN_KIND=Right +RIGHT_ROWS=2000 +RIGHT_KEY=100001 + h % 1000 +EXPECTED_ROWS=200000000 diff --git a/benchmarks/sql_benchmarks/pwmj/benchmarks/q20.benchmark b/benchmarks/sql_benchmarks/pwmj/benchmarks/q20.benchmark new file mode 100644 index 0000000000000..2462b7726aa26 --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/benchmarks/q20.benchmark @@ -0,0 +1,13 @@ +# FULL JOIN, all_match: every streamed key is above every buffered key, so each streamed row matches +# the whole buffered side and the join is bound by the output it builds. The output +# is lhs * rhs rows, so this subgroup streams only 2000 rows. + +subgroup all_match + +template sql_benchmarks/pwmj/pwmj.benchmark.template +QPAD=20 +JOIN=FULL +JOIN_KIND=Full +RIGHT_ROWS=2000 +RIGHT_KEY=100001 + h % 1000 +EXPECTED_ROWS=200000000 diff --git a/benchmarks/sql_benchmarks/pwmj/init/cleanup.sql b/benchmarks/sql_benchmarks/pwmj/init/cleanup.sql new file mode 100644 index 0000000000000..92b811f6f021f --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/init/cleanup.sql @@ -0,0 +1,4 @@ +DROP TABLE IF EXISTS lhs; +DROP TABLE IF EXISTS rhs; +reset datafusion.optimizer.enable_piecewise_merge_join; +reset datafusion.catalog.information_schema; diff --git a/benchmarks/sql_benchmarks/pwmj/init/load.sql b/benchmarks/sql_benchmarks/pwmj/init/load.sql new file mode 100644 index 0000000000000..9bc977f5bfde5 --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/init/load.sql @@ -0,0 +1,23 @@ +-- Builds the two in-memory join inputs. `load` runs before `init` and is not +-- timed, so generating the keys never enters the measurement. +-- +-- Keys are scrambled with a fixed multiplicative hash rather than an RNG, so +-- every run and every branch sees the same data. The multiplier is odd and not +-- a multiple of 5, so `lhs.key` is a permutation of 1..=100000. +-- +-- Parameters, supplied by pwmj.benchmark.template: +-- RIGHT_ROWS streamed row count +-- RIGHT_KEY streamed key expression over `value` and `h` +DROP TABLE IF EXISTS lhs; + +DROP TABLE IF EXISTS rhs; + +CREATE TABLE lhs AS +SELECT CAST(1 + (value * 2654435761) % 100000 AS INT) AS key, + CAST(value AS INT) AS payload +FROM range(100000); + +CREATE TABLE rhs AS +SELECT CAST(${RIGHT_KEY} AS INT) AS key, + CAST(value AS INT) AS payload +FROM (SELECT value, (value * 2654435761) % 2000000 AS h FROM range(${RIGHT_ROWS})) t; diff --git a/benchmarks/sql_benchmarks/pwmj/init/settings.sql b/benchmarks/sql_benchmarks/pwmj/init/settings.sql new file mode 100644 index 0000000000000..686c97c7f809a --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/init/settings.sql @@ -0,0 +1,3 @@ +-- information_schema is enabled so the template can assert the flag took effect. +set datafusion.catalog.information_schema = true; +set datafusion.optimizer.enable_piecewise_merge_join = true; diff --git a/benchmarks/sql_benchmarks/pwmj/pwmj.benchmark.template b/benchmarks/sql_benchmarks/pwmj/pwmj.benchmark.template new file mode 100644 index 0000000000000..edd9da3f21080 --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/pwmj.benchmark.template @@ -0,0 +1,48 @@ +# Shared template for the `pwmj` suite: `PiecewiseMergeJoin`'s classic joins +# (INNER / LEFT / RIGHT / FULL) over a single range predicate, +# `lhs.key < rhs.key`. +# +# Parameters set by each qNN.benchmark: +# QPAD zero-padded query id +# JOIN SQL join keyword (INNER, LEFT, RIGHT, FULL) +# JOIN_KIND plan `join_type` (Inner, Left, Right, Full) +# RIGHT_ROWS streamed row count +# RIGHT_KEY streamed key expression over `value` and its scramble `h` +# EXPECTED_ROWS rows the join returns, asserted before measuring +# +# `lhs` (buffered) holds the keys 1..=100000; a streamed key `b` matches exactly +# the buffered keys 1..b, so each subgroup is a different match regime (see +# init/load.sql). There is no nested-loop arm: at these sizes it is 2 * 10^11 +# predicate evaluations. + +load sql_benchmarks/pwmj/init/load.sql + +init sql_benchmarks/pwmj/init/settings.sql + +name Q${QPAD} +group pwmj + +# PWMJ is off by default. Without the flag every query plans as a nested-loop +# join and the suite measures a different operator than its name claims. +assert I +SELECT value = 'true' FROM information_schema.df_settings WHERE name = 'datafusion.optimizer.enable_piecewise_merge_join'; +---- +true + +# The row count follows from the keys alone, so a wrong join result fails here +# rather than showing up as a plausible timing. +assert I +SELECT count(*) FROM lhs ${JOIN} JOIN rhs ON lhs.key < rhs.key; +---- +${EXPECTED_ROWS} + +# Pin the operator and the join type: the flag only permits PWMJ, and a planner +# that swapped the inputs would measure the mirrored join. `expect_plan` matches +# the plan as `EXPLAIN` displays it; the trailing comma keeps `Left` from +# matching `LeftSemi`/`LeftAnti`/`LeftMark`. +expect_plan PiecewiseMergeJoin: operator=Lt, join_type=${JOIN_KIND}, + +run +SELECT lhs.payload, rhs.payload FROM lhs ${JOIN} JOIN rhs ON lhs.key < rhs.key; + +cleanup sql_benchmarks/pwmj/init/cleanup.sql diff --git a/benchmarks/sql_benchmarks/pwmj/pwmj.suite b/benchmarks/sql_benchmarks/pwmj/pwmj.suite new file mode 100644 index 0000000000000..47b18e670f4da --- /dev/null +++ b/benchmarks/sql_benchmarks/pwmj/pwmj.suite @@ -0,0 +1,7 @@ +description = "Piecewise merge join (classic INNER/LEFT/RIGHT/FULL) SQL benchmarks" + +query_pattern = "q{QUERY_ID_PADDED}.benchmark" + +[[examples]] +command = "cargo run --release --bin benchmark_runner -- pwmj" +description = "Run all piecewise merge join queries."