compute: prune the source imports no export reads

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::<Vec<_>>(),
+            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<mz_compute_types::plan::LirRelationExpr, ()>,
+        used_imports: &BTreeSet<GlobalId>,
     ) -> Result<Option<TimeDependence>, 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))

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

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<RawOptimizerNotice> {
         }
     }
 }
+
+#[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<_>>(), 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<_>>(),
+            vec![READ, UNREAD]
+        );
+    }
+}

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 <empty>
+
+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 <empty>
+
+Target cluster: quickstart
+
+EOF