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,28 @@ 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.
+    ///
+    /// Such a subscribe is always `internal`, because there is no session to
+    /// attribute a `mz_subscriptions` row to. `pack_subscribe_update` depends on
+    /// that: it needs a session uuid, and is only reached for subscribes that
+    /// are not internal.
+    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 +136,22 @@ pub struct ActiveSubscribe {
 }
 
 impl ActiveSubscribe {
+    /// 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

@@ -931,7 +931,11 @@ impl CatalogState {
         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(
+            subscribe
+                .session_uuid()
+                .expect("a subscribe with an introspection row is session-owned"),
+        ));
         packer.push(Datum::String(&subscribe.cluster_id.to_string()));
 
         let start_dt = mz_ore::now::to_datetime(subscribe.start_time);

src/adapter/src/client.rs

@@ -72,7 +72,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 +313,7 @@ impl Client {
         } = response;
 
         let peek_client = PeekClient::new(
-            self.clone(),
+            CoordinatorClient::Session(self.clone()),
             &catalog,
             storage_collections,
             transient_id_gen,

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,10 +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,
+        /// Connection originating the write, so the coordinator can cancel this
+        /// pending write if the connection is cancelled before it commits.
+        /// `None` for coordinator background work, which has no connection.
+        conn_id: Option<ConnectionId>,
         target_id: CatalogItemId,
         target_global_id: GlobalId,
         diffs: Vec<(Row, Diff)>,

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,10 @@ 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>,
+    /// The in-flight hydration-history sweep. 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 +3908,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 +5114,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(),

src/adapter/src/coord/command_handler.rs

@@ -638,25 +638,14 @@ impl Coordinator {
                     as_of,
                     arity,
                     sink_id,
-                    conn_id,
-                    session_uuid,
+                    owner,
                     start_time,
                     read_holds,
                     tx,
                 } => {
                     self.handle_create_internal_subscribe(
-                        *df_desc,
-                        cluster_id,
-                        replica_id,
-                        depends_on,
-                        as_of,
-                        arity,
-                        sink_id,
-                        conn_id,
-                        session_uuid,
-                        start_time,
-                        read_holds,
-                        tx,
+                        *df_desc, cluster_id, replica_id, depends_on, as_of, arity, sink_id, owner,
+                        start_time, read_holds, tx,
                     )
                     .await;
                 }