diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index b79decdd65e02..24d1a00e6bdae 100644 --- a/misc/python/materialize/mzcompose/__init__.py +++ b/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", diff --git a/src/adapter/src/optimize.rs b/src/adapter/src/optimize.rs index 67d272d98073a..3acfd31ded0bb 100644 --- a/src/adapter/src/optimize.rs +++ b/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, } } } diff --git a/src/repr/src/optimize.rs b/src/repr/src/optimize.rs index f919eb0027762..d0a9d36ec7d0f 100644 --- a/src/repr/src/optimize.rs +++ b/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, diff --git a/src/sql/src/plan/lowering.rs b/src/sql/src/plan/lowering.rs index a13ab2f4c3e08..b868f4e9de575 100644 --- a/src/sql/src/plan/lowering.rs +++ b/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, diff --git a/src/sql/src/plan/statement/ddl.rs b/src/sql/src/plan/statement/ddl.rs index 82e9c640e122f..8340886ffb4ef 100644 --- a/src/sql/src/plan/statement/ddl.rs +++ b/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; diff --git a/src/sql/src/plan/statement/dml.rs b/src/sql/src/plan/statement/dml.rs index f86aea68c6bd3..b12e77cc7b860 100644 --- a/src/sql/src/plan/statement/dml.rs +++ b/src/sql/src/plan/statement/dml.rs @@ -636,6 +636,7 @@ impl TryFrom 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(), }, diff --git a/src/sql/src/plan/transform_hir.rs b/src/sql/src/plan/transform_hir.rs index eb64af2f017f2..24f19f688da8d 100644 --- a/src/sql/src/plan/transform_hir.rs +++ b/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() == (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` 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 { + // 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 = 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 = 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::>>()?; + } + _ => 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 { + // 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 diff --git a/src/sql/src/session/vars/definitions.rs b/src/sql/src/session/vars/definitions.rs index c88cb1b1dd1f5..a5a8c286c7a58 100644 --- a/src/sql/src/session/vars/definitions.rs +++ b/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); diff --git a/test/launchdarkly-flag-consistency/mzcompose.py b/test/launchdarkly-flag-consistency/mzcompose.py index dc7aefb7cf8b4..50f2e8d6351eb 100644 --- a/test/launchdarkly-flag-consistency/mzcompose.py +++ b/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 diff --git a/test/sqllogictest/subquery.slt b/test/sqllogictest/subquery.slt index 8174845041336..35a57451ff00d 100644 --- a/test/sqllogictest/subquery.slt +++ b/test/sqllogictest/subquery.slt @@ -763,3 +763,179 @@ Return Target cluster: quickstart EOF + +# FROM-less correlated existence subqueries collapse to a plain Filter, closing +# database-issues#2613 (semijoin) and database-issues#2969 (antijoin). Gated by +# the `enable_simplify_from_less_existence` feature flag (default off in +# production, on in CI/test). The `NOT EXISTS` rewrite must be null-safe +# (`IS NOT TRUE`, not `NOT`), so a row whose predicate evaluates to NULL is +# preserved. +statement ok +CREATE TABLE fl (f1 int) + +statement ok +INSERT INTO fl VALUES (0), (1), (2), (NULL) + +# With the flag OFF, the rows are correct but the plans retain the semijoin +# (database-issues#2613) and antijoin (database-issues#2969) structure. +simple conn=mz_system,user=mz_system +ALTER SYSTEM SET enable_simplify_from_less_existence TO false; +---- +COMPLETE 0 + +query I rowsort +SELECT f1 FROM fl WHERE NOT EXISTS (SELECT 1 WHERE f1 > 0) +---- +0 +NULL + +query I rowsort +SELECT f1 FROM fl WHERE 1 IN (SELECT 1 WHERE f1 > 1) +---- +2 + +query T multiline +EXPLAIN OPTIMIZED PLAN WITH(humanized expressions) AS VERBOSE TEXT FOR +SELECT f1 FROM fl WHERE NOT EXISTS (SELECT 1 WHERE f1 > 0) +---- +Explained Query: + With + cte l0 = + Distinct project=[#0{f1}] + ReadStorage materialize.public.fl + Return + Project (#0{f1}) + Join on=(#0{f1} = #1{f1}) type=differential + ArrangeBy keys=[[#0{f1}]] + ReadStorage materialize.public.fl + ArrangeBy keys=[[#0{f1}]] + Union + Negate + Filter (#0{f1} > 0) + Get l0 + Get l0 + +Source materialize.public.fl + +Target cluster: quickstart + +EOF + +query T multiline +EXPLAIN OPTIMIZED PLAN WITH(humanized expressions) AS VERBOSE TEXT FOR +SELECT f1 FROM fl WHERE 1 IN (SELECT 1 WHERE f1 > 1) +---- +Explained Query: + Project (#0{f1}) + Join on=(#0{f1} = #1{f1}) type=differential + ArrangeBy keys=[[#0{f1}]] + ReadStorage materialize.public.fl + ArrangeBy keys=[[#0{f1}]] + Distinct project=[#0{f1}] + Filter (#0{f1} > 1) + ReadStorage materialize.public.fl + +Source materialize.public.fl + +Target cluster: quickstart + +EOF + +# With the flag ON (the CI/test default), the same queries collapse to a plain +# Filter and the rows are unchanged. +simple conn=mz_system,user=mz_system +ALTER SYSTEM SET enable_simplify_from_less_existence TO true; +---- +COMPLETE 0 + +# database-issues#2969: NOT EXISTS keeps the NULL row (predicate is NULL) and +# the FALSE row (f1 = 0), matching an empty subquery. +query I rowsort +SELECT f1 FROM fl WHERE NOT EXISTS (SELECT 1 WHERE f1 > 0) +---- +0 +NULL + +# database-issues#2613: 1 IN keeps only the row where the predicate is TRUE. +query I rowsort +SELECT f1 FROM fl WHERE 1 IN (SELECT 1 WHERE f1 > 1) +---- +2 + +# The NOT EXISTS plan is a plain Filter using the null-safe `IS NOT TRUE`, with +# no antijoin (no Union/Negate). +query T multiline +EXPLAIN OPTIMIZED PLAN WITH(humanized expressions) AS VERBOSE TEXT FOR +SELECT f1 FROM fl WHERE NOT EXISTS (SELECT 1 WHERE f1 > 0) +---- +Explained Query: + Filter ((#0{f1} > 0)) IS NOT TRUE + ReadStorage materialize.public.fl + +Source materialize.public.fl + filter=(((#0{f1} > 0)) IS NOT TRUE) + +Target cluster: quickstart + +EOF + +# The 1 IN plan is a plain Filter, with no semijoin. +query T multiline +EXPLAIN OPTIMIZED PLAN WITH(humanized expressions) AS VERBOSE TEXT FOR +SELECT f1 FROM fl WHERE 1 IN (SELECT 1 WHERE f1 > 1) +---- +Explained Query: + Filter ((#0{f1} > 1)) IS TRUE + ReadStorage materialize.public.fl + +Source materialize.public.fl + filter=(((#0{f1} > 1)) IS TRUE) + +Target cluster: quickstart + +EOF + +# A window function in a FROM-less quantified subquery must not be inlined into +# the outer Filter by the collapse. `x IN/= ANY/= ALL (SELECT )` normalizes +# to an EXISTS with `` in a Filter predicate, but a window function run over +# the whole outer relation (row_number() = 1, 2, 3, ...) differs from its +# single-row subquery value (always 1). All three `1` rows satisfy `f1 = 1`, so +# the correct answer keeps all three. The buggy inlined form keeps only the one +# row that row_number() numbers 1. +statement ok +CREATE TABLE flw (f1 int) + +statement ok +INSERT INTO flw VALUES (1), (1), (1), (NULL) + +simple conn=mz_system,user=mz_system +ALTER SYSTEM SET enable_simplify_from_less_existence TO true; +---- +COMPLETE 0 + +query I rowsort +SELECT f1 FROM flw WHERE f1 IN (SELECT row_number() OVER ()) +---- +1 +1 +1 + +query I rowsort +SELECT f1 FROM flw WHERE f1 = ANY (SELECT row_number() OVER ()) +---- +1 +1 +1 + +query I rowsort +SELECT f1 FROM flw WHERE f1 = ALL (SELECT row_number() OVER ()) +---- +1 +1 +1 + +# Restore the default so later files see production behavior. +simple conn=mz_system,user=mz_system +ALTER SYSTEM SET enable_simplify_from_less_existence TO false; +---- +COMPLETE 0