adapter: collect durable object hydration history

Adds the collector that fills `mz_internal.mz_object_hydration_history`, and the
background read-then-write plumbing it needs.

A sweep visits one managed user replica per interval and installs an internal
subscribe on that replica which aggregates complete per-worker episodes, maps
runtime export ids to catalog objects, anti-joins against the history table, and
writes the missing rows through the timestamped OCC path. Retention retracts a
bounded batch per sweep on the catalog server, independently of whether that
sweep's collection succeeded.

Reading the target table inside the subscribe is what makes the write idempotent
across concurrent environmentd processes. Two collectors that compute the same
row race for one write timestamp, and the loser observes the winner's append
through its own subscribe and finds nothing left to write.

The read-then-write path grows an `RtwCaller`, because replica pinning,
dependency validation, and write cancellation all differ for a background caller
and have to move together. Background subscribes are owned by the coordinator
rather than a session, so they write no `mz_subscriptions` row, and background
writes take no OCC permit: the sweep has no statement timeout, and its subscribe
must first hydrate a dataflow on a user replica, so holding a permit would let it
stall user DML for as long as that takes.

Collection is off by default and enabled in the mzcompose configuration.
`test/testdrive/hydration-status.td` covers lifecycle ordering, several replicas,
deduplication across sweeps, replica removal, retention, and the deliberate
absence of an object that never reports a completion time.
`test/restart/mzcompose.py` covers survival of an environmentd restart without
duplication, on a two-worker replica.

Closes: SQL-644

Author: aljoscha

doc/user/data/metrics.yml

@@ -97,7 +97,7 @@ metrics:
   source: src/adapter/src/metrics.rs
   visibility: internal
 - name: mz_active_internal_subscribes
-  help: The number of active internal subscribes, which serve frontend-sequenced read-then-write.
+  help: The number of active internal subscribes used by read-then-write operations and background maintenance.
   labels:
   - session_type
   source: src/adapter/src/metrics.rs

misc/python/materialize/mzcompose/__init__.py

@@ -137,6 +137,9 @@ def get_minimal_system_parameters(
     if version < MzVersion.parse_mz("v26.25.0-dev"):
         config["enable_multi_replica_sources"] = "true"
 
+    if version >= MzVersion.parse_mz("v26.39.0-dev"):
+        config["hydration_history_collection_interval"] = "60s"
+
     if sanitizer_enabled():
         config["with_0dt_deployment_max_wait"] = "18000s"
 
@@ -461,6 +464,9 @@ def get_variable_system_parameters(
         VariableSystemParameter(
             "arrangement_size_history_retention_period", "7d", ["1min", "1h", "7d"]
         ),
+        VariableSystemParameter(
+            "hydration_history_retention_period", "30d", ["1min", "1h", "30d"]
+        ),
         VariableSystemParameter(
             "persist_validate_part_bounds_on_read", "false", ["true", "false"]
         ),

misc/python/materialize/parallel_workload/action.py

@@ -2949,6 +2949,16 @@ def __init__(
             "'1h'",
             "'7d'",
         ]
+        self.flags_with_values["hydration_history_collection_interval"] = [
+            "'0s'",
+            "'1s'",
+            "'1min'",
+        ]
+        self.flags_with_values["hydration_history_retention_period"] = [
+            "'1min'",
+            "'1h'",
+            "'30d'",
+        ]
         # Keep these generous: a tight timeout would abort the oracle's own
         # queries (they are retried, but it adds noise). "0s" leaves it unset.
         self.flags_with_values["pg_timestamp_oracle_statement_timeout"] = [

src/adapter-types/src/dyncfgs.rs

@@ -344,6 +344,20 @@ pub const ARRANGEMENT_SIZE_HISTORY_RETENTION_PERIOD: Config<Duration> = Config::
     "How long to retain rows in mz_internal.mz_object_arrangement_size_history.",
 );
 
+/// How often to sweep replicas for completed object hydration episodes.
+pub const HYDRATION_HISTORY_COLLECTION_INTERVAL: Config<Duration> = Config::new(
+    "hydration_history_collection_interval",
+    Duration::ZERO,
+    "How often to record completed object hydration episodes. A zero duration disables collection.",
+);
+
+/// How long to retain completed object hydration episodes.
+pub const HYDRATION_HISTORY_RETENTION_PERIOD: Config<Duration> = Config::new(
+    "hydration_history_retention_period",
+    Duration::from_hours(30 * 24),
+    "How long to retain rows in mz_internal.mz_object_hydration_history.",
+);
+
 /// How frequently the catalog `*_info` metrics (`mz_object_info`,
 /// `mz_cluster_info`, …) are reconciled with the catalog. A zero duration
 /// disables reconciliation.
@@ -470,6 +484,8 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&CONSOLE_OIDC_SCOPES)
         .add(&ARRANGEMENT_SIZE_HISTORY_COLLECTION_INTERVAL)
         .add(&ARRANGEMENT_SIZE_HISTORY_RETENTION_PERIOD)
+        .add(&HYDRATION_HISTORY_COLLECTION_INTERVAL)
+        .add(&HYDRATION_HISTORY_RETENTION_PERIOD)
         .add(&CATALOG_INFO_METRICS_RECONCILE_INTERVAL)
         .add(&PG_TIMESTAMP_ORACLE_STATEMENT_TIMEOUT)
         .add(&FRONTEND_READ_THEN_WRITE)

src/adapter/src/active_compute_sink.rs

@@ -50,10 +50,10 @@ impl ActiveComputeSink {
     }
 
     /// Reports the ID of the connection which created the sink.
-    pub fn connection_id(&self) -> &ConnectionId {
+    pub fn connection_id(&self) -> Option<&ConnectionId> {
         match &self {
-            ActiveComputeSink::Subscribe(subscribe) => &subscribe.conn_id,
-            ActiveComputeSink::CopyTo(copy_to) => &copy_to.conn_id,
+            ActiveComputeSink::Subscribe(subscribe) => subscribe.connection_id(),
+            ActiveComputeSink::CopyTo(copy_to) => Some(&copy_to.conn_id),
         }
     }
 
@@ -90,13 +90,26 @@ pub enum ActiveComputeSinkRetireReason {
     DependencyDropped(DroppedDependency),
 }
 
+/// Ownership and cleanup scope of an active subscribe.
+#[derive(Debug)]
+pub enum ActiveSubscribeOwner {
+    /// The subscribe belongs to a SQL session.
+    Session {
+        conn_id: ConnectionId,
+        session_uuid: Uuid,
+    },
+    /// The subscribe belongs to a coordinator background task.
+    ///
+    /// Always `internal`, since there is no session to attribute a
+    /// `mz_subscriptions` row to.
+    Background,
+}
+
 /// A description of an active subscribe from coord's perspective
 #[derive(Debug)]
 pub struct ActiveSubscribe {
-    /// The ID of the connection which created the subscribe.
-    pub conn_id: ConnectionId,
-    /// The UUID of the session which created the subscribe.
-    pub session_uuid: Uuid,
+    /// The owner responsible for retiring the subscribe.
+    pub owner: ActiveSubscribeOwner,
     /// The ID of the cluster on which the subscribe is running.
     pub cluster_id: ClusterId,
     /// The IDs of the objects on which the subscribe depends.
@@ -121,6 +134,33 @@ pub struct ActiveSubscribe {
 }
 
 impl ActiveSubscribe {
+    /// The session uuid for this subscribe's `mz_subscriptions` row, or `None`
+    /// if it does not appear there.
+    pub fn introspection_session_uuid(&self) -> Option<Uuid> {
+        match &self.owner {
+            ActiveSubscribeOwner::Session { session_uuid, .. } if !self.internal => {
+                Some(*session_uuid)
+            }
+            _ => None,
+        }
+    }
+
+    /// Returns the owning connection, if this is a session subscribe.
+    pub fn connection_id(&self) -> Option<&ConnectionId> {
+        match &self.owner {
+            ActiveSubscribeOwner::Session { conn_id, .. } => Some(conn_id),
+            ActiveSubscribeOwner::Background => None,
+        }
+    }
+
+    /// Returns the owning session UUID, if this is a session subscribe.
+    pub fn session_uuid(&self) -> Option<Uuid> {
+        match self.owner {
+            ActiveSubscribeOwner::Session { session_uuid, .. } => Some(session_uuid),
+            ActiveSubscribeOwner::Background => None,
+        }
+    }
+
     /// Initializes the subscription.
     ///
     /// This method must be called exactly once, after constructing an

src/adapter/src/catalog/builtin_table_updates.rs

@@ -51,6 +51,7 @@ use mz_sql::names::SchemaSpecifier;
 use mz_sql_parser::ast::display::AstDisplay;
 use mz_storage_client::client::TableData;
 use smallvec::smallvec;
+use uuid::Uuid;
 
 // DO NOT add any more imports from `crate` outside of `crate::catalog`.
 use crate::active_compute_sink::ActiveSubscribe;
@@ -926,12 +927,13 @@ impl CatalogState {
         &self,
         id: GlobalId,
         subscribe: &ActiveSubscribe,
+        session_uuid: Uuid,
         diff: Diff,
     ) -> BuiltinTableUpdate<&'static BuiltinTable> {
         let mut row = Row::default();
         let mut packer = row.packer();
         packer.push(Datum::String(&id.to_string()));
-        packer.push(Datum::Uuid(subscribe.session_uuid));
+        packer.push(Datum::Uuid(session_uuid));
         packer.push(Datum::String(&subscribe.cluster_id.to_string()));
 
         let start_dt = mz_ore::now::to_datetime(subscribe.start_time);

src/adapter/src/catalog/state.rs

@@ -3127,7 +3127,9 @@ mod tests {
     #[mz_ore::test(tokio::test)]
     #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `TLS_client_method`
     async fn validate_read_then_write_deep_chain_no_stack_overflow() {
-        use crate::coord::read_then_write::validate_read_then_write_dependencies;
+        use crate::coord::read_then_write::{
+            DependencyPolicy, validate_read_then_write_dependencies,
+        };
 
         Catalog::with_debug(|mut catalog| async move {
             // Deep enough that the previous recursive implementation overflowed
@@ -3142,6 +3144,7 @@ mod tests {
                 &catalog,
                 [CatalogItemId::User(BASE)],
                 usize::MAX,
+                DependencyPolicy::UserDml,
             )
             .expect("deep chain of user views is valid for read-then-write");
 
@@ -3156,7 +3159,9 @@ mod tests {
     #[mz_ore::test(tokio::test)]
     #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `TLS_client_method`
     async fn validate_read_then_write_dependency_limit() {
-        use crate::coord::read_then_write::validate_read_then_write_dependencies;
+        use crate::coord::read_then_write::{
+            DependencyPolicy, validate_read_then_write_dependencies,
+        };
         use crate::error::AdapterError;
 
         Catalog::with_debug(|mut catalog| async move {
@@ -3168,14 +3173,20 @@ mod tests {
             const OBJECTS: usize = DEPTH + 1;
 
             // Exactly at the limit is allowed.
-            validate_read_then_write_dependencies(&catalog, [CatalogItemId::User(BASE)], OBJECTS)
-                .expect("chain at the limit is valid");
+            validate_read_then_write_dependencies(
+                &catalog,
+                [CatalogItemId::User(BASE)],
+                OBJECTS,
+                DependencyPolicy::UserDml,
+            )
+            .expect("chain at the limit is valid");
 
             // One below the limit is rejected with a clean error.
             let err = validate_read_then_write_dependencies(
                 &catalog,
                 [CatalogItemId::User(BASE)],
                 OBJECTS - 1,
+                DependencyPolicy::UserDml,
             )
             .expect_err("chain over the limit is rejected");
             assert!(matches!(

src/adapter/src/client.rs

@@ -63,6 +63,7 @@ use crate::command::{
     SASLChallengeResponse, SASLVerifyProofResponse, SuperuserAttribute,
 };
 use crate::config::{ScopedParameters, ScopedParametersScope, SystemParameterFrontend};
+use crate::coord::read_then_write::DependencyPolicy;
 use crate::coord::{Coordinator, ExecuteContextGuard};
 use crate::error::AdapterError;
 use crate::frontend_read_then_write::{
@@ -72,7 +73,7 @@ use crate::frontend_read_then_write::{
 use crate::metrics::Metrics;
 use crate::optimize::dataflows::{EvalTime, ExprPrepOneShot};
 use crate::optimize::{self, Optimize, OptimizerError};
-use crate::peek_client::{ExecutionLogging, TakeOver};
+use crate::peek_client::{CoordinatorClient, ExecutionLogging, TakeOver};
 use crate::session::{
     EndTransactionAction, PreparedStatement, Session, SessionConfig, StateRevision, TransactionId,
     TransactionStatus,
@@ -313,7 +314,7 @@ impl Client {
         } = response;
 
         let peek_client = PeekClient::new(
-            self.clone(),
+            CoordinatorClient::Session(self.clone()),
             &catalog,
             storage_collections,
             transient_id_gen,
@@ -1973,7 +1974,7 @@ impl SessionClient {
                         "calls to mz_now in write statements",
                     ));
                 }
-                validate_selection_dependencies(&catalog, &depends_on)?;
+                validate_selection_dependencies(&catalog, &depends_on, DependencyPolicy::UserDml)?;
                 return Err(prohibited_in_transaction(&stmt));
             }
         }

src/adapter/src/command.rs

@@ -45,6 +45,7 @@ use mz_timestamp_oracle::TimestampOracle;
 use tokio::sync::{Semaphore, mpsc, oneshot, watch};
 use uuid::Uuid;
 
+use crate::active_compute_sink::ActiveSubscribeOwner;
 use crate::catalog::Catalog;
 use crate::config::{ScopedParameters, ScopedParametersScope, SystemParameterFrontend};
 use crate::coord::appends::{BuiltinTableAppendNotify, WriteResult};
@@ -419,8 +420,7 @@ pub enum Command {
         as_of: mz_repr::Timestamp,
         arity: usize,
         sink_id: GlobalId,
-        conn_id: ConnectionId,
-        session_uuid: Uuid,
+        owner: ActiveSubscribeOwner,
         start_time: mz_ore::now::EpochMillis,
         read_holds: ReadHolds,
         tx: oneshot::Sender<Result<mpsc::UnboundedReceiver<PeekResponseUnary>, AdapterError>>,
@@ -437,14 +437,10 @@ pub enum Command {
     ///   including read-only, a changed target and cancellation, is reported the
     ///   same way in both modes.
     AttemptWrite {
-        /// Connection originating the write. Used so the coordinator can
-        /// cancel this pending write if the connection is cancelled before
-        /// the write commits.
-        conn_id: ConnectionId,
+        attempt: WriteAttempt,
         target_id: CatalogItemId,
         target_global_id: GlobalId,
         diffs: Vec<(Row, Diff)>,
-        write_ts: Option<mz_repr::Timestamp>,
         tx: oneshot::Sender<WriteResult>,
     },
 
@@ -455,6 +451,24 @@ pub enum Command {
     },
 }
 
+/// Who a read-then-write commits on behalf of, and how its timestamp is chosen.
+///
+/// Group commit picking the timestamp requires a connection to answer through,
+/// so that combination is only reachable from a session.
+#[derive(Debug)]
+pub enum WriteAttempt {
+    /// A session's write, cancelled with `conn_id` if the connection goes away
+    /// before it commits. A `write_ts` of `None` lets group commit pick the
+    /// timestamp, which then cannot be reported as passed.
+    Session {
+        conn_id: ConnectionId,
+        write_ts: Option<mz_repr::Timestamp>,
+    },
+    /// Coordinator background work. There is no connection to cancel with, so
+    /// the caller names the timestamp and handles `TimestampPassed` itself.
+    Background { write_ts: mz_repr::Timestamp },
+}
+
 impl Command {
     pub fn session(&self) -> Option<&Session> {
         match self {

src/adapter/src/coord.rs

@@ -130,7 +130,7 @@ 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::{JoinHandle, spawn};
+use mz_ore::task::{AbortOnDropHandle, JoinHandle, spawn};
 use mz_ore::thread::JoinHandleExt;
 use mz_ore::tracing::{OpenTelemetryContext, TracingHandle};
 use mz_ore::url::SensitiveUrl;
@@ -236,6 +236,7 @@ mod caught_up;
 mod command_handler;
 mod ddl;
 pub(crate) mod group_sync;
+mod hydration_history;
 mod indexes;
 mod info_metrics;
 mod introspection;
@@ -386,6 +387,8 @@ pub enum Message {
     ArrangementSizesSnapshot,
     ArrangementSizesWrite(Vec<ArrangementSizeRecord>),
     ArrangementSizesPrune(Vec<BuiltinTableUpdate>),
+    HydrationHistorySchedule,
+    HydrationHistoryRun,
     /// Performs any cleanup and logging actions necessary for
     /// finalizing a statement execution.
     RetireExecute {
@@ -536,6 +539,8 @@ impl Message {
             Message::ArrangementSizesSnapshot => "arrangement_sizes_snapshot",
             Message::ArrangementSizesWrite(_) => "arrangement_sizes_write",
             Message::ArrangementSizesPrune(_) => "arrangement_sizes_prune",
+            Message::HydrationHistorySchedule => "hydration_history_schedule",
+            Message::HydrationHistoryRun => "hydration_history_run",
             Message::RetireExecute { .. } => "retire_execute",
             Message::ExecuteSingleStatementTransaction { .. } => {
                 "execute_single_statement_transaction"
@@ -2064,6 +2069,11 @@ pub struct Coordinator {
     connection_cancel_watches: BTreeMap<ConnectionId, (watch::Sender<bool>, watch::Receiver<bool>)>,
     /// Active introspection subscribes.
     introspection_subscribes: BTreeMap<GlobalId, IntrospectionSubscribe>,
+    /// The last replica visited by the sequential hydration-history sweep.
+    hydration_history_replica_cursor: Option<ReplicaId>,
+    /// Hydration-history sweep, while one is in flight. Aborted when we are
+    /// dropped.
+    hydration_history_sweep: Option<AbortOnDropHandle<()>>,
 
     /// Locks that grant access to a specific object, populated lazily as objects are written to.
     write_locks: BTreeMap<CatalogItemId, Arc<tokio::sync::Mutex<()>>>,
@@ -3899,6 +3909,7 @@ impl Coordinator {
 
             self.schedule_storage_usage_collection().await;
             self.schedule_arrangement_sizes_collection().await;
+            self.schedule_hydration_history_collection();
             self.spawn_privatelink_vpc_endpoints_watch_task();
             self.spawn_statement_logging_task();
             self.spawn_catalog_info_metrics_task();
@@ -5104,6 +5115,8 @@ pub fn serve(
                     active_copies: BTreeMap::new(),
                     connection_cancel_watches: BTreeMap::new(),
                     introspection_subscribes: BTreeMap::new(),
+                    hydration_history_replica_cursor: None,
+                    hydration_history_sweep: None,
                     write_locks: BTreeMap::new(),
                     deferred_write_ops: BTreeMap::new(),
                     pending_writes: Vec::new(),