compute: stop error collections multiplying their cardinality

[antiguru]

Motivation

Closes: CPU-204

An error collection carries a multiplicity nothing reads. A dataflow is in an
error state if its error collection is non-empty, and the error surfaced to the
user is chosen arbitrarily, so an error's diff conveys nothing. Two rendering
patterns nonetheless grow that diff multiplicatively rather than additively, and
because every error in a dataflow consolidates onto one of a handful of distinct
error values, those diffs pile onto single records and reach Diff overflow on
plans of quite ordinary size.

The reported shape is plan-level sharing: a chain of diamond-shaped CTEs over a
dataflow that has any EvalError. The asymmetry that makes this an error-only
problem is worth stating plainly, since it explains why the ok side of the same
plan is fine: the branches of a diamond compute different ok rows, so ok
multiplicity does not compound, but they propagate the identical upstream
error collection, so error multiplicity doubles at every reconvergence.

Description

Two independent multipliers, fixed separately.

Delta join, N copies per join. render_delta_join propagated each input's
pre-existing errors once per delta path: build_update_stream concatenated the
source's errors and every stage's build_halfjoin concatenated a lookup
arrangement's errors, inside the per-path region. With N inputs that is N copies
of every input's errors assembled over N² concat edges, and since a join's
output is another join's input those factors compound through a nested plan.
Input error collections are now gathered once, outside the path loop, and the
two helpers return only the errors they themselves produce. No operator added,
no semantic change: the same union of streams, formed once instead of N times.
This multiplier is independent of the diamond mechanism and can stack on top of
it.

Plan-level sharing, f copies per level. Every reader of a binding
propagates that binding's errors independently, so a binding read f times
contributes its errors f times, and the factors apply again at each further
level of sharing. CollectionBundle::distinct_errs collapses a binding's error
multiplicities to one, applied at Let binding definitions that more than one
Get reads, which holds a dataflow's error multiplicity to the fan-out of a
single level rather than the product across levels. A binding one Get reads
cannot duplicate its own errors, so it is left alone; RenderPlan::reference_counts
supplies that gate.

Three decisions the diff cannot explain:

Why a reduce and not arithmetic on the diffs. Saturating the add, or
collapsing to a sign, looks cheaper and does not work: neither is additive, so
neither cancels when the errors retract, and a partially-retracted error would
persist forever as a phantom. Bounding multiplicity requires reading the
accumulated collection, which means a reduce over an arrangement. LetRec
already collapses its error variable this way and for this reason.

Why the binding definition and not the multi-input operators. Union and the
arrangement sites are where duplicate copies happen to meet again; the binding
is where the duplication originates. Collapsing at the origin is what actually
breaks the compounding across levels, rather than capping one confluence at a
time.

Why every representation in the bundle is collapsed, not just one. A bundle's
raw collection and each of its arrangements carry their own independent error
stream, and they are not even the same content — an arrangement's errors include
the key-formation errors that the raw collection's do not. Which form a consumer
reads is the consumer's choice (delta join reads both, in one operator), and the
binding definition cannot know that choice, so each form has to be collapsed.
The cost of a shared binding is therefore one arrange-plus-reduce for the raw
collection and one reduce per arrangement, the latter reusing the arrangement
that already exists.

Gated on enable_compute_error_distinct, off in production and pinned on for
the sqllogictest suite and mzcompose so the path is exercised before it earns a
production default. No release note: with the flag off, behavior is unchanged.

Deliberately out of scope, noted in the code: imported sources and indexes read
by more than one Get are not collapsed, and ArrangementFlavor::Trace error
traces cannot be rewritten in place by the importing dataflow. The structural
fix that would subsume all of this is an out-of-band per-dataflow error
collection that fallible operators append to once, instead of threading errors
through the plan DAG at all; that is a much larger change, and this is the
bounded version.

Verification

test/sqllogictest/error_semantics.slt gains a 70-level diamond-chain CTE over
an erroring expression, asserting the error still comes back; an overflow on
that path is a panic, which CI fails on.

This test is not yet proven red without the fix. Building sqllogictest
exhausted the available disk in my environment before I could run it in both
directions. Two things could keep it from reproducing: the optimizer may inline
or merge the CTE levels instead of keeping 70 Lets, and the doubling only
compounds where each level's errors are consolidated. Treat the test as
unproven until that runs — reviewers should not read its presence as evidence.

CPU-125 (sqlancer/sqlsmith overflows in CI) is a plausible independent check,
since sqlsmith generates exactly these shared-CTE shapes.

🤖 Generated with Claude Code

https://claude.ai/code/session_01GbtepuqcxmAffaM2qXZ4uA

[linear-code]

CPU-204

[antiguru]

This should be scoped to replicas.

[antiguru]

Done in 7bcb186 — added .scoped(ParameterScope::Replica).

Worth noting this is better than an env-wide flag for more than consistency: it means the collapse can be turned on for one unbilled replica against a real erroring dataflow, which is how the overflow was reproduced in the first place. The env-wide base still applies, so the sqllogictest and mzcompose pinning is unaffected.

---

Generated by Claude Code

[ggevay]

Thank you for the quick fix! Wrote some comments.

[ggevay]

(Maybe make it default to true?)

[antiguru]

Torn, so flagging rather than deciding.

Against: the repo convention is that a new flag defaults off in production and on in test/CI, so the path bakes in sqllogictest, testdrive and parallel-workload before it earns a production default. That's what's wired up now.

For: CPU-204 is firing at customers today and escalating, so shipping dark means the Sentry noise continues until someone remembers to flip it. And the escape hatch is good now that it's replica-scoped — LaunchDarkly can turn it off environment-wide or per replica without a release.

What tips me toward waiting is the measurement you posted below: +6.4% dataflow operators on mz_catalog_server. That's not free, and it argues for a cycle of soak before it's the production default. If you want it on now, I'd rather first take the demanded-forms optimisation (skip the collapse on bundle forms no Get reads, which kills the two-arrangement raw path where ensure_collections kept an unread passthrough) so the number we default on is smaller.

Your call — happy either way, just tell me which.

---

Generated by Claude Code

[ggevay]

The PR description says "No release note: with the flag off, behavior is unchanged". That seems to be not accurate, and I think it needs either a code change or a wording change before merge.

The delta-join rewiring is ungated: delta_join.rs never reads ENABLE_ERROR_DISTINCT, and this flag is consulted solely at render.rs:1079. So on upgrade every production delta join gets a rewired error graph while the collapse itself stays dark, and the rollout story ("pinned on for sqllogictest and mzcompose so the path is exercised before it earns a production default") covers only half the change.

[antiguru]

Correct, and the PR body was wrong. delta_join.rs never reads the flag — it's consulted only in distinct_shared_binding_errs — so with the flag off, delta joins still get the rewired error graph and only the collapse stays dark. "Behavior is unchanged" is false as written.

My preference is the wording change plus a test, not gating, for two reasons. Gating means restoring the errs threading through build_halfjoin/build_halfjoin2/build_halfjoin1/build_update_stream and carrying both paths through a fiddly area, which I think is more risk than it removes. And the failure mode actually worth protecting against isn't the multiplicity change (unobservable, strictly fewer copies) but bundle_errs missing a form and silently dropping errors, so a dataflow stops reporting one — a flag doesn't catch that, a test does. That's the delta-join test I owe you from the error_semantics.slt thread, and I'd rather that be the gate.

If you'd still rather have the switch, say so and I'll gate it — it's your call, and I'll wait for your answer before rewording, so we don't land the weaker option by default.

---

Generated by Claude Code

[ggevay]

Rec bindings never reach this. Both call sites are in lets loops (here and render.rs:1053), and in the LetRec path insert_id stores the raw bundle (render.rs:946) while only err_v receives the distinct (render.rs:989). So a Get rendered before the binding's insert_id resolves to the Variable and is collapsed, but forward reads (later rec bindings in the same block, the body, everything downstream) resolve to the raw bundle and are not. A diamond chain of rec bindings therefore still compounds level over level, which is the shape this PR fixes for lets.

reference_counts already descends RecBind values (render_plan.rs:596-605), so extending the collapse to the rec insertion looks small, and it would be good to have it in this PR rather than a follow-up. Worth noting that sqlsmith emits WMR, so CPU-125 would otherwise keep firing through this door.

[antiguru]

You're right, and this was a real gap rather than a wording one. Confirmed the mechanism exactly as you describe: insert_id at render.rs:945 stores the bundle straight from render_recursive_plan, so only reads rendered before it — which resolve to the feedback Variable, collapsed at :995 — saw collapsed errors. Everything forward resolved to the raw bundle.

Fixed in 0f3ec19 by collapsing the bundle before insert_id. I deliberately left the err extraction at :941 reading the uncollapsed bundle so the existing Distinct recursive err path on the feedback edge is untouched.

Agreed on the CPU-125 reasoning — that door being open is what made this worth fixing here rather than in a follow-up.

---

Generated by Claude Code

[ggevay]

Could this doc note that it bounds an input's errors to one copy per retained form, rather than to one copy outright? Each key form arranges its own copy of the input's errors (context.rs:1163, errs.clone().concat(errs_keyed) arranged per key), so an input retained under two lookup keys contributes its errors twice here. That sits a little awkwardly beside the claim on distinct_errs (context.rs:518) that collapsing holds a dataflow's error multiplicity to the fan-out of a single level.

The shape is reachable: SELECT q0.x, q0.y, q0.q FROM (SELECT x, y, 1/z AS q FROM e) q0, s1, t1 WHERE q0.x = s1.x AND q0.y = t1.y with indexes on s1(x) and t1(y) plans as type=delta with ArrangeBy keys=[[#0{x}], [#1{y}]] on the error-carrying input.

[ggevay]

Small wording point: imported errors do not arrive collapsed to one. They arrive bounded by the exporting dataflow's last-level fan-out. The collapse happens only at multiply-read Let bindings and never at the export itself, and a global read twice within one dataflow is never collapsed at all, since distinct_shared_binding_errs is only ever called with Id::Local.

The bound still holds and skipping Trace is still right, so this is only about the claim being slightly stronger than the guarantee.

[ggevay]

Two notes on this test.

It does not reproduce the overflow. With the flag off, nothing consolidates between levels, so the error records duplicate physically rather than accumulating as a diff. I ran it that way: 34 GB RSS after 10 minutes with no Overflow line, killed before it finished. So the test is genuinely red without the fix, but a future regression would surface as an OOM-killed replica or a timeout rather than a crisp failure, and it would not distinguish this bug from any other blowup. Adding CREATE DEFAULT INDEX per level produces the real thing, Overflow: 4611686018427387904 + 4611686018427387904 at level 63, in seconds. Happy to hand over that file.

Separately, every Get in this query is raw, so only the self.collection branch of distinct_errs runs and the ArrangementFlavor::Local collapse has no coverage at all. A variant where the levels feed a join would exercise that arm and the delta-join change at the same time, the latter currently having no test.

[antiguru]

Thank you for actually running it — that's the answer to the caveat I left in the PR body, and it's worse than "unproven": the test doesn't reproduce the bug. I'd flagged consolidation-between-levels as the thing that could stop it compounding and then had no way to check, since testdrive needs Docker and the optimized build exceeded this container's disk twice. You closed that out with a real measurement.

Yes please to the file. CREATE DEFAULT INDEX per level is the missing piece — an arrangement is what turns duplicated records into an accumulated diff, which is precisely the mechanism the fix targets, so an unindexed chain was never going to show Overflow. A test that fails as a 34 GB OOM also fails the "would a future regression be diagnosable" bar.

Also taking the second point: with every Get raw, only the self.collection branch of distinct_errs is covered and the ArrangementFlavor::Local arm has none. I'll add the join-fed variant so that arm and the delta-join rewiring both get coverage — the latter currently has no test at all, which is the same gap that makes the ungated-rewiring question below matter.

---

Generated by Claude Code

[ggevay]

@frankmcsherry was interested in operator counts on mz_catalog_server before, so we measured the delta on this branch, in an empty environment. Reproducible exactly across repeated runs:

	dataflow operators	addresses
flag off	15,917	40,711
flag on	16,931	42,747
delta	+1,014 (+6.4%)	+2,036 (+5.0%)

[def-]

Some strange stuff happens with multiple replicas when an MV is in error state:

diff --git a/test/cluster/mzcompose.py b/test/cluster/mzcompose.py
index 73c6dbfa93..4730f90851 100644
--- a/test/cluster/mzcompose.py
+++ b/test/cluster/mzcompose.py
@@ -6507,6 +6507,137 @@ def workflow_test_constant_sink(c: Composition) -> None:
                 """))


+def workflow_test_mv_error_multiplicity_divergence(c: Composition) -> None:
+    """
+    Test that two replicas of one cluster do not fight over an errored
+    materialized view's persist shard.
+
+    `enable_compute_error_distinct` decides whether a shared binding keeps the
+    error multiplicity its fan-out gives it, it is replica-scoped, and rendering
+    reads it once, when a dataflow is built. Two replicas that rendered on
+    either side of a flip therefore disagree about that multiplicity, and
+    neither is re-rendered. The MV sink writes `desired - persist` into a shard
+    both replicas share, so a disagreement that reaches the shard has each
+    replica correcting the other's writes at every batch description, forever,
+    with no input activity at all.
+    """
+
+    def replica_def(name: str, host: str) -> str:
+        return f"""
+            {name} (
+                STORAGECTL ADDRESSES ['{host}:2100'],
+                STORAGE ADDRESSES ['{host}:2103'],
+                COMPUTECTL ADDRESSES ['{host}:2101'],
+                COMPUTE ADDRESSES ['{host}:2102'],
+                WORKERS 2
+            )"""
+
+    def assert_mv_errors() -> None:
+        # Reading the MV waits for its shard's write frontier to pass the read
+        # timestamp, so this also waits for the sink to have written the error.
+        try:
+            c.sql_query("SELECT * FROM mv")
+        except DatabaseError as e:
+            assert "division by zero" in str(e), e
+        else:
+            raise RuntimeError("materialized view unexpectedly did not error")
+
+    def distinct_error_operators(replica: str) -> int:
+        with c.sql_cursor(
+            startup_params={"cluster": "cluster1", "cluster_replica": replica}
+        ) as cursor:
+            cursor.execute(
+                b"SELECT count(*) FROM mz_introspection.mz_dataflow_operators "
+                b"WHERE name LIKE 'Distinct errors%'"
+            )
+            return int(cursor.fetchall()[0][0])
+
+    def correction_insertions() -> float:
+        """Cumulative persist sink correction insertions, over both replicas."""
+        return sum(
+            Metrics(
+                c.exec(s, "curl", "localhost:6878/metrics", capture=True).stdout
+            ).get_summed_value("mz_persist_sink_correction_insertions_total")
+            for s in ("clusterd1", "clusterd2")
+        )
+
+    c.up("materialized", "clusterd1", "clusterd2")
+
+    c.sql(
+        """
+        ALTER SYSTEM SET unsafe_enable_unorchestrated_cluster_replicas = true;
+        ALTER SYSTEM SET enable_compute_error_distinct = false;
+        """,
+        port=6877,
+        user="mz_system",
+    )
+
+    c.sql(f"""
+        CREATE CLUSTER cluster1 REPLICAS ({replica_def("replica1", "clusterd1")});
+
+        CREATE TABLE t (a int, b int);
+        INSERT INTO t VALUES (1, 0);
+
+        -- Diamond-shaped bindings over a division by zero. Every level reads the
+        -- level below it twice, so the error reaches the sink with multiplicity
+        -- 8 when the collapse is off and 2 when it is on.
+        CREATE MATERIALIZED VIEW mv IN CLUSTER cluster1 AS
+            WITH
+                c0 AS (SELECT a, a / b AS q FROM t),
+                c1 AS (SELECT a, q FROM c0 WHERE a = 1 UNION ALL SELECT q, a FROM c0 WHERE a = 2),
+                c2 AS (SELECT a, q FROM c1 WHERE a = 1 UNION ALL SELECT q, a FROM c1 WHERE a = 2),
+                c3 AS (SELECT a, q FROM c2 WHERE a = 1 UNION ALL SELECT q, a FROM c2 WHERE a = 2)
+            SELECT * FROM c3;
+        """)
+
+    assert_mv_errors()
+
+    # Flip the flag and add a replica that renders the MV with the collapse on,
+    # while replica1 keeps running its uncollapsed dataflow.
+    c.sql(
+        "ALTER SYSTEM SET enable_compute_error_distinct = true",
+        port=6877,
+        user="mz_system",
+    )
+    c.sql(f"CREATE CLUSTER REPLICA cluster1.{replica_def('replica2', 'clusterd2')}")
+
+    for _ in range(120):
+        hydrated = c.sql_query("""
+            SELECT count(*)
+            FROM mz_internal.mz_compute_hydration_times h
+            JOIN mz_cluster_replicas r ON r.id = h.replica_id
+            JOIN mz_materialized_views v ON v.id = h.object_id
+            WHERE r.name = 'replica2' AND h.time_ns IS NOT NULL
+            """)[0][0]
+        if hydrated:
+            break
+        time.sleep(0.5)
+    else:
+        raise AssertionError("replica2 did not hydrate the materialized view")
+
+    # Without this the rest of the test could pass for the wrong reason, e.g.
+    # because the optimizer stopped sharing the view's bindings.
+    assert distinct_error_operators("replica1") == 0, "replica1 collapsed errors"
+    assert distinct_error_operators("replica2") > 0, "replica2 did not collapse errors"
+
+    # Replicas that agree stop touching the shard once they are hydrated and
+    # never insert into their correction buffers again. Replicas that disagree
+    # keep correcting each other in bursts, at least a dozen insertions a
+    # minute, separated by pauses of up to half a minute.
+    time.sleep(5)
+    before = correction_insertions()
+    time.sleep(60)
+    insertions = correction_insertions() - before
+
+    assert insertions <= 3, (
+        "the replicas keep correcting the materialized view's persist shard: "
+        f"{insertions} correction buffer insertions in a minute, with no input "
+        "activity at all"
+    )
+
+    assert_mv_errors()
+
+
 def workflow_test_memory_limiter(c: Composition) -> None:
     """
     Test that the memory limiter functions as expected.

Running bin/mzcompose --find cluster down && bin/mzcompose --find cluster run test-mv-error-multiplicity-divergence fails with: AssertionError: the replicas keep correcting the materialized view's persist shard: 22.0 correction buffer insertions in a minute, with no input activity at all