Improved replica and dataflow expiration

misc/python/materialize/cli/ci_annotate_errors.py

@@ -125,7 +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\ .*\ has\ exceeded\ expiration
+    | 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/catalog.rs

@@ -98,7 +98,6 @@ pub(crate) mod consistency;
 mod migrate;
 
 mod apply;
-mod dataflow_expiration;
 mod open;
 mod state;
 mod transact;

src/adapter/src/coord.rs

@@ -66,16 +66,6 @@
 //! ```
 //!
 
-use anyhow::Context;
-use chrono::{DateTime, Utc};
-use ipnet::IpNet;
-use mz_adapter_types::dyncfgs::WITH_0DT_DEPLOYMENT_CAUGHT_UP_CHECK_INTERVAL;
-use mz_compute_client::as_of_selection;
-use mz_ore::channel::trigger::Trigger;
-use mz_ore::url::SensitiveUrl;
-use mz_sql::names::{ResolvedIds, SchemaSpecifier};
-use mz_sql::session::user::User;
-use mz_storage_types::read_holds::ReadHold;
 use std::borrow::Cow;
 use std::collections::{BTreeMap, BTreeSet, VecDeque};
 use std::fmt;
@@ -87,17 +77,20 @@ use std::sync::LazyLock;
 use std::sync::{Arc, Mutex};
 use std::thread;
 use std::time::{Duration, Instant};
-use thiserror::Error;
 
+use anyhow::Context;
+use chrono::{DateTime, Utc};
 use derivative::Derivative;
 use differential_dataflow::lattice::Lattice;
 use fail::fail_point;
 use futures::future::{BoxFuture, FutureExt, LocalBoxFuture};
 use futures::StreamExt;
 use http::Uri;
+use ipnet::IpNet;
 use itertools::{Either, Itertools};
 use mz_adapter_types::compaction::CompactionWindow;
 use mz_adapter_types::connection::ConnectionId;
+use mz_adapter_types::dyncfgs::WITH_0DT_DEPLOYMENT_CAUGHT_UP_CHECK_INTERVAL;
 use mz_build_info::BuildInfo;
 use mz_catalog::builtin::{BUILTINS, BUILTINS_STATIC, MZ_STORAGE_USAGE_BY_SHARD};
 use mz_catalog::config::{AwsPrincipalContext, BuiltinItemMigrationConfig, ClusterReplicaSizeMap};
@@ -107,6 +100,7 @@ use mz_catalog::memory::objects::{
     DataSourceDesc, TableDataSource,
 };
 use mz_cloud_resources::{CloudResourceController, VpcEndpointConfig, VpcEndpointEvent};
+use mz_compute_client::as_of_selection;
 use mz_compute_client::controller::error::InstanceMissing;
 use mz_compute_types::dataflows::DataflowDescription;
 use mz_compute_types::plan::Plan;
@@ -117,12 +111,14 @@ use mz_controller_types::{ClusterId, ReplicaId, WatchSetId};
 use mz_expr::{MapFilterProject, OptimizedMirRelationExpr, RowSetFinishing};
 use mz_orchestrator::{OfflineReason, ServiceProcessMetrics};
 use mz_ore::cast::{CastFrom, CastLossy};
+use mz_ore::channel::trigger::Trigger;
 use mz_ore::future::TimeoutError;
 use mz_ore::metrics::MetricsRegistry;
 use mz_ore::now::{EpochMillis, NowFn};
 use mz_ore::task::{spawn, JoinHandle};
 use mz_ore::thread::JoinHandleExt;
 use mz_ore::tracing::{OpenTelemetryContext, TracingHandle};
+use mz_ore::url::SensitiveUrl;
 use mz_ore::vec::VecExt;
 use mz_ore::{assert_none, instrument, soft_assert_or_log, soft_panic_or_log, stack};
 use mz_persist_client::usage::{ShardsUsageReferenced, StorageUsageClient};
@@ -134,11 +130,13 @@ use mz_secrets::cache::CachingSecretsReader;
 use mz_secrets::{SecretsController, SecretsReader};
 use mz_sql::ast::{Raw, Statement};
 use mz_sql::catalog::{CatalogCluster, EnvironmentId};
+use mz_sql::names::{ResolvedIds, SchemaSpecifier};
 use mz_sql::optimizer_metrics::OptimizerMetrics;
 use mz_sql::plan::{
     self, AlterSinkPlan, ConnectionDetails, CreateConnectionPlan, OnTimeoutAction, Params,
     QueryWhen,
 };
+use mz_sql::session::user::User;
 use mz_sql::session::vars::{ConnectionCounter, SystemVars};
 use mz_sql_parser::ast::display::AstDisplay;
 use mz_sql_parser::ast::ExplainStage;
@@ -147,6 +145,7 @@ use mz_storage_client::controller::{CollectionDescription, DataSource};
 use mz_storage_types::connections::inline::{IntoInlineConnection, ReferencedConnection};
 use mz_storage_types::connections::Connection as StorageConnection;
 use mz_storage_types::connections::ConnectionContext;
+use mz_storage_types::read_holds::ReadHold;
 use mz_storage_types::sinks::S3SinkFormat;
 use mz_storage_types::sources::Timeline;
 use mz_timestamp_oracle::postgres_oracle::{
@@ -156,6 +155,7 @@ use mz_timestamp_oracle::WriteTimestamp;
 use mz_transform::dataflow::DataflowMetainfo;
 use opentelemetry::trace::TraceContextExt;
 use serde::Serialize;
+use thiserror::Error;
 use timely::progress::{Antichain, Timestamp as _};
 use tokio::runtime::Handle as TokioHandle;
 use tokio::select;
@@ -178,13 +178,15 @@ use crate::coord::cluster_scheduling::SchedulingDecision;
 use crate::coord::id_bundle::CollectionIdBundle;
 use crate::coord::introspection::IntrospectionSubscribe;
 use crate::coord::peek::PendingPeek;
+use crate::coord::statement_logging::{StatementLogging, StatementLoggingId};
 use crate::coord::timeline::{TimelineContext, TimelineState};
 use crate::coord::timestamp_selection::{TimestampContext, TimestampDetermination};
 use crate::coord::validity::PlanValidity;
 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,
 };
@@ -195,8 +197,6 @@ use crate::util::{ClientTransmitter, ResultExt};
 use crate::webhook::{WebhookAppenderInvalidator, WebhookConcurrencyLimiter};
 use crate::{flags, AdapterNotice, ReadHolds};
 
-use self::statement_logging::{StatementLogging, StatementLoggingId};
-
 pub(crate) mod id_bundle;
 pub(crate) mod in_memory_oracle;
 pub(crate) mod peek;
@@ -743,10 +743,6 @@ pub struct CreateMaterializedViewOptimize {
     /// An optional context set iff the state machine is initiated from
     /// sequencing an EXPLAIN for this statement.
     explain_ctx: ExplainContext,
-    /// Whether the timeline is [`mz_storage_types::sources::Timeline::EpochMilliseconds`].
-    ///
-    /// Used to determine if it is safe to enable dataflow expiration.
-    is_timeline_epoch_ms: bool,
 }
 
 #[derive(Debug)]
@@ -2587,15 +2583,8 @@ impl Coordinator {
                 .catalog()
                 .resolve_full_name(entry.name(), None)
                 .to_string();
-            let is_timeline_epoch_ms = self.get_timeline_context(*id).is_timeline_epoch_ms();
             let (_optimized_plan, physical_plan, _metainfo) = self
-                .optimize_create_continual_task(
-                    &ct,
-                    *id,
-                    self.owned_catalog(),
-                    debug_name,
-                    is_timeline_epoch_ms,
-                )
+                .optimize_create_continual_task(&ct, *id, self.owned_catalog(), debug_name)
                 .expect("builtin CT should optimize successfully");
 
             // Determine an as of for the new continual task.
@@ -2638,7 +2627,6 @@ impl Coordinator {
 
         for entry in ordered_catalog_entries {
             let id = entry.id();
-            let is_timeline_epoch_ms = self.get_timeline_context(id).is_timeline_epoch_ms();
             match entry.item() {
                 CatalogItem::Index(idx) => {
                     // Collect optimizer parameters.
@@ -2669,7 +2657,6 @@ impl Coordinator {
                             entry.name().clone(),
                             idx.on,
                             idx.keys.to_vec(),
-                            is_timeline_epoch_ms,
                         );
                         let global_mir_plan = optimizer.optimize(index_plan)?;
                         let optimized_plan = global_mir_plan.df_desc().clone();
@@ -2680,7 +2667,9 @@ impl Coordinator {
                         (optimized_plan, global_lir_plan)
                     };
 
-                    let (physical_plan, metainfo) = global_lir_plan.unapply();
+                    let (mut physical_plan, metainfo) = global_lir_plan.unapply();
+                    physical_plan.time_dependence =
+                        time_dependence(self.catalog(), physical_plan.import_ids(), None);
                     let metainfo = {
                         // Pre-allocate a vector of transient GlobalIds for each notice.
                         let notice_ids = std::iter::repeat_with(|| self.allocate_transient_id())
@@ -2725,7 +2714,6 @@ impl Coordinator {
                             debug_name,
                             optimizer_config.clone(),
                             self.optimizer_metrics(),
-                            is_timeline_epoch_ms,
                             force_non_monotonic,
                         );
 
@@ -2740,7 +2728,12 @@ impl Coordinator {
                         (optimized_plan, global_lir_plan)
                     };
 
-                    let (physical_plan, metainfo) = global_lir_plan.unapply();
+                    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 metainfo = {
                         // Pre-allocate a vector of transient GlobalIds for each notice.
                         let notice_ids = std::iter::repeat_with(|| self.allocate_transient_id())
@@ -2770,13 +2763,7 @@ impl Coordinator {
                         .resolve_full_name(entry.name(), None)
                         .to_string();
                     let (optimized_plan, physical_plan, metainfo) = self
-                        .optimize_create_continual_task(
-                            ct,
-                            id,
-                            self.owned_catalog(),
-                            debug_name,
-                            is_timeline_epoch_ms,
-                        )?;
+                        .optimize_create_continual_task(ct, id, self.owned_catalog(), debug_name)?;
 
                     let catalog = self.catalog_mut();
                     catalog.set_optimized_plan(id, optimized_plan);

src/adapter/src/coord/introspection.rs

@@ -255,11 +255,7 @@ impl Coordinator {
         let read_holds = self.acquire_read_holds(&id_bundle);
         let as_of = read_holds.least_valid_read();
 
-        // Introspection subscribes only read from system collections, which are always in
-        // the `EpochMilliseconds` timeline.
-        let is_timeline_epoch_ms = true;
-
-        let global_mir_plan = global_mir_plan.resolve(as_of, is_timeline_epoch_ms);
+        let global_mir_plan = global_mir_plan.resolve(as_of);
 
         let span = Span::current();
         Ok(StageResult::Handle(mz_ore::task::spawn_blocking(

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

@@ -84,10 +84,6 @@ impl Coordinator {
             },
         };
 
-        let is_timeline_epoch_ms = self
-            .validate_timeline_context(resolved_ids.0.clone())?
-            .is_timeline_epoch_ms();
-
         // Construct the CatalogItem for this CT and optimize it.
         let mut item = crate::continual_task::ct_item_from_plan(plan, sink_id, resolved_ids)?;
         let full_name = bootstrap_catalog.resolve_full_name(&name, Some(session.conn_id()));
@@ -96,7 +92,6 @@ impl Coordinator {
             sink_id,
             Arc::new(bootstrap_catalog),
             full_name.to_string(),
-            is_timeline_epoch_ms,
         )?;
 
         // Timestamp selection
@@ -166,7 +161,6 @@ impl Coordinator {
         output_id: GlobalId,
         catalog: Arc<dyn OptimizerCatalog>,
         debug_name: String,
-        is_timeline_epoch_ms: bool,
     ) -> Result<
         (
             DataflowDescription<OptimizedMirRelationExpr>,
@@ -200,7 +194,6 @@ impl Coordinator {
             debug_name,
             optimizer_config,
             self.optimizer_metrics(),
-            is_timeline_epoch_ms,
             force_non_monotonic,
         );
 

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

@@ -30,6 +30,7 @@ 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;
@@ -295,7 +296,7 @@ impl Coordinator {
         }: CreateIndexOptimize,
     ) -> Result<StageResult<Box<CreateIndexStage>>, AdapterError> {
         let plan::CreateIndexPlan {
-            index: plan::Index { cluster_id, on, .. },
+            index: plan::Index { cluster_id, .. },
             ..
         } = &plan;
 
@@ -321,7 +322,6 @@ impl Coordinator {
             self.optimizer_metrics(),
         );
         let span = Span::current();
-        let is_timeline_epoch_ms = self.get_timeline_context(*on).is_timeline_epoch_ms();
         Ok(StageResult::Handle(mz_ore::task::spawn_blocking(
             || "optimize create index",
             move || {
@@ -333,7 +333,7 @@ impl Coordinator {
                     let _dispatch_guard = explain_ctx.dispatch_guard();
 
                     let index_plan =
-                        optimize::index::Index::new(plan.name.clone(), plan.index.on, plan.index.keys.clone(), is_timeline_epoch_ms);
+                        optimize::index::Index::new(plan.name.clone(), plan.index.on, plan.index.keys.clone());
 
                     // MIR ⇒ MIR optimization (global)
                     let global_mir_plan = optimizer.catch_unwind_optimize(index_plan)?;
@@ -440,26 +440,24 @@ impl Coordinator {
             owner_id: *self.catalog().get_entry(&on).owner_id(),
         }];
 
-        // 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())
             .take(global_lir_plan.df_meta().optimizer_notices.len())
             .collect::<Vec<_>>();
 
         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()
                     .set_optimized_plan(exported_index_id, global_mir_plan.df_desc().clone());
                 coord
                     .catalog_mut()
-                    .set_physical_plan(exported_index_id, global_lir_plan.df_desc().clone());
-
-                let (mut df_desc, df_meta) = global_lir_plan.unapply();
+                    .set_physical_plan(exported_index_id, df_desc.clone());
 
                 let notice_builtin_updates_fut = coord
                     .process_dataflow_metainfo(df_meta, exported_index_id, session, notice_ids)
@@ -476,11 +474,6 @@ impl Coordinator {
                 let since = coord.least_valid_read(&read_holds);
                 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(
                         df_desc,