compute: stop error collections multiplying their cardinality

[antiguru]

Motivation

Closes: CPU-204
Follow-up: CPU-209

An error collection carries a multiplicity almost 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 about the query's
answer. Several 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

Three multipliers. Two are fixed here, the third only for materialized views.

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.

Sharing within a dataflow, 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 every Let and rec binding definition. This is
the multiplier the reported incidents were made of, and the one the regression
test drives to overflow.

Sharing across objects, f copies per level. A level of sharing expressed as
an indexed view rather than a CTE reads Get(Global), which a binding collapse
never sees, so the importing dataflow accumulates the exported diffs. Only half
of this is closed here, at the persist sink, whose shard another dataflow reads
back verbatim and whose contents are durable. Subscribes and one-shot copies are
read once by a client and never re-imported, so they do not pay for the
arrangement. The sink normalizes after the null assertions, whose errors follow
the ok row's multiplicity rather than the error collection's.

The index half is not fixed, and CPU-209 records why: collapsing at the
export gives the exported arrangement a second trace reader, which pins the
shared spine's physical frontier and makes importing dataflows panic in
cursor_through with upper straddles batch. That was measured four ways,
including a variant that builds the reduce, discards its output, and exports the
original trace, which panics too. So the fault is the reader's existence, not
its output or the frontier probe. Cross-object multiplicity through indexes
therefore stays unbounded. In proportion: object-graph path counts for the
reported incidents top out at 8, against roughly 7×10^30 within one dataflow, so
this half is reachable synthetically but is not what those incidents were.

Two 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 every representation in a 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, and a delta join reads both in one operator, so
the binding definition cannot know that choice.

Behavior and rollout

enable_compute_error_distinct gates the per-binding collapse, and now defaults
on. The divergence that argued for shipping dark is closed: the sink
normalization is ungated, so nothing this flag decides reaches durable state and
two replicas rendering under different values write the same thing. A default
change also arrives with a new binary that every replica is recreated on, whereas
a runtime flip is what leaves two live replicas rendering differently. Production
LaunchDarkly still serves off, so the flag is allowlisted in
launchdarkly-flag-consistency until that is reconciled.

Two parts are deliberately not gated, so "behavior is unchanged with the flag
off" would be wrong:

• The delta-join rewiring is unconditional. It is equivalence-preserving on the
  set of errors and strictly reduces copies, and the failure mode worth guarding
  against is bundle_errs missing a form and silently dropping a dataflow's
  errors, which a test catches and a flag does not.
• The sink normalization is unconditional, because gating it would leave the
  durable content of a materialized view's shard dependent on the flag.

Error counts change, in two places. mz_compute_error_counts_raw_unified sums
error diffs, so a diamond over one failing row previously reported 8 and now
reports 1; unshared plans are unaffected. And anything reading a materialized
view's shard now sees one error per distinct error rather than one per failing
row, which is what the idx2_div_by_zero golden moves record. Neither number is
promised: an object is tainted by having any error at all, and that is the part
that is preserved. The Console does surface these counts, so the change is
visible.

Verification

test/sqllogictest/error_semantics.slt gains a 70-level diamond CTE chain over
an erroring expression, asserting the error still comes back, where an overflow
is a panic CI fails on. Measured both directions:

Flag on	Flag off
PASS	Overflow: 4611686018427387904 + 4611686018427387904

Full file with the flag on: 153/153. With the flag off it panics in
consolidate_updates_slice_slow over DataflowErrorSer and the replica dies.

union rather than union all is what makes the chain reach the overflow: the
distinct arranges the error collection so duplicates consolidate into one record
with a doubling diff. Under union all the records duplicate physically instead
and the query exhausts memory, 34 GB with no completion, without ever
overflowing.

A cross-object chain of indexed views is deliberately not in the suite, since
nothing here bounds it. It belongs with CPU-209.

Thanks to @ggevay for measuring that the first version of that test did not
reproduce the bug at all, for the union all to union diagnosis, for finding
that the cross-object half was still unfixed, and for the reference-count
measurement that removed a gate this PR did not need; and to @def- for the
mzcompose workflow showing two replicas of one cluster correcting each other's
writes to an errored materialized view's shard indefinitely, which the sink
normalization closes.

Operator cost, measured on mz_catalog_server in an empty environment:
+1,014 dataflow operators (+6.4%) with the flag on, measured before the sink
normalization was added.

🤖 Generated with Claude Code

https://claude.ai/code/session_01GbtepuqcxmAffaM2qXZ4uA

[linear-code]

CPU-204

[ggevay]

Thank you for the quick fix! Wrote some comments.

[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

[antiguru]

@def- this is a blocker and the diagnosis holds. Thank you for the runnable repro — that's a failure mode I would not have found by reading.

Confirmed the mechanism: materialized_view.rs:262 builds desired as OkErr::new(ok_collection.inner, err_collection.inner), taking the error collection's diffs verbatim, and the sink writes desired - persist. So an error's multiplicity is durable state in the shard. Two replicas that disagree each see the other's writes as an error to correct, and converge on nothing. No input activity required, exactly as you measured.

The root cause is broader than the flag, and I'd rather fix that than the trigger. Error multiplicity in an MV's shard is a function of plan shape: it is whatever the binding fan-out happens to be, 8 in your repro. Nothing anywhere normalizes it before the write. The flag makes divergence trivially reachable, but any difference in how two replicas of a cluster render the same MV's error graph has the same consequence, and the multiplicity is meaningless in the first place — nothing reads it, which is the premise this whole PR rests on.

So the fix I'd propose is to collapse errors at the persist sink boundary, unconditionally and independent of enable_compute_error_distinct, so shard content is canonical regardless of flag state or plan shape. Then a mixed-multiplicity cluster converges instead of oscillating: a replica whose desired says 1 against a shard holding 8 writes -7 once and stops.

Two notes on alternatives I considered and rejected:

• Cluster-scoping the flag is not a substitute. Rendering reads the config once, when the dataflow is built, so a flip followed by CREATE CLUSTER REPLICA still gives you one replica per value even with a cluster-coherent parameter. It narrows the window rather than closing it.
• Only persist-backed exports are affected. Index exports are per-replica in-memory arrangements and a peek surfaces presence, not multiplicity, so they can differ harmlessly. That is what makes the sink the right and sufficient place.

I also owe a correction on my earlier reply to @antiguru: I sold per-replica enablement as a benefit of the replica scoping, "turn it on for one unbilled replica against a real erroring dataflow". On an MV that is precisely your bug, deliberately induced. That advice was wrong and I withdraw it — until the sink normalizes, the flag must not differ between replicas of one cluster, and validating on a single replica is only safe for index-backed objects.

Would you mind if I take your workflow into the PR as-is? It asserts on the mechanism rather than the symptom, and the distinct_error_operators guard against passing for the wrong reason is the part I would have gotten wrong.

---

Generated by Claude Code

[def-]

Would you mind if I take your workflow into the PR as-is?

You can always take my tests directly! Sometimes the comments are a bit inappropriate for a permanent test since it's my local reproducer, and sometimes references stuff that isn't even in the repo.

[def-]

Nothing else from QA side

[ggevay]

There are some scary errors in CI:

cursor_through: upper straddles batch

[antiguru]

@ggevay The cross-object half you found is unfixed again. I had it working via a collapse at both index exports, but that collapse crashed 19 CI jobs, so it is reverted in 528abd9. Filed as CPU-209, because the blocker turned out to be a general hazard rather than anything error-specific. Summary here so the PR carries it.

Symptom, in dataflows importing an index and unrelated to the erroring plan:

panicked at differential-dataflow/src/trace/implementations/spine_fueled.rs:180:21:
`cursor_through`: `upper` straddles batch

Four variants, each a full local sqllogictest run:

Variant	Result
reduce over the exported error arrangement	panic
same, plus errs.stream added to output_probe	panic
reduce built, output discarded, original trace exported	panic
as_collection over the same arrangement instead of reduce	385/385 pass

Row 3 is the one that decides it: merely constructing the reader panics, even when its output goes nowhere and the trace that gets exported is the original. So this is not about the collapse, the error values, or the ok/error frontier desynchronization I first assumed (row 2 rules the probe out).

Mechanism, read off the code rather than instrumented. TraceAgent's physical frontier is the minimum over outstanding handles. reduce calls source_trace.set_physical_compaction(upper_limit) at its own lagging progress (reduce.rs:243), which pins that minimum, so ArrangementManager::maintenance can no longer advance the exported trace to read_upper() (manager.rs:61). Spine drains pending into merging only while pending[0].upper() <= physical_frontier (spine_fueled.rs:404), so batches stay pending, and cursor_through applies its straddle check only to pending batches (spine_fueled.rs:180). The caller is the importer's mz_join_core, cutting at its own acknowledged frontier (mz_join_core.rs:465), which has no reason to line up with a batch the exporting side left pending. What I have not verified directly is that the reduce handle is the one holding the minimum down.

So the invariant is: an arrangement registered with ArrangementManager must not have a second trace reader in the exporting dataflow. Nothing states or checks it, and as_collection is fine because a stream reader never registers a handle. The last commit on the branch replaces the wrong explanation I had left in export_index with this one.

What that leaves: multiplicity crossing an index boundary is unbounded, materialized views are covered by the sink normalization, and the 64-level indexed-view test came out with the revert. Keeping it in proportion, object-graph path counts for the reported incidents top out at 8 against roughly 7e30 within one dataflow, so I would rather land the intra-dataflow fix and do the index side properly under CPU-209 than hold this. The safe shape looks like collapsing in stream form and re-arranging, given that row 4 passes, at the cost of an extra arrangement. Say the word if you would rather it were one PR.

[ggevay]

Thank you, LGTM, seems ok to me to postpone https://linear.app/materializeinc/issue/CPU-209/a-second-trace-reader-on-an-exported-arrangement-panics-importers-in

Edit: From Claude: You could reference CPU-209 in the export_index NOTE so future readers find the tracker, and attach the removed 64-level indexed-view test to CPU-209 so the eventual fix lands with its regression test.

[antiguru]

Both done.

export_index's NOTE now ends with TODO(CPU-209): bound it without a trace reader. (57dbe0d).

Your indexed-view file is on CPU-209, credited, with a link back to the review comment holding the verbatim version. I put it there as a generator rather than 413 pasted lines, since the levels are mechanical, and noted the one rename it needs (distinct_shared_binding_errs is now distinct_binding_errs). Left two open questions for whoever picks it up: whether it earns its own file, and whether 65 indexes belong in the fast suite.

One more thing surfaced while checking whether this PR covers neighbouring issues: it does not fix CPU-74, self-linear-joins double-counting errors. Measured on a build of this branch, lj_mv still reports 2. The plan does carry a Let, so the collapse fires and resets l0 to multiplicity 1, but Join::Linear then concatenates l0's errors once per input. Resetting at each binding stops multiplicity compounding across levels, which is what CPU-204 is about; it does not bound the fan-out of the final level, which is what CPU-74 is. Analysis and the fix recipe are on that issue. Fixable at the sink alone by normalizing before log_dataflow_errors, independent of CPU-209, as a follow-up.

Waiting on CI for the two comment-only commits, then this is ready from my side.

[antiguru]

Thanks for the thorough reviews! Much appreciated

[def-]

@antiguru Can you check if this makes sense? I'll be out in a few minutes, but this looks serious enough to potentially flag:

1. MEDIUM -- Error collapse rewrites negative accumulations to +1, erasing the negative-multiplicity corruption signal

src/compute/src/render/context.rs:456

The collapse reduce emits ((), Diff::ONE) for every key whose accumulated input is non-empty, including keys that accumulate to a negative diff. That is a sign flip, not a multiplicity collapse: an error collection holding E: -N becomes E: +1. The peek path treats those two states as different conditions, so an index whose error trace goes negative stops reporting Invalid data in source errors, saw retractions (...) and stops emitting the error! event that surfaces this corruption class.

Details

DD's reduce_abelian calls the user logic whenever the consolidated per-key input is non-empty (if !input.is_empty()), and input is the accumulation, so a key at -N yields [((), -N)] and the closure unconditionally pushes +1. A key at exactly 0 consolidates away and correctly produces nothing.

That negative error accumulations are a real, handled state is established in-tree, not hypothetically:

• src/compute/src/compute_state.rs:1682-1697 — the index peek scans the error trace and branches on sign: copies.is_negative() logs error!("index peek encountered negative multiplicities in error trace") and returns a distinct Invalid data in source errors, saw retractions ({}) for row that does not exist message; only copies.is_positive() returns the error itself.
• src/compute/src/logging/compute.rs:620-624 — error counts are packed as a Datum::Int64 rather than a DD diff precisely because "the total per-worker error count might be negative".

Both new call sites reach that trace:

• src/compute/src/render.rs:1086 collapses at every Let/rec binding definition, so a negative accumulation arising in or upstream of any shared binding is +1 by the time it reaches the export. Default-on.
• src/compute/src/render/sinks.rs:140 collapses unconditionally at the MV sink, so the durable shard records +1 and every index or dataflow importing that shard loses the signal, regardless of the flag.

The doc comment justifies the transform with "query semantics depend only on whether an error is present". That holds for positive multiplicities; it does not hold across the sign boundary the peek path keys off. The pre-existing Distinct recursive err in render.rs has the same shape, but it was confined to LetRec error variables; this diff extends it to every binding and to durable MV content.

Secondary consequence: because the collapse is not sign-preserving, a -N contributed by a shared binding can no longer cancel a +N contributed by another branch of the same plan, so a dataflow whose error collection previously summed to zero can now surface an error.

Fix. Make the reduce clamp magnitude while preserving sign, which is equally bounded and equally correct under retraction (the reduce reads the accumulation either way):

|_err, input: &[(_, Diff)], output| {
    let accum: Diff = input.iter().map(|(_v, d)| *d).sum();
    output.push(((), if accum.is_positive() { Diff::ONE } else { -Diff::ONE }));
}

[antiguru]

@ggevay The delta-join test I owed you is tracked as CPU-212, so it does not disappear now that this is merged. It records that the no-gate decision rests on a test that does not exist yet, and carries what is needed to write it: the three read shapes (raw source, local arrangement, imported trace), the two error origins (errors in the collection versus errors from forming the key, which live on different streams), your type=delta query as the shape, and the trap that the plan must be pinned or it can silently degrade to a linear join and cover none of it. Your delta_join.rs thread on the one-copy-per-retained-form bound is linked there too, since the same query exercises it.