[pull] main from MaterializeInc:main

Cargo.toml

@@ -494,7 +494,7 @@ sentry-tracing = "0.49.1"
 serde = { version = "1.0.219", features = ["derive"] }
 serde-aux = "4.7.0"
 serde_bytes = "0.11.19"
-serde_json = "1.0.150"
+serde_json = { version = "1.0.150", features = ["raw_value"] }
 serde_plain = "1.0.2"
 serde_repr = "0.1.20"
 serde_urlencoded = "0.7.1"

doc/user/content/console/_index.md

@@ -5,7 +5,8 @@ disable_list: true
 disable_toc: true
 menus:
   main:
-    weight: 5
+    parent: "get-started"
+    weight: 24
     identifier: 'console'
 aliases:
   - /self-managed/v25.2/console/

doc/user/content/reference/system-catalog/mz_catalog.md

@@ -671,6 +671,11 @@ Field          | Type                 | Meaning
 
 The `mz_views` table contains a row for each view in the system.
 
+A few generated system views in `mz_internal` (the `mz_builtin_*` views,
+which list every builtin object) are shown with a short placeholder in
+`definition`, `create_sql`, and `redacted_create_sql` instead of their full
+SQL, which would embed metadata about every builtin object.
+
 <!-- RELATION_SPEC mz_catalog.mz_views -->
 Field          | Type                 | Meaning
 ---------------|----------------------|----------

doc/user/content/reference/system-catalog/mz_internal.md

@@ -646,6 +646,22 @@ all database objects in the system.
 | `object_id`             | [`text`]     | The ID of the dependent object. Corresponds to [`mz_objects.id`](../mz_catalog/#mz_objects).  |
 | `referenced_object_id`  | [`text`]     | The ID of the referenced object. Corresponds to [`mz_objects.id`](../mz_catalog/#mz_objects). |
 
+## `mz_object_graph_edges`
+
+The `mz_object_graph_edges` view describes the dependency edges between maintained
+objects, as rendered by object dependency graphs.
+
+It unions the dataflow-layer dependencies from
+[`mz_materialization_dependencies`](#mz_materialization_dependencies), restricted to
+indexes, materialized views, sinks, sources, and tables, with the source-to-subsource
+and source-to-table edges that stand in for sources whose subsources carry the data.
+
+<!-- RELATION_SPEC mz_internal.mz_object_graph_edges -->
+| Field           | Type     | Meaning                                                                                |
+| --------------- | -------- | -------------------------------------------------------------------------------------- |
+| `object_id`     | [`text`] | The ID of the dependent object. Corresponds to [`mz_objects.id`](../mz_catalog/#mz_objects).     |
+| `dependency_id` | [`text`] | The ID of the object it depends on. Corresponds to [`mz_objects.id`](../mz_catalog/#mz_objects). |
+
 ## `mz_object_fully_qualified_names`
 
 The `mz_object_fully_qualified_names` view enriches the [`mz_catalog.mz_objects`](/reference/system-catalog/mz_catalog/#mz_objects) view with namespace information.
@@ -1500,6 +1516,8 @@ The `mz_webhook_sources` table contains a row for each webhook source in the sys
 <!-- RELATION_SPEC_UNDOCUMENTED mz_internal.mz_activity_log_thinned -->
 <!-- RELATION_SPEC_UNDOCUMENTED mz_internal.mz_builtin_materialized_views -->
 <!-- RELATION_SPEC_UNDOCUMENTED mz_internal.mz_builtin_sources -->
+<!-- RELATION_SPEC_UNDOCUMENTED mz_internal.mz_builtin_tables -->
+<!-- RELATION_SPEC_UNDOCUMENTED mz_internal.mz_builtin_views -->
 <!-- RELATION_SPEC_UNDOCUMENTED mz_internal.mz_catalog_raw -->
 <!-- RELATION_SPEC_UNDOCUMENTED mz_internal.mz_cluster_replica_size_internal -->
 <!-- RELATION_SPEC_UNDOCUMENTED mz_internal.mz_cluster_workload_classes -->

doc/user/data/examples/create_connection.yml

@@ -656,7 +656,8 @@
       description: |
         *Value:* `text`. Required.
 
-        The name of the AWS PrivateLink service.
+        The name of the AWS VPC endpoint service, which always starts with
+        `com.amazonaws.`.
     - name: "`AVAILABILITY ZONES`"
       description: |
         *Value:* `text[]`. Required.

misc/python/materialize/checks/all_checks/builtin_version_pin.py

@@ -57,9 +57,9 @@
     "mz_sinks": MzVersion.parse_mz("v26.38.0"),
     "mz_ssh_tunnel_connections": MzVersion.parse_mz("v26.37.0"),
     "mz_system_privileges": MzVersion.parse_mz("v26.30.0"),
-    "mz_tables": None,
+    "mz_tables": MzVersion.parse_mz("v26.39.0"),
     "mz_types": None,
-    "mz_views": None,
+    "mz_views": MzVersion.parse_mz("v26.39.0"),
 }
 
 

misc/python/materialize/mzcompose/__init__.py

@@ -116,6 +116,7 @@ def get_minimal_system_parameters(
         "enable_s3_tables_region_check": "false",
         "enable_statement_lifecycle_logging": "true",
         "enable_storage_introspection_logs": "true",
+        "enable_compute_error_distinct": "true",
         "enable_compute_temporal_bucketing": "true",
         "enable_variadic_left_join_lowering": "true",
         "enable_worker_core_affinity": "true",
@@ -781,6 +782,7 @@ def get_default_system_parameters(
     "user_id_pool_batch_size",
     "webhook_max_request_size_bytes",
     "webhook_validation_memory_budget_bytes",
+    "subscribe_max_buffered_bytes",
     "cluster_controller_tick_interval",
     "default_cluster_reconfiguration_timeout",
     "read_then_write_max_dependencies",

misc/python/materialize/parallel_workload/action.py

@@ -2986,6 +2986,7 @@ def __init__(
         self.flags_with_values["enable_compute_temporal_bucketing"] = (
             BOOLEAN_FLAG_VALUES
         )
+        self.flags_with_values["enable_compute_error_distinct"] = BOOLEAN_FLAG_VALUES
         self.flags_with_values["enable_alter_table_add_column"] = BOOLEAN_FLAG_VALUES
         self.flags_with_values["enable_bounded_staleness_isolation"] = (
             BOOLEAN_FLAG_VALUES
@@ -3132,6 +3133,7 @@ def __init__(
             "compute_server_maintenance_interval",
             "compute_dataflow_max_inflight_bytes",
             "compute_dataflow_max_inflight_bytes_cc",
+            "subscribe_max_buffered_bytes",
             "compute_flat_map_fuel",
             "consolidating_vec_growth_dampener",
             "compute_hydration_concurrency",

src/adapter-types/src/dyncfgs.rs

@@ -297,6 +297,25 @@ pub const WEBHOOK_VALIDATION_MEMORY_BUDGET_BYTES: Config<usize> = Config::new(
     "The maximum bytes of temporary storage a webhook CHECK expression may allocate while validating one request.",
 );
 
+/// Budget for the backlog a `SUBSCRIBE` (or `COPY (SUBSCRIBE ...) TO STDOUT`)
+/// may accumulate in environmentd while waiting for a slow client to read.
+///
+/// The subscribe producer runs on the non-blockable coordinator loop, so it
+/// cannot apply backpressure to a slow client. Instead the coordinator retires
+/// the subscribe once its buffered backlog exceeds this budget, bounding the
+/// memory a slow client can make the shared process hold.
+///
+/// The backlog excludes the message the client is currently draining, so this
+/// bounds the accumulation of messages, not the size of any single one.
+/// `max_result_size` is what bounds an individual message. A client that keeps
+/// up holds at most one message at a time, so it stays at a zero backlog and a
+/// large snapshot batch is delivered rather than retired.
+pub const SUBSCRIBE_MAX_BUFFERED_BYTES: Config<usize> = Config::new(
+    "subscribe_max_buffered_bytes",
+    128 * 1024 * 1024,
+    "Maximum bytes a SUBSCRIBE may buffer in environmentd for a slow client before it is retired with an error.",
+);
+
 /// Number of user IDs to pre-allocate in a batch. Pre-allocating IDs avoids
 /// a persist write + oracle call per DDL statement.
 pub const USER_ID_POOL_BATCH_SIZE: Config<u32> = Config::new(
@@ -464,6 +483,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&MCP_REQUEST_TIMEOUT)
         .add(&WEBHOOK_MAX_REQUEST_SIZE_BYTES)
         .add(&WEBHOOK_VALIDATION_MEMORY_BUDGET_BYTES)
+        .add(&SUBSCRIBE_MAX_BUFFERED_BYTES)
         .add(&USER_ID_POOL_BATCH_SIZE)
         .add(&GROUP_COMMIT_MAX_ATTEMPTS)
         .add(&CONSOLE_OIDC_CLIENT_ID)

src/adapter/Cargo.toml

@@ -93,6 +93,7 @@ static_assertions.workspace = true
 timely.workspace = true
 tokio.workspace = true
 tokio-postgres.workspace = true
+tokio-stream.workspace = true
 tracing.workspace = true
 tracing-core.workspace = true
 tracing-opentelemetry.workspace = true

src/adapter/src/active_compute_sink.rs

@@ -10,8 +10,9 @@
 //! Coordinator bookkeeping for active compute sinks.
 
 use std::cmp::Ordering;
-use std::collections::BTreeSet;
+use std::collections::{BTreeSet, VecDeque};
 use std::num::NonZeroUsize;
+use std::sync::{Arc, Mutex};
 
 use mz_adapter_types::connection::ConnectionId;
 use mz_compute_client::protocol::response::SubscribeBatch;
@@ -88,6 +89,62 @@ pub enum ActiveComputeSinkRetireReason {
     /// The compute sink was forcibly terminated because an object it depended on
     /// was dropped.
     DependencyDropped(DroppedDependency),
+    /// The compute sink was retired because its coordinator-side buffer exceeded
+    /// its budget while the client was not reading fast enough. Carries the
+    /// buffered and budget byte counts for the terminal error.
+    BufferExceeded {
+        buffered_bytes: usize,
+        max_buffered_bytes: usize,
+    },
+}
+
+/// Overhead charged to every queued message on top of its payload.
+///
+/// A frontier-only advance carries no rows, so its payload is zero bytes. It
+/// still costs real memory: a node in the unbounded channel and an entry in
+/// `footprints`. Charging payload alone would leave the backlog flat while
+/// those two grow without bound, so the retire check would never trip on a
+/// stalled client that only receives progress messages. The fixed charge makes
+/// the budget bound message *count* as well as payload bytes.
+const SUBSCRIBE_MESSAGE_OVERHEAD_BYTES: usize = 1024;
+
+/// Footprints of the subscribe messages queued to the client but not yet
+/// drained, oldest first. The producer pushes one per message sent, the
+/// client-writer task pops as it drains. FIFO delivery keeps the queue aligned
+/// with the channel.
+///
+/// The message at the front is the one the client is currently draining. It is
+/// always tolerated, however large, so a client working through one big batch is
+/// not retired. Only what is queued behind it counts as backlog.
+#[derive(Debug, Default)]
+pub struct SubscribeBacklogAccounting {
+    /// Per-message footprints, in send order.
+    footprints: VecDeque<usize>,
+    /// Sum of `footprints`.
+    total: usize,
+}
+
+impl SubscribeBacklogAccounting {
+    /// Records a queued message of the given footprint.
+    pub fn push(&mut self, footprint: usize) {
+        self.footprints.push_back(footprint);
+        self.total = self.total.saturating_add(footprint);
+    }
+
+    /// Records the oldest queued message being drained by the client writer.
+    pub fn pop(&mut self) {
+        if let Some(footprint) = self.footprints.pop_front() {
+            self.total = self.total.saturating_sub(footprint);
+        }
+    }
+
+    /// Bytes queued behind the message the client is currently draining. That
+    /// message is tolerated whatever its size, so only this counts against the
+    /// budget.
+    pub fn backlog_size(&self) -> usize {
+        self.total
+            .saturating_sub(self.footprints.front().copied().unwrap_or(0))
+    }
 }
 
 /// A description of an active subscribe from coord's perspective
@@ -105,6 +162,17 @@ pub struct ActiveSubscribe {
     // The responses have the form `PeekResponseUnary` but should perhaps
     // become `SubscribeResponse`.
     pub channel: mpsc::UnboundedSender<PeekResponseUnary>,
+    /// Footprints of the messages queued in `channel` but not yet drained by the
+    /// client writer. Shared with the receiver side, which pops as it drains.
+    ///
+    /// The producer runs on the non-blockable coordinator loop and cannot block
+    /// on a slow client, so instead of applying backpressure the coordinator
+    /// watches `backlog_size` against `max_buffered_bytes` and retires the
+    /// subscribe once the backlog exceeds it.
+    pub backlog_accounting: Arc<Mutex<SubscribeBacklogAccounting>>,
+    /// Budget for the buffered backlog. A snapshot of `subscribe_max_buffered_bytes`
+    /// taken when the subscribe was created.
+    pub max_buffered_bytes: usize,
     /// Whether progress information should be emitted.
     pub emit_progress: bool,
     /// The logical timestamp at which the subscribe began execution.
@@ -154,8 +222,9 @@ impl ActiveSubscribe {
                 }
             }
 
+            let bytes = row_buf.byte_len();
             let row_iter = Box::new(row_buf.into_row_iter());
-            self.send(PeekResponseUnary::Rows(row_iter));
+            self.send(PeekResponseUnary::Rows(row_iter), bytes);
         }
     }
 
@@ -178,7 +247,7 @@ impl ActiveSubscribe {
                 mz_ore::iter::consolidate_update_iter(merged)
             }
             Err(s) => {
-                self.send(PeekResponseUnary::Error(s));
+                self.send(PeekResponseUnary::Error(s), 0);
                 return true;
             }
         };
@@ -376,8 +445,9 @@ impl ActiveSubscribe {
         };
 
         let rows = output_builder.build();
+        let bytes = rows.byte_len();
         let rows = Box::new(rows.into_row_iter());
-        self.send(PeekResponseUnary::Rows(rows));
+        self.send(PeekResponseUnary::Rows(rows), bytes);
 
         // Emit progress message if requested. Don't emit progress for the first
         // batch if the upper is exactly `as_of` (we're guaranteed it is not
@@ -402,15 +472,34 @@ impl ActiveSubscribe {
             ActiveComputeSinkRetireReason::DependencyDropped(d) => {
                 PeekResponseUnary::DependencyDropped(d)
             }
+            ActiveComputeSinkRetireReason::BufferExceeded {
+                buffered_bytes,
+                max_buffered_bytes,
+            } => PeekResponseUnary::Error(
+                AdapterError::SubscribeFellBehind {
+                    buffered_bytes,
+                    max_buffered_bytes,
+                }
+                .to_string(),
+            ),
         };
-        self.send(message);
+        self.send(message, 0);
     }
 
     /// Sends a message to the client if the subscribe has not already completed
     /// and if the client has not already gone away.
-    fn send(&self, response: PeekResponseUnary) {
-        // TODO(benesch): the lack of backpressure here can result in
-        // unbounded memory usage.
+    ///
+    /// `bytes` is the message's payload size. Its footprint (payload plus a fixed
+    /// per-message overhead) is recorded in `backlog_accounting` here and
+    /// released by the receiver side when the message is drained. Overflow of
+    /// the budget is detected by the coordinator after `process_response`
+    /// returns, not here, because this method cannot retire the sink.
+    fn send(&self, response: PeekResponseUnary, bytes: usize) {
+        let footprint = bytes.saturating_add(SUBSCRIBE_MESSAGE_OVERHEAD_BYTES);
+        self.backlog_accounting
+            .lock()
+            .expect("subscribe backlog accounting poisoned")
+            .push(footprint);
         let _ = self.channel.send(response);
     }
 }
@@ -456,6 +545,13 @@ impl ActiveCopyTo {
             ActiveComputeSinkRetireReason::DependencyDropped(dep) => {
                 Err(dep.to_concurrent_dependency_drop())
             }
+            ActiveComputeSinkRetireReason::BufferExceeded {
+                buffered_bytes,
+                max_buffered_bytes,
+            } => Err(AdapterError::SubscribeFellBehind {
+                buffered_bytes,
+                max_buffered_bytes,
+            }),
         };
         let _ = self.tx.send(message);
     }
@@ -473,3 +569,49 @@ pub(crate) struct ActiveCopyFrom {
     /// Context of the SQL session that ran the statement.
     pub ctx: ExecuteContext,
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::active_compute_sink::SubscribeBacklogAccounting;
+
+    /// The backlog excludes the message being drained, and zero-payload
+    /// messages (footprint = overhead only) still accumulate against it.
+    #[mz_ore::test]
+    fn test_subscribe_backlog_accounting() {
+        let mut acc = SubscribeBacklogAccounting::default();
+        assert_eq!(acc.backlog_size(), 0);
+
+        // A single large message is fully tolerated: nothing is queued behind it.
+        acc.push(10_000);
+        assert_eq!(acc.backlog_size(), 0);
+
+        // Near-empty messages (only per-message overhead) still build backlog, so
+        // a flood of frontier-only advances cannot grow without bound.
+        acc.push(1_024);
+        acc.push(1_024);
+        assert_eq!(acc.backlog_size(), 2_048);
+
+        // Draining the oldest message advances the tolerated front.
+        acc.pop();
+        assert_eq!(acc.backlog_size(), 1_024);
+
+        acc.pop();
+        acc.pop();
+        assert_eq!(acc.backlog_size(), 0);
+    }
+
+    /// A client that drains each message before the next is sent never
+    /// accumulates backlog, however many messages flow and however large they
+    /// are. This is the property that keeps a well-behaved subscribe from ever
+    /// being retired, so it is asserted after every step rather than at the end.
+    #[mz_ore::test]
+    fn test_subscribe_backlog_keeping_up_client() {
+        let mut acc = SubscribeBacklogAccounting::default();
+        for i in 0..1_000 {
+            acc.push(1_024 + i * 4_096);
+            assert_eq!(acc.backlog_size(), 0, "message {i} built backlog");
+            acc.pop();
+            assert_eq!(acc.backlog_size(), 0, "message {i} left backlog behind");
+        }
+    }
+}