adriangb opened a new issue, #25341: URL: https://github.com/apache/datafusion/issues/25341
### 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 https://github.com/apache/datafusion/pull/24972 (merged 2026-09-15, which closes https://github.com/apache/datafusion/issues/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 22651d24cc with a release build of `datafusion-cli` (`cargo build -p datafusion-cli --profile ci`) on an Apple Silicon laptop. ```sql 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 https://github.com/apache/datafusion/pull/21363. Both builds give the same results. | Query | main 22651d24cc | 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 https://github.com/apache/datafusion/pull/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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
