diff --git a/src/compute-client/src/controller.rs b/src/compute-client/src/controller.rs index 7dcc5d6b49b92..3d9e145b0076a 100644 --- a/src/compute-client/src/controller.rs +++ b/src/compute-client/src/controller.rs @@ -52,6 +52,7 @@ use mz_expr::row::RowCollection; use mz_ore::cast::CastFrom; use mz_ore::metrics::MetricsRegistry; use mz_ore::now::NowFn; +use mz_ore::soft_assert_or_log; use mz_ore::tracing::OpenTelemetryContext; use mz_persist_types::PersistLocation; use mz_repr::{GlobalId, RelationDesc, Row, Timestamp}; @@ -815,6 +816,45 @@ impl ComputeController { return Err(EmptyAsOfForCopyTo); } + // Validation: the dataflow exports something + // + // An export-less description has nothing to render and no answer to "what do the exports + // read", which the checks below are phrased in terms of. `optimize_dataflow` leaves such a + // description's imports alone for that reason, so one arriving here would fail the import + // check for the wrong reason. + soft_assert_or_log!( + !dataflow.index_exports.is_empty() || !dataflow.sink_exports.is_empty(), + "dataflow {} has no exports", + dataflow.debug_name, + ); + + // The imports the exports actually read. `optimize_dataflow` prunes the import list to + // exactly this set, so the two agree unless a producer stopped pruning. + // + // Computed once and used twice: the check below reports a loose list, and + // `determine_time_dependence` counts through it rather than over the raw list. That + // consumer is the one whose wrong answer hangs an environment: an import no export reads + // would report wall-clock dependence for a dataflow whose exports are constant, earning it + // a dataflow expiration that pins the output frontier days short of the empty antichain, + // and nothing downstream would learn the collection is final. Deriving it from this set + // makes that correct by construction, leaving the prune to reclaim the read hold and the + // persist source. + let used_imports = dataflow.used_import_ids(); + + // Validation: every import is read + // + // The read holds and the persist sources the replicas build are still derived from the raw + // list below, so a loose one describes a dataflow other than the one that will run. A + // logging variant rather than `soft_assert_no_log!`: the walk is paid for above either way, + // so reporting it in production costs only the comparison. + soft_assert_or_log!( + dataflow.import_ids().all(|id| used_imports.contains(&id)), + "dataflow {} imports collections no export reads: imports {:?}, read {:?}", + dataflow.debug_name, + dataflow.import_ids().collect::>(), + used_imports, + ); + // Validation: input collections let storage_ids = dataflow.imported_source_ids().collect(); let mut import_read_holds = self.storage_collections.acquire_read_holds(storage_ids)?; @@ -835,7 +875,7 @@ impl ComputeController { } } let time_dependence = self - .determine_time_dependence(instance_id, &dataflow) + .determine_time_dependence(instance_id, &dataflow, &used_imports) .expect("must exist"); let instance = self.instance_mut(instance_id).expect("validated"); @@ -1017,24 +1057,29 @@ impl ComputeController { } /// Determine the time dependence for a dataflow. + /// + /// `used_imports` are the imports the exports read, as + /// [`DataflowDescription::used_import_ids`] reports them. Only those count: an import no export + /// reads would report wall-clock dependence for a dataflow whose exports are constant, and that + /// earns it a dataflow expiration, which pins its output frontier at the expiration time. A + /// constant export's frontier is the empty antichain, so the pin would hold it days short of + /// the truth and whoever reads that frontier would never learn the collection can no longer + /// change. + /// + /// `optimize_dataflow` prunes the import list to this set, so the two agree and the filtering + /// is a no-op. It is here because this is the consumer whose wrong answer hangs an environment, + /// and deriving the answer from the read set makes it independent of the list staying tight. fn determine_time_dependence( &self, instance_id: ComputeInstanceId, dataflow: &DataflowDescription, + used_imports: &BTreeSet, ) -> Result, TimeDependenceError> { let instance = self .instance(instance_id) .map_err(|err| TimeDependenceError::InstanceMissing(err.0))?; let mut time_dependencies = Vec::new(); - // Only the imports the exports read say anything about how this dataflow's frontier relates - // to wall clock. Counting one that optimization left unused reports wall-clock dependence - // for a dataflow whose exports are constant, and that earns it a dataflow expiration, which - // pins its output frontier at the expiration time. A constant export's frontier is the empty - // antichain, so the pin holds it days short of the truth and whoever reads that frontier - // never learns the collection can no longer change. - let used_imports = dataflow.used_import_ids(); - for id in dataflow .imported_index_ids() .filter(|id| used_imports.contains(id)) diff --git a/src/compute-types/src/dataflows.rs b/src/compute-types/src/dataflows.rs index 96f080604d499..af41b7fab388a 100644 --- a/src/compute-types/src/dataflows.rs +++ b/src/compute-types/src/dataflows.rs @@ -481,11 +481,15 @@ where /// Computes the set of imports the dataflow's exports read, meaning the imports reachable from /// an index export's `on_id` or a sink export's `from`. /// - /// This is not the same as [`Self::import_ids`]. An import that no export reaches contributes - /// neither contents nor frontier to any of them, and one ends up in the description whenever - /// optimization drops the reference to it after the imports were collected, for instance by - /// folding a collection to a constant. The answer covers the dataflow as a whole, so an import - /// only one export reads is still reported, and a dataflow with no exports reports none. + /// A description that has been through the optimizer answers [`Self::import_ids`] here, because + /// the optimizer prunes the import list to what the exports read. The two come apart while a + /// description is still being assembled, and this is what the prune and the assertion guarding + /// it are both defined in terms of. The answer covers the dataflow as a whole, so an import only + /// one export reads is still reported, and a dataflow with no exports reports none. + /// + /// NOTE: On the index side this over-approximates. [`Self::depends_on`] cannot tell which index + /// on a collection a plan will use, so reaching a collection reports every index imported on it. + /// Pruning index imports needs the exact usage information the MIR pipeline collects, not this. /// /// Panics for an export naming a collection that is neither an import nor built exactly once /// here, which is [`Self::depends_on`]'s precondition on its argument. Rendering resolves the diff --git a/src/transform/src/dataflow.rs b/src/transform/src/dataflow.rs index b9c30d0475aad..92f7fe4b1fc4e 100644 --- a/src/transform/src/dataflow.rs +++ b/src/transform/src/dataflow.rs @@ -100,9 +100,13 @@ pub fn optimize_dataflow( transform_ctx.df_meta, )?; + prune_dataflow_source_imports(dataflow); + // Warning: If you want to add a transform call here, consider it very carefully whether it // could accidentally invalidate information that we already derived above in - // `optimize_dataflow_monotonic` or `prune_and_annotate_dataflow_index_imports`. + // `optimize_dataflow_monotonic`, `prune_and_annotate_dataflow_index_imports`, or + // `prune_dataflow_source_imports`. A transform here that drops the last reference to an import + // puts back exactly the discrepancy the two prunes just removed. mz_repr::explain::trace_plan(dataflow); @@ -511,6 +515,46 @@ pub fn optimize_dataflow_snapshot(dataflow: &mut DataflowDesc) -> Result<(), Tra Ok(()) } +/// Restricts the sources imported by `dataflow` to only the ones its exports read. +/// +/// The counterpart to [`prune_and_annotate_dataflow_index_imports`] for source imports. Imports are +/// collected before the global pipeline runs, from the plans as they were written, so a transform +/// can drop the last `Get` of one, for instance by folding a selection to a constant. +/// +/// An import that survives that is not free. Every worker builds a `persist_source` for it and +/// decodes a shard into a stream nobody consumes, and the controller takes a read hold that pins +/// the collection's `since` for as long as the dataflow lives. Both are read off the import list +/// directly, so pruning is what reclaims them. +/// +/// A third consumer, the wall-clock dependence a dataflow reports, is the one whose wrong answer +/// does real damage: it earns a dataflow whose exports can never change again an expiration, which +/// pins their output frontier at the expiration time rather than letting it reach the empty +/// antichain, so nothing downstream learns the collection is final. +/// `ComputeController::determine_time_dependence` derives that from the read set rather than from +/// the import list, so it does not depend on this pass having run. `create_dataflow` also reports a +/// list this pass left loose. +/// +/// The input plans should be normalized with `NormalizeLets`, for the same reason +/// [`prune_and_annotate_dataflow_index_imports`] wants them to be: an unused `Let` binding can +/// otherwise keep alive a `Get` that nothing reads. +#[mz_ore::instrument( + target = "optimizer", + level = "debug", + fields(path.segment = "source_imports") +)] +fn prune_dataflow_source_imports(dataflow: &mut DataflowDesc) { + // NOTE: A description with no exports has no answer to "what do the exports read", and pruning + // everything is the wrong one. `EXPLAIN` builds a peek description without its index export, + // see the conditional `export_index` in `mz_adapter::optimize::peek`. Such a description is + // explained and then dropped, never installed, so leaving its import list alone costs nothing. + if dataflow.index_exports.is_empty() && dataflow.sink_exports.is_empty() { + return; + } + + let used = dataflow.used_import_ids(); + dataflow.source_imports.retain(|id, _| used.contains(id)); +} + /// Restricts the indexes imported by `dataflow` to only the ones it needs. /// It also adds to the `DataflowMetainfo` how each index will be used. /// It also annotates global `Get`s with whether they will be reads from Persist or an index, plus @@ -1373,3 +1417,104 @@ impl DataflowMetainfo { } } } + +#[cfg(test)] +mod tests { + use mz_compute_types::sinks::{ + ComputeSinkConnection, ComputeSinkDesc, SubscribeSinkConnection, + }; + use mz_expr::OptimizedMirRelationExpr; + use mz_repr::{RelationDesc, ReprRelationType, ReprScalarType, SqlRelationType}; + + use super::*; + + const READ: GlobalId = GlobalId::User(1); + const UNREAD: GlobalId = GlobalId::User(2); + const VIEW: GlobalId = GlobalId::Transient(1); + const SINK: GlobalId = GlobalId::Transient(2); + + fn typ() -> ReprRelationType { + ReprRelationType::new(vec![ReprScalarType::Int64.nullable(false)]) + } + + /// A dataflow importing `READ` and `UNREAD` and building `VIEW` from `plan`. It has no exports + /// until `export_subscribe` adds one. + fn dataflow(plan: MirRelationExpr) -> DataflowDesc { + let mut df = DataflowDesc::new("test".to_string()); + df.import_source(READ, SqlRelationType::from_repr(&typ()), false); + df.import_source(UNREAD, SqlRelationType::from_repr(&typ()), false); + df.objects_to_build.push(BuildDesc { + id: VIEW, + plan: OptimizedMirRelationExpr::declare_optimized(plan), + }); + df + } + + fn export_subscribe(df: &mut DataflowDesc) { + df.export_sink( + SINK, + ComputeSinkDesc { + from: VIEW, + from_desc: RelationDesc::new(SqlRelationType::from_repr(&typ()), ["c"]), + connection: ComputeSinkConnection::Subscribe(SubscribeSinkConnection { + output: Vec::new(), + }), + with_snapshot: true, + up_to: Default::default(), + non_null_assertions: Vec::new(), + refresh_schedule: None, + }, + ); + } + + fn get(id: GlobalId) -> MirRelationExpr { + MirRelationExpr::Get { + id: Id::Global(id), + typ: typ(), + access_strategy: AccessStrategy::Persist, + } + } + + fn constant() -> MirRelationExpr { + MirRelationExpr::Constant { + rows: Ok(Vec::new()), + typ: typ(), + } + } + + #[mz_ore::test] + fn prune_drops_the_import_no_export_reads() { + let mut df = dataflow(get(READ)); + export_subscribe(&mut df); + + prune_dataflow_source_imports(&mut df); + + assert_eq!(df.imported_source_ids().collect::>(), vec![READ]); + } + + /// The shape this prune exists for: the optimizer folded the export to a constant, so neither + /// import is read any more even though both are still listed. + #[mz_ore::test] + fn prune_drops_every_import_of_a_constant_export() { + let mut df = dataflow(constant()); + export_subscribe(&mut df); + + prune_dataflow_source_imports(&mut df); + + assert_eq!(df.imported_source_ids().count(), 0); + } + + /// A description with no exports is one `EXPLAIN` builds and never installs. Pruning it against + /// an empty set of exports would strip every import, so the prune leaves it alone. + #[mz_ore::test] + fn prune_leaves_an_export_less_description_alone() { + let mut df = dataflow(get(READ)); + + prune_dataflow_source_imports(&mut df); + + assert_eq!( + df.imported_source_ids().collect::>(), + vec![READ, UNREAD] + ); + } +} diff --git a/test/sqllogictest/explain/materialized_view.slt b/test/sqllogictest/explain/materialized_view.slt index 1f0eed832f810..134302ec85d93 100644 --- a/test/sqllogictest/explain/materialized_view.slt +++ b/test/sqllogictest/explain/materialized_view.slt @@ -210,3 +210,91 @@ Used Indexes: Target cluster: quickstart EOF + + +# An import whose last reference the global pipeline folds away is pruned from the dataflow +# description. It must not show up as a source: the description drives the persist sources the +# replicas build, the read holds the controller takes, and whether the dataflow is taken to depend +# on wall clock. + +statement ok +CREATE TABLE flags (c0 bool); + +statement ok +CREATE MATERIALIZED VIEW mv_folded AS + WITH x AS (SELECT c0 FROM flags WHERE TRUE = c0) + (SELECT true AS c0 FROM x) EXCEPT ALL (SELECT c0 FROM x); + +# Pin the premise. The two branches are the same collection under the filter, which the global +# pipeline sees and local optimization does not, so the table is still an import when the fold +# removes its last reference. Were it to fold locally, the description would never import the table +# and the golden below would hold for the wrong reason. +query T multiline +EXPLAIN LOCALLY OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR +MATERIALIZED VIEW mv_folded; +---- +With + cte l0 = + Filter (#0{c0} = true) + Get materialize.public.flags +Return + Threshold + Union + Project (#1) + Map (true) + Get l0 + Negate + Get l0 + +Target cluster: quickstart + +EOF + +query T multiline +EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR +MATERIALIZED VIEW mv_folded; +---- +materialize.public.mv_folded: + Constant + +Target cluster: quickstart + +EOF + + +# The same state through a different transform. The predicate in `unsatisfiable` cannot hold, so +# that view folds on its own, and inlining it into the join is what drops the last reference to +# `flags`. Unlike the fold above this one does not depend on recognizing a mapped literal as a +# column, so the two goldens do not stand or fall together. `flags` carries no index, so it is a +# source import rather than an index one. +statement ok +CREATE VIEW unsatisfiable AS SELECT c0 FROM flags WHERE c0 IS NULL AND c0 = true; + +statement ok +CREATE MATERIALIZED VIEW mv_empty_join AS + SELECT f.c0 FROM flags f, unsatisfiable u WHERE f.c0 = u.c0; + +query T multiline +EXPLAIN LOCALLY OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR +MATERIALIZED VIEW mv_empty_join; +---- +Project (#0{c0}) + Join on=(#0{c0} = #1{c0}) + Filter (#0{c0}) IS NOT NULL + Get materialize.public.flags + Get materialize.public.unsatisfiable + +Target cluster: quickstart + +EOF + +query T multiline +EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR +MATERIALIZED VIEW mv_empty_join; +---- +materialize.public.mv_empty_join: + Constant + +Target cluster: quickstart + +EOF