diff --git a/src/compute/src/render.rs b/src/compute/src/render.rs index 1d42ccd0d3956..4fb528880f37e 100644 --- a/src/compute/src/render.rs +++ b/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 diff --git a/src/compute/src/render/context.rs b/src/compute/src/render/context.rs index f02bf55673407..e97067a89b276 100644 --- a/src/compute/src/render/context.rs +++ b/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); diff --git a/src/compute/src/render/top_k.rs b/src/compute/src/render/top_k.rs index 54b48540d72b4..a8d5657032a01 100644 --- a/src/compute/src/render/top_k.rs +++ b/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); diff --git a/test/sqllogictest/temporal_bucketing.slt b/test/sqllogictest/temporal_bucketing.slt index b4ea3956aae72..217b2d104ab49 100644 --- a/test/sqllogictest/temporal_bucketing.slt +++ b/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