catalog: unified object arrangement across clusters and replicas

[leedqin]

Motivation

We need to query per object memory across clusters and replicas from the catalog for Maintained Objects UI. There is no performant way of doing that without setting up session variables for each query against introspection. This allows the console to get all the object arrangements in the current moment and get historical data as well. This would also help make some other introspection queries querying object arrangements more simple in other parts of the console.

Design doc: doc/developer/design/20260331_unified_object_arrangement_sizes.md.

Linear issue: https://linear.app/materializeinc/issue/CNS-42/per-object-memory-arrangement-for-maintained-objects-in-the-console

Description

Adds per-object arrangement size tracking, following the introspection
subscribe pattern (same as mz_compute_hydration_times and
mz_compute_error_counts_raw_unified): a live
mz_internal.mz_object_arrangement_sizes source populated by a
per-replica subscribe, and a mz_internal.mz_object_arrangement_size_history
table that accumulates hourly snapshots (7-day retention). Both are
queryable from mz_catalog_server with no session variables or cluster
targeting.

Catalog surface

• New builtin source mz_internal.mz_object_arrangement_sizes (live) and builtin table mz_internal.mz_object_arrangement_size_history (with indexes on object_id and collection_timestamp).
• Two dyncfgs: arrangement_size_collection_interval (1h default) and arrangement_size_retention_period (7d default).

Runtime

• A per-replica introspection subscribe populates the live table, wired
  through the storage controller via a new IntrospectionType variant.
• A coordinator task periodically snapshots the live table and appends
  rows to the history table. Fires at the configured interval with a
  per-environment offset so fleets of envs don't collect in lockstep.
• A startup pruner retracts history rows older than the retention
  period. A migration guard prevents the history table from being
  truncated by schema migration, which would confuse the pruner.
• A hydration gate holds off the first snapshot until all compute
  objects have hydrated at least once, so early rows can't capture
  half-built arrangements.
• Two Prometheus metrics for observability:
  mz_arrangement_sizes_collection_time_seconds (histogram) and
  mz_arrangement_sizes_rows_written_total (counter).

Tests

• Unit tests for the pruner cutoff logic and the hydration check (10 cases).
• sqllogictest for the 10 MiB quantization formula (9 edge cases around the boundary).
• testdrive covering live-table visibility, filter-only MV absence, history accumulation, and drop retraction.

Verification

Verified with 10 Rust unit tests (pruner cutoff + hydration check), a
sqllogictest covering the 10 MiB quantization boundary, and a testdrive
end-to-end (live + history + drop retraction). Also ran locally against
a Postgres source with two replicas and confirmed ~300 history rows
accumulate with stable sizes and advancing collection_timestamp.
Verified with 10 Rust unit tests (pruner cutoff + hydration check), a
sqllogictest covering the 10 MiB quantization boundary, and a testdrive
end-to-end (live + history + drop retraction). Also ran locally against
a Postgres source with two replicas and confirmed ~300 history rows
accumulate with stable sizes and advancing collection_timestamp.

Tips For reviewing:

I would recommend reviewing it per commit

[def-]

After restarting Materialize, mz_object_arrangement_size_history can contain partial samples (collection_timestamps with rows for only a subset of the expected (object_id, replica_id) pairs) during the rehydration window. I'm not sure how bad that is?
With this test:

diff --git a/test/restart/mzcompose.py b/test/restart/mzcompose.py
index c8e4d4ab17..872619992d 100644
--- a/test/restart/mzcompose.py
+++ b/test/restart/mzcompose.py
@@ -976,6 +976,107 @@ def workflow_user_id_no_reuse_after_restart(c: Composition) -> None:
     c.sql("DROP TABLE idreuse_t1")


+def workflow_arrangement_sizes_hydration_gate(c: Composition) -> None:
+    """Regression test: arrangement-sizes hydration gate must hold across restart.
+
+    The race between the introspection-subscribe install and the
+    arrangement-size collection timer is narrow, so we repeat the
+    kill/up/observe cycle several times to keep the test reliable.
+    """
+
+    num_replicas = 2
+    arranged_names = tuple(f"mv{i}" for i in range(1, 11)) + tuple(
+        f"idx{i}" for i in range(1, 11)
+    )
+    expected_count = len(arranged_names) * num_replicas
+    name_filter = "(" + ", ".join(f"'{n}'" for n in arranged_names) + ")"
+    restart_rounds = 5
+
+    c.down(destroy_volumes=True)
+
+    with c.override(
+        Materialized(
+            additional_system_parameter_defaults={
+                "arrangement_size_collection_interval": "500ms",
+            },
+            sanity_restart=False,
+        )
+    ):
+        c.up("materialized")
+
+        mv_ddls = "\n".join(
+            f"CREATE MATERIALIZED VIEW mv{i} IN CLUSTER test AS "
+            f"SELECT a % {i + 1} AS k, SUM(b) AS s FROM t GROUP BY a % {i + 1};"
+            f"CREATE INDEX idx{i} IN CLUSTER test ON mv{i} (k);"
+            for i in range(1, 11)
+        )
+        c.sql(
+            dedent(
+                f"""
+                CREATE CLUSTER test SIZE 'scale=1,workers=1', REPLICATION FACTOR {num_replicas};
+                CREATE TABLE t (a int, b int);
+                INSERT INTO t SELECT g, g % 100 FROM generate_series(1, 100000) g;
+                {mv_ddls}
+                """
+            )
+        )
+
+        def wait_for_full_sample() -> str:
+            deadline = time.time() + 120
+            while time.time() < deadline:
+                rows = c.sql_query(
+                    f"""
+                    SELECT max(collection_timestamp)::text FROM (
+                        SELECT collection_timestamp
+                          FROM mz_internal.mz_object_arrangement_size_history h
+                          JOIN mz_objects o ON o.id = h.object_id
+                         WHERE o.name IN {name_filter}
+                         GROUP BY collection_timestamp
+                        HAVING count(*) = {expected_count}
+                    ) AS s
+                    """
+                )
+                if rows and rows[0][0] is not None:
+                    return rows[0][0]
+                time.sleep(0.5)
+            raise UIError(
+                f"timed out waiting for a sample with {expected_count} rows"
+            )
+
+        all_partials: list[tuple[int, str, int]] = []
+        for round_idx in range(restart_rounds):
+            pre_
restart_ts = wait_for_full_sample()
+
+            c.kill("materialized")
+            c.up("materialized")
+
+            time.sleep(8)
+
+            samples = c.sql_query(
+                f"""
+                SELECT collection_timestamp::text, count(*)
+                  FROM mz_internal.mz_object_arrangement_size_history h
+                  JOIN mz_objects o ON o.id = h.object_id
+                 WHERE o.name IN {name_filter}
+                   AND collection_timestamp > '{pre_restart_ts}'::timestamptz
+                 GROUP BY collection_timestamp
+                 ORDER BY collection_timestamp
+                """
+            )
+            assert (
+                samples
+            ), f"round {round_idx}: no post-restart collection_timestamps found"
+
+            for ts, cnt in samples:
+                if cnt != expected_count:
+                    all_partials.append((round_idx, ts, cnt))
+
+        assert not all_partials, (
+            f"{len(all_partials)} post-restart samples (across {restart_rounds} "
+            f"rounds) had row counts != {expected_count}: {all_partials}"
+        )
+
+
 def workflow_default(c: Composition) -> None:
     def process(name: str) -> None:
         if name == "default":

We can run bin/mzcompose --find restart down && bin/mzcompose --find restart run arrangement-sizes-hydration-gate:

==> mzcompose: test case workflow-arrangement-sizes-hydration-gate failed: builtins.AssertionError: 5 post-restart samples (across 5 rounds) had row counts != 40: [(1, '2026-04-23 04:24:37.167+00', 20), (2, '2026-04-23 04:24:55.667+00', 19), (2, '2026-04-23 04:24:56.408+00', 38), (3, '2026-04-23 04:25:11.167+00', 3), (3, '2026-04-23 04:25:11.877+00', 7)]
Traceback (most recent call last):
  File "/home/deen/git/materialize2/misc/python/materialize/mzcompose/composition.py", line 696, in test_case
    yield
  File "/home/deen/git/materialize2/misc/python/materialize/cli/mzcompose.py", line 857, in handle_composition
    composition.workflow(
    ~~~~~~~~~~~~~~~~~~~~^
        workflow_name, *args.unknown_subargs[1:], *extra_args
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    )
    ^
  File "/home/deen/git/materialize2/misc/python/materialize/mzcompose/composition.py", line 584, in workflow
    func(self)
    ~~~~^^^^^^
  File "/home/deen/git/materialize2/test/restart/mzcompose.py", line 1074, in workflow_arrangement_sizes_hydration_gate
    assert not all_partials, (
           ^^^^^^^^^^^^^^^^
AssertionError: 5 post-restart samples (across 5 rounds) had row counts != 40: [(1, '2026-04-23 04:24:37.167+00', 20), (2, '2026-04-23 04:24:55.667+00', 19), (2, '2026-04-23 04:24:56.408+00', 38), (3, '2026-04-23 04:25:11.167+00', 3), (3, '2026-04-23 04:25:11.877+00', 7)]
mzcompose: error: at least one test case failed

[antiguru]

Left some comments inline. I wonder about the < 10Mib/> 10MiB split. Small objects can have just as much churn as larger ones, why treat them differently? What is the problem we're trying to solve for?

[antiguru]

It's a bit unfortunate you'll need to go through the raw collections, but right now this seems to be the best approach. The views we define on top of the raw collections do more than what you need here, and certainly will be more expensive to maintain.

[antiguru]

Count don't need the aggregate input group size annotation (it's only for hierarchical reductions, not for simple ones.)

[antiguru]

Note this bakes in an assumption that's not guaranteed to be true in the future, which is that one dataflow has one export. If it'd change, we'd double-count. Just calling it out. Once it happens, we'd need to re-think what the granularity for this information is instead.

[antiguru]

This will have a lot of noise independently of the 10MiB gate. Running the subscribe on an otherwise empty cluster shows an update for at least one export every second. You could filter temporary exports t*, as they won't be reflected anywhere, but that doesn't solve the problem.

[antiguru]

Make sure the collection task reacts to change to the interval (without going through an envd restart)

[antiguru]

In the limit this'll mean we never write any data as there can always be an object somewhere that isn't hydrated. This checks for all objects, correct? Alternatively, just write down the size for objects that are hydrated.

Why gate on hydration status?

[leedqin]

yes! I think this was just overly defensive and I assumed that it would make sense to care about the hydration across all replicas for that object. But I removed it so now i am writing down the size for objects that are hydrated.

[antiguru]

Use a DatumVecBorrow here too.

[antiguru]

Suggested change

	Duration::from_secs(60 * 60),
	Duration::from_hours(1),

[antiguru]

Suggested change

	Duration::from_secs(7 * 24 * 60 * 60),
	Duration::from_hours(7 * 24),

[antiguru]

Add a sentinel value (0s?) to disable the collection.

[leedqin]

Left some comments inline. I wonder about the < 10Mib/> 10MiB split. Small objects can have just as much churn as larger ones, why treat them differently? What is the problem we're trying to solve for?

@antiguru That makes sense! I was reasoning about absolute churn rather than relative and the < 10 MiB exact-bytes branch was preserving precision for smaller object. I see that was the wrong abstraction. I am going to drop sub -10 MiB objects from mz_object_arrangments to help reduce the churn.

[leedqin]

@def- The test was happening due to a flag on hydration that is removed now arrangement_sizes_hydration_observed flag with a per-pair filter that runs on every snapshot, looking each
(replica_id, object_id) up in mz_compute_hydration_times and only
emitting a row if it's hydrated. As a result, there will be partial counts but if that needs to be avoided then I can just make the 'size' column Nullable.

Added this regression test 'edb42b2#diff-f65a1b69f0440a372762d0aeb6b80dcc2decd2ff75287c4961449abffa15964'

[antiguru]

Just some minor comments.

[antiguru]

You can override the flag in the testdrive file, too, to avoid scope creep.

[antiguru]

Unfortunately we're not great at namespacing our feature flags. Do you think you could find a prefix for the two flags you're introducing to indicate it's arrangement size logging in adapter? There's also the mz_introspection arrangement size infrastructure, and I'd like to avoid any confusion between the two.

[antiguru]

It might be interesting to report their size during hydration, right?

[leedqin]

Made this change in this latest commit: a5f46ff

[antiguru]

Looks good, thank you!

[def-]

Test changes lgtm

[def-]

I just noticed Nightly was not run for this, but should have been! https://ci.dev.materialize.com/trigger
A rebase would also have been good after x weeks before merging, I think that would have caught the Cargo test failure.