diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py
index a2e156a9e627b..042251f16b068 100644
--- a/misc/python/materialize/mzcompose/__init__.py
+++ b/misc/python/materialize/mzcompose/__init__.py
@@ -284,6 +284,11 @@ def get_variable_system_parameters(
         VariableSystemParameter(
             "enable_columnar_merge_batcher", "true", ["true", "false"]
         ),
+        # On by default so CI exercises the columnar accumulable diff layout, which
+        # is off in production while it earns trust.
+        VariableSystemParameter(
+            "enable_columnar_accumulable_diff", "true", ["true", "false"]
+        ),
         VariableSystemParameter(
             "compute_peek_response_stash_threshold_bytes",
             # 1 MiB, an in-between value
diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py
index 6a008fc235834..f2d61ed41340a 100644
--- a/misc/python/materialize/parallel_workload/action.py
+++ b/misc/python/materialize/parallel_workload/action.py
@@ -3141,6 +3141,7 @@ def __init__(
         self.flags_with_values["enable_compute_sync_mv_sink"] = BOOLEAN_FLAG_VALUES
         self.flags_with_values["enable_column_paged_batcher"] = BOOLEAN_FLAG_VALUES
         self.flags_with_values["enable_columnar_merge_batcher"] = BOOLEAN_FLAG_VALUES
+        self.flags_with_values["enable_columnar_accumulable_diff"] = BOOLEAN_FLAG_VALUES
         self.flags_with_values["enable_column_paged_batcher_spill"] = (
             BOOLEAN_FLAG_VALUES
         )
diff --git a/src/compute-types/src/dyncfgs.rs b/src/compute-types/src/dyncfgs.rs
index 01cf2f5292d9c..1d74ae1ba03f0 100644
--- a/src/compute-types/src/dyncfgs.rs
+++ b/src/compute-types/src/dyncfgs.rs
@@ -90,6 +90,22 @@ pub const ENABLE_COLUMNAR_MERGE_BATCHER: Config<bool> = Config::new(
     ParameterScope::Replica,
 );
 
+/// Store the accumulable reduce's accumulators in columnar form.
+///
+/// The accumulable reduce keeps one accumulator per aggregate in the diff of its
+/// input arrangement. When `true`, that arrangement holds its diffs in a columnar
+/// container, which lays the accumulators out by variant so each pays only for its
+/// own fields. When `false` (the default), the diffs live in a columnation stack at
+/// the width of the largest variant. Read at operator construction time, so flips
+/// take effect on dataflows created after the change.
+pub const ENABLE_COLUMNAR_ACCUMULABLE_DIFF: Config<bool> = Config::new(
+    "enable_columnar_accumulable_diff",
+    false,
+    "Store the accumulable reduce's accumulators in a columnar arrangement diff, laid out by \
+     variant, instead of a columnation stack of fixed-width accumulator enums.",
+    ParameterScope::Replica,
+);
+
 /// Allow the column-paged batcher's pager to evict chunks under memory
 /// pressure. Only meaningful when [`ENABLE_COLUMN_PAGED_BATCHER`] is `true`.
 /// With the spill flag off the pager keeps every chunk resident regardless of
@@ -844,6 +860,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&MV_SINK_ADVANCE_PERSIST_FRONTIERS)
         .add(&ENABLE_COLUMN_PAGED_BATCHER)
         .add(&ENABLE_COLUMNAR_MERGE_BATCHER)
+        .add(&ENABLE_COLUMNAR_ACCUMULABLE_DIFF)
         .add(&ENABLE_COLUMN_PAGED_BATCHER_SPILL)
         .add(&COLUMN_PAGED_BATCHER_BUDGET_FRACTION)
         .add(&COLUMN_PAGED_BATCHER_LZ4)
diff --git a/src/compute/src/extensions/arrange.rs b/src/compute/src/extensions/arrange.rs
index b06e1a56790b7..73cc2a9589608 100644
--- a/src/compute/src/extensions/arrange.rs
+++ b/src/compute/src/extensions/arrange.rs
@@ -15,12 +15,14 @@ use differential_dataflow::difference::Semigroup;
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::arrangement::arrange_core;
 use differential_dataflow::operators::arrange::{Arranged, TraceAgent};
+use differential_dataflow::trace::implementations::BatchContainer;
 use differential_dataflow::trace::implementations::spine_fueled::Spine;
 use differential_dataflow::trace::{Batch, Batcher, Builder, Trace, TraceReader};
 use differential_dataflow::{Collection, Data, ExchangeData, Hashable, VecCollection};
 use mz_compute_types::dyncfgs::{ENABLE_COLUMN_PAGED_BATCHER, ENABLE_COLUMNAR_MERGE_BATCHER};
 use mz_dyncfg::ConfigSet;
 use mz_row_spine::ArcBatch;
+use mz_timely_util::containers::HeapSize;
 use timely::Container;
 use timely::container::{ContainerBuilder, PushInto};
 use timely::dataflow::Stream;
@@ -497,10 +499,10 @@ where
     }
 }
 
-impl<'scope, T, R> ArrangementSize for Arranged<'scope, RowAgent<T, R>>
+impl<'scope, T, DC> ArrangementSize for Arranged<'scope, RowAgent<T, DC::Owned, DC>>
 where
     T: MzTimestamp,
-    R: Semigroup + Ord + MzArrangeData + 'static,
+    DC: BatchContainer<Owned: Semigroup + 'static> + HeapSize,
 {
     fn log_arrangement_size(self) -> Self {
         log_arrangement_size_inner(self, |batch| {
diff --git a/src/compute/src/render/reduce.rs b/src/compute/src/render/reduce.rs
index a5e8afcbb36fa..3d896f2c4eed3 100644
--- a/src/compute/src/render/reduce.rs
+++ b/src/compute/src/render/reduce.rs
@@ -13,10 +13,11 @@
 
 use std::collections::BTreeMap;
 
+use columnar::Columnar;
 use columnation::{Columnation, CopyRegion};
-use dec::OrderedDecimal;
 use differential_dataflow::Diff as _;
 use differential_dataflow::collection::AsCollection;
+use differential_dataflow::columnar::layout::Coltainer;
 use differential_dataflow::consolidation::ConsolidatingContainerBuilder;
 use differential_dataflow::difference::{IsZero, Multiply, Semigroup};
 use differential_dataflow::hashable::Hashable;
@@ -26,7 +27,9 @@ use differential_dataflow::trace::implementations::BatchContainer;
 use differential_dataflow::trace::{Builder, Cursor, Navigable, Trace};
 use differential_dataflow::{Data, VecCollection};
 use itertools::Itertools;
-use mz_compute_types::dyncfgs::{ENABLE_COMPUTE_TEMPORAL_BUCKETING, TEMPORAL_BUCKETING_SUMMARY};
+use mz_compute_types::dyncfgs::{
+    ENABLE_COLUMNAR_ACCUMULABLE_DIFF, ENABLE_COMPUTE_TEMPORAL_BUCKETING, TEMPORAL_BUCKETING_SUMMARY,
+};
 use mz_compute_types::plan::ArrangementStrategy;
 use mz_compute_types::plan::reduce::{
     AccumulablePlan, BasicPlan, BucketedPlan, HierarchicalPlan, KeyValPlan, LirAggregateExpr,
@@ -35,7 +38,7 @@ use mz_compute_types::plan::reduce::{
 use mz_compute_types::plan::scalar::LirScalarExpr;
 use mz_expr::{AggregateFunc, EvalError, SafeMfpPlan};
 use mz_ore::cast::CastLossy;
-use mz_repr::adt::numeric::{self, Numeric, NumericAgg};
+use mz_repr::adt::numeric::{self, Numeric, NumericAgg, OrderedNumericAgg};
 use mz_repr::fixed_length::ExtendDatums;
 use mz_repr::{Datum, DatumVec, Diff, Row, RowArena, SharedRow};
 use mz_timely_util::columnation::ColumnationChunker;
@@ -54,8 +57,8 @@ use crate::render::errors::MaybeValidatingRow;
 use crate::render::reduce::monoids::{ReductionMonoid, get_monoid};
 use crate::render::{ArrangementFlavor, Pairer, RenderTimestamp};
 use crate::typedefs::{
-    ErrBatcher, ErrBuilder, KeyBatcher, RowErrBuilder, RowErrSpine, RowRowAgent, RowRowArrangement,
-    RowRowSpine, RowSpine, RowValSpine,
+    ErrBatcher, ErrBuilder, KeyBatcher, RowAgent, RowErrBuilder, RowErrSpine, RowRowAgent,
+    RowRowArrangement, RowRowSpine, RowSpine, RowValSpine,
 };
 use mz_row_spine::{
     DatumContainer, DatumSeq, RowBatcher, RowBuilder, RowRowBatcher, RowRowBuilder, RowValBatcher,
@@ -1473,6 +1476,50 @@ impl<'scope, T: RenderTimestamp> Context<'scope, T> {
             differential_dataflow::collection::concatenate(collection_scope, to_aggregate)
         };
 
+        // The accumulators travel in the arrangement's diffs. A columnar diff container
+        // lays each `Accum` out by variant, so it occupies only its own variant's
+        // columns rather than the footprint of the largest variant. Both layouts feed
+        // the same reduce operators.
+        if ENABLE_COLUMNAR_ACCUMULABLE_DIFF.get(&self.config_set) {
+            let arranged = collection
+                .mz_arrange::<
+                    ColumnationChunker<_>,
+                    RowBatcher<_, _>,
+                    RowBuilder<_, _, Coltainer<_>>,
+                    RowSpine<_, (Vec<Accum>, Diff), Coltainer<_>>,
+                >(
+                    "ArrangeAccumulable [val: empty]",
+                );
+            self.reduce_accumulable(arranged, full_aggrs, mfp_after)
+        } else {
+            let arranged = collection
+                .mz_arrange::<
+                    ColumnationChunker<_>,
+                    RowBatcher<_, _>,
+                    RowBuilder<_, _>,
+                    RowSpine<_, (Vec<Accum>, Diff)>,
+                >(
+                    "ArrangeAccumulable [val: empty]",
+                );
+            self.reduce_accumulable(arranged, full_aggrs, mfp_after)
+        }
+    }
+
+    /// Reduces arranged accumulators to output rows, and to the errors the accumulated
+    /// values can reveal. Generic over the container holding the diffs, so both diff
+    /// layouts share one rendering of the reduce operators.
+    fn reduce_accumulable<'s, DC>(
+        &self,
+        arranged: Arranged<'s, RowAgent<T, (Vec<Accum>, Diff), DC>>,
+        full_aggrs: Vec<LirAggregateExpr>,
+        mfp_after: Option<SafeMfpPlan<LirScalarExpr>>,
+    ) -> (
+        RowRowArrangement<'s, T>,
+        VecCollection<'s, T, DataflowErrorSer, Diff>,
+    )
+    where
+        DC: BatchContainer<Owned = (Vec<Accum>, Diff)>,
+    {
         // Allocations for the two closures.
         let mut datums1 = DatumVec::new();
         let mut datums2 = DatumVec::new();
@@ -1482,15 +1529,6 @@ impl<'scope, T: RenderTimestamp> Context<'scope, T> {
 
         let error_logger = self.error_logger();
         let err_full_aggrs = full_aggrs.clone();
-        let arranged = collection
-            .mz_arrange::<
-                ColumnationChunker<_>,
-                RowBatcher<_, _>,
-                RowBuilder<_, _>,
-                RowSpine<_, (Vec<Accum>, Diff)>,
-            >(
-                "ArrangeAccumulable [val: empty]",
-            );
         let arranged_output = arranged
             .clone()
             .mz_reduce_abelian::<_, RowRowBuilder<_, _>, RowRowSpine<_, _>, _>(
@@ -1622,7 +1660,7 @@ fn accumulable_zero(aggr_func: &AggregateFunc) -> Accum {
             non_nulls: Diff::ZERO,
         },
         AggregateFunc::SumNumeric => Accum::Numeric {
-            accum: OrderedDecimal(NumericAgg::zero()),
+            accum: OrderedNumericAgg(NumericAgg::zero()),
             pos_infs: Diff::ZERO,
             neg_infs: Diff::ZERO,
             nans: Diff::ZERO,
@@ -1778,7 +1816,7 @@ fn datum_to_accumulator(aggregate_func: &AggregateFunc, datum: Datum) -> Accum {
                 };
 
                 Accum::Numeric {
-                    accum: OrderedDecimal(accum),
+                    accum: OrderedNumericAgg(accum),
                     pos_infs,
                     neg_infs,
                     nans,
@@ -1786,7 +1824,7 @@ fn datum_to_accumulator(aggregate_func: &AggregateFunc, datum: Datum) -> Accum {
                 }
             }
             Datum::Null => Accum::Numeric {
-                accum: OrderedDecimal(NumericAgg::zero()),
+                accum: OrderedNumericAgg(NumericAgg::zero()),
                 pos_infs: Diff::ZERO,
                 neg_infs: Diff::ZERO,
                 nans: Diff::ZERO,
@@ -2018,8 +2056,12 @@ type AccumCount = mz_ore::Overflowing<i128>;
     PartialOrd,
     Ord,
     Serialize,
-    Deserialize
+    Deserialize,
+    Columnar
 )]
+// The columnar container orders references with this derived `Ord`, which must agree with
+// the owned `Ord`. It does because every field's reference type is its owned type.
+#[columnar(derive(PartialEq, Eq, PartialOrd, Ord))]
 enum Accum {
     /// Accumulates boolean values.
     Bool {
@@ -2052,7 +2094,7 @@ enum Accum {
     /// Accumulates arbitrary precision decimals.
     Numeric {
         /// Accumulates non-special values
-        accum: OrderedDecimal<NumericAgg>,
+        accum: OrderedNumericAgg,
         /// Counts +inf
         pos_infs: Diff,
         /// Counts -inf
@@ -2254,7 +2296,7 @@ impl Multiply<Diff> for Accum {
                 // http://speleotrove.com/decimal/dncont.html
                 assert!(!cx.status().rounded(), "Accum::Numeric multiply overflow");
                 Accum::Numeric {
-                    accum: OrderedDecimal(f),
+                    accum: OrderedNumericAgg(f),
                     pos_infs: pos_infs * factor,
                     neg_infs: neg_infs * factor,
                     nans: nans * factor,
@@ -2265,6 +2307,8 @@ impl Multiply<Diff> for Accum {
     }
 }
 
+// The batcher stages updates in columnation chunks before they reach the arrangement,
+// which stores `Accum` in its columnar form.
 impl Columnation for Accum {
     type InnerRegion = CopyRegion<Self>;
 }
@@ -2722,4 +2766,110 @@ mod tests {
         let datum = finalize_accum(&func, &acc, Diff::from(2_i64));
         assert_eq!(datum, Datum::from(0.0_f64));
     }
+
+    /// Accumulators of every variant, in zero, accumulated, and negated states.
+    fn sample_accums() -> Vec<Accum> {
+        let mut cx = numeric::cx_datum();
+        let mut numeric = |s: &str| Datum::from(cx.parse(s).unwrap());
+        let cases: Vec<(AggregateFunc, Vec<Datum>)> = vec![
+            (AggregateFunc::Count, vec![Datum::Null, Datum::Int64(5)]),
+            (
+                AggregateFunc::SumInt64,
+                vec![Datum::Int64(-7), Datum::Int64(i64::MAX)],
+            ),
+            (
+                AggregateFunc::SumUInt16,
+                vec![Datum::UInt16(3), Datum::Null],
+            ),
+            (
+                AggregateFunc::Any,
+                vec![Datum::True, Datum::False, Datum::Null],
+            ),
+            (
+                AggregateFunc::SumFloat64,
+                vec![
+                    Datum::from(1.5_f64),
+                    Datum::from(f64::NAN),
+                    Datum::from(f64::NEG_INFINITY),
+                ],
+            ),
+            (
+                AggregateFunc::SumNumeric,
+                vec![
+                    numeric("-12345.678"),
+                    numeric("9e39"),
+                    numeric("NaN"),
+                    numeric("Infinity"),
+                    Datum::Null,
+                ],
+            ),
+        ];
+        let mut accums = Vec::new();
+        for (func, datums) in cases {
+            let mut sum = accumulable_zero(&func);
+            accums.push(sum);
+            for datum in datums {
+                let accum = datum_to_accumulator(&func, datum);
+                sum.plus_equals(&accum);
+                accums.push(accum);
+                accums.push(accum.multiply(&Diff::from(-1_i64)));
+            }
+            accums.push(sum);
+        }
+        accums
+    }
+
+    #[mz_ore::test]
+    fn accum_columnar_round_trip() {
+        use columnar::bytes::indexed::{DecodedStore, encode};
+        use columnar::{AsBytes, Borrow, BorrowedOf, FromBytes, Index, Len};
+        use differential_dataflow::trace::implementations::BatchContainer;
+
+        let accums = sample_accums();
+        let container = Accum::as_columns(accums.iter());
+        assert_eq!(container.len(), accums.len());
+        let borrowed = container.borrow();
+        for (index, accum) in accums.iter().enumerate() {
+            assert_eq!(Accum::into_owned(borrowed.get(index)), *accum);
+        }
+        for (i, a) in accums.iter().enumerate() {
+            for (j, b) in accums.iter().enumerate() {
+                assert_eq!(borrowed.get(i).cmp(&borrowed.get(j)), a.cmp(b));
+            }
+        }
+
+        let bytes: Vec<&[u8]> = borrowed.as_bytes().map(|(_align, bytes)| bytes).collect();
+        let decoded = BorrowedOf::<Accum>::from_bytes(&mut bytes.into_iter());
+        for (index, accum) in accums.iter().enumerate() {
+            assert_eq!(Accum::into_owned(decoded.get(index)), *accum);
+        }
+        // NOTE: the `i128` columns cannot be `validate`d, see the `Overflowing<i128>` test in
+        // `mz_ore`, so this only decodes.
+        let mut words = Vec::new();
+        encode(&mut words, &borrowed);
+        let decoded = BorrowedOf::<Accum>::from_store(&DecodedStore::new(&words), &mut 0);
+        for (index, accum) in accums.iter().enumerate() {
+            assert_eq!(Accum::into_owned(decoded.get(index)), *accum);
+        }
+
+        // The arrangement's diff container, holding whole `(Vec<Accum>, Diff)` diffs.
+        let diffs: Vec<(Vec<Accum>, Diff)> = accums
+            .chunks(3)
+            .map(|chunk| (chunk.to_vec(), Diff::ONE))
+            .collect();
+        let mut coltainer = Coltainer::<(Vec<Accum>, Diff)>::default();
+        for diff in &diffs {
+            coltainer.push_own(diff);
+        }
+        assert_eq!(coltainer.len(), diffs.len());
+        for (index, diff) in diffs.iter().enumerate() {
+            assert_eq!(
+                <Coltainer<(Vec<Accum>, Diff)>>::into_owned(coltainer.index(index)),
+                *diff
+            );
+        }
+        let mut sum = <Coltainer<(Vec<Accum>, Diff)>>::into_owned(coltainer.index(0));
+        sum.plus_equals(&sum.clone().multiply(&Diff::from(-1_i64)));
+        assert!(sum.is_zero());
+    }
 }
diff --git a/src/compute/src/typedefs.rs b/src/compute/src/typedefs.rs
index 0c71a624f5fbb..3a0883281dbdc 100644
--- a/src/compute/src/typedefs.rs
+++ b/src/compute/src/typedefs.rs
@@ -18,7 +18,7 @@ use differential_dataflow::trace::implementations::merge_batcher::MergeBatcher;
 use differential_dataflow::trace::wrappers::enter::TraceEnter;
 use differential_dataflow::trace::wrappers::frontier::TraceFrontier;
 use mz_repr::Diff;
-use mz_timely_util::columnation::ColInternalMerger;
+use mz_timely_util::columnation::{ColInternalMerger, ColumnationStack};
 
 use mz_row_spine::RowValBuilder;
 
@@ -96,7 +96,7 @@ pub type RowRowAgent<T, R> = TraceAgent<RowRowSpine<T, R>>;
 pub type RowRowArrangement<'scope, T> = Arranged<'scope, RowRowAgent<T, Diff>>;
 pub type RowRowEnter<T, R, TEnter> = TraceEnter<TraceFrontier<RowRowAgent<T, R>>, TEnter>;
 // Row specialized spines and agents.
-pub type RowAgent<T, R> = TraceAgent<RowSpine<T, R>>;
+pub type RowAgent<T, R, DC = ColumnationStack<R>> = TraceAgent<RowSpine<T, R, DC>>;
 pub type RowArrangement<'scope, T> = Arranged<'scope, RowAgent<T, Diff>>;
 pub type RowEnter<T, R, TEnter> = TraceEnter<TraceFrontier<RowAgent<T, R>>, TEnter>;
 
diff --git a/src/ore/src/overflowing.rs b/src/ore/src/overflowing.rs
index 93c5128d205d7..0a15a39c07555 100644
--- a/src/ore/src/overflowing.rs
+++ b/src/ore/src/overflowing.rs
@@ -98,6 +98,7 @@ impl<T: std::fmt::Display> std::fmt::Display for Overflowing<T> {
 #[cfg(feature = "columnar")]
 mod columnar {
     use crate::overflowing::Overflowing;
+    use columnar::bytes::indexed::DecodedStore;
     use columnar::common::PushIndexAs;
     use columnar::{
         AsBytes, Borrow, Clear, Columnar, Container, FromBytes, Index, IndexAs, Len, Push,
@@ -105,16 +106,15 @@ mod columnar {
     use serde::{Deserialize, Serialize};
     use std::ops::Range;
 
-    impl<T: Columnar + Copy + Send> Columnar for Overflowing<T>
+    impl<T: Columnar<Container: PushIndexAs<T>> + Copy + Send> Columnar for Overflowing<T>
     where
-        for<'a> &'a [T]: AsBytes<'a> + FromBytes<'a>,
         Overflowing<T>: From<T>,
     {
         #[inline(always)]
         fn into_owned(other: columnar::Ref<'_, Self>) -> Self {
             other
         }
-        type Container = Overflows<T>;
+        type Container = Overflows<T, T::Container>;
         #[inline(always)]
         fn reborrow<'b, 'a: 'b>(thing: columnar::Ref<'a, Self>) -> columnar::Ref<'b, Self>
         where
@@ -124,9 +124,11 @@ mod columnar {
         }
     }
 
-    /// Columnar container for [`Overflowing`].
+    /// Columnar container for [`Overflowing`], delegating to `T`'s own container `TC`, so
+    /// `Overflowing<i128>` uses columnar's byte-encoded `i128` store rather than requiring
+    /// `&[i128]` to be castable to bytes.
     #[derive(Copy, Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
-    pub struct Overflows<T, TC = Vec<T>>(TC, std::marker::PhantomData<T>);
+    pub struct Overflows<T, TC>(TC, std::marker::PhantomData<T>);
 
     impl<T, TC: Default> Default for Overflows<T, TC> {
         #[inline(always)]
@@ -201,6 +203,13 @@ mod columnar {
         fn from_bytes(bytes: &mut impl Iterator<Item = &'a [u8]>) -> Self {
             Self(TC::from_bytes(bytes), std::marker::PhantomData)
         }
+        #[inline(always)]
+        fn from_store(store: &DecodedStore<'a>, offset: &mut usize) -> Self {
+            Self(TC::from_store(store, offset), std::marker::PhantomData)
+        }
+        fn element_sizes(sizes: &mut Vec<usize>) -> Result<(), String> {
+            TC::element_sizes(sizes)
+        }
     }
 
     impl<T: Copy, TC: Len> Len for Overflows<T, TC> {
@@ -235,10 +244,10 @@ mod columnar {
         }
     }
 
-    impl<T: Copy, TC: Push<T>> Push<&Overflowing<T>> for Overflows<T, TC> {
+    impl<T: Copy, TC: for<'a> Push<&'a T>> Push<&Overflowing<T>> for Overflows<T, TC> {
         #[inline(always)]
         fn push(&mut self, item: &Overflowing<T>) {
-            self.0.push(item.0);
+            self.0.push(&item.0);
         }
     }
 }
@@ -768,4 +777,42 @@ mod test {
         let result = Overflowing::<i8>::MAX.checked_add(Overflowing::<i8>::ONE);
         assert_eq!(result, None);
     }
+
+    #[cfg(feature = "columnar")]
+    #[crate::test]
+    fn test_columnar_i128_round_trip() {
+        use ::columnar::{AsBytes, Borrow, BorrowedOf, Columnar, FromBytes, Index, Len};
+
+        let values = [
+            Overflowing::<i128>::MIN,
+            Overflowing(-7),
+            Overflowing::<i128>::ZERO,
+            Overflowing::<i128>::ONE,
+            Overflowing::<i128>::MAX,
+        ];
+        let container = Overflowing::<i128>::as_columns(values.iter());
+        assert_eq!(container.len(), values.len());
+        let borrowed = container.borrow();
+        for (index, value) in values.iter().enumerate() {
+            assert_eq!(borrowed.get(index), *value);
+        }
+
+        let bytes: Vec<&[u8]> = borrowed.as_bytes().map(|(_align, bytes)| bytes).collect();
+        let decoded = BorrowedOf::<Overflowing<i128>>::from_bytes(&mut bytes.into_iter());
+        assert_eq!(decoded.len(), values.len());
+        for (index, value) in values.iter().enumerate() {
+            assert_eq!(decoded.get(index), *value);
+        }
+
+        // NOTE: columnar's `i128` store does not implement `element_sizes`, so the indexed
+        // store can be decoded but not `validate`d for this type.
+        let mut words = Vec::new();
+        ::columnar::bytes::indexed::encode(&mut words, &borrowed);
+        let store = ::columnar::bytes::indexed::DecodedStore::new(&words);
+        let decoded = BorrowedOf::<Overflowing<i128>>::from_store(&store, &mut 0);
+        assert_eq!(decoded.len(), values.len());
+        for (index, value) in values.iter().enumerate() {
+            assert_eq!(decoded.get(index), *value);
+        }
+    }
 }
diff --git a/src/repr/src/adt/numeric.rs b/src/repr/src/adt/numeric.rs
index 81ba5dbfb681c..341d5af9bf25c 100644
--- a/src/repr/src/adt/numeric.rs
+++ b/src/repr/src/adt/numeric.rs
@@ -12,12 +12,13 @@
 //!
 //! [`rust-dec`]: https://github.com/MaterializeInc/rust-dec/
 
+use std::cmp::Ordering;
 use std::error::Error;
 use std::fmt;
 use std::sync::LazyLock;
 
 use anyhow::bail;
-use dec::{Context, Decimal};
+use dec::{Context, Decimal, OrderedDecimal};
 use mz_ore::cast;
 use mz_persist_types::columnar::FixedSizeCodec;
 use mz_proto::{ProtoType, RustType, TryFromProtoError};
@@ -53,6 +54,35 @@ pub const NUMERIC_AGG_MAX_PRECISION: u8 = NUMERIC_AGG_WIDTH * 3;
 /// A double-width version of [`Numeric`] for use in aggregations.
 pub type NumericAgg = Decimal<NUMERIC_AGG_WIDTH_USIZE>;
 
+/// A [`NumericAgg`] with the total order of [`OrderedDecimal`], storable in columnar form.
+///
+/// Equality and ordering are those of `OrderedDecimal`, so NaN equals NaN and every
+/// value has a defined position. This crate cannot implement `Columnar` for
+/// `OrderedDecimal<NumericAgg>`, as both the trait and the type are foreign, so the
+/// newtype hosts that impl.
+#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
+pub struct OrderedNumericAgg(pub NumericAgg);
+
+impl PartialEq for OrderedNumericAgg {
+    fn eq(&self, other: &Self) -> bool {
+        OrderedDecimal(self.0) == OrderedDecimal(other.0)
+    }
+}
+
+impl Eq for OrderedNumericAgg {}
+
+impl PartialOrd for OrderedNumericAgg {
+    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+        Some(self.cmp(other))
+    }
+}
+
+impl Ord for OrderedNumericAgg {
+    fn cmp(&self, other: &Self) -> Ordering {
+        OrderedDecimal(self.0).cmp(&OrderedDecimal(other.0))
+    }
+}
+
 static CX_DATUM: LazyLock<Context<Numeric>> = LazyLock::new(|| {
     let mut cx = Context::<Numeric>::default();
     cx.set_max_exponent(isize::from(NUMERIC_DATUM_MAX_PRECISION - 1))
@@ -861,6 +891,155 @@ impl FixedSizeCodec<Numeric> for PackedNumeric {
     }
 }
 
+mod columnar_impls {
+    use std::ops::Range;
+
+    use columnar::bytes::indexed::DecodedStore;
+    use columnar::{AsBytes, Borrow, Clear, Columnar, Container, FromBytes, Index, Len, Push};
+
+    use super::{NUMERIC_AGG_WIDTH_USIZE, NumericAgg, OrderedNumericAgg};
+
+    /// Width of the coefficient column. One unit wider than the decimal's coefficient so
+    /// that each element is 56 bytes, a whole number of `u64` words. The indexed byte
+    /// store pads every column to whole words and cannot decode an element width that
+    /// does not divide that padding.
+    const LSU_COLUMN_WIDTH: usize = NUMERIC_AGG_WIDTH_USIZE + 1;
+    /// The raw parts of a [`NumericAgg`], in the order `Decimal::to_raw_parts` returns
+    /// them, with the coefficient units zero-padded to [`LSU_COLUMN_WIDTH`].
+    type Parts = (u32, i32, u8, [u16; LSU_COLUMN_WIDTH]);
+    /// One column per raw part. The coefficient units use `Vec<[u16; N]>` directly rather
+    /// than the array's own columnar container, which would add per-element offsets to a
+    /// fixed-width value.
+    type PartsContainer = (Vec<u32>, Vec<i32>, Vec<u8>, Vec<[u16; LSU_COLUMN_WIDTH]>);
+    type PartsBorrowed<'a> = <PartsContainer as Borrow>::Borrowed<'a>;
+
+    impl Columnar for OrderedNumericAgg {
+        #[inline(always)]
+        fn into_owned(other: columnar::Ref<'_, Self>) -> Self {
+            other
+        }
+        type Container = OrderedNumericAggs;
+        #[inline(always)]
+        fn reborrow<'b, 'a: 'b>(thing: columnar::Ref<'a, Self>) -> columnar::Ref<'b, Self>
+        where
+            Self: 'a,
+        {
+            thing
+        }
+    }
+
+    /// Columnar container for [`OrderedNumericAgg`].
+    ///
+    /// References are owned values rebuilt from the part columns, so comparisons on
+    /// references use the decimal order rather than the raw parts' lexicographic order.
+    #[derive(Copy, Clone, Debug, Default)]
+    pub struct OrderedNumericAggs<TC = PartsContainer>(TC);
+
+    impl Borrow for OrderedNumericAggs {
+        type Ref<'a> = OrderedNumericAgg;
+        type Borrowed<'a> = OrderedNumericAggs<PartsBorrowed<'a>>;
+        #[inline(always)]
+        fn borrow<'a>(&'a self) -> Self::Borrowed<'a> {
+            OrderedNumericAggs(self.0.borrow())
+        }
+        #[inline(always)]
+        fn reborrow<'b, 'a: 'b>(item: Self::Borrowed<'a>) -> Self::Borrowed<'b>
+        where
+            Self: 'a,
+        {
+            OrderedNumericAggs(<PartsContainer as Borrow>::reborrow(item.0))
+        }
+        #[inline(always)]
+        fn reborrow_ref<'b, 'a: 'b>(item: Self::Ref<'a>) -> Self::Ref<'b>
+        where
+            Self: 'a,
+        {
+            item
+        }
+    }
+
+    impl Container for OrderedNumericAggs {
+        #[inline(always)]
+        fn extend_from_self(&mut self, other: Self::Borrowed<'_>, range: Range<usize>) {
+            self.0.extend_from_self(other.0, range);
+        }
+        #[inline(always)]
+        fn reserve_for<'a, I>(&mut self, selves: I)
+        where
+            Self: 'a,
+            I: Iterator<Item = Self::Borrowed<'a>> + Clone,
+        {
+            self.0.reserve_for(selves.map(|s| s.0));
+        }
+    }
+
+    impl<TC: Len> Len for OrderedNumericAggs<TC> {
+        #[inline(always)]
+        fn len(&self) -> usize {
+            self.0.len()
+        }
+    }
+
+    impl Clear for OrderedNumericAggs {
+        #[inline(always)]
+        fn clear(&mut self) {
+            self.0.clear();
+        }
+    }
+
+    impl<'a> Index for OrderedNumericAggs<PartsBorrowed<'a>> {
+        type Ref = OrderedNumericAgg;
+        #[inline(always)]
+        fn get(&self, index: usize) -> Self::Ref {
+            let (digits, exponent, bits, lsu) = self.0.get(index);
+            let mut units = [0u16; NUMERIC_AGG_WIDTH_USIZE];
+            units.copy_from_slice(&lsu[..NUMERIC_AGG_WIDTH_USIZE]);
+            OrderedNumericAgg(NumericAgg::from_raw_parts(*digits, *exponent, *bits, units))
+        }
+    }
+
+    impl Push<OrderedNumericAgg> for OrderedNumericAggs {
+        #[inline(always)]
+        fn push(&mut self, item: OrderedNumericAgg) {
+            let (digits, exponent, bits, units) = item.0.to_raw_parts();
+            let mut lsu = [0u16; LSU_COLUMN_WIDTH];
+            lsu[..NUMERIC_AGG_WIDTH_USIZE].copy_from_slice(&units);
+            let parts: Parts = (digits, exponent, bits, lsu);
+            self.0.push(parts);
+        }
+    }
+
+    impl Push<&OrderedNumericAgg> for OrderedNumericAggs {
+        #[inline(always)]
+        fn push(&mut self, item: &OrderedNumericAgg) {
+            self.push(*item);
+        }
+    }
+
+    impl<'a, TC: AsBytes<'a>> AsBytes<'a> for OrderedNumericAggs<TC> {
+        const SLICE_COUNT: usize = TC::SLICE_COUNT;
+        #[inline(always)]
+        fn get_byte_slice(&self, index: usize) -> (u64, &'a [u8]) {
+            self.0.get_byte_slice(index)
+        }
+    }
+
+    impl<'a, TC: FromBytes<'a>> FromBytes<'a> for OrderedNumericAggs<TC> {
+        const SLICE_COUNT: usize = TC::SLICE_COUNT;
+        #[inline(always)]
+        fn from_bytes(bytes: &mut impl Iterator<Item = &'a [u8]>) -> Self {
+            OrderedNumericAggs(TC::from_bytes(bytes))
+        }
+        #[inline(always)]
+        fn from_store(store: &DecodedStore<'a>, offset: &mut usize) -> Self {
+            OrderedNumericAggs(TC::from_store(store, offset))
+        }
+        fn element_sizes(sizes: &mut Vec<usize>) -> Result<(), String> {
+            TC::element_sizes(sizes)
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use mz_ore::assert_ok;
@@ -961,4 +1140,52 @@ mod tests {
 
         insta::assert_debug_snapshot!(all_numerics);
     }
+
+    #[mz_ore::test]
+    fn ordered_numeric_agg_columnar_round_trip() {
+        use columnar::{AsBytes, Borrow, BorrowedOf, Columnar, FromBytes, Index, Len};
+
+        let mut cx = cx_agg();
+        let values: Vec<OrderedNumericAgg> = [
+            "0",
+            "-0",
+            "1",
+            "-12345.678",
+            "9e39",
+            "9e-39",
+            "123456789012345678901234567890123456789012345678901234567890",
+            "NaN",
+            "Infinity",
+            "-Infinity",
+        ]
+        .into_iter()
+        .map(|s| OrderedNumericAgg(cx.parse(s).unwrap()))
+        .collect();
+
+        let container = OrderedNumericAgg::as_columns(values.iter());
+        assert_eq!(container.len(), values.len());
+        let borrowed = container.borrow();
+        for (index, value) in values.iter().enumerate() {
+            assert_eq!(borrowed.get(index), *value);
+        }
+
+        let bytes: Vec<&[u8]> = borrowed.as_bytes().map(|(_align, bytes)| bytes).collect();
+        let decoded = BorrowedOf::<OrderedNumericAgg>::from_bytes(&mut bytes.into_iter());
+        assert_eq!(decoded.len(), values.len());
+        for (index, value) in values.iter().enumerate() {
+            assert_eq!(decoded.get(index), *value);
+        }
+
+        // The indexed store pads each column to whole words, so decoding through it
+        // also checks that the column layout tolerates that padding.
+        let mut words = Vec::new();
+        columnar::bytes::indexed::encode(&mut words, &borrowed);
+        columnar::bytes::indexed::validate::<BorrowedOf<OrderedNumericAgg>>(&words).unwrap();
+        let store = columnar::bytes::indexed::DecodedStore::new(&words);
+        let decoded = BorrowedOf::<OrderedNumericAgg>::from_store(&store, &mut 0);
+        assert_eq!(decoded.len(), values.len());
+        for (index, value) in values.iter().enumerate() {
+            assert_eq!(decoded.get(index), *value);
+        }
+    }
 }
diff --git a/src/row-spine/src/lib.rs b/src/row-spine/src/lib.rs
index a64795a1c3008..80ca39a875511 100644
--- a/src/row-spine/src/lib.rs
+++ b/src/row-spine/src/lib.rs
@@ -34,6 +34,7 @@ pub static DICTIONARY_COMPRESSION: std::sync::atomic::AtomicBool =
 /// Spines specialized to contain `Row` types in keys and values.
 mod spines {
     use columnation::Columnation;
+    use differential_dataflow::trace::implementations::BatchContainer;
     use differential_dataflow::trace::implementations::Layout;
     use differential_dataflow::trace::implementations::Update;
     use differential_dataflow::trace::implementations::Vector;
@@ -76,9 +77,12 @@ mod spines {
     pub type RowValBuilder<V, T, R> =
         ArcBuilder<crate::dictionary::builders::RowValBuilder<V, T, R>>;
 
-    pub type RowSpine<T, R> = Spine<ArcBatch<OrdKeyBatch<RowLayout<((Row, ()), T, R)>>>>;
+    /// Key-only `Row` spine. `DC` is the diff container, see `RowLayout`.
+    pub type RowSpine<T, R, DC = ColumnationStack<R>> =
+        Spine<ArcBatch<OrdKeyBatch<RowLayout<((Row, ()), T, R), DC>>>>;
     pub type RowBatcher<T, R> = KeyBatcher<Row, T, R>;
-    pub type RowBuilder<T, R> = ArcBuilder<crate::dictionary::builders::RowBuilder<T, R>>;
+    pub type RowBuilder<T, R, DC = ColumnationStack<R>> =
+        ArcBuilder<crate::dictionary::builders::RowBuilder<T, R, DC>>;
 
     pub type ValRowSpine<K, T, R> = Spine<ArcBatch<OrdValBatch<ValRowLayout<((K, Row), T, R)>>>>;
     pub type ValRowBatcher<K, T, R> = KeyValBatcher<K, Row, T, R>;
@@ -116,8 +120,13 @@ mod spines {
     pub struct RowValLayout<U: Update<Key = Row>> {
         phantom: std::marker::PhantomData<U>,
     }
-    pub struct RowLayout<U: Update<Key = Row, Val = ()>> {
-        phantom: std::marker::PhantomData<U>,
+    /// Layout for key-only `Row` updates. `DC` is the diff `BatchContainer`, a
+    /// columnation stack by default.
+    pub struct RowLayout<U, DC = ColumnationStack<<U as Update>::Diff>>
+    where
+        U: Update<Key = Row, Val = ()>,
+    {
+        phantom: std::marker::PhantomData<(U, DC)>,
     }
     /// Mirror of [`RowValLayout`] with the roles swapped: arbitrary `Columnation`
     /// keys with `Row` values stored as packed bytes in a [`DatumContainer`].
@@ -148,15 +157,15 @@ mod spines {
         type DiffContainer = ColumnationStack<U::Diff>;
         type OffsetContainer = OffsetOptimized;
     }
-    impl<U: Update<Key = Row, Val = ()>> Layout for RowLayout<U>
+    impl<U: Update<Key = Row, Val = ()>, DC> Layout for RowLayout<U, DC>
     where
         U::Time: Columnation,
-        U::Diff: Columnation,
+        DC: BatchContainer<Owned = U::Diff>,
     {
         type KeyContainer = DatumContainer;
         type ValContainer = ColumnationStack<()>;
         type TimeContainer = ColumnationStack<U::Time>;
-        type DiffContainer = ColumnationStack<U::Diff>;
+        type DiffContainer = DC;
         type OffsetContainer = OffsetOptimized;
     }
     impl<U: Update<Val = Row>> Layout for ValRowLayout<U>
@@ -903,6 +912,7 @@ mod dictionary {
         use differential_dataflow::lattice::Lattice;
         use differential_dataflow::trace::Builder;
         use differential_dataflow::trace::Description;
+        use differential_dataflow::trace::implementations::BatchContainer;
         use differential_dataflow::trace::implementations::ord_neu::{OrdKeyBatch, OrdKeyBuilder};
         use differential_dataflow::trace::implementations::ord_neu::{OrdValBatch, OrdValBuilder};
         use mz_timely_util::columnar::Column;
@@ -1081,16 +1091,20 @@ mod dictionary {
         pub struct RowBuilder<
             T: Lattice + Timestamp + Columnation,
             R: Ord + Semigroup + Columnation + 'static,
+            DC: BatchContainer<Owned = R> = TimelyStack<R>,
         > {
-            inner: OrdKeyBuilder<RowLayout<((Row, ()), T, R)>, TimelyStack<((Row, ()), T, R)>>,
+            inner: OrdKeyBuilder<RowLayout<((Row, ()), T, R), DC>, TimelyStack<((Row, ()), T, R)>>,
         }
 
-        impl<T: Lattice + Timestamp + Columnation, R: Ord + Semigroup + Columnation + 'static>
-            Builder for RowBuilder<T, R>
+        impl<T, R, DC> Builder for RowBuilder<T, R, DC>
+        where
+            T: Lattice + Timestamp + Columnation,
+            R: Ord + Semigroup + Columnation + 'static,
+            DC: BatchContainer<Owned = R>,
         {
             type Input = TimelyStack<((Row, ()), T, R)>;
             type Time = T;
-            type Output = OrdKeyBatch<RowLayout<((Row, ()), T, R)>>;
+            type Output = OrdKeyBatch<RowLayout<((Row, ()), T, R), DC>>;
 
             fn with_capacity(keys: usize, vals: usize, upds: usize) -> Self {
                 Self {
diff --git a/src/timely-util/src/containers.rs b/src/timely-util/src/containers.rs
index 00dc8b7aafeaa..e6cb9845ef2e1 100644
--- a/src/timely-util/src/containers.rs
+++ b/src/timely-util/src/containers.rs
@@ -15,4 +15,7 @@
 
 //! Reusable containers.
 
+pub mod heap_size;
 pub mod stack;
+
+pub use heap_size::HeapSize;
diff --git a/src/timely-util/src/containers/heap_size.rs b/src/timely-util/src/containers/heap_size.rs
new file mode 100644
index 0000000000000..d997e107b624e
--- /dev/null
+++ b/src/timely-util/src/containers/heap_size.rs
@@ -0,0 +1,40 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+//! Heap size accounting for the containers that back arrangement batches.
+
+use columnar::{AsBytes, Borrow, Columnar};
+use columnation::Columnation;
+use differential_dataflow::columnar::layout::Coltainer;
+
+use crate::columnation::ColumnationStack;
+
+/// A container that can report the heap allocations backing it.
+pub trait HeapSize {
+    /// Calls `callback(size, capacity)`, in bytes, once per allocation backing `self`.
+    fn heap_size(&self, callback: impl FnMut(usize, usize));
+}
+
+impl<T: Columnation> HeapSize for ColumnationStack<T> {
+    fn heap_size(&self, callback: impl FnMut(usize, usize)) {
+        ColumnationStack::heap_size(self, callback)
+    }
+}
+
+impl<C: Columnar> HeapSize for Coltainer<C> {
+    fn heap_size(&self, mut callback: impl FnMut(usize, usize)) {
+        // Columnar containers expose their contents as byte slices, one per column, and each
+        // non-empty column is one `Vec` allocation, so the callback count is right. They do
+        // not expose spare capacity, so each slice reports its length as both size and
+        // capacity, and the capacity is a lower bound.
+        for (_align, bytes) in self.container.borrow().as_bytes() {
+            callback(bytes.len(), bytes.len());
+        }
+    }
+}
diff --git a/test/launchdarkly-flag-consistency/mzcompose.py b/test/launchdarkly-flag-consistency/mzcompose.py
index 618411c9e9835..dca6979751586 100644
--- a/test/launchdarkly-flag-consistency/mzcompose.py
+++ b/test/launchdarkly-flag-consistency/mzcompose.py
@@ -246,6 +246,7 @@
     enable_statement_arrival_logging
     enable_binary_date_bin
     enable_coalesce_case_transform
+    enable_columnar_accumulable_diff
     enable_columnar_merge_batcher
     enable_compute_half_join2
     enable_compute_index_peek_offload