diff --git a/doc/user/data/metrics.yml b/doc/user/data/metrics.yml index 2d7550548da17..8f449a0f4ba65 100644 --- a/doc/user/data/metrics.yml +++ b/doc/user/data/metrics.yml @@ -1135,6 +1135,12 @@ metrics: - replica_id source: src/cluster-client/src/metrics.rs visibility: internal +- name: mz_delete_rows_selection_total + help: "The number of `DELETE ... ROWS` statements by the selection they read: `rows` when the rows are proven contained in the table, `intersection` otherwise." + labels: + - selection + source: src/adapter/src/metrics.rs + visibility: internal - name: mz_determine_timestamp help: The total number of calls to determine_timestamp. labels: diff --git a/src/adapter/src/client.rs b/src/adapter/src/client.rs index f2df00187ced0..6ff8513aecf12 100644 --- a/src/adapter/src/client.rs +++ b/src/adapter/src/client.rs @@ -1902,6 +1902,7 @@ impl SessionClient { assignments: BTreeMap::new(), kind: MutationKind::Insert, returning: insert_plan.returning, + delete_rows: None, } } _ => { diff --git a/src/adapter/src/coord/hydration_history.rs b/src/adapter/src/coord/hydration_history.rs index 6caf7c1b88596..44d13da21ae8b 100644 --- a/src/adapter/src/coord/hydration_history.rs +++ b/src/adapter/src/coord/hydration_history.rs @@ -827,6 +827,7 @@ fn plan_mutation( assignments: BTreeMap::new(), kind, returning: Vec::new(), + delete_rows: None, }) } diff --git a/src/adapter/src/frontend_read_then_write.rs b/src/adapter/src/frontend_read_then_write.rs index b054313d6b6fa..ec74562fc2c2b 100644 --- a/src/adapter/src/frontend_read_then_write.rs +++ b/src/adapter/src/frontend_read_then_write.rs @@ -127,10 +127,13 @@ use mz_cluster_client::ReplicaId; use mz_compute_types::ComputeInstanceId; use mz_expr::Eval; use mz_expr::row::RowCollection; -use mz_expr::{CollectionPlan, Id, LocalId, MirRelationExpr, MirScalarExpr, RowSetFinishing}; +use mz_expr::{ + CollectionPlan, Id, LocalId, MirRelationExpr, MirScalarExpr, OptimizedMirRelationExpr, + RowSetFinishing, +}; use mz_ore::cast::CastFrom; use mz_ore::{soft_assert_or_log, soft_panic_or_log}; -use mz_repr::optimize::OverrideFrom; +use mz_repr::optimize::{OptimizerFeatures, OverrideFrom}; use mz_repr::{CatalogItemId, Diff, GlobalId, RelationDesc, Row, RowArena, Timestamp}; use mz_sql::catalog::CatalogError; use mz_sql::plan::{self, MutationKind, QueryWhen}; @@ -139,6 +142,7 @@ use mz_sql::session::vars::IsolationLevel; use mz_storage_client::client::TableData; use mz_storage_types::sources::Timeline; use mz_timestamp_oracle::TimestampOracle; +use mz_transform::analysis::SubMultiset; use prometheus::Histogram; use timely::progress::Antichain; use tokio::sync::mpsc; @@ -648,6 +652,112 @@ pub(crate) fn contains_mz_now(plan: &plan::ReadThenWritePlan) -> bool { || plan.returning.iter().any(|e| e.contains_temporal()) } +/// Values of the `selection` label of `mz_delete_rows_selection_total`. +const DELETE_ROWS_SELECTION_ROWS: &str = "rows"; +const DELETE_ROWS_SELECTION_INTERSECTION: &str = "intersection"; + +/// Bounds the catalog walk of [`delete_rows_contained`], in nesting depth and +/// in collections analyzed. Exhausting either reports "not proven". +const DELETE_ROWS_MAX_DEPTH: usize = 32; +const DELETE_ROWS_MAX_COLLECTIONS: usize = 1024; + +/// Whether the rows of a `DELETE ... ROWS` are proven to be a sub-multiset of +/// its table at every timestamp. `false` means "not proven". +/// +/// `catalog` must be the snapshot the plan was built against. +fn delete_rows_contained(catalog: &Catalog, delete_rows: &plan::DeleteRows) -> bool { + let config = optimize::OptimizerConfig::from(catalog.system_config()); + let features = config.features.clone(); + // An error here also fails the intersection, which reports it. + let rows = match optimize::view::Optimizer::new(config, None).optimize(delete_rows.rows.clone()) + { + Ok(rows) => rows, + Err(err) => { + tracing::debug!(%err, "cannot optimize DELETE ... ROWS rows"); + return false; + } + }; + let mut prover = ContainmentProver { + catalog, + target: delete_rows.table, + features, + known: BTreeMap::new(), + budget: DELETE_ROWS_MAX_COLLECTIONS, + }; + prover.proves(&rows, 0) +} + +/// Proves containment in `target` across the catalog, by seeding +/// [`SubMultiset`] with the collections an expression reads that are +/// themselves proven contained. +struct ContainmentProver<'a> { + catalog: &'a Catalog, + target: GlobalId, + features: OptimizerFeatures, + /// Memoized answers of [`Self::is_contained`]. + known: BTreeMap, + /// Collections still allowed to be analyzed. + budget: usize, +} + +impl ContainmentProver<'_> { + fn proves(&mut self, expr: &MirRelationExpr, depth: usize) -> bool { + let known = expr + .depends_on() + .into_iter() + .filter(|id| self.is_contained(*id, depth)) + .collect(); + SubMultiset::new(self.target, known).holds(expr, &self.features) + } + + /// Whether the contents of collection `id` at every timestamp are a + /// sub-multiset of the target's contents at that timestamp. + fn is_contained(&mut self, id: GlobalId, depth: usize) -> bool { + if id == self.target { + return true; + } + if let Some(known) = self.known.get(&id) { + return *known; + } + let contained = if depth >= DELETE_ROWS_MAX_DEPTH || self.budget == 0 { + false + } else { + self.budget -= 1; + match self.definition(id) { + Some(expr) => self.proves(&expr, depth + 1), + None => false, + } + }; + self.known.insert(id, contained); + contained + } + + /// The expression collection `id` equals at every timestamp, if there is + /// one. + /// + /// A view is its definition. A materialized view stores its definition at + /// every timestamp below its write frontier, unless a refresh schedule lets + /// the stored contents lag the definition. A replaced materialized view + /// stores an earlier definition at earlier timestamps, so it qualifies + /// only while it has a single version. Every other collection, including + /// tables other than the target, has no definition in terms of the target. + fn definition(&self, id: GlobalId) -> Option> { + let entry = self.catalog.try_get_entry_by_global_id(&id)?; + match entry.item() { + CatalogItem::View(view) => Some(Arc::clone(&view.locally_optimized_expr)), + CatalogItem::MaterializedView(mv) + if mv.refresh_schedule.is_none() + && mv.replacement_target.is_none() + && mv.collections.len() == 1 + && mv.global_id_writes() == id => + { + Some(Arc::clone(&mv.locally_optimized_expr)) + } + _ => None, + } + } +} + /// The timeline whose oracle governs a read-then-write with the given read-side /// [`TimelineContext`]. /// @@ -795,6 +905,38 @@ impl PeekClient { return Err(AdapterError::ReadOnlyTransaction); } + if let Some(delete_rows) = plan.delete_rows.take() { + let target_global_id = catalog.get_entry(&plan.id).latest_global_id(); + let rows_contained = delete_rows.table == target_global_id + && delete_rows_contained(catalog, &delete_rows); + tracing::debug!( + table = %delete_rows.table, + rows_contained, + "DELETE ... ROWS selection" + ); + // Writing `q` in place of `t INTERSECT ALL q` relies on the OCC loop's + // invariants. A write at `T` waits for the selection's frontier to reach + // `T`, so the collections `q` reads are complete through `T - 1`, where + // the proof makes `q` a sub-multiset of `t`. The timestamped write then + // commits to the txns shard, which every table write shares, at exactly + // `T` or not at all, so `t` still holds its `T - 1` contents when the + // retractions apply, whether or not the selection reads `t` itself. The + // blind exit writes at a timestamp it does not choose, and a proven `q` + // never takes it with rows, because it reads `t` directly or through + // collections that follow `t`, whose frontier does not close. + let label = if rows_contained { + plan.selection = delete_rows.rows; + DELETE_ROWS_SELECTION_ROWS + } else { + DELETE_ROWS_SELECTION_INTERSECTION + }; + self.coordinator_client() + .metrics() + .delete_rows_selection + .with_label_values(&[label]) + .inc(); + } + let validation_result = validate_read_then_write( catalog, session, @@ -1209,6 +1351,7 @@ impl PeekClient { assignments, kind, returning: _, + delete_rows: _, } = plan; let expr = selection.clone().lower(catalog.system_config(), None)?; diff --git a/src/adapter/src/metrics.rs b/src/adapter/src/metrics.rs index ef0f66c767bbc..a0dcebb29bef3 100644 --- a/src/adapter/src/metrics.rs +++ b/src/adapter/src/metrics.rs @@ -77,6 +77,7 @@ pub struct Metrics { pub apply_catalog_implications_seconds: Histogram, pub group_commit_catalog_upper_seconds: Histogram, pub occ_retry_count: HistogramVec, + pub delete_rows_selection: IntCounterVec, pub by_cluster: ClusterLabeledMetrics, } @@ -316,6 +317,13 @@ impl Metrics { 0., 1., 2., 3., 5., 10., 25., 50., 100., 200., 300., 500., 750., 1000., ], )), + delete_rows_selection: registry.register(metric!( + name: "mz_delete_rows_selection_total", + help: "The number of `DELETE ... ROWS` statements by the selection they read: \ + `rows` when the rows are proven contained in the table, `intersection` \ + otherwise.", + var_labels: ["selection"], + )), by_cluster: ClusterLabeledMetrics::register_into(registry), } } diff --git a/src/environmentd/tests/read_then_write.rs b/src/environmentd/tests/read_then_write.rs index 46bc9af4fef6e..468f0c133b18a 100644 --- a/src/environmentd/tests/read_then_write.rs +++ b/src/environmentd/tests/read_then_write.rs @@ -2171,3 +2171,215 @@ fn test_serializable_read_sees_own_refresh_mv_write() { "a serializable read must see the session's own write" ); } + +/// How many `DELETE ... ROWS` statements read the given selection so far. +fn delete_rows_selections(server: &test_util::TestServerWithRuntime, selection: &str) -> u64 { + test_util::get_counter_value( + server.metrics_registry(), + "mz_delete_rows_selection_total", + &[("selection", selection)], + ) +} + +/// `DELETE ... ROWS (q)` retracts `q` directly exactly when `q` is proven +/// contained in the table, and otherwise intersects it with the table. The +/// counts are those of the intersection either way. +#[mz_ore::test] +#[allow(clippy::disallowed_methods)] +fn test_delete_rows_selection() { + let server = test_util::TestHarness::default() + .unsafe_mode() + .with_system_parameter_default("enable_delete_rows".to_string(), "true".to_string()) + .with_system_parameter_default("enable_refresh_every_mvs".to_string(), "true".to_string()) + .with_system_parameter_default( + "enable_alter_table_add_column".to_string(), + "true".to_string(), + ) + .start_blocking(); + let mut client = server.connect(postgres::NoTls).unwrap(); + for statement in [ + "CREATE TABLE intents (id INT, done BOOL)", + "CREATE TABLE store (id INT)", + "CREATE TABLE other (id INT, done BOOL)", + "INSERT INTO store VALUES (1)", + "INSERT INTO other VALUES (2, false)", + "CREATE MATERIALIZED VIEW garbage AS SELECT * FROM intents EXCEPT ALL SELECT id, false FROM store", + "CREATE VIEW garbage_view AS SELECT * FROM intents EXCEPT ALL SELECT id, false FROM store", + "CREATE VIEW garbage_indexed AS SELECT * FROM intents EXCEPT ALL SELECT id, false FROM store", + "CREATE INDEX garbage_indexed_idx ON garbage_indexed (id)", + "CREATE VIEW over_garbage AS SELECT * FROM garbage WHERE id > 0", + "CREATE MATERIALIZED VIEW stale WITH (REFRESH EVERY '1000 hours') AS SELECT * FROM intents", + ] { + client.batch_execute(statement).unwrap(); + } + + // Each case resets `intents` to `(1, false)` once and `(2, false)` twice, + // so the garbage is `(2, false)` twice. + let cases = [ + ( + "SELECT * FROM intents EXCEPT ALL SELECT id, false FROM store", + "rows", + 2, + ), + ("SELECT * FROM garbage", "rows", 2), + ("SELECT * FROM garbage_view", "rows", 2), + ("SELECT * FROM garbage_indexed", "rows", 2), + ("SELECT * FROM over_garbage WHERE done = false", "rows", 2), + ( + "SELECT * FROM garbage INTERSECT ALL SELECT * FROM other", + "rows", + 1, + ), + ( + "SELECT * FROM intents UNION ALL SELECT * FROM intents", + "intersection", + 3, + ), + ("SELECT * FROM other", "intersection", 1), + ("SELECT id, NOT done FROM garbage", "intersection", 0), + ("SELECT * FROM stale", "intersection", 0), + ]; + for (query, selection, expected) in cases { + for statement in [ + "DELETE FROM intents", + "INSERT INTO intents VALUES (1, false), (2, false), (2, false)", + ] { + client.batch_execute(statement).unwrap(); + } + let rows_before = delete_rows_selections(&server, "rows"); + let intersection_before = delete_rows_selections(&server, "intersection"); + let deleted = client + .execute(&format!("DELETE FROM intents ROWS ({query})"), &[]) + .unwrap(); + assert_eq!(deleted, expected, "rows deleted by `{query}`"); + let expected_counts = match selection { + "rows" => (1, 0), + _ => (0, 1), + }; + assert_eq!( + ( + delete_rows_selections(&server, "rows") - rows_before, + delete_rows_selections(&server, "intersection") - intersection_before, + ), + expected_counts, + "`{query}` should read the {selection} selection", + ); + let remaining: i64 = client + .query_one("SELECT count(*) FROM intents", &[]) + .unwrap() + .get(0); + assert_eq!( + remaining, + 3 - i64::try_from(expected).unwrap(), + "rows left after `{query}`" + ); + } + + // A view over an earlier version of the table is not the table. + for statement in [ + "CREATE VIEW intents_v0 AS SELECT * FROM intents", + "ALTER TABLE intents ADD COLUMN extra INT", + ] { + client.batch_execute(statement).unwrap(); + } + let intersection_before = delete_rows_selections(&server, "intersection"); + let deleted = client + .execute( + "DELETE FROM intents ROWS (SELECT *, NULL::int FROM intents_v0)", + &[], + ) + .unwrap(); + assert_eq!(deleted, 3); + assert_eq!( + delete_rows_selections(&server, "intersection") - intersection_before, + 1 + ); +} + +/// Concurrent `DELETE ... ROWS` reading a materialized view of the table, not +/// the table itself, must not retract a row more often than the table holds it. +#[mz_ore::test] +#[allow(clippy::disallowed_methods)] +fn test_concurrent_delete_rows_from_view_does_not_over_delete() { + const MULTIPLICITY: i32 = 7; + const NUM_WORKERS: usize = 6; + const ROUNDS: usize = 5; + + let server = test_util::TestHarness::default() + .unsafe_mode() + .with_system_parameter_default("enable_delete_rows".to_string(), "true".to_string()) + .start_blocking(); + let mut setup = server.connect(postgres::NoTls).unwrap(); + for statement in [ + "CREATE TABLE t (id INT)", + "CREATE MATERIALIZED VIEW garbage AS SELECT * FROM t WHERE id = 1", + ] { + setup.batch_execute(statement).unwrap(); + } + + for _round in 0..ROUNDS { + setup.batch_execute("DELETE FROM t").unwrap(); + setup + .execute( + "INSERT INTO t SELECT 1 FROM generate_series(1, $1)", + &[&MULTIPLICITY], + ) + .unwrap(); + setup.batch_execute("INSERT INTO t VALUES (2)").unwrap(); + + // Connecting inside the spawn loop would let the first worker finish + // its DELETE before the last client even exists. + let clients: Vec<_> = (0..NUM_WORKERS) + .map(|_| server.connect(postgres::NoTls).unwrap()) + .collect(); + let barrier = Arc::new(Barrier::new(NUM_WORKERS)); + let total_deleted = Arc::new(AtomicUsize::new(0)); + let mut handles = Vec::new(); + for mut client in clients { + let barrier = Arc::clone(&barrier); + let total_deleted = Arc::clone(&total_deleted); + handles.push(thread::spawn(move || { + barrier.wait(); + let n = client + .execute("DELETE FROM t ROWS (SELECT * FROM garbage)", &[]) + .expect("DELETE ... ROWS under contention should succeed via OCC"); + total_deleted.fetch_add(usize::try_from(n).unwrap(), Ordering::SeqCst); + })); + } + for h in handles { + h.join().expect("worker panicked"); + } + + let remaining: Vec = setup + .query("SELECT id FROM t", &[]) + .unwrap() + .iter() + .map(|r| r.get(0)) + .collect(); + assert_eq!(remaining, vec![2], "only the row outside `garbage` is left"); + assert_eq!( + total_deleted.load(Ordering::SeqCst), + usize::try_from(MULTIPLICITY).unwrap(), + "sum of reported deletes must equal the initial multiplicity", + ); + } + assert_eq!( + delete_rows_selections(&server, "rows"), + u64::try_from(ROUNDS * NUM_WORKERS).unwrap(), + "every DELETE ... ROWS should retract the view's rows directly", + ); + + // Without a retry, the workers did not race, and the counts above would + // hold for serial deletes too. + let metrics = server.metrics_registry().gather(); + let histogram = session_occ_retry_histogram(&metrics); + let zero_retry_bucket = histogram + .get_bucket() + .iter() + .find(|b| b.upper_bound() == 0.0) + .expect("histogram should have a 0-retry bucket"); + assert!( + zero_retry_bucket.cumulative_count() < histogram.get_sample_count(), + "expected at least one DELETE ... ROWS to retry under contention", + ); +} diff --git a/src/sql/src/plan.rs b/src/sql/src/plan.rs index c23aace7f9814..9238a77d804ba 100644 --- a/src/sql/src/plan.rs +++ b/src/sql/src/plan.rs @@ -1230,6 +1230,23 @@ pub struct ReadThenWritePlan { pub assignments: BTreeMap, pub kind: MutationKind, pub returning: Vec, + /// For `DELETE FROM t ROWS (q)`, `q` alongside the `selection` + /// `t INTERSECT ALL q`. + pub delete_rows: Option, +} + +/// The rows a `DELETE FROM t ROWS (q)` retracts, without the intersection with +/// the table. +/// +/// `rows` may replace the plan's `selection` only once `rows` is proven to be a +/// sub-multiset of `table` at every timestamp. The intersection then equals +/// `rows`. Without that proof, `rows` can retract rows the table does not hold. +#[derive(Debug, Clone)] +pub struct DeleteRows { + /// The table version the selection reads. + pub table: GlobalId, + /// `q`, cast to the table's column types. + pub rows: HirRelationExpr, } /// Generated by `ALTER ... IF EXISTS` if the named object did not exist. diff --git a/src/sql/src/plan/query.rs b/src/sql/src/plan/query.rs index 6b52c3158b546..56a6fee3660de 100644 --- a/src/sql/src/plan/query.rs +++ b/src/sql/src/plan/query.rs @@ -740,6 +740,8 @@ pub struct ReadThenWritePlan { /// Map from column index to SET expression. Empty for DELETE statements. pub assignments: BTreeMap, pub finishing: RowSetFinishing, + /// See [`crate::plan::ReadThenWritePlan::delete_rows`]. + pub delete_rows: Option, } pub fn plan_delete_query( @@ -827,6 +829,7 @@ pub fn plan_mutation_query_inner( let desc = item.relation_desc().expect("table has desc"); let relation_type = qcx.relation_type(&get); + let mut delete_rows = None; if let Some(rows) = rows { if !using.is_empty() || selection.is_some() || alias.is_some() { sql_bail!("DELETE ... ROWS cannot be combined with USING, WHERE, or a table alias"); @@ -857,6 +860,17 @@ pub fn plan_mutation_query_inner( ); } let rows = cast_to_table_columns(&qcx, ccx, &desc, &ordering, &target_types, rows)?; + let HirRelationExpr::Get { + id: Id::Global(table), + .. + } = &get + else { + unreachable!("a resolved table is a global get, got {get:?}"); + }; + delete_rows = Some(crate::plan::DeleteRows { + table: *table, + rows: rows.clone(), + }); // `get` is the left input, so every row the selection retracts is a row of the // table, at most as often as the table holds it. get = get.intersect_all(rows); @@ -921,6 +935,7 @@ pub fn plan_mutation_query_inner( selection: get, finishing, assignments: sets, + delete_rows, }) } diff --git a/src/sql/src/plan/statement/dml.rs b/src/sql/src/plan/statement/dml.rs index 0325a683e8f53..290168133be8e 100644 --- a/src/sql/src/plan/statement/dml.rs +++ b/src/sql/src/plan/statement/dml.rs @@ -165,9 +165,17 @@ pub fn plan_read_then_write( mut selection, finishing, assignments, + mut delete_rows, }: query::ReadThenWritePlan, ) -> Result { selection.bind_parameters_and_simplify_offset(scx, QueryLifetime::OneShot, params)?; + if let Some(delete_rows) = &mut delete_rows { + delete_rows.rows.bind_parameters_and_simplify_offset( + scx, + QueryLifetime::OneShot, + params, + )?; + } let mut assignments_outer = BTreeMap::new(); for (idx, mut set) in assignments { set.bind_parameters_and_simplify_offset(scx, QueryLifetime::OneShot, params)?; @@ -182,6 +190,7 @@ pub fn plan_read_then_write( assignments: assignments_outer, kind, returning: Vec::new(), + delete_rows, })) } diff --git a/src/sql/src/rbac.rs b/src/sql/src/rbac.rs index 5ab8465256550..fdbd340c22c73 100644 --- a/src/sql/src/rbac.rs +++ b/src/sql/src/rbac.rs @@ -1462,6 +1462,7 @@ fn generate_rbac_requirements( assignments, kind, returning, + delete_rows: _, }) => { let acl_mode = match kind { MutationKind::Insert => AclMode::INSERT, diff --git a/src/transform/src/analysis.rs b/src/transform/src/analysis.rs index a42552b8f736a..2ae14bca0c0f5 100644 --- a/src/transform/src/analysis.rs +++ b/src/transform/src/analysis.rs @@ -11,6 +11,7 @@ pub mod equivalences; pub mod monotonic; +pub mod sub_multiset; use mz_expr::MirRelationExpr; @@ -21,6 +22,7 @@ pub use common::{Derived, DerivedBuilder, DerivedView}; pub use explain::annotate_plan; pub use non_negative::NonNegative; pub use repr_types::ReprRelationType; +pub use sub_multiset::SubMultiset; pub use subtree::SubtreeSize; pub use unique_keys::UniqueKeys; diff --git a/src/transform/src/analysis/sub_multiset.rs b/src/transform/src/analysis/sub_multiset.rs new file mode 100644 index 0000000000000..915e4dc35f0ae --- /dev/null +++ b/src/transform/src/analysis/sub_multiset.rs @@ -0,0 +1,237 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +//! Proves that an expression is a sub-multiset of a target collection. + +use std::collections::BTreeSet; + +use itertools::Itertools; + +use mz_expr::{Id, MirRelationExpr}; +use mz_repr::GlobalId; +use mz_repr::optimize::OptimizerFeatures; + +use crate::analysis::{Analysis, Arity, Derived, DerivedBuilder, NonNegative}; + +/// Analysis that determines whether an expression is a sub-multiset of the +/// collection `target`: at every time, every row's accumulated count in the +/// expression is at most its accumulated count in `target`. +/// +/// `true` is a proof and `false` means "not proven". Every shape the analysis +/// does not recognize is `false`, which makes the analysis sound only as long +/// as each rule below is. A false `true` lets a caller retract rows a +/// collection does not hold. +/// +/// The analysis trusts its parameters: `target` and every member of `known` +/// must be non-negative collections, and every member of `known` must be a +/// sub-multiset of `target` at every time. Like [`NonNegative`], which it uses, +/// it assumes all `Id::Global` collections are non-negative. +#[derive(Debug)] +pub struct SubMultiset { + target: GlobalId, + known: BTreeSet, +} + +impl SubMultiset { + /// An analysis proving containment in `target`, given that every collection + /// in `known` is already a sub-multiset of `target`. + pub fn new(target: GlobalId, known: BTreeSet) -> Self { + Self { target, known } + } + + /// Whether `expr` is proven to be a sub-multiset of the target. + pub fn holds(self, expr: &MirRelationExpr, features: &OptimizerFeatures) -> bool { + let mut builder = DerivedBuilder::new(features); + builder.require(self); + let derived = builder.visit(expr); + derived + .as_view() + .value::() + .copied() + .unwrap_or(false) + } +} + +impl Analysis for SubMultiset { + type Value = bool; + + fn announce_dependencies(builder: &mut DerivedBuilder) { + builder.require(NonNegative); + builder.require(Arity); + } + + fn derive( + &self, + expr: &MirRelationExpr, + index: usize, + results: &[Self::Value], + depends: &Derived, + ) -> Self::Value { + let non_negative = depends.results::(); + let arity = depends.results::(); + match expr { + MirRelationExpr::Get { + id: Id::Global(id), .. + } => *id == self.target || self.known.contains(id), + // `Let` substitutes its value, so a binding inherits the value's + // result. A `LetRec` binding can be read before its value is + // derived, which `get` reports as not proven, and the `LetRec` arm + // below keeps any result inside one from escaping. + MirRelationExpr::Get { + id: Id::Local(id), .. + } => depends + .bindings() + .get(id) + .and_then(|offset| results.get(*offset)) + .copied() + .unwrap_or(false), + // The body is the last child. + MirRelationExpr::Let { .. } => results[index - 1], + MirRelationExpr::LetRec { .. } => false, + MirRelationExpr::Filter { .. } | MirRelationExpr::ArrangeBy { .. } => { + results[index - 1] + } + // With a non-negative input, a row's count in the output is at + // most its count in the input. + MirRelationExpr::TopK { .. } => results[index - 1] && non_negative[index - 1], + MirRelationExpr::Project { input, outputs } => { + // An identity projection, or one that drops exactly the + // columns a `Map` appended, yields the rows of the `Map`'s + // input. + if is_identity(outputs, arity[index - 1]) { + results[index - 1] + } else if let MirRelationExpr::Map { .. } = &**input { + is_identity(outputs, arity[index - 2]) && results[index - 2] + } else { + false + } + } + MirRelationExpr::Threshold { input } => { + // `max(0, c)` is at most `c` wherever `c` is positive, and the + // target is non-negative. + if results[index - 1] { + return true; + } + // `max(0, p + n1 + n2 ...)` is at most `max(0, p)` when every + // `ni` is non-positive. + let MirRelationExpr::Union { .. } = &**input else { + return false; + }; + let children = children(input, index - 1, depends); + let mut proven = 0; + for (child, offset) in children { + if is_non_positive(child, offset, non_negative) { + continue; + } + if !results[offset] { + return false; + } + proven += 1; + } + proven == 1 + } + MirRelationExpr::Union { .. } => is_proven_intersection(expr, index, results, depends), + _ => false, + } + } +} + +/// Whether `expr`, at post-order `index`, is the negation of a non-negative +/// collection, possibly below operators that keep each accumulation's sign. +/// +/// The optimizer pushes `Negate` below `Map`, `Filter` and `Project`, so the +/// subtracted input of an `EXCEPT ALL` need not have `Negate` at its root. +fn is_non_positive(mut expr: &MirRelationExpr, mut index: usize, non_negative: &[bool]) -> bool { + loop { + match expr { + MirRelationExpr::Negate { .. } => return non_negative[index - 1], + // `Project` sums accumulations, which keeps them non-positive. + MirRelationExpr::Map { input, .. } + | MirRelationExpr::Filter { input, .. } + | MirRelationExpr::Project { input, .. } + | MirRelationExpr::ArrangeBy { input, .. } => { + expr = &**input; + index -= 1; + } + _ => return false, + } + } +} + +/// Whether `outputs` lists the columns `0..arity` in order. +fn is_identity(outputs: &[usize], arity: usize) -> bool { + outputs.len() == arity && outputs.iter().enumerate().all(|(i, c)| i == *c) +} + +/// The children of `expr`, at post-order `index`, in order, with their +/// post-order indexes. +fn children<'a>( + expr: &'a MirRelationExpr, + index: usize, + depends: &Derived, +) -> Vec<(&'a MirRelationExpr, usize)> { + let mut offsets: Vec<_> = depends + .children_of_rev(index, expr.children().count()) + .collect(); + offsets.reverse(); + expr.children().zip_eq(offsets).collect() +} + +/// Recognizes `A INTERSECT ALL B` as planned, +/// `Union(A, Negate(Threshold(Union(A, Y))))` with either union's inputs in +/// either order, and proves it if `A` is proven, or if `Y` is `Negate(B)` and +/// `B` is proven. +/// +/// The expression accumulates to `a - max(0, a + y)`, which is at most `a` for +/// every `y`. With `y = -b` it is `min(a, b)`, so also at most `b`. Both rely on +/// the two occurrences of `A` being structurally equal, and so evaluating to +/// the same collection. +fn is_proven_intersection( + expr: &MirRelationExpr, + index: usize, + results: &[bool], + depends: &Derived, +) -> bool { + let outer = children(expr, index, depends); + let [first, second] = outer.as_slice() else { + return false; + }; + for ((a, a_offset), (negated, negated_offset)) in [(first, second), (second, first)] { + let MirRelationExpr::Negate { input } = negated else { + continue; + }; + let MirRelationExpr::Threshold { input } = &**input else { + continue; + }; + let MirRelationExpr::Union { .. } = &**input else { + continue; + }; + // `Negate` and `Threshold` are unary, so the inner union sits two + // positions below the negation. + let inner = children(input, negated_offset - 2, depends); + let [inner_first, inner_second] = inner.as_slice() else { + continue; + }; + for ((a_copy, _), (y, y_offset)) in + [(inner_first, inner_second), (inner_second, inner_first)] + { + if a_copy != a { + continue; + } + let b_proven = matches!(y, MirRelationExpr::Negate { .. }) && results[y_offset - 1]; + if results[*a_offset] || b_proven { + return true; + } + } + } + false +} + +#[cfg(test)] +mod tests; diff --git a/src/transform/src/analysis/sub_multiset/tests.rs b/src/transform/src/analysis/sub_multiset/tests.rs new file mode 100644 index 0000000000000..23fef572375fa --- /dev/null +++ b/src/transform/src/analysis/sub_multiset/tests.rs @@ -0,0 +1,217 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +use std::collections::BTreeSet; + +use mz_expr::{LocalId, MirRelationExpr, MirScalarExpr}; +use mz_repr::optimize::OptimizerFeatures; +use mz_repr::{Datum, GlobalId, ReprRelationType, ReprScalarType}; + +use super::SubMultiset; + +const TARGET: GlobalId = GlobalId::User(1); +const KNOWN: GlobalId = GlobalId::User(2); +const OTHER: GlobalId = GlobalId::User(3); + +fn typ() -> ReprRelationType { + ReprRelationType::new(vec![ + ReprScalarType::Int64.nullable(false), + ReprScalarType::Int64.nullable(false), + ]) +} + +fn get(id: GlobalId) -> MirRelationExpr { + MirRelationExpr::global_get(id, typ()) +} + +fn local(id: u64) -> MirRelationExpr { + MirRelationExpr::local_get(LocalId::new(id), typ()) +} + +fn proven(expr: &MirRelationExpr) -> bool { + SubMultiset::new(TARGET, BTreeSet::from([KNOWN])).holds(expr, &OptimizerFeatures::default()) +} + +/// `Negate` without the double-negation elimination of [`MirRelationExpr::negate`]. +fn raw_negate(input: MirRelationExpr) -> MirRelationExpr { + MirRelationExpr::Negate { + input: Box::new(input), + } +} + +fn union(inputs: Vec) -> MirRelationExpr { + MirRelationExpr::union_many(inputs, typ()) +} + +fn intersect_all(a: MirRelationExpr, b: MirRelationExpr) -> MirRelationExpr { + union(vec![ + a.clone(), + union(vec![a, b.negate()]).threshold().negate(), + ]) +} + +fn predicate() -> MirScalarExpr { + MirScalarExpr::column(0).call_is_null().not() +} + +#[mz_ore::test] +fn gets() { + assert!(proven(&get(TARGET))); + assert!(proven(&get(KNOWN))); + assert!(!proven(&get(OTHER))); +} + +#[mz_ore::test] +fn filters_and_arrangements_pass_through() { + assert!(proven(&get(TARGET).filter([predicate()]))); + assert!(proven( + &get(TARGET).arrange_by(&[vec![MirScalarExpr::column(0)]]) + )); + assert!(!proven(&get(OTHER).filter([predicate()]))); +} + +#[mz_ore::test] +fn union_all_of_target_is_not_contained() { + assert!(!proven(&union(vec![get(TARGET), get(TARGET)]))); + assert!(!proven(&union(vec![ + get(TARGET).filter([predicate()]), + get(TARGET).filter([predicate().not()]), + ]))); +} + +#[mz_ore::test] +fn join_is_not_contained() { + let join = MirRelationExpr::join(vec![get(TARGET), get(OTHER)], vec![]); + assert!(!proven(&join.project(vec![0, 1]))); +} + +#[mz_ore::test] +fn except_all() { + assert!(proven( + &union(vec![get(TARGET), get(OTHER).negate()]).threshold() + )); + assert!(proven( + &union(vec![get(OTHER).negate(), get(TARGET)]).threshold() + )); + assert!(proven( + &union(vec![get(KNOWN), get(OTHER).negate(), get(OTHER).negate()]).threshold() + )); + assert!(proven(&get(TARGET).threshold())); + // The optimizer pushes `Negate` below `Map` and `Project`. + let one = MirScalarExpr::literal_ok(Datum::Int64(1), ReprScalarType::Int64); + let pushed = get(OTHER).project(vec![0]).negate().map(vec![one.clone()]); + assert!(proven(&union(vec![get(TARGET), pushed]).threshold())); + let positive = get(OTHER).project(vec![0]).map(vec![one]); + assert!(!proven(&union(vec![get(TARGET), positive]).threshold())); + // A doubly negated input adds rows. + assert!(!proven( + &union(vec![get(TARGET), raw_negate(raw_negate(get(OTHER)))]).threshold() + )); + // Two proven inputs can exceed the target. + assert!(!proven( + &union(vec![get(TARGET), get(KNOWN), get(OTHER).negate()]).threshold() + )); + assert!(!proven( + &union(vec![get(TARGET), get(OTHER), get(OTHER).negate()]).threshold() + )); + assert!(!proven( + &union(vec![get(OTHER), get(TARGET).negate()]).threshold() + )); + // Without the threshold, nothing is proven. + assert!(!proven(&union(vec![get(TARGET), get(OTHER).negate()]))); +} + +#[mz_ore::test] +fn intersect_all_is_contained_in_either_input() { + assert!(proven(&intersect_all(get(TARGET), get(OTHER)))); + assert!(proven(&intersect_all(get(OTHER), get(TARGET)))); + assert!(!proven(&intersect_all(get(OTHER), get(OTHER)))); + + // Inputs of both unions in the other order. + let swapped = union(vec![ + union(vec![get(OTHER).negate(), get(TARGET)]) + .threshold() + .negate(), + get(TARGET), + ]); + assert!(proven(&swapped)); + + // With `A` proven, the subtracted input can be anything. + let any_y = union(vec![ + get(TARGET), + union(vec![get(TARGET), get(OTHER)]).threshold().negate(), + ]); + assert!(proven(&any_y)); + // With only `B` proven, it has to be subtracted. + let added_b = union(vec![ + get(OTHER), + union(vec![get(OTHER), get(TARGET)]).threshold().negate(), + ]); + assert!(!proven(&added_b)); + + // The two copies of `A` must be the same expression. + let mismatched = union(vec![ + get(OTHER), + union(vec![get(TARGET), get(TARGET).negate()]) + .threshold() + .negate(), + ]); + assert!(!proven(&mismatched)); +} + +#[mz_ore::test] +fn projections() { + assert!(proven(&get(TARGET).project(vec![0, 1]))); + assert!(!proven(&get(TARGET).project(vec![1, 0]))); + assert!(!proven(&get(TARGET).project(vec![0]))); + assert!(!proven(&get(TARGET).project(vec![0, 0]))); + + let one = MirScalarExpr::literal_ok(Datum::Int64(1), ReprScalarType::Int64); + let mapped = get(TARGET).map(vec![one]); + assert!(!proven(&mapped)); + assert!(proven(&mapped.clone().project(vec![0, 1]))); + assert!(!proven(&mapped.clone().project(vec![0, 2]))); + assert!(!proven(&mapped.project(vec![2, 1]))); +} + +#[mz_ore::test] +fn other_operators_are_not_proven() { + assert!(!proven(&get(TARGET).negate())); + assert!(!proven(&get(TARGET).distinct())); + assert!(!proven(&MirRelationExpr::constant(vec![], typ()))); + assert!(proven(&get(TARGET).top_k(vec![], vec![], None, 1, None))); + assert!(!proven( + &union(vec![get(TARGET), get(OTHER).negate()]).top_k(vec![], vec![], None, 1, None) + )); +} + +#[mz_ore::test] +fn let_bindings() { + let let_target = MirRelationExpr::Let { + id: LocalId::new(1), + value: Box::new(get(TARGET)), + body: Box::new(intersect_all(get(OTHER), local(1))), + }; + assert!(proven(&let_target)); + + let let_other = MirRelationExpr::Let { + id: LocalId::new(1), + value: Box::new(get(OTHER)), + body: Box::new(local(1).filter([predicate()])), + }; + assert!(!proven(&let_other)); + + let letrec = MirRelationExpr::LetRec { + ids: vec![LocalId::new(1)], + values: vec![get(TARGET)], + limits: vec![None], + body: Box::new(local(1)), + }; + assert!(!proven(&letrec)); +} diff --git a/test/sqllogictest/delete_rows.slt b/test/sqllogictest/delete_rows.slt index e186202f948d7..a7e5c30e83147 100644 --- a/test/sqllogictest/delete_rows.slt +++ b/test/sqllogictest/delete_rows.slt @@ -331,3 +331,235 @@ query IT SELECT * FROM t ---- 1 x + +# When the query is provably contained in the table, the write retracts the +# query's rows without intersecting them with the table. The result is the same +# either way. `src/environmentd/tests/read_then_write.rs` observes which +# selection runs. + +statement ok +CREATE TABLE gc_intents (id int, key text, val text, done bool) + +statement ok +CREATE TABLE gc_store (id int, key text, val text) + +statement ok +INSERT INTO gc_store VALUES (1, 'a', 'x'), (3, 'c', 'z') + +statement ok +CREATE MATERIALIZED VIEW garbage AS + SELECT * FROM gc_intents EXCEPT ALL SELECT id, key, val, false FROM gc_store + +statement ok +CREATE VIEW garbage_view AS + SELECT * FROM gc_intents EXCEPT ALL SELECT id, key, val, false FROM gc_store + +statement ok +CREATE VIEW garbage_indexed AS + SELECT * FROM gc_intents EXCEPT ALL SELECT id, key, val, false FROM gc_store + +statement ok +CREATE INDEX garbage_indexed_idx ON garbage_indexed (id) + +statement ok +INSERT INTO gc_intents VALUES + (1, 'a', 'x', false), (1, 'a', 'x', false), (2, 'b', 'y', false), (3, 'c', 'z', false) + +simple +DELETE FROM gc_intents ROWS ( + SELECT * FROM gc_intents EXCEPT ALL SELECT id, key, val, false FROM gc_store +) +---- +COMPLETE 2 + +query ITTTI rowsort +SELECT *, count(*) FROM gc_intents GROUP BY id, key, val, done +---- +1 a x false 1 +3 c z false 1 + +statement ok +INSERT INTO gc_intents VALUES (1, 'a', 'x', false), (2, 'b', 'y', false), (2, 'b', 'y', false) + +simple +DELETE FROM gc_intents ROWS (SELECT * FROM garbage) +---- +COMPLETE 3 + +query ITTTI rowsort +SELECT *, count(*) FROM gc_intents GROUP BY id, key, val, done +---- +1 a x false 1 +3 c z false 1 + +query I +SELECT count(*) FROM garbage +---- +0 + +statement ok +INSERT INTO gc_intents VALUES (1, 'a', 'x', false), (4, 'd', 'w', true) + +simple +DELETE FROM gc_intents ROWS (SELECT * FROM garbage_view) +---- +COMPLETE 2 + +statement ok +INSERT INTO gc_intents VALUES (1, 'a', 'x', false), (4, 'd', 'w', true) + +simple +DELETE FROM gc_intents ROWS (SELECT * FROM garbage_indexed WHERE id = 4) +---- +COMPLETE 1 + +query ITTTI rowsort +SELECT *, count(*) FROM gc_intents GROUP BY id, key, val, done +---- +1 a x false 2 +3 c z false 1 + +simple +DELETE FROM gc_intents ROWS (SELECT * FROM garbage_indexed) +---- +COMPLETE 1 + +query ITTTI rowsort +SELECT *, count(*) FROM gc_intents GROUP BY id, key, val, done +---- +1 a x false 1 +3 c z false 1 + +query I +SELECT count(*) FROM garbage_indexed +---- +0 + +# A materialized view with a refresh schedule holds rows the table has since +# lost, so its rows are not contained in the table and are intersected with it. + +simple conn=mz_system,user=mz_system +ALTER SYSTEM SET enable_refresh_every_mvs = true +---- +COMPLETE 0 + +statement ok +CREATE TABLE s (a int) + +statement ok +INSERT INTO s VALUES (1), (1), (2) + +statement ok +CREATE MATERIALIZED VIEW stale WITH (REFRESH EVERY '1000 hours') AS SELECT * FROM s + +query I rowsort +SELECT * FROM stale +---- +1 +1 +2 + +statement ok +DELETE FROM s WHERE a = 1 + +simple +DELETE FROM s ROWS (SELECT * FROM stale) +---- +COMPLETE 1 + +query I +SELECT count(*) FROM s +---- +0 + +statement ok +INSERT INTO s VALUES (1) + +query I +SELECT * FROM s +---- +1 + +# The union of the table with itself holds each row twice as often as the +# table, so each row is deleted only as often as the table holds it. + +statement ok +INSERT INTO s VALUES (1), (2) + +simple +DELETE FROM s ROWS (SELECT * FROM s UNION ALL SELECT * FROM s) +---- +COMPLETE 3 + +query I +SELECT count(*) FROM s +---- +0 + +statement ok +INSERT INTO s VALUES (3) + +query I +SELECT * FROM s +---- +3 + +# The rows of another table are intersected with the target. + +statement ok +CREATE TABLE s2 (a int) + +statement ok +INSERT INTO s2 VALUES (3), (3), (4) + +simple +DELETE FROM s ROWS (SELECT * FROM s2) +---- +COMPLETE 1 + +query I +SELECT count(*) FROM s +---- +0 + +# A view over an earlier version of the table is not the table. + +simple conn=mz_system,user=mz_system +ALTER SYSTEM SET enable_alter_table_add_column = true +---- +COMPLETE 0 + +statement ok +CREATE TABLE versioned (a int) + +statement ok +INSERT INTO versioned VALUES (1), (1), (2) + +statement ok +CREATE VIEW versioned_old AS SELECT * FROM versioned + +statement ok +ALTER TABLE versioned ADD COLUMN b int + +statement ok +INSERT INTO versioned VALUES (1, 10) + +simple +DELETE FROM versioned ROWS (SELECT a, NULL::int FROM versioned_old) +---- +COMPLETE 3 + +query II rowsort +SELECT * FROM versioned +---- +1 10 + +simple +DELETE FROM versioned ROWS (SELECT a, 10 FROM versioned_old) +---- +COMPLETE 1 + +query I +SELECT count(*) FROM versioned +---- +0