refactor: use the null-aware anti join for every NOT IN key shape

#25339 made the null-aware
`LeftAnti` hash join take more than one key and a residual filter. The two
workarounds for the old executor limits are no longer necessary:

* More than one key: the rule built a null-aware `LeftMark` join and a
  filter on the mark. It now builds the null-aware `LeftAnti` join.
* A residual filter: the rule gave up, and the caller materialized the
  UNKNOWN rows with three mark joins, two of them nested loop joins. It now
  builds the null-aware `LeftAnti` join, which applies the residual filter
  when it looks for UNKNOWN rows. The plan in `joins.slt` is the same as on
  `main` again, and the single anti join that `null_aware_anti_join.slt`
  pins for a residual filter stays.

The executor reads the first key as the `NOT IN` value. If the `IN`
equality does not become that key, the rule still gives up and the caller
materializes the result.

A constant `NOT IN` value is projected as a column for a correlated
subquery too, so the value is the first key and the correlation the second.
This corrects two `null_aware_anti_join.slt` results that were pinned to
wrong answers, and three constant `NOT IN` queries that failed to plan now
run. All the new results agree with DuckDB 1.5.2.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: adriangb

datafusion/optimizer/src/decorrelate_predicate_subquery.rs

@@ -488,12 +488,12 @@ struct MarkJoin {
 
 /// The join keys of the join that replaces an `IN` or `NOT IN` predicate.
 struct JoinKeys {
-    /// The equalities that the hash join can use as keys. The `IN` predicate
-    /// is the first one when its value holds a column; the others are the
-    /// correlation.
-    equijoin_keys: Vec<(Expr, Expr)>,
     /// The part of the join filter that the split could not turn into keys.
     residual_filter: Option<Expr>,
+    /// True if the `IN` equality is the first equality that the hash join can
+    /// use as a key, which is where a null-aware hash join reads the `IN`
+    /// value. The other keys are the correlation.
+    value_is_first_key: bool,
     /// True if the `IN` value or the subquery column it is compared with can
     /// be NULL inside the scope of an outer row, see
     /// [`key_may_be_null_in_scope`]. Only then can `IN` be UNKNOWN, so only
@@ -519,9 +519,12 @@ impl JoinKeys {
             left_schema,
             right_schema,
         )?;
+        let value_is_first_key = equijoin_keys.first().is_some_and(|(value, column)| {
+            value == &in_value.value && column == &in_value.subquery_column
+        });
         Ok(Self {
-            equijoin_keys,
             residual_filter,
+            value_is_first_key,
             value_may_be_null: in_value.may_be_null_in_scope(
                 left_schema,
                 right_schema,
@@ -764,8 +767,6 @@ fn build_join(
         }
         other => other,
     };
-    // The columns of the outer plan, which an anti join keeps as they are.
-    let outer_columns = left.schema().columns();
     let left = projected_left.as_ref().unwrap_or(left);
 
     let join_filter = match (&in_value, join_filter_opt) {
@@ -799,32 +800,19 @@ fn build_join(
     };
 
     // A `NOT IN` in a filter builds a `LeftAnti` join, and needs null-aware
-    // semantics when the value can be NULL in scope. The null-aware `LeftAnti`
-    // executor takes one key only (see `NullAwareMode::try_new`), and no hash
-    // join can mark the UNKNOWN rows of a residual filter
-    // (https://github.com/apache/datafusion/issues/25336). So:
-    //
-    // * A residual filter: give up here. The caller then materializes the
-    //   UNKNOWN rows with more joins, see `in_subquery_value_mark_join`.
-    // * More than one key: the null-aware `LeftMark` executor takes any number
-    //   of keys, the others being the scope of the outer row. Build that join
-    //   instead and keep the rows whose mark is FALSE, which is `NOT IN` under
-    //   three-valued logic.
-    // * One key: the null-aware `LeftAnti` join below.
-    let anti_join_as_mark = match &join_keys {
-        Some(keys) if join_type == JoinType::LeftAnti && keys.value_may_be_null => {
-            if keys.residual_filter.is_some() {
-                return Ok(None);
-            }
-            keys.equijoin_keys.len() > 1
-        }
-        _ => false,
-    };
-    let join_type = if anti_join_as_mark {
-        JoinType::LeftMark
-    } else {
-        join_type
-    };
+    // semantics when the value can be NULL in scope. The null-aware hash join
+    // reads its first key as the `NOT IN` value and the other keys as the
+    // scope of the outer row (see `HashJoinExec::null_aware`). If the `IN`
+    // equality did not become that first key, give up here. The caller then
+    // materializes the UNKNOWN rows with more joins, see
+    // `in_subquery_value_mark_join`.
+    if let Some(keys) = &join_keys
+        && join_type == JoinType::LeftAnti
+        && keys.value_may_be_null
+        && !keys.value_is_first_key
+    {
+        return Ok(None);
+    }
 
     if matches!(join_type, JoinType::LeftMark | JoinType::RightMark) {
         let right_schema = sub_query_alias.schema();
@@ -884,18 +872,6 @@ fn build_join(
             )?
             .build()?;
 
-        // `NOT IN` keeps the rows whose mark is FALSE. `NOT mark` is TRUE for
-        // those rows only, and the projection removes the mark column again.
-        let new_plan = if anti_join_as_mark {
-            let mark = Expr::Column(Column::new(Some(alias.to_string()), "mark"));
-            LogicalPlanBuilder::from(new_plan)
-                .filter(not(mark))?
-                .project(outer_columns.into_iter().map(Expr::from))?
-                .build()?
-        } else {
-            new_plan
-        };
-
         debug!(
             "predicate subquery optimized:\n{}",
             new_plan.display_indent()
@@ -909,8 +885,7 @@ fn build_join(
 
     // Null-aware semantics are only needed for a `NOT IN` anti join, which
     // follows three-valued logic. `NOT EXISTS` and `IN` are two-valued, and
-    // `join_keys` is `None` for them. The join here has one key and no
-    // residual filter: the other shapes were handled above.
+    // `join_keys` is `None` for them.
     let null_aware = join_keys
         .as_ref()
         .is_some_and(|keys| keys.value_may_be_null);
@@ -1895,11 +1870,11 @@ mod tests {
         )
     }
 
-    /// The same rewrite must not fire for a correlated subquery: the
-    /// correlation predicate is a second equi-join key, and null-aware hash
-    /// joins accept only one.
+    /// The same rewrite fires for a correlated subquery. The projected value
+    /// is the first equi-join key, which is where the null-aware hash join
+    /// reads the `NOT IN` value, and the correlation is the second.
     #[test]
-    fn constant_not_in_correlated_subquery_becomes_a_mark_join() -> Result<()> {
+    fn constant_not_in_correlated_subquery() -> Result<()> {
         let outer_scan = nullable_scalar_mark_scan("outer_t")?;
         let inner_scan = nullable_scalar_mark_scan("inner_t")?;
 
@@ -1920,14 +1895,12 @@ mod tests {
             plan,
             @"
         Projection: outer_t.id, outer_t.grp [id:Int32;N, grp:Int32;N]
-          Filter: NOT __correlated_sq_1.mark [id:Int32;N, grp:Int32;N, __correlated_sq_1_value:Int32, mark:Boolean;N]
-            LeftMark Join:  Filter: __correlated_sq_1_value = __correlated_sq_1.id AND outer_t.grp = __correlated_sq_1.grp null_aware [id:Int32;N, grp:Int32;N, __correlated_sq_1_value:Int32, mark:Boolean;N]
-              Projection: outer_t.id, outer_t.grp, Int32(3) AS __correlated_sq_1_value [id:Int32;N, grp:Int32;N, __correlated_sq_1_value:Int32]
-                TableScan: outer_t [id:Int32;N, grp:Int32;N]
-              Projection: __correlated_sq_1.id, __correlated_sq_1.grp [id:Int32;N, grp:Int32;N]
-                SubqueryAlias: __correlated_sq_1 [id:Int32;N, grp:Int32;N]
-                  Projection: inner_t.id, inner_t.grp [id:Int32;N, grp:Int32;N]
-                    TableScan: inner_t [id:Int32;N, grp:Int32;N]
+          LeftAnti Join:  Filter: __correlated_sq_1_value = __correlated_sq_1.id AND outer_t.grp = __correlated_sq_1.grp null_aware [id:Int32;N, grp:Int32;N, __correlated_sq_1_value:Int32]
+            Projection: outer_t.id, outer_t.grp, Int32(3) AS __correlated_sq_1_value [id:Int32;N, grp:Int32;N, __correlated_sq_1_value:Int32]
+              TableScan: outer_t [id:Int32;N, grp:Int32;N]
+            SubqueryAlias: __correlated_sq_1 [id:Int32;N, grp:Int32;N]
+              Projection: inner_t.id, inner_t.grp [id:Int32;N, grp:Int32;N]
+                TableScan: inner_t [id:Int32;N, grp:Int32;N]
         "
         )
     }

datafusion/sqllogictest/test_files/joins.slt

@@ -1988,22 +1988,11 @@ 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)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)
-06)----------TableScan: join_t1 projection=[t1_id, t1_name, t1_int]
-07)----------SubqueryAlias: __correlated_sq_2
-08)------------Projection: CAST(join_t2.t2_id AS Int64) + Int64(1)
-09)--------------TableScan: join_t2 projection=[t2_id]
-10)--------SubqueryAlias: __correlated_sq_3
-11)----------Projection: CAST(join_t2.t2_id AS Int64) + Int64(1)
-12)------------Filter: CAST(join_t2.t2_id AS Int64) + Int64(1) IS NULL
-13)--------------TableScan: join_t2 projection=[t2_id]
-14)------SubqueryAlias: __correlated_sq_4
-15)--------Projection: CAST(join_t2.t2_id AS Int64) + Int64(1)
-16)----------TableScan: join_t2 projection=[t2_id]
+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]
 
 # In subquery to join with outer filter
 

datafusion/sqllogictest/test_files/null_aware_anti_join.slt

@@ -655,11 +655,15 @@ CREATE TABLE naconst_corr_t2(id INT, g INT) AS VALUES (1, 1), (NULL, 2);
 # A non-equality correlation stays in the join filter. Row (1, 1) sees the NULL
 # of (NULL, 2), so its `NOT IN` is UNKNOWN; row (2, 2) sees an empty subquery,
 # so its `NOT IN` is TRUE. Expected results verified with DuckDB.
-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) ORDER BY id;
+----
+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 NOT (3 IN (SELECT id FROM naconst_corr_t2 WHERE naconst_corr_t2.g > naconst_corr_t1.g)) ORDER BY id;
+----
+2
 
 statement ok
 DROP TABLE naconst_corr_t1;
@@ -683,17 +687,17 @@ query I
 SELECT z FROM naconst_corr_t3 WHERE 1 NOT IN (SELECT naconst_corr_t4.id FROM naconst_corr_t4 WHERE naconst_corr_t4.z = naconst_corr_t3.z) ORDER BY z;
 ----
 7
-10
+NULL
 
-# Pinned to today's behaviour, which is incorrect. See
-# https://github.com/apache/datafusion/issues/25336 -- the fix flips this.
 query I
 SELECT z FROM naconst_corr_t3 WHERE 2 NOT IN (SELECT naconst_corr_t4.id FROM naconst_corr_t4 WHERE naconst_corr_t4.z = naconst_corr_t3.z) ORDER BY z;
 ----
-10
+NULL
 
-query error DataFusion error: Error during planning: null_aware LeftAnti join requires equi\-join keys, but the join has none
+query I
 SELECT z FROM naconst_corr_t3 WHERE 1 NOT IN (SELECT naconst_corr_t4.id FROM naconst_corr_t4 WHERE naconst_corr_t4.z > naconst_corr_t3.z) ORDER BY z;
+----
+NULL
 
 # The projected value column must be the FIRST equi-join key: a null-aware hash
 # join reads `on[0]` as the `NOT IN` value key and `on[1..]` as correlation
@@ -702,16 +706,19 @@ query TT
 EXPLAIN SELECT z FROM naconst_corr_t3 WHERE 1 NOT IN (SELECT naconst_corr_t4.id FROM naconst_corr_t4 WHERE naconst_corr_t4.z = naconst_corr_t3.z);
 ----
 logical_plan
-01)LeftAnti Join: naconst_corr_t3.z = __correlated_sq_1.z Filter: Int64(1) = __correlated_sq_1.naconst_corr_t4.id null_aware
-02)--TableScan: naconst_corr_t3 projection=[z]
-03)--SubqueryAlias: __correlated_sq_1
-04)----Projection: CAST(naconst_corr_t4.id AS Int64), naconst_corr_t4.z
-05)------TableScan: naconst_corr_t4 projection=[id, z]
+01)Projection: naconst_corr_t3.z
+02)--LeftAnti Join: __correlated_sq_1_value = __correlated_sq_1.naconst_corr_t4.id, naconst_corr_t3.z = __correlated_sq_1.z null_aware
+03)----Projection: naconst_corr_t3.z, Int64(1) AS __correlated_sq_1_value
+04)------TableScan: naconst_corr_t3 projection=[z]
+05)----SubqueryAlias: __correlated_sq_1
+06)------Projection: CAST(naconst_corr_t4.id AS Int64), naconst_corr_t4.z
+07)--------TableScan: naconst_corr_t4 projection=[id, z]
 physical_plan
-01)HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(z@0, z@1)], filter=1 = naconst_corr_t4.id@0, null_aware
-02)--DataSourceExec: partitions=1, partition_sizes=[1]
-03)--ProjectionExec: expr=[CAST(id@0 AS Int64) as naconst_corr_t4.id, z@1 as z]
-04)----DataSourceExec: partitions=1, partition_sizes=[1]
+01)HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(__correlated_sq_1_value@1, naconst_corr_t4.id@0), (z@0, z@1)], projection=[z@0], null_aware
+02)--ProjectionExec: expr=[z@0 as z, 1 as __correlated_sq_1_value]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+04)--ProjectionExec: expr=[CAST(id@0 AS Int64) as naconst_corr_t4.id, z@1 as z]
+05)----DataSourceExec: partitions=1, partition_sizes=[1]
 
 statement ok
 DROP TABLE naconst_corr_t3;

datafusion/sqllogictest/test_files/subquery_projection.slt

@@ -369,9 +369,8 @@ 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.
+# value and the correlation. A function key over non-nullable columns that
+# cannot be NULL 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');
 
@@ -385,9 +384,9 @@ SELECT * FROM t1 WHERE upper(t1.s) NOT IN (SELECT t2.s FROM t2 WHERE t2.k = t1.k
 
 # `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
+# first and the correlation second, and is null-aware because the value
+# expression can be NULL. `main` tests the columns of the key instead, which
+# are not nullable, and keeps the `k = 1` row. That 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);
@@ -398,36 +397,29 @@ 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]
+01)LeftAnti Join: nullif(CAST(t1.k AS Int64), Int64(1)) = __correlated_sq_1.t2.k + Int64(10), t1.k = __correlated_sq_1.k null_aware
+02)--TableScan: t1 projection=[k]
+03)--SubqueryAlias: __correlated_sq_1
+04)----Projection: CAST(t2.k AS Int64) + Int64(10), t2.k
+05)------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]
+01)HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(nullif(t1.k,Int64(1))@1, t2.k + Int64(10)@0), (k@0, k@1)], projection=[k@0], null_aware
+02)--ProjectionExec: expr=[k@0 as k, nullif(CAST(k@0 AS Int64), 1) as nullif(t1.k,Int64(1))]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+04)--ProjectionExec: expr=[CAST(k@0 AS Int64) + 10 as t2.k + Int64(10), k@0 as k]
+05)----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.
+# A non-equality correlation leaves one key and a residual join filter. The
+# null-aware `LeftAnti` hash join applies the residual filter when it looks
+# for UNKNOWN rows. 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);
 
@@ -453,30 +445,17 @@ EXPLAIN SELECT k FROM ra WHERE NULLIF(ra.k, 1) NOT IN (SELECT rb.k FROM rb WHERE
 ----
 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]
+02)--LeftAnti Join: nullif(CAST(ra.k AS Int64), Int64(1)) = __correlated_sq_1.rb.k Filter: __correlated_sq_1.z > ra.z null_aware
+03)----TableScan: ra projection=[k, z]
+04)----SubqueryAlias: __correlated_sq_1
+05)------Projection: CAST(rb.k AS Int64), rb.z
+06)--------TableScan: rb projection=[k, 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]
+01)HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(nullif(ra.k,Int64(1))@2, rb.k@0)], filter=z@1 > z@0, projection=[k@0], null_aware
+02)--ProjectionExec: expr=[k@0 as k, z@1 as z, nullif(CAST(k@0 AS Int64), 1) as nullif(ra.k,Int64(1))]
 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]
+04)--ProjectionExec: expr=[CAST(k@0 AS Int64) as rb.k, z@1 as z]
+05)----DataSourceExec: partitions=1, partition_sizes=[1]
 
 statement ok
 DROP TABLE ra;