compute: select the arrange merge batcher by dyncfg

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

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
         )

src/compute-types/src/dyncfgs.rs

@@ -48,19 +48,45 @@ pub const ENABLE_ERROR_DISTINCT: Config<bool> = 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
 /// explicitly so the spill path is measured.
 pub const ENABLE_COLUMN_PAGED_BATCHER: Config<bool> = 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<bool> = 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)

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<ColumnationStack<_>>` + `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`.

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<LirScalarExpr>,
         thinning: Vec<usize>,
-        use_paged_path: bool,
+        batcher: ArrangementBatcher,
     ) -> (
         Arranged<'scope, RowRowAgent<T, Diff>>,
         VecCollection<'scope, T, DataflowErrorSer, Diff>,
@@ -1270,22 +1267,28 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> {
 
         let exchange =
             ExchangeCore::<ColumnBuilder<_>, _>::new_core(columnar_exchange::<Row, Row, T, Diff>);
-        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,

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::<ColumnBuilder<_>, _>::new_core(
                 columnar_exchange::<Row, Row, T, Diff>,
             );
-            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);
         }

src/row-spine/src/lib.rs

@@ -57,12 +57,15 @@ mod spines {
     pub type RowRowBatcher<T, R> = KeyValBatcher<Row, Row, T, R>;
     pub type RowRowBuilder<T, R> = ArcBuilder<crate::dictionary::builders::RowRowBuilder<T, R>>;
 
-    /// `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<T, R> =
@@ -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<

src/timely-util/src/columnar.rs

@@ -64,6 +64,15 @@ pub type Col2KeyBatcher<K, T, R> = Col2ValBatcher<K, (), T, R>;
 /// real one via [`merge_batcher::ColumnMergeBatcher::set_pager`].
 pub type Col2ValPagedBatcher<K, V, T, R> = 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<K, V, T, R> = MergeBatcher<batcher::ColumnMerger<(K, V), T, R>>;
+
 /// 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

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