From 293c46eb28daf05b082ba7b9a96adf4af938ad80 Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Fri, 11 Sep 2026 23:59:17 +0200 Subject: [PATCH 1/2] test: cover the arrangement bucketing site and name the reduce one The runtime fixtures claimed to cover the arrangement re-encode in `ensure_collections`, but a temporal filter above a GROUP BY buckets the keyed `(key, val)` stream inside `render_reduce`, which bypasses `ensure_collections` entirely. So no fixture reached either arrangement site, and the one site whose result propagates was untested: `arrange_collection` hands back a passthrough that becomes the bundle's collection for every later consumer. Add an index on a temporal-filtered view, which routes through the loop that builds the requested arrangements, and pin its plan so the fixture cannot silently stop reaching the site if lowering stops marking the `ArrangeBy`. Point the reduce fixture's comment at `render_reduce`, and rewrite the block header to say which site each fixture covers. Two neighboring comments went stale when bucketing became columnar-native and the edge collapsed to a single representation: there is no re-encode at the TopK site any more, and both legs of the mixed `Union` are columnar whatever their strategy, so the interesting part is that one leg skipped the bucketer. Co-Authored-By: Claude Opus 5 (1M context) --- test/sqllogictest/temporal_bucketing.slt | 69 +++++++++++++++++++----- 1 file changed, 57 insertions(+), 12 deletions(-) diff --git a/test/sqllogictest/temporal_bucketing.slt b/test/sqllogictest/temporal_bucketing.slt index 217b2d104ab49..63f2811dd2fc5 100644 --- a/test/sqllogictest/temporal_bucketing.slt +++ b/test/sqllogictest/temporal_bucketing.slt @@ -278,12 +278,12 @@ 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. +# Runtime tests. These turn `enable_compute_temporal_bucketing` on and assert +# that the bucketed dataflows still produce the correct logical results. There +# is one fixture per site that applies bucketing, because each site hands the +# bucketer a differently shaped stream: the keyed `(key, val)` stream in +# `render_reduce`, the TopK input, a consolidating `Union` whose legs carry +# different strategies, and an arrangement built by `ensure_collections`. # ----------------------------------------------------------------------------- simple conn=mz_system,user=mz_system @@ -309,8 +309,9 @@ 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`. +# Bucketed Reduce: temporal filter above a GROUP BY. `render_reduce` buckets the +# keyed `(key, val)` stream itself, so this reaches `reduce.rs` rather than +# either arrangement site. statement ok CREATE MATERIALIZED VIEW rt_reduce AS SELECT k, count(*) @@ -325,8 +326,8 @@ SELECT * FROM rt_reduce 2 1 3 1 -# Bucketed TopK: temporal filter under ORDER BY ... LIMIT. Hits the re-encode -# in `top_k.rs`. +# Bucketed TopK: temporal filter under ORDER BY ... LIMIT. Buckets the TopK +# input in `top_k.rs`. statement ok CREATE MATERIALIZED VIEW rt_topk AS SELECT k @@ -344,8 +345,8 @@ SELECT * FROM rt_topk # 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. +# strategies `[TemporalBucketing, Direct]`, so `concat_many` sees one bucketed +# leg and one that skipped the bucketer. query T multiline EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR CREATE MATERIALIZED VIEW rt_union AS @@ -387,3 +388,47 @@ SELECT * FROM rt_union 1 1 3 + +# Bucketed arrangement: an index on a temporal-filtered view. `ensure_collections` +# buckets in the loop that builds the requested arrangements, and hands the result +# to `arrange_collection`, whose passthrough then serves as the bundle's collection +# for every later consumer. The plan is pinned because the site is only reached +# while the `ArrangeBy` carries `strategy=TemporalBucketing`. +statement ok +CREATE VIEW rt_indexed AS +SELECT k FROM rt_events WHERE event_time + INTERVAL '45 day' > mz_now() + +query T multiline +EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR +CREATE DEFAULT INDEX ON rt_indexed +---- +materialize.public.rt_indexed_primary_idx: + ArrangeBy + strategy=TemporalBucketing + raw=true + arrangements[0]={ key=[#0{k}], permutation=id, thinning=() } + Get::PassArrangements materialize.public.rt_indexed + raw=true + +materialize.public.rt_indexed: + Get::Collection materialize.public.rt_events + raw=true + +Source materialize.public.rt_events + project=(#0) + filter=((mz_now() < timestamp_to_mz_timestamp((#1{event_time} + 45 days)))) + +Target cluster: quickstart + +EOF + +statement ok +CREATE DEFAULT INDEX ON rt_indexed + +query I rowsort +SELECT k FROM rt_indexed +---- +1 +1 +2 +3 From 46960e7823c5dd3920a358dd2f12ceb71a1b3dd3 Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Mon, 14 Sep 2026 17:38:38 +0200 Subject: [PATCH 2/2] test: assert the arrange passthrough and drop an expired row The index fixture's arrangement is the terminal consumer, so nothing in that dataflow reads the passthrough collection `arrange_collection` returns, and no unit test asserted it either. Add one that captures the passthrough under an erroring key and checks it forwards every input record unchanged, and reword the fixture comment to claim only what the index shape covers. The runtime rows all sat inside the temporal window, so a bucketer that leaked an expired update would have stayed green. Add a row whose retraction time precedes every fixture's as_of. Each fixture filters on the same predicate below its operator, so the pinned results are unchanged. Co-Authored-By: Claude Fable 5.1 --- src/compute/src/render/context.rs | 42 ++++++++++++++++++++++++ test/sqllogictest/temporal_bucketing.slt | 14 +++++--- 2 files changed, 51 insertions(+), 5 deletions(-) diff --git a/src/compute/src/render/context.rs b/src/compute/src/render/context.rs index dcda673932f39..dde38c4d13a08 100644 --- a/src/compute/src/render/context.rs +++ b/src/compute/src/render/context.rs @@ -1569,6 +1569,48 @@ mod tests { assert!(!err.is_empty()); } + /// The passthrough forwards every input record, including those whose key + /// evaluation errored, so a consumer of the bundle's collection sees the + /// unarranged input rather than the ok side of the arrangement. + #[mz_ore::test] + fn arrange_collection_passthrough_forwards_input() { + let rows = test_rows(); + let mut expected: Vec<(Row, Timestamp, Diff)> = rows + .iter() + .map(|(row, t)| (row.clone(), Timestamp::from(*t), Diff::ONE)) + .collect(); + expected.sort(); + + let key = vec![LirScalarExpr::literal( + Err(EvalError::DivisionByZero), + ReprScalarType::Int32, + )]; + let captured = timely::execute_directly(move |worker| { + worker.dataflow::(|scope| { + let (mut input, collection) = scope.new_collection(); + let (_arranged, _errs, passthrough) = + CollectionBundle::::arrange_collection( + &"col".to_string(), + vec_to_columnar(collection), + key, + vec![0, 1], + ArrangementBatcher::Columnation, + ); + let captured = columnar_to_vec(passthrough).inner.capture(); + + let max_time = rows.iter().map(|(_, t)| *t).max().unwrap_or(0); + for (row, time) in rows { + input.update_at(row, Timestamp::from(time), Diff::ONE); + } + input.advance_to(Timestamp::from(max_time + 1)); + input.flush(); + captured + }) + }); + + assert_eq!(extract_row_updates(captured), expected); + } + fn extract_row_updates( captured: Captured<(Row, Timestamp, Diff)>, ) -> Vec<(Row, Timestamp, Diff)> { diff --git a/test/sqllogictest/temporal_bucketing.slt b/test/sqllogictest/temporal_bucketing.slt index 63f2811dd2fc5..0db99d8d87948 100644 --- a/test/sqllogictest/temporal_bucketing.slt +++ b/test/sqllogictest/temporal_bucketing.slt @@ -283,7 +283,7 @@ EOF # is one fixture per site that applies bucketing, because each site hands the # bucketer a differently shaped stream: the keyed `(key, val)` stream in # `render_reduce`, the TopK input, a consolidating `Union` whose legs carry -# different strategies, and an arrangement built by `ensure_collections`. +# different strategies, and the arrangement loop in `ensure_collections`. # ----------------------------------------------------------------------------- simple conn=mz_system,user=mz_system @@ -295,13 +295,16 @@ 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. +# wall clock at query time, keeping the results deterministic. The k=4 row +# expired long ago, so every fixture must drop it: its retraction time precedes +# the dataflow's as_of, which is the case a leaking bucketer would get wrong. 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') + (3, '2999-01-04 00:00:00'), + (4, '2000-01-01 00:00:00') statement ok CREATE TABLE rt_other (k INT NOT NULL) @@ -391,8 +394,9 @@ SELECT * FROM rt_union # Bucketed arrangement: an index on a temporal-filtered view. `ensure_collections` # buckets in the loop that builds the requested arrangements, and hands the result -# to `arrange_collection`, whose passthrough then serves as the bundle's collection -# for every later consumer. The plan is pinned because the site is only reached +# to `arrange_collection`. The index is the terminal consumer, so this covers the +# bucketer feeding the arrangement and not the passthrough collection, which has +# no reader in this shape. The plan is pinned because the site is only reached # while the `ArrangeBy` carries `strategy=TemporalBucketing`. statement ok CREATE VIEW rt_indexed AS