From 8103f7787b60a9e404dcc92a0364da2bd3f66d8a Mon Sep 17 00:00:00 2001 From: Qindeel Ishtiaq Date: Fri, 11 Sep 2026 18:54:03 -0400 Subject: [PATCH 1/6] envd: collect and record replica swap metrics (#38788) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### Motivation On swap-based replica sizes, the catalog only exposes the combined `heap_bytes` / `heap_limit`. clusterd already measures swap separately (`VmSwap` from `/proc/self/status`) and reports it as `swap_bytes` on `/api/usage-metrics`, but the Kubernetes orchestrator folds it into `heap_bytes` (and, for compatibility, into `disk_bytes`) and drops the standalone value, so it never reaches SQL. Deriving it downstream is unreliable: `heap_bytes - memory_bytes` mixes clusterd's VmRSS with the metrics-server working set (different instruments, different sampling instants), and `disk_bytes` only equals swap on swap-enabled sizes, which are not distinguishable from public catalog relations. Recording swap first-class lets the Console show how much of a replica's memory is spilled to disk. Fixes [CNS-154](https://linear.app/materializeinc/issue/CNS-154/add-swap-metrics-to-database-catalog) ### Description * Add `swap_bytes` to `ServiceProcessMetrics`. The Kubernetes orchestrator keeps the clusterd-reported value (the existing swap-into-`disk_bytes` shim is unchanged); the process orchestrator reports `None`. * Append `swap_bytes` as a nullable `uint8` column at the end of `REPLICA_METRICS_HISTORY_DESC`. This is a backward-compatible persist schema change, so existing history is preserved and older rows read `NULL` (same pattern as `heap_bytes` / `heap_limit` in 96fa447160). * Record it in `mz_internal.mz_cluster_replica_metrics_history` and expose it on `mz_internal.mz_cluster_replica_metrics`. * Add `swap_percent` to `mz_internal.mz_cluster_replica_utilization` and `mz_cluster_replica_utilization_history`, computed as `swap_bytes / heap_limit * 100`. No clusterd changes are needed: every running clusterd already reports `swap_bytes`, so the metric lights up on existing replicas as soon as environmentd upgrades. Two decisions worth reviewer attention: * **`swap_percent` denominator**: `heap_limit`, matching the sibling columns' "percent of the total allocation" contract, so `heap_percent - swap_percent` reads as the RAM share and the two stack in a gauge. The spill fraction (swap as a share of `heap_bytes`) is derivable as `swap_percent / heap_percent` or from the raw columns. * **Follow-up**: once the Console reads `swap_bytes` directly, the orchestrator shim that folds swap into `disk_bytes` can be retired (see the comment in `get_metrics`). ### Verification Regenerated the explain goldens (`system-cluster.slt`, `catalog_server_explain.slt`) and `autogenerated/mz_internal.slt`, and added the new columns to `mz_catalog_server_index_accounting.slt`; all four files pass against a fresh catalog. The system-catalog docs page is updated to match the column comments. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --- .../content/sql/system-catalog/mz_internal.md | 4 ++++ .../catalog/open/builtin_schema_migration.rs | 10 +++++++++ src/catalog/src/builtin/mz_internal.rs | 21 +++++++++++++++++-- src/controller/src/lib.rs | 1 + src/orchestrator-kubernetes/src/lib.rs | 1 + src/orchestrator-process/src/lib.rs | 1 + src/orchestrator/src/lib.rs | 1 + src/storage-client/src/healthcheck.rs | 1 + .../autogenerated/mz_internal.slt | 4 ++++ test/sqllogictest/catalog_server_explain.slt | 12 +++++------ .../mz_catalog_server_index_accounting.slt | 2 ++ test/sqllogictest/system-cluster.slt | 2 +- 12 files changed, 51 insertions(+), 9 deletions(-) diff --git a/doc/user/content/sql/system-catalog/mz_internal.md b/doc/user/content/sql/system-catalog/mz_internal.md index d36965fa807ed..0778a125b2121 100644 --- a/doc/user/content/sql/system-catalog/mz_internal.md +++ b/doc/user/content/sql/system-catalog/mz_internal.md @@ -227,6 +227,7 @@ usage. | `disk_bytes` | [`uint8`] | Approximate disk usage, in bytes. | `heap_bytes` | [`uint8`] | Approximate heap (RAM + swap) usage, in bytes. | `heap_limit` | [`uint8`] | Available heap (RAM + swap) space, in bytes. +| `swap_bytes` | [`uint8`] | Approximate swap usage, in bytes. ## `mz_cluster_replica_metrics_history` @@ -252,6 +253,7 @@ history is retained across restarts. | `occurred_at` | [`timestamp with time zone`] | Wall-clock timestamp at which the event occurred. | `heap_bytes` | [`uint8`] | Approximate heap (RAM + swap) usage, in bytes. | `heap_limit` | [`uint8`] | Available heap (RAM + swap) space, in bytes. +| `swap_bytes` | [`uint8`] | Approximate swap usage, in bytes. ## `mz_cluster_replica_statuses` @@ -298,6 +300,7 @@ At this time, we do not make any guarantees about the exactness or freshness of | `memory_percent` | [`double precision`] | Approximate RAM usage, in percent of the total allocation. | `disk_percent` | [`double precision`] | Approximate disk usage, in percent of the total allocation. | `heap_percent` | [`double precision`] | Approximate heap (RAM + swap) usage, in percent of the total allocation. +| `swap_percent` | [`double precision`] | Approximate swap usage, in percent of the total heap allocation. ## `mz_cluster_replica_utilization_history` @@ -316,6 +319,7 @@ At this time, we do not make any guarantees about the exactness or freshness of | `memory_percent` | [`double precision`] | Approximate RAM usage, in percent of the total allocation. | `disk_percent` | [`double precision`] | Approximate disk usage, in percent of the total allocation. | `heap_percent` | [`double precision`] | Approximate heap (RAM + swap) usage, in percent of the total allocation. +| `swap_percent` | [`double precision`] | Approximate swap usage, in percent of the total heap allocation. | `occurred_at` | [`timestamp with time zone`] | Wall-clock timestamp at which the event occurred. ## `mz_cluster_replica_history` diff --git a/src/adapter/src/catalog/open/builtin_schema_migration.rs b/src/adapter/src/catalog/open/builtin_schema_migration.rs index 34653cd13aad0..e21d3da4179bb 100644 --- a/src/adapter/src/catalog/open/builtin_schema_migration.rs +++ b/src/adapter/src/catalog/open/builtin_schema_migration.rs @@ -462,6 +462,16 @@ static MIGRATIONS: LazyLock> = LazyLock::new(|| { MZ_INTERNAL_SCHEMA, "mz_object_dependencies", ), + // Adding the `swap_bytes` column to `mz_cluster_replica_metrics_history` + // is a backward-compatible append, so the shard schema can evolve in + // place. See the NOTE above: this version must stay at the workspace's + // current dev version until the change ships. + MigrationStep::evolution( + "26.43.0-dev.0", + CatalogItemType::Source, + MZ_INTERNAL_SCHEMA, + "mz_cluster_replica_metrics_history", + ), ] }); diff --git a/src/catalog/src/builtin/mz_internal.rs b/src/catalog/src/builtin/mz_internal.rs index 006492fbdf5d2..ccde5b89bd957 100644 --- a/src/catalog/src/builtin/mz_internal.rs +++ b/src/catalog/src/builtin/mz_internal.rs @@ -2734,6 +2734,7 @@ pub static MZ_CLUSTER_REPLICA_METRICS_HISTORY: LazyLock = "Approximate heap (RAM + swap) usage, in bytes.", ), ("heap_limit", "Available heap (RAM + swap) space, in bytes."), + ("swap_bytes", "Approximate swap usage, in bytes."), ]), is_retained_metrics_object: false, access: vec![PUBLIC_SELECT], @@ -2752,6 +2753,7 @@ pub static MZ_CLUSTER_REPLICA_METRICS: LazyLock = LazyLock::new(|| .with_column("disk_bytes", SqlScalarType::UInt64.nullable(true)) .with_column("heap_bytes", SqlScalarType::UInt64.nullable(true)) .with_column("heap_limit", SqlScalarType::UInt64.nullable(true)) + .with_column("swap_bytes", SqlScalarType::UInt64.nullable(true)) .with_key(vec![0, 1]) .finish(), column_comments: BTreeMap::from_iter([ @@ -2768,6 +2770,7 @@ pub static MZ_CLUSTER_REPLICA_METRICS: LazyLock = LazyLock::new(|| "Approximate heap (RAM + swap) usage, in bytes.", ), ("heap_limit", "Available heap (RAM + swap) space, in bytes."), + ("swap_bytes", "Approximate swap usage, in bytes."), ]), sql: " SELECT @@ -2778,7 +2781,8 @@ SELECT memory_bytes, disk_bytes, heap_bytes, - heap_limit + heap_limit, + swap_bytes FROM mz_internal.mz_cluster_replica_metrics_history JOIN mz_cluster_replicas r ON r.id = replica_id ORDER BY replica_id, process_id, occurred_at DESC", @@ -2805,6 +2809,7 @@ ORDER BY replica_id, process_id, occurred_at DESC", ("disk_bytes", SemanticType::ByteCount), ("heap_bytes", SemanticType::ByteCount), ("heap_limit", SemanticType::ByteCount), + ("swap_bytes", SemanticType::ByteCount), ] }, }), @@ -5421,6 +5426,7 @@ pub static MZ_CLUSTER_REPLICA_UTILIZATION: LazyLock = LazyLock::new .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("swap_percent", SqlScalarType::Float64.nullable(true)) .finish(), column_comments: BTreeMap::from_iter([ ("replica_id", "The ID of a cluster replica."), @@ -5441,6 +5447,10 @@ pub static MZ_CLUSTER_REPLICA_UTILIZATION: LazyLock = LazyLock::new "heap_percent", "Approximate heap (RAM + swap) usage, in percent of the total allocation.", ), + ( + "swap_percent", + "Approximate swap usage, in percent of the total heap allocation.", + ), ]), sql: " SELECT @@ -5449,7 +5459,8 @@ SELECT m.cpu_nano_cores::float8 / NULLIF(s.cpu_nano_cores, 0) * 100 AS cpu_percent, m.memory_bytes::float8 / NULLIF(s.memory_bytes, 0) * 100 AS memory_percent, m.disk_bytes::float8 / NULLIF(s.disk_bytes, 0) * 100 AS disk_percent, - m.heap_bytes::float8 / NULLIF(m.heap_limit, 0) * 100 AS heap_percent + m.heap_bytes::float8 / NULLIF(m.heap_limit, 0) * 100 AS heap_percent, + m.swap_bytes::float8 / NULLIF(m.heap_limit, 0) * 100 AS swap_percent FROM mz_catalog.mz_cluster_replicas AS r JOIN mz_catalog.mz_cluster_replica_sizes AS s ON r.size = s.size @@ -5486,6 +5497,7 @@ pub static MZ_CLUSTER_REPLICA_UTILIZATION_HISTORY: LazyLock = .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("swap_percent", SqlScalarType::Float64.nullable(true)) .with_column( "occurred_at", SqlScalarType::TimestampTz { precision: None }.nullable(false), @@ -5510,6 +5522,10 @@ pub static MZ_CLUSTER_REPLICA_UTILIZATION_HISTORY: LazyLock = "heap_percent", "Approximate heap (RAM + swap) usage, in percent of the total allocation.", ), + ( + "swap_percent", + "Approximate swap usage, in percent of the total heap allocation.", + ), ( "occurred_at", "Wall-clock timestamp at which the event occurred.", @@ -5523,6 +5539,7 @@ SELECT m.memory_bytes::float8 / NULLIF(s.memory_bytes, 0) * 100 AS memory_percent, m.disk_bytes::float8 / NULLIF(s.disk_bytes, 0) * 100 AS disk_percent, m.heap_bytes::float8 / NULLIF(m.heap_limit, 0) * 100 AS heap_percent, + m.swap_bytes::float8 / NULLIF(m.heap_limit, 0) * 100 AS swap_percent, m.occurred_at FROM mz_catalog.mz_cluster_replicas AS r diff --git a/src/controller/src/lib.rs b/src/controller/src/lib.rs index 6e675eb3001d1..7547d1d403ee5 100644 --- a/src/controller/src/lib.rs +++ b/src/controller/src/lib.rs @@ -636,6 +636,7 @@ impl Controller { Datum::TimestampTz(now_tz), m.heap_bytes.into(), m.heap_limit.into(), + m.swap_bytes.into(), ]); (row.clone(), mz_repr::Diff::ONE) }) diff --git a/src/orchestrator-kubernetes/src/lib.rs b/src/orchestrator-kubernetes/src/lib.rs index 3515b7d07b495..98be7d1b6889b 100644 --- a/src/orchestrator-kubernetes/src/lib.rs +++ b/src/orchestrator-kubernetes/src/lib.rs @@ -1592,6 +1592,7 @@ impl OrchestratorWorker { }; process_metrics.heap_limit = usage.heap_limit; + process_metrics.swap_bytes = usage.swap_bytes; } process_metrics diff --git a/src/orchestrator-process/src/lib.rs b/src/orchestrator-process/src/lib.rs index 2c1c599d5fe2d..3eb515a7eb5c7 100644 --- a/src/orchestrator-process/src/lib.rs +++ b/src/orchestrator-process/src/lib.rs @@ -559,6 +559,7 @@ impl OrchestratorWorker { disk_bytes: None, heap_bytes: None, heap_limit: None, + swap_bytes: None, }); } Ok(metrics) diff --git a/src/orchestrator/src/lib.rs b/src/orchestrator/src/lib.rs index 5503065867aa1..e70accf1e1eed 100644 --- a/src/orchestrator/src/lib.rs +++ b/src/orchestrator/src/lib.rs @@ -154,6 +154,7 @@ pub struct ServiceProcessMetrics { pub disk_bytes: Option, pub heap_bytes: Option, pub heap_limit: Option, + pub swap_bytes: Option, } /// A simple language for describing assertions about a label's existence and value. diff --git a/src/storage-client/src/healthcheck.rs b/src/storage-client/src/healthcheck.rs index 1c164a2f24dd3..f3792b2b0e3b0 100644 --- a/src/storage-client/src/healthcheck.rs +++ b/src/storage-client/src/healthcheck.rs @@ -174,6 +174,7 @@ pub static REPLICA_METRICS_HISTORY_DESC: LazyLock = LazyLock::new( ) .with_column("heap_bytes", SqlScalarType::UInt64.nullable(true)) .with_column("heap_limit", SqlScalarType::UInt64.nullable(true)) + .with_column("swap_bytes", SqlScalarType::UInt64.nullable(true)) .finish() }); diff --git a/test/sqllogictest/autogenerated/mz_internal.slt b/test/sqllogictest/autogenerated/mz_internal.slt index 66cd3a52e3a4e..9f170e9453025 100644 --- a/test/sqllogictest/autogenerated/mz_internal.slt +++ b/test/sqllogictest/autogenerated/mz_internal.slt @@ -149,6 +149,7 @@ memory_bytes uint8 Approximate␠RAM␠usage,␠in␠bytes. disk_bytes uint8 Approximate␠disk␠usage,␠in␠bytes. heap_bytes uint8 Approximate␠heap␠(RAM␠+␠swap)␠usage,␠in␠bytes. heap_limit uint8 Available␠heap␠(RAM␠+␠swap)␠space,␠in␠bytes. +swap_bytes uint8 Approximate␠swap␠usage,␠in␠bytes. query TTT SELECT name, type, comment FROM objects WHERE schema = 'mz_internal' AND object = 'mz_cluster_replica_metrics_history' ORDER BY position @@ -161,6 +162,7 @@ disk_bytes uint8 Approximate␠disk␠usage,␠in␠bytes. occurred_at timestamp␠with␠time␠zone Wall-clock␠timestamp␠at␠which␠the␠event␠occurred. heap_bytes uint8 Approximate␠heap␠(RAM␠+␠swap)␠usage,␠in␠bytes. heap_limit uint8 Available␠heap␠(RAM␠+␠swap)␠space,␠in␠bytes. +swap_bytes uint8 Approximate␠swap␠usage,␠in␠bytes. query TTT SELECT name, type, comment FROM objects WHERE schema = 'mz_internal' AND object = 'mz_cluster_replica_statuses' ORDER BY position @@ -189,6 +191,7 @@ cpu_percent double␠precision Approximate␠CPU␠usage,␠in␠percent␠of memory_percent double␠precision Approximate␠RAM␠usage,␠in␠percent␠of␠the␠total␠allocation. disk_percent double␠precision Approximate␠disk␠usage,␠in␠percent␠of␠the␠total␠allocation. heap_percent double␠precision Approximate␠heap␠(RAM␠+␠swap)␠usage,␠in␠percent␠of␠the␠total␠allocation. +swap_percent double␠precision Approximate␠swap␠usage,␠in␠percent␠of␠the␠total␠heap␠allocation. query TTT SELECT name, type, comment FROM objects WHERE schema = 'mz_internal' AND object = 'mz_cluster_replica_utilization_history' ORDER BY position @@ -199,6 +202,7 @@ cpu_percent double␠precision Approximate␠CPU␠usage,␠in␠percent␠of memory_percent double␠precision Approximate␠RAM␠usage,␠in␠percent␠of␠the␠total␠allocation. disk_percent double␠precision Approximate␠disk␠usage,␠in␠percent␠of␠the␠total␠allocation. heap_percent double␠precision Approximate␠heap␠(RAM␠+␠swap)␠usage,␠in␠percent␠of␠the␠total␠allocation. +swap_percent double␠precision Approximate␠swap␠usage,␠in␠percent␠of␠the␠total␠heap␠allocation. occurred_at timestamp␠with␠time␠zone Wall-clock␠timestamp␠at␠which␠the␠event␠occurred. query TTT diff --git a/test/sqllogictest/catalog_server_explain.slt b/test/sqllogictest/catalog_server_explain.slt index 4fb1faa189f0a..113e1100c41d0 100644 --- a/test/sqllogictest/catalog_server_explain.slt +++ b/test/sqllogictest/catalog_server_explain.slt @@ -709,7 +709,7 @@ mz_internal.mz_cluster_replica_metrics_ind: mz_internal.mz_cluster_replica_metrics: →Map/Filter/Project - Project: #0..=#4, #6, #7 + Project: #0..=#4, #6..=#8 →Non-monotonic TopK Group By #0, #1 Order By #5 desc nulls_first @@ -8451,8 +8451,8 @@ Explained Query: →One-Shot Delta Join [%0:mz_cluster_replicas » %1:mz_cluster_replica_sizes[#0{size}] » %2:mz_cluster_replica_metrics[#0{replica_id}]] path %0: after %2: - Project: #0, #4, #10..=#13 - Map: ((uint8_to_double(#5{cpu_nano_cores}) / uint8_to_double(case when (0 = uint8_to_numeric(#1{cpu_nano_cores})) then null else #1{cpu_nano_cores} end)) * 100), ((uint8_to_double(#6{memory_bytes}) / uint8_to_double(case when (0 = uint8_to_numeric(#2{memory_bytes})) then null else #2{memory_bytes} end)) * 100), ((uint8_to_double(#7{disk_bytes}) / uint8_to_double(case when (0 = uint8_to_numeric(#3{disk_bytes})) then null else #3{disk_bytes} end)) * 100), ((uint8_to_double(#8{heap_bytes}) / uint8_to_double(case when (0 = uint8_to_numeric(#9{heap_limit})) then null else #9{heap_limit} end)) * 100) + Project: #0, #4, #11..=#13, #15, #16 + Map: ((uint8_to_double(#5{cpu_nano_cores}) / uint8_to_double(case when (0 = uint8_to_numeric(#1{cpu_nano_cores})) then null else #1{cpu_nano_cores} end)) * 100), ((uint8_to_double(#6{memory_bytes}) / uint8_to_double(case when (0 = uint8_to_numeric(#2{memory_bytes})) then null else #2{memory_bytes} end)) * 100), ((uint8_to_double(#7{disk_bytes}) / uint8_to_double(case when (0 = uint8_to_numeric(#3{disk_bytes})) then null else #3{disk_bytes} end)) * 100), uint8_to_double(case when (0 = uint8_to_numeric(#9{heap_limit})) then null else #9{heap_limit} end), ((uint8_to_double(#8{heap_bytes}) / #14) * 100), ((uint8_to_double(#10{swap_bytes}) / #14) * 100) →Unarranged Raw Stream →Fused with Child Map/Filter/Project Project: #0, #3 @@ -8478,8 +8478,8 @@ Explained Query: →One-Shot Delta Join [%0:mz_cluster_replicas » %1:mz_cluster_replica_sizes[#0{size}] » %2:mz_cluster_replica_metrics_history[#0{replica_id}]] path %0: after %2: - Project: #0, #4, #11..=#14, #8 - Map: ((uint8_to_double(#5{cpu_nano_cores}) / uint8_to_double(case when (0 = uint8_to_numeric(#1{cpu_nano_cores})) then null else #1{cpu_nano_cores} end)) * 100), ((uint8_to_double(#6{memory_bytes}) / uint8_to_double(case when (0 = uint8_to_numeric(#2{memory_bytes})) then null else #2{memory_bytes} end)) * 100), ((uint8_to_double(#7{disk_bytes}) / uint8_to_double(case when (0 = uint8_to_numeric(#3{disk_bytes})) then null else #3{disk_bytes} end)) * 100), ((uint8_to_double(#9{heap_bytes}) / uint8_to_double(case when (0 = uint8_to_numeric(#10{heap_limit})) then null else #10{heap_limit} end)) * 100) + Project: #0, #4, #12..=#14, #16, #17, #8 + Map: ((uint8_to_double(#5{cpu_nano_cores}) / uint8_to_double(case when (0 = uint8_to_numeric(#1{cpu_nano_cores})) then null else #1{cpu_nano_cores} end)) * 100), ((uint8_to_double(#6{memory_bytes}) / uint8_to_double(case when (0 = uint8_to_numeric(#2{memory_bytes})) then null else #2{memory_bytes} end)) * 100), ((uint8_to_double(#7{disk_bytes}) / uint8_to_double(case when (0 = uint8_to_numeric(#3{disk_bytes})) then null else #3{disk_bytes} end)) * 100), uint8_to_double(case when (0 = uint8_to_numeric(#10{heap_limit})) then null else #10{heap_limit} end), ((uint8_to_double(#9{heap_bytes}) / #15) * 100), ((uint8_to_double(#11{swap_bytes}) / #15) * 100) →Unarranged Raw Stream →Fused with Child Map/Filter/Project Project: #0, #3 @@ -10527,7 +10527,7 @@ Explained Query: →Differential Join %0:l4[#0{entity_name}, #1{name}] » %1[#0{entity_name}, #1{column_name}] →Arranged l4 →Arrange (#0{entity_name}, #1{column_name}) - →Constant (283 rows) + →Constant (284 rows) →Return →Union →Map/Filter/Project diff --git a/test/sqllogictest/mz_catalog_server_index_accounting.slt b/test/sqllogictest/mz_catalog_server_index_accounting.slt index 75b12a7d85855..5e47356389bf6 100644 --- a/test/sqllogictest/mz_catalog_server_index_accounting.slt +++ b/test/sqllogictest/mz_catalog_server_index_accounting.slt @@ -274,6 +274,7 @@ mz_cluster_replica_metrics heap_limit mz_cluster_replica_metrics memory_bytes mz_cluster_replica_metrics process_id mz_cluster_replica_metrics replica_id +mz_cluster_replica_metrics swap_bytes mz_cluster_replica_metrics_history cpu_nano_cores mz_cluster_replica_metrics_history disk_bytes mz_cluster_replica_metrics_history heap_bytes @@ -282,6 +283,7 @@ mz_cluster_replica_metrics_history memory_bytes mz_cluster_replica_metrics_history occurred_at mz_cluster_replica_metrics_history process_id mz_cluster_replica_metrics_history replica_id +mz_cluster_replica_metrics_history swap_bytes mz_cluster_replica_name_history id mz_cluster_replica_name_history new_name mz_cluster_replica_name_history occurred_at diff --git a/test/sqllogictest/system-cluster.slt b/test/sqllogictest/system-cluster.slt index c3325b8959e6f..8b6f37efd2f51 100644 --- a/test/sqllogictest/system-cluster.slt +++ b/test/sqllogictest/system-cluster.slt @@ -410,7 +410,7 @@ ORDER BY r.id; ---- Explained Query: Finish order_by=[#0{id} asc nulls_last] output=[#0..=#5] - Project (#11{id}, #12{name}, #0{id}, #14{size}, #1{name}, #29) + Project (#11{id}, #12{name}, #0{id}, #14{size}, #1{name}, #30) Map (((uint8_to_double(#25{memory_bytes}) / uint8_to_double(case when (0 = uint8_to_numeric(#19{memory_bytes})) then null else #19{memory_bytes} end)) * 100)) Join on=(#0{id} = #13{cluster_id} AND #11{id} = #22{replica_id} AND #14{size} = #15{size}) type=delta ArrangeBy keys=[[#0{id}]] From c4885878f14af161cee1835e8d7e74553bcfc80a Mon Sep 17 00:00:00 2001 From: Materialize Date: Sat, 12 Sep 2026 00:00:00 +0000 Subject: [PATCH 2/6] LICENSE: update change date --- LICENSE | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/LICENSE b/LICENSE index 0cc0ff1ff8409..f0e5e32d33ff2 100644 --- a/LICENSE +++ b/LICENSE @@ -13,7 +13,7 @@ Business Source License 1.1 Licensor: Materialize, Inc. -Licensed Work: Materialize Version 20260911 +Licensed Work: Materialize Version 20260912 The Licensed Work is © 2026 Materialize, Inc. Additional Use Grant: Within a single installation of the Licensed Work, @@ -39,7 +39,7 @@ Additional Use Grant: Within a single installation of the Licensed Work, whose definitions are controlled by such third parties. -Change Date: September 11, 2030 +Change Date: September 12, 2030 Change License: Apache License, Version 2.0 From 1e56d9c9c4747d21a435e90821304ca603a0ff5d Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Fri, 11 Sep 2026 20:16:53 -0400 Subject: [PATCH 3/6] compute: rewire mz_join_core to a container builder (#38368) Rewires `mz_join_core` to take a container builder instead of an output container, then uses that to let a linear join's last stage write the node's output edge directly. Follows up on the review note on #37784. `vec_to_columnar` encoded the last stage's output after the fact whenever no finalization closure followed it. The stage now writes a `ConsolidatingColumnBuilder`, which drops the operator hop and the `Vec` of owned `Row`s that used to sit on the edge in between. The builder choice is per stage rather than global. A `Column` is the wrong intermediate for a consumer that re-encodes what it reads, and both a following stage's arrangement and a finalization closure do exactly that: a `Vec` hands them moved `Row` allocations where a `Column` would copy row bytes first. Every non-terminal stage therefore keeps the `Vec` accumulator, and an error-capable closure keeps it even when terminal, because its output passes through `ok_err` and the demux materializes the ok side as a `Vec` regardless. The delta join's `vec_to_columnar` is not addressed here. Its stages chain through a `Vec`-typed `half_join` input and its output passes through three more operators before the node boundary, so it needs its own change. Posted by Claude Code --- src/compute/src/render/join/linear_join.rs | 215 +++++++++++++++----- src/compute/src/render/join/mz_join_core.rs | 22 +- 2 files changed, 171 insertions(+), 66 deletions(-) diff --git a/src/compute/src/render/join/linear_join.rs b/src/compute/src/render/join/linear_join.rs index 0b960c688614f..53cd8fa218dec 100644 --- a/src/compute/src/render/join/linear_join.rs +++ b/src/compute/src/render/join/linear_join.rs @@ -36,8 +36,11 @@ use mz_timely_util::columnar::{ Col2ValBatcher, Col2ValColBatcher, Col2ValPagedBatcher, columnar_exchange, }; use mz_timely_util::operator::{CollectionExt, StreamExt}; +use timely::ContainerBuilder; +use timely::container::{CapacityContainerBuilder, PushInto}; use timely::dataflow::channels::pact::{ExchangeCore, Pipeline}; use timely::dataflow::operators::OkErr; +use timely::dataflow::operators::generic::Operator; use timely::dataflow::{Scope, Stream}; use crate::extensions::arrange::{ArrangementBatcher, MzArrangeCore}; @@ -100,15 +103,21 @@ impl LinearJoinSpec { } } - /// Render a join operator according to this specification. - fn render<'s, T, Tr1, Tr2, L, I>( + /// Render a join operator according to this specification, assembling its + /// output through `CB`. + /// + /// The `DifferentialDataflow` implementation builds its own `Vec` output and + /// cannot be handed a container builder, so that arm re-encodes through `CB`. + /// The `Materialize` implementation writes `CB` directly. + fn render<'s, T, Tr1, Tr2, L, I, CB>( &self, arranged1: Arranged<'s, Tr1>, arranged2: Arranged<'s, Tr2>, result: L, - ) -> VecCollection<'s, T, I::Item, Diff> + ) -> Stream<'s, T, CB::Container> where T: Lattice + timely::progress::Timestamp, + CB: ContainerBuilder + PushInto<(I::Item, T, Diff)> + 'static, Tr1: TraceReader + Clone + 'static, Tr2: TraceReader + Clone + 'static, BatchCursor: Cursor