Initial refresh support

Author: Siddhartha Sahu

src/adapter/src/catalog/dataflow_expiration.rs

@@ -6,78 +6,32 @@
 //! Helper function for dataflow expiration checks.
 
 use crate::catalog::Catalog;
-use differential_dataflow::lattice::Lattice;
-use mz_catalog::memory::objects::CatalogItem;
-use mz_expr::CollectionPlan;
-use mz_repr::{GlobalId, Timestamp};
+use mz_compute_types::dataflows::{RefreshDep, RefreshDepIndex};
+use mz_repr::GlobalId;
 
 impl Catalog {
-    /// Whether the catalog entry `id` or any of its transitive dependencies is a materialized view
-    /// with a refresh schedule. Used to disable dataflow expiration if found.
-    pub(crate) fn item_has_transitive_refresh_schedule(&self, id: GlobalId) -> bool {
-        let test_has_transitive_refresh_schedule = |dep: GlobalId| -> bool {
-            if let Some(mv) = self.get_entry(&dep).materialized_view() {
-                return mv.refresh_schedule.is_some();
-            }
-            false
-        };
-        test_has_transitive_refresh_schedule(id)
-            || self
-                .state()
-                .transitive_uses(id)
-                .any(test_has_transitive_refresh_schedule)
-    }
     /// Recursive function.
-    pub(crate) fn compute_expiration_with_refresh(
+    pub(crate) fn get_refresh_dependencies(
         &self,
-        id: GlobalId,
-        replica_expiration: Timestamp,
-    ) -> Timestamp {
-        let entry = self.get_entry(&id);
-        // TODO: use a queue to deduplicate work.
-        // TODO: don't use Timestamp::MAX as the initial value.
-        match &entry.item {
-            CatalogItem::MaterializedView(mv) => {
-                let mut new_expiration = mv.raw_expr.depends_on().into_iter().fold(
-                    Timestamp::MAX,
-                    |new_expiration, dep| {
-                        // Pick the minimum refresh time of all dependencies.
-                        new_expiration
-                            .meet(&self.compute_expiration_with_refresh(dep, replica_expiration))
-                    },
-                );
-                if let Some(refresh_schedule) = &mv.refresh_schedule {
-                    if let Some(next_refresh) = refresh_schedule.round_up_timestamp(new_expiration)
-                    {
-                        new_expiration = next_refresh;
-                    }
-                }
-                new_expiration
-            }
-            CatalogItem::View(view) => view.raw_expr.depends_on().into_iter().fold(
-                Timestamp::MAX,
-                |new_expiration, dep| {
-                    new_expiration
-                        .meet(&self.compute_expiration_with_refresh(dep, replica_expiration))
-                },
-            ),
-            CatalogItem::Index(index) => {
-                self.compute_expiration_with_refresh(index.on, replica_expiration)
-            }
-            CatalogItem::Func(_) => {
-                todo!()
-            }
-            CatalogItem::ContinualTask(_) => {
-                todo!()
+        deps: impl Iterator<Item = GlobalId>,
+        deps_tree: &mut Vec<RefreshDep>,
+    ) -> Option<RefreshDepIndex> {
+        let start = deps_tree.len();
+        for dep in deps {
+            let entry = self.get_entry(&dep);
+            let refresh_dep_index =
+                self.get_refresh_dependencies(entry.uses().into_iter(), deps_tree);
+            let refresh_schedule = entry
+                .materialized_view()
+                .and_then(|mv| mv.refresh_schedule.clone());
+            if refresh_dep_index.is_some() || refresh_schedule.is_some() {
+                deps_tree.push(RefreshDep {
+                    refresh_dep_index,
+                    refresh_schedule,
+                });
             }
-            // These will not have a transitive dependency on REFRESH.
-            CatalogItem::Table(_)
-            | CatalogItem::Source(_)
-            | CatalogItem::Sink(_)
-            | CatalogItem::Log(_)
-            | CatalogItem::Type(_)
-            | CatalogItem::Secret(_)
-            | CatalogItem::Connection(_) => replica_expiration,
         }
+        let end = deps_tree.len();
+        (end > start).then_some(RefreshDepIndex { start, end })
     }
 }

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

@@ -442,7 +442,6 @@ impl Coordinator {
 
         // Collect properties for `DataflowExpirationDesc`.
         let transitive_upper = self.least_valid_write(&id_bundle);
-        let has_transitive_refresh_schedule = self.catalog.item_has_transitive_refresh_schedule(on);
 
         // Pre-allocate a vector of transient GlobalIds for each notice.
         let notice_ids = std::iter::repeat_with(|| self.allocate_transient_id())
@@ -477,9 +476,6 @@ impl Coordinator {
                 df_desc.set_as_of(since);
 
                 df_desc.dataflow_expiration_desc.transitive_upper = Some(transitive_upper);
-                df_desc
-                    .dataflow_expiration_desc
-                    .has_transitive_refresh_schedule = has_transitive_refresh_schedule;
 
                 coord
                     .ship_dataflow_and_notice_builtin_table_updates(

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

@@ -21,7 +21,7 @@ use mz_repr::explain::{ExprHumanizerExt, TransientItem};
 use mz_repr::optimize::OptimizerFeatures;
 use mz_repr::optimize::OverrideFrom;
 use mz_repr::refresh_schedule::RefreshSchedule;
-use mz_repr::{Datum, GlobalId, Row, Timestamp};
+use mz_repr::{Datum, GlobalId, Row};
 use mz_sql::ast::ExplainStage;
 use mz_sql::catalog::CatalogError;
 use mz_sql::names::ResolvedIds;
@@ -575,33 +575,10 @@ impl Coordinator {
 
         // Collect properties for `DataflowExpirationDesc`.
         let transitive_upper = self.least_valid_write(&id_bundle);
-        let has_transitive_refresh_schedule = refresh_schedule.is_some()
-            || raw_expr
-                .depends_on()
-                .into_iter()
-                .any(|id| self.catalog.item_has_transitive_refresh_schedule(id));
-
-        // Collect properties for `DataflowExpirationDesc`.
-        let transitive_upper = self.least_valid_write(&id_bundle);
-        let replica_expiration = Timestamp::default(); // TODO: Can we get replica_expiration information here somehow?
-
-        let mut expiration_with_refresh =
-            raw_expr
-                .depends_on()
-                .into_iter()
-                .fold(Timestamp::MAX, |new_expiration, id| {
-                    new_expiration.meet(
-                        &self
-                            .catalog
-                            .compute_expiration_with_refresh(id, replica_expiration),
-                    )
-                });
-        if let Some(refresh_schedule) = &refresh_schedule {
-            if let Some(next_refresh) = refresh_schedule.round_up_timestamp(expiration_with_refresh)
-            {
-                expiration_with_refresh = next_refresh;
-            }
-        }
+        let mut refresh_deps = Vec::new();
+        let refresh_deps_index = self
+            .catalog
+            .get_refresh_dependencies(raw_expr.depends_on().into_iter(), &mut refresh_deps);
 
         let read_holds_owned;
         let read_holds = if let Some(txn_reads) = self.txn_read_holds.get(session.conn_id()) {
@@ -697,9 +674,8 @@ impl Coordinator {
                 df_desc.until = until;
 
                 df_desc.dataflow_expiration_desc.transitive_upper = Some(transitive_upper);
-                df_desc
-                    .dataflow_expiration_desc
-                    .has_transitive_refresh_schedule = has_transitive_refresh_schedule;
+                df_desc.dataflow_expiration_desc.refresh_deps = refresh_deps;
+                df_desc.dataflow_expiration_desc.refresh_deps_index = refresh_deps_index;
 
                 let storage_metadata = coord.catalog.state().storage_metadata();
 

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

@@ -324,7 +324,6 @@ impl Coordinator {
             cluster_id,
             plan:
                 plan::SubscribePlan {
-                    from,
                     copy_to,
                     emit_progress,
                     output,
@@ -339,10 +338,6 @@ impl Coordinator {
 
         // Collect properties for `DataflowExpirationDesc`.
         let transitive_upper = self.least_valid_write(&id_bundle);
-        let has_transitive_refresh_schedule = from
-            .depends_on()
-            .into_iter()
-            .any(|id| self.catalog.item_has_transitive_refresh_schedule(id));
 
         let sink_id = global_lir_plan.sink_id();
 
@@ -366,9 +361,6 @@ impl Coordinator {
         let (mut df_desc, df_meta) = global_lir_plan.unapply();
 
         df_desc.dataflow_expiration_desc.transitive_upper = Some(transitive_upper);
-        df_desc
-            .dataflow_expiration_desc
-            .has_transitive_refresh_schedule = has_transitive_refresh_schedule;
 
         // Emit notices.
         self.emit_optimizer_notices(ctx.session(), &df_meta.optimizer_notices);

src/compute-types/src/dataflows.proto

@@ -49,9 +49,18 @@ message ProtoDataflowDescription {
   }
 
   message ProtoDataflowExpirationDesc {
+    message ProtoRefreshDepIndex {
+      uint64 start = 1;
+      uint64 end = 2;
+    }
+    message ProtoRefreshDep {
+      optional ProtoRefreshDepIndex refresh_dep_index = 1;
+      optional mz_repr.refresh_schedule.ProtoRefreshSchedule refresh_schedule = 2;
+    }
     optional mz_repr.antichain.ProtoU64Antichain transitive_upper = 1;
-    bool has_transitive_refresh_schedule = 2;
     bool is_timeline_epoch_ms = 3;
+    repeated ProtoRefreshDep refresh_deps = 4;
+    optional ProtoRefreshDepIndex refresh_deps_index = 5;
   }
 
   repeated ProtoSourceImport source_imports = 1;