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

src/compute/src/render.rs

@@ -1416,11 +1416,16 @@ impl<'scope, T: RenderTimestamp + MaybeBucketByTime> Context<'scope, T> {
                             .get(&self.config_set)
                             .try_into()
                             .expect("must fit");
+                        // Temporal bucketing operates on `Vec`: decode the edge
+                        // into it, then re-encode the result so this Union input
+                        // is a columnar edge like every other.
                         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

@@ -54,7 +54,7 @@ use timely::progress::{Antichain, Timestamp};
 use crate::compute_state::ComputeState;
 use crate::extensions::arrange::{ArrangementBatcher, KeyCollection, MzArrange, MzArrangeCore};
 use crate::extensions::reduce::MzReduce;
-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::{
@@ -1168,14 +1168,15 @@ 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.
-                CollectionEdge::Vec(T::maybe_apply_temporal_bucketing(
+                // 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
+                // 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
             };
@@ -1199,26 +1200,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 consumes and produces a `Vec` edge, so
-                    // decode here. This is the sanctioned leaf decode where a
-                    // `Vec`-internal operator meets the columnar edge.
-                    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 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 batcher = ArrangementBatcher::from_config(config_set);
                 let (oks, errs_keyed, passthrough) =
                     Self::arrange_collection(&name, oks, key.clone(), thinning.clone(), batcher);

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. 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
         };
@@ -162,8 +163,7 @@ impl<'scope, T: crate::render::RenderTimestamp + crate::render::MaybeBucketByTim
                     // per-row evaluation only runs for column or otherwise fallible
                     // limits. On the `Vec` edge `into_vec` is the identity, so the
                     // `Vec` path is unchanged. On a columnar edge it is a narrow
-                    // sanctioned decode
-                    // confined to this rare path.
+                    // sanctioned decode confined to this rare path.
                     let errors = ok_input.clone().into_vec().flat_map(move |row| {
                         let temp_storage = mz_repr::RowArena::new();
                         let datums = datum_vec.borrow_with(&row);

test/sqllogictest/temporal_bucketing.slt

@@ -276,3 +276,114 @@ materialize.public.mv_except_distinct_buckets_at_reduce:
 Target cluster: quickstart
 
 EOF
+
+# -----------------------------------------------------------------------------
+# Runtime tests. 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, which reaches `concat_many` as a columnar edge like the other leg.
+# -----------------------------------------------------------------------------
+
+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]`. 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