@@ -85,9 +85,8 @@ impl<'scope, T: crate::render::RenderTimestamp> Context<'scope, T> {
8585
8686/// Output ok-session container builder for [`flat_map_stage`].
8787///
88- /// Consolidating like the err builder, but emits `Column<(Row, T, Diff)>` so
89- /// the FlatMap output travels as the columnar edge. Output rows are freshly
90- /// built by the mfp, so the owned give into staging is a move, not a new alloc.
88+ /// Consolidating like the err builder, but emits `Column<(Row, T, Diff)>`. The mfp builds
89+ /// each output row fresh, so the owned give into staging is a move.
9190type FlatMapOk < T > = ConsolidatingColumnBuilder < Row , T , Diff > ;
9291/// Output err-session container builder for [`flat_map_stage`].
9392type FlatMapErr < T > = ConsolidatingContainerBuilder < Vec < ( DataflowErrorSer , T , Diff ) > > ;
@@ -479,8 +478,7 @@ mod tests {
479478 assert_eq ! ( vec_updates, extract_sorted_columns( col_captured) ) ;
480479 }
481480
482- /// Decodes a capture of the columnar FlatMap output into sorted owned
483- /// `(row, time, diff)` updates.
481+ /// Decodes a capture of the columnar output into sorted `(row, time, diff)` updates.
484482 fn extract_sorted_columns (
485483 captured : std:: sync:: mpsc:: Receiver <
486484 timely:: dataflow:: operators:: capture:: Event < Timestamp , Column < ( Row , Timestamp , Diff ) > > ,
@@ -508,10 +506,8 @@ mod tests {
508506
509507 #[ mz_ore:: test]
510508 fn flat_map_output_consolidates_within_batch ( ) {
511- // Two distinct input rows whose table-function expansions overlap once
512- // the mfp projects away the differing `stop` column. The overlapping
513- // output rows land at the same time in one batch, so the consolidating
514- // columnar output builder must fold them into summed diffs.
509+ // Two input rows whose expansions overlap once the mfp projects away the
510+ // differing `stop` column, so the output builder has duplicates to fold.
515511 let captured = timely:: execute_directly ( move |worker| {
516512 worker. dataflow :: < Timestamp , _ , _ > ( |scope| {
517513 let ( mut input, collection) = scope. new_collection ( ) ;
@@ -521,8 +517,7 @@ mod tests {
521517 LirScalarExpr :: literal_ok( Datum :: Int64 ( 1 ) , ReprScalarType :: Int64 ) ,
522518 ] ;
523519 let func = TableFunc :: GenerateSeriesInt64 ;
524- // Project to only the generated value (column 2), collapsing the
525- // two input rows' distinct (start, stop) prefixes.
520+ // Project to the generated value alone, collapsing the distinct prefixes.
526521 let mfp = MapFilterProject :: < LirScalarExpr > :: new ( 3 )
527522 . project ( vec ! [ 2 ] )
528523 . into_plan ( )
@@ -539,9 +534,7 @@ mod tests {
539534 usize:: MAX ,
540535 ) ;
541536 let captured = oks. capture ( ) ;
542- // Both rows at t=0: generate_series(1, 2) -> {1, 2},
543- // generate_series(1, 3) -> {1, 2, 3}. Generated 1 and 2 appear on
544- // both, so they must fold to a diff of two.
537+ // Both at t=0: {1, 2} and {1, 2, 3}, so 1 and 2 fold to a diff of two.
545538 input. advance_to ( Timestamp :: from ( 0_u64 ) ) ;
546539 input. update ( input_row ( 2 ) , Diff :: ONE ) ;
547540 input. update ( input_row ( 3 ) , Diff :: ONE ) ;
0 commit comments