adapter: move arrangement size snapshotting off the coordinator loop

src/adapter/src/coord.rs

@@ -316,6 +316,17 @@ impl IdPool {
     }
 }
 
+/// A row for `mz_object_arrangement_size_history`, prepared off-thread by the
+/// arrangement sizes snapshot task and stamped with a collection timestamp at
+/// write time.
+#[derive(Debug)]
+pub struct ArrangementSizeRecord {
+    pub replica_id: String,
+    pub object_id: String,
+    pub size: i64,
+    pub hydration_complete: bool,
+}
+
 #[derive(Debug)]
 pub enum Message {
     Command(OpenTelemetryContext, Command),
@@ -359,6 +370,7 @@ pub enum Message {
     StorageUsagePrune(Vec<BuiltinTableUpdate>),
     ArrangementSizesSchedule,
     ArrangementSizesSnapshot,
+    ArrangementSizesWrite(Vec<ArrangementSizeRecord>),
     ArrangementSizesPrune(Vec<BuiltinTableUpdate>),
     /// Performs any cleanup and logging actions necessary for
     /// finalizing a statement execution.
@@ -509,6 +521,7 @@ impl Message {
             Message::StorageUsagePrune(_) => "storage_usage_prune",
             Message::ArrangementSizesSchedule => "arrangement_sizes_schedule",
             Message::ArrangementSizesSnapshot => "arrangement_sizes_snapshot",
+            Message::ArrangementSizesWrite(_) => "arrangement_sizes_write",
             Message::ArrangementSizesPrune(_) => "arrangement_sizes_prune",
             Message::RetireExecute { .. } => "retire_execute",
             Message::ExecuteSingleStatementTransaction { .. } => {

src/adapter/src/coord/introspection.rs

@@ -28,6 +28,9 @@
 //!   failure), `handle_introspection_subscribe_batch` reacts on the corresponding error responses
 //!   by reinstalling the failed introspection subscribes.
 
+use std::collections::BTreeSet;
+use std::time::{Duration, Instant};
+
 use anyhow::bail;
 use derivative::Derivative;
 use mz_adapter_types::dyncfgs::ENABLE_INTROSPECTION_SUBSCRIBES;
@@ -72,6 +75,13 @@ pub(super) struct IntrospectionSubscribe {
     /// introspection data around in the meantime makes for a better UX than removing it.
     #[derivative(Debug = "ignore")]
     deferred_write: Option<StorageWriteOp>,
+    /// When this subscribe first appended data to the target storage collection, if it has.
+    ///
+    /// Until then, the target collection may still contain rows written by a previous incarnation
+    /// of this subscribe (before an environmentd restart, or before the target replica
+    /// reconnected). Consumers that must not observe such stale rows, like the
+    /// `mz_object_arrangement_size_history` snapshots, use this to judge per-replica freshness.
+    first_data_at: Option<Instant>,
 }
 
 impl IntrospectionSubscribe {
@@ -144,6 +154,7 @@ impl Coordinator {
             replica_id,
             spec,
             deferred_write: None,
+            first_data_at: None,
         };
         self.introspection_subscribes.insert(id, subscribe);
 
@@ -410,6 +421,9 @@ impl Coordinator {
         // Ensure that the contents of the target storage collection are cleaned when the new
         // subscribe starts reporting data.
         subscribe.deferred_write = Some(subscribe.delete_write_op());
+        // Until then, the collection serves the previous subscribe's data, which the replica may
+        // have invalidated by restarting.
+        subscribe.first_data_at = None;
 
         self.introspection_subscribes.insert(new_id, subscribe);
         self.sequence_introspection_subscribe(new_id, spec, cluster_id, replica_id)
@@ -468,13 +482,37 @@ impl Coordinator {
                 .update_introspection_collection(subscribe.spec.introspection_type, op);
         }
 
+        subscribe.first_data_at.get_or_insert_with(Instant::now);
+
         self.controller.storage.update_introspection_collection(
             subscribe.spec.introspection_type,
             StorageWriteOp::Append {
                 updates: new_updates,
             },
         );
     }
+
+    /// Returns the IDs of replicas whose introspection subscribe of the given type first
+    /// delivered data at least `margin` ago.
+    ///
+    /// Rows for other replicas in the corresponding storage collection are not trustworthy: they
+    /// either predate this environmentd process or were written before the replica reconnected,
+    /// and may describe a previous incarnation of the replica. The margin accounts for the
+    /// subscribe's first append becoming visible to readers only asynchronously (the collection
+    /// manager flushes writes in batches, and snapshot reads use an oracle timestamp that trails
+    /// the wall clock).
+    pub(super) fn fresh_introspection_replicas(
+        &self,
+        introspection_type: IntrospectionType,
+        margin: Duration,
+    ) -> BTreeSet<String> {
+        self.introspection_subscribes
+            .values()
+            .filter(|s| s.spec.introspection_type == introspection_type)
+            .filter(|s| s.first_data_at.is_some_and(|at| at.elapsed() >= margin))
+            .map(|s| s.replica_id.to_string())
+            .collect()
+    }
 }
 
 impl Staged for IntrospectionSubscribeStage {
@@ -576,11 +614,11 @@ const SUBSCRIBES: &[SubscribeSpec] = &[
     // differential logs where each `+1` row represents one byte of heap delta;
     // after consolidation, `COUNT(*)` is the current arrangement size in bytes.
     //
-    // The `HAVING` floor drops objects below 10 MiB. Below that threshold the
-    // heap-size collection wiggles by a few bytes per second from ordinary
-    // allocator activity, and emitting exact bytes would push a downstream
-    // update on every wiggle. Quantizing to the nearest 10 MiB keeps the
-    // emitted size stable across in-bucket wiggle.
+    // Sizes are quantized to the nearest 10 MiB: the heap-size collection
+    // wiggles by a few bytes per second from ordinary allocator activity, and
+    // emitting exact bytes would push a downstream update on every wiggle.
+    // Arrangements below 5 MiB quantize to a size of 0. They are kept rather
+    // than dropped so that every arranged object is present in the collection.
     //
     // `mz_dataflow_addresses.address[1]` is the root of each operator's address
     // tree, which equals the owning `dataflow_id` — so we can go addresses →
@@ -610,7 +648,6 @@ const SUBSCRIBES: &[SubscribeSpec] = &[
             ) AS rs ON rs.operator_id = od.operator_id
             WHERE ce.export_id NOT LIKE 't%'
             GROUP BY ce.export_id
-            HAVING COUNT(*) >= 10485760
         )",
     },
 ];