From 664988f02f9b113654dcc4a7e49256a59e42b7db Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Tue, 25 Aug 2026 13:43:49 +0200 Subject: [PATCH] compute: select the arrange merge batcher by dyncfg Compute arrange sites could already choose between the columnation merge batcher and a columnar one, but only in the form of the paged batcher, so the columnation-versus-columnar comparison came entangled with the pager and its spill budget. Separate the two axes: `enable_columnar_merge_batcher` selects the resident columnar batcher, which holds the same `Column` chains and feeds the same builder as the paged arm but merges through `ColumnMerger` with no pager and no spill accounting. The three type parameters `mz_arrange_core` takes are one unit rather than three independent knobs, because the chunker's container and the builder's input are both pinned to `Batcher::Output`. `ArrangementBatcher` names the three resulting arms and resolves them from the config set in one place, so neither arrange site encodes the precedence rule. `enable_column_paged_batcher` keeps precedence: it asks for the same columnar chains and additionally routes them through the pager, so its existing meaning and rollout are unchanged. `Col2ValColBatcher` wraps `ColumnMerger`, which until now existed only for benchmarks. It reuses `RowRowColPagedBuilder`, whose `Input` is `Column` rather than `PagedColumn` and which therefore never depended on paging; that builder's documentation now says so, while the rename its name invites is left out to keep this change on the flag. The flag is off in production and on in the CI configuration so the columnar arm is exercised before it earns a production default, and parallel workload varies it alongside the paged flag. Co-Authored-By: Claude Opus 5 (1M context) --- misc/python/materialize/mzcompose/__init__.py | 5 +++ .../materialize/parallel_workload/action.py | 1 + src/compute-types/src/dyncfgs.rs | 39 +++++++++++++++--- src/compute/src/extensions/arrange.rs | 39 ++++++++++++++++++ src/compute/src/render/context.rs | 41 ++++++++++--------- src/compute/src/render/join/linear_join.rs | 28 ++++++++----- src/row-spine/src/lib.rs | 18 ++++---- src/timely-util/src/columnar.rs | 9 ++++ .../mzcompose.py | 1 + 9 files changed, 138 insertions(+), 43 deletions(-) diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index 8c11565eb5007..9ce47a13dc0a5 100644 --- a/misc/python/materialize/mzcompose/__init__.py +++ b/misc/python/materialize/mzcompose/__init__.py @@ -258,6 +258,11 @@ def get_variable_system_parameters( VariableSystemParameter( "compute_apply_column_demands", "true", ["true", "false"] ), + # On by default so CI exercises the columnar merge batcher, which is + # off in production while it earns trust. + VariableSystemParameter( + "enable_columnar_merge_batcher", "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 68485da625c3d..a64d4b3b3a094 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/misc/python/materialize/parallel_workload/action.py @@ -3024,6 +3024,7 @@ def __init__( self.flags_with_values["enable_coalesce_case_transform"] = BOOLEAN_FLAG_VALUES 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_column_paged_batcher_spill"] = ( BOOLEAN_FLAG_VALUES ) diff --git a/src/compute-types/src/dyncfgs.rs b/src/compute-types/src/dyncfgs.rs index 8b516d22ca5ab..2f7a2c7586549 100644 --- a/src/compute-types/src/dyncfgs.rs +++ b/src/compute-types/src/dyncfgs.rs @@ -48,10 +48,15 @@ pub const ENABLE_ERROR_DISTINCT: Config = Config::new( /// `true`, arrange operators use `Col2ValPagedBatcher` (in /// `mz_timely_util::columnar`) and `RowRowColPagedBuilder` (in /// `mz_row_spine`), the columnar-native batcher that the pager can spill -/// (gated by [`ENABLE_COLUMN_PAGED_BATCHER_SPILL`]). When `false` (the -/// default), the same arrange sites use the legacy `Col2ValBatcher` / -/// `RowRowBuilder` (columnation-merger) path. Read at operator construction -/// time. Flips take effect on dataflows created after the change. +/// (gated by [`ENABLE_COLUMN_PAGED_BATCHER_SPILL`]). Read at operator +/// construction time. Flips take effect on dataflows created after the +/// change. +/// +/// Takes precedence over [`ENABLE_COLUMNAR_MERGE_BATCHER`]: both select +/// columnar chains, and this one additionally routes them through the pager. +/// With both `false` the arrange sites use the columnation +/// `Col2ValBatcher` / `RowRowBuilder` path. See +/// `mz_compute::extensions::arrange::ArrangementBatcher` for the resolution. /// /// Disabled by default while the new path is stabilizing. /// `DifferentialJoinHydration*` feature-benchmark scenarios opt in @@ -59,8 +64,29 @@ pub const ENABLE_ERROR_DISTINCT: Config = Config::new( pub const ENABLE_COLUMN_PAGED_BATCHER: Config = Config::new( "enable_column_paged_batcher", false, - "Use the columnar-native paged merge batcher at arrange sites. When `false` (default), \ - arranges fall back to the legacy columnation `Col2ValBatcher` / `RowRowBuilder` path.", + "Use the columnar-native paged merge batcher at arrange sites. Takes precedence over \ + enable_columnar_merge_batcher; with both false, arranges use the columnation \ + `Col2ValBatcher` / `RowRowBuilder` path.", + ParameterScope::Replica, +); + +/// Use the resident columnar merge batcher at arrange sites. When `true`, +/// arrange operators use `Col2ValColBatcher` (in `mz_timely_util::columnar`) +/// and `RowRowColPagedBuilder` (in `mz_row_spine`): the same `Column` chains +/// and the same builder as the paged arm, merged by `ColumnMerger` with no +/// pager and no spill budget. When `false` (the default), the arrange sites +/// use the columnation `Col2ValBatcher` / `RowRowBuilder` path. Read at +/// operator construction time. Flips take effect on dataflows created after +/// the change. +/// +/// This is the columnation-versus-columnar axis on its own, so the two paths +/// can be compared without the pager in the measurement. It is ignored while +/// [`ENABLE_COLUMN_PAGED_BATCHER`] is `true`. +pub const ENABLE_COLUMNAR_MERGE_BATCHER: Config = Config::new( + "enable_columnar_merge_batcher", + false, + "Use the resident columnar merge batcher at arrange sites, instead of the columnation \ + one. Ignored when enable_column_paged_batcher is true.", ParameterScope::Replica, ); @@ -706,6 +732,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet { .add(&SUBSCRIBE_SNAPSHOT_OPTIMIZATION) .add(&MV_SINK_ADVANCE_PERSIST_FRONTIERS) .add(&ENABLE_COLUMN_PAGED_BATCHER) + .add(&ENABLE_COLUMNAR_MERGE_BATCHER) .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 5aa383e0f11fa..b06e1a56790b7 100644 --- a/src/compute/src/extensions/arrange.rs +++ b/src/compute/src/extensions/arrange.rs @@ -18,6 +18,8 @@ use differential_dataflow::operators::arrange::{Arranged, TraceAgent}; 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 timely::Container; use timely::container::{ContainerBuilder, PushInto}; @@ -34,6 +36,43 @@ use crate::typedefs::{ KeyAgent, KeyValAgent, MzArrangeData, MzData, MzTimestamp, RowAgent, RowRowAgent, RowValAgent, }; +/// Which merge batcher an arrange site should instantiate. +/// +/// The three parameters `mz_arrange_core` takes are one unit, not three +/// knobs: the chunker's container and the builder's input are both pinned to +/// `Batcher::Output`, so the chunker and builder follow from the batcher and +/// a call site has to spell out a whole arm per variant. +pub enum ArrangementBatcher { + /// `Chunker>` + `Col2ValBatcher` + `RowRowBuilder`. + /// Chains are columnation stacks. + Columnation, + /// `ColumnChunker` + `Col2ValColBatcher` + `RowRowColPagedBuilder`. + /// Chains are resident `Column`s. + Columnar, + /// `ColumnChunker` + `Col2ValPagedBatcher` + `RowRowColPagedBuilder`. + /// Chains are `Column`s routed through the pager, which may spill them. + ColumnarPaged, +} + +impl ArrangementBatcher { + /// Resolve the batcher from the replica's config set. + /// + /// `ENABLE_COLUMN_PAGED_BATCHER` wins over + /// `ENABLE_COLUMNAR_MERGE_BATCHER`, because it asks for the same columnar + /// chains plus paging. Call this once per arrange site at operator + /// construction time, so a dataflow keeps one batcher for its whole life + /// even if the flags flip underneath it. + pub fn from_config(config: &ConfigSet) -> Self { + if ENABLE_COLUMN_PAGED_BATCHER.get(config) { + Self::ColumnarPaged + } else if ENABLE_COLUMNAR_MERGE_BATCHER.get(config) { + Self::Columnar + } else { + Self::Columnation + } + } +} + /// Extension trait to arrange data. pub trait MzArrange<'scope>: MzArrangeCore<'scope> { /// Arranges a stream of `(Key, Val)` updates by `Key` into a trace of type `Tr`. diff --git a/src/compute/src/render/context.rs b/src/compute/src/render/context.rs index 2f65a61f0e025..e3a6351ec857e 100644 --- a/src/compute/src/render/context.rs +++ b/src/compute/src/render/context.rs @@ -21,8 +21,8 @@ use differential_dataflow::trace::{Cursor, Navigable, TraceReader}; use differential_dataflow::{AsCollection, Data, VecCollection}; use mz_compute_types::dataflows::DataflowDescription; use mz_compute_types::dyncfgs::{ - ENABLE_COLUMN_PAGED_BATCHER, ENABLE_COMPUTE_RENDER_FUELED_AS_SPECIFIC_COLLECTION, - ENABLE_COMPUTE_TEMPORAL_BUCKETING, TEMPORAL_BUCKETING_SUMMARY, + ENABLE_COMPUTE_RENDER_FUELED_AS_SPECIFIC_COLLECTION, ENABLE_COMPUTE_TEMPORAL_BUCKETING, + TEMPORAL_BUCKETING_SUMMARY, }; use mz_compute_types::plan::scalar::{LirScalarExpr, mfp_mir_to_lir_plan, mfp_plan_lir_to_mir}; use mz_compute_types::plan::{ArrangementStrategy, AvailableCollections}; @@ -34,7 +34,9 @@ use mz_repr::{DatumVec, DatumVecBorrow, Diff, GlobalId, Row, RowArena, SharedRow use mz_storage_types::controller::CollectionMetadata; use mz_timely_util::columnar::batcher; use mz_timely_util::columnar::builder::ColumnBuilder; -use mz_timely_util::columnar::{Col2ValBatcher, Col2ValPagedBatcher, columnar_exchange}; +use mz_timely_util::columnar::{ + Col2ValBatcher, Col2ValColBatcher, Col2ValPagedBatcher, columnar_exchange, +}; use mz_timely_util::columnation::ColumnationChunker; use timely::ContainerBuilder; use timely::container::{CapacityContainerBuilder, PushInto}; @@ -47,7 +49,7 @@ use timely::progress::operate::FrontierInterest; use timely::progress::{Antichain, Timestamp}; use crate::compute_state::ComputeState; -use crate::extensions::arrange::{KeyCollection, MzArrange, MzArrangeCore}; +use crate::extensions::arrange::{ArrangementBatcher, KeyCollection, MzArrange, MzArrangeCore}; use crate::extensions::reduce::MzReduce; use crate::render::columnar::CollectionEdge; use crate::render::errors::{DataflowErrorSer, ErrorLogger}; @@ -1173,14 +1175,9 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> { } else { oks }; - let use_paged_path = ENABLE_COLUMN_PAGED_BATCHER.get(config_set); - let (oks, errs_keyed, passthrough) = Self::arrange_collection( - &name, - oks, - key.clone(), - thinning.clone(), - use_paged_path, - ); + let batcher = ArrangementBatcher::from_config(config_set); + let (oks, errs_keyed, passthrough) = + Self::arrange_collection(&name, oks, key.clone(), thinning.clone(), batcher); let errs_concat: KeyCollection<_, _, _> = errs.clone().concat(errs_keyed).into(); self.collection = Some((CollectionEdge::Vec(passthrough), errs)); let errs = @@ -1214,7 +1211,7 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> { oks: VecCollection<'scope, T, Row, Diff>, key: Vec, thinning: Vec, - use_paged_path: bool, + batcher: ArrangementBatcher, ) -> ( Arranged<'scope, RowRowAgent>, VecCollection<'scope, T, DataflowErrorSer, Diff>, @@ -1270,22 +1267,28 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> { let exchange = ExchangeCore::, _>::new_core(columnar_exchange::); - let oks = if use_paged_path { - ok_stream.mz_arrange_core::< + let oks = match batcher { + ArrangementBatcher::ColumnarPaged => ok_stream.mz_arrange_core::< _, batcher::ColumnChunker<_>, Col2ValPagedBatcher<_, _, _, _>, RowRowColPagedBuilder<_, _>, RowRowSpine<_, _>, - >(exchange, name) - } else { - ok_stream.mz_arrange_core::< + >(exchange, name), + ArrangementBatcher::Columnar => ok_stream.mz_arrange_core::< + _, + batcher::ColumnChunker<_>, + Col2ValColBatcher<_, _, _, _>, + RowRowColPagedBuilder<_, _>, + RowRowSpine<_, _>, + >(exchange, name), + ArrangementBatcher::Columnation => ok_stream.mz_arrange_core::< _, batcher::Chunker<_>, Col2ValBatcher<_, _, _, _>, RowRowBuilder<_, _>, RowRowSpine<_, _>, - >(exchange, name) + >(exchange, name), }; ( oks, diff --git a/src/compute/src/render/join/linear_join.rs b/src/compute/src/render/join/linear_join.rs index 13d6d910d0d91..84de916c96c47 100644 --- a/src/compute/src/render/join/linear_join.rs +++ b/src/compute/src/render/join/linear_join.rs @@ -19,9 +19,7 @@ use differential_dataflow::operators::arrange::arrangement::Arranged; use differential_dataflow::trace::cursor::{BatchCursor, BatchKey, BatchVal}; use differential_dataflow::trace::{Cursor, Navigable, TraceReader}; use differential_dataflow::{AsCollection, Data, VecCollection}; -use mz_compute_types::dyncfgs::{ - ENABLE_COLUMN_PAGED_BATCHER, ENABLE_MZ_JOIN_CORE, LINEAR_JOIN_YIELDING, -}; +use mz_compute_types::dyncfgs::{ENABLE_MZ_JOIN_CORE, LINEAR_JOIN_YIELDING}; use mz_compute_types::plan::join::JoinClosure; use mz_compute_types::plan::join::linear_join::{LinearJoinPlan, LinearStagePlan}; use mz_dyncfg::ConfigSet; @@ -30,13 +28,15 @@ use mz_repr::fixed_length::ExtendDatums; use mz_repr::{DatumVec, Diff, Row, RowArena, SharedRow}; use mz_timely_util::columnar::batcher; use mz_timely_util::columnar::builder::ColumnBuilder; -use mz_timely_util::columnar::{Col2ValBatcher, Col2ValPagedBatcher, columnar_exchange}; +use mz_timely_util::columnar::{ + Col2ValBatcher, Col2ValColBatcher, Col2ValPagedBatcher, columnar_exchange, +}; use mz_timely_util::operator::{CollectionExt, StreamExt}; use timely::dataflow::Scope; use timely::dataflow::channels::pact::{ExchangeCore, Pipeline}; use timely::dataflow::operators::OkErr; -use crate::extensions::arrange::MzArrangeCore; +use crate::extensions::arrange::{ArrangementBatcher, MzArrangeCore}; use crate::render::RenderTimestamp; use crate::render::context::{ArrangementFlavor, CollectionBundle, Context}; use crate::render::errors::DataflowErrorSer; @@ -390,22 +390,28 @@ where let exchange = ExchangeCore::, _>::new_core( columnar_exchange::, ); - let arranged = if ENABLE_COLUMN_PAGED_BATCHER.get(&self.config_set) { - keyed.mz_arrange_core::< + let arranged = match ArrangementBatcher::from_config(&self.config_set) { + ArrangementBatcher::ColumnarPaged => keyed.mz_arrange_core::< _, batcher::ColumnChunker<_>, Col2ValPagedBatcher<_, _, _, _>, RowRowColPagedBuilder<_, _>, RowRowSpine<_, _>, - >(exchange, "JoinStage") - } else { - keyed.mz_arrange_core::< + >(exchange, "JoinStage"), + ArrangementBatcher::Columnar => keyed.mz_arrange_core::< + _, + batcher::ColumnChunker<_>, + Col2ValColBatcher<_, _, _, _>, + RowRowColPagedBuilder<_, _>, + RowRowSpine<_, _>, + >(exchange, "JoinStage"), + ArrangementBatcher::Columnation => keyed.mz_arrange_core::< _, batcher::Chunker<_>, Col2ValBatcher<_, _, _, _>, RowRowBuilder<_, _>, RowRowSpine<_, _>, - >(exchange, "JoinStage") + >(exchange, "JoinStage"), }; joined = JoinedFlavor::Local(arranged); } diff --git a/src/row-spine/src/lib.rs b/src/row-spine/src/lib.rs index f7495ddc8dbfb..ad84b647029af 100644 --- a/src/row-spine/src/lib.rs +++ b/src/row-spine/src/lib.rs @@ -57,12 +57,15 @@ mod spines { pub type RowRowBatcher = KeyValBatcher; pub type RowRowBuilder = ArcBuilder>; - /// `RowRowBuilder` variant that consumes [`Column`] chunks. Pairs with - /// [`Col2ValPagedBatcher`] for the spillable arrange path. Installs a - /// dictionary codec at seal time, gathering statistics from the sealed - /// `Column` chain, so paged arrangements compress on the same footing as the - /// columnation-fed [`RowRowBuilder`]. + /// `RowRowBuilder` variant that consumes [`Column`] chunks. Pairs with any + /// batcher whose chains are `Column`s, spillable + /// ([`Col2ValPagedBatcher`]) or resident ([`Col2ValColBatcher`]) alike, so + /// the `Paged` in the name records where it started rather than a + /// restriction. Installs a dictionary codec at seal time, gathering + /// statistics from the sealed `Column` chain, so columnar arrangements + /// compress on the same footing as the columnation-fed [`RowRowBuilder`]. /// + /// [`Col2ValColBatcher`]: mz_timely_util::columnar::Col2ValColBatcher /// [`Col2ValPagedBatcher`]: mz_timely_util::columnar::Col2ValPagedBatcher /// [`Column`]: mz_timely_util::columnar::Column pub type RowRowColPagedBuilder = @@ -1198,8 +1201,9 @@ mod dictionary { } } - /// Paged counterpart of [`RowRowBuilder`] that consumes [`Column`] - /// chunks instead of columnation stacks. Mirrors `RowRowBuilder::seal`: + /// Counterpart of [`RowRowBuilder`] that consumes [`Column`] chunks + /// instead of columnation stacks, whether or not the batcher that + /// produced them pages. Mirrors `RowRowBuilder::seal`: /// it gathers key and value statistics from the sealed chain and /// installs codecs directly, then drops the per-container stats gatherer. pub struct RowRowColPagedBuilder< diff --git a/src/timely-util/src/columnar.rs b/src/timely-util/src/columnar.rs index 572c68a3c32ce..d10fd370c899f 100644 --- a/src/timely-util/src/columnar.rs +++ b/src/timely-util/src/columnar.rs @@ -64,6 +64,15 @@ pub type Col2KeyBatcher = Col2ValBatcher; /// real one via [`merge_batcher::ColumnMergeBatcher::set_pager`]. pub type Col2ValPagedBatcher = merge_batcher::ColumnMergeBatcher<(K, V), T, R>; +/// Columnar-native counterpart to [`Col2ValBatcher`], holding [`Column`] +/// chunks rather than columnation stacks and merging them through +/// [`batcher::ColumnMerger`]. +/// +/// Pairs with [`batcher::ColumnChunker`] and any builder whose `Input` is +/// `Column<((K, V), T, R)>`. Unlike [`Col2ValPagedBatcher`] the chains stay +/// resident, so this arm carries no pager and no spill budget. +pub type Col2ValColBatcher = MergeBatcher>; + /// A container based on a columnar store, encoded in aligned bytes. /// /// The type can represent typed data, bytes from Timely, or an aligned allocation. The name diff --git a/test/launchdarkly-flag-consistency/mzcompose.py b/test/launchdarkly-flag-consistency/mzcompose.py index af702e504d697..c06b322c565f2 100644 --- a/test/launchdarkly-flag-consistency/mzcompose.py +++ b/test/launchdarkly-flag-consistency/mzcompose.py @@ -240,6 +240,7 @@ enable_statement_arrival_logging enable_binary_date_bin enable_coalesce_case_transform + enable_columnar_merge_batcher enable_compute_half_join2 enable_compute_render_fueled_as_specific_collection enable_date_bin_hopping