From 05b527287fccbea2954a57ae95b29789fde9fa54 Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Mon, 20 Jul 2026 23:54:05 +0200 Subject: [PATCH 1/2] compute: columnar Get/Mfp output via as_collection_core Make `as_collection_core` build its ok output into a `ConsolidatingColumnBuilder` and return a `CollectionEdge::Columnar`, flipping the Get and Mfp producers to emit the columnar edge. This is the first producer flip: the columnar arm goes live for Get/Mfp-fed edges, which Wave 1 already made every consumer accept natively or decode only at a sanctioned leaf. Rework the identity fast-path so an unarranged trivial MFP hands the input edge straight through (a columnar producer stays columnar with no `ColumnarToVec` hop) instead of forcing `as_specific_collection` to decode it to `Vec`. A keyed identity still reads its arrangement as a `Vec` edge. `as_specific_collection` stays the row-based consumer leaf. `ConsolidatingColumnBuilder` folds within-batch duplicates, matching the row-based `ConsolidatingContainerBuilder` this replaced, and emits a `Column` instead of a `Vec`. It stages owned `(Row, T, Diff)` tuples to consolidate in place, so the producer gives its records owned. `mfp_plan.evaluate` already produces a fresh owned `Row` per result, so this is a move into staging, not a new allocation. The borrowed-push no-owned-Row pattern is a consumer optimization for reading an existing columnar batch; it never applied to a producer computing new rows. The internal fueled `flat_map` chain (`CollectionBundle::flat_map`, `ArrangementFlavor::flat_map`, `flat_map_core_fallible`, `PendingWork:: do_work`) carried a `PushInto<(D, T, Diff)>` bound that constrained the element type up front. The bound was never exercised by those bodies (the caller's `logic` performs the `give`), so relax it to `DCB: ContainerBuilder` and drop the now-unused `D`, letting the producer choose its own builder and push shape through the same fuel machinery. Sink and `ensure_collections` temporal bucketing decode the produced edge at their existing sanctioned leaves. Tests: a `Get -> ArrangeBy` end-to-end unit test asserting the producer emits columnar and the arrange input keeps the columnar passthrough; an identity-fast-path test asserting the input edge variant is preserved; and a within-batch consolidation test asserting duplicate output rows fold to a single record with summed diff. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/compute/src/render.rs | 6 +- src/compute/src/render/context.rs | 331 ++++++++++++++++++++++++------ src/compute/src/render/reduce.rs | 2 +- src/compute/src/render/sinks.rs | 8 +- 4 files changed, 281 insertions(+), 66 deletions(-) diff --git a/src/compute/src/render.rs b/src/compute/src/render.rs index c8607c22246cc..1ea3080e59a03 100644 --- a/src/compute/src/render.rs +++ b/src/compute/src/render.rs @@ -1276,7 +1276,7 @@ impl<'scope, T: RenderTimestamp + MaybeBucketByTime> Context<'scope, T> { self.until.clone(), &self.config_set, ); - CollectionBundle::from_collections(oks, errs) + CollectionBundle::from_edge(oks, errs) } mz_compute_types::plan::GetPlan::Collection(mfp) => { let (oks, errs) = collection.as_collection_core( @@ -1285,7 +1285,7 @@ impl<'scope, T: RenderTimestamp + MaybeBucketByTime> Context<'scope, T> { self.until.clone(), &self.config_set, ); - CollectionBundle::from_collections(oks, errs) + CollectionBundle::from_edge(oks, errs) } } } @@ -1305,7 +1305,7 @@ impl<'scope, T: RenderTimestamp + MaybeBucketByTime> Context<'scope, T> { self.until.clone(), &self.config_set, ); - CollectionBundle::from_collections(oks, errs) + CollectionBundle::from_edge(oks, errs) } } FlatMap { diff --git a/src/compute/src/render/context.rs b/src/compute/src/render/context.rs index 36bef4b1b7536..afce351953452 100644 --- a/src/compute/src/render/context.rs +++ b/src/compute/src/render/context.rs @@ -36,6 +36,7 @@ use mz_storage_types::controller::CollectionMetadata; use mz_timely_util::columnar::Column; use mz_timely_util::columnar::batcher; use mz_timely_util::columnar::builder::ColumnBuilder; +use mz_timely_util::columnar::consolidate::ConsolidatingColumnBuilder; use mz_timely_util::columnar::{ Col2ValBatcher, Col2ValColBatcher, Col2ValPagedBatcher, columnar_exchange, }; @@ -304,7 +305,7 @@ impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> { /// The `max_demand` parameter limits the number of columns decoded from the /// input. Only the first `max_demand` columns are decoded. Pass `usize::MAX` to /// decode all columns. - pub fn flat_map( + pub fn flat_map( &self, key: Option<&Row>, max_demand: usize, @@ -314,8 +315,7 @@ impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> { VecCollection<'scope, T, DataflowErrorSer, Diff>, ) where - D: Data, - DCB: ContainerBuilder + PushInto<(D, T, Diff)>, + DCB: ContainerBuilder, L: for<'a, 'b> FnMut( &'a mut DatumVecBorrow<'b>, T, @@ -329,7 +329,7 @@ impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> { // decode (and the activation-scoped arena it decodes into). match &self { ArrangementFlavor::Local(oks, errs) => { - let (oks, mfp_errs) = CollectionBundle::::flat_map_core_fallible::<_, _, DCB, _>( + let (oks, mfp_errs) = CollectionBundle::::flat_map_core_fallible::<_, DCB, _>( oks.clone(), key, max_demand, @@ -341,7 +341,7 @@ impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> { (oks, errs) } ArrangementFlavor::Trace(_, oks, errs) => { - let (oks, mfp_errs) = CollectionBundle::::flat_map_core_fallible::<_, _, DCB, _>( + let (oks, mfp_errs) = CollectionBundle::::flat_map_core_fallible::<_, DCB, _>( oks.clone(), key, max_demand, @@ -703,7 +703,7 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> { /// The `max_demand` parameter limits the number of columns decoded from the /// input. Only the first `max_demand` columns are decoded. Pass `usize::MAX` to /// decode all columns. - pub fn flat_map( + pub fn flat_map( &self, key_val: Option<(Vec, Option)>, max_demand: usize, @@ -713,8 +713,7 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> { VecCollection<'scope, T, DataflowErrorSer, Diff>, ) where - D: Data, - DCB: ContainerBuilder + PushInto<(D, T, Diff)>, + DCB: ContainerBuilder, L: for<'a> FnMut( &'a mut DatumVecBorrow<'_>, T, @@ -730,7 +729,7 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> { if let Some((key, val)) = key_val { self.arrangement(&key) .expect("Should have ensured during planning that this arrangement exists.") - .flat_map::<_, DCB, _>(val.as_ref(), max_demand, logic) + .flat_map::(val.as_ref(), max_demand, logic) } else { let (oks, errs) = self .collection @@ -752,7 +751,7 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> { /// callback writes ok results into the first session and errors into the second, returning /// the number of records produced. See [`ArrangementFlavor::flat_map`] for the fuel /// rationale. - fn flat_map_core_fallible( + fn flat_map_core_fallible( trace: Arranged<'scope, Tr>, key: Option<&< as Cursor>::KeyContainer as BatchContainer>::Owned>, max_demand: usize, @@ -767,8 +766,10 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> { for<'a> BatchCursor: Cursor: ExtendDatums, Val<'a>: ExtendDatums, Time = T, Diff = mz_repr::Diff>, < as Cursor>::KeyContainer as BatchContainer>::Owned: PartialEq, - D: Data, - DCB: ContainerBuilder + PushInto<(D, T, Diff)>, + // The builder accepts whatever `logic` gives it, so the push bound lives at the `give` + // call site rather than here. This lets a caller push borrowed records into a columnar + // builder that has no owned-tuple `Push`. + DCB: ContainerBuilder, // `logic` receives the key and value already decoded into a `DatumVecBorrow`. The decode // (and its arena/`DatumVec`) lives in the per-activation closure below, so it is scoped to // a single scheduling invocation rather than to the operator. @@ -994,13 +995,13 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> { until: Antichain, config_set: &ConfigSet, ) -> ( - VecCollection<'scope, T, mz_repr::Row, Diff>, + CollectionEdge<'scope, T>, VecCollection<'scope, T, DataflowErrorSer, Diff>, ) { // Unwrap the stable-serialization row wrapper, seeking works on // plain rows. let key_val = key_val.map(|(key, val)| (key, val.map(|val| val.0))); - // If the MFP is trivial, we can just call `as_collection`. + // If the MFP is trivial, we can just return the collection. // In the case that we weren't going to apply the `key_val` optimization, // this path results in a slightly smaller and faster // dataflow graph, and is intended to fix @@ -1013,7 +1014,21 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> { if mfp_plan.is_identity() && !has_key_val { let key = key_val.map(|(k, _v)| k); - return self.as_specific_collection(key.as_deref(), config_set); + return match key { + // Unarranged identity: hand the edge straight through, so a + // columnar producer stays columnar without a `ColumnarToVec` hop. + None => self + .collection + .clone() + .expect("The unarranged collection doesn't exist."), + // Keyed identity reads an existing arrangement, which is + // row-based. Wrap it as a `Vec` edge. `as_specific_collection` + // stays the consumer leaf. + Some(key) => { + let (oks, errs) = self.as_specific_collection(Some(&key), config_set); + (CollectionEdge::Vec(oks), errs) + } + }; } // Apply demand-based column pruning. We round-trip through MIR @@ -1033,48 +1048,54 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> { // Wrap in an `Rc` so that lifetimes work out. let until = std::rc::Rc::new(until); - let (stream, errors) = self - .flat_map::<_, ConsolidatingContainerBuilder>, _>( - key_val, - max_demand, - move |row_datums, time, diff, ok_session, err_session| { - let mut row_builder = SharedRow::get(); - let until = std::rc::Rc::clone(&until); - let temp_storage = RowArena::new(); - let row_iter = row_datums.iter(); - let mut datums_local = datum_vec.borrow(); - datums_local.extend(row_iter); - let event_time = time.event_time(); - let mut work: usize = 0; - for result in mfp_plan.evaluate( - &mut datums_local, - &temp_storage, - event_time, - diff.clone(), - move |time| !until.less_equal(time), - &mut row_builder, - ) { - work += 1; - match result { - Ok((row, event_time, diff)) => { - // Copy the whole time, and re-populate event time. - let mut time: T = time.clone(); - *time.event_time_mut() = event_time; - ok_session.give((row, time, diff)); - } - Err((e, event_time, diff)) => { - // Copy the whole time, and re-populate event time. - let mut time: T = time.clone(); - *time.event_time_mut() = event_time; - err_session.give((e, time, diff)); - } + // The ok output is built into a `Column`, so this producer emits the + // columnar edge. `ConsolidatingColumnBuilder` folds within-batch + // duplicates, matching the row-based `ConsolidatingContainerBuilder` + // this replaced. It stages owned `(Row, T, Diff)` tuples to consolidate + // in place, so the records are given owned; `mfp_plan.evaluate` already + // produces a fresh owned `Row` per result, so this is a move into + // staging, not a new allocation. + let (stream, errors) = self.flat_map::, _>( + key_val, + max_demand, + move |row_datums, time, diff, ok_session, err_session| { + let mut row_builder = SharedRow::get(); + let until = std::rc::Rc::clone(&until); + let temp_storage = RowArena::new(); + let row_iter = row_datums.iter(); + let mut datums_local = datum_vec.borrow(); + datums_local.extend(row_iter); + let event_time = time.event_time(); + let mut work: usize = 0; + for result in mfp_plan.evaluate( + &mut datums_local, + &temp_storage, + event_time, + diff.clone(), + move |time| !until.less_equal(time), + &mut row_builder, + ) { + work += 1; + match result { + Ok((row, event_time, diff)) => { + // Copy the whole time, and re-populate event time. + let mut time: T = time.clone(); + *time.event_time_mut() = event_time; + ok_session.give((row, time, diff)); + } + Err((e, event_time, diff)) => { + // Copy the whole time, and re-populate event time. + let mut time: T = time.clone(); + *time.event_time_mut() = event_time; + err_session.give((e, time, diff)); } } - work - }, - ); + } + work + }, + ); - (stream.as_collection(), errors) + (CollectionEdge::Columnar(stream.as_collection()), errors) } pub fn ensure_collections( mut self, @@ -1140,11 +1161,18 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> { .try_into() .expect("must fit"); bucketed = true; - T::maybe_apply_temporal_bucketing(oks.inner, as_of.clone(), summary) + // Temporal bucketing consumes and produces a `Vec` edge, so + // decode here. This is the sanctioned leaf decode where a + // `Vec`-internal operator meets the columnar edge. + CollectionEdge::Vec(T::maybe_apply_temporal_bucketing( + oks.into_vec().inner, + as_of.clone(), + summary, + )) } else { oks }; - self.collection = Some((CollectionEdge::Vec(oks), errs)); + self.collection = Some((oks, errs)); } for (key, _, thinning) in collections.arranged { if !self.arranged.contains_key(&key) { @@ -1420,7 +1448,7 @@ where } /// Perform roughly `fuel` work through the cursor, applying `logic` and sending results to /// the two output sessions. - fn do_work( + fn do_work( &mut self, key: Option<&C::Key<'_>>, logic: &mut L, @@ -1428,8 +1456,9 @@ where ok_output: &mut OutputBuilderSession<'_, C::Time, DCB>, err_output: &mut OutputBuilderSession<'_, C::Time, ECB>, ) where - D: Data, - DCB: ContainerBuilder + PushInto<(D, C::Time, C::Diff)>, + // The push bound lives at `logic`'s `give` call site, not here, so a caller can push + // borrowed records into a columnar builder that has no owned-tuple `Push`. + DCB: ContainerBuilder, L: FnMut( C::Key<'_>, C::Val<'_>, @@ -1560,7 +1589,7 @@ fn walk_cursor( #[cfg(test)] mod tests { use differential_dataflow::input::Input; - use mz_expr::EvalError; + use mz_expr::{EvalError, MapFilterProject}; use mz_repr::{Datum, ReprScalarType, Timestamp}; use timely::dataflow::operators::Capture; use timely::dataflow::operators::capture::{Event, Extract}; @@ -1700,4 +1729,186 @@ mod tests { assert!(!err_vec.is_empty()); assert_eq!(err_vec, err_col); } + + fn extract_row_updates( + captured: Captured<(Row, Timestamp, Diff)>, + ) -> Vec<(Row, Timestamp, Diff)> { + let mut updates: Vec<_> = captured + .extract() + .into_iter() + .flat_map(|(_, data)| data) + .collect(); + updates.sort(); + updates + } + + /// A `Get -> ArrangeBy` chain carries the columnar arm end to end. A + /// non-identity MFP drives `as_collection_core` down its columnar producer + /// path, and feeding that edge into the arrange input keeps the columnar + /// passthrough, so no `ColumnarToVec` sits on the arrange path. + /// + /// The producer output is checked against the projected input. Arrange + /// correctness itself is covered by `arrange_collection_arms_agree`; here we + /// only assert the variant survives the hand-off. + #[mz_ore::test] + fn get_arrange_by_carries_columnar_end_to_end() { + let rows = vec![ + (Row::pack_slice(&[Datum::Int64(1), Datum::Int64(10)]), 0u64), + (Row::pack_slice(&[Datum::Int64(2), Datum::Int64(20)]), 1), + (Row::pack_slice(&[Datum::Int64(1), Datum::Int64(10)]), 1), + ]; + // Project away column 1; the output row carries only column 0. A + // projection is non-identity, so `as_collection_core` takes the columnar + // producer path rather than the identity passthrough. + let mfp = MapFilterProject::::new(2) + .project(vec![0]) + .into_plan() + .expect("mfp"); + let mut expected: Vec<(Row, Timestamp, Diff)> = rows + .iter() + .map(|(r, t)| { + let col0 = r.iter().next().unwrap(); + (Row::pack_slice(&[col0]), Timestamp::from(*t), Diff::ONE) + }) + .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 producer_is_columnar = matches!(edge, CollectionEdge::Columnar(_)); + // Tee the producer output for a content check, then feed the + // original edge into the arrange input. + let produced = edge.clone().into_vec().inner.capture(); + let (_arranged, _arrange_errs, passthrough) = + CollectionBundle::::arrange_collection( + &"arrange".to_string(), + edge, + vec![LirScalarExpr::column(0)], + vec![0], + ArrangementBatcher::Columnation, + ); + let passthrough_is_columnar = + matches!(passthrough, CollectionEdge::Columnar(_)); + + let max_time = rows.iter().map(|(_, t)| *t).max().unwrap(); + for (row, time) in rows { + input.update_at(row, Timestamp::from(time), Diff::ONE); + } + input.advance_to(Timestamp::from(max_time + 1)); + input.flush(); + (producer_is_columnar, passthrough_is_columnar, produced) + }) + }); + + assert!( + producer_is_columnar, + "a non-identity MFP must produce a columnar edge" + ); + assert!( + passthrough_is_columnar, + "the arrange input must keep the columnar passthrough (no ColumnarToVec)" + ); + assert_eq!(extract_row_updates(produced), expected); + } + + /// The reworked identity fast-path returns the unarranged input edge + /// unchanged, so a columnar producer stays columnar and a `Vec` producer + /// stays `Vec` with no repack in either direction. + #[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::(); + let (_err_input, errs) = scope.new_collection::(); + let edge = if columnar_input { + CollectionEdge::Columnar(vec_to_columnar(collection)) + } else { + CollectionEdge::Vec(collection) + }; + let bundle = CollectionBundle::from_edge(edge, errs); + let identity = MapFilterProject::::new(1) + .into_plan() + .expect("identity mfp"); + let (out, _errs) = + bundle.as_collection_core(identity, None, Antichain::new(), &config_set); + let is_columnar = matches!(out, CollectionEdge::Columnar(_)); + input.update(Row::pack_slice(&[Datum::Int64(1)]), Diff::ONE); + input.advance_to(Timestamp::from(1u64)); + input.flush(); + is_columnar + }) + }); + assert_eq!( + is_columnar, columnar_input, + "the identity fast-path must preserve the input edge variant" + ); + } + } + + /// The columnar producer folds within-batch duplicates: input rows that + /// project to the same output row at the same time collapse to one record + /// with summed diff, matching the row-based `ConsolidatingContainerBuilder` + /// this replaced. A plain `ColumnBuilder` would emit both records. + #[mz_ore::test] + fn as_collection_core_consolidates_within_batch() { + // `[1, 10]` and `[1, 20]` both project (dropping column 1) to `[1]` at + // t=0, so their `+1` diffs must fold to `+2`. `[2, 30]` projects to a + // distinct record. + let rows = vec![ + Row::pack_slice(&[Datum::Int64(1), Datum::Int64(10)]), + Row::pack_slice(&[Datum::Int64(1), Datum::Int64(20)]), + Row::pack_slice(&[Datum::Int64(2), Datum::Int64(30)]), + ]; + let mfp = MapFilterProject::::new(2) + .project(vec![0]) + .into_plan() + .expect("mfp"); + let expected = vec![ + ( + Row::pack_slice(&[Datum::Int64(1)]), + Timestamp::from(0u64), + Diff::from(2), + ), + ( + Row::pack_slice(&[Datum::Int64(2)]), + Timestamp::from(0u64), + Diff::ONE, + ), + ]; + + 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); + assert!( + matches!(edge, CollectionEdge::Columnar(_)), + "a non-identity MFP must produce a columnar edge" + ); + let captured = edge.into_vec().inner.capture(); + // Feed all rows at the same time in one batch so the fold is + // within-batch, not a downstream re-consolidation. + for row in rows { + input.update_at(row, Timestamp::from(0u64), Diff::ONE); + } + input.advance_to(Timestamp::from(1u64)); + input.flush(); + captured + }) + }); + + assert_eq!(extract_row_updates(captured), expected); + } } diff --git a/src/compute/src/render/reduce.rs b/src/compute/src/render/reduce.rs index 3d896f2c4eed3..57487a1970491 100644 --- a/src/compute/src/render/reduce.rs +++ b/src/compute/src/render/reduce.rs @@ -111,7 +111,7 @@ impl<'scope, T: RenderTimestamp> Context<'scope, T> { let (key_val_input, err) = input .enter_region(inner) - .flat_map::<_, ConsolidatingContainerBuilder>, _>( + .flat_map::>, _>( input_key.map(|k| (k, None)), max_demand, move |row_datums, time, diff, ok_session, err_session| { diff --git a/src/compute/src/render/sinks.rs b/src/compute/src/render/sinks.rs index c96e0be462b7b..b66807febc5ac 100644 --- a/src/compute/src/render/sinks.rs +++ b/src/compute/src/render/sinks.rs @@ -80,12 +80,16 @@ impl<'g, T: RenderTimestamp> Context<'g, T> { let mut mfp = MapFilterProject::::new(unthinned_arity); mfp.permute_fn(|c| permutation[c], thinning.len() + key.len()); let mfp_plan = mfp.into_plan().expect("MFP planning failed"); - bundle.as_collection_core( + // 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, - ) + ); + (oks.into_vec(), errs) }; // Attach logging of dataflow errors. From 45cd71bacb7e2d48e6b464d34075ecec228469bf Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Thu, 10 Sep 2026 20:17:14 +0200 Subject: [PATCH 2/2] doc: tighten the columnar Get/Mfp-output comments Give the missing-push-bound rationale a single owner, drop the comparisons against the row-based builder, and cut the test doc to what the asserts do not already state. Co-Authored-By: Claude Opus 5 (1M context) --- src/compute/src/render/context.rs | 45 +++++++++---------------------- 1 file changed, 12 insertions(+), 33 deletions(-) diff --git a/src/compute/src/render/context.rs b/src/compute/src/render/context.rs index afce351953452..f0e33fa7f5c94 100644 --- a/src/compute/src/render/context.rs +++ b/src/compute/src/render/context.rs @@ -766,9 +766,8 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> { for<'a> BatchCursor: Cursor: ExtendDatums, Val<'a>: ExtendDatums, Time = T, Diff = mz_repr::Diff>, < as Cursor>::KeyContainer as BatchContainer>::Owned: PartialEq, - // The builder accepts whatever `logic` gives it, so the push bound lives at the `give` - // call site rather than here. This lets a caller push borrowed records into a columnar - // builder that has no owned-tuple `Push`. + // No push bound here: it lives at `logic`'s `give` call site, so a caller can push + // borrowed records into a columnar builder that has no owned-tuple `Push`. DCB: ContainerBuilder, // `logic` receives the key and value already decoded into a `DatumVecBorrow`. The decode // (and its arena/`DatumVec`) lives in the per-activation closure below, so it is scoped to @@ -1015,15 +1014,12 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> { if mfp_plan.is_identity() && !has_key_val { let key = key_val.map(|(k, _v)| k); return match key { - // Unarranged identity: hand the edge straight through, so a - // columnar producer stays columnar without a `ColumnarToVec` hop. + // Unarranged identity hands the edge straight through, so a columnar + // producer stays columnar. None => self .collection .clone() .expect("The unarranged collection doesn't exist."), - // Keyed identity reads an existing arrangement, which is - // row-based. Wrap it as a `Vec` edge. `as_specific_collection` - // stays the consumer leaf. Some(key) => { let (oks, errs) = self.as_specific_collection(Some(&key), config_set); (CollectionEdge::Vec(oks), errs) @@ -1048,13 +1044,9 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> { // Wrap in an `Rc` so that lifetimes work out. let until = std::rc::Rc::new(until); - // The ok output is built into a `Column`, so this producer emits the - // columnar edge. `ConsolidatingColumnBuilder` folds within-batch - // duplicates, matching the row-based `ConsolidatingContainerBuilder` - // this replaced. It stages owned `(Row, T, Diff)` tuples to consolidate - // in place, so the records are given owned; `mfp_plan.evaluate` already - // produces a fresh owned `Row` per result, so this is a move into - // staging, not a new allocation. + // `ConsolidatingColumnBuilder` folds within-batch duplicates. It consolidates in + // place, so records are given owned, which costs nothing here because + // `mfp_plan.evaluate` already produces a fresh `Row` per result. let (stream, errors) = self.flat_map::, _>( key_val, max_demand, @@ -1161,9 +1153,7 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> { .try_into() .expect("must fit"); bucketed = true; - // Temporal bucketing consumes and produces a `Vec` edge, so - // decode here. This is the sanctioned leaf decode where a - // `Vec`-internal operator meets the columnar edge. + // Temporal bucketing is `Vec`-internal, so decode here. CollectionEdge::Vec(T::maybe_apply_temporal_bucketing( oks.into_vec().inner, as_of.clone(), @@ -1456,8 +1446,6 @@ where ok_output: &mut OutputBuilderSession<'_, C::Time, DCB>, err_output: &mut OutputBuilderSession<'_, C::Time, ECB>, ) where - // The push bound lives at `logic`'s `give` call site, not here, so a caller can push - // borrowed records into a columnar builder that has no owned-tuple `Push`. DCB: ContainerBuilder, L: FnMut( C::Key<'_>, @@ -1742,14 +1730,8 @@ mod tests { updates } - /// A `Get -> ArrangeBy` chain carries the columnar arm end to end. A - /// non-identity MFP drives `as_collection_core` down its columnar producer - /// path, and feeding that edge into the arrange input keeps the columnar - /// passthrough, so no `ColumnarToVec` sits on the arrange path. - /// - /// The producer output is checked against the projected input. Arrange - /// correctness itself is covered by `arrange_collection_arms_agree`; here we - /// only assert the variant survives the hand-off. + /// Arrange correctness itself is covered by `arrange_collection_arms_agree`; this only + /// asserts the columnar variant survives the hand-off. #[mz_ore::test] fn get_arrange_by_carries_columnar_end_to_end() { let rows = vec![ @@ -1757,9 +1739,8 @@ mod tests { (Row::pack_slice(&[Datum::Int64(2), Datum::Int64(20)]), 1), (Row::pack_slice(&[Datum::Int64(1), Datum::Int64(10)]), 1), ]; - // Project away column 1; the output row carries only column 0. A - // projection is non-identity, so `as_collection_core` takes the columnar - // producer path rather than the identity passthrough. + // A projection is non-identity, so `as_collection_core` takes the producer path + // rather than the identity passthrough. let mfp = MapFilterProject::::new(2) .project(vec![0]) .into_plan() @@ -1783,8 +1764,6 @@ mod tests { let (edge, _errs) = bundle.as_collection_core(mfp, None, Antichain::new(), &config_set); 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. let produced = edge.clone().into_vec().inner.capture(); let (_arranged, _arrange_errs, passthrough) = CollectionBundle::::arrange_collection(