compute: Retire the fueled as_specific_collection flag

misc/python/materialize/mzcompose/__init__.py

@@ -700,7 +700,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

@@ -3272,7 +3272,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

@@ -532,14 +532,6 @@ pub const COMPUTE_FLAT_MAP_FUEL: Config<usize> = Config::new(
     ParameterScope::Replica,
 );
 
-/// 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.",
-    ParameterScope::Environment,
-);
-
 /// Whether to apply logical backpressure in compute dataflows.
 pub const ENABLE_COMPUTE_LOGICAL_BACKPRESSURE: Config<bool> = Config::new(
     "enable_compute_logical_backpressure",
@@ -837,7 +829,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

@@ -1291,17 +1291,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)
                     }
                 }
@@ -1316,12 +1311,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

@@ -21,10 +21,7 @@ use differential_dataflow::trace::implementations::BatchContainer;
 use differential_dataflow::trace::{Cursor, Navigable, TraceReader};
 use differential_dataflow::{AsCollection, VecCollection};
 use mz_compute_types::dataflows::DataflowDescription;
-use mz_compute_types::dyncfgs::{
-    ENABLE_COMPUTE_RENDER_FUELED_AS_SPECIFIC_COLLECTION, ENABLE_COMPUTE_TEMPORAL_BUCKETING,
-    TEMPORAL_BUCKETING_SUMMARY,
-};
+use mz_compute_types::dyncfgs::{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};
 use mz_dyncfg::ConfigSet;
@@ -60,7 +57,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.
 ///
@@ -239,40 +236,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
@@ -639,16 +602,14 @@ 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. The unkeyed path
     /// returns the unarranged `.collection` edge with its variant intact.
     pub fn as_specific_collection(
         &self,
         key: Option<&[LirScalarExpr]>,
-        config_set: &ConfigSet,
     ) -> (
         CollectionEdge<'scope, T>,
         VecCollection<'scope, T, DataflowErrorSer, Diff>,
@@ -667,30 +628,21 @@ 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) {
-                    // Output is 1:1 from the already-consolidated cursor, so a
-                    // non-consolidating `ColumnBuilder` suffices. `max_demand` is
-                    // `usize::MAX` because the materialized collection carries every column.
-                    let (ok, err) = arranged.flat_map_ok::<ColumnBuilder<(Row, T, Diff)>, _>(
-                        None,
-                        usize::MAX,
-                        {
-                            // `give` copies the bytes into the column, so one buffer
-                            // serves every record.
-                            let mut row_buf = Row::default();
-                            move |borrow, t, r, ok_session| {
-                                row_buf.packer().extend(borrow.iter());
-                                ok_session.give((&row_buf, &t, &r));
-                                1
-                            }
-                        },
-                    );
-                    (CollectionEdge::Columnar(ok.as_collection()), err)
-                } else {
-                    #[allow(deprecated)]
-                    let (oks, errs) = arranged.as_collection();
-                    (CollectionEdge::Vec(oks), errs)
-                }
+                // Output is 1:1 from the already-consolidated cursor, so a
+                // non-consolidating `ColumnBuilder` suffices. `max_demand` is
+                // `usize::MAX` because the materialized collection carries every column.
+                let (ok, err) =
+                    arranged.flat_map_ok::<ColumnBuilder<(Row, T, Diff)>, _>(None, usize::MAX, {
+                        // `give` copies the bytes into the column, so one buffer
+                        // serves every record.
+                        let mut row_buf = Row::default();
+                        move |borrow, t, r, ok_session| {
+                            row_buf.packer().extend(borrow.iter());
+                            ok_session.give((&row_buf, &t, &r));
+                            1
+                        }
+                    });
+                (CollectionEdge::Columnar(ok.as_collection()), err)
             }
         }
     }
@@ -994,7 +946,6 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> {
         mfp_plan: MfpPlan<LirScalarExpr>,
         key_val: Option<(Vec<LirScalarExpr>, Option<StableRow>)>,
         until: Antichain<mz_repr::Timestamp>,
-        config_set: &ConfigSet,
     ) -> (
         CollectionEdge<'scope, T>,
         VecCollection<'scope, T, DataflowErrorSer, Diff>,
@@ -1022,7 +973,7 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> {
                     .collection
                     .clone()
                     .expect("The unarranged collection doesn't exist."),
-                Some(key) => self.as_specific_collection(Some(&key), config_set),
+                Some(key) => self.as_specific_collection(Some(&key)),
             };
         }
 
@@ -1135,7 +1086,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).
@@ -1575,7 +1526,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;
@@ -1753,15 +1703,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(_));
                     let produced = edge.clone().into_vec().inner.capture();
                     let (_arranged, _arrange_errs, passthrough) =
@@ -1802,7 +1750,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>();
@@ -1816,8 +1763,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));
@@ -1863,14 +1809,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"
@@ -1902,8 +1846,6 @@ mod tests {
             .collect();
         expected.sort();
 
-        // A populated set, because `ConfigSet::default()` panics 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();
@@ -1929,7 +1871,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

@@ -60,7 +60,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

@@ -259,8 +259,9 @@ where
                         .expect("The unarranged collection doesn't exist."),
                     // A source key materializes an existing arrangement, which
                     // `as_specific_collection` presents as a columnar edge.
-                    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

@@ -251,7 +251,6 @@
     enable_compute_half_join2
     enable_compute_index_peek_offload
     enable_compute_peek_row_iteration_limit
-    enable_compute_render_fueled_as_specific_collection
     enable_date_bin_hopping
     enable_default_connection_validation
     enable_dequadratic_eqprop_map