diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py
index fadf6a22fa8b0..034bdc2bb5a11 100644
--- a/misc/python/materialize/mzcompose/__init__.py
+++ b/misc/python/materialize/mzcompose/__init__.py
@@ -116,6 +116,7 @@ def get_minimal_system_parameters(
         "enable_s3_tables_region_check": "false",
         "enable_statement_lifecycle_logging": "true",
         "enable_storage_introspection_logs": "true",
+        "enable_compute_error_distinct": "true",
         "enable_compute_temporal_bucketing": "true",
         "enable_variadic_left_join_lowering": "true",
         "enable_worker_core_affinity": "true",
diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py
index d4c19d23076f0..c121ec6a3fb79 100644
--- a/misc/python/materialize/parallel_workload/action.py
+++ b/misc/python/materialize/parallel_workload/action.py
@@ -2986,6 +2986,7 @@ def __init__(
         self.flags_with_values["enable_compute_temporal_bucketing"] = (
             BOOLEAN_FLAG_VALUES
         )
+        self.flags_with_values["enable_compute_error_distinct"] = BOOLEAN_FLAG_VALUES
         self.flags_with_values["enable_alter_table_add_column"] = BOOLEAN_FLAG_VALUES
         self.flags_with_values["enable_bounded_staleness_isolation"] = (
             BOOLEAN_FLAG_VALUES
diff --git a/src/compute-types/src/dyncfgs.rs b/src/compute-types/src/dyncfgs.rs
index 19ca6b4cd02fb..bfc0802b28541 100644
--- a/src/compute-types/src/dyncfgs.rs
+++ b/src/compute-types/src/dyncfgs.rs
@@ -23,6 +23,25 @@ pub const ENABLE_HALF_JOIN2: Config<bool> = Config::new(
     "Whether compute should use `half_join2` rather than DD's `half_join` to render delta joins.",
 );
 
+/// Whether rendering should collapse error multiplicities to one where it arranges errors.
+///
+/// Error semantics depend only on whether an error is present, so its multiplicity carries no
+/// information a consumer reads. Left uncollapsed, a shared collection contributes its errors once
+/// per plan path that reads it, and because those factors apply again at each level of sharing they
+/// compound multiplicatively until the `Diff` overflows.
+///
+/// Governs sharing within a dataflow only. Sharing across objects is bounded unconditionally, by
+/// normalizing at every boundary another dataflow can read, so what this flag decides is never
+/// durable state and two replicas rendering under different values still write the same thing.
+/// Environment-scoped for now because nothing needs finer granularity, not because finer would be
+/// unsafe.
+pub const ENABLE_ERROR_DISTINCT: Config<bool> = Config::new(
+    "enable_compute_error_distinct",
+    true,
+    "Whether compute rendering should collapse error multiplicities to one where it arranges \
+     errors.",
+);
+
 /// Use the column-paged merge batcher code path at arrange sites. When
 /// `true`, arrange operators use `Col2ValPagedBatcher` (in
 /// `mz_timely_util::columnar`) and `RowRowColPagedBuilder` (in
@@ -548,6 +567,7 @@ pub const MV_SINK_ADVANCE_PERSIST_FRONTIERS: Config<bool> = Config::new(
 pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
     configs
         .add(&ENABLE_HALF_JOIN2)
+        .add(&ENABLE_ERROR_DISTINCT)
         .add(&ENABLE_MZ_JOIN_CORE)
         .add(&ENABLE_SYNC_MV_SINK)
         .add(&ENABLE_CORRECTION_V2)
diff --git a/src/compute/src/render.rs b/src/compute/src/render.rs
index 82fc821cd8604..ccfe93b902c97 100644
--- a/src/compute/src/render.rs
+++ b/src/compute/src/render.rs
@@ -125,7 +125,8 @@ use mz_compute_types::dataflows::{DataflowDescription, IndexDesc};
 use mz_compute_types::dyncfgs::{
     COMPUTE_APPLY_COLUMN_DEMANDS, COMPUTE_LOGICAL_BACKPRESSURE_INFLIGHT_SLACK,
     COMPUTE_LOGICAL_BACKPRESSURE_MAX_RETAINED_CAPABILITIES, ENABLE_COMPUTE_LOGICAL_BACKPRESSURE,
-    ENABLE_COMPUTE_TEMPORAL_BUCKETING, SUBSCRIBE_SNAPSHOT_OPTIMIZATION, TEMPORAL_BUCKETING_SUMMARY,
+    ENABLE_COMPUTE_TEMPORAL_BUCKETING, ENABLE_ERROR_DISTINCT, SUBSCRIBE_SNAPSHOT_OPTIMIZATION,
+    TEMPORAL_BUCKETING_SUMMARY,
 };
 use mz_compute_types::plan::render_plan::{
     self, BindStage, LetBind, LetFreePlan, RecBind, RenderPlan,
@@ -711,6 +712,17 @@ impl<'g> Context<'g, mz_repr::Timestamp> {
         let key = &idx.key;
         match bundle.arrangement(key) {
             Some(ArrangementFlavor::Local(mut oks, mut errs)) => {
+                // NOTE: Do not give an exported arrangement a second reader that holds a trace
+                // handle, such as a `reduce`. Such a reader pins the shared spine's physical
+                // frontier at its own lagging progress, and `ArrangementManager::maintenance` can
+                // then no longer advance it, so batches pile up in `Spine::pending`. A cursor is
+                // only checked for straddling over pending batches, so an importing dataflow's
+                // `cursor_through` eventually panics with `upper` straddles batch. Watching
+                // `errs.stream` in `output_probe` does not help, and neither does discarding the
+                // reader's output. Stream-level readers like `as_collection` are unaffected. This is
+                // why error multiplicity is not collapsed here, leaving multiplicity that crosses
+                // an index boundary unbounded. TODO(CPU-209): bound it without a trace reader.
+
                 // Ensure that the frontier does not advance past the expiration time, if set.
                 // Otherwise, we might write down incorrect data.
                 if let Some(&expiration) = self.dataflow_expiration.as_option() {
@@ -906,6 +918,7 @@ impl<'scope> Context<'scope, Product<mz_repr::Timestamp, PointStamp<u64>>> {
                                 .render_letfree_plan(object_id, value, binding)
                                 .leave_region(self.scope)
                         });
+                let bundle = self.distinct_binding_errs(bundle);
                 self.insert_id(Id::Local(id), bundle);
             }
 
@@ -939,6 +952,11 @@ impl<'scope> Context<'scope, Product<mz_repr::Timestamp, PointStamp<u64>>> {
                 // here to cause that to happen.
                 let (oks, mut err) = bundle.collection.clone().unwrap();
                 let oks = oks.into_vec();
+                // Collapses what forward reads see. `err_v` below feeds reads rendered before this
+                // binding and is collapsed separately; without this, a `Get` in a later rec binding
+                // or in the body resolves to the bundle stored here and compounds level over level,
+                // which is exactly what the collapse prevents for non-recursive bindings.
+                let bundle = self.distinct_binding_errs(bundle);
                 self.insert_id(Id::Local(id), bundle);
                 let (oks_v, err_v) = variables.remove(&Id::Local(id)).unwrap();
 
@@ -1044,6 +1062,7 @@ impl<'scope, T: RenderTimestamp + MaybeBucketByTime> Context<'scope, T> {
                                 .render_letfree_plan(object_id, value, binding)
                                 .leave_region(self.scope)
                         });
+                let bundle = self.distinct_binding_errs(bundle);
                 self.insert_id(Id::Local(id), bundle);
             }
         }
@@ -1056,6 +1075,25 @@ impl<'scope, T: RenderTimestamp + MaybeBucketByTime> Context<'scope, T> {
         })
     }
 
+    /// Collapses a binding's error multiplicities.
+    ///
+    /// Applied to every binding, not only the multiply-read ones. Gating on the reference count
+    /// would be a pure optimization, since collapsing a binding one `Get` reads is harmless, and
+    /// there is almost nothing to gate: `NormalizeLets` inlines single-use bindings, so the ones
+    /// reaching rendering are shared. See [`CollectionBundle::distinct_errs`] for why the collapse
+    /// is needed at all, and why a binding's definition is the place for it rather than the
+    /// multi-input operators where the duplicate copies happen to meet again.
+    fn distinct_binding_errs(
+        &self,
+        bundle: CollectionBundle<'scope, T>,
+    ) -> CollectionBundle<'scope, T> {
+        if ENABLE_ERROR_DISTINCT.get(&self.config_set) {
+            bundle.distinct_errs()
+        } else {
+            bundle
+        }
+    }
+
     /// Renders a let-free plan to a differential dataflow, producing the collection of results.
     fn render_letfree_plan(
         &self,
diff --git a/src/compute/src/render/context.rs b/src/compute/src/render/context.rs
index 6e6b80119b601..c5f20a8691947 100644
--- a/src/compute/src/render/context.rs
+++ b/src/compute/src/render/context.rs
@@ -48,6 +48,7 @@ use timely::progress::{Antichain, Timestamp};
 
 use crate::compute_state::ComputeState;
 use crate::extensions::arrange::{KeyCollection, MzArrange, MzArrangeCore};
+use crate::extensions::reduce::MzReduce;
 use crate::render::columnar::CollectionEdge;
 use crate::render::errors::{DataflowErrorSer, ErrorLogger};
 use crate::render::{LinearJoinSpec, MaybeBucketByTime, RenderTimestamp};
@@ -435,6 +436,41 @@ impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> {
         }
     }
 }
+/// Rewrites an arranged error collection to hold each of its errors once.
+///
+/// Sound because query semantics depend only on whether an error is present, and correct under
+/// retraction only because it reads the accumulated collection: no pointwise function of the input
+/// diffs (a saturating add, a sign) can collapse multiplicity and still cancel when the errors
+/// retract.
+///
+/// NOTE: One consumer does read the multiplicity. Error-count introspection reports it as the
+/// number of failing rows, so `log_dataflow_errors` must see the collection before this collapses
+/// it, or a dataflow reports one error however many rows failed. Collapse after the logging, never
+/// before.
+pub(crate) fn distinct_arranged_errs<'a, T: RenderTimestamp>(
+    errs: Arranged<'a, ErrAgent<T, Diff>>,
+    name: &str,
+) -> Arranged<'a, ErrAgent<T, Diff>> {
+    errs.mz_reduce_abelian::<_, ErrBuilder<_, _>, ErrSpine<_, _>, _>(
+        name,
+        |_err, _input, output| output.push(((), Diff::ONE)),
+    )
+}
+
+/// Rewrites an error collection to hold each of its errors once.
+///
+/// Costs an arrangement more than [`distinct_arranged_errs`], which reuses the arrangement its
+/// input already has.
+pub(crate) fn distinct_errs_collection<'a, T: RenderTimestamp>(
+    errs: VecCollection<'a, T, DataflowErrorSer, Diff>,
+) -> VecCollection<'a, T, DataflowErrorSer, Diff> {
+    let errs: KeyCollection<_, _, _> = errs.into();
+    let errs = errs
+        .mz_arrange::<ColumnationChunker<_>, ErrBatcher<_, _>, ErrBuilder<_, _>, ErrSpine<_, _>>(
+            "Arrange errors",
+        );
+    distinct_arranged_errs(errs, "Distinct errors").as_collection(|err, _| err.clone())
+}
 
 /// A bundle of the various ways a collection can be represented.
 ///
@@ -507,6 +543,50 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> {
         }
     }
 
+    /// Collapses the multiplicity of every error this bundle carries to one.
+    ///
+    /// Belongs at the definition of a binding that more than one `Get` reads. Each reader
+    /// propagates the binding's errors independently, so a binding read `f` times contributes its
+    /// errors `f` times to the dataflow's error output, and those factors apply again at every
+    /// further level of sharing: a chain of diamonds multiplies rather than adds, and reaches
+    /// `Diff` overflow at a depth plans really do have. Collapsing at each definition holds the
+    /// dataflow's error multiplicity to the fan-out of a single level.
+    ///
+    /// Sound because error semantics depend only on whether an error is present, and correct under
+    /// retraction only because it reads the accumulated collection: no pointwise function of the
+    /// input diffs (a saturating add, a sign) can collapse multiplicity and still cancel when the
+    /// errors retract.
+    ///
+    /// Collapses every form the bundle offers, not just one. Each form carries its own error
+    /// stream, and those streams differ in content as well as identity: 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 within one operator, so a
+    /// binding's definition cannot know which form to collapse.
+    ///
+    /// NOTE: Leaves imported arrangements (`ArrangementFlavor::Trace`) alone, whose error traces
+    /// this dataflow cannot rewrite in place. Their errors arrive bounded by the exporting
+    /// dataflow's last level of sharing rather than collapsed to one, since nothing collapses at an
+    /// export. A global read more than once within one dataflow is not collapsed either, because
+    /// only local bindings reach this.
+    pub fn distinct_errs(mut self) -> Self {
+        if let Some((oks, errs)) = self.collection.take() {
+            self.collection = Some((oks, distinct_errs_collection(errs)));
+        }
+        for (key, flavor) in std::mem::take(&mut self.arranged) {
+            let flavor = match flavor {
+                ArrangementFlavor::Local(oks, errs) => {
+                    // Names the key, not the binding: an operator name carrying a `LocalId` would
+                    // churn the introspection goldens every time the optimizer renumbers locals.
+                    let name = format!("Distinct errors[{key:?}]");
+                    ArrangementFlavor::Local(oks, distinct_arranged_errs(errs, &name))
+                }
+                flavor @ ArrangementFlavor::Trace(..) => flavor,
+            };
+            self.arranged.insert(key, flavor);
+        }
+        self
+    }
+
     /// Brings the collection bundle into a region.
     pub fn enter_region<'inner>(&self, region: Scope<'inner, T>) -> CollectionBundle<'inner, T> {
         CollectionBundle {
diff --git a/src/compute/src/render/join/delta_join.rs b/src/compute/src/render/join/delta_join.rs
index a16a60764c70d..cfac33a55887a 100644
--- a/src/compute/src/render/join/delta_join.rs
+++ b/src/compute/src/render/join/delta_join.rs
@@ -55,9 +55,6 @@ impl<'scope, T: RenderTimestamp> Context<'scope, T> {
     ) -> CollectionBundle<'scope, T> {
         // We create a new region to contain the dataflow paths for the delta join.
         let (oks, errs) = self.scope.clone().region_named("Join(Delta)", |inner| {
-            // Collects error streams for the ambient scope.
-            let mut inner_errs = Vec::new();
-
             // Our plan is to iterate through each input relation, and attempt
             // to find a plan that maximally uses existing keys (better: uses
             // existing arrangements, to which we have access).
@@ -86,6 +83,11 @@ impl<'scope, T: RenderTimestamp> Context<'scope, T> {
                 })
                 .collect::<Vec<_>>();
 
+            // Collects error streams for the ambient scope, seeded with the errors the inputs
+            // arrive with. Every path reads every input, but an input's pre-existing errors belong
+            // in the output once, not once per path. See [`bundle_errs`].
+            let mut inner_errs: Vec<_> = inputs.iter().flat_map(bundle_errs).collect();
+
             for path_plan in join_plan.path_plans {
                 // Deconstruct the stages of the path plan.
                 let DeltaPathPlan {
@@ -288,12 +290,51 @@ fn prune_bundle<'scope, T: RenderTimestamp>(
     }
 }
 
+/// The error collections of every collection and arrangement `bundle` holds.
+///
+/// A delta join reads every input from every one of its paths, but an input's pre-existing errors
+/// belong in the join's error output once. Propagating them per path would multiply their
+/// multiplicities by the number of paths, and because a join's output is another join's input, those
+/// factors compound multiplicatively through a nested plan until the `Diff` overflows. Error
+/// semantics depend only on presence, so the extra copies buy nothing.
+///
+/// Expects the pruned bundle (see [`prune_bundle`]), so that it yields errors only for the
+/// collections some path actually reads.
+///
+/// NOTE: This bounds an input's errors to one copy per retained form, not to one copy outright.
+/// Each form carries its own error collection, and an arrangement's is a distinct stream built from
+/// the raw one plus that key's key-formation errors, so an input the join reads under two lookup
+/// keys still contributes its errors twice. Reachable whenever a delta path set needs an
+/// error-carrying input arranged by more than one key.
+fn bundle_errs<'scope, T: RenderTimestamp>(
+    bundle: &CollectionBundle<'scope, T>,
+) -> Vec<VecCollection<'scope, T, DataflowErrorSer, Diff>> {
+    let mut collected = Vec::with_capacity(bundle.arranged.len() + 1);
+    if let Some((_oks, errs)) = &bundle.collection {
+        collected.push(errs.clone());
+    }
+    for flavor in bundle.arranged.values() {
+        let errs = match flavor {
+            ArrangementFlavor::Local(_oks, errs) => errs.clone().as_collection(|k, _v| k.clone()),
+            ArrangementFlavor::Trace(_id, _oks, errs) => {
+                errs.clone().as_collection(|k, _v| k.clone())
+            }
+        };
+        collected.push(errs);
+    }
+    collected
+}
+
 /// Constructs a `half_join` against the arrangement held by a collection bundle.
 ///
 /// This wrapper demuxes over the two flavors of arrangement (dataflow-local or imported trace)
 /// that the bundle might hold for `lookup_key`, dispatching to the generic [`build_halfjoin_trace`]
 /// for each. `source_precedes_lookup` selects the tie-breaking comparison: `le` if the source
 /// relation precedes the lookup relation in the total order on relations, otherwise `lt`.
+///
+/// The returned error collection holds only the errors this stage produces. The errors `bundle`
+/// already carries are the caller's to propagate, once, rather than once per delta path that looks
+/// the input up. See [`bundle_errs`].
 fn build_halfjoin<'scope, T>(
     updates: VecCollection<'scope, T, (Row, T), Diff>,
     prev_key: Vec<LirScalarExpr>,
@@ -311,7 +352,7 @@ where
     T: RenderTimestamp,
 {
     match bundle.arrangement(&lookup_key) {
-        Some(ArrangementFlavor::Local(oks, errs)) => {
+        Some(ArrangementFlavor::Local(oks, _errs)) => {
             let (oks, errs2) = if source_precedes_lookup {
                 build_halfjoin_trace::<_, RowRowAgent<_, _>, _>(
                     updates,
@@ -333,9 +374,9 @@ where
                     config_set,
                 )
             };
-            (oks, errs2.concat(errs.as_collection(|k, _v| k.clone())))
+            (oks, errs2)
         }
-        Some(ArrangementFlavor::Trace(_, oks, errs)) => {
+        Some(ArrangementFlavor::Trace(_, oks, _errs)) => {
             let (oks, errs2) = if source_precedes_lookup {
                 build_halfjoin_trace::<_, RowRowEnter<_, _, _>, _>(
                     updates,
@@ -357,7 +398,7 @@ where
                     config_set,
                 )
             };
-            (oks, errs2.concat(errs.as_collection(|k, _v| k.clone())))
+            (oks, errs2)
         }
         None => panic!("Arrangement promised by the planner is absent!"),
     }
@@ -658,6 +699,10 @@ where
 /// With a `source_key`, demuxes over the two flavors of arrangement the bundle might hold for that
 /// key, dispatching to the generic [`build_update_stream_trace`]. Without a `source_key`, the source
 /// relation is consumed as a raw (unarranged) collection via [`build_update_stream_stream`].
+///
+/// The returned error collection holds only the errors the initial closure produces. The errors
+/// `bundle` already carries are the caller's to propagate, once, rather than once per delta path.
+/// See [`bundle_errs`].
 fn build_update_stream<'scope, T>(
     bundle: &CollectionBundle<'scope, T>,
     as_of: Antichain<mz_repr::Timestamp>,
@@ -675,32 +720,28 @@ where
         // No source key means a single-time dataflow (e.g. a `SELECT`) whose plan was truncated to
         // this one path, letting us hydrate the source from its raw collection instead of an
         // arrangement.
-        let (oks, errs) = bundle
+        let (oks, _errs) = bundle
             .collection
             .clone()
             .expect("The unarranged collection doesn't exist.");
-        let (oks, errs2) =
-            build_update_stream_stream(oks.into_vec(), as_of, source_relation, initial_closure);
-        return (oks, errs2.concat(errs));
+        return build_update_stream_stream(oks.into_vec(), as_of, source_relation, initial_closure);
     };
     match bundle.arrangement(&source_key) {
-        Some(ArrangementFlavor::Local(oks, errs)) => {
-            let (oks, errs2) = build_update_stream_trace::<_, RowRowAgent<_, _>>(
+        Some(ArrangementFlavor::Local(oks, _errs)) => {
+            build_update_stream_trace::<_, RowRowAgent<_, _>>(
                 oks,
                 as_of,
                 source_relation,
                 initial_closure,
-            );
-            (oks, errs2.concat(errs.as_collection(|k, _v| k.clone())))
+            )
         }
-        Some(ArrangementFlavor::Trace(_, oks, errs)) => {
-            let (oks, errs2) = build_update_stream_trace::<_, RowRowEnter<_, _, _>>(
+        Some(ArrangementFlavor::Trace(_, oks, _errs)) => {
+            build_update_stream_trace::<_, RowRowEnter<_, _, _>>(
                 oks,
                 as_of,
                 source_relation,
                 initial_closure,
-            );
-            (oks, errs2.concat(errs.as_collection(|k, _v| k.clone())))
+            )
         }
         None => panic!("Arrangement promised by the planner is absent!"),
     }
diff --git a/src/compute/src/render/sinks.rs b/src/compute/src/render/sinks.rs
index 4dbdf88a4eec9..c96e0be462b7b 100644
--- a/src/compute/src/render/sinks.rs
+++ b/src/compute/src/render/sinks.rs
@@ -30,7 +30,7 @@ use timely::progress::Antichain;
 
 use crate::compute_state::SinkToken;
 use crate::logging::compute::LogDataflowErrors;
-use crate::render::context::Context;
+use crate::render::context::{Context, distinct_errs_collection};
 use crate::render::errors::DataflowErrorSer;
 use crate::render::{RenderTimestamp, StartSignal};
 
@@ -130,6 +130,16 @@ impl<'g, T: RenderTimestamp> Context<'g, T> {
             err_collection = err_collection.concat(null_errs);
         }
 
+        // Normalize what the sink persists. A materialized view's error multiplicity is durable
+        // state another dataflow reads back verbatim, so leaving it at this dataflow's fan-out both
+        // multiplies across the object graph and makes the written value depend on plan shape.
+        // Placed after the null assertions so their errors, whose multiplicity follows the ok row's,
+        // are covered too. Only the persist-backed sink is re-importable: a subscribe or a one-shot
+        // copy is read once by a client, so neither pays for the arrangement.
+        if matches!(sink.connection, ComputeSinkConnection::MaterializedView(_)) {
+            err_collection = distinct_errs_collection(err_collection);
+        }
+
         let region_name = match sink.connection {
             ComputeSinkConnection::Subscribe(_) => format!("SubscribeSink({:?})", sink_id),
             ComputeSinkConnection::MaterializedView(_) => {
diff --git a/test/launchdarkly-flag-consistency/mzcompose.py b/test/launchdarkly-flag-consistency/mzcompose.py
index 074467a98fc16..bd8d9cb6baac4 100644
--- a/test/launchdarkly-flag-consistency/mzcompose.py
+++ b/test/launchdarkly-flag-consistency/mzcompose.py
@@ -526,6 +526,7 @@
     "compute_subscribe_snapshot_optimization",
     "enable_cast_elimination",
     "enable_compute_correction_v2",
+    "enable_compute_error_distinct",
     "enable_compute_temporal_bucketing",
     "enable_new_outer_join_lowering",
     "enable_upsert_v2",
diff --git a/test/sqllogictest/error_semantics.slt b/test/sqllogictest/error_semantics.slt
index 967ca375be35c..3462e888974f3 100644
--- a/test/sqllogictest/error_semantics.slt
+++ b/test/sqllogictest/error_semantics.slt
@@ -93,3 +93,95 @@ select coalesce(a, (select a/b from test)) from test;
 
 query error Evaluation error: division by zero
 select *, case when a = 5 then (select a/b from test) else 7 end from test;
+
+# A chain of diamond-shaped CTEs over an erroring expression. Each level reads the level
+# below it twice, so an undeduplicated error collection doubles its multiplicity per level
+# and overflows `Diff` well before the chain ends. The multiplicity of an error is not
+# observable, so the only thing to assert is that the error still comes back, and does so
+# without an overflow (which CI catches as a panic).
+#
+# `union` rather than `union all` is what makes this reach the overflow: the distinct arranges
+# the error collection, so duplicates consolidate into one record carrying a doubling diff.
+# Under `union all` nothing consolidates between levels, the records duplicate physically
+# instead, and the query exhausts memory without ever overflowing.
+
+statement ok
+create table diamond (a int, b int)
+
+statement ok
+insert into diamond values (1, 0)
+
+query error division by zero
+with
+  c0 as (select a, a / b as q from diamond),
+  c1 as (select a, q from c0 where a = 1 union select q, a from c0 where a = 2),
+  c2 as (select a, q from c1 where a = 1 union select q, a from c1 where a = 2),
+  c3 as (select a, q from c2 where a = 1 union select q, a from c2 where a = 2),
+  c4 as (select a, q from c3 where a = 1 union select q, a from c3 where a = 2),
+  c5 as (select a, q from c4 where a = 1 union select q, a from c4 where a = 2),
+  c6 as (select a, q from c5 where a = 1 union select q, a from c5 where a = 2),
+  c7 as (select a, q from c6 where a = 1 union select q, a from c6 where a = 2),
+  c8 as (select a, q from c7 where a = 1 union select q, a from c7 where a = 2),
+  c9 as (select a, q from c8 where a = 1 union select q, a from c8 where a = 2),
+  c10 as (select a, q from c9 where a = 1 union select q, a from c9 where a = 2),
+  c11 as (select a, q from c10 where a = 1 union select q, a from c10 where a = 2),
+  c12 as (select a, q from c11 where a = 1 union select q, a from c11 where a = 2),
+  c13 as (select a, q from c12 where a = 1 union select q, a from c12 where a = 2),
+  c14 as (select a, q from c13 where a = 1 union select q, a from c13 where a = 2),
+  c15 as (select a, q from c14 where a = 1 union select q, a from c14 where a = 2),
+  c16 as (select a, q from c15 where a = 1 union select q, a from c15 where a = 2),
+  c17 as (select a, q from c16 where a = 1 union select q, a from c16 where a = 2),
+  c18 as (select a, q from c17 where a = 1 union select q, a from c17 where a = 2),
+  c19 as (select a, q from c18 where a = 1 union select q, a from c18 where a = 2),
+  c20 as (select a, q from c19 where a = 1 union select q, a from c19 where a = 2),
+  c21 as (select a, q from c20 where a = 1 union select q, a from c20 where a = 2),
+  c22 as (select a, q from c21 where a = 1 union select q, a from c21 where a = 2),
+  c23 as (select a, q from c22 where a = 1 union select q, a from c22 where a = 2),
+  c24 as (select a, q from c23 where a = 1 union select q, a from c23 where a = 2),
+  c25 as (select a, q from c24 where a = 1 union select q, a from c24 where a = 2),
+  c26 as (select a, q from c25 where a = 1 union select q, a from c25 where a = 2),
+  c27 as (select a, q from c26 where a = 1 union select q, a from c26 where a = 2),
+  c28 as (select a, q from c27 where a = 1 union select q, a from c27 where a = 2),
+  c29 as (select a, q from c28 where a = 1 union select q, a from c28 where a = 2),
+  c30 as (select a, q from c29 where a = 1 union select q, a from c29 where a = 2),
+  c31 as (select a, q from c30 where a = 1 union select q, a from c30 where a = 2),
+  c32 as (select a, q from c31 where a = 1 union select q, a from c31 where a = 2),
+  c33 as (select a, q from c32 where a = 1 union select q, a from c32 where a = 2),
+  c34 as (select a, q from c33 where a = 1 union select q, a from c33 where a = 2),
+  c35 as (select a, q from c34 where a = 1 union select q, a from c34 where a = 2),
+  c36 as (select a, q from c35 where a = 1 union select q, a from c35 where a = 2),
+  c37 as (select a, q from c36 where a = 1 union select q, a from c36 where a = 2),
+  c38 as (select a, q from c37 where a = 1 union select q, a from c37 where a = 2),
+  c39 as (select a, q from c38 where a = 1 union select q, a from c38 where a = 2),
+  c40 as (select a, q from c39 where a = 1 union select q, a from c39 where a = 2),
+  c41 as (select a, q from c40 where a = 1 union select q, a from c40 where a = 2),
+  c42 as (select a, q from c41 where a = 1 union select q, a from c41 where a = 2),
+  c43 as (select a, q from c42 where a = 1 union select q, a from c42 where a = 2),
+  c44 as (select a, q from c43 where a = 1 union select q, a from c43 where a = 2),
+  c45 as (select a, q from c44 where a = 1 union select q, a from c44 where a = 2),
+  c46 as (select a, q from c45 where a = 1 union select q, a from c45 where a = 2),
+  c47 as (select a, q from c46 where a = 1 union select q, a from c46 where a = 2),
+  c48 as (select a, q from c47 where a = 1 union select q, a from c47 where a = 2),
+  c49 as (select a, q from c48 where a = 1 union select q, a from c48 where a = 2),
+  c50 as (select a, q from c49 where a = 1 union select q, a from c49 where a = 2),
+  c51 as (select a, q from c50 where a = 1 union select q, a from c50 where a = 2),
+  c52 as (select a, q from c51 where a = 1 union select q, a from c51 where a = 2),
+  c53 as (select a, q from c52 where a = 1 union select q, a from c52 where a = 2),
+  c54 as (select a, q from c53 where a = 1 union select q, a from c53 where a = 2),
+  c55 as (select a, q from c54 where a = 1 union select q, a from c54 where a = 2),
+  c56 as (select a, q from c55 where a = 1 union select q, a from c55 where a = 2),
+  c57 as (select a, q from c56 where a = 1 union select q, a from c56 where a = 2),
+  c58 as (select a, q from c57 where a = 1 union select q, a from c57 where a = 2),
+  c59 as (select a, q from c58 where a = 1 union select q, a from c58 where a = 2),
+  c60 as (select a, q from c59 where a = 1 union select q, a from c59 where a = 2),
+  c61 as (select a, q from c60 where a = 1 union select q, a from c60 where a = 2),
+  c62 as (select a, q from c61 where a = 1 union select q, a from c61 where a = 2),
+  c63 as (select a, q from c62 where a = 1 union select q, a from c62 where a = 2),
+  c64 as (select a, q from c63 where a = 1 union select q, a from c63 where a = 2),
+  c65 as (select a, q from c64 where a = 1 union select q, a from c64 where a = 2),
+  c66 as (select a, q from c65 where a = 1 union select q, a from c65 where a = 2),
+  c67 as (select a, q from c66 where a = 1 union select q, a from c66 where a = 2),
+  c68 as (select a, q from c67 where a = 1 union select q, a from c67 where a = 2),
+  c69 as (select a, q from c68 where a = 1 union select q, a from c68 where a = 2),
+  c70 as (select a, q from c69 where a = 1 union select q, a from c69 where a = 2)
+select * from c70
diff --git a/test/testdrive-old-kafka-src-syntax/top-1-monotonic.td b/test/testdrive-old-kafka-src-syntax/top-1-monotonic.td
index 203471e5eada5..c05dd0cc6c4c8 100644
--- a/test/testdrive-old-kafka-src-syntax/top-1-monotonic.td
+++ b/test/testdrive-old-kafka-src-syntax/top-1-monotonic.td
@@ -314,3 +314,35 @@ $ kafka-ingest format=avro topic=top1 schema=${schema} timestamp=5
 "Dataflow: order_by_desc_limit" MonotonicTop1
 "Dataflow: order_by_limit" "Arranged MonotonicTop1 partial [val: empty]"
 "Dataflow: order_by_limit" MonotonicTop1
+"Dataflow: group_by_in_top_1" "Arrange bundle err"
+"Dataflow: group_by_in_top_1" "Arrange bundle err"
+"Dataflow: group_by_in_top_1" "Arrange errors"
+"Dataflow: group_by_in_top_1" "Distinct errors"
+"Dataflow: group_by_in_top_1" "Distinct errors[[Column(0)]]"
+"Dataflow: group_by_in_top_1" "Distinct errors[[Column(0)]]"
+"Dataflow: group_by_limit" "Arrange bundle err"
+"Dataflow: group_by_limit" "Distinct errors[[]]"
+"Dataflow: group_by_order_by_in_top_1" "Arrange bundle err"
+"Dataflow: group_by_order_by_in_top_1" "Arrange bundle err"
+"Dataflow: group_by_order_by_in_top_1" "Arrange errors"
+"Dataflow: group_by_order_by_in_top_1" "Distinct errors"
+"Dataflow: group_by_order_by_in_top_1" "Distinct errors[[Column(0)]]"
+"Dataflow: group_by_order_by_in_top_1" "Distinct errors[[Column(0)]]"
+"Dataflow: limit_only" "Arrange bundle err"
+"Dataflow: limit_only" "Distinct errors[[]]"
+"Dataflow: order_by_desc_limit" "Arrange bundle err"
+"Dataflow: order_by_desc_limit" "Distinct errors[[]]"
+"Dataflow: order_by_limit" "Arrange bundle err"
+"Dataflow: order_by_limit" "Distinct errors[[]]"
+"Dataflow: group_by_in_top_1" "Arrange errors"
+"Dataflow: group_by_in_top_1" "Distinct errors"
+"Dataflow: group_by_limit" "Arrange errors"
+"Dataflow: group_by_limit" "Distinct errors"
+"Dataflow: group_by_order_by_in_top_1" "Arrange errors"
+"Dataflow: group_by_order_by_in_top_1" "Distinct errors"
+"Dataflow: limit_only" "Arrange errors"
+"Dataflow: limit_only" "Distinct errors"
+"Dataflow: order_by_desc_limit" "Arrange errors"
+"Dataflow: order_by_desc_limit" "Distinct errors"
+"Dataflow: order_by_limit" "Arrange errors"
+"Dataflow: order_by_limit" "Distinct errors"
diff --git a/test/testdrive/introspection-sources.td b/test/testdrive/introspection-sources.td
index 1300e662974f6..d836ae00ba8d4 100644
--- a/test/testdrive/introspection-sources.td
+++ b/test/testdrive/introspection-sources.td
@@ -489,7 +489,7 @@ mv_div_by_zero   1
   JOIN mz_objects o ON (c.export_id = o.id)
   ORDER BY name
 idx1_div_by_zero 3
-idx2_div_by_zero 3
+idx2_div_by_zero 1
 mv_div_by_zero   3
 
 > DELETE FROM zeros
@@ -508,7 +508,7 @@ mv_div_by_zero   3
   JOIN mz_objects o ON (c.export_id = o.id)
   ORDER BY name
 idx1_div_by_zero 2
-idx2_div_by_zero 2
+idx2_div_by_zero 1
 mv_div_by_zero   2
 
 > DROP INDEX idx1_div_by_zero
@@ -517,7 +517,7 @@ mv_div_by_zero   2
   FROM mz_introspection.mz_compute_error_counts c
   JOIN mz_objects o ON (c.export_id = o.id)
   ORDER BY name
-idx2_div_by_zero 2
+idx2_div_by_zero 1
 mv_div_by_zero   2
 
 > DROP MATERIALIZED VIEW mv_div_by_zero
diff --git a/test/testdrive/kafka-avro-upsert-sinks.td b/test/testdrive/kafka-avro-upsert-sinks.td
index b7d7b347a3393..c2bf9965ce7a2 100644
--- a/test/testdrive/kafka-avro-upsert-sinks.td
+++ b/test/testdrive/kafka-avro-upsert-sinks.td
@@ -447,3 +447,9 @@ contains:upsert key could not be validated as unique
 "Dataflow: materialize.public.input_with_deletions_keyed" "Reduced Fallibly MinsMaxesHierarchical"
 "Dataflow: materialize.public.input_with_deletions_keyed" "Reduced Fallibly MinsMaxesHierarchical"
 "Dataflow: materialize.public.input_with_deletions_keyed" "Reduced Fallibly MinsMaxesHierarchical"
+"Dataflow: materialize.public.input_keyed" "Arrange errors"
+"Dataflow: materialize.public.input_keyed" "Distinct errors"
+"Dataflow: materialize.public.input_keyed_ab" "Arrange errors"
+"Dataflow: materialize.public.input_keyed_ab" "Distinct errors"
+"Dataflow: materialize.public.input_with_deletions_keyed" "Arrange errors"
+"Dataflow: materialize.public.input_with_deletions_keyed" "Distinct errors"
diff --git a/test/testdrive/materializations.td b/test/testdrive/materializations.td
index 8fe17b5c20289..38280d337ba54 100644
--- a/test/testdrive/materializations.td
+++ b/test/testdrive/materializations.td
@@ -496,3 +496,15 @@ c    d
 "Dataflow: materialize.public.test7" AccumulableErrorCheck
 "Dataflow: materialize.public.test7" "ArrangeAccumulable [val: empty]"
 "Dataflow: materialize.public.test7" ReduceAccumulable
+"Dataflow: materialize.public.test1" "Arrange errors"
+"Dataflow: materialize.public.test1" "Distinct errors"
+"Dataflow: materialize.public.test2" "Arrange errors"
+"Dataflow: materialize.public.test2" "Distinct errors"
+"Dataflow: materialize.public.test4" "Arrange errors"
+"Dataflow: materialize.public.test4" "Distinct errors"
+"Dataflow: materialize.public.test6" "Arrange errors"
+"Dataflow: materialize.public.test6" "Distinct errors"
+"Dataflow: materialize.public.test7" "Arrange bundle err"
+"Dataflow: materialize.public.test7" "Arrange errors"
+"Dataflow: materialize.public.test7" "Distinct errors"
+"Dataflow: materialize.public.test7" "Distinct errors[[]]"
diff --git a/test/testdrive/monotonic.td b/test/testdrive/monotonic.td
index 7428d3de28feb..eea687c6ea778 100644
--- a/test/testdrive/monotonic.td
+++ b/test/testdrive/monotonic.td
@@ -243,3 +243,11 @@ Target cluster: quickstart
 "Dataflow: materialize.public.i8_primary_idx" "ArrangeBy[[Column(0, \"b\")]]-errors"
 "Dataflow: materialize.public.i9_primary_idx" "ArrangeBy[[Column(0, \"b\")]]"
 "Dataflow: materialize.public.i9_primary_idx" "ArrangeBy[[Column(0, \"b\")]]-errors"
+"Dataflow: materialize.public.m1" "Arrange errors"
+"Dataflow: materialize.public.m1" "Distinct errors"
+"Dataflow: materialize.public.monotonic_fused" "Arrange errors"
+"Dataflow: materialize.public.monotonic_fused" "Distinct errors"
+"Dataflow: materialize.public.monotonic_max" "Arrange errors"
+"Dataflow: materialize.public.monotonic_max" "Distinct errors"
+"Dataflow: materialize.public.monotonic_min" "Arrange errors"
+"Dataflow: materialize.public.monotonic_min" "Distinct errors"
diff --git a/test/testdrive/top-1-monotonic.td b/test/testdrive/top-1-monotonic.td
index 13eecd6fdb4bc..ddd2d4376bfc9 100644
--- a/test/testdrive/top-1-monotonic.td
+++ b/test/testdrive/top-1-monotonic.td
@@ -311,3 +311,35 @@ $ kafka-ingest format=avro topic=top1 schema=${schema} timestamp=5
 "Dataflow: group_by_order_by_in_top_1" "Arranged DistinctBy"
 "Dataflow: group_by_order_by_in_top_1" DistinctBy
 "Dataflow: group_by_order_by_in_top_1" DistinctByErrorCheck
+"Dataflow: group_by_in_top_1" "Arrange bundle err"
+"Dataflow: group_by_in_top_1" "Arrange bundle err"
+"Dataflow: group_by_in_top_1" "Arrange errors"
+"Dataflow: group_by_in_top_1" "Distinct errors"
+"Dataflow: group_by_in_top_1" "Distinct errors[[Column(0)]]"
+"Dataflow: group_by_in_top_1" "Distinct errors[[Column(0)]]"
+"Dataflow: group_by_limit" "Arrange bundle err"
+"Dataflow: group_by_limit" "Distinct errors[[]]"
+"Dataflow: group_by_order_by_in_top_1" "Arrange bundle err"
+"Dataflow: group_by_order_by_in_top_1" "Arrange bundle err"
+"Dataflow: group_by_order_by_in_top_1" "Arrange errors"
+"Dataflow: group_by_order_by_in_top_1" "Distinct errors"
+"Dataflow: group_by_order_by_in_top_1" "Distinct errors[[Column(0)]]"
+"Dataflow: group_by_order_by_in_top_1" "Distinct errors[[Column(0)]]"
+"Dataflow: limit_only" "Arrange bundle err"
+"Dataflow: limit_only" "Distinct errors[[]]"
+"Dataflow: order_by_desc_limit" "Arrange bundle err"
+"Dataflow: order_by_desc_limit" "Distinct errors[[]]"
+"Dataflow: order_by_limit" "Arrange bundle err"
+"Dataflow: order_by_limit" "Distinct errors[[]]"
+"Dataflow: group_by_in_top_1" "Arrange errors"
+"Dataflow: group_by_in_top_1" "Distinct errors"
+"Dataflow: group_by_limit" "Arrange errors"
+"Dataflow: group_by_limit" "Distinct errors"
+"Dataflow: group_by_order_by_in_top_1" "Arrange errors"
+"Dataflow: group_by_order_by_in_top_1" "Distinct errors"
+"Dataflow: limit_only" "Arrange errors"
+"Dataflow: limit_only" "Distinct errors"
+"Dataflow: order_by_desc_limit" "Arrange errors"
+"Dataflow: order_by_desc_limit" "Distinct errors"
+"Dataflow: order_by_limit" "Arrange errors"
+"Dataflow: order_by_limit" "Distinct errors"
diff --git a/test/testdrive/unified-compute-introspection.td b/test/testdrive/unified-compute-introspection.td
index 9e23821a8caec..8b8ae48ba5017 100644
--- a/test/testdrive/unified-compute-introspection.td
+++ b/test/testdrive/unified-compute-introspection.td
@@ -52,7 +52,7 @@ mv_div_by_zero   r1 1
   JOIN mz_cluster_replicas r ON c.replica_id = r.id
   WHERE c.count != 0
 idx1_div_by_zero r1 3
-idx2_div_by_zero r1 3
+idx2_div_by_zero r1 1
 mv_div_by_zero   r1 3
 
 > DELETE FROM zeros
@@ -73,7 +73,7 @@ mv_div_by_zero   r1 3
   JOIN mz_cluster_replicas r ON c.replica_id = r.id
   WHERE c.count != 0
 idx1_div_by_zero r1 2
-idx2_div_by_zero r1 2
+idx2_div_by_zero r1 1
 mv_div_by_zero   r1 2
 
 > ALTER CLUSTER test SET (REPLICATION FACTOR 2)
@@ -85,8 +85,8 @@ mv_div_by_zero   r1 2
   WHERE c.count != 0
 idx1_div_by_zero r1 2
 idx1_div_by_zero r2 2
-idx2_div_by_zero r1 2
-idx2_div_by_zero r2 2
+idx2_div_by_zero r1 1
+idx2_div_by_zero r2 1
 mv_div_by_zero   r1 2
 mv_div_by_zero   r2 2
 
@@ -100,7 +100,7 @@ mv_div_by_zero   r2 2
   JOIN mz_cluster_replicas r ON c.replica_id = r.id
   WHERE c.count != 0
 idx1_div_by_zero r1 2
-idx2_div_by_zero r1 2
+idx2_div_by_zero r1 1
 mv_div_by_zero   r1 2
 
 # Test that error logging is retracted when objects are dropped.
@@ -112,7 +112,7 @@ mv_div_by_zero   r1 2
   JOIN mz_objects o ON c.object_id = o.id
   JOIN mz_cluster_replicas r ON c.replica_id = r.id
   WHERE c.count != 0
-idx2_div_by_zero r1 2
+idx2_div_by_zero r1 1
 mv_div_by_zero   r1 2
 
 > DROP MATERIALIZED VIEW mv_div_by_zero