compute: columnar Constant output

src/compute/src/render.rs

@@ -146,6 +146,7 @@ use mz_timely_util::scope_label::ScopeExt;
 use timely::PartialOrder;
 use timely::container::CapacityContainerBuilder;
 use timely::dataflow::channels::pact::Pipeline;
+use timely::dataflow::operators::core::to_stream::ToStreamBuilder;
 use timely::dataflow::operators::vec::ToStream;
 use timely::dataflow::operators::vec::{BranchWhen, Filter};
 use timely::dataflow::operators::{Capability, Operator, Probe, probe};
@@ -169,6 +170,7 @@ use crate::render::context::{ArrangementFlavor, Context};
 use crate::render::errors::DataflowErrorSer;
 use crate::typedefs::{ErrBatcher, ErrBuilder, ErrSpine, KeyBatcher, MzTimestamp};
 use mz_row_spine::{DatumSeq, RowRowBatcher, RowRowBuilder};
+use mz_timely_util::columnar::consolidate::ConsolidatingColumnBuilder;
 
 pub(crate) mod columnar;
 pub mod context;
@@ -1177,22 +1179,31 @@ impl<'scope, T: RenderTimestamp + MaybeBucketByTime> Context<'scope, T> {
                 // We should advance times in constant collections to start from `as_of`.
                 let as_of_frontier = self.as_of_frontier.clone();
                 let until = self.until.clone();
-                let ok_collection = rows
-                    .into_iter()
-                    .filter_map(move |(row, mut time, diff)| {
-                        time.advance_by(as_of_frontier.borrow());
-                        if !until.less_equal(&time) {
-                            Some((
-                                row,
-                                <T as Refines<mz_repr::Timestamp>>::to_inner(time),
-                                diff,
-                            ))
-                        } else {
-                            None
-                        }
-                    })
-                    .to_stream(self.scope)
-                    .as_collection();
+                // Build the ok rows columnar, so this literal source emits the
+                // columnar edge. Advancing times to `as_of` can collapse
+                // distinct original times onto one time, so rows the planner
+                // left distinct may become duplicates here; a
+                // `ConsolidatingColumnBuilder` folds those within the batch (the
+                // rows are already owned, so the give is a move into staging).
+                let ok_collection = CollectionEdge::Columnar(
+                    rows.into_iter()
+                        .filter_map(move |(row, mut time, diff)| {
+                            time.advance_by(as_of_frontier.borrow());
+                            if !until.less_equal(&time) {
+                                Some((
+                                    row,
+                                    <T as Refines<mz_repr::Timestamp>>::to_inner(time),
+                                    diff,
+                                ))
+                            } else {
+                                None
+                            }
+                        })
+                        .to_stream_with_builder::<_, ConsolidatingColumnBuilder<Row, T, Diff>>(
+                            self.scope,
+                        )
+                        .as_collection(),
+                );
 
                 let mut error_time: mz_repr::Timestamp = Timestamp::minimum();
                 error_time.advance_by(self.as_of_frontier.borrow());
@@ -1208,7 +1219,7 @@ impl<'scope, T: RenderTimestamp + MaybeBucketByTime> Context<'scope, T> {
                     .to_stream(self.scope)
                     .as_collection();
 
-                CollectionBundle::from_collections(ok_collection, err_collection)
+                CollectionBundle::from_edge(ok_collection, err_collection)
             }
             Get { id, keys, plan } => {
                 // Recover the collection from `self` and then apply `mfp` to it.

test/sqllogictest/degenerate.slt

@@ -19,3 +19,36 @@ INSERT INTO t1 VALUES (NULL)
 query I
 SELECT * FROM t1 WHERE NOT (f1 = f1)
 ----
+
+# Constant collections render into a columnar edge (node P2). Duplicate constant
+# rows land at the same (advanced) time, so the producer folds them within the
+# batch; the query result must still show the correct multiplicity.
+query I rowsort
+SELECT a FROM (VALUES (1), (1), (2)) t(a)
+----
+1
+1
+2
+
+# A constant feeding an indexed view exercises the columnar producer against an
+# ArrangeBy consumer, and the aggregate below against a Reduce consumer.
+statement ok
+CREATE VIEW cv AS SELECT a FROM (VALUES (1), (1), (2), (3)) t(a)
+
+statement ok
+CREATE DEFAULT INDEX ON cv
+
+query II rowsort
+SELECT a, count(*) FROM cv GROUP BY a
+----
+1
+2
+2
+1
+3
+1
+
+query I
+SELECT sum(a) FROM cv
+----
+7