perf: use one null-aware mark join for hashable IN subqueries in projections

benchmarks/sql_benchmarks/README.md

@@ -48,7 +48,7 @@ in the community:
 | `predicate_eval`      | Conjunctive (AND) filter-evaluation micro-benchmarks; each subgroup is a different predicate pattern, to test how an adaptive predicate-ordering system behaves across them ([#11262](https://github.com/apache/datafusion/issues/11262)). Subgroups (`--subgroup`): `costsel`, `cost`, `selectivity`, `cardinality`, `width`, `scale`, `neutral`, `correlation`, `drift`, `nulls`. The suite sets no engine config of its own, so by default it measures DataFusion's built-in left-deep `AND` short-circuit; point it at a system under test by exporting that system's own DataFusion setting (the harness builds its `SessionConfig` with `SessionConfig::from_env`). Every query is a `count(*)`, and the counts are checked in under `predicate_eval/results/`, so `--result-mode validate` also checks that a reordering under test still returns the same rows; the checked-in counts were persisted at the suite defaults (`PRED_ROWS=1000000`, `PRED_FILL=30`), so validation assumes those (the `scale` and `width` subgroups pin their own values per query and validate at any setting). |
 | `parquet_row_filter_skip` | Micro-benchmark for the per-row-group fully-matched RowFilter skip on Parquet scans ([#23696](https://github.com/apache/datafusion/issues/23696)). Subgroups (`--subgroup`): `skip` (clustered key, most row groups fully matched by statistics so the per-row filter is skipped), `control` (scrambled key, no row group is ever fully matched). Size the data with `PRED_ROWS` and the row-group size with `RG_SIZE`. |
 | `spill_views` | Sort and `GROUP BY` queries that spill `StringView` and `BinaryView` columns read from Parquet ([#23564](https://github.com/apache/datafusion/issues/23564)). Each query sets its own memory limit and `target_partitions = 4`, so it spills in any environment. Subgroups (`--subgroup`): `repeated` (sort with a low-cardinality payload that shares one dictionary buffer, where spill compaction can remove data), `distinct` (sort and `GROUP BY` on all-distinct strings, where compaction cannot remove data). Set the memory limit of each subgroup with `SPILL_VIEWS_LIMIT_REPEATED` (default `40M`) and `SPILL_VIEWS_LIMIT_DISTINCT` (default `96M`), for example to let both sides of a comparison finish when one of them runs out of memory at the default. The queries must still spill, so check a raised limit with `EXPLAIN ANALYZE` and keep it well below the working set: `q05` stops spilling at about `192M`. Give a whole number of K, M or G units, because the run asserts the limit that the memory pool reports back. |
-| `projection_subquery` | `IN`, `NOT IN` and `EXISTS` subqueries that sit in the `SELECT` list instead of a filter ([#25341](https://github.com/apache/datafusion/issues/25341)). Every query projects the boolean subquery result and aggregates it, so the measured cost is the decorrelation plan and not the size of the output. `q07` correlates on `<` instead of `=`, so it keeps the nested-loop plan and is the control that must not change. Size the two inline tables with `PSQ_ROWS` (default `30000`). The counts are checked in under `projection_subquery/results/`, so `--result-mode validate` also proves that a change to the decorrelation still returns the same rows; the checked-in counts were persisted at the default `PSQ_ROWS`, so validation assumes that value. |
+| `projection_subquery` | `IN`, `NOT IN` and `EXISTS` subqueries that sit in the `SELECT` list instead of a filter ([#25341](https://github.com/apache/datafusion/issues/25341)). Every query projects the boolean subquery result and aggregates it, so the measured cost is the decorrelation plan and not the size of the output. `q07` correlates on `<` instead of `=`, so the correlation stays a residual filter of the mark join. Size the two inline tables with `PSQ_ROWS` (default `30000`). The counts are checked in under `projection_subquery/results/`, so `--result-mode validate` also proves that a change to the decorrelation still returns the same rows; the checked-in counts were persisted at the default `PSQ_ROWS`, so validation assumes that value. |
 
 # Running Benchmarks
 

benchmarks/sql_benchmarks/projection_subquery/projection_subquery.suite

@@ -1,4 +1,4 @@
-description = "IN / NOT IN / EXISTS subqueries that appear in the SELECT list instead of a filter (apache/datafusion#25341). Each query projects the boolean result of a subquery and aggregates it, so the measured work is the decorrelation plan and not the output size. The decorrelation turns one projected IN into mark joins; when the join predicate is not hashable, or the correlation is not an equality, the plan keeps a nested-loop mark join and the cost grows with outer rows times inner rows. Query 07 correlates on `<` and stays on that path, so it is the control that must not change. The two tables are built inline by the load SQL and are sized with PSQ_ROWS."
+description = "IN / NOT IN / EXISTS subqueries that appear in the SELECT list instead of a filter (apache/datafusion#25341). Each query projects the boolean result of a subquery and aggregates it, so the measured work is the decorrelation plan and not the output size. The decorrelation turns one projected IN into mark joins. Query 07 correlates on `<`, so the correlation stays a residual filter of the mark join. The two tables are built inline by the load SQL and are sized with PSQ_ROWS."
 
 query_pattern = "q{QUERY_ID_PADDED}.benchmark"
 

benchmarks/sql_benchmarks/projection_subquery/queries/q07.sql

@@ -1,6 +1,6 @@
--- Shape: IN correlated on `<` instead of `=`. There is no equality to hash on,
--- so this query keeps the nested-loop mark joins. It is the control: its cost
--- must stay the same when the hashable cases get faster.
+-- Shape: IN correlated on `<` instead of `=`. The `IN` equality is the hash key
+-- and the `<` correlation stays a residual filter of the one null-aware mark
+-- join, which applies it when it decides whether a NULL makes the mark UNKNOWN.
 SELECT
   count(*) FILTER (WHERE m) AS true_count,
   count(*) FILTER (WHERE m IS NULL) AS null_count

datafusion/optimizer/src/decorrelate.rs

@@ -30,7 +30,7 @@ use datafusion_common::{
     Column, DFSchemaRef, HashMap, Result, ScalarValue, assert_or_internal_err, plan_err,
 };
 use datafusion_expr::expr::{Alias, GroupingSet};
-use datafusion_expr::logical_plan::Join;
+use datafusion_expr::logical_plan::{Join, JoinType};
 use datafusion_expr::simplify::SimplifyContext;
 use datafusion_expr::utils::{
     collect_subquery_cols, conjunction, find_join_exprs, split_conjunction,
@@ -76,6 +76,21 @@ pub struct PullUpCorrelatedExpr {
     /// whether we have converted a scalar aggregation into a group aggregation. When unnesting
     /// lateral joins, we need to produce a left outer join in such cases.
     pub pulled_up_scalar_agg: bool,
+    /// Every correlated conjunct that a `Filter` of the subquery applies,
+    /// before `remove_duplicated_filter` drops the ones that the `IN`
+    /// predicate already covers. The outer references are kept, so that a
+    /// subquery column and an outer column with the same qualified name stay
+    /// apart.
+    ///
+    /// `join_filters` holds only the conjuncts that the join still needs.
+    /// This list is what the subquery enforces on its own rows. The caller
+    /// uses it to tell if a join key can be NULL inside the scope of an outer
+    /// row: `x IN (SELECT y FROM .. WHERE y = x)` keeps every NULL `y` out of
+    /// its result, although `join_filters` no longer says so.
+    ///
+    /// The list is cleared when the pull up passes a node that can put a NULL
+    /// back into such a column: an outer join, a union or a grouping set.
+    pub correlated_filters: Vec<Expr>,
 }
 
 impl Default for PullUpCorrelatedExpr {
@@ -97,6 +112,7 @@ impl PullUpCorrelatedExpr {
             collected_count_expr_map: HashMap::new(),
             pull_up_having_expr: None,
             pulled_up_scalar_agg: false,
+            correlated_filters: Vec::new(),
         }
     }
 
@@ -186,6 +202,12 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr {
     }
 
     fn f_up(&mut self, plan: LogicalPlan) -> Result<Transformed<LogicalPlan>> {
+        // A node that null-extends or regroups rows can put a NULL back into
+        // a column that a correlated filter below it rejected, so the filters
+        // recorded so far no longer bound what the subquery returns.
+        if !self.correlated_filters.is_empty() && may_reintroduce_nulls(&plan) {
+            self.correlated_filters.clear();
+        }
         let subquery_schema = plan.schema();
         match &plan {
             LogicalPlan::Filter(plan_filter) => {
@@ -195,6 +217,11 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr {
                         .iter()
                         .filter(|e| e.contains_outer())
                         .all(|&e| can_pullup_over_aggregation(e));
+                for expr in &subquery_filter_exprs {
+                    if expr.contains_outer() && !self.correlated_filters.contains(expr) {
+                        self.correlated_filters.push((*expr).clone());
+                    }
+                }
                 let (mut join_filters, subquery_filters) =
                     find_join_exprs(subquery_filter_exprs)?;
                 if let Some(in_predicate) = &self.in_predicate_opt {
@@ -548,6 +575,27 @@ impl PullUpCorrelatedExpr {
     }
 }
 
+/// Whether `plan` can output a NULL in a column that a filter below it keeps
+/// free of NULLs: an outer join fills the columns of its unmatched side with
+/// NULL, a union adds the rows of its other inputs, and a grouping set fills
+/// the columns it leaves out with NULL, such as the grand total row of `ROLLUP`.
+fn may_reintroduce_nulls(plan: &LogicalPlan) -> bool {
+    match plan {
+        LogicalPlan::Join(join) => {
+            matches!(
+                join.join_type,
+                JoinType::Left | JoinType::Right | JoinType::Full
+            )
+        }
+        LogicalPlan::Union(_) => true,
+        LogicalPlan::Aggregate(aggregate) => aggregate
+            .group_expr
+            .iter()
+            .any(|expr| matches!(expr, Expr::GroupingSet(_))),
+        _ => false,
+    }
+}
+
 /// Whether every input of `join` that holds outer references is a side whose
 /// rows the join preserves, so a correlated filter below it can be pulled above
 /// the join without changing the result.