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.

Author: claude

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};
@@ -813,25 +813,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::<Vec<_>>(),
-            dataflow.used_import_ids(),
+            used_imports,
         );
 
         // Validation: input collections
@@ -854,7 +872,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");
@@ -1036,31 +1054,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<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();
 
-        // 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() {

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