Projected IN subqueries plan two extra nested-loop mark joins per subquery and become quadratic

[adriangb]

Describe the bug

An IN or NOT IN subquery in a SELECT list plans three mark joins for each subquery. Two of those joins have no join predicate. They become NestedLoopJoinExec: join_type=RightMark over the full outer side and the full inner side. The cost of the query is then quadratic in the number of rows.

The results are correct. Only the speed is bad. A query that must take about 0.03 s takes more than 70 s on 200000 rows. If you wrap the subquery in COALESCE, the expression is duplicated and you get six mark joins. That query takes more than 140 s.

Projected IN subqueries were decorrelated by #24972 (merged 2026-09-15, which closes #23022). Before that change these queries were not supported. So this is not a regression against an older release. But the plan that the new code makes is much more costly than necessary.

The cause is in in_subquery_value_mark_join in datafusion/optimizer/src/decorrelate_predicate_subquery.rs. To make three-valued logic visible in a projected column, it builds three mark joins for each subquery:

1. matched, the real predicate.
2. subquery_has_null, the subquery filtered to col IS NULL and joined with no predicate.
3. subquery_non_empty, the subquery joined with no predicate.

A CASE expression then combines the three mark columns. For an uncorrelated subquery the last two joins have no join predicate at all, so they can only plan as nested loop joins.

To Reproduce

Tested on commit 22651d2 with a release build of datafusion-cli (cargo build -p datafusion-cli --profile ci) on an Apple Silicon laptop.

CREATE TABLE outer_t AS SELECT CAST(v AS INT) AS id, CAST(v % 1000 AS INT) AS z FROM (SELECT unnest(generate_series(1, 200000)) AS v);
CREATE TABLE inner_t AS SELECT CASE WHEN v % 97 = 0 THEN NULL ELSE CAST(v * 2 AS INT) END AS id, CAST(v % 1000 AS INT) AS z FROM (SELECT unnest(generate_series(1, 200000)) AS v);

-- B1 bare IN in the SELECT list
SELECT count(*) FILTER (WHERE m), count(*) FILTER (WHERE m IS NULL) FROM (SELECT id, id IN (SELECT id FROM inner_t) AS m FROM outer_t);
-- B2 COALESCE shape
SELECT count(*) FILTER (WHERE m) FROM (SELECT id, COALESCE((id IN (SELECT id FROM inner_t))::boolean, false) AS m FROM outer_t);
-- B3 correlated equality
SELECT count(*) FILTER (WHERE m), count(*) FILTER (WHERE m IS NULL) FROM (SELECT id, id IN (SELECT i.id FROM inner_t i WHERE i.z = o.z) AS m FROM outer_t o);
-- B4 two IN subqueries in separate columns
SELECT count(*) FILTER (WHERE a), count(*) FILTER (WHERE b) 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);
-- B5 correlated EXISTS
SELECT count(*) FILTER (WHERE e) FROM (SELECT id, EXISTS (SELECT 1 FROM inner_t i WHERE i.id = o.id) AS e FROM outer_t o);

Measured times. The column "single mark join" is the same queries with one null aware LeftMark hash join for each subquery, which is the approach of #21363. Both builds give the same results.

Query	main 22651d2	single mark join
B1	72.6 s	0.03 s
B2	148.8 s	0.05 s
B3	1.4 s	0.08 s
B4	105.4 s	0.06 s
B5	0.03 s	0.02 s

To see the plan shape on small tables, use t1(id), t2(id) and t3(id) and run EXPLAIN SELECT id, id IN (SELECT id FROM t3) AS m3, id IN (SELECT id FROM t2) AS m2 FROM t1. On main you get six mark joins for two subqueries:

logical_plan
01)Projection: t1.id, __correlated_sq_1.mark IS NOT DISTINCT FROM Boolean(true) OR (__correlated_sq_2.mark OR t1.id IS NULL AND __correlated_sq_3.mark) IS NOT DISTINCT FROM Boolean(true) AND __correlated_sq_1.mark IS DISTINCT FROM Boolean(true) AND Boolean(NULL) AS m3, __correlated_sq_4.mark IS NOT DISTINCT FROM Boolean(true) OR (__correlated_sq_5.mark OR t1.id IS NULL AND __correlated_sq_6.mark) IS NOT DISTINCT FROM Boolean(true) AND __correlated_sq_4.mark IS DISTINCT FROM Boolean(true) AND Boolean(NULL) AS m2
02)--LeftMark Join:
03)----LeftMark Join:
04)------LeftMark Join: t1.id = __correlated_sq_4.id null_aware
05)--------LeftMark Join:
06)----------LeftMark Join:
07)------------LeftMark Join: t1.id = __correlated_sq_1.id null_aware
08)--------------TableScan: t1 projection=[id]
09)--------------SubqueryAlias: __correlated_sq_1
10)----------------TableScan: t3 projection=[id]
11)------------SubqueryAlias: __correlated_sq_2
12)--------------Filter: t3.id IS NULL
13)----------------TableScan: t3 projection=[id]
14)----------SubqueryAlias: __correlated_sq_3
15)------------TableScan: t3 projection=[id]
16)--------SubqueryAlias: __correlated_sq_4
17)----------TableScan: t2 projection=[id]
18)------SubqueryAlias: __correlated_sq_5
19)--------Filter: t2.id IS NULL
20)----------TableScan: t2 projection=[id]
21)----SubqueryAlias: __correlated_sq_6
22)------TableScan: t2 projection=[id]
physical_plan
01)ProjectionExec: expr=[...]
02)--NestedLoopJoinExec: join_type=RightMark
03)----DataSourceExec: partitions=1, partition_sizes=[1]
04)----NestedLoopJoinExec: join_type=RightMark
05)------FilterExec: id@0 IS NULL
06)--------DataSourceExec: partitions=1, partition_sizes=[1]
07)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
08)--------HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0)], null_aware
09)----------CoalescePartitionsExec
10)------------NestedLoopJoinExec: join_type=RightMark
11)--------------DataSourceExec: partitions=1, partition_sizes=[1]
12)--------------NestedLoopJoinExec: join_type=RightMark
13)----------------FilterExec: id@0 IS NULL
14)------------------DataSourceExec: partitions=1, partition_sizes=[1]
15)----------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
16)------------------HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0)], null_aware
17)--------------------DataSourceExec: partitions=1, partition_sizes=[1]
18)--------------------DataSourceExec: partitions=1, partition_sizes=[1]
19)----------DataSourceExec: partitions=1, partition_sizes=[1]

Expected behavior

One mark join for each subquery. No nested loop join.

A single LeftMark join is already exact under three-valued logic when the join filter is hashable only. build_join makes the join null aware when the keys can be NULL, which came from #21585. When the keys cannot be NULL, a plain mark join is exact. The three join materialization is necessary only when a non-equality correlated predicate stays as a residual join filter, because the hash join cannot mark UNKNOWN for a residual predicate.

A pull request that makes this change is in preparation.

Additional context

EXISTS in a SELECT list is not affected. See B5, which is fast on main, because EXISTS has two-valued logic and needs one mark join only.

The same subquery shape in a WHERE clause is also not affected. The Filter path builds one semi join, anti join or mark join for each subquery.