From 101eb384c124a7c4caf542f3066196a62080f92b Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 13:06:47 +0000 Subject: [PATCH 1/3] compute: prune the source imports no export reads Counting only the read imports in `determine_time_dependence` fixes the frontier confusion at the consumer, but the discrepancy it reads around is a gap of its own, and it costs more than that one symptom. Source imports are collected before the global MIR pipeline runs, from the plans as they were written. `prune_and_annotate_dataflow_index_imports` tightens the index imports at the end of the pipeline, and `DataflowBuilder::import_into_dataflow` says that is by design, but `source_imports` never had a counterpart: the three passes that touch it after optimization only annotate. A transform that drops the last `Get` of an imported collection leaves the import behind. Every worker then builds a `persist_source` for it and decodes a shard into a stream nobody consumes, the controller takes a read hold that pins the collection's `since` for the life of the dataflow, and the dataflow reports a wall-clock dependence its exports do not have. So restore the invariant rather than read around it. `prune_dataflow_source_imports` runs at the end of `optimize_dataflow`, beside the index prune, and drops what no export reads. `ComputeController::create_dataflow` soft-asserts that what reaches it is tight, so a producer that stops pruning fails a test rather than reaching a replica. `determine_time_dependence` goes back to counting every import, which is now the same set, and `used_import_ids` becomes the definition both the prune and the assertion are written in terms of. A description with no exports is exempt. `EXPLAIN` builds a peek description without its index export, and pruning that against an empty export set would strip every import. Such a description is explained and dropped, never installed. Tests: unit tests for the prune, including the export-less case, and an `EXPLAIN MATERIALIZED VIEW` golden pinning that a folded import stops showing up as a source. That golden is paired with an `EXPLAIN LOCALLY OPTIMIZED PLAN` one asserting the table is still read at that stage, so the case cannot go vacuous if the fold ever moves out of the global pipeline. The integration test from the previous commit still covers the hang end to end. --- src/compute-client/src/controller.rs | 45 ++++-- src/compute-types/src/dataflows.rs | 14 +- src/transform/src/dataflow.rs | 142 +++++++++++++++++- .../explain/materialized_view.slt | 50 ++++++ 4 files changed, 230 insertions(+), 21 deletions(-) diff --git a/src/compute-client/src/controller.rs b/src/compute-client/src/controller.rs index 7dcc5d6b49b92..c3dbb7cfe44cb 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_no_log; use mz_ore::tracing::OpenTelemetryContext; use mz_persist_types::PersistLocation; use mz_repr::{GlobalId, RelationDesc, Row, Timestamp}; @@ -815,6 +816,27 @@ impl ComputeController { return Err(EmptyAsOfForCopyTo); } + // Validation: every import is read + // + // The import list is meant to be exactly the imports the exports read. `optimize_dataflow` + // prunes it to that, and the read holds, the time dependence, and the persist sources the + // replicas build are all derived from it below, so each of them describes a dataflow other + // than the one that will run once the list is loose. This catches a producer that stops + // pruning. + // + // `soft_assert_no_log!` rather than a logging variant because the check walks the plan and + // this runs per peek, and the walk is worth paying for only where it can fail a test. + soft_assert_no_log!( + { + let used = dataflow.used_import_ids(); + dataflow.import_ids().all(|id| used.contains(&id)) + }, + "dataflow {} imports collections no export reads: imports {:?}, read {:?}", + dataflow.debug_name, + dataflow.import_ids().collect::>(), + dataflow.used_import_ids(), + ); + // Validation: input collections let storage_ids = dataflow.imported_source_ids().collect(); let mut import_read_holds = self.storage_collections.acquire_read_holds(storage_ids)?; @@ -1027,28 +1049,21 @@ impl ComputeController { .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 + // Every import counts, which is only the right answer because the import list is the set of + // imports the exports read. 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 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)) - { + // 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. `create_dataflow` asserts + // the list is tight before we get here. + for id in dataflow.imported_index_ids() { let dependence = instance .get_time_dependence(id) .map_err(|err| TimeDependenceError::CollectionMissing(err.0))?; time_dependencies.push(dependence); } - 'source: for id in dataflow - .imported_source_ids() - .filter(|id| used_imports.contains(id)) - { + 'source: for id in dataflow.imported_source_ids() { // We first check whether the id is backed by a compute object, in which case we use // the time dependence we know. This is true for storage sinks. for instance in self.instances.values() { 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..87f794a1eccad 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,41 @@ 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, the controller takes a read hold that pins the +/// collection's `since` for as long as the dataflow lives, and the dataflow reports a wall-clock +/// dependence none of its exports have. The last of those is the dangerous one. It earns a dataflow +/// whose exports can never change again an expiration, and that pins their output frontier at the +/// expiration time rather than letting it reach the empty antichain, so nothing downstream learns +/// the collection is final. +/// +/// 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 +1412,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..7ba5c4dc81339 100644 --- a/test/sqllogictest/explain/materialized_view.slt +++ b/test/sqllogictest/explain/materialized_view.slt @@ -210,3 +210,53 @@ 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 From b2aca48ba14dac5648d561248f595abc82a573fe Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 14:25:44 +0000 Subject: [PATCH 2/3] compute: cover the source-import prune with a second fold The `EXCEPT ALL` golden depends on the optimizer recognizing a mapped literal as the column it was filtered on, so a change to that reasoning would take the only dataflow-level coverage of the prune with it. This second shape reaches 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`. Its `EXPLAIN LOCALLY OPTIMIZED PLAN` golden pins the premise the same way: the table is still read when the imports are collected, so the prune has something to do. Reaching this state at all needs the fold to be one only the dataflow-level pipeline can see. A view that folds to a constant locally is inlined as a constant and never imports its inputs, which is why `LIMIT 0` behind a view does not qualify. --- .../explain/materialized_view.slt | 38 +++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/test/sqllogictest/explain/materialized_view.slt b/test/sqllogictest/explain/materialized_view.slt index 7ba5c4dc81339..134302ec85d93 100644 --- a/test/sqllogictest/explain/materialized_view.slt +++ b/test/sqllogictest/explain/materialized_view.slt @@ -260,3 +260,41 @@ materialize.public.mv_folded: 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 From 4dbaaf5826353afe56b434c30cfe91520e59d2a6 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 19 Aug 2026 11:54:56 +0000 Subject: [PATCH 3/3] compute: derive the time dependence from the read set, and say so in production `soft_assert_no_log!` is silent when soft assertions are off, so with the import list as the only input to a dataflow's wall-clock dependence, production rested entirely on the prune covering every path that installs a dataflow. A path slipping past would pin the output frontier at the replica expiration, leave the collection never reporting final, and say nothing anywhere, which is the silence SQL-635 presented as. `create_dataflow` now computes `used_import_ids` once and uses it twice. `determine_time_dependence` takes it and counts through it rather than over the raw import list, so the consumer whose wrong answer hangs an environment is correct whether or not the prune ran. The tightness check reports against the same set, and because the walk is paid for regardless it can afford `soft_assert_or_log!`, which reports a loose list in production instead of nothing. The prune keeps its job. Read holds and persist sources are read off the import list directly, so it is what reclaims them. An export-less description gets its own check. It has no answer to "what do the exports read", so the import check would have failed it for the wrong reason, and the prune already leaves such a description alone. Stating the two invariants separately keeps both messages accurate. --- src/compute-client/src/controller.rs | 80 +++++++++++++++++++--------- src/transform/src/dataflow.rs | 17 +++--- 2 files changed, 66 insertions(+), 31 deletions(-) diff --git a/src/compute-client/src/controller.rs b/src/compute-client/src/controller.rs index c3dbb7cfe44cb..3d9e145b0076a 100644 --- a/src/compute-client/src/controller.rs +++ b/src/compute-client/src/controller.rs @@ -52,7 +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_no_log; +use mz_ore::soft_assert_or_log; use mz_ore::tracing::OpenTelemetryContext; use mz_persist_types::PersistLocation; use mz_repr::{GlobalId, RelationDesc, Row, Timestamp}; @@ -816,25 +816,43 @@ impl ComputeController { return Err(EmptyAsOfForCopyTo); } - // Validation: every import is read + // Validation: the dataflow exports something // - // The import list is meant to be exactly the imports the exports read. `optimize_dataflow` - // prunes it to that, and the read holds, the time dependence, and the persist sources the - // replicas build are all derived from it below, so each of them describes a dataflow other - // than the one that will run once the list is loose. This catches a producer that stops - // pruning. + // 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. // - // `soft_assert_no_log!` rather than a logging variant because the check walks the plan and - // this runs per peek, and the walk is worth paying for only where it can fail a test. - soft_assert_no_log!( - { - let used = dataflow.used_import_ids(); - dataflow.import_ids().all(|id| used.contains(&id)) - }, + // 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::>(), - dataflow.used_import_ids(), + used_imports, ); // Validation: input collections @@ -857,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"); @@ -1039,31 +1057,43 @@ 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(); - // Every import counts, which is only the right answer because the import list is the set of - // imports the exports read. 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. `create_dataflow` asserts - // the list is tight before we get here. - for id in dataflow.imported_index_ids() { + for id in dataflow + .imported_index_ids() + .filter(|id| used_imports.contains(id)) + { let dependence = instance .get_time_dependence(id) .map_err(|err| TimeDependenceError::CollectionMissing(err.0))?; time_dependencies.push(dependence); } - 'source: for id in dataflow.imported_source_ids() { + 'source: for id in dataflow + .imported_source_ids() + .filter(|id| used_imports.contains(id)) + { // We first check whether the id is backed by a compute object, in which case we use // the time dependence we know. This is true for storage sinks. for instance in self.instances.values() { diff --git a/src/transform/src/dataflow.rs b/src/transform/src/dataflow.rs index 87f794a1eccad..92f7fe4b1fc4e 100644 --- a/src/transform/src/dataflow.rs +++ b/src/transform/src/dataflow.rs @@ -522,12 +522,17 @@ pub fn optimize_dataflow_snapshot(dataflow: &mut DataflowDesc) -> Result<(), Tra /// 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, the controller takes a read hold that pins the -/// collection's `since` for as long as the dataflow lives, and the dataflow reports a wall-clock -/// dependence none of its exports have. The last of those is the dangerous one. It earns a dataflow -/// whose exports can never change again an expiration, and that pins their output frontier at the -/// expiration time rather than letting it reach the empty antichain, so nothing downstream learns -/// the collection is final. +/// 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