bench: add projection_subquery SQL benchmark suite

benchmarks/README.md

@@ -1009,6 +1009,32 @@ Several queries are included to test sort merge joins under various workloads.
 
 ./bench.sh run smj
 ```
+
+## Projection Subquery
+
+This benchmark measures `IN`, `NOT IN` and `EXISTS` subqueries that sit in the `SELECT` list instead of a filter
+(see <https://github.com/apache/datafusion/issues/25341>).
+
+A projected `IN` must return `NULL`, and not `false`, when there is no match and the inner side holds a `NULL`.
+The decorrelation therefore turns one projected `IN` into more than one mark join.
+If a mark join has no hashable join predicate, it runs as a nested-loop join, and the cost grows with the outer row count times the inner row count.
+
+Every query projects the boolean subquery result and aggregates it, so the measured cost is the plan and not the size of the output.
+Query `q07` correlates on `<` instead of `=`, so it stays on the nested-loop path.
+It is the control: its time must not change when the hashable shapes get faster.
+
+The two tables are built inline by the suite's load SQL, so there is no data step.
+Set `PSQ_ROWS` to change the row count in each table.
+The default is 30,000, which keeps the nested-loop plans at a few hundred milliseconds.
+The checked-in result files hold the counts for that default, so `--result-mode validate` needs it.
+
+### Example Run
+
+```bash
+# No need to generate data: the suite's load SQL builds the two tables inline
+
+./bench.sh run projection_subquery
+```
 ## Cancellation
 
 Test performance of cancelling queries.

benchmarks/bench.sh

@@ -115,6 +115,10 @@ parquet_row_filter_skip: Per-RG fully-matched RowFilter skip on Parquet (apache/
 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)
+projection_subquery:    IN / NOT IN / EXISTS subqueries in the SELECT list (see https://github.com/apache/datafusion/issues/25341); each query projects the
+                          boolean subquery result and aggregates it, so the cost is the decorrelation plan and not the output size
+                          (q07 correlates on '<' instead of '=', so it keeps the nested-loop plan and acts as the control)
+                          (data generated inline by the suite's load SQL; knob: PSQ_ROWS)
 
 # ClickBench Benchmarks
 clickbench_1:           ClickBench queries against a single parquet file
@@ -279,6 +283,10 @@ main() {
                     # Data is generated inline by the suite's load SQL from range().
                     echo "null_aware_join: no external data to generate"
                     ;;
+                projection_subquery)
+                    # Data is generated inline by the suite's load SQL.
+                    echo "projection_subquery: no external data to generate"
+                    ;;
                 asof_join)
                     data_asof_join
                     ;;
@@ -530,6 +538,9 @@ main() {
                 null_aware_join)
                     run_null_aware_join
                     ;;
+                projection_subquery)
+                    run_projection_subquery
+                    ;;
                 asof_join)
                     run_asof_join
                     ;;
@@ -969,6 +980,24 @@ run_null_aware_join() {
       bash -c "$SQL_CARGO_COMMAND"
 }
 
+# Runs the projection_subquery suite: IN / NOT IN / EXISTS subqueries that sit
+# in the SELECT list instead of a filter (see
+# https://github.com/apache/datafusion/issues/25341). The load SQL builds the
+# two tables inline, so there is no data step. Each query projects the boolean
+# subquery result and aggregates it, so the measured cost is the decorrelation
+# plan and not the size of the output. Query 07 correlates on '<' instead of
+# '=', so it keeps the nested-loop plan and acts as the control.
+# Knob (string-substituted into the load SQL, not engine config):
+#   PSQ_ROWS  rows in each of the two tables (default 30_000; the checked-in
+#             result files hold the counts for that value)
+run_projection_subquery() {
+    echo "Running projection_subquery benchmark (rows=${PSQ_ROWS:-30000})..."
+    debug_run env BENCH_NAME=projection_subquery \
+      PSQ_ROWS="${PSQ_ROWS:-30000}" \
+      ${QUERY:+BENCH_QUERY="${QUERY}"}  \
+      bash -c "$SQL_CARGO_COMMAND"
+}
+
 # Runs the tpch in memory (needs tpch parquet data)
 run_tpch_mem() {
     SCALE_FACTOR=$1

benchmarks/sql_benchmarks/README.md

@@ -47,6 +47,7 @@ in the community:
 | `wide_schema`         | Small-projection queries on a wide (1024-col, 256-file) synthetic dataset; runs `wide` + `narrow` subgroups for comparison |
 | `predicate_eval`      | Conjunctive (AND) filter-evaluation micro-benchmarks; each subgroup is a different predicate pattern, to test how an adaptive predicate-ordering system behaves across them ([#11262](https://github.com/apache/datafusion/issues/11262)). Subgroups (`--subgroup`): `costsel`, `cost`, `selectivity`, `cardinality`, `width`, `scale`, `neutral`, `correlation`, `drift`, `nulls`. The suite sets no engine config of its own, so by default it measures DataFusion's built-in left-deep `AND` short-circuit; point it at a system under test by exporting that system's own DataFusion setting (the harness builds its `SessionConfig` with `SessionConfig::from_env`). Every query is a `count(*)`, and the counts are checked in under `predicate_eval/results/`, so `--result-mode validate` also checks that a reordering under test still returns the same rows; the checked-in counts were persisted at the suite defaults (`PRED_ROWS=1000000`, `PRED_FILL=30`), so validation assumes those (the `scale` and `width` subgroups pin their own values per query and validate at any setting). |
 | `parquet_row_filter_skip` | Micro-benchmark for the per-row-group fully-matched RowFilter skip on Parquet scans ([#23696](https://github.com/apache/datafusion/issues/23696)). Subgroups (`--subgroup`): `skip` (clustered key, most row groups fully matched by statistics so the per-row filter is skipped), `control` (scrambled key, no row group is ever fully matched). Size the data with `PRED_ROWS` and the row-group size with `RG_SIZE`. |
+| `projection_subquery` | `IN`, `NOT IN` and `EXISTS` subqueries that sit in the `SELECT` list instead of a filter ([#25341](https://github.com/apache/datafusion/issues/25341)). Every query projects the boolean subquery result and aggregates it, so the measured cost is the decorrelation plan and not the size of the output. `q07` correlates on `<` instead of `=`, so it keeps the nested-loop plan and is the control that must not change. Size the two inline tables with `PSQ_ROWS` (default `30000`). The counts are checked in under `projection_subquery/results/`, so `--result-mode validate` also proves that a change to the decorrelation still returns the same rows; the checked-in counts were persisted at the default `PSQ_ROWS`, so validation assumes that value. |
 
 # Running Benchmarks
 
@@ -174,6 +175,7 @@ Some benchmarks use custom environment variables as outlined below:
 | PRED_ROWS                    | Used in the predicate_eval benchmark to size the synthetic table (the `scale` subgroup overrides this per query), and in the parquet_row_filter_skip benchmark to size the generated Parquet datasets (default `10000000` there). | `1000000`     |
 | PRED_FILL                    | Used in the predicate_eval benchmark as the string-column width knob (filler chars per marker).                          | `30`          |
 | RG_SIZE                      | Used in the parquet_row_filter_skip benchmark as the Parquet `max_row_group_size` for the generated datasets.            | `1000000`     |
+| PSQ_ROWS                     | Used in the projection_subquery benchmark to size the two inline tables.                                                 | `30000`       |
 
 ## How it works
 

benchmarks/sql_benchmarks/projection_subquery/benchmarks/q01.benchmark

@@ -0,0 +1,3 @@
+template sql_benchmarks/projection_subquery/projection_subquery.benchmark.template
+QPAD=01
+NAME=q01_in_bare

benchmarks/sql_benchmarks/projection_subquery/benchmarks/q02.benchmark

@@ -0,0 +1,3 @@
+template sql_benchmarks/projection_subquery/projection_subquery.benchmark.template
+QPAD=02
+NAME=q02_in_coalesce

benchmarks/sql_benchmarks/projection_subquery/benchmarks/q03.benchmark

@@ -0,0 +1,3 @@
+template sql_benchmarks/projection_subquery/projection_subquery.benchmark.template
+QPAD=03
+NAME=q03_in_correlated_eq

benchmarks/sql_benchmarks/projection_subquery/benchmarks/q04.benchmark

@@ -0,0 +1,3 @@
+template sql_benchmarks/projection_subquery/projection_subquery.benchmark.template
+QPAD=04
+NAME=q04_in_two_columns

benchmarks/sql_benchmarks/projection_subquery/benchmarks/q05.benchmark

@@ -0,0 +1,3 @@
+template sql_benchmarks/projection_subquery/projection_subquery.benchmark.template
+QPAD=05
+NAME=q05_not_in_bare

benchmarks/sql_benchmarks/projection_subquery/benchmarks/q06.benchmark

@@ -0,0 +1,3 @@
+template sql_benchmarks/projection_subquery/projection_subquery.benchmark.template
+QPAD=06
+NAME=q06_exists_correlated

benchmarks/sql_benchmarks/projection_subquery/benchmarks/q07.benchmark

@@ -0,0 +1,3 @@
+template sql_benchmarks/projection_subquery/projection_subquery.benchmark.template
+QPAD=07
+NAME=q07_in_correlated_residual

benchmarks/sql_benchmarks/projection_subquery/init/cleanup.sql

@@ -0,0 +1,2 @@
+DROP TABLE IF EXISTS outer_t;
+DROP TABLE IF EXISTS inner_t;

benchmarks/sql_benchmarks/projection_subquery/load/tables.sql

@@ -0,0 +1,25 @@
+-- Two small integer tables for the projected IN / EXISTS subquery benchmarks.
+--
+-- `outer_t.id` is dense on 1..PSQ_ROWS and `inner_t.id` is the even numbers,
+-- so about half of the outer keys have a match. Every 97th inner key is NULL,
+-- which keeps three-valued logic in play: a projected `IN` must return NULL,
+-- not false, when there is no match and the inner side holds a NULL. That is
+-- the reason the decorrelation needs more than one mark join.
+--
+-- `z` is a low-cardinality column (1000 groups) used by the correlated
+-- queries, once with an equality correlation and once with a `<` correlation.
+--
+-- The tables are deliberately small. The plans under test are quadratic in
+-- outer rows times inner rows, so a larger PSQ_ROWS makes the suite take
+-- minutes per query instead of seconds.
+CREATE TABLE outer_t AS
+SELECT
+  CAST(value AS INT) AS id,
+  CAST(value % 1000 AS INT) AS z
+FROM generate_series(1, ${PSQ_ROWS:-30000});
+
+CREATE TABLE inner_t AS
+SELECT
+  CASE WHEN value % 97 = 0 THEN NULL ELSE CAST(value * 2 AS INT) END AS id,
+  CAST(value % 1000 AS INT) AS z
+FROM generate_series(1, ${PSQ_ROWS:-30000});

benchmarks/sql_benchmarks/projection_subquery/projection_subquery.benchmark.template

@@ -0,0 +1,45 @@
+# Shared template for every projection_subquery benchmark. Each qNN.benchmark
+# includes this template with parameters:
+#   QPAD  zero-padded query id / query file stem  (e.g. 01)
+#   NAME  criterion display name                  (e.g. q01_in_bare)
+# Optional, read by the load script as ${...:-default}:
+#   PSQ_ROWS  rows in each of the two tables (default 30000)
+#
+# The tables are always named `outer_t` and `inner_t`, so the load, the asserts
+# and the cleanup are the same for every query.
+#
+# Every query is an aggregate over the projected subquery result, so the output
+# is two numbers and the measurement is the plan and not the result transfer.
+# The counts are checked in under results/, so --result-mode validate also
+# proves that a change to the decorrelation still returns the same rows. The
+# checked-in counts hold for the default PSQ_ROWS only.
+
+load sql_benchmarks/projection_subquery/load/tables.sql
+
+name ${NAME}
+group projection_subquery
+
+# Both tables have exactly PSQ_ROWS rows.
+assert I
+SELECT count(*) = ${PSQ_ROWS:-30000} FROM outer_t;
+----
+true
+
+assert I
+SELECT count(*) = ${PSQ_ROWS:-30000} FROM inner_t;
+----
+true
+
+# The inner key must keep its NULLs, because three-valued logic is what forces
+# the extra mark joins. A load that loses them makes the suite measure the
+# wrong plan.
+assert I
+SELECT count(*) > 0 FROM inner_t WHERE id IS NULL;
+----
+true
+
+run sql_benchmarks/projection_subquery/queries/q${QPAD}.sql
+
+result sql_benchmarks/projection_subquery/results/${NAME}.csv
+
+cleanup sql_benchmarks/projection_subquery/init/cleanup.sql

benchmarks/sql_benchmarks/projection_subquery/projection_subquery.suite

@@ -0,0 +1,23 @@
+description = "IN / NOT IN / EXISTS subqueries that appear in the SELECT list instead of a filter (apache/datafusion#25341). Each query projects the boolean result of a subquery and aggregates it, so the measured work is the decorrelation plan and not the output size. The decorrelation turns one projected IN into mark joins; when the join predicate is not hashable, or the correlation is not an equality, the plan keeps a nested-loop mark join and the cost grows with outer rows times inner rows. Query 07 correlates on `<` and stays on that path, so it is the control that must not change. The two tables are built inline by the load SQL and are sized with PSQ_ROWS."
+
+query_pattern = "q{QUERY_ID_PADDED}.benchmark"
+
+[[options]]
+name = "rows"
+short = "r"
+env = "PSQ_ROWS"
+default = "30000"
+values = ["30000", "..."]
+help = "Sets the number of rows in each of the two generated tables. The checked-in result files hold the counts for the default value, so --result-mode validate needs the default."
+
+[[examples]]
+command = "cargo run --release --bin benchmark_runner -- projection_subquery"
+description = "Run every query with the default 30,000 rows per table."
+
+[[examples]]
+command = "cargo run --release --bin benchmark_runner -- projection_subquery --query 1 --result-mode validate"
+description = "Run query 01 and compare the counts with the checked-in result file."
+
+[[examples]]
+command = "cargo run --release --bin benchmark_runner -- projection_subquery -r 5000"
+description = "Run every query on smaller tables."

benchmarks/sql_benchmarks/projection_subquery/queries/q01.sql

@@ -0,0 +1,7 @@
+-- Shape: bare uncorrelated IN in the SELECT list. The subquery has no
+-- correlation, so the only join key is the outer key against the inner key.
+-- This is the plain form of the quadratic plan.
+SELECT
+  count(*) FILTER (WHERE m) AS true_count,
+  count(*) FILTER (WHERE m IS NULL) AS null_count
+FROM (SELECT id, id IN (SELECT id FROM inner_t) AS m FROM outer_t);

benchmarks/sql_benchmarks/projection_subquery/queries/q02.sql

@@ -0,0 +1,6 @@
+-- Shape: the same bare IN, wrapped in COALESCE so the NULL result becomes
+-- false. The wrapper is the common way to use a projected IN, and it must not
+-- stop the decorrelation.
+SELECT
+  count(*) FILTER (WHERE m) AS true_count
+FROM (SELECT id, COALESCE((id IN (SELECT id FROM inner_t))::boolean, false) AS m FROM outer_t);

benchmarks/sql_benchmarks/projection_subquery/queries/q03.sql

@@ -0,0 +1,6 @@
+-- Shape: IN correlated on an equality. The correlation adds a second join key,
+-- so the whole join condition stays hashable.
+SELECT
+  count(*) FILTER (WHERE m) AS true_count,
+  count(*) FILTER (WHERE m IS NULL) AS null_count
+FROM (SELECT id, id IN (SELECT i.id FROM inner_t i WHERE i.z = o.z) AS m FROM outer_t o);

benchmarks/sql_benchmarks/projection_subquery/queries/q04.sql

@@ -0,0 +1,12 @@
+-- Shape: two independent projected IN subqueries in one SELECT list. Each one
+-- is decorrelated on its own, so this measures whether the cost of the first
+-- one simply doubles.
+SELECT
+  count(*) FILTER (WHERE a) AS a_count,
+  count(*) FILTER (WHERE b) AS b_count
+FROM (
+  SELECT
+    id IN (SELECT id FROM inner_t) AS a,
+    id IN (SELECT id FROM inner_t WHERE z < 500) AS b
+  FROM outer_t
+);

benchmarks/sql_benchmarks/projection_subquery/queries/q05.sql

@@ -0,0 +1,6 @@
+-- Shape: NOT IN in the SELECT list. It is the negation of q01 and it has the
+-- same three-valued logic, so it must plan the same way.
+SELECT
+  count(*) FILTER (WHERE m) AS true_count,
+  count(*) FILTER (WHERE m IS NULL) AS null_count
+FROM (SELECT id, id NOT IN (SELECT id FROM inner_t) AS m FROM outer_t);

benchmarks/sql_benchmarks/projection_subquery/queries/q06.sql

@@ -0,0 +1,6 @@
+-- Shape: correlated EXISTS in the SELECT list. EXISTS has no NULL result, so
+-- one mark join is enough. It is the reference for what the IN queries should
+-- cost.
+SELECT
+  count(*) FILTER (WHERE e) AS true_count
+FROM (SELECT id, EXISTS (SELECT 1 FROM inner_t i WHERE i.id = o.id) AS e FROM outer_t o);

benchmarks/sql_benchmarks/projection_subquery/queries/q07.sql

@@ -0,0 +1,7 @@
+-- Shape: IN correlated on `<` instead of `=`. There is no equality to hash on,
+-- so this query keeps the nested-loop mark joins. It is the control: its cost
+-- must stay the same when the hashable cases get faster.
+SELECT
+  count(*) FILTER (WHERE m) AS true_count,
+  count(*) FILTER (WHERE m IS NULL) AS null_count
+FROM (SELECT id, id IN (SELECT i.id FROM inner_t i WHERE i.z < o.z) AS m FROM outer_t o);

benchmarks/sql_benchmarks/projection_subquery/results/q01_in_bare.csv

@@ -0,0 +1,2 @@
+true_count|null_count
+14846|15154

benchmarks/sql_benchmarks/projection_subquery/results/q02_in_coalesce.csv

@@ -0,0 +1,2 @@
+true_count
+14846

benchmarks/sql_benchmarks/projection_subquery/results/q03_in_correlated_eq.csv

@@ -0,0 +1,2 @@
+true_count|null_count
+15|9270

benchmarks/sql_benchmarks/projection_subquery/results/q04_in_two_columns.csv

@@ -0,0 +1,2 @@
+a_count|b_count
+14846|7423

benchmarks/sql_benchmarks/projection_subquery/results/q05_not_in_bare.csv

@@ -0,0 +1,2 @@
+true_count|null_count
+0|15154

benchmarks/sql_benchmarks/projection_subquery/results/q06_exists_correlated.csv

@@ -0,0 +1,2 @@
+true_count
+14846

benchmarks/sql_benchmarks/projection_subquery/results/q07_in_correlated_residual.csv

@@ -0,0 +1,2 @@
+true_count|null_count
+7408|22487