diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index 042251f16b068..8b0df12ec2b52 100644 --- a/misc/python/materialize/mzcompose/__init__.py +++ b/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", diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py index 8b4239957807f..2bd87b9b69e10 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/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", diff --git a/src/compute-types/src/dyncfgs.rs b/src/compute-types/src/dyncfgs.rs index 1d74ae1ba03f0..b302720179f7e 100644 --- a/src/compute-types/src/dyncfgs.rs +++ b/src/compute-types/src/dyncfgs.rs @@ -532,14 +532,6 @@ pub const COMPUTE_FLAT_MAP_FUEL: Config = 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 = 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 = 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) diff --git a/src/compute/src/render.rs b/src/compute/src/render.rs index 40dda6ab6f634..9b38b18038104 100644 --- a/src/compute/src/render.rs +++ b/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) } } diff --git a/src/compute/src/render/context.rs b/src/compute/src/render/context.rs index 5e008facb0b16..60312eab830f6 100644 --- a/src/compute/src/render/context.rs +++ b/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::, _>( - 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::, _>(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, key_val: Option<(Vec, Option)>, until: Antichain, - 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( #[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::(|scope| { let (mut input, collection) = scope.new_collection(); let (_err_input, errs) = scope.new_collection::(); 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::(|scope| { let (mut input, collection) = scope.new_collection::(); @@ -1816,8 +1763,7 @@ mod tests { let identity = MapFilterProject::::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::(|scope| { let (mut input, collection) = scope.new_collection(); let (_err_input, errs) = scope.new_collection::(); 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::(|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(); diff --git a/src/compute/src/render/flat_map.rs b/src/compute/src/render/flat_map.rs index 0ceaa92002e6c..b70217d9f8b68 100644 --- a/src/compute/src/render/flat_map.rs +++ b/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 { diff --git a/src/compute/src/render/join/linear_join.rs b/src/compute/src/render/join/linear_join.rs index 15e6751f87b5a..ffc42656ad7d7 100644 --- a/src/compute/src/render/join/linear_join.rs +++ b/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); diff --git a/src/compute/src/render/sinks.rs b/src/compute/src/render/sinks.rs index b66807febc5ac..35518b34d11d4 100644 --- a/src/compute/src/render/sinks.rs +++ b/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) }; diff --git a/test/launchdarkly-flag-consistency/mzcompose.py b/test/launchdarkly-flag-consistency/mzcompose.py index dca6979751586..8c5dc60223268 100644 --- a/test/launchdarkly-flag-consistency/mzcompose.py +++ b/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