compute: columnar Constant output (P2)

Build the `Constant` arm of `render_plan_expr` into a `Column` and return a
`CollectionEdge::Columnar`, flipping the constant literal source to emit the
columnar edge. The err collection stays row-based (out of scope).

The rows go through a `ConsolidatingColumnBuilder` via `to_stream_with_builder`
rather than a direct columnar build. The planner (`FoldConstants`) consolidates
constant rows by `(row, time)` at optimization time, but this arm then advances
every time to `as_of`, which can collapse distinct original times onto one
time and so reintroduce duplicates at the same `(row, time)`. Consolidating here
folds those within the batch, matching the standing producer rule. The constant
rows are already owned, so the give is a move into staging, not a new
allocation. The `as_of` advancement and `until` filtering are unchanged.

Test: extends degenerate.slt with a constant carrying duplicate rows (asserting
multiplicity survives the columnar edge) and a constant feeding an indexed view
and an aggregate, exercising the columnar producer against ArrangeBy and Reduce
consumers.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: antiguru

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