From 5b113723ed08c0618290d17f03dc83dcd7b2e736 Mon Sep 17 00:00:00 2001 From: Darius Jahandarie Date: Wed, 9 Sep 2026 02:22:16 +0900 Subject: [PATCH] compute: pack accumulable reduce accumulators at their natural width The accumulable reduce's input arrangement carries one `Accum` per aggregate in its diff. `Accum` is an enum sized for its widest variant (`Numeric`, a 64 byte `Decimal<27>` plus four counters) and aligned to 16 by the `i128` in its siblings, so every aggregate costs 112 bytes even where `SimpleNumber` needs 24, `Float` 48, or `Bool` 16. A reduce with seven sums over integer and double columns pays about 816 bytes per group where 279 would do, and that arrangement dominates the dataflow's memory. Add `Accums`, a newtype over `Vec` that stores the same accumulators as a self-delimiting sequence of tagged slots at their natural widths, and let `enable_packed_accumulable_diff` select it as the arrangement diff. `Accum` stays the arithmetic type: every operation on `Accums` decodes a slot, reuses `Accum`'s `Semigroup`, `IsZero`, and `Multiply` impls, and re-encodes, so the accumulation domain (`i128`, `Decimal<27>`) and every output type are unchanged. Rendering is generic over a small `AccumulableDiff` trait implemented for both `Vec` and `Accums`; the `Vec` path is untouched. The flag defaults off in production and on in CI. Locally, on 100k groups with four integer and three double sums, the input arrangement drops from 835 to 298 bytes per record and the output arrangement is byte-identical. Co-Authored-By: Claude Opus 5 --- misc/python/materialize/mzcompose/__init__.py | 5 + .../materialize/parallel_workload/action.py | 1 + src/compute-types/src/dyncfgs.rs | 15 + src/compute/src/render/reduce.rs | 637 ++++++++++++++---- src/compute/src/render/reduce/tests.rs | 505 ++++++++++++++ .../mzcompose.py | 1 + 6 files changed, 1042 insertions(+), 122 deletions(-) create mode 100644 src/compute/src/render/reduce/tests.rs diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index 1be7ba3c7ec07..b587d7f4369fb 100644 --- a/misc/python/materialize/mzcompose/__init__.py +++ b/misc/python/materialize/mzcompose/__init__.py @@ -271,6 +271,11 @@ def get_variable_system_parameters( VariableSystemParameter( "enable_columnar_merge_batcher", "true", ["true", "false"] ), + # On by default so CI exercises the packed accumulable diff layout, which + # is off in production while it earns trust. + VariableSystemParameter( + "enable_packed_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 194d7b2f31516..849f3c95eaa81 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/misc/python/materialize/parallel_workload/action.py @@ -3104,6 +3104,7 @@ def __init__( self.flags_with_values["enable_column_paged_batcher_spill"] = ( BOOLEAN_FLAG_VALUES ) + self.flags_with_values["enable_packed_accumulable_diff"] = BOOLEAN_FLAG_VALUES # Fractions of the *cgroup* memory limit, which under mzcompose's # process orchestrator is the whole container budget shared by # environmentd and every replica, not one replica's allowance. A diff --git a/src/compute-types/src/dyncfgs.rs b/src/compute-types/src/dyncfgs.rs index 9d2c8ce0e119a..6bf4bdf58720f 100644 --- a/src/compute-types/src/dyncfgs.rs +++ b/src/compute-types/src/dyncfgs.rs @@ -44,6 +44,20 @@ pub const ENABLE_ERROR_DISTINCT: Config = Config::new( ParameterScope::Environment, ); +/// Select the packed byte layout for the accumulable reduce's arrangement diff. +/// +/// The accumulators are stored at their natural widths instead of at the size of +/// the widest enum variant, so a group's diff shrinks to the sum of its slot +/// sizes. Read at operator construction time, so flips take effect on dataflows +/// created after the change. +pub const ENABLE_PACKED_ACCUMULABLE_DIFF: Config = Config::new( + "enable_packed_accumulable_diff", + false, + "Select the packed byte layout for the accumulable reduce's arrangement diff, instead of a \ + vector of fixed-size accumulator enums.", + ParameterScope::Replica, +); + /// 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 @@ -702,6 +716,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet { configs .add(&ENABLE_HALF_JOIN2) .add(&ENABLE_ERROR_DISTINCT) + .add(&ENABLE_PACKED_ACCUMULABLE_DIFF) .add(&ENABLE_MZ_JOIN_CORE) .add(&ENABLE_SYNC_MV_SINK) .add(&ENABLE_CORRECTION_V2) diff --git a/src/compute/src/render/reduce.rs b/src/compute/src/render/reduce.rs index a5e8afcbb36fa..06c00bf66faea 100644 --- a/src/compute/src/render/reduce.rs +++ b/src/compute/src/render/reduce.rs @@ -13,7 +13,7 @@ use std::collections::BTreeMap; -use columnation::{Columnation, CopyRegion}; +use columnation::{Columnation, CopyRegion, Region}; use dec::OrderedDecimal; use differential_dataflow::Diff as _; use differential_dataflow::collection::AsCollection; @@ -24,9 +24,11 @@ use differential_dataflow::operators::arrange::{Arranged, TraceAgent}; use differential_dataflow::trace::cursor::{BatchCursor, BatchDiff, BatchValOwn}; use differential_dataflow::trace::implementations::BatchContainer; use differential_dataflow::trace::{Builder, Cursor, Navigable, Trace}; -use differential_dataflow::{Data, VecCollection}; +use differential_dataflow::{Data, ExchangeData, VecCollection}; use itertools::Itertools; -use mz_compute_types::dyncfgs::{ENABLE_COMPUTE_TEMPORAL_BUCKETING, TEMPORAL_BUCKETING_SUMMARY}; +use mz_compute_types::dyncfgs::{ + ENABLE_COMPUTE_TEMPORAL_BUCKETING, ENABLE_PACKED_ACCUMULABLE_DIFF, TEMPORAL_BUCKETING_SUMMARY, +}; use mz_compute_types::plan::ArrangementStrategy; use mz_compute_types::plan::reduce::{ AccumulablePlan, BasicPlan, BucketedPlan, HierarchicalPlan, KeyValPlan, LirAggregateExpr, @@ -35,7 +37,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_AGG_WIDTH_USIZE, Numeric, NumericAgg}; use mz_repr::fixed_length::ExtendDatums; use mz_repr::{Datum, DatumVec, Diff, Row, RowArena, SharedRow}; use mz_timely_util::columnation::ColumnationChunker; @@ -1357,50 +1359,63 @@ impl<'scope, T: RenderTimestamp> Context<'scope, T> { fn build_accumulable<'s>( &self, collection: VecCollection<'s, T, (Row, Row), Diff>, - AccumulablePlan { - full_aggrs, - simple_aggrs, - distinct_aggrs, - }: AccumulablePlan, + plan: AccumulablePlan, key_arity: usize, mfp_after: Option>, ) -> ( RowRowArrangement<'s, T>, VecCollection<'s, T, DataflowErrorSer, Diff>, ) { - let collection_scope = collection.scope(); - // we must have called this function with something to reduce - if full_aggrs.len() == 0 || simple_aggrs.len() + distinct_aggrs.len() != full_aggrs.len() { + if plan.full_aggrs.len() == 0 + || plan.simple_aggrs.len() + plan.distinct_aggrs.len() != plan.full_aggrs.len() + { self.error_logger().soft_panic_or_log( "Incorrect numbers of aggregates in accummulable reduction rendering", &format!( "full_aggrs={}, simple_aggrs={}, distinct_aggrs={}", - full_aggrs.len(), - simple_aggrs.len(), - distinct_aggrs.len(), + plan.full_aggrs.len(), + plan.simple_aggrs.len(), + plan.distinct_aggrs.len(), ), ); } + if ENABLE_PACKED_ACCUMULABLE_DIFF.get(&self.config_set) { + self.build_accumulable_with::(collection, plan, key_arity, mfp_after) + } else { + self.build_accumulable_with::>(collection, plan, key_arity, mfp_after) + } + } + + /// Renders [`Self::build_accumulable`] over the arrangement diff type `D`. + fn build_accumulable_with<'s, D: AccumulableDiff>( + &self, + collection: VecCollection<'s, T, (Row, Row), Diff>, + AccumulablePlan { + full_aggrs, + simple_aggrs, + distinct_aggrs, + }: AccumulablePlan, + key_arity: usize, + mfp_after: Option>, + ) -> ( + RowRowArrangement<'s, T>, + VecCollection<'s, T, DataflowErrorSer, Diff>, + ) { + let collection_scope = collection.scope(); + // Some of the aggregations may have the `distinct` bit set, which means that they'll // need to be extracted from `collection` and be subjected to `distinct` with `key`. // Other aggregations can be directly moved in to the `diff` field. // - // In each case, the resulting collection should have `data` shaped as `(key, ())` - // and a `diff` that is a vector with length `3 * aggrs.len()`. The three values are - // generally the count, and then two aggregation-specific values. The size could be - // reduced if we want to specialize for the aggregations. + // In each case, the resulting collection has `data` shaped as `(key, ())` + // and a `diff` holding one accumulator per aggregate plus the record count. - // Instantiate a default vector for diffs with the correct types at each + // Instantiate a default set of diffs with the correct types at each // position. - let zero_diffs: (Vec<_>, Diff) = ( - full_aggrs - .iter() - .map(|f| accumulable_zero(&f.func)) - .collect(), - Diff::ZERO, - ); + let zero_diffs: (D, Diff) = (D::zero(&full_aggrs), Diff::ZERO); + let layout = zero_diffs.0.layout(); let mut to_aggregate = Vec::new(); if simple_aggrs.len() > 0 { @@ -1408,6 +1423,7 @@ impl<'scope, T: RenderTimestamp> Context<'scope, T> { let collection = collection.clone(); let easy_cases = collection.explode_one({ let zero_diffs = zero_diffs.clone(); + let layout = layout.clone(); move |(key, row)| { let mut diffs = zero_diffs.clone(); // Try to unpack only the datums we need. Unfortunately, since we @@ -1422,7 +1438,11 @@ impl<'scope, T: RenderTimestamp> Context<'scope, T> { datum = row_iter.next().unwrap(); } let datum = datum.1; - diffs.0[*datum_index] = datum_to_accumulator(&aggr.func, datum); + diffs.0.set( + &layout, + *datum_index, + datum_to_accumulator(&aggr.func, datum), + ); diffs.1 = Diff::ONE; } ((key, ()), diffs) @@ -1455,10 +1475,15 @@ impl<'scope, T: RenderTimestamp> Context<'scope, T> { .as_collection(move |key_val_iter, _| pairer.split(key_val_iter)) .explode_one({ let zero_diffs = zero_diffs.clone(); + let layout = layout.clone(); move |(key, row)| { let datum = row.iter().next().unwrap(); let mut diffs = zero_diffs.clone(); - diffs.0[datum_index] = datum_to_accumulator(&aggr.func, datum); + diffs.0.set( + &layout, + datum_index, + datum_to_accumulator(&aggr.func, datum), + ); diffs.1 = Diff::ONE; ((key, ()), diffs) } @@ -1479,6 +1504,8 @@ impl<'scope, T: RenderTimestamp> Context<'scope, T> { let mfp_after1 = mfp_after.clone(); let mfp_after2 = mfp_after.filter(|mfp| mfp.could_error()); let full_aggrs2 = full_aggrs.clone(); + let mut accums1 = Vec::new(); + let mut accums2 = Vec::new(); let error_logger = self.error_logger(); let err_full_aggrs = full_aggrs.clone(); @@ -1487,7 +1514,7 @@ impl<'scope, T: RenderTimestamp> Context<'scope, T> { ColumnationChunker<_>, RowBatcher<_, _>, RowBuilder<_, _>, - RowSpine<_, (Vec, Diff)>, + RowSpine<_, (D, Diff)>, >( "ArrangeAccumulable [val: empty]", ); @@ -1498,6 +1525,7 @@ impl<'scope, T: RenderTimestamp> Context<'scope, T> { { move |key, input, output| { let (ref accums, total) = input[0].1; + let accums = accums.as_accums(&mut accums1); let temp_storage = RowArena::new(); let mut datums_local = datums1.borrow(); @@ -1523,6 +1551,7 @@ impl<'scope, T: RenderTimestamp> Context<'scope, T> { "AccumulableErrorCheck", move |key, input, output| { let (ref accums, total) = input[0].1; + let accums = accums.as_accums(&mut accums2); for (aggr, accum) in err_full_aggrs.iter().zip_eq(accums) { // We first test here if inputs without net-positive records are present, // producing an error to the logs and to the query output if that is the case. @@ -2269,6 +2298,461 @@ impl Columnation for Accum { type InnerRegion = CopyRegion; } +/// A packed sequence of [`Accum`]s, one per aggregate. +/// +/// Slots sit back to back, each a one-byte tag naming the [`Accum`] variant +/// followed by that variant's fixed-size payload: +/// +/// | tag | variant | payload fields | payload bytes | +/// |-----|---------|----------------|---------------| +/// | 0 | `Bool` | `trues`, `falses` | 16 | +/// | 1 | `SimpleNumber` | `accum: i128`, `non_nulls` | 24 | +/// | 2 | `Float` | `accum: i128`, `pos_infs`, `neg_infs`, `nans`, `non_nulls` | 48 | +/// | 3 | `Numeric` | `digits: u32`, `exponent: i32`, `bits: u8`, `lsu: [u16; NUMERIC_AGG_WIDTH]`, `pos_infs`, `neg_infs`, `nans`, `non_nulls` | 95 | +/// +/// Counters are the `i64` inside a [`Diff`], the `Numeric` fields are +/// [`dec::Decimal::to_raw_parts`], and every integer is little-endian and read +/// unaligned. Payloads are written field by field rather than as struct bytes, +/// so no padding enters the encoding. +/// +/// The layout is self-delimiting and carries no count header: a decoder reads a +/// tag, consumes that tag's payload, and stops when the bytes run out. +#[derive( + Clone, + Debug, + Default, + PartialEq, + Eq, + PartialOrd, + Ord, + Serialize, + Deserialize +)] +struct Accums(Vec); + +impl Accums { + const TAG_BOOL: u8 = 0; + const TAG_SIMPLE_NUMBER: u8 = 1; + const TAG_FLOAT: u8 = 2; + const TAG_NUMERIC: u8 = 3; + + /// Packs `accums` into slots, in order. + fn pack(accums: impl IntoIterator) -> Accums { + let mut bytes = Vec::new(); + for accum in accums { + let start = bytes.len(); + bytes.resize(start + Self::slot_len(Self::tag(&accum)), 0); + Self::write_slot(&mut bytes[start..], &accum); + } + Accums(bytes) + } + + /// Decodes the slots, in order. + fn decode(&self) -> impl Iterator + '_ { + let mut offset = 0; + std::iter::from_fn(move || { + if offset >= self.0.len() { + return None; + } + let len = Self::slot_len(self.0[offset]); + let accum = Self::decode_slot(&self.0[offset..]); + offset += len; + Some(accum) + }) + } + + /// Byte offset of every slot, in order. + fn slot_offsets(&self) -> Vec { + let mut offsets = Vec::new(); + let mut offset = 0; + while offset < self.0.len() { + offsets.push(offset); + offset += Self::slot_len(self.0[offset]); + } + offsets + } + + /// The tag naming `accum`'s variant. + fn tag(accum: &Accum) -> u8 { + match accum { + Accum::Bool { .. } => Self::TAG_BOOL, + Accum::SimpleNumber { .. } => Self::TAG_SIMPLE_NUMBER, + Accum::Float { .. } => Self::TAG_FLOAT, + Accum::Numeric { .. } => Self::TAG_NUMERIC, + } + } + + /// Encoded length of a slot with tag `tag`, tag byte included. + fn slot_len(tag: u8) -> usize { + let payload = match tag { + Self::TAG_BOOL => 2 * 8, + Self::TAG_SIMPLE_NUMBER => 16 + 8, + Self::TAG_FLOAT => 16 + 4 * 8, + Self::TAG_NUMERIC => 4 + 4 + 1 + 2 * NUMERIC_AGG_WIDTH_USIZE + 4 * 8, + tag => panic!("invalid Accums slot tag: {tag}"), + }; + 1 + payload + } + + /// Encodes `accum` into `dst`, which must be at least the slot's length. + fn write_slot(dst: &mut [u8], accum: &Accum) { + let mut w = SlotWriter { dst, pos: 0 }; + match accum { + Accum::Bool { trues, falses } => { + w.u8(Self::TAG_BOOL); + w.diff(*trues); + w.diff(*falses); + } + Accum::SimpleNumber { accum, non_nulls } => { + w.u8(Self::TAG_SIMPLE_NUMBER); + w.count(*accum); + w.diff(*non_nulls); + } + Accum::Float { + accum, + pos_infs, + neg_infs, + nans, + non_nulls, + } => { + w.u8(Self::TAG_FLOAT); + w.count(*accum); + w.diff(*pos_infs); + w.diff(*neg_infs); + w.diff(*nans); + w.diff(*non_nulls); + } + Accum::Numeric { + accum, + pos_infs, + neg_infs, + nans, + non_nulls, + } => { + let (digits, exponent, bits, lsu) = accum.0.to_raw_parts(); + w.u8(Self::TAG_NUMERIC); + w.u32(digits); + w.i32(exponent); + w.u8(bits); + w.lsu(&lsu); + w.diff(*pos_infs); + w.diff(*neg_infs); + w.diff(*nans); + w.diff(*non_nulls); + } + } + debug_assert_eq!(w.pos, Self::slot_len(Self::tag(accum))); + } + + /// Decodes the slot at the start of `bytes`. + fn decode_slot(bytes: &[u8]) -> Accum { + let tag = bytes[0]; + // Bound the reader to this slot so that a field list that disagrees + // with `slot_len` panics instead of reading into the next slot. + let mut r = SlotReader { + src: &bytes[..Self::slot_len(tag)], + pos: 1, + }; + let accum = match tag { + Self::TAG_BOOL => Accum::Bool { + trues: r.diff(), + falses: r.diff(), + }, + Self::TAG_SIMPLE_NUMBER => Accum::SimpleNumber { + accum: r.count(), + non_nulls: r.diff(), + }, + Self::TAG_FLOAT => Accum::Float { + accum: r.count(), + pos_infs: r.diff(), + neg_infs: r.diff(), + nans: r.diff(), + non_nulls: r.diff(), + }, + Self::TAG_NUMERIC => { + let digits = r.u32(); + let exponent = r.i32(); + let bits = r.u8(); + let lsu = r.lsu(); + Accum::Numeric { + accum: OrderedDecimal(NumericAgg::from_raw_parts(digits, exponent, bits, lsu)), + pos_infs: r.diff(), + neg_infs: r.diff(), + nans: r.diff(), + non_nulls: r.diff(), + } + } + tag => panic!("invalid Accums slot tag: {tag}"), + }; + debug_assert_eq!(r.pos, Self::slot_len(tag)); + accum + } +} + +impl IsZero for Accums { + fn is_zero(&self) -> bool { + self.decode().all(|accum| accum.is_zero()) + } +} + +impl Semigroup for Accums { + fn plus_equals(&mut self, rhs: &Self) { + // An empty value stands in for the identity, matching differential's + // `Semigroup` and `IsZero` impls for `Vec`: an empty vector is zero, + // and adding to it adopts the other side. + if self.0.is_empty() { + self.0.clone_from(&rhs.0); + return; + } + if rhs.0.is_empty() { + return; + } + + let mut offset = 0; + while offset < self.0.len() { + let tag = self.0[offset]; + let len = Self::slot_len(tag); + // NOTE: both sides descend from the same plan's aggregate list, so + // their slots line up. Misalignment means a diff built elsewhere + // reached this arrangement, and adding through it would silently + // corrupt the accumulation. + let aligned = rhs.0.len() >= offset + len && rhs.0[offset] == tag; + if !aligned { + mz_ore::soft_panic_or_log!("Accums slot mismatch at offset {offset}"); + return; + } + + let mut accum = Self::decode_slot(&self.0[offset..]); + accum.plus_equals(&Self::decode_slot(&rhs.0[offset..])); + Self::write_slot(&mut self.0[offset..offset + len], &accum); + offset += len; + } + } +} + +impl Multiply for Accums { + type Output = Self; + + fn multiply(mut self, factor: &Diff) -> Self { + let mut offset = 0; + while offset < self.0.len() { + let len = Self::slot_len(self.0[offset]); + let accum = Self::decode_slot(&self.0[offset..]).multiply(factor); + Self::write_slot(&mut self.0[offset..offset + len], &accum); + offset += len; + } + self + } +} + +impl Columnation for Accums { + type InnerRegion = AccumsRegion; +} + +/// Region for [`Accums`], deferring to the region of its backing bytes. +#[derive(Default)] +struct AccumsRegion { + inner: as Columnation>::InnerRegion, +} + +impl Region for AccumsRegion { + type Item = Accums; + + unsafe fn copy(&mut self, item: &Self::Item) -> Self::Item { + Accums(unsafe { self.inner.copy(&item.0) }) + } + + fn clear(&mut self) { + self.inner.clear(); + } + + fn reserve_items<'a, I>(&mut self, items: I) + where + Self: 'a, + I: Iterator + Clone, + { + self.inner.reserve_items(items.map(|accums| &accums.0)); + } + + fn reserve_regions<'a, I>(&mut self, regions: I) + where + Self: 'a, + I: Iterator + Clone, + { + self.inner.reserve_regions(regions.map(|r| &r.inner)); + } + + fn heap_size(&self, callback: impl FnMut(usize, usize)) { + self.inner.heap_size(callback); + } +} + +/// Cursor writing little-endian fields into one [`Accums`] slot. +struct SlotWriter<'a> { + dst: &'a mut [u8], + pos: usize, +} + +impl SlotWriter<'_> { + fn bytes(&mut self, bytes: &[u8]) { + self.dst[self.pos..self.pos + bytes.len()].copy_from_slice(bytes); + self.pos += bytes.len(); + } + + fn u8(&mut self, v: u8) { + self.bytes(&v.to_le_bytes()); + } + + fn u32(&mut self, v: u32) { + self.bytes(&v.to_le_bytes()); + } + + fn i32(&mut self, v: i32) { + self.bytes(&v.to_le_bytes()); + } + + fn diff(&mut self, v: Diff) { + self.bytes(&v.into_inner().to_le_bytes()); + } + + fn count(&mut self, v: AccumCount) { + self.bytes(&v.into_inner().to_le_bytes()); + } + + fn lsu(&mut self, lsu: &[u16; NUMERIC_AGG_WIDTH_USIZE]) { + for unit in lsu { + self.bytes(&unit.to_le_bytes()); + } + } +} + +/// Cursor reading little-endian fields from one [`Accums`] slot. +struct SlotReader<'a> { + src: &'a [u8], + pos: usize, +} + +impl SlotReader<'_> { + fn array(&mut self) -> [u8; N] { + let bytes = self.src[self.pos..self.pos + N].try_into().unwrap(); + self.pos += N; + bytes + } + + fn u8(&mut self) -> u8 { + u8::from_le_bytes(self.array()) + } + + fn u16(&mut self) -> u16 { + u16::from_le_bytes(self.array()) + } + + fn u32(&mut self) -> u32 { + u32::from_le_bytes(self.array()) + } + + fn i32(&mut self) -> i32 { + i32::from_le_bytes(self.array()) + } + + fn diff(&mut self) -> Diff { + Diff::from(i64::from_le_bytes(self.array())) + } + + fn count(&mut self) -> AccumCount { + AccumCount::from(i128::from_le_bytes(self.array())) + } + + fn lsu(&mut self) -> [u16; NUMERIC_AGG_WIDTH_USIZE] { + let mut lsu = [0; NUMERIC_AGG_WIDTH_USIZE]; + for unit in lsu.iter_mut() { + *unit = self.u16(); + } + lsu + } +} + +/// The diff type of the accumulable reduce's input arrangement, holding one +/// [`Accum`] per aggregate in the plan's `full_aggrs` order. +trait AccumulableDiff: + Semigroup + Multiply + Default + Ord + Columnation + ExchangeData +{ + /// Where each aggregate's accumulator sits within a value. + /// + /// Fixed by the plan's aggregate list, so it is computed once from + /// [`Self::zero`] and handed to every [`Self::set`] rather than + /// rediscovered per row. + type Layout: Clone + 'static; + + /// The accumulator identities for `aggrs`, in order. + fn zero(aggrs: &[LirAggregateExpr]) -> Self; + + /// The layout of `self` and of every value cloned from it. + fn layout(&self) -> Self::Layout; + + /// Overwrites the accumulator at `idx`, which must already hold the same + /// [`Accum`] variant as `accum`. `layout` must come from [`Self::layout`] + /// on this value or one it was cloned from. + fn set(&mut self, layout: &Self::Layout, idx: usize, accum: Accum); + + /// The accumulators, in aggregate order. + /// + /// `scratch` backs the returned slice for implementations that have to + /// materialize; whatever it held is discarded. + fn as_accums<'a>(&'a self, scratch: &'a mut Vec) -> &'a [Accum]; +} + +impl AccumulableDiff for Vec { + type Layout = (); + + fn zero(aggrs: &[LirAggregateExpr]) -> Self { + aggrs + .iter() + .map(|aggr| accumulable_zero(&aggr.func)) + .collect() + } + + fn layout(&self) -> Self::Layout {} + + fn set(&mut self, _layout: &Self::Layout, idx: usize, accum: Accum) { + self[idx] = accum; + } + + fn as_accums<'a>(&'a self, _scratch: &'a mut Vec) -> &'a [Accum] { + self + } +} + +impl AccumulableDiff for Accums { + type Layout = Vec; + + fn zero(aggrs: &[LirAggregateExpr]) -> Self { + Accums::pack(aggrs.iter().map(|aggr| accumulable_zero(&aggr.func))) + } + + fn layout(&self) -> Self::Layout { + self.slot_offsets() + } + + fn set(&mut self, layout: &Self::Layout, idx: usize, accum: Accum) { + let offset = layout[idx]; + let tag = Accums::tag(&accum); + // A tag change would resize the slot and shift every slot after it, so + // refuse the write rather than corrupt the layout. + if self.0[offset] != tag { + mz_ore::soft_panic_or_log!("Accums slot mismatch at index {idx}"); + return; + } + Accums::write_slot(&mut self.0[offset..offset + Accums::slot_len(tag)], &accum); + } + + fn as_accums<'a>(&'a self, scratch: &'a mut Vec) -> &'a [Accum] { + scratch.clear(); + scratch.extend(self.decode()); + scratch + } +} + /// Monoids for in-place compaction of monotonic streams. mod monoids { @@ -2631,95 +3115,4 @@ mod window_agg_helpers { } #[cfg(test)] -mod tests { - use super::*; - - /// The saturating conversion that `float_to_fixed_point` replaces. Used to - /// assert that the new wrapping conversion agrees on the in-range values - /// where the old conversion was already correct. - #[allow(clippy::as_conversions)] - fn saturating_convert(n: f64) -> i128 { - (n * FLOAT_SCALE) as i128 - } - - #[mz_ore::test] - fn float_to_fixed_point_matches_saturating_in_range() { - // For values whose scaled magnitude comfortably fits in an `i128`, the - // wrapping conversion must produce exactly the same result the previous - // saturating cast did. - let cases = [ - 0.0, - -0.0, - 1.0, - -1.0, - 0.1, - -0.1, - 0.5, - -0.5, - 3.25, - -3.25, - 123456.789, - -123456.789, - 1e10, - -1e10, - 1e20, - -1e20, - 5e30, // large, but scaled magnitude still fits comfortably in i128 - -5e30, - ]; - for n in cases { - assert_eq!( - float_to_fixed_point(n), - saturating_convert(n), - "mismatch for n = {n}" - ); - } - } - - #[mz_ore::test] - fn float_to_fixed_point_truncates_toward_zero() { - // 1.75 * 2^24 = 29360128, exactly representable. - assert_eq!(float_to_fixed_point(1.75), 29_360_128); - assert_eq!(float_to_fixed_point(-1.75), -29_360_128); - - // Fractional results truncate toward zero, matching the previous cast. - let frac = 0.123_456_7_f64; - assert_eq!(float_to_fixed_point(frac), saturating_convert(frac)); - assert_eq!(float_to_fixed_point(-frac), saturating_convert(-frac)); - assert_eq!(float_to_fixed_point(-frac), -float_to_fixed_point(frac)); - } - - #[mz_ore::test] - fn float_to_fixed_point_subnormals_round_to_zero() { - assert_eq!(float_to_fixed_point(0.0), 0); - assert_eq!(float_to_fixed_point(-0.0), 0); - assert_eq!(float_to_fixed_point(f64::MIN_POSITIVE / 2.0), 0); - assert_eq!(float_to_fixed_point(5e-324), 0); // smallest subnormal - } - - #[mz_ore::test] - fn float_to_fixed_point_cancels_large_finite_values() { - // Regression test for database-issues#11265: large finite values that - // individually overflow the fixed-point domain must still sum to the - // correct result when their mathematical sum is representable. The - // previous saturating conversion produced `i128::MAX + i128::MIN == -1`. - for &n in &[1.1e31_f64, 1e32, 5e33, 1e284] { - assert_eq!( - float_to_fixed_point(n).wrapping_add(float_to_fixed_point(-n)), - 0, - "n = {n} did not cancel with -n" - ); - } - } - - #[mz_ore::test] - fn float_to_fixed_point_sum_via_accumulator() { - // Exercise the full accumulate-then-finalize path for the reported case. - let func = AggregateFunc::SumFloat64; - let mut acc = accumulable_zero(&func); - acc.plus_equals(&datum_to_accumulator(&func, Datum::from(1.1e31_f64))); - acc.plus_equals(&datum_to_accumulator(&func, Datum::from(-1.1e31_f64))); - let datum = finalize_accum(&func, &acc, Diff::from(2_i64)); - assert_eq!(datum, Datum::from(0.0_f64)); - } -} +mod tests; diff --git a/src/compute/src/render/reduce/tests.rs b/src/compute/src/render/reduce/tests.rs new file mode 100644 index 0000000000000..ebdc06eff1974 --- /dev/null +++ b/src/compute/src/render/reduce/tests.rs @@ -0,0 +1,505 @@ +// 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. + +//! Tests for the accumulable reduce's accumulators: the fixed-point conversion +//! float sums accumulate through, and the packed [`Accums`] encoding, which is +//! held to the same results as a `Vec` under the same operations. + +use columnation::ColumnStack; +use rand::rngs::SmallRng; +use rand::{Rng, SeedableRng}; + +use super::*; + +/// The saturating conversion that `float_to_fixed_point` replaces. Used to +/// assert that the new wrapping conversion agrees on the in-range values +/// where the old conversion was already correct. +#[allow(clippy::as_conversions)] +fn saturating_convert(n: f64) -> i128 { + (n * FLOAT_SCALE) as i128 +} + +#[mz_ore::test] +fn float_to_fixed_point_matches_saturating_in_range() { + // For values whose scaled magnitude comfortably fits in an `i128`, the + // wrapping conversion must produce exactly the same result the previous + // saturating cast did. + let cases = [ + 0.0, + -0.0, + 1.0, + -1.0, + 0.1, + -0.1, + 0.5, + -0.5, + 3.25, + -3.25, + 123456.789, + -123456.789, + 1e10, + -1e10, + 1e20, + -1e20, + 5e30, // large, but scaled magnitude still fits comfortably in i128 + -5e30, + ]; + for n in cases { + assert_eq!( + float_to_fixed_point(n), + saturating_convert(n), + "mismatch for n = {n}" + ); + } +} + +#[mz_ore::test] +fn float_to_fixed_point_truncates_toward_zero() { + // 1.75 * 2^24 = 29360128, exactly representable. + assert_eq!(float_to_fixed_point(1.75), 29_360_128); + assert_eq!(float_to_fixed_point(-1.75), -29_360_128); + + // Fractional results truncate toward zero, matching the previous cast. + let frac = 0.123_456_7_f64; + assert_eq!(float_to_fixed_point(frac), saturating_convert(frac)); + assert_eq!(float_to_fixed_point(-frac), saturating_convert(-frac)); + assert_eq!(float_to_fixed_point(-frac), -float_to_fixed_point(frac)); +} + +#[mz_ore::test] +fn float_to_fixed_point_subnormals_round_to_zero() { + assert_eq!(float_to_fixed_point(0.0), 0); + assert_eq!(float_to_fixed_point(-0.0), 0); + assert_eq!(float_to_fixed_point(f64::MIN_POSITIVE / 2.0), 0); + assert_eq!(float_to_fixed_point(5e-324), 0); // smallest subnormal +} + +#[mz_ore::test] +fn float_to_fixed_point_cancels_large_finite_values() { + // Regression test for database-issues#11265: large finite values that + // individually overflow the fixed-point domain must still sum to the + // correct result when their mathematical sum is representable. The + // previous saturating conversion produced `i128::MAX + i128::MIN == -1`. + for &n in &[1.1e31_f64, 1e32, 5e33, 1e284] { + assert_eq!( + float_to_fixed_point(n).wrapping_add(float_to_fixed_point(-n)), + 0, + "n = {n} did not cancel with -n" + ); + } +} + +#[mz_ore::test] +fn float_to_fixed_point_sum_via_accumulator() { + // Exercise the full accumulate-then-finalize path for the reported case. + let func = AggregateFunc::SumFloat64; + let mut acc = accumulable_zero(&func); + acc.plus_equals(&datum_to_accumulator(&func, Datum::from(1.1e31_f64))); + acc.plus_equals(&datum_to_accumulator(&func, Datum::from(-1.1e31_f64))); + let datum = finalize_accum(&func, &acc, Diff::from(2_i64)); + assert_eq!(datum, Datum::from(0.0_f64)); +} + +fn diff(v: i64) -> Diff { + Diff::from(v) +} + +fn count(v: i128) -> AccumCount { + AccumCount::from(v) +} + +fn numeric_agg(s: &str) -> OrderedDecimal { + OrderedDecimal(numeric::cx_agg().parse(s).unwrap()) +} + +fn numeric_datum(s: &str) -> Datum<'static> { + Datum::from(numeric::cx_datum().parse(s).unwrap()) +} + +fn aggr(func: AggregateFunc) -> LirAggregateExpr { + LirAggregateExpr { + func, + expr: LirScalarExpr::column(0), + distinct: false, + } +} + +/// Every aggregate function `reduction_type` routes to the accumulable +/// reduce, which is exactly the set `Accums` has to encode. +fn accumulable_funcs() -> Vec { + let funcs = vec![ + AggregateFunc::SumInt16, + AggregateFunc::SumInt32, + AggregateFunc::SumInt64, + AggregateFunc::SumUInt16, + AggregateFunc::SumUInt32, + AggregateFunc::SumUInt64, + AggregateFunc::SumFloat32, + AggregateFunc::SumFloat64, + AggregateFunc::SumNumeric, + AggregateFunc::Count, + AggregateFunc::Any, + AggregateFunc::All, + AggregateFunc::Dummy, + ]; + for func in &funcs { + assert_eq!( + reduction_type(func), + ReductionType::Accumulable, + "func={func:?}" + ); + } + funcs +} + +/// A random input datum for `func`, drawn from the domain +/// `datum_to_accumulator` accepts for it, including `Null` and the special +/// float and numeric values. +/// +/// Magnitudes stay small so that no sequence of `plus_equals` and +/// `multiply` can overflow an accumulator. Overflow is not a difference +/// between the two representations (both defer to [`Accum`]), but +/// `Accum::Numeric` panics on it, which would just make the test flaky. +fn random_datum(func: &AggregateFunc, rng: &mut SmallRng) -> Datum<'static> { + if *func == AggregateFunc::Dummy { + return Datum::Dummy; + } + if rng.random_bool(0.15) { + return Datum::Null; + } + match func { + AggregateFunc::SumInt16 => Datum::Int16(rng.random_range(-1000..=1000)), + AggregateFunc::SumInt32 => Datum::Int32(rng.random_range(-1000..=1000)), + AggregateFunc::SumInt64 => Datum::Int64(rng.random_range(-1000..=1000)), + AggregateFunc::SumUInt16 => Datum::UInt16(rng.random_range(0..=1000)), + AggregateFunc::SumUInt32 => Datum::UInt32(rng.random_range(0..=1000)), + AggregateFunc::SumUInt64 => Datum::UInt64(rng.random_range(0..=1000)), + AggregateFunc::SumFloat32 => Datum::from(match rng.random_range(0..6) { + 0 => f32::NAN, + 1 => f32::INFINITY, + 2 => f32::NEG_INFINITY, + _ => rng.random_range(-1000.0..1000.0_f32), + }), + AggregateFunc::SumFloat64 => Datum::from(match rng.random_range(0..6) { + 0 => f64::NAN, + 1 => f64::INFINITY, + 2 => f64::NEG_INFINITY, + _ => rng.random_range(-1000.0..1000.0_f64), + }), + AggregateFunc::SumNumeric => match rng.random_range(0..6) { + 0 => numeric_datum("nan"), + 1 => numeric_datum("inf"), + 2 => numeric_datum("-inf"), + 3 => numeric_datum("0.00000001234"), + _ => Datum::from(Numeric::from(rng.random_range(-1000..=1000))), + }, + AggregateFunc::Count => Datum::Int32(rng.random_range(-1000..=1000)), + AggregateFunc::Any | AggregateFunc::All => { + if rng.random_bool(0.5) { + Datum::True + } else { + Datum::False + } + } + func => panic!("not an accumulable aggregate: {func:?}"), + } +} + +#[mz_ore::test] +fn accums_slot_sizes() { + // The slot sizes are the point of the packed layout, so pin them: a + // change here is a change to the arrangement's memory footprint. + let cases = [ + ( + Accum::Bool { + trues: diff(1), + falses: diff(1), + }, + 1 + 16, + ), + ( + Accum::SimpleNumber { + accum: count(1), + non_nulls: diff(1), + }, + 1 + 24, + ), + ( + Accum::Float { + accum: count(1), + pos_infs: diff(1), + neg_infs: diff(1), + nans: diff(1), + non_nulls: diff(1), + }, + 1 + 48, + ), + ( + Accum::Numeric { + accum: numeric_agg("1.5"), + pos_infs: diff(1), + neg_infs: diff(1), + nans: diff(1), + non_nulls: diff(1), + }, + 1 + 95, + ), + ]; + + for (accum, len) in cases { + assert_eq!(Accums::pack([accum]).0.len(), len, "accum={accum:?}"); + } + + let total: usize = cases.iter().map(|(_, len)| len).sum(); + let packed = Accums::pack(cases.map(|(accum, _)| accum)); + assert_eq!(packed.0.len(), total); + assert_eq!(packed.decode().count(), cases.len()); +} + +#[mz_ore::test] +fn accums_roundtrip() { + let cases = [ + Accum::Bool { + trues: diff(0), + falses: diff(0), + }, + Accum::Bool { + trues: diff(i64::MAX), + falses: diff(i64::MIN), + }, + Accum::Bool { + trues: diff(-7), + falses: diff(9), + }, + Accum::SimpleNumber { + accum: count(0), + non_nulls: diff(0), + }, + Accum::SimpleNumber { + accum: count(i128::MAX), + non_nulls: diff(i64::MAX), + }, + Accum::SimpleNumber { + accum: count(i128::MIN), + non_nulls: diff(i64::MIN), + }, + Accum::SimpleNumber { + accum: count(-1), + non_nulls: diff(-3), + }, + Accum::Float { + accum: count(0), + pos_infs: diff(0), + neg_infs: diff(0), + nans: diff(0), + non_nulls: diff(0), + }, + Accum::Float { + accum: count(i128::MAX), + pos_infs: diff(i64::MAX), + neg_infs: diff(i64::MIN), + nans: diff(-5), + non_nulls: diff(11), + }, + Accum::Float { + accum: count(i128::MIN), + pos_infs: diff(-1), + neg_infs: diff(1), + nans: diff(0), + non_nulls: diff(-1), + }, + Accum::Numeric { + accum: OrderedDecimal(NumericAgg::zero()), + pos_infs: diff(0), + neg_infs: diff(0), + nans: diff(0), + non_nulls: diff(0), + }, + Accum::Numeric { + accum: numeric_agg("123456789012345678901234567890123456789012345678901234567890"), + pos_infs: diff(3), + neg_infs: diff(-4), + nans: diff(7), + non_nulls: diff(-9), + }, + Accum::Numeric { + accum: numeric_agg("-9.999999999999999e-100"), + pos_infs: diff(i64::MIN), + neg_infs: diff(i64::MAX), + nans: diff(1), + non_nulls: diff(1), + }, + Accum::Numeric { + accum: numeric_agg("1e100"), + pos_infs: diff(0), + neg_infs: diff(0), + nans: diff(0), + non_nulls: diff(1), + }, + ]; + + for accum in cases { + let decoded: Vec<_> = Accums::pack([accum]).decode().collect(); + assert_eq!(decoded, vec![accum], "accum={accum:?}"); + } + + // The same values in one buffer, so slot walking has to find each + // boundary rather than reading a single slot. + let decoded: Vec<_> = Accums::pack(cases).decode().collect(); + assert_eq!(decoded, cases.to_vec()); +} + +#[mz_ore::test] +fn accums_region_roundtrip() { + // `AccumsRegion` is the only unsafe code in the encoding and every batch + // passes through it, so pin the copy, read-back, clear cycle. + let aggrs = [ + aggr(AggregateFunc::SumNumeric), + aggr(AggregateFunc::Count), + aggr(AggregateFunc::SumFloat64), + aggr(AggregateFunc::Any), + ]; + let zero = ::zero(&aggrs); + let layout = zero.layout(); + let mut filled = zero.clone(); + let datums = [ + numeric_datum("1.5"), + Datum::Int32(1), + Datum::from(2.5_f64), + Datum::True, + ]; + for (idx, (aggr, datum)) in aggrs.iter().zip(datums).enumerate() { + filled.set(&layout, idx, datum_to_accumulator(&aggr.func, datum)); + } + assert_ne!(filled, zero); + + let mut stack = ColumnStack::::default(); + stack.copy(&zero); + stack.copy(&filled); + assert_eq!(stack.len(), 2); + assert_eq!(stack[0], zero); + assert_eq!(stack[1], filled); + assert_eq!( + stack[1].decode().collect::>(), + filled.decode().collect::>() + ); + + let mut heap_bytes = 0; + stack.heap_size(|size, _capacity| heap_bytes += size); + assert!( + heap_bytes >= zero.0.len() + filled.0.len(), + "heap_size must account for the packed bytes, got {heap_bytes}" + ); + + stack.clear(); + assert_eq!(stack.len(), 0); + stack.copy(&filled); + assert_eq!(stack[0], filled); +} + +#[mz_ore::test] +fn accums_empty_is_identity() { + let aggrs = [aggr(AggregateFunc::SumInt64), aggr(AggregateFunc::Any)]; + let mut filled = ::zero(&aggrs); + let layout = filled.layout(); + filled.set( + &layout, + 0, + datum_to_accumulator(&AggregateFunc::SumInt64, Datum::Int64(7)), + ); + filled.set( + &layout, + 1, + datum_to_accumulator(&AggregateFunc::Any, Datum::True), + ); + + let mut lhs = Accums::default(); + assert!(lhs.is_zero()); + lhs.plus_equals(&filled); + assert_eq!(lhs, filled); + + let mut lhs = filled.clone(); + lhs.plus_equals(&Accums::default()); + assert_eq!(lhs, filled); + + assert_eq!( + Accums::default().multiply(&diff(3)), + Accums::default(), + "multiplying the identity yields the identity" + ); +} + +#[mz_ore::test] +fn accums_agree_with_vec_accum() { + let mut rng = SmallRng::seed_from_u64(0); + let funcs = accumulable_funcs(); + + // One layout per function isolates each variant; the wider layouts make + // slot walking cross variant boundaries. + let mut layouts: Vec> = + funcs.iter().cloned().map(|func| vec![func]).collect(); + layouts.push(funcs.clone()); + layouts.push(vec![ + AggregateFunc::SumNumeric, + AggregateFunc::Count, + AggregateFunc::SumFloat64, + AggregateFunc::Any, + AggregateFunc::SumInt32, + ]); + + // Reused across layouts and steps without clearing, so `as_accums` has to + // discard whatever it finds there. + let mut scratch = vec![Accum::Bool { + trues: diff(99), + falses: diff(99), + }]; + + for layout in layouts { + let aggrs: Vec<_> = layout.iter().cloned().map(aggr).collect(); + + let mut vec_state = as AccumulableDiff>::zero(&aggrs); + let mut packed_state = ::zero(&aggrs); + let vec_layout = vec_state.layout(); + let packed_layout = packed_state.layout(); + assert_eq!(packed_state.as_accums(&mut scratch), &vec_state[..]); + + // The `explode_one` path: overwrite each slot in place. + for (idx, func) in layout.iter().enumerate() { + let accum = datum_to_accumulator(func, random_datum(func, &mut rng)); + vec_state.set(&vec_layout, idx, accum); + packed_state.set(&packed_layout, idx, accum); + } + + for step in 0..64 { + if rng.random_bool(0.6) { + let rhs: Vec = layout + .iter() + .map(|func| datum_to_accumulator(func, random_datum(func, &mut rng))) + .collect(); + vec_state.plus_equals(&rhs); + packed_state.plus_equals(&Accums::pack(rhs)); + } else { + let factor = diff(rng.random_range(-3..=3)); + vec_state = vec_state.multiply(&factor); + packed_state = packed_state.multiply(&factor); + } + + assert_eq!( + packed_state.as_accums(&mut scratch), + &vec_state[..], + "layout={layout:?}, step={step}" + ); + assert_eq!( + packed_state.is_zero(), + vec_state.is_zero(), + "layout={layout:?}, step={step}" + ); + } + } +} diff --git a/test/launchdarkly-flag-consistency/mzcompose.py b/test/launchdarkly-flag-consistency/mzcompose.py index 860d76ab2ee0b..ac840f498ced7 100644 --- a/test/launchdarkly-flag-consistency/mzcompose.py +++ b/test/launchdarkly-flag-consistency/mzcompose.py @@ -269,6 +269,7 @@ enable_notices_for_index_already_exists enable_notices_for_index_empty_key enable_off_thread_optimization + enable_packed_accumulable_diff enable_password_auth enable_paused_cluster_readhold_downgrade enable_persist_streaming_compaction