Move time dependence into controller

This doesn't yet handle timelines correctly.

Signed-off-by: Moritz Hoffmann <[email protected]>

Author: antiguru

misc/python/materialize/cli/ci_annotate_errors.py

@@ -125,8 +125,7 @@
     | restart-materialized-1\ *|\ thread\ 'coordinator'\ panicked\ at\ 'external\ operation\ .*\ failed\ unrecoverably.*
     # Expected in cluster test
     | cluster-clusterd[12]-1\ .*\ halting\ process:\ new\ timely\ configuration\ does\ not\ match\ existing\ timely\ configuration
-    | cluster-clusterd1-1\ .*\ not\ less\ than\ expiration
-    | cluster-clusterd1-1\ .*\ Replica\ expired
+    | cluster-clusterd1-1\ .*\ replica\ expired
     # Emitted by tests employing explicit mz_panic()
     | forced\ panic
     # Emitted by broken_statements.slt in order to stop panic propagation, as 'forced panic' will unwantedly panic the `environmentd` thread.

src/adapter/src/coord.rs

@@ -186,7 +186,6 @@ use crate::error::AdapterError;
 use crate::explain::insights::PlanInsightsContext;
 use crate::explain::optimizer_trace::{DispatchGuard, OptimizerTrace};
 use crate::metrics::Metrics;
-use crate::optimize::dataflow_expiration::time_dependence;
 use crate::optimize::dataflows::{
     dataflow_import_id_bundle, ComputeInstanceSnapshot, DataflowBuilder,
 };
@@ -2667,9 +2666,7 @@ impl Coordinator {
                         (optimized_plan, global_lir_plan)
                     };
 
-                    let (mut physical_plan, metainfo) = global_lir_plan.unapply();
-                    physical_plan.time_dependence =
-                        time_dependence(self.catalog(), physical_plan.import_ids(), None);
+                    let (physical_plan, metainfo) = global_lir_plan.unapply();
                     let metainfo = {
                         // Pre-allocate a vector of transient GlobalIds for each notice.
                         let notice_ids = std::iter::repeat_with(|| self.allocate_transient_id())
@@ -2728,12 +2725,7 @@ impl Coordinator {
                         (optimized_plan, global_lir_plan)
                     };
 
-                    let (mut physical_plan, metainfo) = global_lir_plan.unapply();
-                    physical_plan.time_dependence = time_dependence(
-                        self.catalog(),
-                        physical_plan.import_ids(),
-                        mv.refresh_schedule.clone(),
-                    );
+                    let (physical_plan, metainfo) = global_lir_plan.unapply();
                     let metainfo = {
                         // Pre-allocate a vector of transient GlobalIds for each notice.
                         let notice_ids = std::iter::repeat_with(|| self.allocate_transient_id())
@@ -2762,12 +2754,8 @@ impl Coordinator {
                         .catalog()
                         .resolve_full_name(entry.name(), None)
                         .to_string();
-                    let (optimized_plan, mut physical_plan, metainfo) = self
+                    let (optimized_plan, physical_plan, metainfo) = self
                         .optimize_create_continual_task(ct, id, self.owned_catalog(), debug_name)?;
-                    // Filter our own id as it is not known yet.
-                    let ids = physical_plan.import_ids().filter(|x| *x != id);
-                    physical_plan.time_dependence = time_dependence(self.catalog(), ids, None);
-
                     let catalog = self.catalog_mut();
                     catalog.set_optimized_plan(id, optimized_plan);
                     catalog.set_physical_plan(id, physical_plan);
@@ -3104,7 +3092,7 @@ impl Coordinator {
                 catalog.expire().await;
             }
         }
-        .boxed_local()
+            .boxed_local()
     }
 
     /// Obtain a read-only Catalog reference.

src/adapter/src/coord/sequencer/inner/create_continual_task.rs

@@ -39,7 +39,6 @@ use crate::catalog;
 use crate::command::ExecuteResponse;
 use crate::coord::Coordinator;
 use crate::error::AdapterError;
-use crate::optimize::dataflow_expiration::time_dependence;
 use crate::optimize::dataflows::dataflow_import_id_bundle;
 use crate::optimize::{self, Optimize, OptimizerCatalog};
 use crate::session::Session;
@@ -126,10 +125,6 @@ impl Coordinator {
 
         let () = self
             .catalog_transact_with_side_effects(Some(session), ops, |coord| async {
-                // We're referencing ourselves, so filter out our ID.
-                let ids = physical_plan.import_ids().filter(|x| *x != sink_id);
-                physical_plan.time_dependence = time_dependence(coord.catalog(), ids, None);
-
                 let catalog = coord.catalog_mut();
                 catalog.set_optimized_plan(sink_id, optimized_plan);
                 catalog.set_physical_plan(sink_id, physical_plan.clone());

src/adapter/src/coord/sequencer/inner/create_index.rs

@@ -30,7 +30,6 @@ use crate::coord::{
 use crate::error::AdapterError;
 use crate::explain::explain_dataflow;
 use crate::explain::optimizer_trace::OptimizerTrace;
-use crate::optimize::dataflow_expiration::time_dependence;
 use crate::optimize::dataflows::dataflow_import_id_bundle;
 use crate::optimize::{self, Optimize};
 use crate::session::Session;
@@ -448,9 +447,6 @@ impl Coordinator {
         let transact_result = self
             .catalog_transact_with_side_effects(Some(session), ops, |coord| async {
                 let (mut df_desc, df_meta) = global_lir_plan.unapply();
-                df_desc.time_dependence =
-                    time_dependence(coord.catalog(), df_desc.import_ids(), None);
-
                 // Save plan structures.
                 coord
                     .catalog_mut()

src/adapter/src/coord/sequencer/inner/create_materialized_view.rs

@@ -45,7 +45,6 @@ use crate::error::AdapterError;
 use crate::explain::explain_dataflow;
 use crate::explain::explain_plan;
 use crate::explain::optimizer_trace::OptimizerTrace;
-use crate::optimize::dataflow_expiration::time_dependence;
 use crate::optimize::dataflows::dataflow_import_id_bundle;
 use crate::optimize::{self, Optimize};
 use crate::session::Session;
@@ -649,10 +648,6 @@ impl Coordinator {
             .catalog_transact_with_side_effects(Some(session), ops, |coord| async {
                 let output_desc = global_lir_plan.desc().clone();
                 let (mut df_desc, df_meta) = global_lir_plan.unapply();
-
-                df_desc.time_dependence =
-                    time_dependence(coord.catalog(), df_desc.import_ids(), refresh_schedule);
-
                 // Save plan structures.
                 coord
                     .catalog_mut()

src/adapter/src/coord/sequencer/inner/subscribe.rs

@@ -23,7 +23,6 @@ use crate::coord::{
     SubscribeStage, SubscribeTimestampOptimizeLir, TargetCluster,
 };
 use crate::error::AdapterError;
-use crate::optimize::dataflow_expiration::time_dependence;
 use crate::optimize::Optimize;
 use crate::session::{Session, TransactionOps};
 use crate::{optimize, AdapterNotice, ExecuteContext, TimelineContext};
@@ -348,8 +347,7 @@ impl Coordinator {
         };
         active_subscribe.initialize();
 
-        let (mut df_desc, df_meta) = global_lir_plan.unapply();
-        df_desc.time_dependence = time_dependence(self.catalog(), df_desc.import_ids(), None);
+        let (df_desc, df_meta) = global_lir_plan.unapply();
 
         // Emit notices.
         self.emit_optimizer_notices(ctx.session(), &df_meta.optimizer_notices);

src/adapter/src/optimize.rs

@@ -53,7 +53,6 @@
 //! repository.
 
 pub mod copy_to;
-pub mod dataflow_expiration;
 pub mod dataflows;
 pub mod index;
 pub mod materialized_view;

src/adapter/src/optimize/dataflow_expiration.rs

@@ -785,6 +785,7 @@ mod tests {
     use mz_storage_types::read_holds::ReadHoldError;
     use mz_storage_types::sources::SourceExportDataConfig;
     use mz_storage_types::sources::{GenericSourceConnection, SourceDesc};
+    use mz_storage_types::time_dependence::TimeDependence;
 
     use super::*;
 
@@ -947,6 +948,13 @@ mod tests {
             }
             Ok(holds)
         }
+
+        fn get_time_dependence(
+            &self,
+            _id: GlobalId,
+        ) -> Result<Option<TimeDependence>, StorageError<Self::Timestamp>> {
+            unimplemented!()
+        }
     }
 
     fn dataflow(