mz-optimize-memory

# Optimize Cluster Memory Reduce what a Materialize compute cluster spends on memory, with every claim backed by a measurement. The deliverable of an engagement is a set of per-item verdicts with measured evidence, validated changes expressed in the user's own deployment tooling, and a right-sized replica at the end. Freed memory is not saved money until the replica is resized down. For freshness problems (an object lagging behind wall-clock time) use the mz-debug-freshness skill instead. Memory work and freshness work share instruments but have different workflows. ## Ground rules - Ask before changing anything, and before reading the user's data. Catalog, introspection and EXPLAIN reads need no permission; a SELECT over the user's relations (a cardinality probe, the match-rate query, an exactness proof) reads their data and is asked for first. Anything that creates, alters, drops, or writes rows, in the user's environment or on their machine (including a local Docker container), is proposed first and run only after a yes. Approval covers a class of change, not each statement: an approved experiment cluster covers building and dropping candidate objects on it, and anything beyond it needs a fresh ask. - Never change objects on production clusters, even with permission to experiment. Object definitions usually have a source of truth outside Materialize (mz-deploy, dbt, or deploy scripts), and a catalog-only change is silently reverted, or half-reverted, by the next deploy. Accepted changes are delivered as edits to that source of truth, or as DDL for the user to run where the catalog is the source. - Experiments run on a dedicated experiment cluster that the user creates (you supply the commands). Never build or rebuild candidate objects on a production cluster: an object built there hydrates on the production replica and spends production memory and CPU on an experiment. Read-only `EXPLAIN ANALYZE` and introspection reads against production are diagnosis and are fine; anything that hydrates goes to the experiment cluster. - Do not recommend `REFRESH EVERY` materialized views as an optimization lever in any situation. - Every verdict rests on a measurement: both sides sized, the join key read from the plan, the consumer count counted. A graph-shape or plan-text argument alone is not a verdict, and an estimate gates an experiment but never settles one: shared arrangements restructure join plans in ways static bytes-per-row arithmetic (the width math of step 5) cannot see, so an estimate can err either way, and any non-trivial index or boundary verdict (a boundary: an indexed view placed at a (collection, key) point several consumers share) gets a measured build before it ships. - Leave-alone is the default outcome of a check, and it stays in your notes. Check many things (a hint that measures correct, an exact min/max or TopK whose arrangement is just the input it must retain to answer retractions, a single-copy arrangement no signature applies to), change only what a measured signature calls for, and report only the opportunities. A change without such a signature is churn: a rebuild for no measured saving. State a leave-alone only for an item the user asked about (a proposal item, a named object) or when the residual footprint needs explaining, such as why the replica cannot go lower. - Batch by confidence. Changes you are at least about 97% sure of (a hint retune that `mz_expected_group_size_advice` measured, an index drop that no plan reads, removal of columns the plan proves unread) can ship in one batch; anything less certain is applied alone and re-measured before the next, so each saving stays attributable and an interaction (a new index restructuring a neighbour's plan) shows up as its own number. Record the estimate per change and the measurement per batch either way. ## Connection and tools Prefer the Materialize MCP developer endpoint when it is available. Its tools describe themselves; what matters for this skill: every cluster-bound read (`EXPLAIN ANALYZE`, `mz_introspection`, `EXPLAIN` of user objects) goes through the `query` tool with its cluster argument, plus its `cluster_replica` argument on a multi-replica cluster (any extra replica counts, including an unbilled support replica on Cloud). The endpoint's other tool, `query_system_catalog`, is a trap here: it does not refuse those statements (the `mz_introspection` relations pass its name check) and answers them about the session's default cluster: with exactly one replica there the answer looks valid (often an empty result, or another cluster's numbers) and carries no error; with several it fails with `log source reads must target a replica`, with none with `has no replicas available to service request`. Never send a cluster-bound read there. The tools cannot SET anything, so the cluster is always the argument, never a session setting. On servers before v26.40 the endpoint's own instructions say never to query `mz_introspection.mz_dataflow_arrangement_sizes`; that rule predates the `query` tool and concerns the catalog tool only (later servers say so themselves), so it does not apply to `query` with a cluster argument (what that relation's `id` joins to is in "Reading introspection"). Tool responses are bare row arrays without column names, numbers as strings, capped by default at 1 MB per response and 60 seconds per request (both per-environment settings an operator can move), so keep SELECT lists short and in a fixed order, and give census queries a `LIMIT` or an aggregation rather than listing every arrangement of a big cluster (a `LIMIT` bounds the result, not the work of a join). A timed-out request cancels the peek but sometimes not the work: its dataflow can run to completion on the cluster, so do not resend it unchanged. Timestamps arrive as milliseconds since the epoch unless cast to `text`, and object names outside the session's database need the database prefix, or the statement fails with `unknown catalog item` when a schema of that name exists in the session's database and with `unknown schema` when it does not. If no MCP server is configured, or its `query` tool is absent (older Materialize, or disabled by the operator; `query` needs v26.30 and its `cluster_replica` argument v26.33), suggest setting one up (the mz-health-check skill covers client configuration), or ask the user for a SQL connection string instead. Whenever you ask for a connection string, warn the user that it should be for a role scoped to what the current phase needs (read-only for diagnosis; see "Making changes", below, for the experiment role), never an admin or superuser. Check that the scoping is real before relying on it: `SHOW enable_rbac_checks`. Cloud enforces RBAC by default, while self-managed instances default to `enableRbac: false` and the emulator image turns the checks off, and with the checks off the GRANTs and ownership gates on user objects stop being enforced, so a scoped role protects nothing there (the gates on system objects, and `restrict_to_user_objects` below, are separate and still apply). Ask also that the role not have `restrict_to_user_objects` set, which hides the system catalog and makes `EXPLAIN ANALYZE` fail on objects the role does not own, even with the RBAC checks off. All writes happen outside MCP. The "Making changes" section defines who executes them and where. ## Reading introspection Rules for every introspection read, over MCP or SQL alike: - **Measure hydrated dataflows only.** Introspection state (everything `EXPLAIN ANALYZE` and `mz_introspection` read) fills as hydration completes: a still-hydrating dataflow returns empty or partial numbers, and what it does return measures catching up, not steady state. Gate every measurement on `mz_internal.mz_hydration_statuses` (maintained by the control plane, so it answers even when the replica is busy). This applies equally after every rebuild you perform mid-engagement: rebuild, wait for hydrated, then re-measure. Right after a replica crash the relation can briefly serve the previous incarnation's rows, so re-read it once the replica is back before trusting a `true`. For a cluster that cannot reach hydration at all, see "Intake: a cluster that cannot hydrate". - Introspection is served by the target cluster and is session-scoped: `SET cluster = ...` first (or the cluster argument of the MCP `query` tool). On a cluster with several replicas, target one replica too (`SET cluster_replica = ...`, or the tool's `cluster_replica` argument): introspection reads on a multi-replica cluster fail outright otherwise ("log source reads must target a replica"), readings legitimately differ across replicas, and the one you measure must itself be hydrated. - Results can lag a few seconds behind DDL; if a query returns empty rows right after a change, wait and re-run before concluding anything. - `mz_introspection.mz_dataflow_arrangement_sizes.id` is a dataflow id (`uint8`), never joinable to `mz_objects.id`: reach the catalog through `mz_introspection.mz_compute_exports` (`dataflow_id` to `export_id`), or match `name` (`Dataflow: <db>.<schema>.<object>`, the name the object was created under, not its current name: a blue/green deploy or a dbt run creates objects under a staging name and swaps them in, so the dataflow keeps the staging name and only the export path finds it). - The docs' dataflow troubleshooting page (https://materialize.com/docs/transform-data/dataflow-troubleshooting/) carries the census and per-operator queries this skill builds on and the CPU-side instruments it does not repeat (`mz_scheduling_elapsed`, `mz_compute_operator_durations_histogram`, the `mz_dataflow_addresses` region walk); the freshness page (https://materialize.com/docs/transform-data/freshness-troubleshooting/) covers lag. ## Intake: a cluster that cannot hydrate Users often arrive with a cluster that is OOMing, not with a stable but oversized one. Do not start the measurement workflow against an OOM-looping replica: a replica that dies mid-hydration yields no usable measurements ("Reading introspection"). Confirm the state first from two control-plane relations, which answer even while the replica is looping: `mz_internal.mz_hydration_statuses` keeps reporting `hydrated = false` across the replica's restarts (a single `false` reading only says still hydrating; a cluster with no replica rows in `mz_cluster_replicas` also stays `false`, with `replica_id` NULL), and `mz_internal.mz_cluster_replica_status_history` confirms that MEMORY is the cause (`offline` rows with reason `oom-killed`, which covers every memory-class limit, RAM, heap and spill disk alike, without saying which; any other failure leaves the reason NULL; the current `mz_cluster_replica_statuses` often reads `online` between two kills, so it cannot; the emulator's process orchestrator records no reason, so there count the `offline` rows). The second check matters because a cluster can also fail to hydrate on a compute grind (a cross join or a skewed key grinding one worker for hours with flat memory: the quadratic-join shape, below; `mz_introspection.mz_scheduling_elapsed_per_worker` summed per worker shows the one busy worker whenever the replica still answers) or on starved inputs, and a bigger replica fixes neither. The loop is over, and measurement meaningful, once every object on the cluster is hydrated and the latest history row per process is `online`; kills that recur after hydration completes are the steady-state case this skill exists for, which ends only when the cluster holds less state (the levers below) or the replica gets bigger. Offer two ways to reach measurable ground, both executed by the user: - Develop against a smaller input data set until the pipeline is optimized, then scale the data back up. - Temporarily size the cluster (or a clone of it) large enough to hydrate without OOMing, measure and optimize there, then right-size down. The oversized replica is a measurement platform, not the fix. ## Cost model The high-level model: - Indexes and materialized views are maintained INCREMENTALLY by persistent dataflows that update their results as inputs change. That is why they need memory continuously: the operators must retain enough state to react to any input change. - Most of the retained state lives in ARRANGEMENTS: in-memory collections of rows organized by a key, maintained as inputs change. The kind of arrangement decides which lever can reach it. INDEX arrangements are shareable: `CREATE INDEX` maintains an arrangement of the indexed collection that other dataflows can read, but only WITHIN the index's cluster. Every other arrangement is intra-dataflow: private working state visible only to the dataflow that built it. - Intra-dataflow arrangements sit on operators' inputs and outputs. An operator's INPUT arrangement is either an explicit plan node or built inside the operator with no node of its own; both show in `EXPLAIN ANALYZE MEMORY`. The explicit node has three spellings in EXPLAIN: `ArrangeBy` in `EXPLAIN OPTIMIZED PLAN`; `ArrangeBy` with one `arrangements[i]` line per key in `EXPLAIN PHYSICAL PLAN` (verbose text, that statement's default); and `Arrange (key)` in the arrow text that a bare `EXPLAIN <object>` or `EXPLAIN PHYSICAL PLAN AS TEXT` prints. Some operators can REUSE an arrangement that already exists on their input (an index, or an upstream operator's arranged output) instead of building one, and some leave their OUTPUT arranged so the next operator in the same dataflow can reuse it. Reuse is within a dataflow; sharing across dataflows always needs an index. Those two properties, reuse of an input arrangement and an arranged output, decide what an index can replace; the operator table below gives them per operator. - Only operators that retain state cost memory. A join's output is a plain, unarranged collection, so "this join emits 500M rows" is not a memory cost by itself. Look at what the output feeds: hierarchical MIN/MAX reduces, TopKs, and explicit arrangements retain. Accumulable aggregates (sum/count/avg) collapse to per-group accumulators and stay cheap. - Not all retained memory is arrangement state: the persist sink at the end of every materialized view holds its own buffer, largest at hydration and drained afterwards, so that share of the peak is observable only during a hydration and a replica restart is how you get another one. See "Memory beyond arrangements" below. - A plain VIEW is a saved query, not a computed thing: every consuming dataflow INLINES the view's definition and the optimizer replans it in that consumer's context, pruning columns to that consumer's demand. Five dataflows over one view means five independent copies of its operators and arrangements. This is why a big view in the SQL is not a thing you will find in the dataflow census: its cost appears inlined inside each consumer, which is also why mining measured arrangements beats reading view SQL. Two objects stop the inlining and make a view computed once, with different products: - an INDEX on the view computes it once per cluster, and its product IS an arrangement, which consumers on that cluster read directly, joins included; - a MATERIALIZED VIEW computes it once globally and persists the results, readable from any cluster, but persisted results do NOT end in an arrangement: each consuming dataflow arranges what it reads by whatever key it needs. So an MV dedups the computation but not the arrangement state; indexing the MV on each reading cluster recovers the missing half for the consumers that read it on the index's key, one shared arrangement instead of one private copy each. - An index's own dataflow can hold more arrangements than the one it exports: indexing a computed view also computes the view, with whatever intra-dataflow arrangements that takes. The exported arrangement is the shareable product, not the dataflow's whole footprint. ### Operator-level detail Per physical-plan operator, with input arrangements counted whether printed as a node or built internally: | Operator | Input arrangement | Reuses an existing input arrangement | Output arranged, reusable | Memory | |---|---|---|---|---| | Get, Constant, Map/Filter/Project, FlatMap, Negate | none | reads through one without holding it | no | none of their own | | Explicit arrangement node (`ArrangeBy` / `Arrange`, spellings above) | builds one arrangement per key it prints (default text: `Arrange (key)`, one `(key)` each; verbose: one `arrangements[i]` line each); the node printed as `Unarranged Raw Stream` (verbose: `raw=true` with no `arrangements` line) builds nothing and only hands an existing arrangement through as a plain collection | yes, silently: it lists only the arrangements it builds; keys that the input already provides pass through unlisted, and a node whose keys all exist is elided from the plan | yes, one per key built | rows times width per key built; nothing for the raw-stream form | | Join, `type=differential` (linear) | both inputs of every stage are arranged: an input relation reuses an existing arrangement or gets a printed Arrange; the running intermediate result is one side of every stage after the first and is arranged internally (`JoinStage`) | yes | no | input relations plus every intermediate | | Join, `type=delta` | every input relation, by each key it is probed on: reuses an existing arrangement or gets a printed Arrange; no intermediate is arranged | yes | no | input relations only, no intermediates | | Reduce | internal, by the group key (hierarchical variants by (hash, group key)) | no | yes, by the group key | see Reduce below | | Distinct (a `SELECT DISTINCT` or key-only `GROUP BY`; `Distinct` in the optimized plan, `Reduce::Distinct` in the verbose physical plan, `Distinct GroupAggregate` in the arrow text and EXPLAIN ANALYZE) | internal, by all its columns (`Arranged DistinctBy`) | no | yes, by the whole row | about two records per group, input plus output, multiplicity only | | TopK | internal, per hierarchy level, by (hash, group key) | no | only `Monotonic Top1` | non-monotonic: input per level plus a reduced arrangement per level; `MonotonicTopK`: one stage, about (workers x limit) records per group while updates flow and near the limit once quiesced, plus a reduced arrangement; `Monotonic Top1`: one record per group | | Threshold | Arrange by the whole row (all columns, in order), printed by the physical plan only | in principle, never in practice (its input is a Union) | yes, by the whole row | one whole-row arrangement plus the output | | Consolidating Union, LetRec boundary, monotonic-input consolidation | no arrangement; a merge batcher instead | no | no | transient, see Consolidation below | | Persist sink (MV output, absent from EXPLAIN) | no arrangement; a correction buffer instead, logged as `mv_sink(<id>)::write` | no | no | see "Memory beyond arrangements" | **Reduce.** A Reduce arranges its own input in its own layout (group key plus the aggregate arguments), so it never consumes an existing arrangement and no index can replace that input arrangement: an index on the group key is read as a full scan, and the Reduce, a bare `count(*)` included, still builds its own (measured). What that input arrangement holds depends on the aggregate class: - Accumulable aggregates (sum, count, avg) carry their accumulators in the diff with empty values, so the input consolidates to one record per group and the arranged output adds a second: about two records per group, however many rows feed it. Cheap per row, not per group: the pair costs roughly 150 bytes per group for one accumulator (`count`, `sum`) and 300 for an `avg` (two), so a very large number of very small groups is where an accumulable stops being cheap. Do not calibrate this on a `generate_series` rig: over groups that are all alike the accumulators store compactly and the figure reads ten times lower. - A plain `SELECT DISTINCT` or key-only `GROUP BY` is the Distinct row above. An aggregate-level DISTINCT (`count(DISTINCT v)`) instead adds an arrangement pair sized by groups times distinct values. `DISTINCT ON` is not a Reduce at all but a TopK with `limit=1`. - Basic aggregates (`string_agg`, `jsonb_agg`, and the other non-accumulable, non-hierarchical ones) keep every qualifying input row of the group, thinned to the aggregate's argument columns. Identical thinned rows consolidate into one record carrying a multiplicity, so the record count is the distinct (group key, argument) tuples per group, not the rows: a low-cardinality argument column makes a large group small. - Non-monotonic min/max (the bucketed hierarchical rendering) keeps the same full input, thinned the same way; no exact incremental extremal aggregate can hold less, because a retraction of the current extremum must be answerable from what is retained. The hierarchy then adds two arrangements per bucket level (`Arranged MinsMaxesHierarchical input` and `Reduced Fallibly MinsMaxesHierarchical`), and every level whose bucket count exceeds the group's retained record count holds a near-full copy of the input, which is what GROUP SIZE hints trim. A monotonic min/max (append-only input) escapes this and keeps the winner in the diff, one record per group; an MV is never monotonic, whatever its definition (its sink may retract), so only a dataflow reading an append-only source directly gets that rendering. A GROUP BY mixing classes is planned as separate Reduces (one for the accumulables, one for the hierarchicals, one per basic aggregate, such as `string_agg` or `jsonb_agg`) joined on the group key; that join reads the Reduces' output arrangements directly and adds none of its own. Map/filter/project work fused into a Reduce leaves the output arranged (the `Fused with Child Map/Filter/Project` line is printed under Get and FlatMap nodes too, so by itself it is no sign of an arrangement); the fusion does not happen when the projection emits a mapped column, as `avg` does, and the `Mfp` left above the Reduce costs the next consumer a new arrangement. **TopK.** The non-monotonic rendering arranges its input per hierarchy level (`Arranged TopK input` in `mz_arrangement_sizes`) and keeps a reduced arrangement per level (`Reduced TopK input`), keyed by (hash, group key); the reduced arrangements hold the retractions of the rows the hierarchy does not emit, once in total rather than per level and concentrated in the last levels, near-full there whenever groups exceed the limit and empty when they fit inside it. Like Reduce, no TopK rendering consumes an existing arrangement, so no index can replace its input arrangement, and only `Monotonic Top1` leaves its result arranged, so a join over the other two gets a printed Arrange. An MV's output is never monotonic (see Reduce). **Join implementations.** A DIFFERENTIAL join over more than two inputs runs as a series of binary joins whose intermediate results are standing arrangements, proportional to the intermediate sizes, which can dwarf both the inputs and the output. A DELTA join maintains no intermediates: it streams each input's updates against the other inputs' existing arrangements, so its memory cost is exactly its input arrangements, which shared indexes can supply. Reading this in `EXPLAIN ANALYZE MEMORY`: the memory printed on a `Differential Join` operator line is exactly the eliminable intermediate state; the input arrangements are the separate Arrange lines below it, and those are kept under either implementation. Which implementation the plan picked (`type=differential` vs `type=delta`) often matters more than how big the join is, and it can be changed: supplying the input arrangements as indexes lets the planner pick a delta join ("Flipping differential joins to delta", references/indexes.md). **Consolidation.** A consolidating Union (printed `Consolidating Union`; the planner marks a Union when any of its inputs is a Negate, so EXCEPT and the outer-join lowerings carry one), the consolidation at a LetRec loop boundary, and the input consolidation of monotonic reduces and TopKs hold updates in a merge batcher until each update's timestamp completes, then emit them and keep nothing: no maintained trace. Their memory still shows in `mz_arrangement_sizes` (batcher records and bytes are summed into the same `records` and `size` columns under the consolidating operator's id: `UnionConsolidation` for a Union, `LetRecConsolidation` at a LetRec boundary, `Consolidated ReduceMonotonic input`, `Consolidated MonotonicTopK input` and `Consolidated MonotonicTop1 input` for the monotonic operators) and the row disappears once the batcher drains. Normally that is a hydration-time transient. It is standing state when a temporal filter feeds the consolidation: every future-dated retraction is parked until its time arrives, one record per live row for as long as the row lives. The near-term ones (within the `compute_temporal_bucketing_summary` horizon, two seconds by default) sit in the consuming operator's batcher and the rest in a separate `Temporal delay` operator just ahead of it (temporal bucketing), which logs them the same way under its own operator id. ### Memory beyond arrangements This matters most at hydration. The persist sink at the end of every materialized view stashes updates in a correction buffer until they are written. At hydration that buffer holds the MV's entire snapshot (nothing is written until the snapshot is complete), and the written copy is kept until persist reads it back, so the sink's peak is two copies of the MV's output: it scales with what the MV emits, not with what it reads or with the dataflow's own arrangements, which are separate lines in the same census. A rehydration of an existing MV peaks at the same two copies by another route: the sink writes the difference between the recomputed output and what the shard already holds, so it buffers both, with opposite signs, until they cancel. The sink does not appear in EXPLAIN, but its buffer IS in the arrangement introspection: the operator `mv_sink(<id>)::write` in `mz_arrangement_sizes`, rolled into the dataflow and object totals. When an MV's own plan applies the `mz_now()` predicate with nothing arranging or consolidating between it and the sink (a filter-and-project MV), the buffer also parks every future-dated retraction until its time arrives, one record per live row. A future-dated retraction is parked exactly once, by the first time-batching operator after the filter, so any intervening arrangement absorbs it instead, and a consumer that reads the filtered view through an index or another MV never parks it at all (its plan shows `ReadIndex` or the MV read, not the `mz_now()` Filter). A sink line that stays big after hydration is therefore either pending future-dated retractions (the line drains the moment they fire) or a write backlog, the sink not keeping up with persist. A consolidation batcher line reads the same way (see the Consolidation paragraph above). What the arrangement sum misses around the sink: its in-flight batches being appended to persist (only the `::write` buffer is logged, never the append) and allocator overhead beyond the logged `size`, largest right after a rebuild. To watch the sink buffer directly, read the `mz_persist_sink_correction_*` metrics via `mz_introspection.mz_cluster_prometheus_metrics` (`metric_name` column; cluster-wide counters and per-worker high-water gauges, no per-sink label). A replica's total heap (`heap_bytes` in `mz_internal.mz_cluster_replica_metrics`, RAM plus swap) therefore exceeds its arrangement sum, and post-hydration arrangement totals understate the peak twice over: they miss that gap, and the arrangements themselves peak at hydration, when each merge batcher holds the entire snapshot until the first seal, and trace merges (the batch merging inside an arrangement) transiently hold both source batches and their merged output (one index export measured 4.4x its settled size twelve seconds after its build). That is why replicas are sized from the measured whole-cluster hydration peak (see "Making changes", below), never from arrangement totals. The lever for a peak dominated by one MV's sink is in the lever table ("MV hydration spikes"). Read `heap_bytes`, not `memory_bytes`, for that comparison. On Cloud and with the self-managed defaults, replicas run with swap enabled and no scratch disk, so arrangements are ordinary swappable heap: `memory_bytes` counts resident RAM only and can fall below the arrangement sum once cold pages are paged out, `heap_limit` is the enforced RAM-plus-swap ceiling, and `disk_bytes` there reports swap, not disk. Treat swap as transparent unless there is so much of it that hydration time or freshness suffers. Under the process orchestrator (the emulator, and a local environmentd) the orchestrator reports NULL for `heap_bytes`, `heap_limit` and `disk_bytes`, so `memory_bytes` is all you get; it counts resident RAM, and a container on a host with swap enabled can still page out, so a falling `memory_bytes` there is not by itself freed memory. `mz_internal.mz_cluster_replica_utilization` reports the same readings as percentages of the replica's allocation (`cpu_percent`, `memory_percent`, and