compute: Re-encode temporal-bucketing output as the columnar edge

Temporal bucketing (`maybe_apply_temporal_bucketing`) is `Vec`-internal:
it consumes and produces a `StreamVec`. The four sites that drive it
previously decoded the edge with `into_vec` and then re-wrapped the `Vec`
result as `CollectionEdge::Vec`, leaving the last `Vec` producer that fed
a Union's `concat_many`.

Re-encode each bucketed result with `vec_to_columnar` so the output edge
is columnar, matching every other producer after the producer wave. The
bucketer stays on `Vec` internally; only its output boundary changes.
This is a non-consolidating leaf encode, since the bucketer output is not
consolidated and the prior `CollectionEdge::Vec` wrap was non-consolidating.

With every producer now columnar, a consolidating Union that mixes a
bucketed input with a `Direct` input feeds `concat_many` two columnar
edges instead of a mixed pair.

Adds runtime coverage to `temporal_bucketing.slt` under
`enable_compute_temporal_bucketing`: bucketed Reduce, bucketed TopK, and
an `EXCEPT ALL` whose Union carries `[TemporalBucketing, Direct]`.

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

Author: antiguru

src/compute/src/render.rs

@@ -1378,11 +1378,17 @@ impl<'scope, T: RenderTimestamp + MaybeBucketByTime> Context<'scope, T> {
                             .get(&self.config_set)
                             .try_into()
                             .expect("must fit");
+                        // Temporal bucketing (node C8) is `Vec`-internal: decode
+                        // the edge into it, then re-encode the `Vec` result to
+                        // columnar so this Union input stays columnar and
+                        // `concat_many` sees no mixed variants.
                         let os = os.into_vec();
-                        CollectionEdge::Vec(T::maybe_apply_temporal_bucketing(
-                            os.inner,
-                            self.as_of_frontier.clone(),
-                            summary,
+                        CollectionEdge::Columnar(vec_to_columnar(
+                            T::maybe_apply_temporal_bucketing(
+                                os.inner,
+                                self.as_of_frontier.clone(),
+                                summary,
+                            ),
                         ))
                     } else {
                         os

src/compute/src/render/context.rs

@@ -51,7 +51,7 @@ use timely::progress::{Antichain, Timestamp};
 
 use crate::compute_state::ComputeState;
 use crate::extensions::arrange::{KeyCollection, MzArrange, MzArrangeCore};
-use crate::render::columnar::CollectionEdge;
+use crate::render::columnar::{CollectionEdge, vec_to_columnar};
 use crate::render::errors::{DataflowErrorSer, ErrorLogger};
 use crate::render::{LinearJoinSpec, MaybeBucketByTime, RenderTimestamp};
 use crate::typedefs::{
@@ -1083,14 +1083,15 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> {
                     .try_into()
                     .expect("must fit");
                 bucketed = true;
-                // Temporal bucketing (node C8) still consumes and produces a
-                // `Vec` edge, so decode here. This leaf decode is the sanctioned
-                // seam until the bucketing operator learns the columnar form.
-                CollectionEdge::Vec(T::maybe_apply_temporal_bucketing(
+                // Temporal bucketing (node C8) is `Vec`-internal: it consumes and
+                // produces a `Vec` stream. Decode the edge into it, then re-encode
+                // the `Vec` result to columnar at the boundary so the bucketed
+                // output edge stays columnar like every other producer.
+                CollectionEdge::Columnar(vec_to_columnar(T::maybe_apply_temporal_bucketing(
                     oks.into_vec().inner,
                     as_of.clone(),
                     summary,
-                ))
+                )))
             } else {
                 oks
             };
@@ -1114,27 +1115,26 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> {
                 } else {
                     strategy
                 };
-                let oks = if matches!(effective_strategy, ArrangementStrategy::TemporalBucketing)
-                    && ENABLE_COMPUTE_TEMPORAL_BUCKETING.get(config_set)
-                {
-                    let summary: mz_repr::Timestamp = TEMPORAL_BUCKETING_SUMMARY
-                        .get(config_set)
-                        .try_into()
-                        .expect("must fit");
-                    bucketed = true;
-                    // Temporal bucketing (node C8) still consumes and produces a
-                    // `Vec` edge, so decode here. This leaf decode is the
-                    // sanctioned seam until the bucketing operator learns the
-                    // columnar form.
-                    let oks = oks.into_vec();
-                    CollectionEdge::Vec(T::maybe_apply_temporal_bucketing(
-                        oks.inner,
-                        as_of.clone(),
-                        summary,
-                    ))
-                } else {
-                    oks
-                };
+                let oks =
+                    if matches!(effective_strategy, ArrangementStrategy::TemporalBucketing)
+                        && ENABLE_COMPUTE_TEMPORAL_BUCKETING.get(config_set)
+                    {
+                        let summary: mz_repr::Timestamp = TEMPORAL_BUCKETING_SUMMARY
+                            .get(config_set)
+                            .try_into()
+                            .expect("must fit");
+                        bucketed = true;
+                        // Temporal bucketing (node C8) is `Vec`-internal: it consumes
+                        // and produces a `Vec` stream. Decode the edge into it, then
+                        // re-encode the `Vec` result to columnar at the boundary so the
+                        // bucketed output edge stays columnar like every other producer.
+                        let oks = oks.into_vec();
+                        CollectionEdge::Columnar(vec_to_columnar(
+                            T::maybe_apply_temporal_bucketing(oks.inner, as_of.clone(), summary),
+                        ))
+                    } else {
+                        oks
+                    };
                 let use_paged_path = ENABLE_COLUMN_PAGED_BATCHER.get(config_set);
                 let (oks, errs_keyed, passthrough) = Self::arrange_collection(
                     &name,

src/compute/src/render/top_k.rs

@@ -49,7 +49,7 @@ use timely::dataflow::operators::generic::builder_rc::OperatorBuilder;
 use crate::extensions::arrange::{ArrangementSize, KeyCollection, MzArrange};
 use crate::extensions::reduce::{ClearContainer, MzReduce};
 use crate::render::Pairer;
-use crate::render::columnar::CollectionEdge;
+use crate::render::columnar::{CollectionEdge, vec_to_columnar};
 use crate::render::context::{ArrangementFlavor, CollectionBundle, Context};
 use crate::render::errors::DataflowErrorSer;
 use crate::render::errors::MaybeValidatingRow;
@@ -110,11 +110,12 @@ impl<'scope, T: crate::render::RenderTimestamp + crate::render::MaybeBucketByTim
                  `mz_now()` has been const-folded and no temporal bucketing is set",
             );
         }
-        // Temporal bucketing consumes and produces a `Vec` stream, so decode the
-        // edge here. This is the sanctioned leaf decode (node C8 analog). It only
-        // fires under `ENABLE_COMPUTE_TEMPORAL_BUCKETING` and the `TemporalBucketing`
-        // strategy, both off on the common path, so the columnar edge otherwise
-        // flows straight through.
+        // Temporal bucketing is `Vec`-internal: it consumes and produces a `Vec`
+        // stream. Decode the edge into it, then re-encode the `Vec` result to
+        // columnar at the boundary so the bucketed input edge stays columnar. It
+        // only fires under `ENABLE_COMPUTE_TEMPORAL_BUCKETING` and the
+        // `TemporalBucketing` strategy, both off on the common path, so the
+        // columnar edge otherwise flows straight through.
         let ok_input = if matches!(
             temporal_bucketing_strategy,
             ArrangementStrategy::TemporalBucketing
@@ -124,11 +125,11 @@ impl<'scope, T: crate::render::RenderTimestamp + crate::render::MaybeBucketByTim
                 .get(&self.config_set)
                 .try_into()
                 .expect("must fit");
-            CollectionEdge::Vec(T::maybe_apply_temporal_bucketing(
+            CollectionEdge::Columnar(vec_to_columnar(T::maybe_apply_temporal_bucketing(
                 ok_input.into_vec().inner,
                 self.as_of_frontier.clone(),
                 summary,
-            ))
+            )))
         } else {
             ok_input
         };

test/sqllogictest/temporal_bucketing.slt

@@ -276,3 +276,114 @@ materialize.public.mv_except_distinct_buckets_at_reduce:
 Target cluster: quickstart
 
 EOF
+
+# -----------------------------------------------------------------------------
+# Runtime tests (node Ta). With `enable_compute_temporal_bucketing` on, the
+# bucketed dataflow edges are re-encoded to the columnar representation instead
+# of re-wrapping `Vec`. These tests turn the flag on and assert the bucketed
+# dataflows still produce the correct logical results, and that a consolidating
+# `Union` concatenating a bucketed input with a `Direct` input yields the right
+# output (the mixed `concat_many` case, now columnar on both legs).
+# -----------------------------------------------------------------------------
+
+simple conn=mz_system,user=mz_system
+ALTER SYSTEM SET enable_compute_temporal_bucketing = true
+----
+COMPLETE 0
+
+statement ok
+CREATE TABLE rt_events (k INT NOT NULL, event_time TIMESTAMP NOT NULL)
+
+# Far-future event times so the temporal predicates hold regardless of the
+# wall clock at query time, keeping the results deterministic.
+statement ok
+INSERT INTO rt_events VALUES
+  (1, '2999-01-01 00:00:00'),
+  (1, '2999-01-02 00:00:00'),
+  (2, '2999-01-03 00:00:00'),
+  (3, '2999-01-04 00:00:00')
+
+statement ok
+CREATE TABLE rt_other (k INT NOT NULL)
+
+statement ok
+INSERT INTO rt_other VALUES (2), (99)
+
+# Bucketed Reduce: temporal filter above a GROUP BY. Hits the arrangement
+# re-encode in `context.rs`.
+statement ok
+CREATE MATERIALIZED VIEW rt_reduce AS
+SELECT k, count(*)
+FROM rt_events
+WHERE event_time + INTERVAL '45 day' > mz_now()
+GROUP BY k
+
+query II rowsort
+SELECT * FROM rt_reduce
+----
+1  2
+2  1
+3  1
+
+# Bucketed TopK: temporal filter under ORDER BY ... LIMIT. Hits the re-encode
+# in `top_k.rs`.
+statement ok
+CREATE MATERIALIZED VIEW rt_topk AS
+SELECT k
+FROM rt_events
+WHERE event_time + INTERVAL '45 day' > mz_now()
+ORDER BY event_time
+LIMIT 3
+
+query I rowsort
+SELECT * FROM rt_topk
+----
+1
+1
+2
+
+# Mixed Union: `EXCEPT ALL` of a temporal-filtered leg (bucketed) against a
+# plain relation (Direct) lowers to a consolidating `Union` with per-input
+# strategies `[TemporalBucketing, Direct]`. After node Ta both legs reach
+# `concat_many` as columnar edges.
+query T multiline
+EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR
+CREATE MATERIALIZED VIEW rt_union AS
+SELECT k FROM rt_events WHERE event_time + INTERVAL '45 day' > mz_now()
+EXCEPT ALL
+SELECT k FROM rt_other
+----
+materialize.public.rt_union:
+  Threshold::Basic ensure_arrangement={ key=[#0], permutation=id, thinning=() }
+    ArrangeBy
+      raw=false
+      arrangements[0]={ key=[#0], permutation=id, thinning=() }
+      Union consolidate_output=true
+        temporal_bucketing_strategies=[TemporalBucketing, Direct]
+        Get::Collection materialize.public.rt_events
+          raw=true
+        Negate
+          Get::PassArrangements materialize.public.rt_other
+            raw=true
+
+Source materialize.public.rt_events
+  project=(#0)
+  filter=((mz_now() < timestamp_to_mz_timestamp((#1{event_time} + 45 days))))
+Source materialize.public.rt_other
+
+Target cluster: quickstart
+
+EOF
+
+statement ok
+CREATE MATERIALIZED VIEW rt_union AS
+SELECT k FROM rt_events WHERE event_time + INTERVAL '45 day' > mz_now()
+EXCEPT ALL
+SELECT k FROM rt_other
+
+query I rowsort
+SELECT * FROM rt_union
+----
+1
+1
+3