sql: collapse FROM-less correlated existence subqueries to a filter

misc/python/materialize/mzcompose/__init__.py

@@ -97,6 +97,7 @@ def get_minimal_system_parameters(
         "enable_replacement_materialized_views": "true",
         "enable_cluster_schedule_refresh": "true",
         "enable_s3_tables_region_check": "false",
+        "enable_simplify_from_less_existence": "true",
         "enable_statement_lifecycle_logging": "true",
         "enable_storage_introspection_logs": "true",
         "enable_compute_temporal_bucketing": "true",
@@ -236,6 +237,11 @@ def get_variable_system_parameters(
             "false",
             ["true", "false"],
         ),
+        VariableSystemParameter(
+            "enable_simplify_from_less_existence",
+            "false",
+            ["true", "false"],
+        ),
         VariableSystemParameter(
             "enable_upsert_v2",
             "false",

src/adapter/src/optimize.rs

@@ -347,6 +347,9 @@ impl From<&OptimizerConfig> for mz_sql::plan::HirToMirConfig {
             enable_simplify_quantified_comparisons: config
                 .features
                 .enable_simplify_quantified_comparisons,
+            enable_simplify_from_less_existence: config
+                .features
+                .enable_simplify_from_less_existence,
         }
     }
 }

src/repr/src/optimize.rs

@@ -134,6 +134,8 @@ optimizer_feature_flags!({
     // See the feature flag of the same name.
     enable_simplify_quantified_comparisons: bool,
     // See the feature flag of the same name.
+    enable_simplify_from_less_existence: bool,
+    // See the feature flag of the same name.
     enable_coalesce_case_transform: bool,
     // See the feature flag of the same name.
     enable_will_distinct_propagation: bool,

src/sql/src/plan/lowering.rs

@@ -141,6 +141,7 @@ pub struct Config {
     pub enable_variadic_left_join_lowering: bool,
     pub enable_cast_elimination: bool,
     pub enable_simplify_quantified_comparisons: bool,
+    pub enable_simplify_from_less_existence: bool,
 }
 
 impl Default for Config {
@@ -150,6 +151,7 @@ impl Default for Config {
             enable_variadic_left_join_lowering: false,
             enable_cast_elimination: false,
             enable_simplify_quantified_comparisons: false,
+            enable_simplify_from_less_existence: false,
         }
     }
 }
@@ -161,6 +163,7 @@ impl From<&SystemVars> for Config {
             enable_variadic_left_join_lowering: vars.enable_variadic_left_join_lowering(),
             enable_cast_elimination: vars.enable_cast_elimination(),
             enable_simplify_quantified_comparisons: vars.enable_simplify_quantified_comparisons(),
+            enable_simplify_from_less_existence: vars.enable_simplify_from_less_existence(),
         }
     }
 }
@@ -207,6 +210,9 @@ impl HirRelationExpr {
                     &mut other,
                     context.config.enable_simplify_quantified_comparisons,
                 )?;
+                if context.config.enable_simplify_from_less_existence {
+                    transform_hir::simplify_from_less_existence_subqueries(&mut other)?;
+                }
                 transform_hir::fuse_window_functions(&mut other, &context)?;
                 MirRelationExpr::constant(vec![vec![]], ReprRelationType::new(vec![])).let_in(
                     &mut id_gen,

src/sql/src/plan/statement/ddl.rs

@@ -4950,6 +4950,7 @@ pub fn unplan_create_cluster(
                 enable_cast_elimination: _,
                 enable_case_literal_transform: _,
                 enable_simplify_quantified_comparisons: _,
+                enable_simplify_from_less_existence: _,
                 enable_coalesce_case_transform: _,
                 enable_will_distinct_propagation: _,
             } = optimizer_feature_overrides;

src/sql/src/plan/statement/dml.rs

@@ -636,6 +636,7 @@ impl TryFrom<ExplainPlanOptionExtracted> for ExplainConfig {
                 enable_cast_elimination: Default::default(),
                 enable_case_literal_transform: Default::default(),
                 enable_simplify_quantified_comparisons: Default::default(),
+                enable_simplify_from_less_existence: Default::default(),
                 enable_coalesce_case_transform: Default::default(),
                 enable_will_distinct_propagation: Default::default(),
             },

src/sql/src/plan/transform_hir.rs

@@ -16,13 +16,13 @@ use std::{iter, mem};
 use itertools::Itertools;
 use mz_expr::WindowFrame;
 use mz_expr::func::variadic::RecordCreate;
-use mz_expr::visit::Visit;
+use mz_expr::visit::{Visit, VisitChildren};
 use mz_expr::{ColumnOrder, UnaryFunc, VariadicFunc};
 use mz_ore::stack::RecursionLimitError;
 use mz_repr::{ColumnName, SqlColumnType, SqlRelationType, SqlScalarType};
 
 use crate::plan::hir::{
-    AbstractExpr, AggregateFunc, AggregateWindowExpr, HirRelationExpr, HirScalarExpr,
+    AbstractExpr, AggregateFunc, AggregateWindowExpr, ColumnRef, HirRelationExpr, HirScalarExpr,
     ValueWindowExpr, ValueWindowFunc, WindowExpr,
 };
 use crate::plan::{AggregateExpr, WindowExprType};
@@ -307,6 +307,205 @@ pub fn try_simplify_quantified_comparisons(
     walk_relation(expr, &[], simplify_join_on)
 }
 
+/// Collapses `EXISTS` over a FROM-less subquery into an equivalent scalar
+/// predicate on the outer row, so that decorrelation produces a plain `Filter`
+/// rather than a semijoin (for `EXISTS`) or an antijoin (for `NOT EXISTS`).
+///
+/// A FROM-less subquery is a chain of `Map`, `Project`, and `Filter` nodes over
+/// a single-row `Constant` (the join identity of a query with no `FROM`
+/// clause). Such a subquery yields exactly one row when every `Filter`
+/// predicate is `TRUE` and zero rows otherwise, so
+///
+/// ```text
+/// EXISTS(<from-less subquery with predicates p1, p2, ...>) == (p1 AND p2 AND ...) IS TRUE
+/// ```
+///
+/// evaluated on the outer row. The `IS TRUE` is mandatory for null safety. An
+/// empty subquery (some predicate `FALSE` or `NULL`) must make `EXISTS` return
+/// `FALSE`, which `IS TRUE` reproduces while a bare predicate would leak `NULL`.
+/// `NOT EXISTS` then becomes `NOT ((...) IS TRUE)`, which is `... IS NOT TRUE`
+/// and likewise null-safe.
+///
+/// The rewrite fires only on correlated subqueries, where the predicate
+/// references at least one outer column. This keeps it to the pure existence
+/// check that a genuine anti/semi-join would otherwise be lowered to, and it
+/// avoids changing whether an uncorrelated erroring subquery is evaluated when
+/// the outer relation is empty.
+///
+/// This closes database-issues#2613 (`x IN (SELECT ... WHERE p)`, which
+/// [`try_simplify_quantified_comparisons`] has already turned into an `EXISTS`)
+/// and database-issues#2969 (`NOT EXISTS (SELECT ... WHERE p)`). It must run
+/// after [`try_simplify_quantified_comparisons`].
+pub fn simplify_from_less_existence_subqueries(
+    expr: &mut HirRelationExpr,
+) -> Result<(), RecursionLimitError> {
+    // `try_visit_mut_post` walks every relation node, and because
+    // `VisitChildren<Self>` for `HirRelationExpr` descends into the bodies of
+    // `Exists`/`Select` subqueries, it reaches existence checks at every nesting
+    // level. Post-order guarantees a subquery body is simplified before the
+    // `Exists` that encloses it.
+    expr.try_visit_mut_post(&mut |rel| {
+        rel.try_visit_mut_children(|scalar: &mut HirScalarExpr| {
+            scalar.try_visit_mut_pre(&mut |e| {
+                if let HirScalarExpr::Exists(input, _name) = e {
+                    if let Some(pred) = from_less_existence_predicate(input) {
+                        *e = pred.call_unary(UnaryFunc::IsTrue(mz_expr::func::IsTrue));
+                    }
+                }
+                Ok(())
+            })
+        })
+    })
+}
+
+/// If `sub` is a FROM-less subquery (see
+/// [`simplify_from_less_existence_subqueries`]) whose existence check is
+/// correlated on the outer row, returns the predicate `p1 AND p2 AND ...`
+/// expressed in the outer row's frame. Returns `None` otherwise.
+fn from_less_existence_predicate(sub: &HirRelationExpr) -> Option<HirScalarExpr> {
+    // A FROM-less subquery is a linear Map/Project/Filter chain over a single-row
+    // `Constant`. Both properties of the base are load-bearing for soundness: the
+    // single row is what lets EXISTS reduce to "the predicate holds on that row",
+    // and the constant is what lets its columns be inlined into the lifted
+    // predicate below. A 0-row, multi-row, or non-constant base is a genuine
+    // anti/semi-join and bails at the `_` arm.
+    //
+    // Record the chain top to bottom here; it is replayed bottom to top below.
+    let mut chain = Vec::new();
+    let mut cur = sub;
+    let (row, typ) = loop {
+        match cur {
+            HirRelationExpr::Filter { input, .. }
+            | HirRelationExpr::Map { input, .. }
+            | HirRelationExpr::Project { input, .. } => {
+                chain.push(cur);
+                cur = input.as_ref();
+            }
+            HirRelationExpr::Constant { rows, typ } if rows.len() == 1 => break (&rows[0], typ),
+            _ => return None,
+        }
+    };
+
+    // `env` holds the value of each column of the current relation, expressed in
+    // the subquery's own frame. Because level-0 references are resolved as we go,
+    // env entries only ever contain constants and outer (level >= 1) references.
+    let mut env: Vec<HirScalarExpr> = row
+        .iter()
+        .zip_eq(typ.column_types.iter())
+        .map(|(datum, col_type)| HirScalarExpr::literal(datum, col_type.scalar_type.clone()))
+        .collect();
+
+    // Replay the chain bottom to top so each node sees the `env` built by the nodes
+    // beneath it: `Map` extends `env`, `Filter` reads it, `Project` permutes it.
+    let mut preds: Vec<HirScalarExpr> = Vec::new();
+    for node in chain.iter().rev() {
+        match node {
+            HirRelationExpr::Filter { predicates, .. } => {
+                for predicate in predicates {
+                    preds.push(resolve_local_columns(predicate, &env)?);
+                }
+            }
+            HirRelationExpr::Map { scalars, .. } => {
+                for scalar in scalars {
+                    let resolved = resolve_local_columns(scalar, &env)?;
+                    env.push(resolved);
+                }
+            }
+            HirRelationExpr::Project { outputs, .. } => {
+                env = outputs
+                    .iter()
+                    .map(|i| env.get(*i).cloned())
+                    .collect::<Option<Vec<_>>>()?;
+            }
+            _ => unreachable!("chain only contains Filter, Map, and Project nodes"),
+        }
+    }
+
+    let mut pred = HirScalarExpr::variadic_and(preds);
+
+    // `pred` is built only from predicates that `resolve_local_columns` accepted,
+    // and that rejects any subquery, so `pred` contains no nested subqueries. Every
+    // column reference is therefore in the subquery's own frame at nesting depth 0,
+    // and an outer reference is exactly one with `level > 0`.
+
+    // Require correlation: the predicate must reference an outer column. Without
+    // correlation this is not the existence check a genuine anti/semi-join lowers
+    // to, and firing would risk changing when a constant erroring predicate is
+    // evaluated.
+    let mut correlated = false;
+    pred.visit_post(&mut |e| {
+        if let HirScalarExpr::Column(col, _name) = e {
+            if col.level > 0 {
+                correlated = true;
+            }
+        }
+    });
+    if !correlated {
+        return None;
+    }
+
+    // Lift the predicate out of the subquery: references to the immediately
+    // enclosing (outer) scope move down one level.
+    pred.visit_mut_post(&mut |e| {
+        if let HirScalarExpr::Column(col, _name) = e {
+            if col.level > 0 {
+                col.level -= 1;
+            }
+        }
+    });
+
+    Some(pred)
+}
+
+/// Returns `expr` with every reference to the current scope (a [`ColumnRef`]
+/// with `level == 0`) replaced by its value from `env`. Returns `None` if `expr`
+/// cannot be soundly lifted into the outer scope, or references a column absent
+/// from `env`.
+fn resolve_local_columns(expr: &HirScalarExpr, env: &[HirScalarExpr]) -> Option<HirScalarExpr> {
+    // Every scalar in the FROM-less body is substituted into the outer scope, so
+    // reject any that cannot be evaluated equivalently there. The match is
+    // exhaustive on purpose: a new `HirScalarExpr` variant must be classified here
+    // rather than silently treated as liftable.
+    let mut unliftable = false;
+    expr.visit_post(&mut |e| {
+        let liftable = match e {
+            // Row-local: the value depends only on the row, so it is the same in
+            // the subquery's frame and the outer frame.
+            HirScalarExpr::Column(..)
+            | HirScalarExpr::Parameter(..)
+            | HirScalarExpr::Literal(..)
+            | HirScalarExpr::CallUnmaterializable(..)
+            | HirScalarExpr::CallUnary { .. }
+            | HirScalarExpr::CallBinary { .. }
+            | HirScalarExpr::CallVariadic { .. }
+            | HirScalarExpr::If { .. } => true,
+            // A subquery carries its own nested scopes that this flat substitution
+            // does not handle. A window function over the single-row body (e.g.
+            // `row_number() OVER ()` is always 1) is not the same function over the
+            // multi-row outer relation. Neither may cross the subquery boundary.
+            HirScalarExpr::Exists(..)
+            | HirScalarExpr::Select(..)
+            | HirScalarExpr::Windowing(..) => false,
+        };
+        unliftable |= !liftable;
+    });
+    if unliftable {
+        return None;
+    }
+
+    let mut expr = expr.clone();
+    let mut ok = true;
+    expr.visit_mut_post(&mut |e| {
+        if let HirScalarExpr::Column(ColumnRef { level: 0, column }, _name) = e {
+            match env.get(*column) {
+                Some(value) => *e = value.clone(),
+                None => ok = false,
+            }
+        }
+    });
+    ok.then_some(expr)
+}
+
 /// An empty parameter type map.
 ///
 /// These transformations are expected to run after parameters are bound, so

src/sql/src/session/vars/definitions.rs

@@ -2274,6 +2274,12 @@ feature_flags!(
         default: true,
         enable_for_item_parsing: false,
     },
+    {
+        name: enable_simplify_from_less_existence,
+        desc: "Allow the optimizer to collapse EXISTS/NOT EXISTS over a FROM-less correlated subquery into a plain Filter during HIR-to-MIR lowering.",
+        default: false,
+        enable_for_item_parsing: false,
+    },
     {
         name: enable_coalesce_case_transform,
         desc: "Allow the optimizer to push `COALESCE` into `CASE WHEN`.",
@@ -2316,6 +2322,7 @@ impl From<&super::SystemVars> for OptimizerFeatures {
             enable_cast_elimination: vars.enable_cast_elimination(),
             enable_case_literal_transform: vars.enable_case_literal_transform(),
             enable_simplify_quantified_comparisons: vars.enable_simplify_quantified_comparisons(),
+            enable_simplify_from_less_existence: vars.enable_simplify_from_less_existence(),
             enable_coalesce_case_transform: vars.enable_coalesce_case_transform(),
             enable_will_distinct_propagation: vars.enable_will_distinct_propagation(),
         }
@@ -2358,6 +2365,7 @@ mod tests {
             enable_cast_elimination,
             enable_case_literal_transform,
             enable_simplify_quantified_comparisons,
+            enable_simplify_from_less_existence,
             enable_coalesce_case_transform,
             enable_will_distinct_propagation,
         } = false_features;
@@ -2388,6 +2396,7 @@ mod tests {
         set_var!(enable_cast_elimination);
         set_var!(enable_case_literal_transform);
         set_var!(enable_simplify_quantified_comparisons);
+        set_var!(enable_simplify_from_less_existence);
         set_var!(enable_coalesce_case_transform);
         set_var!(enable_will_distinct_propagation);
 

test/launchdarkly-flag-consistency/mzcompose.py

@@ -266,6 +266,7 @@
     enable_replica_targeted_materialized_views
     enable_s3_tables_region_check
     enable_session_timelines
+    enable_simplify_from_less_existence
     enable_simplify_quantified_comparisons
     enable_time_at_time_zone
     enable_unlimited_retain_history