From 3311df27ecd29fc15bdb9e44028627e78c5fdc15 Mon Sep 17 00:00:00 2001 From: Jon Currey Date: Wed, 23 Sep 2026 08:35:14 -0400 Subject: [PATCH 1/6] orchestrator-kubernetes: log early retries of a failed command at warn level (#39060) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### Motivation `OrchestratorWorker::handle_command` retries every failed `EnsureService`, `DropService` and `ListServices` call forever (`Retry::default()`, backoff starting at 125ms and capped at 10s), and logged every failed attempt with `tracing::error!`, which Sentry records as an event. A Kubernetes API error that clears on retry therefore reaches Sentry even though the command succeeds a moment later. [DATABASE-BACKEND-503](https://materializeinc.sentry.io/issues/7746503161/) ([CLO-288](https://linear.app/materializeinc/issue/CLO-288)) is one such case: on two staging environments running v26.43.0-rc.1 and rc.2, environmentd's service account was refused `list` on StatefulSets at startup, `statefulsets.apps is forbidden`. Each environment produced three events within the same second and none after, which matches three failed attempts followed by a success at about 875ms. The Cloud team attributes it to RBAC that had not yet propagated. ### Description - The first `RETRY_WARN_ATTEMPTS` (5) attempts of a failed worker command log at warn level, and later attempts at error level. With the existing backoff, the first error-level attempt comes about 4s after the first failure, so a persistent failure still reaches Sentry within seconds. - Both levels carry the attempt number and the time since the first attempt as structured fields. The message text is unchanged, so existing Sentry grouping is unaffected. - The retry behavior itself is unchanged. No tests were added. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5.5 --- src/orchestrator-kubernetes/src/lib.rs | 30 ++++++++++++++++++++++---- 1 file changed, 26 insertions(+), 4 deletions(-) diff --git a/src/orchestrator-kubernetes/src/lib.rs b/src/orchestrator-kubernetes/src/lib.rs index 98be7d1b6889b..a4eb62945c2a5 100644 --- a/src/orchestrator-kubernetes/src/lib.rs +++ b/src/orchestrator-kubernetes/src/lib.rs @@ -65,6 +65,11 @@ pub mod util; const FIELD_MANAGER: &str = "environmentd"; const NODE_FAILURE_THRESHOLD_SECONDS: i64 = 30; +/// How many attempts of a failed worker command log at warn level before later attempts log at +/// error level. With the retry backoff starting at 125ms, the first error-level attempt comes +/// about 4s after the first failure, so a Kubernetes API error that clears on retry, such as a +/// new service account's RBAC that has not yet propagated, does not reach Sentry. +const RETRY_WARN_ATTEMPTS: usize = 5; const POD_TEMPLATE_HASH_ANNOTATION: &str = "environmentd.materialize.cloud/pod-template-hash"; @@ -1461,12 +1466,29 @@ impl OrchestratorWorker { F: Fn() -> U, U: Future>, { + let start = Instant::now(); Retry::default() .clamp_backoff(Duration::from_secs(10)) - .retry_async(|_| { - f().map_err( - |error| tracing::error!(%cmd_type, "orchestrator call failed: {error}"), - ) + .retry_async(|state| { + f().map_err(move |error| { + let attempt = state.i + 1; + let elapsed = start.elapsed(); + if state.i < RETRY_WARN_ATTEMPTS { + tracing::warn!( + %cmd_type, + attempt, + ?elapsed, + "orchestrator call failed: {error}" + ); + } else { + tracing::error!( + %cmd_type, + attempt, + ?elapsed, + "orchestrator call failed: {error}" + ); + } + }) }) .await .expect("always retries on error") From 2b5d1b4d4c8d372b0287920bc3f44b2c7ee21616 Mon Sep 17 00:00:00 2001 From: Patrick Butler Date: Wed, 23 Sep 2026 10:13:26 -0400 Subject: [PATCH 2/6] SS-480 Iceberg Sink: Write UUID columns as text instead of fixed binary (#39033) Adds a type override for iceberg sinks so that UUIDs are written as text instead of fixed binarys, as Unity Catalog doesn't have support for fixed types. --- src/arrow-util/src/builder.rs | 11 ++ src/storage-types/src/sinks.rs | 3 + src/storage/src/sink/iceberg.rs | 184 +++++++++++++++++++++++++++-- test/iceberg/legacy-uuid-column.td | 85 +++++++++++++ test/iceberg/mzcompose.py | 54 +++++++++ test/iceberg/uuid-column.td | 97 +++++++++++++++ 6 files changed, 426 insertions(+), 8 deletions(-) create mode 100644 test/iceberg/legacy-uuid-column.td create mode 100644 test/iceberg/uuid-column.td diff --git a/src/arrow-util/src/builder.rs b/src/arrow-util/src/builder.rs index bdfa9943e175a..5164df3ad3336 100644 --- a/src/arrow-util/src/builder.rs +++ b/src/arrow-util/src/builder.rs @@ -974,6 +974,17 @@ impl ArrowColumn { (ColBuilder::LargeStringBuilder(builder), Datum::Interval(iv)) => { builder.append_value(iv.to_string()) } + // Uuid-to-string conversion for destinations that don't support a uuid type + // natively. Iceberg's spec does have a `uuid` primitive, but our bridge to it + // converts on the Arrow `DataType` alone and so cannot reach it; see + // `iceberg_type_overrides`. `Display` renders the lowercase hyphenated form, + // matching what Materialize itself produces for `uuid::text`. + (ColBuilder::StringBuilder(builder), Datum::Uuid(val)) => { + builder.append_value(val.to_string()) + } + (ColBuilder::LargeStringBuilder(builder), Datum::Uuid(val)) => { + builder.append_value(val.to_string()) + } (ColBuilder::Decimal128Builder(builder), Datum::Numeric(mut dec)) => { if dec.0.is_special() { anyhow::bail!("Cannot represent special numeric value {} in parquet", dec) diff --git a/src/storage-types/src/sinks.rs b/src/storage-types/src/sinks.rs index 694fc42f8c0f0..d5c3dd4c8a5a0 100644 --- a/src/storage-types/src/sinks.rs +++ b/src/storage-types/src/sinks.rs @@ -679,6 +679,8 @@ pub const ICEBERG_UINT64_DECIMAL_PRECISION: u8 = 20; /// - `UInt64` -> `Decimal128(20, 0)` /// - `MzTimestamp` (which uses UInt64) -> `Decimal128(20, 0)` /// - `Interval` -> string (`LargeUtf8`) +/// - `Uuid` -> string (`Utf8`). Unity Catalog does not support fixed binary types, so we use string +/// instead of `FixedSizeBinary(16)`. /// /// Pass this to `mz_arrow_util::builder::desc_to_schema_with_overrides` /// when producing the Arrow schema for an iceberg sink, and to @@ -701,6 +703,7 @@ pub fn iceberg_type_overrides( "mz_timestamp".to_string(), )), SqlScalarType::Interval => Some((DataType::LargeUtf8, "interval".to_string())), + SqlScalarType::Uuid => Some((DataType::Utf8, "uuid".to_string())), _ => None, } } diff --git a/src/storage/src/sink/iceberg.rs b/src/storage/src/sink/iceberg.rs index 2f73e194f27ba..5522320c41b0f 100644 --- a/src/storage/src/sink/iceberg.rs +++ b/src/storage/src/sink/iceberg.rs @@ -100,8 +100,8 @@ use futures::StreamExt; use iceberg::ErrorKind; use iceberg::arrow::{arrow_schema_to_schema, schema_to_arrow_schema}; use iceberg::spec::{ - DataFile, FormatVersion, Snapshot, StructType, read_data_files_from_avro, - write_data_files_to_avro, + DataFile, FormatVersion, NestedField, PrimitiveType, Snapshot, StructType, Type, + read_data_files_from_avro, write_data_files_to_avro, }; use iceberg::spec::{Schema, SchemaRef}; use iceberg::table::Table; @@ -968,6 +968,53 @@ async fn try_commit_batch( } } +/// Whether the sink can write into a table that already carries `current`, +/// where `expected` is the schema the sink would create today. +/// +/// Field ids, names and nullability have to match. Types have to match too, +/// except that a column may be `fixed[16]` where `expected` has `string`. Uuid +/// columns were written as fixed binary before, and a table holding one stays +/// writable, because the writer builds its Arrow columns from the table's own +/// schema rather than from `expected`. Rejecting it would strand the sink. +/// +/// NOTE: the tolerance keys on the Iceberg types alone, so it also accepts a +/// `fixed[16]` column where the relation has `text`. Such a table fails on the +/// first row written instead of here. Only a table created outside Materialize +/// can be in that state, since the sink never creates one. +fn is_compatible(current: &Schema, expected: &Schema) -> bool { + current + .identifier_field_ids() + .eq(expected.identifier_field_ids()) + && struct_is_compatible(current.as_struct(), expected.as_struct()) +} + +fn struct_is_compatible(current: &StructType, expected: &StructType) -> bool { + current.fields().len() == expected.fields().len() + && std::iter::zip(current.fields(), expected.fields()) + .all(|(c, e)| field_is_compatible(c, e)) +} + +fn field_is_compatible(current: &NestedField, expected: &NestedField) -> bool { + current.id == expected.id + && current.name == expected.name + && current.required == expected.required + && match (&*current.field_type, &*expected.field_type) { + // A uuid column written before uuids became strings. + (Type::Primitive(PrimitiveType::Fixed(16)), Type::Primitive(PrimitiveType::String)) => { + true + } + (Type::Struct(c), Type::Struct(e)) => struct_is_compatible(c, e), + (Type::List(c), Type::List(e)) => { + field_is_compatible(&c.element_field, &e.element_field) + } + (Type::Map(c), Type::Map(e)) => { + field_is_compatible(&c.key_field, &e.key_field) + && field_is_compatible(&c.value_field, &e.value_field) + } + (c, e) => c == e, + } +} + /// Load an existing Iceberg table or create it if it doesn't exist. async fn load_or_create_table( catalog: &dyn Catalog, @@ -984,11 +1031,7 @@ async fn load_or_create_table( // Table exists, return it // TODO: Add proper schema evolution/validation to ensure compatibility let current_schema = table.metadata().current_schema(); - if !(current_schema.as_struct().eq(schema.as_struct()) - && current_schema - .identifier_field_ids() - .eq(schema.identifier_field_ids())) - { + if !is_compatible(current_schema, schema) { anyhow::bail!( "Iceberg table '{}' schema does not match expected schema. \ Current schema: {:?}, expected schema: {:?}", @@ -2032,7 +2075,7 @@ where #[cfg(test)] mod tests { - use iceberg::spec::{PrimitiveType, Type}; + use iceberg::spec::{ListType, PrimitiveType, Type}; use iceberg::writer::file_writer::location_generator::LocationGenerator; use mz_repr::SqlScalarType; use mz_storage_types::sinks::ICEBERG_UINT64_DECIMAL_PRECISION; @@ -2135,6 +2178,14 @@ mod tests { DataType::Decimal128(ICEBERG_UINT64_DECIMAL_PRECISION, 0) ); + // Interval should override to LargeUtf8 + let result = iceberg_type_overrides(&SqlScalarType::Interval); + assert_eq!(result.unwrap().0, DataType::LargeUtf8); + + // Uuid should override to Utf8 + let result = iceberg_type_overrides(&SqlScalarType::Uuid); + assert_eq!(result.unwrap().0, DataType::Utf8); + // Other types should return None (use default) assert!(iceberg_type_overrides(&SqlScalarType::Int32).is_none()); assert!(iceberg_type_overrides(&SqlScalarType::String).is_none()); @@ -2197,6 +2248,123 @@ mod tests { assert_eq!(*field.field_type, Type::Primitive(PrimitiveType::String)); } + /// A uuid column must reach the Iceberg table as `string`, not as the + /// `fixed[16]` that the default `FixedSizeBinary(16)` mapping would produce. + #[mz_ore::test] + fn test_iceberg_uuid_override() { + let result = iceberg_type_overrides(&SqlScalarType::Uuid); + assert_eq!(result.unwrap().0, DataType::Utf8); + + let desc = mz_repr::RelationDesc::builder() + .with_column("id", SqlScalarType::Int32.nullable(false)) + .with_column("u", SqlScalarType::Uuid.nullable(true)) + .finish(); + + let (arrow_schema, iceberg_schema) = + relation_desc_to_iceberg_schema(&desc).expect("schema conversion should succeed"); + + assert_eq!(arrow_schema.field(1).data_type(), &DataType::Utf8); + + let field = iceberg_schema + .as_struct() + .field_by_name("u") + .expect("field should exist"); + assert_eq!(*field.field_type, Type::Primitive(PrimitiveType::String)); + assert_ne!( + *field.field_type, + Type::Primitive(PrimitiveType::Fixed(16)), + "uuid must not fall back to the default FixedSizeBinary(16) mapping" + ); + } + + /// Rebuilds `schema` with the named field's type replaced, so a fixture can + /// differ from what the sink derives in exactly one type. + fn with_field_type(schema: &Schema, name: &str, ty: Type) -> Schema { + let fields = schema.as_struct().fields().iter().map(|f| { + let mut field = (**f).clone(); + if field.name == name { + field.field_type = Box::new(ty.clone()); + } + Arc::new(field) + }); + Schema::builder() + .with_fields(fields) + .build() + .expect("valid schema") + } + + /// A table created before uuid columns became strings holds `fixed[16]` and + /// must stay writable, while any other difference is still a mismatch. + #[mz_ore::test] + fn test_is_compatible_accepts_legacy_uuid() { + let desc = mz_repr::RelationDesc::builder() + .with_column("id", SqlScalarType::Int32.nullable(false)) + .with_column("u", SqlScalarType::Uuid.nullable(true)) + .finish(); + let (_, expected) = + relation_desc_to_iceberg_schema(&desc).expect("schema conversion should succeed"); + + assert!(is_compatible(&expected, &expected)); + + // The halt this tolerance exists to prevent. + let legacy = with_field_type(&expected, "u", Type::Primitive(PrimitiveType::Fixed(16))); + assert!(is_compatible(&legacy, &expected)); + + // Only that direction: nothing produces a string column where the sink + // derives fixed binary. + assert!(!is_compatible(&expected, &legacy)); + + // An unrelated type difference is still a mismatch. + let wrong_type = with_field_type(&expected, "u", Type::Primitive(PrimitiveType::Long)); + assert!(!is_compatible(&wrong_type, &expected)); + + // So is a missing column. + let truncated = Schema::builder() + .with_fields(expected.as_struct().fields().iter().take(1).cloned()) + .build() + .expect("valid schema"); + assert!(!is_compatible(&truncated, &expected)); + } + + /// The tolerance reaches a uuid nested inside a list, which the type + /// overrides remap just like a top-level column. + #[mz_ore::test] + fn test_is_compatible_accepts_legacy_uuid_in_list() { + let desc = mz_repr::RelationDesc::builder() + .with_column("id", SqlScalarType::Int32.nullable(false)) + .with_column( + "us", + SqlScalarType::List { + element_type: Box::new(SqlScalarType::Uuid), + custom_id: None, + } + .nullable(true), + ) + .finish(); + let (_, expected) = + relation_desc_to_iceberg_schema(&desc).expect("schema conversion should succeed"); + + let Type::List(list) = &*expected + .as_struct() + .field_by_name("us") + .expect("field should exist") + .field_type + else { + panic!("expected a list type"); + }; + let mut legacy_element = (*list.element_field).clone(); + legacy_element.field_type = Box::new(Type::Primitive(PrimitiveType::Fixed(16))); + let legacy = with_field_type( + &expected, + "us", + Type::List(ListType { + element_field: Arc::new(legacy_element), + }), + ); + + assert!(is_compatible(&legacy, &expected)); + } + #[mz_ore::test] fn test_iceberg_range_schema() { // Test full schema conversion with range column diff --git a/test/iceberg/legacy-uuid-column.td b/test/iceberg/legacy-uuid-column.td new file mode 100644 index 0000000000000..fb85248c88c32 --- /dev/null +++ b/test/iceberg/legacy-uuid-column.td @@ -0,0 +1,85 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +# A sink pointed at a table whose uuid column predates the uuid-as-string +# mapping must keep writing into it. The table holds `fixed[16]` where the sink +# would now create `string`, and a strict schema comparison would halt the sink +# forever. The workflow creates that table through the catalog REST API before +# this file runs. + +> CREATE SECRET legacy_uuid_access_key_secret AS '${arg.s3-access-key}' + +> CREATE CONNECTION legacy_uuid_polaris TO ICEBERG CATALOG ( + CATALOG TYPE = 'REST', + URL = 'http://polaris:8181/api/catalog', + CREDENTIAL = 'root:root', + WAREHOUSE = 'default_catalog', + SCOPE = 'PRINCIPAL_ROLE:ALL' + ); + +> CREATE TABLE legacy_uuid_rows (id int, u uuid); + +> INSERT INTO legacy_uuid_rows VALUES + (1, 'a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11'), + (2, NULL); + +> CREATE SINK legacy_uuid_sink + FROM legacy_uuid_rows + INTO ICEBERG CATALOG CONNECTION legacy_uuid_polaris ( + NAMESPACE 'default_namespace', + TABLE 'legacy_uuid_table' + ) + KEY (id) NOT ENFORCED + MODE UPSERT + WITH (COMMIT INTERVAL '1s'); + +$ sleep-is-probably-flaky-i-have-justified-my-need-with-a-comment duration=10s + +# Before the schema comparison was relaxed this reported `stalled` with +# "schema does not match expected schema", and stayed there across restarts. +> SELECT status, error FROM mz_internal.mz_sink_statuses WHERE name = 'legacy_uuid_sink' +running + +$ duckdb-execute name=legacy_uuid_iceberg +CREATE SECRET s3_secret_legacy_uuid (TYPE S3, KEY_ID 'tduser', SECRET '${arg.s3-access-key}', ENDPOINT '${arg.aws-endpoint}', URL_STYLE 'path', USE_SSL false, REGION 'minio'); +SET unsafe_enable_version_guessing = true; +CREATE SECRET polaris_secret_legacy_uuid (TYPE iceberg, CLIENT_ID 'root', CLIENT_SECRET 'root', ENDPOINT 'http://polaris:8181/api/catalog'); +ATTACH 'default_catalog' AS polaris_legacy_uuid (TYPE iceberg, SECRET polaris_secret_legacy_uuid, ACCESS_DELEGATION_MODE 'none'); + +# The column keeps its original type. The sink writes into the table as it +# found it rather than rewriting the schema, which is what makes the existing +# binary builder the right one for these rows. +$ duckdb-query name=legacy_uuid_iceberg +SELECT typeof(u) FROM polaris_legacy_uuid.default_namespace.legacy_uuid_table WHERE id = 1 +BLOB + +# The 16 bytes of the uuid, hex-encoded, confirming the values themselves +# survive the legacy encoding. +$ duckdb-query name=legacy_uuid_iceberg +SELECT id, hex(u) FROM polaris_legacy_uuid.default_namespace.legacy_uuid_table ORDER BY id +1 A0EEBC999C0B4EF8BB6D6BB9BD380A11 +2 + +# Updates and inserts keep flowing after the table has been accepted. +> UPDATE legacy_uuid_rows SET u = 'ffffffff-ffff-ffff-ffff-ffffffffffff' WHERE id = 1; + +> INSERT INTO legacy_uuid_rows VALUES (3, '11111111-2222-3333-4444-555555555555'); + +$ sleep-is-probably-flaky-i-have-justified-my-need-with-a-comment duration=10s + +$ duckdb-query name=legacy_uuid_iceberg +SELECT id, hex(u) FROM polaris_legacy_uuid.default_namespace.legacy_uuid_table ORDER BY id +1 FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF +2 +3 11111111222233334444555555555555 + +> SELECT status FROM mz_internal.mz_sink_statuses WHERE name = 'legacy_uuid_sink' +running + +> DROP SINK legacy_uuid_sink diff --git a/test/iceberg/mzcompose.py b/test/iceberg/mzcompose.py index a0b9a4f79ee9c..a7802771c578a 100644 --- a/test/iceberg/mzcompose.py +++ b/test/iceberg/mzcompose.py @@ -126,6 +126,7 @@ def workflow_smoke(c: Composition) -> None: "--var=aws-endpoint=minio:9000", "catalog.td", "nested-records.td", + "uuid-column.td", "key-validation.td", ) @@ -863,3 +864,56 @@ def workflow_range_noncanonical(c: Composition) -> None: "--var=aws-endpoint=minio:9000", "range-noncanonical.td", ) + + +def workflow_legacy_uuid_column(c: Composition) -> None: + """A table whose uuid column predates the uuid-as-string mapping must keep + working. + + Such a table holds `fixed[16]` where the sink would now create `string`, so + a strict schema comparison halts the sink forever. The table is created here + through the catalog's REST API, because nothing in this composition can + produce one otherwise: the sink only ever creates tables with the current + mapping.""" + key = _setup(c) + + polaris_port = c.port("polaris", 8181) + base_url = f"http://localhost:{polaris_port}" + access_token = get_polaris_access_token(c) + + # Field ids and nullability have to match what the sink derives from + # `CREATE TABLE legacy_uuid_rows (id int, u uuid)`, so that the only + # difference left is the uuid column's type. + create_table = { + "name": "legacy_uuid_table", + "schema": { + "type": "struct", + "schema-id": 0, + "fields": [ + {"id": 1, "name": "id", "required": False, "type": "int"}, + {"id": 2, "name": "u", "required": False, "type": "fixed[16]"}, + ], + }, + } + req = urllib.request.Request( + f"{base_url}/api/catalog/v1/default_catalog" + f"/namespaces/default_namespace/tables", + data=json.dumps(create_table).encode(), + headers={ + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json", + }, + method="POST", + ) + resp = urllib.request.urlopen(req) + created = json.loads(resp.read()) + fields = created["metadata"]["schemas"][0]["fields"] + assert ( + fields[1]["type"] == "fixed[16]" + ), f"table was not created with a legacy uuid column: {fields}" + + c.run_testdrive_files( + f"--var=s3-access-key={key}", + "--var=aws-endpoint=minio:9000", + "legacy-uuid-column.td", + ) diff --git a/test/iceberg/uuid-column.td b/test/iceberg/uuid-column.td new file mode 100644 index 0000000000000..bbfde712ef81b --- /dev/null +++ b/test/iceberg/uuid-column.td @@ -0,0 +1,97 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +# A uuid column must reach the Iceberg table as a string column. The default +# Arrow mapping for uuid is FixedSizeBinary(16), which iceberg-rust turns into +# `fixed[16]` (a blob to readers), so `iceberg_type_overrides` remaps uuid to +# Utf8. Both the column type and the text form of the values are asserted: the +# type alone would still pass if the values arrived as raw bytes rendered as +# text. + +> CREATE SECRET uuid_access_key_secret AS '${arg.s3-access-key}' + +> CREATE CONNECTION uuid_polaris TO ICEBERG CATALOG ( + CATALOG TYPE = 'REST', + URL = 'http://polaris:8181/api/catalog', + CREDENTIAL = 'root:root', + WAREHOUSE = 'default_catalog', + SCOPE = 'PRINCIPAL_ROLE:ALL' + ); + +# Row 2 is uppercase on input so the assertions pin the lowercase hyphenated +# rendering, and row 3 covers the null path, which bypasses the uuid builder +# arm entirely. +> CREATE TABLE uuid_rows (id int, u uuid); + +> INSERT INTO uuid_rows VALUES + (1, 'a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11'), + (2, 'A0EEBC99-9C0B-4EF8-BB6D-6BB9BD380A22'), + (3, NULL); + +> CREATE SINK uuid_sink + FROM uuid_rows + INTO ICEBERG CATALOG CONNECTION uuid_polaris ( + NAMESPACE 'default_namespace', + TABLE 'uuid_table' + ) + KEY (id) NOT ENFORCED + MODE UPSERT + WITH (COMMIT INTERVAL '1s'); + +$ sleep-is-probably-flaky-i-have-justified-my-need-with-a-comment duration=10s + +> SELECT status FROM mz_internal.mz_sink_statuses WHERE name = 'uuid_sink' +running + +$ duckdb-execute name=uuid_iceberg +CREATE SECRET s3_secret_uuid (TYPE S3, KEY_ID 'tduser', SECRET '${arg.s3-access-key}', ENDPOINT '${arg.aws-endpoint}', URL_STYLE 'path', USE_SSL false, REGION 'minio'); +SET unsafe_enable_version_guessing = true; +CREATE SECRET polaris_secret_uuid (TYPE iceberg, CLIENT_ID 'root', CLIENT_SECRET 'root', ENDPOINT 'http://polaris:8181/api/catalog'); +ATTACH 'default_catalog' AS polaris_uuid (TYPE iceberg, SECRET polaris_secret_uuid, ACCESS_DELEGATION_MODE 'none'); + +# The column type as the catalog reports it. This is BLOB when uuid falls +# through to the default FixedSizeBinary(16) mapping. +$ duckdb-query name=uuid_iceberg +SELECT typeof(u) FROM polaris_uuid.default_namespace.uuid_table WHERE id = 1 +VARCHAR + +$ duckdb-query name=uuid_iceberg +SELECT id, u FROM polaris_uuid.default_namespace.uuid_table ORDER BY id +1 a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11 +2 a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a22 +3 + +# Reading the data files directly distinguishes a schema-only regression from a +# data-path one. +$ duckdb-query name=uuid_iceberg +SELECT id, u FROM iceberg_scan('s3://test-bucket/default_namespace/uuid_table') ORDER BY id +1 a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11 +2 a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a22 +3 + +# The upsert path writes through the same builder as the initial snapshot, and +# an update also exercises the equality-delete writer for a string column. +> UPDATE uuid_rows SET u = 'ffffffff-ffff-ffff-ffff-ffffffffffff' WHERE id = 1; + +> UPDATE uuid_rows SET u = NULL WHERE id = 2; + +> UPDATE uuid_rows SET u = '11111111-2222-3333-4444-555555555555' WHERE id = 3; + +$ sleep-is-probably-flaky-i-have-justified-my-need-with-a-comment duration=10s + +$ duckdb-query name=uuid_iceberg +SELECT id, u FROM polaris_uuid.default_namespace.uuid_table ORDER BY id +1 ffffffff-ffff-ffff-ffff-ffffffffffff +2 +3 11111111-2222-3333-4444-555555555555 + +> SELECT status FROM mz_internal.mz_sink_statuses WHERE name = 'uuid_sink' +running + +> DROP SINK uuid_sink From 41360296ccde3273ef287c5cf0e9d7e03eb61fff Mon Sep 17 00:00:00 2001 From: Dov Alperin Date: Wed, 23 Sep 2026 11:35:22 -0400 Subject: [PATCH 3/6] controller: use the arrangement maintenance policy for storage replicas (#39041) ### Motivation Storage replicas hardcode an arrangement exertion proportionality of 1337 in `mz_controller::clusters`, while compute replicas read theirs from the `arrangement_exert_proportionality` dyncfg, defaulting to 16. With the storage and compute runtimes unified on one replica, there is no reason for the two kinds of arrangement to run different maintenance policies. The proportionality halves per layer below the largest batch, so 1337 lets the maintenance policy request consolidation of any non-empty layer within about ten levels of the largest one. Because the policy is consulted once per operator activation, every idle scheduling turn can then lift a freshly published small batch into the largest batch on its own, and the largest batch is rewritten once per published batch instead of once per doubling. On a skewed 105-source upsert hydration of about 407 GB of state (8 workers, 400cc), the synchronous v2 stash took 56 to 83 minutes at 1337 and 44 minutes at 16, level with the RocksDB-backed v1 on time and CPU. ### Description Storage's Timely config now reads `arrangement_exert_proportionality`, the same replica-scoped dyncfg compute uses, so both run at 16 by default and move together when it is changed. The value is read once per replica provisioning, as before. Zero still disables optional maintenance. No new dyncfg is introduced. Release note: None. --- src/controller/src/clusters.rs | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/src/controller/src/clusters.rs b/src/controller/src/clusters.rs index 813b4cf62a4e4..bab909173e76b 100644 --- a/src/controller/src/clusters.rs +++ b/src/controller/src/clusters.rs @@ -702,11 +702,6 @@ impl Controller { let persist_pubsub_url = self.persist_pubsub_url.clone(); let secrets_args = self.secrets_args.to_flags(); - // TODO(teskje): use the same values as for compute? - let storage_proto_timely_config = TimelyConfig { - arrangement_exert_proportionality: 1337, - ..Default::default() - }; // These configure the replica's process rather than environmentd's, so // they are `ParameterScope::Replica` and must be read through this // replica's scoped overrides. They are baked into the process @@ -714,9 +709,16 @@ impl Controller { // environment-wide value or the override reaches the replica only when // it is next provisioned. let overrides = self.replica_dyncfg_overrides.get(&replica_id); + // Storage and compute arrangements share one maintenance policy, so a + // unified replica runs both kinds of arrangement under the same reach. + let arrangement_exert_proportionality = + ARRANGEMENT_EXERT_PROPORTIONALITY.get_with_overrides(&self.dyncfg, overrides); + let storage_proto_timely_config = TimelyConfig { + arrangement_exert_proportionality, + ..Default::default() + }; let compute_proto_timely_config = TimelyConfig { - arrangement_exert_proportionality: ARRANGEMENT_EXERT_PROPORTIONALITY - .get_with_overrides(&self.dyncfg, overrides), + arrangement_exert_proportionality, enable_zero_copy: ENABLE_TIMELY_ZERO_COPY.get_with_overrides(&self.dyncfg, overrides), enable_zero_copy_lgalloc: ENABLE_TIMELY_ZERO_COPY_LGALLOC .get_with_overrides(&self.dyncfg, overrides), From cb328365128bf6d4fc186f771952db05987a4f07 Mon Sep 17 00:00:00 2001 From: Sang Jun Bak Date: Wed, 23 Sep 2026 11:42:16 -0400 Subject: [PATCH 4/6] catalog, adapter: convert mz_object_global_ids to a materialized view (#38848) Converts mz_object_global_ids to a builtin materialized view. Read commit messages for more details ### Motivation Closes [SQL-700](https://linear.app/materializeinc/issue/SQL-700) ### Description Part of the builtin migration work. The migration is pretty straight forward. ### Verification - Created a new SLT to test the SQL itself. - Diffed this relation using a customer's workload replay against main and this branch. --- .../content/sql/system-catalog/mz_internal.md | 3 +- .../src/catalog/builtin_table_updates.rs | 27 +-- .../catalog/open/builtin_schema_migration.rs | 9 + src/catalog/src/builtin.rs | 2 +- src/catalog/src/builtin/mz_internal.rs | 114 ++++++++--- src/pgrepr-consts/src/oid.rs | 2 +- test/sqllogictest/catalog_server_explain.slt | 50 ++++- .../information_schema_tables.slt | 2 +- test/sqllogictest/mz_object_global_ids.slt | 187 ++++++++++++++++++ test/testdrive/catalog.td | 4 +- 10 files changed, 335 insertions(+), 65 deletions(-) create mode 100644 test/sqllogictest/mz_object_global_ids.slt diff --git a/doc/user/content/sql/system-catalog/mz_internal.md b/doc/user/content/sql/system-catalog/mz_internal.md index 88fa7ffab9a2d..3afbe03d3ecd2 100644 --- a/doc/user/content/sql/system-catalog/mz_internal.md +++ b/doc/user/content/sql/system-catalog/mz_internal.md @@ -24,7 +24,8 @@ reference these objects is not allowed. ## `mz_object_global_ids` -The `mz_object_global_ids` table maps Materialize catalog item IDs to global IDs. +The `mz_object_global_ids` materialized view maps Materialize catalog item +IDs to global IDs. | Field | Type | Meaning | diff --git a/src/adapter/src/catalog/builtin_table_updates.rs b/src/adapter/src/catalog/builtin_table_updates.rs index b05dc8a6a1de4..b6653b1305ad2 100644 --- a/src/adapter/src/catalog/builtin_table_updates.rs +++ b/src/adapter/src/catalog/builtin_table_updates.rs @@ -18,16 +18,14 @@ use mz_catalog::builtin::{ BuiltinTable, MZ_AGGREGATES, MZ_ARRAY_TYPES, MZ_BASE_TYPES, MZ_CLUSTER_REPLICA_SIZE_INTERNAL, MZ_CLUSTER_REPLICA_SIZES, MZ_COLUMNS, MZ_EGRESS_IPS, MZ_FUNCTIONS, MZ_HISTORY_RETENTION_STRATEGIES, MZ_INDEX_COLUMNS, MZ_LICENSE_KEYS, MZ_LIST_TYPES, - MZ_MAP_TYPES, MZ_MATERIALIZED_VIEW_REFRESH_STRATEGIES, MZ_OBJECT_GLOBAL_IDS, MZ_OPERATORS, - MZ_PSEUDO_TYPES, MZ_REPLACEMENTS, MZ_ROLE_AUTH, MZ_SESSIONS, MZ_SOURCE_REFERENCES, - MZ_STORAGE_USAGE_BY_SHARD, MZ_SUBSCRIPTIONS, MZ_TYPE_PG_METADATA, MZ_TYPES, - MZ_WEBHOOKS_SOURCES, + MZ_MAP_TYPES, MZ_MATERIALIZED_VIEW_REFRESH_STRATEGIES, MZ_OPERATORS, MZ_PSEUDO_TYPES, + MZ_REPLACEMENTS, MZ_ROLE_AUTH, MZ_SESSIONS, MZ_SOURCE_REFERENCES, MZ_STORAGE_USAGE_BY_SHARD, + MZ_SUBSCRIPTIONS, MZ_TYPE_PG_METADATA, MZ_TYPES, MZ_WEBHOOKS_SOURCES, }; use mz_catalog::durable::SourceReferences; use mz_catalog::memory::error::Error; use mz_catalog::memory::objects::{ - CatalogEntry, CatalogItem, DataSourceDesc, Func, Index, MaterializedView, Table, - TableDataSource, Type, + CatalogItem, DataSourceDesc, Func, Index, MaterializedView, Table, TableDataSource, Type, }; use mz_expr::MirScalarExpr; use mz_license_keys::ValidatedLicenseKey; @@ -265,26 +263,9 @@ impl CatalogState { updates.push(self.pack_history_retention_strategy_update(id, cw, diff)); } - updates.extend(Self::pack_item_global_id_update(entry, diff)); - updates } - fn pack_item_global_id_update( - entry: &CatalogEntry, - diff: Diff, - ) -> impl Iterator> + use<'_> { - let id = entry.id().to_string(); - let global_ids = entry.global_ids(); - global_ids.map(move |global_id| { - BuiltinTableUpdate::row( - &*MZ_OBJECT_GLOBAL_IDS, - Row::pack_slice(&[Datum::String(&id), Datum::String(&global_id.to_string())]), - diff, - ) - }) - } - fn pack_history_retention_strategy_update( &self, id: CatalogItemId, diff --git a/src/adapter/src/catalog/open/builtin_schema_migration.rs b/src/adapter/src/catalog/open/builtin_schema_migration.rs index ff7dc229281b6..48e295d21bf2a 100644 --- a/src/adapter/src/catalog/open/builtin_schema_migration.rs +++ b/src/adapter/src/catalog/open/builtin_schema_migration.rs @@ -478,6 +478,15 @@ static MIGRATIONS: LazyLock> = LazyLock::new(|| { MZ_INTERNAL_SCHEMA, "mz_replica_hydration_history", ), + // Converting mz_object_global_ids from a builtin table to a + // materialized view over mz_catalog_raw changes its catalog + // fingerprint. + MigrationStep::replacement( + "26.44.0-dev.0", + CatalogItemType::MaterializedView, + MZ_INTERNAL_SCHEMA, + "mz_object_global_ids", + ), ] }); diff --git a/src/catalog/src/builtin.rs b/src/catalog/src/builtin.rs index fae6d3e356974..7d10a3a92531a 100644 --- a/src/catalog/src/builtin.rs +++ b/src/catalog/src/builtin.rs @@ -1227,7 +1227,7 @@ pub static BUILTINS_STATIC: LazyLock>> = LazyLock::ne Builtin::View(&MZ_OBJECTS_ID_NAMESPACE_TYPES), Builtin::View(&MZ_OBJECT_HISTORY), Builtin::View(&MZ_OBJECT_LIFETIMES), - Builtin::Table(&MZ_OBJECT_GLOBAL_IDS), + Builtin::MaterializedView(&MZ_OBJECT_GLOBAL_IDS), Builtin::View(&MZ_ARRANGEMENT_SHARING_PER_WORKER), Builtin::View(&MZ_ARRANGEMENT_SHARING), Builtin::View(&MZ_ARRANGEMENT_SIZES_PER_WORKER), diff --git a/src/catalog/src/builtin/mz_internal.rs b/src/catalog/src/builtin/mz_internal.rs index 6cea054d21a28..4f60443577a11 100644 --- a/src/catalog/src/builtin/mz_internal.rs +++ b/src/catalog/src/builtin/mz_internal.rs @@ -34,7 +34,8 @@ use crate::memory::objects::DataSourceDesc; use super::{ ANALYTICS_SELECT, BuiltinConnection, BuiltinIndex, BuiltinMaterializedView, BuiltinSource, BuiltinTable, BuiltinView, Cardinality, LinkProperties, MONITOR_REDACTED_SELECT, - MONITOR_SELECT, Ontology, OntologyLink, PUBLIC_SELECT, SUPPORT_SELECT, + MONITOR_SELECT, Ontology, OntologyLink, PUBLIC_SELECT, RUNTIME_ALTERABLE_FINGERPRINT_SENTINEL, + SUPPORT_SELECT, }; pub static MZ_CATALOG_RAW: LazyLock = LazyLock::new(|| BuiltinSource { @@ -4151,36 +4152,89 @@ pub static MZ_OBJECT_FULLY_QUALIFIED_NAMES: LazyLock = LazyLock::ne }), }); -pub static MZ_OBJECT_GLOBAL_IDS: LazyLock = LazyLock::new(|| BuiltinTable { - name: "mz_object_global_ids", - schema: MZ_INTERNAL_SCHEMA, - oid: oid::VIEW_MZ_OBJECT_GLOBAL_IDS_OID, - desc: RelationDesc::builder() - .with_column("id", SqlScalarType::String.nullable(false)) - .with_column("global_id", SqlScalarType::String.nullable(false)) - .finish(), - column_comments: BTreeMap::from_iter([ - ( - "id", - "The ID of the object. Corresponds to `mz_objects.id`.", +pub static MZ_OBJECT_GLOBAL_IDS: LazyLock = + LazyLock::new(|| BuiltinMaterializedView { + name: "mz_object_global_ids", + schema: MZ_INTERNAL_SCHEMA, + oid: oid::MV_MZ_OBJECT_GLOBAL_IDS_OID, + desc: RelationDesc::builder() + .with_column("id", SqlScalarType::String.nullable(false)) + .with_column("global_id", SqlScalarType::String.nullable(false)) + .finish(), + column_comments: BTreeMap::from_iter([ + ( + "id", + "The ID of the object. Corresponds to `mz_objects.id`.", + ), + ("global_id", "The global ID of the object."), + ]), + sql: Box::leak( + format!( + " +IN CLUSTER mz_catalog_server +WITH ( + ASSERT NOT NULL id, + ASSERT NOT NULL global_id +) AS +WITH + -- `Item` rows carry user items, temporary items, and runtime-alterable builtins + items AS ( + SELECT + mz_internal.parse_catalog_id(data->'key'->'gid') AS id, + mz_internal.parse_catalog_id(data->'value'->'global_id') AS global_id, + data->'value'->'extra_versions' AS extra_versions + FROM mz_internal.mz_catalog_raw + WHERE data->>'kind' = 'Item' + ), + -- Create a one to many mapping between an object and its global IDs. Tables and + -- materialized views are the only objects with multiple global IDs. + item_versions AS ( + SELECT i.id, mz_internal.parse_catalog_id(v.version->'global_id') AS global_id + FROM items i + CROSS JOIN LATERAL jsonb_array_elements(i.extra_versions) AS v(version) + ), + builtin_mappings AS ( + SELECT + 's' || (data->'value'->>'catalog_id') AS id, + 's' || (data->'value'->>'global_id') AS global_id + FROM mz_internal.mz_catalog_raw + WHERE + data->>'kind' = 'GidMapping' AND + -- Exclude runtime-alterable builtins since they're already included in `items` + data->'value'->>'fingerprint' != '{RUNTIME_ALTERABLE_FINGERPRINT_SENTINEL}' + ), + introspection_source_indexes AS ( + SELECT + 'si' || (data->'value'->>'catalog_id') AS id, + 'si' || (data->'value'->>'global_id') AS global_id + FROM mz_internal.mz_catalog_raw + WHERE data->>'kind' = 'ClusterIntrospectionSourceIndex' + ) +SELECT id, global_id FROM items +UNION ALL +SELECT id, global_id FROM item_versions +UNION ALL +SELECT id, global_id FROM builtin_mappings +UNION ALL +SELECT id, global_id FROM introspection_source_indexes" + ) + .into_boxed_str(), ), - ("global_id", "The global ID of the object."), - ]), - is_retained_metrics_object: false, - access: vec![PUBLIC_SELECT], - ontology: Some(Ontology { - entity_name: "object_global_id", - description: "Mapping between CatalogItemId (SQL layer) and GlobalId (runtime layer)", - links: &const { - [OntologyLink { - name: "id_references", - target: "object", - properties: LinkProperties::fk("id", "id", Cardinality::ManyToOne), - }] - }, - column_semantic_types: &[("id", SemanticType::CatalogItemId)], - }), -}); + is_retained_metrics_object: false, + access: vec![PUBLIC_SELECT], + ontology: Some(Ontology { + entity_name: "object_global_id", + description: "Mapping between CatalogItemId (SQL layer) and GlobalId (runtime layer)", + links: &const { + [OntologyLink { + name: "id_references", + target: "object", + properties: LinkProperties::fk("id", "id", Cardinality::ManyToOne), + }] + }, + column_semantic_types: &[("id", SemanticType::CatalogItemId)], + }), + }); // TODO (SangJunBak): Remove once mz_object_history is released and used in the Console https://github.com/MaterializeInc/console/issues/3342 pub static MZ_OBJECT_LIFETIMES: LazyLock = LazyLock::new(|| BuiltinView { diff --git a/src/pgrepr-consts/src/oid.rs b/src/pgrepr-consts/src/oid.rs index d768fbba66338..5d0e360c6459f 100644 --- a/src/pgrepr-consts/src/oid.rs +++ b/src/pgrepr-consts/src/oid.rs @@ -771,7 +771,7 @@ pub const MV_MZ_SQL_SERVER_SOURCE_TABLES_OID: u32 = 17057; pub const TABLE_MZ_LICENSE_KEYS_OID: u32 = 17058; pub const TABLE_MZ_ROLE_AUTH_OID: u32 = 17059; pub const MV_MZ_ICEBERG_SINKS_OID: u32 = 17060; -pub const VIEW_MZ_OBJECT_GLOBAL_IDS_OID: u32 = 17061; +pub const MV_MZ_OBJECT_GLOBAL_IDS_OID: u32 = 17061; pub const TABLE_MZ_REPLACEMENTS_OID: u32 = 17062; pub const VIEW_PG_AUTHID_CORE_OID: u32 = 17063; pub const INDEX_PG_AUTHID_CORE_IND_OID: u32 = 17064; diff --git a/test/sqllogictest/catalog_server_explain.slt b/test/sqllogictest/catalog_server_explain.slt index 113e1100c41d0..8196ea04a4222 100644 --- a/test/sqllogictest/catalog_server_explain.slt +++ b/test/sqllogictest/catalog_server_explain.slt @@ -5127,7 +5127,7 @@ mz_catalog.mz_materialized_views: Project: #4, #0, #5, #1, #6, #2, #9, #8, #3, #7 Map: {=r/s1, s1=r/s1}, "s1" →Arrange (#1{schema_name}, #2{name}) - →Constant (44 rows) + →Constant (45 rows) →Arrange (empty key) (#0{schema_name}) (#0{schema_name}, #1{name}) →Fused with Child Map/Filter/Project Project: #4, #3, #5 @@ -5472,7 +5472,7 @@ mz_catalog.mz_tables: Project: #3, #0, #4, #1, #5, #2, #6, #6, #6 Map: "s1", null →Arrange (#1{schema_name}, #2{name}) - →Constant (30 rows) + →Constant (29 rows) →Arrange (#0{schema_name}) (#0{schema_name}, #1{name}) →Fused with Child Map/Filter/Project Project: #5, #4, #6 @@ -5889,7 +5889,7 @@ mz_internal.mz_object_dependencies: Project: #4..=#6, #2 Filter: case when (#7{ref_kind} = "func") then (#3{object_type} = "8") else case when (#7{ref_kind} = "type") then (#3{object_type} = "7") else ((#3{object_type} != "7") AND (#3{object_type} != "8")) end end →Arrange (#0{object_schema}..=#2{object_type}) (#3{ref_schema}, #4{ref_name}) - →Constant (1024 rows) + →Constant (1027 rows) →Arrange (#1{schema_name}..=#3{object_type}) →Fused with Child Map/Filter/Project Project: #5, #4, #2, #3 @@ -5925,6 +5925,44 @@ Target cluster: mz_catalog_server EOF +query T multiline +EXPLAIN MATERIALIZED VIEW "mz_internal"."mz_object_global_ids"; +---- +mz_internal.mz_object_global_ids: + →Union + →Fused with Child Map/Filter/Project + Project: #3, #4 + Filter: ("Item" = #2) + Map: parse_catalog_id(((#0{data} -> "key") -> "gid")), parse_catalog_id((#1 -> "global_id")) + →Read mz_internal.mz_catalog_raw + →Fused with Child Map/Filter/Project + Project: #0, #3 + Map: parse_catalog_id((#2{version} -> "global_id")) + →Table Function jsonb_array_elements(#1{extra_versions}) + →Fused with Child Map/Filter/Project + Project: #3, #4 + Filter: ("Item" = #2) + Map: parse_catalog_id(((#0{data} -> "key") -> "gid")), (#1 -> "extra_versions") + →Read mz_internal.mz_catalog_raw + →Fused with Child Map/Filter/Project + Project: #3, #4 + Filter: ("" != (#1 ->> "fingerprint")) AND ("GidMapping" = #2) + Map: ("s" || (#1 ->> "catalog_id")), ("s" || (#1 ->> "global_id")) + →Read mz_internal.mz_catalog_raw + →Fused with Child Map/Filter/Project + Project: #3, #4 + Filter: ("ClusterIntrospectionSourceIndex" = #2) + Map: ("si" || (#1 ->> "catalog_id")), ("si" || (#1 ->> "global_id")) + →Read mz_internal.mz_catalog_raw + +Source mz_internal.mz_catalog_raw + project=(#0..=#2) + map=((#0{data} -> "value"), (#0{data} ->> "kind")) + +Target cluster: mz_catalog_server + +EOF + query T multiline EXPLAIN MATERIALIZED VIEW "mz_internal"."mz_overridden_system_parameters"; ---- @@ -8336,7 +8374,7 @@ query T multiline EXPLAIN SELECT * FROM "mz_internal"."mz_builtin_materialized_views"; ---- Explained Query (fast path): - →Constant (44 rows) + →Constant (45 rows) Target cluster: mz_catalog_server @@ -8356,7 +8394,7 @@ query T multiline EXPLAIN SELECT * FROM "mz_internal"."mz_builtin_tables"; ---- Explained Query (fast path): - →Constant (30 rows) + →Constant (29 rows) Target cluster: mz_catalog_server @@ -10284,7 +10322,7 @@ Explained Query: Project: #3, #4 Filter: case when (#2{ref_kind} = "func") then (#5{object_type} = "8") else case when (#2{ref_kind} = "type") then (#5{object_type} = "7") else ((#5{object_type} != "7") AND (#5{object_type} != "8")) end end →Unarranged Raw Stream - →Constant (1024 rows) + →Constant (1027 rows) →Arrange (#1{schema_name}..=#3{object_type}) →Fused with Child Map/Filter/Project Project: #5, #4, #2, #3 diff --git a/test/sqllogictest/information_schema_tables.slt b/test/sqllogictest/information_schema_tables.slt index 5b27ee141814f..eef88c8d3ba9a 100644 --- a/test/sqllogictest/information_schema_tables.slt +++ b/test/sqllogictest/information_schema_tables.slt @@ -506,7 +506,7 @@ VIEW materialize mz_internal mz_object_global_ids -BASE TABLE +MATERIALIZED VIEW materialize mz_internal mz_object_graph_edges diff --git a/test/sqllogictest/mz_object_global_ids.slt b/test/sqllogictest/mz_object_global_ids.slt new file mode 100644 index 0000000000000..df7e53dc79083 --- /dev/null +++ b/test/sqllogictest/mz_object_global_ids.slt @@ -0,0 +1,187 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +# Behavioural tests for mz_internal.mz_object_global_ids. + +mode cockroach + +reset-server + +simple conn=mz_system,user=mz_system +ALTER SYSTEM SET enable_alter_table_add_column = true +---- +COMPLETE 0 + +statement ok +CREATE SCHEMA gids + +statement ok +SET search_path = gids + +# Setup multiple user items with one global ID. +statement ok +CREATE TABLE t (a int) + +statement ok +CREATE VIEW v AS SELECT a FROM t + +statement ok +CREATE MATERIALIZED VIEW mv AS SELECT count(*) AS c FROM t + +statement ok +CREATE INDEX t_idx ON t (a) + +statement ok +CREATE SECRET pw AS 'hunter2' + +statement ok +CREATE CONNECTION kc TO KAFKA (BROKER 'kafka:9092', SASL MECHANISMS 'PLAIN', SASL USERNAME 'u', SASL PASSWORD = SECRET pw) WITH (VALIDATE = false) + +statement ok +CREATE TYPE custom_list AS LIST (ELEMENT TYPE = int4) + +statement ok +CREATE SOURCE lg FROM LOAD GENERATOR COUNTER + +statement ok +CREATE TABLE lg_tbl FROM SOURCE lg (REFERENCE counter) + +# Every user item in the schema has exactly one global ID. +query TT rowsort +SELECT o.name, count(*) +FROM mz_internal.mz_object_global_ids g +JOIN mz_objects o ON o.id = g.id +JOIN mz_schemas s ON s.id = o.schema_id +WHERE s.name = 'gids' +GROUP BY o.name +---- +custom_list 1 +kc 1 +lg 1 +lg_tbl 1 +mv 1 +pw 1 +t 1 +t_idx 1 +v 1 + +# Setup a table that will have multiple versions +statement ok +CREATE TABLE versioned (a int) + +# A new column adds a version, and each version has its own GlobalId. +statement ok +ALTER TABLE versioned ADD COLUMN b text + +query T +SELECT count(*) +FROM mz_internal.mz_object_global_ids +WHERE id = (SELECT id FROM mz_tables WHERE name = 'versioned') +---- +2 + +statement ok +ALTER TABLE versioned ADD COLUMN c int + +query T +SELECT count(*) +FROM mz_internal.mz_object_global_ids +WHERE id = (SELECT id FROM mz_tables WHERE name = 'versioned') +---- +3 + +# APPLY REPLACEMENT moves a materialized view to the replacement's catalog ID. +# Assert that one of its global IDs is not equal to its catalog ID and that +# the global IDs are unique. +statement ok +CREATE MATERIALIZED VIEW rmv AS SELECT a FROM t + +statement ok +CREATE REPLACEMENT MATERIALIZED VIEW rmv_r FOR rmv AS SELECT a FROM t WHERE a IS DISTINCT FROM 0 + +statement ok +ALTER MATERIALIZED VIEW rmv APPLY REPLACEMENT rmv_r + +query IB +SELECT count(DISTINCT global_id), bool_and(global_id = id) +FROM mz_internal.mz_object_global_ids +WHERE id = (SELECT id FROM mz_materialized_views WHERE name = 'rmv') +---- +2 false + +# Global IDs are unique, and the mapping covers mz_objects exactly: no +# duplicate global ID, no object without a row, no row without an object. +query III +SELECT + (SELECT count(*) FROM ( + SELECT global_id FROM mz_internal.mz_object_global_ids + GROUP BY global_id HAVING count(*) > 1)), + (SELECT count(*) FROM mz_objects o WHERE NOT EXISTS ( + SELECT 1 FROM mz_internal.mz_object_global_ids g WHERE g.id = o.id)), + (SELECT count(*) FROM mz_internal.mz_object_global_ids g WHERE NOT EXISTS ( + SELECT 1 FROM mz_objects o WHERE o.id = g.id)) +---- +0 0 0 + +# Assert that temporary objects are included. + +statement ok +CREATE TEMPORARY VIEW tv AS SELECT a FROM t + +query T +SELECT count(*) +FROM mz_internal.mz_object_global_ids +WHERE id = (SELECT id FROM mz_views WHERE name = 'tv') +---- +1 + + +# Assert builtin items are included. +query T +SELECT count(*) +FROM mz_internal.mz_object_global_ids +WHERE id = (SELECT id FROM mz_views WHERE name = 'mz_object_fully_qualified_names') +---- +1 + +query T +SELECT count(*) > 0 +FROM mz_internal.mz_object_global_ids g +JOIN mz_functions f ON f.id = g.id +WHERE f.name = 'parse_catalog_id' +---- +true + +# mz_analytics is a runtime-alterable builtin. Assert it has exactly one row. +query T +SELECT count(*) +FROM mz_internal.mz_object_global_ids +WHERE id = (SELECT id FROM mz_connections WHERE name = 'mz_analytics') +---- +1 + +# Introspection source indexes are recorded in their own durable collection. +query T +SELECT count(*) > 0 FROM mz_internal.mz_object_global_ids WHERE id LIKE 'si%' +---- +true + +# Assert that dropped items leave no rows. + +# Drop the temporary view explicitly since it's not in the `gids` schema. +statement ok +DROP VIEW tv + +statement ok +DROP SCHEMA gids CASCADE + +query T +SELECT count(*) FROM mz_internal.mz_object_global_ids WHERE id LIKE 'u%' +---- +0 diff --git a/test/testdrive/catalog.td b/test/testdrive/catalog.td index 645afa00580b8..d0baa38912fd3 100644 --- a/test/testdrive/catalog.td +++ b/test/testdrive/catalog.td @@ -608,7 +608,6 @@ mz_history_retention_strategies "" mz_license_keys "" mz_materialized_view_refresh_strategies "" mz_object_arrangement_size_history "" -mz_object_global_ids "" mz_object_hydration_history "" mz_optimizer_notices "" mz_replacements "" @@ -735,6 +734,7 @@ mz_mysql_source_tables mz_catalog_server "" mz_network_policies mz_catalog_server "" mz_network_policy_rules mz_catalog_server "" mz_object_dependencies mz_catalog_server "" +mz_object_global_ids mz_catalog_server "" mz_overridden_system_parameters mz_catalog_server "" mz_pending_cluster_replicas mz_catalog_server "" mz_postgres_source_tables mz_catalog_server "" @@ -838,7 +838,7 @@ test_table "" # `SHOW TABLES` and `mz_tables` should agree. > SELECT COUNT(*) FROM mz_tables WHERE id LIKE 's%' -30 +29 # There is one entry in mz_indexes for each field_number/expression of the index. > SELECT COUNT(id) FROM mz_indexes WHERE id LIKE 's%' From a628512863763f828e6c49b33bce6e773efd7f2a Mon Sep 17 00:00:00 2001 From: Qindeel Ishtiaq Date: Wed, 23 Sep 2026 12:08:56 -0400 Subject: [PATCH 5/6] catalog: add memory and swap columns to console utilization views (#39034) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### Motivation Swap metrics shipped in v26.43.0 on `mz_cluster_replica_metrics`, but the Console's cluster utilization graphs don't read that relation. They read the indexed `mz_console_cluster_utilization_overview*` views, which have no swap column, so the data can't reach the UI. ### Change Three columns added to both view bodies, and so to all three indexed views: ``` ram_percent = MAX(memory_bytes) / memory_bytes_per_process swap_percent = MAX(swap_bytes) / memory_bytes_per_process heap_limit_percent = MAX(heap_limit) / memory_bytes_per_process ``` All three are denominated in a single process's RAM allocation so they share one axis: `ram_percent` is the bar, `swap_percent` stacks on top of it, and `heap_limit_percent` marks where RAM + swap runs out, above 1.0. A replica that is spilling draws above the 100% RAM line — the state the existing columns cannot express. ### MAX rather than SUM These aggregate across processes with `MAX`, joining `heap_percent` rather than `cpu_percent`/`memory_percent`/`disk_percent`, which are `SUM`. A replica dies when any *one* process is OOM-killed, not when its average crosses a limit. Four processes at 63/6/6/6 GiB against a 64 GiB per-process limit is 32% under SUM and 98% under MAX; the replica is killed either way. On a chart with a RAM limit line drawn on it, SUM would show a comfortable bar for a replica that is actually crossing that limit. Swap is what happens *at* that limit, so the reasoning that already made `heap_percent` a worst-process metric applies here too. At `scale=1` the two are identical, so this only changes the multi-process sizes. ### Cost on mz_catalog_server Peak RAM and peak swap are taken as **independent** `MAX`es. On a skewed multi-process replica they can describe different processes, making the stacked total an upper bound rather than an observed state. Selecting them row-consistently needs an argmax over processes (`DISTINCT ON (occurred_at, replica_id) ORDER BY heap_bytes DESC`, joined back). I implemented and measured that: **+6 operators and +3 arrangements in each of the three indexed views**. Since idle catalog_server cost tracks operator count rather than data volume, that is a real standing charge on a shared cluster. The independent form folds into the hierarchical aggregate already present in each body — `Aggregations: max, max` becomes `max, max, max, max, max` — leaving counts identical to before: | View | operators | arrangements | |---|---|---| | `_overview` | 80 → 80 | 20 → 20 | | `_overview_24h` | 80 → 80 | 20 → 20 | | `_overview_3h` | 33 → 33 | 9 → 9 | If the exact-stack semantics turn out to matter in practice, the argmax version is a contained change with a known price. ### Notes for reviewers - Values are fractions in `0..1`, matching the other `_percent` columns in these views, which the Console scales client-side. This differs from `mz_cluster_replica_utilization`, where `swap_percent` is scaled in the view itself. - No builtin schema migration entry is needed — these are `BuiltinView`s, not persist-backed collections. - `swap_bytes` is null on replica sizes with swap disabled and on the process orchestrator, so `swap_percent` is null there rather than zero. `heap_limit` has its own null path via the `/proc/meminfo` fallback in `collect_heap_limit()`. - Unlike `heap_percent`, these have no SUM fallback when `heap_limit` is null; they go null instead, which seemed more honest for a chart with a limit line on it. ### Follow-up A Console PR consumes these as a stacked memory/swap chart, plus a cluster-list column. It needs version-gated query builders, since the three builders in `replicaUtilizationHistory.ts` currently have none and would break on environments predating these columns. ### Tests `test_smoketest_all_builtins` covers this — it plans every builtin and asserts the declared `RelationDesc` matches the SQL's output columns. Regenerated `catalog_server_explain.slt` and `mz_catalog_server_index_accounting.slt`. ### Release notes No user-facing release note. These views are `mz_internal` and marked `RELATION_SPEC_UNDOCUMENTED`; they exist for the Console. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --- src/catalog/src/builtin/mz_internal.rs | 52 ++++++++- test/sqllogictest/catalog_server_explain.slt | 104 +++++++++--------- test/sqllogictest/cockroach/srfs.slt | 4 +- .../console_cluster_utilization_shares.slt | 101 +++++++++++++++++ .../mz_catalog_server_index_accounting.slt | 9 ++ test/testdrive/explain-analyze.td | 4 +- 6 files changed, 215 insertions(+), 59 deletions(-) create mode 100644 test/sqllogictest/console_cluster_utilization_shares.slt diff --git a/src/catalog/src/builtin/mz_internal.rs b/src/catalog/src/builtin/mz_internal.rs index 4f60443577a11..92e7ba0f58ca3 100644 --- a/src/catalog/src/builtin/mz_internal.rs +++ b/src/catalog/src/builtin/mz_internal.rs @@ -7952,6 +7952,31 @@ JOIN root_times r USING (id)", }, }), }); +/// Worst-process shares of one process's RAM allocation, shared by the console +/// utilization view bodies. All three divide by the same denominator so they +/// stack on one axis: `ram_percent` is the bar, `swap_of_ram_percent` sits on +/// top, and `heap_limit_percent` marks where RAM plus swap runs out, above 1.0. +/// Assumes the metrics relation is bound as `m` and the sizes relation as `s`. +/// +/// NOTE: worst-process shares, not replica totals. `memory_percent` alongside +/// them sums across processes instead; the two agree at `scale=1`, but above it +/// these report "the worst process is at 98% of its own allocation" where +/// `memory_percent` reports "the replica is at 32% overall". A replica dies on +/// one process being OOM-killed, so the chart's limit line wants the former. +/// +/// NOTE: peak RAM and peak swap are independent maxima, so under multi-process +/// skew they can describe different processes and the stack is an upper bound. +/// +/// NOTE: null, not zero, where a size cannot swap or `heap_limit` goes +/// unreported. Coalescing would report "swapped nothing" for "cannot swap". +/// +/// NOTE: these recombine to `heap_percent` where the maxima agree, so prefer +/// that column over a fourth spelling of the same quantity. +const CONSOLE_MEMORY_SHARES_SQL: &str = "\ + MAX(m.memory_bytes::float8) / NULLIF(s.memory_bytes, 0) AS ram_percent, + MAX(m.swap_bytes::float8) / NULLIF(s.memory_bytes, 0) AS swap_of_ram_percent, + MAX(m.heap_limit::float8) / NULLIF(s.memory_bytes, 0) AS heap_limit_percent"; + /// The output relation shared by all `mz_console_cluster_utilization_overview*` /// views. Every (bucket size, retention) variant produces the same columns so /// the Console can swap between them based on the selected time range. @@ -7993,6 +8018,9 @@ fn console_cluster_utilization_overview_desc() -> RelationDesc { "max_heap_at", SqlScalarType::TimestampTz { precision: None }.nullable(false), ) + .with_column("ram_percent", SqlScalarType::Float64.nullable(true)) + .with_column("swap_of_ram_percent", SqlScalarType::Float64.nullable(true)) + .with_column("heap_limit_percent", SqlScalarType::Float64.nullable(true)) .with_column("max_cpu_percent", SqlScalarType::Float64.nullable(true)) .with_column( "max_cpu_at", @@ -8055,7 +8083,8 @@ replica_metrics_history AS ( COALESCE( MAX(m.heap_bytes::float8 / NULLIF(m.heap_limit, 0)), SUM(m.memory_bytes::float8) / NULLIF(s.memory_bytes, 0) / NULLIF(s.processes, 0) - ) AS heap_percent + ) AS heap_percent, +{shares} FROM replica_history AS r INNER JOIN mz_catalog.mz_cluster_replica_sizes AS s ON r.size = s.size @@ -8085,6 +8114,9 @@ replica_utilization_history_binned AS ( m.total_memory_bytes, m.heap_bytes, m.heap_percent, + m.ram_percent, + m.swap_of_ram_percent, + m.heap_limit_percent, m.size, date_bin('{bin}', m.occurred_at, '1970-01-01'::timestamp) AS bucket_start FROM replica_metrics_history AS m @@ -8130,9 +8162,11 @@ max_memory_and_disk AS ( OPTIONS (DISTINCT ON INPUT GROUP SIZE = {group_size}) ORDER BY bucket_start, replica_id, COALESCE(memory_and_disk_percent, 0) DESC ), --- For each (replica, bucket), take the sample with the highest heap. +-- For each (replica, bucket), take the sample with the highest heap. The RAM +-- and swap shares ride along, so they describe that sample rather than the +-- bucket's peak RAM, which is what max_memory reports. max_heap AS ( - SELECT DISTINCT ON (bucket_start, replica_id) bucket_start, replica_id, heap_percent, occurred_at + SELECT DISTINCT ON (bucket_start, replica_id) bucket_start, replica_id, heap_percent, ram_percent, swap_of_ram_percent, heap_limit_percent, occurred_at FROM replica_utilization_history_binned OPTIONS (DISTINCT ON INPUT GROUP SIZE = {group_size}) ORDER BY bucket_start, replica_id, COALESCE(heap_bytes, 0) DESC @@ -8170,6 +8204,9 @@ SELECT max_memory_and_disk.occurred_at AS max_memory_and_disk_at, max_heap.heap_percent, max_heap.occurred_at AS max_heap_at, + max_heap.ram_percent, + max_heap.swap_of_ram_percent, + max_heap.heap_limit_percent, max_cpu.cpu_percent AS max_cpu_percent, max_cpu.occurred_at AS max_cpu_at, replica_offline_event_history.offline_events, @@ -8200,6 +8237,7 @@ LEFT JOIN replica_offline_event_history USING (bucket_start, replica_id)"#, bin = bin, retention = retention, group_size = group_size, + shares = CONSOLE_MEMORY_SHARES_SQL, ) } @@ -8220,6 +8258,9 @@ fn console_cluster_utilization_unbinned_3h_desc() -> RelationDesc { .with_column("memory_percent", SqlScalarType::Float64.nullable(true)) .with_column("disk_percent", SqlScalarType::Float64.nullable(true)) .with_column("heap_percent", SqlScalarType::Float64.nullable(true)) + .with_column("ram_percent", SqlScalarType::Float64.nullable(true)) + .with_column("swap_of_ram_percent", SqlScalarType::Float64.nullable(true)) + .with_column("heap_limit_percent", SqlScalarType::Float64.nullable(true)) .with_column( "memory_and_disk_percent", SqlScalarType::Float64.nullable(true), @@ -8265,6 +8306,7 @@ replica_metrics AS ( MAX(m.heap_bytes::float8 / NULLIF(m.heap_limit, 0)), SUM(m.memory_bytes::float8) / NULLIF(s.memory_bytes, 0) / NULLIF(s.processes, 0) ) AS heap_percent, +{shares}, CASE WHEN SUM(m.disk_bytes::float8) IS NULL AND SUM(m.memory_bytes::float8) IS NULL THEN NULL ELSE (COALESCE(SUM(m.memory_bytes::float8), 0) + COALESCE(SUM(m.disk_bytes::float8), 0)) @@ -8297,6 +8339,9 @@ SELECT m.memory_percent, m.disk_percent, m.heap_percent, + m.ram_percent, + m.swap_of_ram_percent, + m.heap_limit_percent, m.memory_and_disk_percent FROM replica_metrics AS m /* Most recent replica name as of the sample time. */ @@ -8310,6 +8355,7 @@ CROSS JOIN LATERAL ( LIMIT 1 ) AS replica_name_history"#, retention = retention, + shares = CONSOLE_MEMORY_SHARES_SQL, ) } diff --git a/test/sqllogictest/catalog_server_explain.slt b/test/sqllogictest/catalog_server_explain.slt index 8196ea04a4222..898186c238260 100644 --- a/test/sqllogictest/catalog_server_explain.slt +++ b/test/sqllogictest/catalog_server_explain.slt @@ -857,7 +857,7 @@ query T multiline EXPLAIN INDEX "mz_internal"."mz_console_cluster_utilization_overview_24h_ind"; ---- mz_internal.mz_console_cluster_utilization_overview_24h_ind: - →Arrange (#17{cluster_id}) + →Arrange (#20{cluster_id}) →Stream mz_internal.mz_console_cluster_utilization_overview_24h mz_internal.mz_console_cluster_utilization_overview_24h: @@ -889,14 +889,14 @@ mz_internal.mz_console_cluster_utilization_overview_24h: cte l2 = →Differential Join %0[#0..=#6] » %1[#0..=#6] after %1: - Project: #0, #1, #7, #10, #11, #13..=#15, #17..=#20 - Map: uint8_to_double(case when (0 = uint8_to_numeric(#6{processes})) then null else #6{processes} end), ((#9 / uint8_to_double(case when (0 = uint8_to_numeric(#3{cpu_nano_cores})) then null else #3{cpu_nano_cores} end)) / #12), ((#10 / uint8_to_double(case when (0 = uint8_to_numeric(#4{memory_bytes})) then null else #4{memory_bytes} end)) / #12), ((#11 / uint8_to_double(case when (0 = uint8_to_numeric(#5{disk_bytes})) then null else #5{disk_bytes} end)) / #12), uint8_to_double(#6{processes}), (uint8_to_double(#5{disk_bytes}) * #16), (uint8_to_double(#4{memory_bytes}) * #16), coalesce(#8, #14), timestamptz_bin(00:05:00, #0{occurred_at}, 1970-01-01 00:00:00 UTC) + Project: #0, #1, #7, #13, #14, #16, #18, #19, #21..=#27 + Map: uint8_to_double(case when (0 = uint8_to_numeric(#6{processes})) then null else #6{processes} end), ((#12 / uint8_to_double(case when (0 = uint8_to_numeric(#3{cpu_nano_cores})) then null else #3{cpu_nano_cores} end)) / #15), uint8_to_double(case when (0 = uint8_to_numeric(#4{memory_bytes})) then null else #4{memory_bytes} end), ((#13 / #17) / #15), ((#14 / uint8_to_double(case when (0 = uint8_to_numeric(#5{disk_bytes})) then null else #5{disk_bytes} end)) / #15), uint8_to_double(#6{processes}), (uint8_to_double(#5{disk_bytes}) * #20), (uint8_to_double(#4{memory_bytes}) * #20), coalesce(#8, #18), (#9 / #17), (#10 / #17), (#11 / #17), timestamptz_bin(00:05:00, #0{occurred_at}, 1970-01-01 00:00:00 UTC) →Temporally-Bucketed Bucketed Hierarchical GroupAggregate (buckets: 268435456 16777216 1048576 65536 4096 256 16) - Aggregations: max, max + Aggregations: max, max, max, max, max Key: - Project: #6, #0, #1, #3..=#5, #2 + Project: #7, #0, #1, #3..=#5, #2 →Fused with Child Map/Filter/Project - Project: #0..=#5, #9..=#11 + Project: #0..=#5, #7, #9..=#12 →Read l1 →Temporally-Bucketed Accumulable GroupAggregate Simple aggregates: sum(uint8_to_double(#6{cpu_nano_cores})), sum(uint8_to_double(#7{memory_bytes})), sum(uint8_to_double(#8{disk_bytes})) @@ -906,7 +906,7 @@ mz_internal.mz_console_cluster_utilization_overview_24h: Project: #0..=#9 →Read l1 cte l3 = - →Delta Join [%0[#1{replica_id}] » %1[#1{replica_id}, #3{bucket_start}] » %2[#1{replica_id}, #3{bucket_start}] » %3[#1{replica_id}, #4{bucket_start}] » %4[#1{replica_id}, #3{bucket_start}] » %5:l0[#0{replica_id}]] [%1[#1{replica_id}, #3{bucket_start}] » %0[#1{replica_id}, #3{bucket_start}] » %2[#1{replica_id}, #3{bucket_start}] » %3[#1{replica_id}, #4{bucket_start}] » %4[#1{replica_id}, #3{bucket_start}] » %5:l0[#0{replica_id}]] [%2[#1{replica_id}, #3{bucket_start}] » %0[#1{replica_id}, #3{bucket_start}] » %1[#1{replica_id}, #3{bucket_start}] » %3[#1{replica_id}, #4{bucket_start}] » %4[#1{replica_id}, #3{bucket_start}] » %5:l0[#0{replica_id}]] [%3[#1{replica_id}, #4{bucket_start}] » %0[#1{replica_id}, #3{bucket_start}] » %1[#1{replica_id}, #3{bucket_start}] » %2[#1{replica_id}, #3{bucket_start}] » %4[#1{replica_id}, #3{bucket_start}] » %5:l0[#0{replica_id}]] [%4[#1{replica_id}, #3{bucket_start}] » %0[#1{replica_id}, #3{bucket_start}] » %1[#1{replica_id}, #3{bucket_start}] » %2[#1{replica_id}, #3{bucket_start}] » %3[#1{replica_id}, #4{bucket_start}] » %5:l0[#0{replica_id}]] [%5:l0[#0{replica_id}] » %0[#1{replica_id}] » %1[#1{replica_id}, #3{bucket_start}] » %2[#1{replica_id}, #3{bucket_start}] » %3[#1{replica_id}, #4{bucket_start}] » %4[#1{replica_id}, #3{bucket_start}]] + →Delta Join [%0[#1{replica_id}] » %1[#1{replica_id}, #3{bucket_start}] » %2[#1{replica_id}, #3{bucket_start}] » %3[#1{replica_id}, #4{bucket_start}] » %4[#1{replica_id}, #6{bucket_start}] » %5:l0[#0{replica_id}]] [%1[#1{replica_id}, #3{bucket_start}] » %0[#1{replica_id}, #3{bucket_start}] » %2[#1{replica_id}, #3{bucket_start}] » %3[#1{replica_id}, #4{bucket_start}] » %4[#1{replica_id}, #6{bucket_start}] » %5:l0[#0{replica_id}]] [%2[#1{replica_id}, #3{bucket_start}] » %0[#1{replica_id}, #3{bucket_start}] » %1[#1{replica_id}, #3{bucket_start}] » %3[#1{replica_id}, #4{bucket_start}] » %4[#1{replica_id}, #6{bucket_start}] » %5:l0[#0{replica_id}]] [%3[#1{replica_id}, #4{bucket_start}] » %0[#1{replica_id}, #3{bucket_start}] » %1[#1{replica_id}, #3{bucket_start}] » %2[#1{replica_id}, #3{bucket_start}] » %4[#1{replica_id}, #6{bucket_start}] » %5:l0[#0{replica_id}]] [%4[#1{replica_id}, #6{bucket_start}] » %0[#1{replica_id}, #3{bucket_start}] » %1[#1{replica_id}, #3{bucket_start}] » %2[#1{replica_id}, #3{bucket_start}] » %3[#1{replica_id}, #4{bucket_start}] » %5:l0[#0{replica_id}]] [%5:l0[#0{replica_id}] » %0[#1{replica_id}] » %1[#1{replica_id}, #3{bucket_start}] » %2[#1{replica_id}, #3{bucket_start}] » %3[#1{replica_id}, #4{bucket_start}] » %4[#1{replica_id}, #6{bucket_start}]] →Arrange (#1{replica_id}) (#1{replica_id}, #3{bucket_start}) →Map/Filter/Project Project: #0..=#3 @@ -915,7 +915,7 @@ mz_internal.mz_console_cluster_utilization_overview_24h: Order By #4 desc nulls_first Limit 1 →Fused with Child Map/Filter/Project - Project: #0, #1, #6, #11, #12 + Project: #0, #1, #6, #14, #15 Map: coalesce(#3{memory_bytes}, 0) →Read l2 →Arrange (#1{replica_id}, #3{bucket_start}) @@ -926,7 +926,7 @@ mz_internal.mz_console_cluster_utilization_overview_24h: Order By #4 desc nulls_first Limit 1 →Fused with Child Map/Filter/Project - Project: #0, #1, #7, #11, #12 + Project: #0, #1, #7, #14, #15 Map: coalesce(#4{disk_bytes}, 0) →Read l2 →Arrange (#1{replica_id}, #3{bucket_start}) @@ -937,7 +937,7 @@ mz_internal.mz_console_cluster_utilization_overview_24h: Order By #4 desc nulls_first Limit 1 →Fused with Child Map/Filter/Project - Project: #0, #1, #5, #11, #12 + Project: #0, #1, #5, #14, #15 Map: coalesce(#5{cpu_percent}, 0) →Read l2 →Arrange (#1{replica_id}, #4{bucket_start}) @@ -948,18 +948,18 @@ mz_internal.mz_console_cluster_utilization_overview_24h: Order By #6 desc nulls_first Limit 1 →Fused with Child Map/Filter/Project - Project: #0, #1, #6, #7, #11..=#13 - Map: case when ((#3{memory_bytes}) IS NULL AND (#4{disk_bytes}) IS NULL) then null else ((coalesce(#3{memory_bytes}, 0) + coalesce(#4{disk_bytes}, 0)) / case when (0 = (#9{total_memory_bytes} + #8{total_disk_bytes})) then null else (#9{total_memory_bytes} + #8{total_disk_bytes}) end) end, coalesce(#12{memory_and_disk_percent}, 0) + Project: #0, #1, #6, #7, #14..=#16 + Map: case when ((#3{memory_bytes}) IS NULL AND (#4{disk_bytes}) IS NULL) then null else ((coalesce(#3{memory_bytes}, 0) + coalesce(#4{disk_bytes}, 0)) / case when (0 = (#9{total_memory_bytes} + #8{total_disk_bytes})) then null else (#9{total_memory_bytes} + #8{total_disk_bytes}) end) end, coalesce(#15{memory_and_disk_percent}, 0) →Read l2 - →Arrange (#1{replica_id}, #3{bucket_start}) + →Arrange (#1{replica_id}, #6{bucket_start}) →Map/Filter/Project - Project: #0..=#3 + Project: #0..=#6 →Non-monotonic TopK - Group By #3, #1 - Order By #4 desc nulls_first + Group By #6, #1 + Order By #7 desc nulls_first Limit 1 →Fused with Child Map/Filter/Project - Project: #0, #1, #10..=#12 + Project: #0, #1, #10..=#15 Map: coalesce(#2{heap_bytes}, 0) →Read l2 →Arrange (#0{replica_id}) @@ -1003,22 +1003,22 @@ mz_internal.mz_console_cluster_utilization_overview_24h: Key: (#0{replica_id}) →Return →Map/Filter/Project - Project: #1, #0, #2..=#5, #10, #8, #9, #11..=#13, #6, #7, #17, #18, #16, #15, #14 + Project: #1, #0, #2..=#5, #10, #8, #9, #11, #12, #16, #13..=#15, #6, #7, #20, #21, #19, #18, #17 Map: (#1{bucket_start} + 00:05:00) →Union →Map/Filter/Project - Project: #0..=#17 + Project: #0..=#20 Map: null →Consolidating Union →Negate Diffs →Fused with Child Map/Filter/Project - Project: #1, #3, #2, #0, #5, #4, #7, #6, #9..=#11, #8, #13, #12, #14..=#16 + Project: #1, #3, #2, #0, #5, #4, #7, #6, #9..=#11, #8, #13..=#16, #12, #17..=#19 →Read l5 →Fused with Child Map/Filter/Project - Project: #1, #3, #2, #0, #5, #4, #7, #6, #9..=#11, #8, #13, #12, #14..=#16 + Project: #1, #3, #2, #0, #5, #4, #7, #6, #9..=#11, #8, #13..=#16, #12, #17..=#19 →Read l4 →Fused with Child Map/Filter/Project - Project: #1, #3, #2, #0, #5, #4, #7, #6, #9..=#11, #8, #13, #12, #14..=#17 + Project: #1, #3, #2, #0, #5, #4, #7, #6, #9..=#11, #8, #13..=#16, #12, #17..=#20 →Read l5 Used Indexes: @@ -1071,14 +1071,14 @@ mz_internal.mz_console_cluster_utilization_overview_3h: cte l1 = →Differential Join %0[#0..=#7] » %1[#0..=#7] after %1: - Project: #0..=#3, #13..=#17 - Map: uint8_to_double(case when (0 = uint8_to_numeric(#7{processes})) then null else #7{processes} end), ((#9 / uint8_to_double(case when (0 = uint8_to_numeric(#4{cpu_nano_cores})) then null else #4{cpu_nano_cores} end)) / #12), ((#10 / uint8_to_double(case when (0 = uint8_to_numeric(#5{memory_bytes})) then null else #5{memory_bytes} end)) / #12), ((#11 / uint8_to_double(case when (0 = uint8_to_numeric(#6{disk_bytes})) then null else #6{disk_bytes} end)) / #12), coalesce(#8, #14), case when ((#10) IS NULL AND (#11) IS NULL) then null else ((coalesce(#10, 0) + coalesce(#11, 0)) / case when (0 = ((uint8_to_double(#5{memory_bytes}) + uint8_to_double(#6{disk_bytes})) * uint8_to_double(#7{processes}))) then null else ((uint8_to_double(#5{memory_bytes}) + uint8_to_double(#6{disk_bytes})) * uint8_to_double(#7{processes})) end) end + Project: #0..=#3, #16, #18..=#24 + Map: uint8_to_double(case when (0 = uint8_to_numeric(#7{processes})) then null else #7{processes} end), ((#12 / uint8_to_double(case when (0 = uint8_to_numeric(#4{cpu_nano_cores})) then null else #4{cpu_nano_cores} end)) / #15), uint8_to_double(case when (0 = uint8_to_numeric(#5{memory_bytes})) then null else #5{memory_bytes} end), ((#13 / #17) / #15), ((#14 / uint8_to_double(case when (0 = uint8_to_numeric(#6{disk_bytes})) then null else #6{disk_bytes} end)) / #15), coalesce(#8, #18), (#9 / #17), (#10 / #17), (#11 / #17), case when ((#13) IS NULL AND (#14) IS NULL) then null else ((coalesce(#13, 0) + coalesce(#14, 0)) / case when (0 = ((uint8_to_double(#5{memory_bytes}) + uint8_to_double(#6{disk_bytes})) * uint8_to_double(#7{processes}))) then null else ((uint8_to_double(#5{memory_bytes}) + uint8_to_double(#6{disk_bytes})) * uint8_to_double(#7{processes})) end) end →Temporally-Bucketed Bucketed Hierarchical GroupAggregate (buckets: 268435456 16777216 1048576 65536 4096 256 16) - Aggregations: max + Aggregations: max, max, max, max Key: - Project: #7, #0, #2, #1, #4..=#6, #3 + Project: #8, #0, #2, #1, #4..=#6, #3 →Fused with Child Map/Filter/Project - Project: #0..=#6, #10..=#12 + Project: #0..=#6, #8, #10..=#13 →Read l0 →Temporally-Bucketed Accumulable GroupAggregate Simple aggregates: sum(uint8_to_double(#7{cpu_nano_cores})), sum(uint8_to_double(#8{memory_bytes})), sum(uint8_to_double(#9{disk_bytes})) @@ -1124,7 +1124,7 @@ query T multiline EXPLAIN INDEX "mz_internal"."mz_console_cluster_utilization_overview_ind"; ---- mz_internal.mz_console_cluster_utilization_overview_ind: - →Arrange (#17{cluster_id}) + →Arrange (#20{cluster_id}) →Stream mz_internal.mz_console_cluster_utilization_overview mz_internal.mz_console_cluster_utilization_overview: @@ -1156,14 +1156,14 @@ mz_internal.mz_console_cluster_utilization_overview: cte l2 = →Differential Join %0[#0..=#6] » %1[#0..=#6] after %1: - Project: #0, #1, #7, #10, #11, #13..=#15, #17..=#20 - Map: uint8_to_double(case when (0 = uint8_to_numeric(#6{processes})) then null else #6{processes} end), ((#9 / uint8_to_double(case when (0 = uint8_to_numeric(#3{cpu_nano_cores})) then null else #3{cpu_nano_cores} end)) / #12), ((#10 / uint8_to_double(case when (0 = uint8_to_numeric(#4{memory_bytes})) then null else #4{memory_bytes} end)) / #12), ((#11 / uint8_to_double(case when (0 = uint8_to_numeric(#5{disk_bytes})) then null else #5{disk_bytes} end)) / #12), uint8_to_double(#6{processes}), (uint8_to_double(#5{disk_bytes}) * #16), (uint8_to_double(#4{memory_bytes}) * #16), coalesce(#8, #14), timestamptz_bin(01:00:00, #0{occurred_at}, 1970-01-01 00:00:00 UTC) + Project: #0, #1, #7, #13, #14, #16, #18, #19, #21..=#27 + Map: uint8_to_double(case when (0 = uint8_to_numeric(#6{processes})) then null else #6{processes} end), ((#12 / uint8_to_double(case when (0 = uint8_to_numeric(#3{cpu_nano_cores})) then null else #3{cpu_nano_cores} end)) / #15), uint8_to_double(case when (0 = uint8_to_numeric(#4{memory_bytes})) then null else #4{memory_bytes} end), ((#13 / #17) / #15), ((#14 / uint8_to_double(case when (0 = uint8_to_numeric(#5{disk_bytes})) then null else #5{disk_bytes} end)) / #15), uint8_to_double(#6{processes}), (uint8_to_double(#5{disk_bytes}) * #20), (uint8_to_double(#4{memory_bytes}) * #20), coalesce(#8, #18), (#9 / #17), (#10 / #17), (#11 / #17), timestamptz_bin(01:00:00, #0{occurred_at}, 1970-01-01 00:00:00 UTC) →Temporally-Bucketed Bucketed Hierarchical GroupAggregate (buckets: 268435456 16777216 1048576 65536 4096 256 16) - Aggregations: max, max + Aggregations: max, max, max, max, max Key: - Project: #6, #0, #1, #3..=#5, #2 + Project: #7, #0, #1, #3..=#5, #2 →Fused with Child Map/Filter/Project - Project: #0..=#5, #9..=#11 + Project: #0..=#5, #7, #9..=#12 →Read l1 →Temporally-Bucketed Accumulable GroupAggregate Simple aggregates: sum(uint8_to_double(#6{cpu_nano_cores})), sum(uint8_to_double(#7{memory_bytes})), sum(uint8_to_double(#8{disk_bytes})) @@ -1173,7 +1173,7 @@ mz_internal.mz_console_cluster_utilization_overview: Project: #0..=#9 →Read l1 cte l3 = - →Delta Join [%0[#1{replica_id}] » %1[#1{replica_id}, #3{bucket_start}] » %2[#1{replica_id}, #3{bucket_start}] » %3[#1{replica_id}, #4{bucket_start}] » %4[#1{replica_id}, #3{bucket_start}] » %5:l0[#0{replica_id}]] [%1[#1{replica_id}, #3{bucket_start}] » %0[#1{replica_id}, #3{bucket_start}] » %2[#1{replica_id}, #3{bucket_start}] » %3[#1{replica_id}, #4{bucket_start}] » %4[#1{replica_id}, #3{bucket_start}] » %5:l0[#0{replica_id}]] [%2[#1{replica_id}, #3{bucket_start}] » %0[#1{replica_id}, #3{bucket_start}] » %1[#1{replica_id}, #3{bucket_start}] » %3[#1{replica_id}, #4{bucket_start}] » %4[#1{replica_id}, #3{bucket_start}] » %5:l0[#0{replica_id}]] [%3[#1{replica_id}, #4{bucket_start}] » %0[#1{replica_id}, #3{bucket_start}] » %1[#1{replica_id}, #3{bucket_start}] » %2[#1{replica_id}, #3{bucket_start}] » %4[#1{replica_id}, #3{bucket_start}] » %5:l0[#0{replica_id}]] [%4[#1{replica_id}, #3{bucket_start}] » %0[#1{replica_id}, #3{bucket_start}] » %1[#1{replica_id}, #3{bucket_start}] » %2[#1{replica_id}, #3{bucket_start}] » %3[#1{replica_id}, #4{bucket_start}] » %5:l0[#0{replica_id}]] [%5:l0[#0{replica_id}] » %0[#1{replica_id}] » %1[#1{replica_id}, #3{bucket_start}] » %2[#1{replica_id}, #3{bucket_start}] » %3[#1{replica_id}, #4{bucket_start}] » %4[#1{replica_id}, #3{bucket_start}]] + →Delta Join [%0[#1{replica_id}] » %1[#1{replica_id}, #3{bucket_start}] » %2[#1{replica_id}, #3{bucket_start}] » %3[#1{replica_id}, #4{bucket_start}] » %4[#1{replica_id}, #6{bucket_start}] » %5:l0[#0{replica_id}]] [%1[#1{replica_id}, #3{bucket_start}] » %0[#1{replica_id}, #3{bucket_start}] » %2[#1{replica_id}, #3{bucket_start}] » %3[#1{replica_id}, #4{bucket_start}] » %4[#1{replica_id}, #6{bucket_start}] » %5:l0[#0{replica_id}]] [%2[#1{replica_id}, #3{bucket_start}] » %0[#1{replica_id}, #3{bucket_start}] » %1[#1{replica_id}, #3{bucket_start}] » %3[#1{replica_id}, #4{bucket_start}] » %4[#1{replica_id}, #6{bucket_start}] » %5:l0[#0{replica_id}]] [%3[#1{replica_id}, #4{bucket_start}] » %0[#1{replica_id}, #3{bucket_start}] » %1[#1{replica_id}, #3{bucket_start}] » %2[#1{replica_id}, #3{bucket_start}] » %4[#1{replica_id}, #6{bucket_start}] » %5:l0[#0{replica_id}]] [%4[#1{replica_id}, #6{bucket_start}] » %0[#1{replica_id}, #3{bucket_start}] » %1[#1{replica_id}, #3{bucket_start}] » %2[#1{replica_id}, #3{bucket_start}] » %3[#1{replica_id}, #4{bucket_start}] » %5:l0[#0{replica_id}]] [%5:l0[#0{replica_id}] » %0[#1{replica_id}] » %1[#1{replica_id}, #3{bucket_start}] » %2[#1{replica_id}, #3{bucket_start}] » %3[#1{replica_id}, #4{bucket_start}] » %4[#1{replica_id}, #6{bucket_start}]] →Arrange (#1{replica_id}) (#1{replica_id}, #3{bucket_start}) →Map/Filter/Project Project: #0..=#3 @@ -1182,7 +1182,7 @@ mz_internal.mz_console_cluster_utilization_overview: Order By #4 desc nulls_first Limit 1 →Fused with Child Map/Filter/Project - Project: #0, #1, #6, #11, #12 + Project: #0, #1, #6, #14, #15 Map: coalesce(#3{memory_bytes}, 0) →Read l2 →Arrange (#1{replica_id}, #3{bucket_start}) @@ -1193,7 +1193,7 @@ mz_internal.mz_console_cluster_utilization_overview: Order By #4 desc nulls_first Limit 1 →Fused with Child Map/Filter/Project - Project: #0, #1, #7, #11, #12 + Project: #0, #1, #7, #14, #15 Map: coalesce(#4{disk_bytes}, 0) →Read l2 →Arrange (#1{replica_id}, #3{bucket_start}) @@ -1204,7 +1204,7 @@ mz_internal.mz_console_cluster_utilization_overview: Order By #4 desc nulls_first Limit 1 →Fused with Child Map/Filter/Project - Project: #0, #1, #5, #11, #12 + Project: #0, #1, #5, #14, #15 Map: coalesce(#5{cpu_percent}, 0) →Read l2 →Arrange (#1{replica_id}, #4{bucket_start}) @@ -1215,18 +1215,18 @@ mz_internal.mz_console_cluster_utilization_overview: Order By #6 desc nulls_first Limit 1 →Fused with Child Map/Filter/Project - Project: #0, #1, #6, #7, #11..=#13 - Map: case when ((#3{memory_bytes}) IS NULL AND (#4{disk_bytes}) IS NULL) then null else ((coalesce(#3{memory_bytes}, 0) + coalesce(#4{disk_bytes}, 0)) / case when (0 = (#9{total_memory_bytes} + #8{total_disk_bytes})) then null else (#9{total_memory_bytes} + #8{total_disk_bytes}) end) end, coalesce(#12{memory_and_disk_percent}, 0) + Project: #0, #1, #6, #7, #14..=#16 + Map: case when ((#3{memory_bytes}) IS NULL AND (#4{disk_bytes}) IS NULL) then null else ((coalesce(#3{memory_bytes}, 0) + coalesce(#4{disk_bytes}, 0)) / case when (0 = (#9{total_memory_bytes} + #8{total_disk_bytes})) then null else (#9{total_memory_bytes} + #8{total_disk_bytes}) end) end, coalesce(#15{memory_and_disk_percent}, 0) →Read l2 - →Arrange (#1{replica_id}, #3{bucket_start}) + →Arrange (#1{replica_id}, #6{bucket_start}) →Map/Filter/Project - Project: #0..=#3 + Project: #0..=#6 →Non-monotonic TopK - Group By #3, #1 - Order By #4 desc nulls_first + Group By #6, #1 + Order By #7 desc nulls_first Limit 1 →Fused with Child Map/Filter/Project - Project: #0, #1, #10..=#12 + Project: #0, #1, #10..=#15 Map: coalesce(#2{heap_bytes}, 0) →Read l2 →Arrange (#0{replica_id}) @@ -1270,22 +1270,22 @@ mz_internal.mz_console_cluster_utilization_overview: Key: (#0{replica_id}) →Return →Map/Filter/Project - Project: #1, #0, #2..=#5, #10, #8, #9, #11..=#13, #6, #7, #17, #18, #16, #15, #14 + Project: #1, #0, #2..=#5, #10, #8, #9, #11, #12, #16, #13..=#15, #6, #7, #20, #21, #19, #18, #17 Map: (#1{bucket_start} + 01:00:00) →Union →Map/Filter/Project - Project: #0..=#17 + Project: #0..=#20 Map: null →Consolidating Union →Negate Diffs →Fused with Child Map/Filter/Project - Project: #1, #3, #2, #0, #5, #4, #7, #6, #9..=#11, #8, #13, #12, #14..=#16 + Project: #1, #3, #2, #0, #5, #4, #7, #6, #9..=#11, #8, #13..=#16, #12, #17..=#19 →Read l5 →Fused with Child Map/Filter/Project - Project: #1, #3, #2, #0, #5, #4, #7, #6, #9..=#11, #8, #13, #12, #14..=#16 + Project: #1, #3, #2, #0, #5, #4, #7, #6, #9..=#11, #8, #13..=#16, #12, #17..=#19 →Read l4 →Fused with Child Map/Filter/Project - Project: #1, #3, #2, #0, #5, #4, #7, #6, #9..=#11, #8, #13, #12, #14..=#17 + Project: #1, #3, #2, #0, #5, #4, #7, #6, #9..=#11, #8, #13..=#16, #12, #17..=#20 →Read l5 Used Indexes: @@ -8570,7 +8570,7 @@ EXPLAIN SELECT * FROM "mz_internal"."mz_console_cluster_utilization_overview"; ---- Explained Query (fast path): →Map/Filter/Project - Project: #1..=#17, #0, #18 + Project: #1..=#20, #0, #21 →Indexed mz_internal.mz_console_cluster_utilization_overview (using mz_internal.mz_console_cluster_utilization_overview_ind) Used Indexes: @@ -8585,7 +8585,7 @@ EXPLAIN SELECT * FROM "mz_internal"."mz_console_cluster_utilization_overview_24h ---- Explained Query (fast path): →Map/Filter/Project - Project: #1..=#17, #0, #18 + Project: #1..=#20, #0, #21 →Indexed mz_internal.mz_console_cluster_utilization_overview_24h (using mz_internal.mz_console_cluster_utilization_overview_24h_ind) Used Indexes: @@ -8600,7 +8600,7 @@ EXPLAIN SELECT * FROM "mz_internal"."mz_console_cluster_utilization_overview_3h" ---- Explained Query (fast path): →Map/Filter/Project - Project: #1, #0, #2..=#9 + Project: #1, #0, #2..=#12 →Indexed mz_internal.mz_console_cluster_utilization_overview_3h (using mz_internal.mz_console_cluster_utilization_overview_3h_ind) Used Indexes: diff --git a/test/sqllogictest/cockroach/srfs.slt b/test/sqllogictest/cockroach/srfs.slt index e8e5aa17e4007..05bffe2393638 100644 --- a/test/sqllogictest/cockroach/srfs.slt +++ b/test/sqllogictest/cockroach/srfs.slt @@ -1331,8 +1331,8 @@ mz_compute_operator_hydration_statuses_per_worker 3 mz_compute_operator_hydration_statuses_per_worker 3 mz_compute_operator_hydration_statuses_per_worker 3 mz_connections 3 -mz_console_cluster_utilization_overview 18 -mz_console_cluster_utilization_overview_24h 18 +mz_console_cluster_utilization_overview 21 +mz_console_cluster_utilization_overview_24h 21 mz_console_cluster_utilization_overview_3h 2 mz_databases 3 mz_dataflow_addresses_per_worker 1 diff --git a/test/sqllogictest/console_cluster_utilization_shares.slt b/test/sqllogictest/console_cluster_utilization_shares.slt new file mode 100644 index 0000000000000..d13d3aeaa90b0 --- /dev/null +++ b/test/sqllogictest/console_cluster_utilization_shares.slt @@ -0,0 +1,101 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +mode cockroach + +reset-server + +# Pins the arithmetic of ram_percent / swap_of_ram_percent / heap_limit_percent +# in the mz_console_cluster_utilization_overview* views. Those views read +# mz_cluster_replica_metrics_history, which cannot be populated deterministically, +# so the expressions are exercised here over tables of the same shape. Keep in +# sync with CONSOLE_MEMORY_SHARES_SQL in src/catalog/src/builtin/mz_internal.rs. + +statement ok +CREATE TABLE replica_sizes (size text, memory_bytes uint8) + +statement ok +CREATE TABLE replica_metrics ( + replica_id text, + size text, + process_id uint8, + memory_bytes uint8, + swap_bytes uint8, + heap_limit uint8 +) + +statement ok +INSERT INTO replica_sizes VALUES ('normal', 100), ('zero-mem', 0) + +# r1 single process. r2 skewed, RAM and swap peaking on the same process. +# r3 skewed in opposite directions, so the independent maxima describe different +# processes. r4 has no swap reported, r5 no heap limit. r6 sits on a size whose +# memory_bytes is 0, exercising the NULLIF guard. +statement ok +INSERT INTO replica_metrics VALUES + ('r1', 'normal', 0, 40, 0, 125), + ('r2', 'normal', 0, 90, 20, 125), + ('r2', 'normal', 1, 30, 5, 125), + ('r3', 'normal', 0, 90, 2, 125), + ('r3', 'normal', 1, 30, 60, 125), + ('r4', 'normal', 0, 50, NULL, 125), + ('r5', 'normal', 0, 50, 10, NULL), + ('r6', 'zero-mem', 0, 50, 10, 125) + +# MAX across processes, each divided by a *single* process's RAM allocation. +# A SUM would report r2 at 1.2 rather than 0.9; multiplying the denominator by +# the process count would report 0.45. +query TRRR +SELECT + m.replica_id, + MAX(m.memory_bytes::float8) / NULLIF(s.memory_bytes, 0) AS ram_percent, + MAX(m.swap_bytes::float8) / NULLIF(s.memory_bytes, 0) AS swap_of_ram_percent, + MAX(m.heap_limit::float8) / NULLIF(s.memory_bytes, 0) AS heap_limit_percent +FROM replica_metrics AS m +JOIN replica_sizes AS s ON s.size = m.size +GROUP BY m.replica_id, s.memory_bytes +ORDER BY m.replica_id +---- +r1 0.4 0 1.25 +r2 0.9 0.2 1.25 +r3 0.9 0.6 1.25 +r4 0.5 NULL 1.25 +r5 0.5 0.1 NULL +r6 NULL NULL NULL + +# r3's stack exceeds its heap ceiling because peak RAM and peak swap come from +# different processes. Documented upper-bound behaviour, not a bug. +query TB +SELECT + m.replica_id, + (MAX(m.memory_bytes::float8) + MAX(m.swap_bytes::float8)) / NULLIF(s.memory_bytes, 0) + > MAX(m.heap_limit::float8) / NULLIF(s.memory_bytes, 0) AS stack_exceeds_ceiling +FROM replica_metrics AS m +JOIN replica_sizes AS s ON s.size = m.size +WHERE m.replica_id IN ('r2', 'r3') +GROUP BY m.replica_id, s.memory_bytes +ORDER BY m.replica_id +---- +r2 false +r3 true + +# A size that cannot swap reports null, never 0: "cannot swap" must not read as +# "swapped nothing". Same for an unreported heap limit. +query TBB +SELECT + m.replica_id, + MAX(m.swap_bytes::float8) IS NULL AS swap_is_null, + MAX(m.heap_limit::float8) IS NULL AS heap_limit_is_null +FROM replica_metrics AS m +WHERE m.replica_id IN ('r4', 'r5') +GROUP BY m.replica_id +ORDER BY m.replica_id +---- +r4 true false +r5 false true diff --git a/test/sqllogictest/mz_catalog_server_index_accounting.slt b/test/sqllogictest/mz_catalog_server_index_accounting.slt index 5e47356389bf6..df6e5e4f66f08 100644 --- a/test/sqllogictest/mz_catalog_server_index_accounting.slt +++ b/test/sqllogictest/mz_catalog_server_index_accounting.slt @@ -401,6 +401,7 @@ mz_console_cluster_utilization_overview bucket_end mz_console_cluster_utilization_overview bucket_start mz_console_cluster_utilization_overview cluster_id mz_console_cluster_utilization_overview disk_percent +mz_console_cluster_utilization_overview heap_limit_percent mz_console_cluster_utilization_overview heap_percent mz_console_cluster_utilization_overview max_cpu_at mz_console_cluster_utilization_overview max_cpu_percent @@ -414,12 +415,15 @@ mz_console_cluster_utilization_overview memory_and_disk_percent mz_console_cluster_utilization_overview memory_percent mz_console_cluster_utilization_overview name mz_console_cluster_utilization_overview offline_events +mz_console_cluster_utilization_overview ram_percent mz_console_cluster_utilization_overview replica_id mz_console_cluster_utilization_overview size +mz_console_cluster_utilization_overview swap_of_ram_percent mz_console_cluster_utilization_overview_24h bucket_end mz_console_cluster_utilization_overview_24h bucket_start mz_console_cluster_utilization_overview_24h cluster_id mz_console_cluster_utilization_overview_24h disk_percent +mz_console_cluster_utilization_overview_24h heap_limit_percent mz_console_cluster_utilization_overview_24h heap_percent mz_console_cluster_utilization_overview_24h max_cpu_at mz_console_cluster_utilization_overview_24h max_cpu_percent @@ -433,18 +437,23 @@ mz_console_cluster_utilization_overview_24h memory_and_disk_percent mz_console_cluster_utilization_overview_24h memory_percent mz_console_cluster_utilization_overview_24h name mz_console_cluster_utilization_overview_24h offline_events +mz_console_cluster_utilization_overview_24h ram_percent mz_console_cluster_utilization_overview_24h replica_id mz_console_cluster_utilization_overview_24h size +mz_console_cluster_utilization_overview_24h swap_of_ram_percent mz_console_cluster_utilization_overview_3h cluster_id mz_console_cluster_utilization_overview_3h cpu_percent mz_console_cluster_utilization_overview_3h disk_percent +mz_console_cluster_utilization_overview_3h heap_limit_percent mz_console_cluster_utilization_overview_3h heap_percent mz_console_cluster_utilization_overview_3h memory_and_disk_percent mz_console_cluster_utilization_overview_3h memory_percent mz_console_cluster_utilization_overview_3h name mz_console_cluster_utilization_overview_3h occurred_at +mz_console_cluster_utilization_overview_3h ram_percent mz_console_cluster_utilization_overview_3h replica_id mz_console_cluster_utilization_overview_3h size +mz_console_cluster_utilization_overview_3h swap_of_ram_percent mz_databases id mz_databases name mz_databases oid diff --git a/test/testdrive/explain-analyze.td b/test/testdrive/explain-analyze.td index c7b719bd4c2d4..ab7225d8c85d7 100644 --- a/test/testdrive/explain-analyze.td +++ b/test/testdrive/explain-analyze.td @@ -68,7 +68,7 @@ materialize.blue.t_idx > EXPLAIN ANALYZE CPU FOR INDEX mz_internal.mz_console_cluster_utilization_overview_ind; operator total_elapsed ----------------------------------------- -"Arrange (#17{cluster_id})" +"Arrange (#20{cluster_id})" " Stream " "Returning Map/Filter/Project" " Union" @@ -98,7 +98,7 @@ operator total_elapsed "l3 = Delta Join [%0 » %1 » %2 » %3 » %4 » %5] [%1 » %0 » %2 » %3 » %4 » %5] [%2 » %0 » %1 » %3 » %4 » %5] [%3 » %0 » %1 » %2 » %4 » %5] [%4 » %0 » %1 » %2 » %3 » %5] [%5 » %0 » %1 » %2 » %3 » %4]" " Arrange (#0{replica_id})" " Arranged l0" -" Arrange (#1{replica_id}, #3{bucket_start})" +" Arrange (#1{replica_id}, #6{bucket_start})" " Map/Filter/Project" " Non-monotonic TopK" " Read l2" From 3c5d05b3ec96716b81a42845ca966bdc36f52a78 Mon Sep 17 00:00:00 2001 From: Michael Greenberg Date: Wed, 23 Sep 2026 12:13:35 -0400 Subject: [PATCH 6/6] sql: clarify LIR version policy (#39065) ### Motivation https://materializeinc.slack.com/archives/C08AENAP691/p1790010546649459 ### Description Clarifies the LIR versioning policy, in comments and messages. This will have to change when LIR pinning is deployed. ### Verification N/A --- src/compute-types/src/plan.rs | 30 ++++++++++++++++++++---- src/compute-types/tests/func_registry.rs | 16 ++++++------- src/compute-types/tests/lir_schema.rs | 24 +++++++++---------- 3 files changed, 43 insertions(+), 27 deletions(-) diff --git a/src/compute-types/src/plan.rs b/src/compute-types/src/plan.rs index 7e1e8336df13a..7c822cfec3497 100644 --- a/src/compute-types/src/plan.rs +++ b/src/compute-types/src/plan.rs @@ -209,15 +209,35 @@ impl std::fmt::Display for LirId { /// Version of the stable LIR serialization format. /// -/// Bump this when the serialized representation of [`LirRelationExpr`] or -/// anything it transitively contains changes, or when a scalar function's -/// declared properties change. Two snapshot tests key off it: the schema -/// snapshot in `tests/lir_schema.rs` against +/// Once a version has shipped, meaning a released Materialize durably stores +/// pinned plans in it, bump this when the serialized representation of +/// [`LirRelationExpr`] or anything it transitively contains changes, or when +/// a scalar function's declared properties change. Whether the current +/// version has shipped is recorded in [`LIR_VERSION_POLICY`]. Two snapshot +/// tests key off it: the schema snapshot in `tests/lir_schema.rs` against /// `tests/snapshots/lir_v{LIR_VERSION}.json`, and the function property /// registry in `tests/func_registry.rs` against -/// `tests/snapshots/func_registry_v{LIR_VERSION}.json`. +/// `tests/snapshots/func_registry.json` with a per-version digest in +/// `tests/snapshots/func_registry_digests.json`. pub const LIR_VERSION: u64 = 1; +/// What a change to the stable LIR format must do at the current +/// [`LIR_VERSION`]. The snapshot tests print it when a snapshot goes stale. +/// +/// This is the one place that records whether the current version has +/// shipped. Once a released Materialize durably stores pinned plans, replace +/// the text with the shipped policy: bump `LIR_VERSION` rather than rewriting +/// the shipped version's snapshots, so the old schema stays available to +/// migration tooling. +pub const LIR_VERSION_POLICY: &str = "\ +The current LIR version has NOT shipped: no released Materialize stores pinned +LIR plans yet, so nothing depends on the checked-in snapshots. Regenerate them +in place and include the diff in your PR. + +Once pinned plans are durably stored, a change like this must instead bump +LIR_VERSION in src/compute-types/src/plan.rs so it lands as a new version. +See doc/developer/design/20260826_pinned_lir.md."; + pub use constant_rows_serde::ConstantRows; /// Serializes `LirRelationNode::Constant`'s rows through the named diff --git a/src/compute-types/tests/func_registry.rs b/src/compute-types/tests/func_registry.rs index 4c2f178db9c00..10095b4b4eea1 100644 --- a/src/compute-types/tests/func_registry.rs +++ b/src/compute-types/tests/func_registry.rs @@ -24,6 +24,8 @@ //! when producing it, and must bump `LIR_VERSION` so pinned plans of the //! old version are replanned. Additions are the exception: a stored plan //! cannot reference a variant that did not exist when it was written. +//! Whether the current version has shipped is recorded in +//! [`LIR_VERSION_POLICY`]. //! * `func_registry_source.json`, the sources. Changes here need judgment, a //! body may or may not have changed behavior, so they are reported without //! demanding a bump. @@ -36,7 +38,7 @@ use std::collections::{BTreeMap, BTreeSet}; -use mz_compute_types::plan::LIR_VERSION; +use mz_compute_types::plan::{LIR_VERSION, LIR_VERSION_POLICY}; use mz_expr::func::registry::{FuncRegistry, Record}; use serde::Serialize; use sha2::{Digest, Sha256}; @@ -257,10 +259,7 @@ fn func_registry_snapshot() { this is not a function change. The input_types and output_type fields\n\ record what the sample probed, and a removed `name[label]` record is a\n\ deleted labeled sample. In that case regenerate without bumping.\n\n\ - If LIR version {LIR_VERSION} has already shipped, bump LIR_VERSION in\n\ - src/compute-types/src/plan.rs so the change lands as a new version and\n\ - version {LIR_VERSION}'s digest in '{DIGESTS_PATH}' stays as it is.\n\ - If version {LIR_VERSION} is unshipped, regenerating in place is fine.\n\n\ + {LIR_VERSION_POLICY}\n\n\ {regenerate}", describe(&properties_diff), ); @@ -282,10 +281,9 @@ fn func_registry_snapshot() { "Scalar function registry changed without affecting stored plans!\n\n\ {}\n\n\ For a body_fingerprint change, review whether the new body alters the\n\ - result for any input. If it does and LIR version {LIR_VERSION} has\n\ - already shipped, bump LIR_VERSION in src/compute-types/src/plan.rs so\n\ - stored plans are replanned. Otherwise, or if version {LIR_VERSION} is\n\ - unshipped, regenerating in place is fine.\n\n\ + result for any input. If it does, treat it as a change to the stable\n\ + format.\n\n\ + {LIR_VERSION_POLICY}\n\n\ {regenerate}", informational.join("\n\n"), ); diff --git a/src/compute-types/tests/lir_schema.rs b/src/compute-types/tests/lir_schema.rs index 80f486c772c44..94a89d5b3476f 100644 --- a/src/compute-types/tests/lir_schema.rs +++ b/src/compute-types/tests/lir_schema.rs @@ -30,10 +30,9 @@ //! because they form the storage codec for `SourceData` (`ProtoRow` on the //! Ok side, `ProtoEvalError` inside `ProtoDataflowError` on the Err side). //! -//! While a given LIR version is unshipped, regenerating its snapshot in place -//! is fine. Once pinned plans are durably stored, a schema change must instead -//! bump [`LIR_VERSION`], which targets a fresh snapshot file and leaves the -//! old schema in place for migration tooling. +//! Whether the current version has shipped, and so whether a schema change +//! regenerates the snapshot in place or bumps [`LIR_VERSION`] to target a +//! fresh snapshot file, is recorded once in [`LIR_VERSION_POLICY`]. //! //! There are some subtleties to tracing. [`serde_reflection::Registry`] reuses //! existing serde machinery, which means that we have to work carefully to ensure @@ -50,7 +49,8 @@ use mz_compute_types::plan::scalar::{LirScalarExpr, LiteralValue}; use mz_compute_types::plan::threshold::ThresholdPlan; use mz_compute_types::plan::top_k::TopKPlan; use mz_compute_types::plan::{ - ArrangementStrategy, ConstantRows, GetPlan, LIR_VERSION, LirRelationExpr, LirRelationNode, + ArrangementStrategy, ConstantRows, GetPlan, LIR_VERSION, LIR_VERSION_POLICY, LirRelationExpr, + LirRelationNode, }; use mz_expr::func::{TimezoneTime, ToCharTimestamp}; use mz_expr::like_pattern::Matcher; @@ -509,18 +509,16 @@ fn lir_schema_snapshot() { }); if expected != actual { panic!( - "The serialized stable LIR schema changed!\n\n\ + "The serialized stable LIR schema changed.\n\n\ The serde type graph reachable from LirRelationExpr no longer matches\n\ - '{path}'. This affects any durably stored LIR plan.\n\n\ + '{path}'. This is expected whenever a type in that graph changes, and\n\ + the snapshot exists so the change shows up as a reviewable diff.\n\n\ What changed:\n{diff}\n\n\ Full detail: diff '{path}' against the freshly traced schema at\n\ '{CURRENT_PATH}'.\n\n\ - If LIR version {LIR_VERSION} has already shipped, bump LIR_VERSION in\n\ - src/compute-types/src/plan.rs so the change lands as a new version.\n\ - If version {LIR_VERSION} is unshipped, regenerating in place is fine.\n\n\ - Then regenerate the snapshot and review the diff:\n\n \ - REWRITE=1 cargo test -p mz-compute-types --test lir_schema\n\n\ - See doc/developer/design/20260311_optimizer_customer_tradeoff.md.\n", + {LIR_VERSION_POLICY}\n\n\ + To regenerate:\n\n \ + REWRITE=1 cargo test -p mz-compute-types --test lir_schema\n", diff = schema_diff(&expected, &actual), ); }