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