compute: Retire the fueled as_specific_collection flag

`as_specific_collection`'s keyed path was gated on
`ENABLE_COMPUTE_RENDER_FUELED_AS_SPECIFIC_COLLECTION` (default true): the
fueled branch packs the arrangement cursor into a `ColumnBuilder` and
returns the columnar edge, while the off branch called the deprecated
`ArrangementFlavor::as_collection` and returned a real-data `Vec`. The
off branch was the last non-leaf real-data `Vec` producer.

The fueled path is a complete functional replacement: it materializes
every `(row, t, diff)` from the same cursor with all columns decoded, so
production (flag on) already used it. Delete the off branch, always take
the fueled columnar path, and retire the flag from `dyncfgs.rs`, its
config-set registration, and the LaunchDarkly and parallel-workload flag
lists.

With the flag gone, `ArrangementFlavor::as_collection` has no callers, so
remove it. The `config_set` argument then existed only to reach the flag,
threaded through `as_specific_collection` and `as_collection_core` (whose
sole `config_set` use was forwarding it), so drop it from both signatures
and their callers.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: antiguru

misc/python/materialize/mzcompose/__init__.py

@@ -588,7 +588,6 @@ def get_default_system_parameters(
     "compute_mv_sink_advance_persist_frontiers",
     "compute_prometheus_introspection_scrape_interval",
     "enable_compute_replica_expiration",
-    "enable_compute_render_fueled_as_specific_collection",
     "compute_logical_backpressure_max_retained_capabilities",
     "compute_logical_backpressure_inflight_slack",
     "persist_fetch_semaphore_cost_adjustment",

misc/python/materialize/parallel_workload/action.py

@@ -1911,7 +1911,6 @@ def __init__(
             "enable_compute_replica_expiration",
             "compute_mv_sink_advance_persist_frontiers",
             "compute_replica_expiration_offset",
-            "enable_compute_render_fueled_as_specific_collection",
             "compute_temporal_bucketing_summary",
             "enable_compute_logical_backpressure",
             "enable_replica_targeted_materialized_views",

src/compute-types/src/dyncfgs.rs

@@ -346,13 +346,6 @@ pub const COMPUTE_FLAT_MAP_FUEL: Config<usize> = Config::new(
     "The amount of output the flat-map operator produces before yielding.",
 );
 
-/// Whether to render `as_specific_collection` using a fueled flat-map operator.
-pub const ENABLE_COMPUTE_RENDER_FUELED_AS_SPECIFIC_COLLECTION: Config<bool> = Config::new(
-    "enable_compute_render_fueled_as_specific_collection",
-    true,
-    "When enabled, renders `as_specific_collection` using a fueled flat-map operator.",
-);
-
 /// Whether to apply logical backpressure in compute dataflows.
 pub const ENABLE_COMPUTE_LOGICAL_BACKPRESSURE: Config<bool> = Config::new(
     "enable_compute_logical_backpressure",
@@ -520,7 +513,6 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&COMPUTE_APPLY_COLUMN_DEMANDS)
         .add(&COMPUTE_FLAT_MAP_FUEL)
         .add(&CONSOLIDATING_VEC_GROWTH_DAMPENER)
-        .add(&ENABLE_COMPUTE_RENDER_FUELED_AS_SPECIFIC_COLLECTION)
         .add(&ENABLE_COMPUTE_LOGICAL_BACKPRESSURE)
         .add(&COMPUTE_LOGICAL_BACKPRESSURE_MAX_RETAINED_CAPABILITIES)
         .add(&COMPUTE_LOGICAL_BACKPRESSURE_INFLIGHT_SLACK)

src/compute/src/render.rs

@@ -1259,17 +1259,12 @@ impl<'scope, T: RenderTimestamp + MaybeBucketByTime> Context<'scope, T> {
                             mfp,
                             Some((key, row)),
                             self.until.clone(),
-                            &self.config_set,
                         );
                         CollectionBundle::from_edge(oks, errs)
                     }
                     mz_compute_types::plan::GetPlan::Collection(mfp) => {
-                        let (oks, errs) = collection.as_collection_core(
-                            mfp,
-                            None,
-                            self.until.clone(),
-                            &self.config_set,
-                        );
+                        let (oks, errs) =
+                            collection.as_collection_core(mfp, None, self.until.clone());
                         CollectionBundle::from_edge(oks, errs)
                     }
                 }
@@ -1284,12 +1279,8 @@ impl<'scope, T: RenderTimestamp + MaybeBucketByTime> Context<'scope, T> {
                 if mfp.is_identity() {
                     input
                 } else {
-                    let (oks, errs) = input.as_collection_core(
-                        mfp,
-                        input_key_val,
-                        self.until.clone(),
-                        &self.config_set,
-                    );
+                    let (oks, errs) =
+                        input.as_collection_core(mfp, input_key_val, self.until.clone());
                     CollectionBundle::from_edge(oks, errs)
                 }
             }

src/compute/src/render/context.rs

@@ -22,8 +22,7 @@ use differential_dataflow::trace::{Cursor, Navigable, TraceReader};
 use differential_dataflow::{AsCollection, 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_COLUMN_PAGED_BATCHER, 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};
@@ -57,7 +56,7 @@ use crate::render::{LinearJoinSpec, MaybeBucketByTime, RenderTimestamp};
 use crate::typedefs::{
     ErrAgent, ErrBatcher, ErrBuilder, ErrEnter, ErrSpine, RowRowAgent, RowRowEnter, RowRowSpine,
 };
-use mz_row_spine::{DatumSeq, RowRowBuilder, RowRowColPagedBuilder};
+use mz_row_spine::{RowRowBuilder, RowRowColPagedBuilder};
 
 /// Dataflow-local collections and arrangements.
 ///
@@ -236,40 +235,6 @@ pub enum ArrangementFlavor<'scope, T: RenderTimestamp> {
 }
 
 impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> {
-    /// Presents `self` as a stream of updates.
-    ///
-    /// Deprecated: This function is not fueled and hence risks flattening the whole arrangement.
-    ///
-    /// This method presents the contents as they are, without further computation.
-    /// If you have logic that could be applied to each record, consider using the
-    /// `flat_map` methods which allows this and can reduce the work done.
-    #[deprecated(note = "Use `flat_map` instead.")]
-    pub fn as_collection(
-        &self,
-    ) -> (
-        VecCollection<'scope, T, Row, Diff>,
-        VecCollection<'scope, T, DataflowErrorSer, Diff>,
-    ) {
-        let mut datums = DatumVec::new();
-        let logic = move |k: DatumSeq, v: DatumSeq| {
-            let temp_storage = RowArena::new();
-            let mut datums_borrow = datums.borrow();
-            k.extend_datums(&temp_storage, &mut datums_borrow, None);
-            v.extend_datums(&temp_storage, &mut datums_borrow, None);
-            SharedRow::pack(&**datums_borrow)
-        };
-        match &self {
-            ArrangementFlavor::Local(oks, errs) => (
-                oks.clone().as_collection(logic),
-                errs.clone().as_collection(|k, &()| k.clone()),
-            ),
-            ArrangementFlavor::Trace(_, oks, errs) => (
-                oks.clone().as_collection(logic),
-                errs.clone().as_collection(|k, &()| k.clone()),
-            ),
-        }
-    }
-
     /// Constructs and applies logic to elements of `self` and returns the results.
     ///
     /// The `logic` callback receives a borrow of the decoded datum vector, a timestamp, a
@@ -558,9 +523,8 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> {
     /// Therefore, it should be used when the appropriate transformation
     /// was planned as part of a following MFP.
     ///
-    /// If `key` is specified, the function converts the arrangement to a collection. It uses either
-    /// the fueled `flat_map` or `as_collection` method, depending on the flag
-    /// [`ENABLE_COMPUTE_RENDER_FUELED_AS_SPECIFIC_COLLECTION`].
+    /// If `key` is specified, the function converts the arrangement to a collection using a
+    /// fueled `flat_map` operator.
     ///
     /// The keyed path materializes the arrangement as the columnar edge, so an
     /// arrangement-producing operator (Reduce, Threshold, bucketed TopK) whose
@@ -570,7 +534,6 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> {
     pub fn as_specific_collection(
         &self,
         key: Option<&[LirScalarExpr]>,
-        config_set: &ConfigSet,
     ) -> (
         CollectionEdge<'scope, T>,
         VecCollection<'scope, T, DataflowErrorSer, Diff>,
@@ -589,28 +552,22 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> {
                 let arranged = self.arranged.get(key).unwrap_or_else(|| {
                     panic!("The collection arranged by {:?} doesn't exist.", key)
                 });
-                if ENABLE_COMPUTE_RENDER_FUELED_AS_SPECIFIC_COLLECTION.get(config_set) {
-                    // Decode all columns (max_demand usize::MAX) and pack each cursor record
-                    // into a `Column`, so the materialized collection carries the columnar edge.
-                    // Output is 1:1 from the already-consolidated cursor, so a non-consolidating
-                    // `ColumnBuilder` matches the row-based `CapacityContainerBuilder` this
-                    // replaced; the packed row is pushed borrowed, holding no owned `Row` per
-                    // record.
-                    let (ok, err) = arranged.flat_map_ok::<ColumnBuilder<(Row, T, Diff)>, _>(
-                        None,
-                        usize::MAX,
-                        |borrow, t, r, ok_session| {
-                            let row = SharedRow::pack(borrow.iter());
-                            ok_session.give((&row, &t, &r));
-                            1
-                        },
-                    );
-                    (CollectionEdge::Columnar(ok.as_collection()), err)
-                } else {
-                    #[allow(deprecated)]
-                    let (oks, errs) = arranged.as_collection();
-                    (CollectionEdge::Vec(oks), errs)
-                }
+                // Decode all columns (max_demand usize::MAX) and pack each cursor record
+                // into a `Column`, so the materialized collection carries the columnar edge.
+                // Output is 1:1 from the already-consolidated cursor, so a non-consolidating
+                // `ColumnBuilder` matches the row-based `CapacityContainerBuilder` this
+                // replaced; the packed row is pushed borrowed, holding no owned `Row` per
+                // record.
+                let (ok, err) = arranged.flat_map_ok::<ColumnBuilder<(Row, T, Diff)>, _>(
+                    None,
+                    usize::MAX,
+                    |borrow, t, r, ok_session| {
+                        let row = SharedRow::pack(borrow.iter());
+                        ok_session.give((&row, &t, &r));
+                        1
+                    },
+                );
+                (CollectionEdge::Columnar(ok.as_collection()), err)
             }
         }
     }
@@ -921,7 +878,6 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> {
         mfp_plan: MfpPlan<LirScalarExpr>,
         key_val: Option<(Vec<LirScalarExpr>, Option<Row>)>,
         until: Antichain<mz_repr::Timestamp>,
-        config_set: &ConfigSet,
     ) -> (
         CollectionEdge<'scope, T>,
         VecCollection<'scope, T, DataflowErrorSer, Diff>,
@@ -949,7 +905,7 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> {
                 // Keyed identity materializes an existing arrangement.
                 // `as_specific_collection` returns the columnar edge, so the
                 // reduce/threshold/topk result carries columnar downstream.
-                Some(key) => self.as_specific_collection(Some(&key), config_set),
+                Some(key) => self.as_specific_collection(Some(&key)),
             };
         }
 
@@ -1066,7 +1022,7 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> {
         let form_raw_collection = collections.raw || will_create_arrangement;
         if form_raw_collection && self.collection.is_none() {
             let (oks, errs) =
-                self.as_collection_core(input_mfp, input_key.map(|k| (k, None)), until, config_set);
+                self.as_collection_core(input_mfp, input_key.map(|k| (k, None)), until);
             // Apply temporal bucketing when the lowering selected `TemporalBucketing` and
             // we will build at least one arrangement. This path fires when the collection
             // must be formed from scratch (e.g., from an arrangement via as_collection_core).
@@ -1523,7 +1479,6 @@ fn walk_cursor<C, F>(
 #[cfg(test)]
 mod tests {
     use differential_dataflow::input::Input;
-    use mz_compute_types::dyncfgs::all_dyncfgs;
     use mz_expr::{EvalError, MapFilterProject};
     use mz_repr::{Datum, ReprScalarType, Timestamp};
     use timely::dataflow::operators::Capture;
@@ -1727,15 +1682,13 @@ mod tests {
             .collect();
         expected.sort();
 
-        let config_set = ConfigSet::default();
         let (producer_is_columnar, passthrough_is_columnar, produced) =
             timely::execute_directly(move |worker| {
                 worker.dataflow::<Timestamp, _, _>(|scope| {
                     let (mut input, collection) = scope.new_collection();
                     let (_err_input, errs) = scope.new_collection::<DataflowErrorSer, Diff>();
                     let bundle = CollectionBundle::from_edge(CollectionEdge::Vec(collection), errs);
-                    let (edge, _errs) =
-                        bundle.as_collection_core(mfp, None, Antichain::new(), &config_set);
+                    let (edge, _errs) = bundle.as_collection_core(mfp, None, Antichain::new());
                     let producer_is_columnar = matches!(edge, CollectionEdge::Columnar(_));
                     // Tee the producer output for a content check, then feed the
                     // original edge into the arrange input.
@@ -1778,7 +1731,6 @@ mod tests {
     #[mz_ore::test]
     fn as_collection_core_identity_passes_edge_through() {
         for columnar_input in [false, true] {
-            let config_set = ConfigSet::default();
             let is_columnar = timely::execute_directly(move |worker| {
                 worker.dataflow::<Timestamp, _, _>(|scope| {
                     let (mut input, collection) = scope.new_collection::<Row, Diff>();
@@ -1792,8 +1744,7 @@ mod tests {
                     let identity = MapFilterProject::<LirScalarExpr>::new(1)
                         .into_plan()
                         .expect("identity mfp");
-                    let (out, _errs) =
-                        bundle.as_collection_core(identity, None, Antichain::new(), &config_set);
+                    let (out, _errs) = bundle.as_collection_core(identity, None, Antichain::new());
                     let is_columnar = matches!(out, CollectionEdge::Columnar(_));
                     input.update(Row::pack_slice(&[Datum::Int64(1)]), Diff::ONE);
                     input.advance_to(Timestamp::from(1u64));
@@ -1839,14 +1790,12 @@ mod tests {
             ),
         ];
 
-        let config_set = ConfigSet::default();
         let captured = timely::execute_directly(move |worker| {
             worker.dataflow::<Timestamp, _, _>(|scope| {
                 let (mut input, collection) = scope.new_collection();
                 let (_err_input, errs) = scope.new_collection::<DataflowErrorSer, Diff>();
                 let bundle = CollectionBundle::from_edge(CollectionEdge::Vec(collection), errs);
-                let (edge, _errs) =
-                    bundle.as_collection_core(mfp, None, Antichain::new(), &config_set);
+                let (edge, _errs) = bundle.as_collection_core(mfp, None, Antichain::new());
                 assert!(
                     matches!(edge, CollectionEdge::Columnar(_)),
                     "a non-identity MFP must produce a columnar edge"
@@ -1887,9 +1836,6 @@ mod tests {
             .collect();
         expected.sort();
 
-        // A populated set so the fueled-materialization flag resolves to its
-        // default (`true`); `ConfigSet::default()` alone would panic on lookup.
-        let config_set = all_dyncfgs(ConfigSet::default());
         let (is_columnar, captured) = timely::execute_directly(move |worker| {
             worker.dataflow::<Timestamp, _, _>(|scope| {
                 let (mut input, collection) = scope.new_collection();
@@ -1915,7 +1861,7 @@ mod tests {
                     0..1,
                     ArrangementFlavor::Local(arranged, err_arranged),
                 );
-                let (edge, _errs) = bundle.as_specific_collection(Some(&key), &config_set);
+                let (edge, _errs) = bundle.as_specific_collection(Some(&key));
                 let is_columnar = matches!(edge, CollectionEdge::Columnar(_));
                 let captured = edge.into_vec().inner.capture();
 

src/compute/src/render/flat_map.rs

@@ -61,7 +61,7 @@ impl<'scope, T: crate::render::RenderTimestamp> Context<'scope, T> {
                 .collection
                 .clone()
                 .expect("The unarranged collection doesn't exist."),
-            Some(key) => input.as_specific_collection(Some(key), &self.config_set),
+            Some(key) => input.as_specific_collection(Some(key)),
         };
 
         let (oks, errs) = match edge {

src/compute/src/render/join/linear_join.rs

@@ -266,8 +266,9 @@ where
                     // columnar edge. `differential_join` forms the source key off
                     // this edge below, so the columnar source has no `ColumnarToVec`
                     // hop.
-                    Some(key) => inputs[linear_plan.source_relation]
-                        .as_specific_collection(Some(key), &self.config_set),
+                    Some(key) => {
+                        inputs[linear_plan.source_relation].as_specific_collection(Some(key))
+                    }
                 };
                 errors.push(errs.enter_region(inner));
                 let joined = joined.enter_region(inner);

src/compute/src/render/sinks.rs

@@ -83,12 +83,8 @@ impl<'g, T: RenderTimestamp> Context<'g, T> {
             // The sink serializes rows, so decode to `Vec` here. This is the
             // sanctioned sink leaf, the same seam as the raw-collection arm
             // above.
-            let (oks, errs) = bundle.as_collection_core(
-                mfp_plan,
-                Some((key.clone(), None)),
-                self.until.clone(),
-                &self.config_set,
-            );
+            let (oks, errs) =
+                bundle.as_collection_core(mfp_plan, Some((key.clone(), None)), self.until.clone());
             (oks.into_vec(), errs)
         };
 

test/launchdarkly-flag-consistency/mzcompose.py

@@ -244,7 +244,6 @@
     enable_coalesce_case_transform
     enable_cluster_controller
     enable_compute_half_join2
-    enable_compute_render_fueled_as_specific_collection
     enable_date_bin_hopping
     enable_default_connection_validation
     enable_dequadratic_eqprop_map