adriangb commented on code in PR #25338:
URL: https://github.com/apache/datafusion/pull/25338#discussion_r4057417829
##########
datafusion/sqllogictest/test_files/subquery_projection.slt:
##########
@@ -97,3 +97,528 @@ FROM outer_values o;
3 NULL
4 true
5 NULL
+
+# Plan shapes and NULL semantics of a projected IN subquery.
+#
+# `n1.id` holds a NULL, `n2.id` holds a NULL, and `n3.id` holds none. The mark
+# column of a LeftMark join carries the three-valued result on its own when the
+# join filter is hashable only, so one join per subquery is enough.
+
+statement ok
+CREATE TABLE n1(id INT, z INT) AS VALUES (1, 10), (2, 20), (NULL, 30), (4, 40);
+
+statement ok
+CREATE TABLE n2(id INT, z INT) AS VALUES (1, 5), (NULL, 50);
+
+statement ok
+CREATE TABLE n3(id INT) AS VALUES (1), (2);
+
+# One hash mark join per subquery. There is no materialization join, so no
+# nested loop join over outer x inner rows.
+query TT
+EXPLAIN SELECT id, id IN (SELECT id FROM n3) AS m3, id IN (SELECT id FROM n2)
AS m2 FROM n1;
+----
+logical_plan
+01)Projection: n1.id, __correlated_sq_1.mark AS m3, __correlated_sq_2.mark AS
m2
+02)--LeftMark Join: n1.id = __correlated_sq_2.id null_aware
+03)----LeftMark Join: n1.id = __correlated_sq_1.id null_aware
+04)------TableScan: n1 projection=[id]
+05)------SubqueryAlias: __correlated_sq_1
+06)--------TableScan: n3 projection=[id]
+07)----SubqueryAlias: __correlated_sq_2
+08)------TableScan: n2 projection=[id]
+physical_plan
+01)ProjectionExec: expr=[id@0 as id, mark@1 as m3, mark@2 as m2]
+02)--HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0)],
null_aware
+03)----HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0)],
null_aware
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------DataSourceExec: partitions=1, partition_sizes=[1]
+06)----DataSourceExec: partitions=1, partition_sizes=[1]
+
+# A non-equality correlation stays a residual join filter, so this query keeps
+# the three-join materialization.
+query TT
+EXPLAIN SELECT id, id IN (SELECT n2.id FROM n2 WHERE n2.z < n1.z) AS m FROM n1;
+----
+logical_plan
+01)Projection: n1.id, CASE WHEN __correlated_sq_1.mark THEN Boolean(true) WHEN
__correlated_sq_2.mark OR n1.id IS NULL AND __correlated_sq_3.mark THEN
Boolean(NULL) ELSE Boolean(false) END AS m
+02)--LeftMark Join: Filter: __correlated_sq_3.z < n1.z
+03)----LeftMark Join: Filter: __correlated_sq_2.z < n1.z
+04)------LeftMark Join: n1.id = __correlated_sq_1.id Filter:
__correlated_sq_1.z < n1.z
+05)--------TableScan: n1 projection=[id, z]
+06)--------SubqueryAlias: __correlated_sq_1
+07)----------TableScan: n2 projection=[id, z]
+08)------SubqueryAlias: __correlated_sq_2
+09)--------Projection: n2.z
+10)----------Filter: n2.id IS NULL
+11)------------TableScan: n2 projection=[id, z]
+12)----SubqueryAlias: __correlated_sq_3
+13)------TableScan: n2 projection=[z]
+physical_plan
+01)ProjectionExec: expr=[id@0 as id, CASE WHEN mark@1 THEN true WHEN mark@2 OR
id@0 IS NULL AND mark@3 THEN NULL ELSE false END as m]
+02)--NestedLoopJoinExec: join_type=RightMark, filter=z@1 < z@0,
projection=[id@0, mark@2, mark@3, mark@4]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+04)----NestedLoopJoinExec: join_type=RightMark, filter=z@1 < z@0
+05)------FilterExec: id@0 IS NULL, projection=[z@1]
+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)], filter=z@1 < z@0
+09)----------DataSourceExec: partitions=1, partition_sizes=[1]
+10)----------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query IB rowsort
+SELECT id, id IN (SELECT n2.id FROM n2 WHERE n2.z < n1.z) AS m FROM n1;
+----
+1 true
+2 false
+4 false
+NULL NULL
+
+query IBBB rowsort
+SELECT
+ id,
+ id IN (SELECT id FROM n3) AS m3,
+ EXISTS (SELECT 1 FROM n2 WHERE n2.id = n1.id) AS e2,
+ id NOT IN (SELECT id FROM n3 WHERE n3.id > 1) AS nn3
+FROM n1;
+----
+1 true true true
+2 true false false
+4 false false true
+NULL NULL false NULL
+
+query IB rowsort
+SELECT id, id IN (SELECT id FROM n2) AS m FROM n1;
+----
+1 true
+2 NULL
+4 NULL
+NULL NULL
+
+query IB rowsort
+SELECT id, id NOT IN (SELECT id FROM n2) AS m FROM n1;
+----
+1 false
+2 NULL
+4 NULL
+NULL NULL
+
+query IB rowsort
+SELECT id, id IN (SELECT id FROM n3) AS m FROM n1;
+----
+1 true
+2 true
+4 false
+NULL NULL
+
+query IT rowsort
+SELECT id, CASE WHEN NOT (id IN (SELECT id FROM n2)) THEN 'a' ELSE 'b' END AS
c FROM n1;
+----
+1 b
+2 b
+4 b
+NULL b
+
+query IB rowsort
+SELECT z, sum(id) IN (SELECT id FROM n3) AS m FROM n1 GROUP BY z;
+----
+10 true
+20 true
+30 NULL
+40 false
+
+query IB rowsort
+SELECT id, COALESCE((id IN (SELECT id FROM n3))::boolean, false) AS matched
FROM n1;
+----
+1 true
+2 true
+4 false
+NULL false
+
+query IT rowsort
+SELECT id, CASE WHEN id NOT IN (SELECT n2.id FROM n2 WHERE n2.z < n1.z) THEN
'a' ELSE 'b' END AS c FROM n1;
+----
+1 b
+2 a
+4 a
+NULL b
+
+statement ok
+DROP TABLE n1;
+
+statement ok
+DROP TABLE n2;
+
+statement ok
+DROP TABLE n3;
+
+# Nullable key expressions over non-nullable columns.
+#
+# `nn.id` and `nn.s` are not nullable, but a key expression over them can still
+# be NULL. `NULLIF(id, 1)` is NULL for `id = 1`, and `TRY_CAST(s AS INT)` is
+# NULL when the text is not a number. The join must be null-aware for these
+# keys, so the mark is NULL and `IN` gives UNKNOWN.
+
+statement ok
+CREATE TABLE nn(id INT NOT NULL, s VARCHAR NOT NULL) AS VALUES (1, '1'), (2,
'x'), (4, '4');
+
+statement ok
+CREATE TABLE r3(id INT NOT NULL) AS VALUES (1), (2);
+
+statement ok
+CREATE TABLE r3n(id INT NOT NULL) AS VALUES (1), (2), (5);
+
+# The nullable key expression keeps the plan at one null-aware mark join.
+query TT
+EXPLAIN SELECT id, NULLIF(id, 1) IN (SELECT id FROM r3) AS m FROM nn;
+----
+logical_plan
+01)Projection: nn.id, __correlated_sq_1.mark AS m
+02)--LeftMark Join: nullif(CAST(nn.id AS Int64), Int64(1)) =
__correlated_sq_1.r3.id null_aware
+03)----TableScan: nn projection=[id]
+04)----SubqueryAlias: __correlated_sq_1
+05)------Projection: CAST(r3.id AS Int64)
+06)--------TableScan: r3 projection=[id]
+physical_plan
+01)ProjectionExec: expr=[id@0 as id, mark@1 as m]
+02)--HashJoinExec: mode=CollectLeft, join_type=LeftMark,
on=[(nullif(nn.id,Int64(1))@1, r3.id@0)], projection=[id@0, mark@2], null_aware
+03)----ProjectionExec: expr=[id@0 as id, nullif(CAST(id@0 AS Int64), 1) as
nullif(nn.id,Int64(1))]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)----ProjectionExec: expr=[CAST(id@0 AS Int64) as r3.id]
+06)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# `NULLIF(id, 1)` is NULL for `id = 1`, and `r3` has no NULL, so the answer is
+# UNKNOWN for that row.
+query IB rowsort
+SELECT id, NULLIF(id, 1) IN (SELECT id FROM r3) AS m FROM nn;
+----
+1 NULL
+2 true
+4 false
+
+# `TRY_CAST('x' AS INT)` is NULL, so the answer is UNKNOWN for that row.
+query IB rowsort
+SELECT id, TRY_CAST(s AS INT) IN (SELECT id FROM r3) AS m FROM nn;
+----
+1 true
+2 NULL
+4 false
+
+# The same on the subquery side: the output of the subquery holds a NULL, so a
+# row with no match is UNKNOWN.
+query IB rowsort
+SELECT id, id IN (SELECT NULLIF(id, 5) FROM r3n) AS m FROM nn;
+----
+1 true
+2 true
+4 NULL
+
+# `NOT IN` reads the same mark column, negated.
+query IB rowsort
+SELECT id, NULLIF(id, 1) NOT IN (SELECT id FROM r3) AS m FROM nn;
+----
+1 NULL
+2 false
+4 true
+
+# The `NOT IN` filter path builds a LeftAnti join and reads the key nullability
+# the same way. UNKNOWN does not pass a filter, so `id = 1` drops out.
+query I rowsort
+SELECT id FROM nn WHERE NULLIF(id, 1) NOT IN (SELECT id FROM r3);
+----
+4
+
+# An empty subquery gives `false` also for a NULL key, and the plan stays one
+# null-aware mark join.
+statement ok
+CREATE TABLE r_empty(id INT NOT NULL) AS SELECT * FROM r3 WHERE false;
+
+query TT
+EXPLAIN SELECT id, NULLIF(id, 1) IN (SELECT id FROM r_empty) AS m FROM nn;
+----
+logical_plan
+01)Projection: nn.id, __correlated_sq_1.mark AS m
+02)--LeftMark Join: nullif(CAST(nn.id AS Int64), Int64(1)) =
__correlated_sq_1.r_empty.id null_aware
+03)----TableScan: nn projection=[id]
+04)----SubqueryAlias: __correlated_sq_1
+05)------Projection: CAST(r_empty.id AS Int64)
+06)--------TableScan: r_empty projection=[id]
+physical_plan
+01)ProjectionExec: expr=[id@0 as id, mark@1 as m]
+02)--HashJoinExec: mode=CollectLeft, join_type=LeftMark,
on=[(nullif(nn.id,Int64(1))@1, r_empty.id@0)], projection=[id@0, mark@2],
null_aware
+03)----ProjectionExec: expr=[id@0 as id, nullif(CAST(id@0 AS Int64), 1) as
nullif(nn.id,Int64(1))]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)----ProjectionExec: expr=[CAST(id@0 AS Int64) as r_empty.id]
+06)------DataSourceExec: partitions=1, partition_sizes=[0]
+
+query IB rowsort
+SELECT id, NULLIF(id, 1) IN (SELECT id FROM r_empty) AS m FROM nn;
+----
+1 false
+2 false
+4 false
+
+query IB rowsort
+SELECT id, NULLIF(id, 1) NOT IN (SELECT id FROM r_empty) AS m FROM nn;
+----
+1 true
+2 true
+4 true
+
+statement ok
+DROP TABLE r_empty;
+
+# A correlated `NOT IN` filter builds a `LeftAnti` join with two keys: the
+# value and the correlation. A null-aware `LeftAnti` hash join supports one key
+# only, so a function key over non-nullable columns must not make this join
+# null-aware.
+statement ok
+CREATE TABLE t1(k INT NOT NULL, s VARCHAR NOT NULL) AS VALUES (1, 'a'), (2,
'b');
+
+statement ok
+CREATE TABLE t2(k INT NOT NULL, s VARCHAR NOT NULL) AS VALUES (1, 'B'), (2,
'B');
+
+query IT rowsort
+SELECT * FROM t1 WHERE upper(t1.s) NOT IN (SELECT t2.s FROM t2 WHERE t2.k =
t1.k);
+----
+1 a
+
+# `NULLIF(k, 1)` is NULL for `k = 1`, and that group of `t2` is not empty, so
+# the correct result has no row for `k = 1`. The join has two keys, the value
+# and the correlation, and the null-aware `LeftAnti` executor takes one key
+# only. The `NOT IN` becomes a null-aware mark join, which takes any number of
+# keys, and a filter on the mark. `main` keeps the `k = 1` row, which is
+# https://github.com/apache/datafusion/issues/25347.
+query I rowsort
+SELECT k FROM t1 WHERE NULLIF(t1.k, 1) NOT IN (SELECT t2.k + 10 FROM t2 WHERE
t2.k = t1.k);
+----
+2
+
+query TT
+EXPLAIN SELECT k FROM t1 WHERE NULLIF(t1.k, 1) NOT IN (SELECT t2.k + 10 FROM
t2 WHERE t2.k = t1.k);
+----
+logical_plan
+01)Projection: t1.k
+02)--Filter: NOT __correlated_sq_1.mark
+03)----LeftMark Join: nullif(CAST(t1.k AS Int64), Int64(1)) =
__correlated_sq_1.t2.k + Int64(10), t1.k = __correlated_sq_1.k null_aware
+04)------TableScan: t1 projection=[k]
+05)------SubqueryAlias: __correlated_sq_1
+06)--------Projection: CAST(t2.k AS Int64) + Int64(10), t2.k
+07)----------TableScan: t2 projection=[k]
+physical_plan
+01)FilterExec: NOT mark@1, projection=[k@0]
+02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+03)----HashJoinExec: mode=CollectLeft, join_type=LeftMark,
on=[(nullif(t1.k,Int64(1))@1, t2.k + Int64(10)@0), (k@0, k@1)],
projection=[k@0, mark@2], null_aware
+04)------ProjectionExec: expr=[k@0 as k, nullif(CAST(k@0 AS Int64), 1) as
nullif(t1.k,Int64(1))]
+05)--------DataSourceExec: partitions=1, partition_sizes=[1]
+06)------ProjectionExec: expr=[CAST(k@0 AS Int64) + 10 as t2.k + Int64(10),
k@0 as k]
+07)--------DataSourceExec: partitions=1, partition_sizes=[1]
+
+statement ok
+DROP TABLE t1;
+
+statement ok
+DROP TABLE t2;
+
+# A non-equality correlation leaves one key and a residual join filter. No hash
+# join can mark the UNKNOWN rows of a residual filter
+# (https://github.com/apache/datafusion/issues/25336), so a `NOT IN` whose
+# value can be NULL does not become an anti join. It becomes the mark joins
+# that materialize its three-valued result, the same plan as for a projected
+# `IN`, and a filter on that result. For `k = 1` the key is NULL and the
+# correlated subquery result is empty, and `NULL NOT IN (<empty set>)` is TRUE.
+# Both rows are correct.
+statement ok
+CREATE TABLE ra(k INT NOT NULL, z INT NOT NULL) AS VALUES (1, 10), (2, 20);
+
+statement ok
+CREATE TABLE rb(k INT NOT NULL, z INT NOT NULL) AS VALUES (5, 50);
+
+query I rowsort
+SELECT k FROM ra WHERE NULLIF(ra.k, 1) NOT IN (SELECT rb.k FROM rb WHERE rb.z
< ra.z);
+----
+1
+2
+
+# The same shape where the correlated subquery result is not empty for the NULL
+# key: `NULL NOT IN ({5})` is UNKNOWN, so `2` is the only row. `main` builds a
+# plain anti join here and also keeps `1`.
+query I rowsort
+SELECT k FROM ra WHERE NULLIF(ra.k, 1) NOT IN (SELECT rb.k FROM rb WHERE rb.z
> ra.z);
+----
+2
Review Comment:
**Result changed here, and the old value was wrong.** This expectation was
`1` and `2`; it is now `2` only. DuckDB 1.5.2 agrees with the new value.
`NULLIF(ra.k, 1)` is NULL for the row `k = 1`, and the correlated subquery
for that row gives `{5}`, which is not empty. `NULL NOT IN ({5})` is UNKNOWN,
so the row must not appear.
A residual filter stays on this join, and no hash join can mark the UNKNOWN
rows of a residual filter. The `NOT IN` now becomes the three mark joins that
materialize its three-valued result, the plan that #24972 already uses for a
projected `IN`.
```
D SELECT ra.k, NULLIF(ra.k,1) AS key, (SELECT list(rb.k) FROM rb WHERE rb.z
> ra.z) AS sub FROM ra ORDER BY ra.k;
┌───────┬───────┬──────────┐
│ k │ key │ sub │
├───────┼───────┼──────────┤
│ 1 │ NULL │ [5] │
│ 2 │ 2 │ [5] │
└───────┴───────┴──────────┘
D SELECT k FROM ra WHERE NULLIF(ra.k, 1) NOT IN (SELECT rb.k FROM rb WHERE
rb.z > ra.z);
┌───┐
│ k │
├───┤
│ 2 │
└───┘
```
The row above, with `rb.z < ra.z`, keeps its expectation of `1` and `2`. Its
subquery result is empty for the NULL key, and `NULL NOT IN (<empty set>)` is
TRUE.
This is the executor gap in
https://github.com/apache/datafusion/issues/25336.
https://github.com/apache/datafusion/pull/25339 fixes the executor, and this
shape can go back to one anti join when that lands.
##########
datafusion/sqllogictest/test_files/subquery_projection.slt:
##########
@@ -97,3 +97,528 @@ FROM outer_values o;
3 NULL
4 true
5 NULL
+
+# Plan shapes and NULL semantics of a projected IN subquery.
+#
+# `n1.id` holds a NULL, `n2.id` holds a NULL, and `n3.id` holds none. The mark
+# column of a LeftMark join carries the three-valued result on its own when the
+# join filter is hashable only, so one join per subquery is enough.
+
+statement ok
+CREATE TABLE n1(id INT, z INT) AS VALUES (1, 10), (2, 20), (NULL, 30), (4, 40);
+
+statement ok
+CREATE TABLE n2(id INT, z INT) AS VALUES (1, 5), (NULL, 50);
+
+statement ok
+CREATE TABLE n3(id INT) AS VALUES (1), (2);
+
+# One hash mark join per subquery. There is no materialization join, so no
+# nested loop join over outer x inner rows.
+query TT
+EXPLAIN SELECT id, id IN (SELECT id FROM n3) AS m3, id IN (SELECT id FROM n2)
AS m2 FROM n1;
+----
+logical_plan
+01)Projection: n1.id, __correlated_sq_1.mark AS m3, __correlated_sq_2.mark AS
m2
+02)--LeftMark Join: n1.id = __correlated_sq_2.id null_aware
+03)----LeftMark Join: n1.id = __correlated_sq_1.id null_aware
+04)------TableScan: n1 projection=[id]
+05)------SubqueryAlias: __correlated_sq_1
+06)--------TableScan: n3 projection=[id]
+07)----SubqueryAlias: __correlated_sq_2
+08)------TableScan: n2 projection=[id]
+physical_plan
+01)ProjectionExec: expr=[id@0 as id, mark@1 as m3, mark@2 as m2]
+02)--HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0)],
null_aware
+03)----HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0)],
null_aware
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------DataSourceExec: partitions=1, partition_sizes=[1]
+06)----DataSourceExec: partitions=1, partition_sizes=[1]
+
+# A non-equality correlation stays a residual join filter, so this query keeps
+# the three-join materialization.
+query TT
+EXPLAIN SELECT id, id IN (SELECT n2.id FROM n2 WHERE n2.z < n1.z) AS m FROM n1;
+----
+logical_plan
+01)Projection: n1.id, CASE WHEN __correlated_sq_1.mark THEN Boolean(true) WHEN
__correlated_sq_2.mark OR n1.id IS NULL AND __correlated_sq_3.mark THEN
Boolean(NULL) ELSE Boolean(false) END AS m
+02)--LeftMark Join: Filter: __correlated_sq_3.z < n1.z
+03)----LeftMark Join: Filter: __correlated_sq_2.z < n1.z
+04)------LeftMark Join: n1.id = __correlated_sq_1.id Filter:
__correlated_sq_1.z < n1.z
+05)--------TableScan: n1 projection=[id, z]
+06)--------SubqueryAlias: __correlated_sq_1
+07)----------TableScan: n2 projection=[id, z]
+08)------SubqueryAlias: __correlated_sq_2
+09)--------Projection: n2.z
+10)----------Filter: n2.id IS NULL
+11)------------TableScan: n2 projection=[id, z]
+12)----SubqueryAlias: __correlated_sq_3
+13)------TableScan: n2 projection=[z]
+physical_plan
+01)ProjectionExec: expr=[id@0 as id, CASE WHEN mark@1 THEN true WHEN mark@2 OR
id@0 IS NULL AND mark@3 THEN NULL ELSE false END as m]
+02)--NestedLoopJoinExec: join_type=RightMark, filter=z@1 < z@0,
projection=[id@0, mark@2, mark@3, mark@4]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+04)----NestedLoopJoinExec: join_type=RightMark, filter=z@1 < z@0
+05)------FilterExec: id@0 IS NULL, projection=[z@1]
+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)], filter=z@1 < z@0
+09)----------DataSourceExec: partitions=1, partition_sizes=[1]
+10)----------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query IB rowsort
+SELECT id, id IN (SELECT n2.id FROM n2 WHERE n2.z < n1.z) AS m FROM n1;
+----
+1 true
+2 false
+4 false
+NULL NULL
+
+query IBBB rowsort
+SELECT
+ id,
+ id IN (SELECT id FROM n3) AS m3,
+ EXISTS (SELECT 1 FROM n2 WHERE n2.id = n1.id) AS e2,
+ id NOT IN (SELECT id FROM n3 WHERE n3.id > 1) AS nn3
+FROM n1;
+----
+1 true true true
+2 true false false
+4 false false true
+NULL NULL false NULL
+
+query IB rowsort
+SELECT id, id IN (SELECT id FROM n2) AS m FROM n1;
+----
+1 true
+2 NULL
+4 NULL
+NULL NULL
+
+query IB rowsort
+SELECT id, id NOT IN (SELECT id FROM n2) AS m FROM n1;
+----
+1 false
+2 NULL
+4 NULL
+NULL NULL
+
+query IB rowsort
+SELECT id, id IN (SELECT id FROM n3) AS m FROM n1;
+----
+1 true
+2 true
+4 false
+NULL NULL
+
+query IT rowsort
+SELECT id, CASE WHEN NOT (id IN (SELECT id FROM n2)) THEN 'a' ELSE 'b' END AS
c FROM n1;
+----
+1 b
+2 b
+4 b
+NULL b
+
+query IB rowsort
+SELECT z, sum(id) IN (SELECT id FROM n3) AS m FROM n1 GROUP BY z;
+----
+10 true
+20 true
+30 NULL
+40 false
+
+query IB rowsort
+SELECT id, COALESCE((id IN (SELECT id FROM n3))::boolean, false) AS matched
FROM n1;
+----
+1 true
+2 true
+4 false
+NULL false
+
+query IT rowsort
+SELECT id, CASE WHEN id NOT IN (SELECT n2.id FROM n2 WHERE n2.z < n1.z) THEN
'a' ELSE 'b' END AS c FROM n1;
+----
+1 b
+2 a
+4 a
+NULL b
+
+statement ok
+DROP TABLE n1;
+
+statement ok
+DROP TABLE n2;
+
+statement ok
+DROP TABLE n3;
+
+# Nullable key expressions over non-nullable columns.
+#
+# `nn.id` and `nn.s` are not nullable, but a key expression over them can still
+# be NULL. `NULLIF(id, 1)` is NULL for `id = 1`, and `TRY_CAST(s AS INT)` is
+# NULL when the text is not a number. The join must be null-aware for these
+# keys, so the mark is NULL and `IN` gives UNKNOWN.
+
+statement ok
+CREATE TABLE nn(id INT NOT NULL, s VARCHAR NOT NULL) AS VALUES (1, '1'), (2,
'x'), (4, '4');
+
+statement ok
+CREATE TABLE r3(id INT NOT NULL) AS VALUES (1), (2);
+
+statement ok
+CREATE TABLE r3n(id INT NOT NULL) AS VALUES (1), (2), (5);
+
+# The nullable key expression keeps the plan at one null-aware mark join.
+query TT
+EXPLAIN SELECT id, NULLIF(id, 1) IN (SELECT id FROM r3) AS m FROM nn;
+----
+logical_plan
+01)Projection: nn.id, __correlated_sq_1.mark AS m
+02)--LeftMark Join: nullif(CAST(nn.id AS Int64), Int64(1)) =
__correlated_sq_1.r3.id null_aware
+03)----TableScan: nn projection=[id]
+04)----SubqueryAlias: __correlated_sq_1
+05)------Projection: CAST(r3.id AS Int64)
+06)--------TableScan: r3 projection=[id]
+physical_plan
+01)ProjectionExec: expr=[id@0 as id, mark@1 as m]
+02)--HashJoinExec: mode=CollectLeft, join_type=LeftMark,
on=[(nullif(nn.id,Int64(1))@1, r3.id@0)], projection=[id@0, mark@2], null_aware
+03)----ProjectionExec: expr=[id@0 as id, nullif(CAST(id@0 AS Int64), 1) as
nullif(nn.id,Int64(1))]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)----ProjectionExec: expr=[CAST(id@0 AS Int64) as r3.id]
+06)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# `NULLIF(id, 1)` is NULL for `id = 1`, and `r3` has no NULL, so the answer is
+# UNKNOWN for that row.
+query IB rowsort
+SELECT id, NULLIF(id, 1) IN (SELECT id FROM r3) AS m FROM nn;
+----
+1 NULL
+2 true
+4 false
+
+# `TRY_CAST('x' AS INT)` is NULL, so the answer is UNKNOWN for that row.
+query IB rowsort
+SELECT id, TRY_CAST(s AS INT) IN (SELECT id FROM r3) AS m FROM nn;
+----
+1 true
+2 NULL
+4 false
+
+# The same on the subquery side: the output of the subquery holds a NULL, so a
+# row with no match is UNKNOWN.
+query IB rowsort
+SELECT id, id IN (SELECT NULLIF(id, 5) FROM r3n) AS m FROM nn;
+----
+1 true
+2 true
+4 NULL
+
+# `NOT IN` reads the same mark column, negated.
+query IB rowsort
+SELECT id, NULLIF(id, 1) NOT IN (SELECT id FROM r3) AS m FROM nn;
+----
+1 NULL
+2 false
+4 true
+
+# The `NOT IN` filter path builds a LeftAnti join and reads the key nullability
+# the same way. UNKNOWN does not pass a filter, so `id = 1` drops out.
+query I rowsort
+SELECT id FROM nn WHERE NULLIF(id, 1) NOT IN (SELECT id FROM r3);
+----
+4
+
+# An empty subquery gives `false` also for a NULL key, and the plan stays one
+# null-aware mark join.
+statement ok
+CREATE TABLE r_empty(id INT NOT NULL) AS SELECT * FROM r3 WHERE false;
+
+query TT
+EXPLAIN SELECT id, NULLIF(id, 1) IN (SELECT id FROM r_empty) AS m FROM nn;
+----
+logical_plan
+01)Projection: nn.id, __correlated_sq_1.mark AS m
+02)--LeftMark Join: nullif(CAST(nn.id AS Int64), Int64(1)) =
__correlated_sq_1.r_empty.id null_aware
+03)----TableScan: nn projection=[id]
+04)----SubqueryAlias: __correlated_sq_1
+05)------Projection: CAST(r_empty.id AS Int64)
+06)--------TableScan: r_empty projection=[id]
+physical_plan
+01)ProjectionExec: expr=[id@0 as id, mark@1 as m]
+02)--HashJoinExec: mode=CollectLeft, join_type=LeftMark,
on=[(nullif(nn.id,Int64(1))@1, r_empty.id@0)], projection=[id@0, mark@2],
null_aware
+03)----ProjectionExec: expr=[id@0 as id, nullif(CAST(id@0 AS Int64), 1) as
nullif(nn.id,Int64(1))]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)----ProjectionExec: expr=[CAST(id@0 AS Int64) as r_empty.id]
+06)------DataSourceExec: partitions=1, partition_sizes=[0]
+
+query IB rowsort
+SELECT id, NULLIF(id, 1) IN (SELECT id FROM r_empty) AS m FROM nn;
+----
+1 false
+2 false
+4 false
+
+query IB rowsort
+SELECT id, NULLIF(id, 1) NOT IN (SELECT id FROM r_empty) AS m FROM nn;
+----
+1 true
+2 true
+4 true
+
+statement ok
+DROP TABLE r_empty;
+
+# A correlated `NOT IN` filter builds a `LeftAnti` join with two keys: the
+# value and the correlation. A null-aware `LeftAnti` hash join supports one key
+# only, so a function key over non-nullable columns must not make this join
+# null-aware.
+statement ok
+CREATE TABLE t1(k INT NOT NULL, s VARCHAR NOT NULL) AS VALUES (1, 'a'), (2,
'b');
+
+statement ok
+CREATE TABLE t2(k INT NOT NULL, s VARCHAR NOT NULL) AS VALUES (1, 'B'), (2,
'B');
+
+query IT rowsort
+SELECT * FROM t1 WHERE upper(t1.s) NOT IN (SELECT t2.s FROM t2 WHERE t2.k =
t1.k);
+----
+1 a
+
+# `NULLIF(k, 1)` is NULL for `k = 1`, and that group of `t2` is not empty, so
+# the correct result has no row for `k = 1`. The join has two keys, the value
+# and the correlation, and the null-aware `LeftAnti` executor takes one key
+# only. The `NOT IN` becomes a null-aware mark join, which takes any number of
+# keys, and a filter on the mark. `main` keeps the `k = 1` row, which is
+# https://github.com/apache/datafusion/issues/25347.
+query I rowsort
+SELECT k FROM t1 WHERE NULLIF(t1.k, 1) NOT IN (SELECT t2.k + 10 FROM t2 WHERE
t2.k = t1.k);
+----
+2
Review Comment:
**Result changed here, and the old value was wrong.** This expectation was
`1` and `2`; it is now `2` only. DuckDB 1.5.2 agrees with the new value.
`NULLIF(t1.k, 1)` is NULL for the row `k = 1`. The correlated subquery for
that row gives `{11}`, which is not empty, so `NULL NOT IN ({11})` is UNKNOWN
and the row must not appear.
The old plan could not say that. A null-aware `LeftAnti` join takes one key
only, and this join has two, the value and the correlation, so the join stayed
a plain anti join and kept the row. The `NOT IN` now becomes a null-aware
`LeftMark` join, which takes any number of keys, plus a filter on the mark.
```
D SELECT t1.k, NULLIF(t1.k,1) AS key, (SELECT list(t2.k+10) FROM t2 WHERE
t2.k=t1.k) AS sub FROM t1 ORDER BY t1.k;
┌───────┬───────┬───────────┐
│ k │ key │ sub │
├───────┼───────┼───────────┤
│ 1 │ NULL │ [11] │
│ 2 │ 2 │ [12] │
└───────┴───────┴───────────┘
D SELECT k FROM t1 WHERE NULLIF(t1.k, 1) NOT IN (SELECT t2.k + 10 FROM t2
WHERE t2.k = t1.k);
┌───┐
│ k │
├───┤
│ 2 │
└───┘
```
This closes the part of https://github.com/apache/datafusion/issues/25347
that a single key cannot express.
##########
datafusion/sqllogictest/test_files/null_aware_anti_join.slt:
##########
@@ -641,19 +641,23 @@ ORDER BY 1;
statement ok
DROP TABLE naconst_clash;
-# A constant value expression with a non-equality correlation leaves the
-# null-aware join without any equi-join key. Only `HashJoinExec` implements
-# null-aware semantics and it needs a key, so the planner reports the gap
-# instead of falling back to a nested loop join that ignores the NULLs and
-# silently returns wrong results.
+# A constant value expression with a non-equality correlation leaves a residual
+# join filter, and no hash join can mark the UNKNOWN rows of a residual filter.
+# The `NOT IN` does not become an anti join. It becomes the mark joins that
+# materialize its three-valued result, the same plan as for a projected `IN`,
+# and a filter on that result. For `id = 1` the correlated subquery result is
+# `{NULL}`, so `3 NOT IN (...)` is UNKNOWN and the row is dropped. For `id = 2`
+# the result is empty and the row is kept.
statement ok
CREATE TABLE naconst_corr_t1(id INT, g INT) AS VALUES (1, 1), (2, 2);
statement ok
CREATE TABLE naconst_corr_t2(id INT, g INT) AS VALUES (1, 1), (NULL, 2);
-query error DataFusion error: Error during planning: null_aware LeftAnti join
requires equi\-join keys, but the join has none
+query I
SELECT id FROM naconst_corr_t1 WHERE 3 NOT IN (SELECT id FROM naconst_corr_t2
WHERE naconst_corr_t2.g > naconst_corr_t1.g);
+----
+2
Review Comment:
**This query gave a planning error before, and it now gives a result.**
DuckDB 1.5.2 agrees with the result.
The expectation was:
```
query error DataFusion error: Error during planning: null_aware LeftAnti
join requires equi-join keys, but the join has none
```
`3` is a constant, so the `IN` equality is not an equi-join key, and the
correlation `g > g` is a residual filter. The rule asked for a null-aware anti
join that the planner cannot build, so the query failed. It now becomes the
mark joins that materialize the three-valued result, which need no equi-join
key.
```
D SELECT a.id, a.g, (SELECT list(b.id) FROM naconst_corr_t2 b WHERE b.g >
a.g) AS sub FROM naconst_corr_t1 a ORDER BY a.id;
┌───────┬───────┬──────────┐
│ id │ g │ sub │
├───────┼───────┼──────────┤
│ 1 │ 1 │ [NULL] │
│ 2 │ 2 │ NULL │
└───────┴───────┴──────────┘
D SELECT id FROM naconst_corr_t1 WHERE 3 NOT IN (SELECT id FROM
naconst_corr_t2 WHERE naconst_corr_t2.g > naconst_corr_t1.g);
┌────┐
│ id │
├────┤
│ 2 │
└────┘
```
For `id = 1` the subquery gives `{NULL}`, so `3 NOT IN ({NULL})` is UNKNOWN
and the row goes. For `id = 2` the subquery is empty, so the answer is TRUE and
the row stays.
##########
datafusion/sqllogictest/test_files/subquery_projection.slt:
##########
@@ -97,3 +97,528 @@ FROM outer_values o;
3 NULL
4 true
5 NULL
+
+# Plan shapes and NULL semantics of a projected IN subquery.
+#
+# `n1.id` holds a NULL, `n2.id` holds a NULL, and `n3.id` holds none. The mark
+# column of a LeftMark join carries the three-valued result on its own when the
+# join filter is hashable only, so one join per subquery is enough.
+
+statement ok
+CREATE TABLE n1(id INT, z INT) AS VALUES (1, 10), (2, 20), (NULL, 30), (4, 40);
+
+statement ok
+CREATE TABLE n2(id INT, z INT) AS VALUES (1, 5), (NULL, 50);
+
+statement ok
+CREATE TABLE n3(id INT) AS VALUES (1), (2);
+
+# One hash mark join per subquery. There is no materialization join, so no
+# nested loop join over outer x inner rows.
+query TT
+EXPLAIN SELECT id, id IN (SELECT id FROM n3) AS m3, id IN (SELECT id FROM n2)
AS m2 FROM n1;
+----
+logical_plan
+01)Projection: n1.id, __correlated_sq_1.mark AS m3, __correlated_sq_2.mark AS
m2
+02)--LeftMark Join: n1.id = __correlated_sq_2.id null_aware
+03)----LeftMark Join: n1.id = __correlated_sq_1.id null_aware
+04)------TableScan: n1 projection=[id]
+05)------SubqueryAlias: __correlated_sq_1
+06)--------TableScan: n3 projection=[id]
+07)----SubqueryAlias: __correlated_sq_2
+08)------TableScan: n2 projection=[id]
+physical_plan
+01)ProjectionExec: expr=[id@0 as id, mark@1 as m3, mark@2 as m2]
+02)--HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0)],
null_aware
+03)----HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0)],
null_aware
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------DataSourceExec: partitions=1, partition_sizes=[1]
+06)----DataSourceExec: partitions=1, partition_sizes=[1]
+
+# A non-equality correlation stays a residual join filter, so this query keeps
+# the three-join materialization.
+query TT
+EXPLAIN SELECT id, id IN (SELECT n2.id FROM n2 WHERE n2.z < n1.z) AS m FROM n1;
+----
+logical_plan
+01)Projection: n1.id, CASE WHEN __correlated_sq_1.mark THEN Boolean(true) WHEN
__correlated_sq_2.mark OR n1.id IS NULL AND __correlated_sq_3.mark THEN
Boolean(NULL) ELSE Boolean(false) END AS m
+02)--LeftMark Join: Filter: __correlated_sq_3.z < n1.z
+03)----LeftMark Join: Filter: __correlated_sq_2.z < n1.z
+04)------LeftMark Join: n1.id = __correlated_sq_1.id Filter:
__correlated_sq_1.z < n1.z
+05)--------TableScan: n1 projection=[id, z]
+06)--------SubqueryAlias: __correlated_sq_1
+07)----------TableScan: n2 projection=[id, z]
+08)------SubqueryAlias: __correlated_sq_2
+09)--------Projection: n2.z
+10)----------Filter: n2.id IS NULL
+11)------------TableScan: n2 projection=[id, z]
+12)----SubqueryAlias: __correlated_sq_3
+13)------TableScan: n2 projection=[z]
+physical_plan
+01)ProjectionExec: expr=[id@0 as id, CASE WHEN mark@1 THEN true WHEN mark@2 OR
id@0 IS NULL AND mark@3 THEN NULL ELSE false END as m]
+02)--NestedLoopJoinExec: join_type=RightMark, filter=z@1 < z@0,
projection=[id@0, mark@2, mark@3, mark@4]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+04)----NestedLoopJoinExec: join_type=RightMark, filter=z@1 < z@0
+05)------FilterExec: id@0 IS NULL, projection=[z@1]
+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)], filter=z@1 < z@0
+09)----------DataSourceExec: partitions=1, partition_sizes=[1]
+10)----------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query IB rowsort
+SELECT id, id IN (SELECT n2.id FROM n2 WHERE n2.z < n1.z) AS m FROM n1;
+----
+1 true
+2 false
+4 false
+NULL NULL
+
+query IBBB rowsort
+SELECT
+ id,
+ id IN (SELECT id FROM n3) AS m3,
+ EXISTS (SELECT 1 FROM n2 WHERE n2.id = n1.id) AS e2,
+ id NOT IN (SELECT id FROM n3 WHERE n3.id > 1) AS nn3
+FROM n1;
+----
+1 true true true
+2 true false false
+4 false false true
+NULL NULL false NULL
+
+query IB rowsort
+SELECT id, id IN (SELECT id FROM n2) AS m FROM n1;
+----
+1 true
+2 NULL
+4 NULL
+NULL NULL
+
+query IB rowsort
+SELECT id, id NOT IN (SELECT id FROM n2) AS m FROM n1;
+----
+1 false
+2 NULL
+4 NULL
+NULL NULL
+
+query IB rowsort
+SELECT id, id IN (SELECT id FROM n3) AS m FROM n1;
+----
+1 true
+2 true
+4 false
+NULL NULL
+
+query IT rowsort
+SELECT id, CASE WHEN NOT (id IN (SELECT id FROM n2)) THEN 'a' ELSE 'b' END AS
c FROM n1;
+----
+1 b
+2 b
+4 b
+NULL b
+
+query IB rowsort
+SELECT z, sum(id) IN (SELECT id FROM n3) AS m FROM n1 GROUP BY z;
+----
+10 true
+20 true
+30 NULL
+40 false
+
+query IB rowsort
+SELECT id, COALESCE((id IN (SELECT id FROM n3))::boolean, false) AS matched
FROM n1;
+----
+1 true
+2 true
+4 false
+NULL false
+
+query IT rowsort
+SELECT id, CASE WHEN id NOT IN (SELECT n2.id FROM n2 WHERE n2.z < n1.z) THEN
'a' ELSE 'b' END AS c FROM n1;
+----
+1 b
+2 a
+4 a
+NULL b
+
+statement ok
+DROP TABLE n1;
+
+statement ok
+DROP TABLE n2;
+
+statement ok
+DROP TABLE n3;
+
+# Nullable key expressions over non-nullable columns.
+#
+# `nn.id` and `nn.s` are not nullable, but a key expression over them can still
+# be NULL. `NULLIF(id, 1)` is NULL for `id = 1`, and `TRY_CAST(s AS INT)` is
+# NULL when the text is not a number. The join must be null-aware for these
+# keys, so the mark is NULL and `IN` gives UNKNOWN.
+
+statement ok
+CREATE TABLE nn(id INT NOT NULL, s VARCHAR NOT NULL) AS VALUES (1, '1'), (2,
'x'), (4, '4');
+
+statement ok
+CREATE TABLE r3(id INT NOT NULL) AS VALUES (1), (2);
+
+statement ok
+CREATE TABLE r3n(id INT NOT NULL) AS VALUES (1), (2), (5);
+
+# The nullable key expression keeps the plan at one null-aware mark join.
+query TT
+EXPLAIN SELECT id, NULLIF(id, 1) IN (SELECT id FROM r3) AS m FROM nn;
+----
+logical_plan
+01)Projection: nn.id, __correlated_sq_1.mark AS m
+02)--LeftMark Join: nullif(CAST(nn.id AS Int64), Int64(1)) =
__correlated_sq_1.r3.id null_aware
+03)----TableScan: nn projection=[id]
+04)----SubqueryAlias: __correlated_sq_1
+05)------Projection: CAST(r3.id AS Int64)
+06)--------TableScan: r3 projection=[id]
+physical_plan
+01)ProjectionExec: expr=[id@0 as id, mark@1 as m]
+02)--HashJoinExec: mode=CollectLeft, join_type=LeftMark,
on=[(nullif(nn.id,Int64(1))@1, r3.id@0)], projection=[id@0, mark@2], null_aware
+03)----ProjectionExec: expr=[id@0 as id, nullif(CAST(id@0 AS Int64), 1) as
nullif(nn.id,Int64(1))]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)----ProjectionExec: expr=[CAST(id@0 AS Int64) as r3.id]
+06)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# `NULLIF(id, 1)` is NULL for `id = 1`, and `r3` has no NULL, so the answer is
+# UNKNOWN for that row.
+query IB rowsort
+SELECT id, NULLIF(id, 1) IN (SELECT id FROM r3) AS m FROM nn;
+----
+1 NULL
+2 true
+4 false
+
+# `TRY_CAST('x' AS INT)` is NULL, so the answer is UNKNOWN for that row.
+query IB rowsort
+SELECT id, TRY_CAST(s AS INT) IN (SELECT id FROM r3) AS m FROM nn;
+----
+1 true
+2 NULL
+4 false
+
+# The same on the subquery side: the output of the subquery holds a NULL, so a
+# row with no match is UNKNOWN.
+query IB rowsort
+SELECT id, id IN (SELECT NULLIF(id, 5) FROM r3n) AS m FROM nn;
+----
+1 true
+2 true
+4 NULL
+
+# `NOT IN` reads the same mark column, negated.
+query IB rowsort
+SELECT id, NULLIF(id, 1) NOT IN (SELECT id FROM r3) AS m FROM nn;
+----
+1 NULL
+2 false
+4 true
+
+# The `NOT IN` filter path builds a LeftAnti join and reads the key nullability
+# the same way. UNKNOWN does not pass a filter, so `id = 1` drops out.
+query I rowsort
+SELECT id FROM nn WHERE NULLIF(id, 1) NOT IN (SELECT id FROM r3);
+----
+4
+
+# An empty subquery gives `false` also for a NULL key, and the plan stays one
+# null-aware mark join.
+statement ok
+CREATE TABLE r_empty(id INT NOT NULL) AS SELECT * FROM r3 WHERE false;
+
+query TT
+EXPLAIN SELECT id, NULLIF(id, 1) IN (SELECT id FROM r_empty) AS m FROM nn;
+----
+logical_plan
+01)Projection: nn.id, __correlated_sq_1.mark AS m
+02)--LeftMark Join: nullif(CAST(nn.id AS Int64), Int64(1)) =
__correlated_sq_1.r_empty.id null_aware
+03)----TableScan: nn projection=[id]
+04)----SubqueryAlias: __correlated_sq_1
+05)------Projection: CAST(r_empty.id AS Int64)
+06)--------TableScan: r_empty projection=[id]
+physical_plan
+01)ProjectionExec: expr=[id@0 as id, mark@1 as m]
+02)--HashJoinExec: mode=CollectLeft, join_type=LeftMark,
on=[(nullif(nn.id,Int64(1))@1, r_empty.id@0)], projection=[id@0, mark@2],
null_aware
+03)----ProjectionExec: expr=[id@0 as id, nullif(CAST(id@0 AS Int64), 1) as
nullif(nn.id,Int64(1))]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)----ProjectionExec: expr=[CAST(id@0 AS Int64) as r_empty.id]
+06)------DataSourceExec: partitions=1, partition_sizes=[0]
+
+query IB rowsort
+SELECT id, NULLIF(id, 1) IN (SELECT id FROM r_empty) AS m FROM nn;
+----
+1 false
+2 false
+4 false
+
+query IB rowsort
+SELECT id, NULLIF(id, 1) NOT IN (SELECT id FROM r_empty) AS m FROM nn;
+----
+1 true
+2 true
+4 true
+
+statement ok
+DROP TABLE r_empty;
+
+# A correlated `NOT IN` filter builds a `LeftAnti` join with two keys: the
+# value and the correlation. A null-aware `LeftAnti` hash join supports one key
+# only, so a function key over non-nullable columns must not make this join
+# null-aware.
+statement ok
+CREATE TABLE t1(k INT NOT NULL, s VARCHAR NOT NULL) AS VALUES (1, 'a'), (2,
'b');
+
+statement ok
+CREATE TABLE t2(k INT NOT NULL, s VARCHAR NOT NULL) AS VALUES (1, 'B'), (2,
'B');
+
+query IT rowsort
+SELECT * FROM t1 WHERE upper(t1.s) NOT IN (SELECT t2.s FROM t2 WHERE t2.k =
t1.k);
+----
+1 a
+
+# `NULLIF(k, 1)` is NULL for `k = 1`, and that group of `t2` is not empty, so
+# the correct result has no row for `k = 1`. The join has two keys, the value
+# and the correlation, and the null-aware `LeftAnti` executor takes one key
+# only. The `NOT IN` becomes a null-aware mark join, which takes any number of
+# keys, and a filter on the mark. `main` keeps the `k = 1` row, which is
+# https://github.com/apache/datafusion/issues/25347.
+query I rowsort
+SELECT k FROM t1 WHERE NULLIF(t1.k, 1) NOT IN (SELECT t2.k + 10 FROM t2 WHERE
t2.k = t1.k);
+----
+2
+
+query TT
+EXPLAIN SELECT k FROM t1 WHERE NULLIF(t1.k, 1) NOT IN (SELECT t2.k + 10 FROM
t2 WHERE t2.k = t1.k);
+----
+logical_plan
+01)Projection: t1.k
+02)--Filter: NOT __correlated_sq_1.mark
+03)----LeftMark Join: nullif(CAST(t1.k AS Int64), Int64(1)) =
__correlated_sq_1.t2.k + Int64(10), t1.k = __correlated_sq_1.k null_aware
+04)------TableScan: t1 projection=[k]
+05)------SubqueryAlias: __correlated_sq_1
+06)--------Projection: CAST(t2.k AS Int64) + Int64(10), t2.k
+07)----------TableScan: t2 projection=[k]
+physical_plan
+01)FilterExec: NOT mark@1, projection=[k@0]
+02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+03)----HashJoinExec: mode=CollectLeft, join_type=LeftMark,
on=[(nullif(t1.k,Int64(1))@1, t2.k + Int64(10)@0), (k@0, k@1)],
projection=[k@0, mark@2], null_aware
+04)------ProjectionExec: expr=[k@0 as k, nullif(CAST(k@0 AS Int64), 1) as
nullif(t1.k,Int64(1))]
+05)--------DataSourceExec: partitions=1, partition_sizes=[1]
+06)------ProjectionExec: expr=[CAST(k@0 AS Int64) + 10 as t2.k + Int64(10),
k@0 as k]
+07)--------DataSourceExec: partitions=1, partition_sizes=[1]
+
+statement ok
+DROP TABLE t1;
+
+statement ok
+DROP TABLE t2;
+
+# A non-equality correlation leaves one key and a residual join filter. No hash
+# join can mark the UNKNOWN rows of a residual filter
+# (https://github.com/apache/datafusion/issues/25336), so a `NOT IN` whose
+# value can be NULL does not become an anti join. It becomes the mark joins
+# that materialize its three-valued result, the same plan as for a projected
+# `IN`, and a filter on that result. For `k = 1` the key is NULL and the
+# correlated subquery result is empty, and `NULL NOT IN (<empty set>)` is TRUE.
+# Both rows are correct.
+statement ok
+CREATE TABLE ra(k INT NOT NULL, z INT NOT NULL) AS VALUES (1, 10), (2, 20);
+
+statement ok
+CREATE TABLE rb(k INT NOT NULL, z INT NOT NULL) AS VALUES (5, 50);
+
+query I rowsort
+SELECT k FROM ra WHERE NULLIF(ra.k, 1) NOT IN (SELECT rb.k FROM rb WHERE rb.z
< ra.z);
+----
+1
+2
+
+# The same shape where the correlated subquery result is not empty for the NULL
+# key: `NULL NOT IN ({5})` is UNKNOWN, so `2` is the only row. `main` builds a
+# plain anti join here and also keeps `1`.
+query I rowsort
+SELECT k FROM ra WHERE NULLIF(ra.k, 1) NOT IN (SELECT rb.k FROM rb WHERE rb.z
> ra.z);
+----
+2
+
+query TT
+EXPLAIN SELECT k FROM ra WHERE NULLIF(ra.k, 1) NOT IN (SELECT rb.k FROM rb
WHERE rb.z > ra.z);
+----
+logical_plan
+01)Projection: ra.k
+02)--Filter: NOT CASE WHEN __correlated_sq_2.mark THEN Boolean(true) WHEN
__correlated_sq_3.mark OR nullif(CAST(ra.k AS Int64), Int64(1)) IS NULL AND
__correlated_sq_4.mark THEN Boolean(NULL) ELSE Boolean(false) END
+03)----Projection: ra.k, __correlated_sq_2.mark, __correlated_sq_3.mark,
__correlated_sq_4.mark
+04)------LeftMark Join: Filter: __correlated_sq_4.z > ra.z
+05)--------LeftMark Join: Filter: __correlated_sq_3.z > ra.z
+06)----------LeftMark Join: nullif(CAST(ra.k AS Int64), Int64(1)) =
__correlated_sq_2.rb.k Filter: __correlated_sq_2.z > ra.z
+07)------------TableScan: ra projection=[k, z]
+08)------------SubqueryAlias: __correlated_sq_2
+09)--------------Projection: CAST(rb.k AS Int64), rb.z
+10)----------------TableScan: rb projection=[k, z]
+11)----------EmptyRelation: rows=0
+12)--------SubqueryAlias: __correlated_sq_4
+13)----------TableScan: rb projection=[z]
+physical_plan
+01)FilterExec: NOT CASE WHEN mark@1 THEN true WHEN mark@2 OR nullif(CAST(k@0
AS Int64), 1) IS NULL AND mark@3 THEN NULL ELSE false END, projection=[k@0]
+02)--NestedLoopJoinExec: join_type=RightMark, filter=z@1 > z@0,
projection=[k@0, mark@2, mark@3, mark@4]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+04)----NestedLoopJoinExec: join_type=RightMark, filter=z@1 > z@0
+05)------EmptyExec
+06)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+07)--------HashJoinExec: mode=CollectLeft, join_type=RightMark, on=[(rb.k@0,
nullif(ra.k,Int64(1))@2)], filter=z@1 > z@0, projection=[k@0, z@1, mark@3]
+08)----------ProjectionExec: expr=[CAST(k@0 AS Int64) as rb.k, z@1 as z]
+09)------------DataSourceExec: partitions=1, partition_sizes=[1]
+10)----------ProjectionExec: expr=[k@0 as k, z@1 as z, nullif(CAST(k@0 AS
Int64), 1) as nullif(ra.k,Int64(1))]
+11)------------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# The correlation names no column of the subquery, so it stays as a residual
+# filter and the subquery is either the whole of `ic` or empty. A NULL in
+# `ic.id` then makes the answer UNKNOWN only for the rows whose correlation
+# holds. `main` builds a null-aware anti join that does not apply the residual
+# when it looks for a NULL, sees one in `ic`, and drops every row. This is the
+# shape whose plan `joins.slt` pins.
+statement ok
+CREATE TABLE oc(id INT, g INT) AS VALUES (1, 1), (2, 0), (NULL, 0);
+
+statement ok
+CREATE TABLE ic(id INT) AS VALUES (5), (NULL);
+
+# `g > 0` holds for `id = 1` only, so its subquery is `{5, NULL}` and
+# `1 NOT IN {5, NULL}` is UNKNOWN. The other two rows have an empty subquery,
+# and `<anything> NOT IN (<empty set>)` is TRUE, the NULL row included.
+query I rowsort
+SELECT id FROM oc WHERE oc.id NOT IN (SELECT ic.id FROM ic WHERE oc.g > 0);
+----
+2
+NULL
Review Comment:
**New test, and it is the one that shows what the `joins.slt` plan change
buys.** `main` returns no rows for this query. DuckDB 1.5.2 returns the two
rows below.
The correlation `oc.g > 0` names no column of `ic`, so it cannot become a
join key and stays as a residual filter. The subquery is therefore the whole of
`ic` for an outer row whose `g > 0`, and empty for every other row.
```
D SELECT oc.id, oc.g, (SELECT list(ic.id) FROM ic WHERE oc.g > 0) AS sub
FROM oc ORDER BY oc.id;
┌───────┬───────┬─────────────┐
│ id │ g │ sub │
├───────┼───────┼─────────────┤
│ 1 │ 1 │ [5, NULL] │
│ 2 │ 0 │ NULL │
│ NULL │ 0 │ NULL │
└───────┴───────┴─────────────┘
D SELECT id FROM oc WHERE oc.id NOT IN (SELECT ic.id FROM ic WHERE oc.g > 0);
┌───────┐
│ id │
├───────┤
│ 2 │
│ NULL │
└───────┘
```
`main` builds one null-aware anti join and keeps the residual filter on it.
That executor does not apply the residual when it looks for a NULL, so it finds
the NULL in `ic` and treats every outer row as UNKNOWN, including the two whose
subquery is empty. It returns nothing.
The NULL row is kept on purpose: its subquery is empty, and `NULL NOT IN
(<empty set>)` is TRUE.
This is the same shape as the `joins.slt` query below, which had no result
test.
##########
datafusion/sqllogictest/test_files/joins.slt:
##########
@@ -1988,11 +1988,22 @@ where join_t1.t1_id + 12 not in
(select join_t2.t2_id + 1 from join_t2 where join_t1.t1_int > 0)
----
logical_plan
-01)LeftAnti Join: CAST(join_t1.t1_id AS Int64) + Int64(12) =
__correlated_sq_1.join_t2.t2_id + Int64(1) Filter: join_t1.t1_int > UInt32(0)
null_aware
-02)--TableScan: join_t1 projection=[t1_id, t1_name, t1_int]
-03)--SubqueryAlias: __correlated_sq_1
-04)----Projection: CAST(join_t2.t2_id AS Int64) + Int64(1)
-05)------TableScan: join_t2 projection=[t2_id]
+01)Projection: join_t1.t1_id, join_t1.t1_name, join_t1.t1_int
+02)--Filter: NOT CASE WHEN __correlated_sq_2.mark THEN Boolean(true) WHEN
__correlated_sq_3.mark OR CAST(join_t1.t1_id AS Int64) + Int64(12) IS NULL AND
__correlated_sq_4.mark THEN Boolean(NULL) ELSE Boolean(false) END
+03)----LeftMark Join: Filter: join_t1.t1_int > UInt32(0)
+04)------LeftMark Join: Filter: join_t1.t1_int > UInt32(0)
+05)--------LeftMark Join: CAST(join_t1.t1_id AS Int64) + Int64(12) =
__correlated_sq_2.join_t2.t2_id + Int64(1) Filter: join_t1.t1_int > UInt32(0)
Review Comment:
**Only the plan changed here. The result of this query is the same before
and after, and it is correct in both.** I want to flag the cost, because this
file tests the plan and not the result.
```
SELECT t1_id, t1_name, t1_int FROM join_t1 WHERE join_t1.t1_id + 12 NOT IN
(SELECT join_t2.t2_id + 1 FROM join_t2 WHERE join_t1.t1_int > 0);
-- main, this branch and DuckDB 1.5.2 all give: 22 b 2
```
`join_t2.t2_id` holds no NULL in this file, so the old single anti join was
correct for this data. The plan is now three joins for the same answer, and two
of them are nested loop joins that carry only the outer-side predicate.
The reason is that the shape is not correct in general. The correlation
`join_t1.t1_int > 0` names no column of the subquery, so it stays as a residual
filter, and the null-aware anti join executor ignores the residual when it
decides whether a NULL makes the result UNKNOWN. Put one NULL in the subquery
column and `main` drops every row:
```sql
CREATE TABLE jt1(t1_id INT, t1_int INT) AS VALUES (11,1),(22,2),(33,0);
CREATE TABLE jt2(t2_id INT) AS VALUES (23),(NULL);
SELECT t1_id, t1_int FROM jt1 WHERE jt1.t1_id + 12 NOT IN (SELECT jt2.t2_id
+ 1 FROM jt2 WHERE jt1.t1_int > 0);
-- main: (no rows)
-- this branch: 33 0
-- DuckDB: 33 0
```
The row `t1_id = 33` has `t1_int = 0`, so its correlated subquery is empty
and `NOT IN (<empty set>)` is TRUE.
`subquery_projection.slt` now has this shape as a result test, with the
tables `oc` and `ic`, so the gain is no longer only in this comment.
https://github.com/apache/datafusion/issues/25336 is the executor gap behind
it. When https://github.com/apache/datafusion/pull/25339 lands, this shape can
go back to one anti join and this plan becomes small again. I can also hold
this file at the old plan and let only the projected `IN` take the new path, if
you would rather not pay three joins for a `WHERE` clause today.
##########
datafusion/optimizer/src/decorrelate.rs:
##########
@@ -184,6 +195,11 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr {
.all(|&e| can_pullup_over_aggregation(e));
let (mut join_filters, subquery_filters) =
find_join_exprs(subquery_filter_exprs)?;
+ for expr in &join_filters {
Review Comment:
Thank you, this is a real regression. I took your fix in 10ccc3e0a4.
`correlated_filters` now keeps the outer references, and `filter_rejects_null`
matches each side of a conjunct against the side of the join that the key is
on. Your two queries are in `subquery_projection.slt` (tables `sh_o` and
`sh_t`), and DuckDB 1.5.2 gives the same results: NULL for the NULL row, and 8
for the NOT IN filter.
##########
datafusion/sqllogictest/test_files/subquery_projection.slt:
##########
@@ -97,3 +97,528 @@ FROM outer_values o;
3 NULL
4 true
5 NULL
+
+# Plan shapes and NULL semantics of a projected IN subquery.
+#
+# `n1.id` holds a NULL, `n2.id` holds a NULL, and `n3.id` holds none. The mark
+# column of a LeftMark join carries the three-valued result on its own when the
+# join filter is hashable only, so one join per subquery is enough.
+
+statement ok
+CREATE TABLE n1(id INT, z INT) AS VALUES (1, 10), (2, 20), (NULL, 30), (4, 40);
+
+statement ok
+CREATE TABLE n2(id INT, z INT) AS VALUES (1, 5), (NULL, 50);
+
+statement ok
+CREATE TABLE n3(id INT) AS VALUES (1), (2);
+
+# One hash mark join per subquery. There is no materialization join, so no
+# nested loop join over outer x inner rows.
+query TT
+EXPLAIN SELECT id, id IN (SELECT id FROM n3) AS m3, id IN (SELECT id FROM n2)
AS m2 FROM n1;
+----
+logical_plan
+01)Projection: n1.id, __correlated_sq_1.mark AS m3, __correlated_sq_2.mark AS
m2
+02)--LeftMark Join: n1.id = __correlated_sq_2.id null_aware
+03)----LeftMark Join: n1.id = __correlated_sq_1.id null_aware
+04)------TableScan: n1 projection=[id]
+05)------SubqueryAlias: __correlated_sq_1
+06)--------TableScan: n3 projection=[id]
+07)----SubqueryAlias: __correlated_sq_2
+08)------TableScan: n2 projection=[id]
+physical_plan
+01)ProjectionExec: expr=[id@0 as id, mark@1 as m3, mark@2 as m2]
+02)--HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0)],
null_aware
+03)----HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0)],
null_aware
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------DataSourceExec: partitions=1, partition_sizes=[1]
+06)----DataSourceExec: partitions=1, partition_sizes=[1]
+
+# A non-equality correlation stays a residual join filter, so this query keeps
+# the three-join materialization.
+query TT
+EXPLAIN SELECT id, id IN (SELECT n2.id FROM n2 WHERE n2.z < n1.z) AS m FROM n1;
+----
+logical_plan
+01)Projection: n1.id, CASE WHEN __correlated_sq_1.mark THEN Boolean(true) WHEN
__correlated_sq_2.mark OR n1.id IS NULL AND __correlated_sq_3.mark THEN
Boolean(NULL) ELSE Boolean(false) END AS m
+02)--LeftMark Join: Filter: __correlated_sq_3.z < n1.z
+03)----LeftMark Join: Filter: __correlated_sq_2.z < n1.z
+04)------LeftMark Join: n1.id = __correlated_sq_1.id Filter:
__correlated_sq_1.z < n1.z
+05)--------TableScan: n1 projection=[id, z]
+06)--------SubqueryAlias: __correlated_sq_1
+07)----------TableScan: n2 projection=[id, z]
+08)------SubqueryAlias: __correlated_sq_2
+09)--------Projection: n2.z
+10)----------Filter: n2.id IS NULL
+11)------------TableScan: n2 projection=[id, z]
+12)----SubqueryAlias: __correlated_sq_3
+13)------TableScan: n2 projection=[z]
+physical_plan
+01)ProjectionExec: expr=[id@0 as id, CASE WHEN mark@1 THEN true WHEN mark@2 OR
id@0 IS NULL AND mark@3 THEN NULL ELSE false END as m]
+02)--NestedLoopJoinExec: join_type=RightMark, filter=z@1 < z@0,
projection=[id@0, mark@2, mark@3, mark@4]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+04)----NestedLoopJoinExec: join_type=RightMark, filter=z@1 < z@0
+05)------FilterExec: id@0 IS NULL, projection=[z@1]
+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)], filter=z@1 < z@0
+09)----------DataSourceExec: partitions=1, partition_sizes=[1]
+10)----------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query IB rowsort
+SELECT id, id IN (SELECT n2.id FROM n2 WHERE n2.z < n1.z) AS m FROM n1;
+----
+1 true
+2 false
+4 false
+NULL NULL
+
+query IBBB rowsort
+SELECT
+ id,
+ id IN (SELECT id FROM n3) AS m3,
+ EXISTS (SELECT 1 FROM n2 WHERE n2.id = n1.id) AS e2,
+ id NOT IN (SELECT id FROM n3 WHERE n3.id > 1) AS nn3
+FROM n1;
+----
+1 true true true
+2 true false false
+4 false false true
+NULL NULL false NULL
+
+query IB rowsort
+SELECT id, id IN (SELECT id FROM n2) AS m FROM n1;
+----
+1 true
+2 NULL
+4 NULL
+NULL NULL
+
+query IB rowsort
+SELECT id, id NOT IN (SELECT id FROM n2) AS m FROM n1;
+----
+1 false
+2 NULL
+4 NULL
+NULL NULL
+
+query IB rowsort
+SELECT id, id IN (SELECT id FROM n3) AS m FROM n1;
+----
+1 true
+2 true
+4 false
+NULL NULL
+
+query IT rowsort
+SELECT id, CASE WHEN NOT (id IN (SELECT id FROM n2)) THEN 'a' ELSE 'b' END AS
c FROM n1;
+----
+1 b
+2 b
+4 b
+NULL b
+
+query IB rowsort
+SELECT z, sum(id) IN (SELECT id FROM n3) AS m FROM n1 GROUP BY z;
+----
+10 true
+20 true
+30 NULL
+40 false
+
+query IB rowsort
+SELECT id, COALESCE((id IN (SELECT id FROM n3))::boolean, false) AS matched
FROM n1;
+----
+1 true
+2 true
+4 false
+NULL false
+
+query IT rowsort
+SELECT id, CASE WHEN id NOT IN (SELECT n2.id FROM n2 WHERE n2.z < n1.z) THEN
'a' ELSE 'b' END AS c FROM n1;
+----
+1 b
+2 a
+4 a
+NULL b
+
+statement ok
+DROP TABLE n1;
+
+statement ok
+DROP TABLE n2;
+
+statement ok
+DROP TABLE n3;
+
+# Nullable key expressions over non-nullable columns.
+#
+# `nn.id` and `nn.s` are not nullable, but a key expression over them can still
+# be NULL. `NULLIF(id, 1)` is NULL for `id = 1`, and `TRY_CAST(s AS INT)` is
+# NULL when the text is not a number. The join must be null-aware for these
+# keys, so the mark is NULL and `IN` gives UNKNOWN.
+
+statement ok
+CREATE TABLE nn(id INT NOT NULL, s VARCHAR NOT NULL) AS VALUES (1, '1'), (2,
'x'), (4, '4');
+
+statement ok
+CREATE TABLE r3(id INT NOT NULL) AS VALUES (1), (2);
+
+statement ok
+CREATE TABLE r3n(id INT NOT NULL) AS VALUES (1), (2), (5);
+
+# The nullable key expression keeps the plan at one null-aware mark join.
+query TT
+EXPLAIN SELECT id, NULLIF(id, 1) IN (SELECT id FROM r3) AS m FROM nn;
+----
+logical_plan
+01)Projection: nn.id, __correlated_sq_1.mark AS m
+02)--LeftMark Join: nullif(CAST(nn.id AS Int64), Int64(1)) =
__correlated_sq_1.r3.id null_aware
+03)----TableScan: nn projection=[id]
+04)----SubqueryAlias: __correlated_sq_1
+05)------Projection: CAST(r3.id AS Int64)
+06)--------TableScan: r3 projection=[id]
+physical_plan
+01)ProjectionExec: expr=[id@0 as id, mark@1 as m]
+02)--HashJoinExec: mode=CollectLeft, join_type=LeftMark,
on=[(nullif(nn.id,Int64(1))@1, r3.id@0)], projection=[id@0, mark@2], null_aware
+03)----ProjectionExec: expr=[id@0 as id, nullif(CAST(id@0 AS Int64), 1) as
nullif(nn.id,Int64(1))]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)----ProjectionExec: expr=[CAST(id@0 AS Int64) as r3.id]
+06)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# `NULLIF(id, 1)` is NULL for `id = 1`, and `r3` has no NULL, so the answer is
+# UNKNOWN for that row.
+query IB rowsort
+SELECT id, NULLIF(id, 1) IN (SELECT id FROM r3) AS m FROM nn;
+----
+1 NULL
+2 true
+4 false
+
+# `TRY_CAST('x' AS INT)` is NULL, so the answer is UNKNOWN for that row.
+query IB rowsort
+SELECT id, TRY_CAST(s AS INT) IN (SELECT id FROM r3) AS m FROM nn;
+----
+1 true
+2 NULL
+4 false
+
+# The same on the subquery side: the output of the subquery holds a NULL, so a
+# row with no match is UNKNOWN.
+query IB rowsort
+SELECT id, id IN (SELECT NULLIF(id, 5) FROM r3n) AS m FROM nn;
+----
+1 true
+2 true
+4 NULL
+
+# `NOT IN` reads the same mark column, negated.
+query IB rowsort
+SELECT id, NULLIF(id, 1) NOT IN (SELECT id FROM r3) AS m FROM nn;
+----
+1 NULL
+2 false
+4 true
+
+# The `NOT IN` filter path builds a LeftAnti join and reads the key nullability
+# the same way. UNKNOWN does not pass a filter, so `id = 1` drops out.
+query I rowsort
+SELECT id FROM nn WHERE NULLIF(id, 1) NOT IN (SELECT id FROM r3);
+----
+4
+
+# An empty subquery gives `false` also for a NULL key, and the plan stays one
+# null-aware mark join.
+statement ok
+CREATE TABLE r_empty(id INT NOT NULL) AS SELECT * FROM r3 WHERE false;
+
+query TT
+EXPLAIN SELECT id, NULLIF(id, 1) IN (SELECT id FROM r_empty) AS m FROM nn;
+----
+logical_plan
+01)Projection: nn.id, __correlated_sq_1.mark AS m
+02)--LeftMark Join: nullif(CAST(nn.id AS Int64), Int64(1)) =
__correlated_sq_1.r_empty.id null_aware
+03)----TableScan: nn projection=[id]
+04)----SubqueryAlias: __correlated_sq_1
+05)------Projection: CAST(r_empty.id AS Int64)
+06)--------TableScan: r_empty projection=[id]
+physical_plan
+01)ProjectionExec: expr=[id@0 as id, mark@1 as m]
+02)--HashJoinExec: mode=CollectLeft, join_type=LeftMark,
on=[(nullif(nn.id,Int64(1))@1, r_empty.id@0)], projection=[id@0, mark@2],
null_aware
+03)----ProjectionExec: expr=[id@0 as id, nullif(CAST(id@0 AS Int64), 1) as
nullif(nn.id,Int64(1))]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)----ProjectionExec: expr=[CAST(id@0 AS Int64) as r_empty.id]
+06)------DataSourceExec: partitions=1, partition_sizes=[0]
+
+query IB rowsort
+SELECT id, NULLIF(id, 1) IN (SELECT id FROM r_empty) AS m FROM nn;
+----
+1 false
+2 false
+4 false
+
+query IB rowsort
+SELECT id, NULLIF(id, 1) NOT IN (SELECT id FROM r_empty) AS m FROM nn;
+----
+1 true
+2 true
+4 true
+
+statement ok
+DROP TABLE r_empty;
+
+# A correlated `NOT IN` filter builds a `LeftAnti` join with two keys: the
+# value and the correlation. A null-aware `LeftAnti` hash join supports one key
+# only, so a function key over non-nullable columns must not make this join
+# null-aware.
+statement ok
+CREATE TABLE t1(k INT NOT NULL, s VARCHAR NOT NULL) AS VALUES (1, 'a'), (2,
'b');
+
+statement ok
+CREATE TABLE t2(k INT NOT NULL, s VARCHAR NOT NULL) AS VALUES (1, 'B'), (2,
'B');
+
+query IT rowsort
+SELECT * FROM t1 WHERE upper(t1.s) NOT IN (SELECT t2.s FROM t2 WHERE t2.k =
t1.k);
+----
+1 a
+
+# `NULLIF(k, 1)` is NULL for `k = 1`, and that group of `t2` is not empty, so
+# the correct result has no row for `k = 1`. The join has two keys, the value
+# and the correlation, and the null-aware `LeftAnti` executor takes one key
+# only. The `NOT IN` becomes a null-aware mark join, which takes any number of
+# keys, and a filter on the mark. `main` keeps the `k = 1` row, which is
+# https://github.com/apache/datafusion/issues/25347.
+query I rowsort
+SELECT k FROM t1 WHERE NULLIF(t1.k, 1) NOT IN (SELECT t2.k + 10 FROM t2 WHERE
t2.k = t1.k);
+----
+2
+
+query TT
+EXPLAIN SELECT k FROM t1 WHERE NULLIF(t1.k, 1) NOT IN (SELECT t2.k + 10 FROM
t2 WHERE t2.k = t1.k);
+----
+logical_plan
+01)Projection: t1.k
+02)--Filter: NOT __correlated_sq_1.mark
+03)----LeftMark Join: nullif(CAST(t1.k AS Int64), Int64(1)) =
__correlated_sq_1.t2.k + Int64(10), t1.k = __correlated_sq_1.k null_aware
+04)------TableScan: t1 projection=[k]
+05)------SubqueryAlias: __correlated_sq_1
+06)--------Projection: CAST(t2.k AS Int64) + Int64(10), t2.k
+07)----------TableScan: t2 projection=[k]
+physical_plan
+01)FilterExec: NOT mark@1, projection=[k@0]
+02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+03)----HashJoinExec: mode=CollectLeft, join_type=LeftMark,
on=[(nullif(t1.k,Int64(1))@1, t2.k + Int64(10)@0), (k@0, k@1)],
projection=[k@0, mark@2], null_aware
+04)------ProjectionExec: expr=[k@0 as k, nullif(CAST(k@0 AS Int64), 1) as
nullif(t1.k,Int64(1))]
+05)--------DataSourceExec: partitions=1, partition_sizes=[1]
+06)------ProjectionExec: expr=[CAST(k@0 AS Int64) + 10 as t2.k + Int64(10),
k@0 as k]
+07)--------DataSourceExec: partitions=1, partition_sizes=[1]
+
+statement ok
+DROP TABLE t1;
+
+statement ok
+DROP TABLE t2;
+
+# A non-equality correlation leaves one key and a residual join filter. No hash
+# join can mark the UNKNOWN rows of a residual filter
+# (https://github.com/apache/datafusion/issues/25336), so a `NOT IN` whose
+# value can be NULL does not become an anti join. It becomes the mark joins
+# that materialize its three-valued result, the same plan as for a projected
+# `IN`, and a filter on that result. For `k = 1` the key is NULL and the
+# correlated subquery result is empty, and `NULL NOT IN (<empty set>)` is TRUE.
+# Both rows are correct.
+statement ok
+CREATE TABLE ra(k INT NOT NULL, z INT NOT NULL) AS VALUES (1, 10), (2, 20);
+
+statement ok
+CREATE TABLE rb(k INT NOT NULL, z INT NOT NULL) AS VALUES (5, 50);
+
+query I rowsort
+SELECT k FROM ra WHERE NULLIF(ra.k, 1) NOT IN (SELECT rb.k FROM rb WHERE rb.z
< ra.z);
+----
+1
+2
+
+# The same shape where the correlated subquery result is not empty for the NULL
+# key: `NULL NOT IN ({5})` is UNKNOWN, so `2` is the only row. `main` builds a
+# plain anti join here and also keeps `1`.
+query I rowsort
+SELECT k FROM ra WHERE NULLIF(ra.k, 1) NOT IN (SELECT rb.k FROM rb WHERE rb.z
> ra.z);
+----
+2
+
+query TT
+EXPLAIN SELECT k FROM ra WHERE NULLIF(ra.k, 1) NOT IN (SELECT rb.k FROM rb
WHERE rb.z > ra.z);
+----
+logical_plan
+01)Projection: ra.k
+02)--Filter: NOT CASE WHEN __correlated_sq_2.mark THEN Boolean(true) WHEN
__correlated_sq_3.mark OR nullif(CAST(ra.k AS Int64), Int64(1)) IS NULL AND
__correlated_sq_4.mark THEN Boolean(NULL) ELSE Boolean(false) END
+03)----Projection: ra.k, __correlated_sq_2.mark, __correlated_sq_3.mark,
__correlated_sq_4.mark
+04)------LeftMark Join: Filter: __correlated_sq_4.z > ra.z
+05)--------LeftMark Join: Filter: __correlated_sq_3.z > ra.z
+06)----------LeftMark Join: nullif(CAST(ra.k AS Int64), Int64(1)) =
__correlated_sq_2.rb.k Filter: __correlated_sq_2.z > ra.z
+07)------------TableScan: ra projection=[k, z]
+08)------------SubqueryAlias: __correlated_sq_2
+09)--------------Projection: CAST(rb.k AS Int64), rb.z
+10)----------------TableScan: rb projection=[k, z]
+11)----------EmptyRelation: rows=0
+12)--------SubqueryAlias: __correlated_sq_4
+13)----------TableScan: rb projection=[z]
+physical_plan
+01)FilterExec: NOT CASE WHEN mark@1 THEN true WHEN mark@2 OR nullif(CAST(k@0
AS Int64), 1) IS NULL AND mark@3 THEN NULL ELSE false END, projection=[k@0]
+02)--NestedLoopJoinExec: join_type=RightMark, filter=z@1 > z@0,
projection=[k@0, mark@2, mark@3, mark@4]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+04)----NestedLoopJoinExec: join_type=RightMark, filter=z@1 > z@0
+05)------EmptyExec
+06)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+07)--------HashJoinExec: mode=CollectLeft, join_type=RightMark, on=[(rb.k@0,
nullif(ra.k,Int64(1))@2)], filter=z@1 > z@0, projection=[k@0, z@1, mark@3]
+08)----------ProjectionExec: expr=[CAST(k@0 AS Int64) as rb.k, z@1 as z]
+09)------------DataSourceExec: partitions=1, partition_sizes=[1]
+10)----------ProjectionExec: expr=[k@0 as k, z@1 as z, nullif(CAST(k@0 AS
Int64), 1) as nullif(ra.k,Int64(1))]
+11)------------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# The correlation names no column of the subquery, so it stays as a residual
+# filter and the subquery is either the whole of `ic` or empty. A NULL in
+# `ic.id` then makes the answer UNKNOWN only for the rows whose correlation
+# holds. `main` builds a null-aware anti join that does not apply the residual
+# when it looks for a NULL, sees one in `ic`, and drops every row. This is the
+# shape whose plan `joins.slt` pins.
+statement ok
+CREATE TABLE oc(id INT, g INT) AS VALUES (1, 1), (2, 0), (NULL, 0);
+
+statement ok
+CREATE TABLE ic(id INT) AS VALUES (5), (NULL);
+
+# `g > 0` holds for `id = 1` only, so its subquery is `{5, NULL}` and
+# `1 NOT IN {5, NULL}` is UNKNOWN. The other two rows have an empty subquery,
+# and `<anything> NOT IN (<empty set>)` is TRUE, the NULL row included.
+query I rowsort
+SELECT id FROM oc WHERE oc.id NOT IN (SELECT ic.id FROM ic WHERE oc.g > 0);
+----
+2
+NULL
+
+statement ok
+DROP TABLE oc;
+
+statement ok
+DROP TABLE ic;
+
+statement ok
+DROP TABLE ra;
+
+statement ok
+DROP TABLE rb;
+
+statement ok
+DROP TABLE nn;
+
+statement ok
+DROP TABLE r3;
+
+statement ok
+DROP TABLE r3n;
+
+# A correlation that repeats the `IN` predicate.
+#
+# `x IN (SELECT y FROM .. WHERE y = x)` writes the `IN` equality a second time.
+# The decorrelation drops the duplicate and adds the same equality back as the
+# join filter, so the scope of the subquery is gone by the time the join is
+# built. The result of this shape is never UNKNOWN: every row the subquery
+# keeps for an outer row satisfies `y = x`, so `y` is not NULL there and the
+# subquery result is either empty or `{x}`. The join must therefore not be
+# null-aware. All results below agree with DuckDB 1.5.2.
+
+statement ok
+CREATE TABLE co(id INT, k INT) AS VALUES
(1,1),(2,1),(NULL,1),(2,2),(NULL,3),(5,NULL),(9,9);
+
+statement ok
+CREATE TABLE ci(id INT, k INT) AS VALUES (1,1),(NULL,2),(7,1),(3,NULL);
+
+# The mark join carries the whole result and is not null-aware.
+query TT
+EXPLAIN SELECT co.id, co.k IN (SELECT ci.k FROM ci WHERE ci.k = co.k) AS m
FROM co;
+----
+logical_plan
+01)Projection: co.id, __correlated_sq_1.mark AS m
+02)--LeftMark Join: co.k = __correlated_sq_1.k
+03)----TableScan: co projection=[id, k]
+04)----SubqueryAlias: __correlated_sq_1
+05)------TableScan: ci projection=[k]
+physical_plan
+01)ProjectionExec: expr=[id@0 as id, mark@1 as m]
+02)--HashJoinExec: mode=CollectLeft, join_type=RightMark, on=[(k@0, k@1)],
projection=[id@0, mark@2]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+04)----DataSourceExec: partitions=1, partition_sizes=[1]
+
+query IIB
+SELECT co.id, co.k, co.k IN (SELECT ci.k FROM ci WHERE ci.k = co.k) AS m FROM
co ORDER BY k, id;
+----
+1 1 true
+2 1 true
+NULL 1 true
+2 2 true
+NULL 3 false
+9 9 false
+5 NULL false
+
+# The `IN` value carries a second correlation, which stays a join key.
+query IIB
+SELECT co.id, co.k, co.id IN (SELECT ci.id FROM ci WHERE ci.id = co.id AND
ci.k = co.k) AS m FROM co ORDER BY k, id;
+----
+1 1 true
+2 1 false
+NULL 1 false
+2 2 false
+NULL 3 false
+9 9 false
+5 NULL false
+
+# The same shape in a `WHERE` clause builds a `LeftAnti` join, which must not
be
+# null-aware either. `main` gives no row here, which is
+# https://github.com/apache/datafusion/issues/25480.
+query II
+SELECT co.id, co.k FROM co WHERE co.k NOT IN (SELECT ci.k FROM ci WHERE ci.k =
co.k) ORDER BY k, id;
+----
+NULL 3
+9 9
+5 NULL
+
+# The same shape with an expression as the value. The subquery output is then a
+# cast, and the join keys refer to it by its column name. The correlation still
+# names the expression as the query writes it, and the join is not null-aware.
+query IIB
+SELECT co.id, co.k, (co.k + 0) IN (SELECT ci.k FROM ci WHERE ci.k = co.k + 0)
AS m FROM co ORDER BY k, id;
+----
+1 1 true
+2 1 true
+NULL 1 true
+2 2 true
+NULL 3 false
+9 9 false
+5 NULL false
+
+# A grouping set above the correlated filter is a different problem. The
+# grand-total row of `ROLLUP` exists for every outer row, so a miss is UNKNOWN
+# here, not FALSE. The pull up moves the filter above the aggregate, which
+# loses that row; the same plan gives a wrong `EXISTS` too. That is a bug in
+# the pull up, https://github.com/apache/datafusion/issues/25519, and is not
+# changed here: the three rows for `k = 3`, `k = 9` and `k = NULL` should be
+# NULL.
+query IIB
+SELECT co.id, co.k, co.k IN (SELECT ci.k FROM ci WHERE ci.k = co.k GROUP BY
ROLLUP(ci.k)) AS m FROM co ORDER BY k, id;
+----
+1 1 true
+2 1 true
+NULL 1 true
+2 2 true
+NULL 3 false
+9 9 false
+5 NULL false
Review Comment:
**Result changed here, and the new value is the wrong one. Please read this
one before the others.**
The three rows for `k = 3`, `k = 9` and `k = NULL` were NULL in the previous
commit, and they are FALSE again now. FALSE is what `main` gives. DuckDB 1.5.2
gives NULL:
```
D SELECT co.id, co.k, co.k IN (SELECT ci.k FROM ci WHERE ci.k = co.k GROUP
BY ROLLUP(ci.k)) AS m FROM co ORDER BY k, id;
┌───────┬───────┬───────┐
│ id │ k │ m │
│ ... │ ... │ ... │
│ NULL │ 3 │ NULL │
│ 9 │ 9 │ NULL │
│ 5 │ NULL │ NULL │
└───────┴───────┴───────┘
```
The NULL came from a guard that this commit removes on purpose. The guard
cleared a flag for an outer join, a union or a grouping set, which made this
one query correct without fixing its cause. The cause is in the pull up: it
moves the correlated filter above the aggregate, and the grand-total row of
`ROLLUP` is then one row for the whole table instead of one row for each outer
row. The same plan gives a wrong `EXISTS` for the same tables, which the guard
never touched:
```
# both main and this branch
SELECT co.id, co.k, EXISTS (SELECT 1 FROM ci WHERE ci.k = co.k GROUP BY
ROLLUP(ci.k)) AS e FROM co;
-- gives false for k = 3, 9 and NULL; DuckDB gives true for every row
```
So the guard was not a fix for the bug, only for one of its symptoms, and
keeping it means this rule carries a list of plan nodes that can put a NULL
back into a column. I filed the cause as
https://github.com/apache/datafusion/issues/25519 and left this query at the
`main` result until that is fixed.
Tell me if you would rather keep the guard until #25519 lands. The change is
one match arm, and the tests stay green either way.
##########
datafusion/optimizer/src/decorrelate_predicate_subquery.rs:
##########
@@ -222,39 +241,89 @@ fn rewrite_inner_subqueries(
Ok((cur_input, expr_without_subqueries.data))
}
+/// Rewrites an `IN` subquery that gives a value, for example in a SELECT list.
+/// The value follows SQL three-valued logic: TRUE for a match, FALSE for a
miss
+/// and NULL (UNKNOWN) when the answer depends on a NULL.
+///
+/// There are two paths:
+///
+/// * One mark join. When the mark column is already exact under three-valued
+/// logic (see [`MarkJoin::three_valued_exact`]), the mark column is the
+/// answer and this single join is the full rewrite. This is the usual case.
+/// * Three mark joins. A residual non-equality filter stays on the join in the
+/// other case. The mark column then only tells TRUE from not-TRUE, so the
+/// UNKNOWN cases must be materialized: one more join tells if the subquery
+/// gives a NULL, and one more tells if the subquery gives any row. A `CASE`
+/// expression puts the three marks together. The two extra joins have no
+/// join predicate, so use them only when the first path cannot apply.
fn in_subquery_value_mark_join(
left: &LogicalPlan,
subquery: &LogicalPlan,
expr: Expr,
negated: bool,
alias: &Arc<AliasGenerator>,
) -> Result<Option<(LogicalPlan, Expr)>> {
+ // An outer reference in the value belongs to an enclosing subquery. It
+ // cannot be resolved in the joins this builds, and `build_join` would read
+ // it as a constant, so leave the predicate for the enclosing rule.
+ if expr.contains_outer() {
+ return Ok(None);
+ }
+
let output_expr = subquery
.head_output_expr()?
.map_or(plan_err!("single expression required."), Ok)?;
let in_predicate = Expr::eq(expr.clone(), output_expr.clone());
- let Some((matched_plan, matched)) =
- mark_join(left, subquery, Some(&in_predicate), false, alias)?
+ let Some(MarkJoin {
+ plan: matched_plan,
+ mark: matched,
+ three_valued_exact,
+ }) = mark_join(left, subquery, Some(&in_predicate), false, alias)?
else {
return Ok(None);
};
- // SQL IN needs three facts per outer row to distinguish FALSE from
UNKNOWN.
- let null_subquery = LogicalPlanBuilder::from(subquery.clone())
- .filter(output_expr.is_null())?
- .build()?;
- let Some((null_plan, subquery_has_null)) =
- mark_join(&matched_plan, &null_subquery, None, false, alias)?
- else {
- return Ok(None);
- };
- let Some((final_plan, subquery_non_empty)) =
- mark_join(&null_plan, subquery, None, false, alias)?
+ // The mark column is the full answer when it is exact. Negation does not
+ // change that, because NOT UNKNOWN is UNKNOWN.
+ if three_valued_exact {
+ return Ok(Some((
+ matched_plan,
+ if negated { not(matched) } else { matched },
+ )));
+ }
+
+ // The value is not matched here, so the answer is UNKNOWN when the
+ // subquery gives a NULL, and also when the value is NULL and the subquery
+ // gives any row at all. One mark join answers both: it keeps a subquery
row
+ // in the scope of the outer row when the subquery value is NULL, or when
+ // the outer value is NULL and the row is in scope at all.
+ //
+ // For an outer value that is not NULL the mark reads "the subquery gives a
+ // NULL". For an outer value that is NULL it reads "the subquery gives a
+ // row", which is the weaker fact that this case needs, and which the first
+ // reading implies. `IS NULL` is two-valued on both sides, so neither test
+ // adds an UNKNOWN of its own.
+ let unknown_alias = alias.next("__correlated_sq");
Review Comment:
Agreed. I removed 50d6e4a3a from this PR. I also removed 1a9d84229a, which
only fixed an error message that the merged join caused. The fallback for a
residual filter is now the same plan as on main.
After the rebase onto #25339, a NOT IN filter with a residual filter no
longer uses the fallback. It becomes one null-aware anti join, because that
join now applies the residual filter. The `nai_res_og` plan that #25339 pins
stays as it is, and the `joins.slt` plan is the same as on main again.
--
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]
