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

[antiguru]

Re-encode the temporal-bucketing output back to the columnar edge; the operator itself still works over Vec internally. Removes one mixed-variant edge source ahead of the concat_many teardown.

Columnar dataflow-edge migration. Design doc: doc/developer/design/20260720_columnar_dataflow_edges.md (#37744).

Part of CPU-51.

[linear-code]

CPU-51

[def-]

QA LLM Review

1. MEDIUM -- Bucketed TopK input now round-trips Row → Column → Row

src/compute/src/render/top_k.rs:128

The new vec_to_columnar encode at the TopK bucketing site is undone one operator later: map_topk_key's columnar arm calls Columnar::into_owned(row) for every record, whereas the Vec arm it displaces moved the already-owned Row for free. A bucketed TopK therefore pays a full columnar re-encode plus a second per-record Row allocation that the pre-change path did not, on every input record, and nothing downstream consumes the columnar form.

Details

Chain after the change, for the plan shape the file's own mv_topk_firing fixture pins down (TopK::Basic over Get::Collection, whose import producer already emits columnar):

Columnar → into_vec() = ColumnarToVec (one owned Row per record) → bucket → vec_to_columnar (full byte copy of rows, times, diffs) → map_topk_key columnar arm at top_k.rs:647, Columnar::into_owned(row) at top_k.rs:667 (a second owned Row per record).

Before the change: ColumnarToVec → bucket → map_topk_key Vec arm at top_k.rs:637, which is oks.map(|row| (key_row, row)) and reuses the owned Row. So the seam goes from one decode to decode + encode + decode.

map_topk_key's own doc already records that the columnar arm "does not avoid the per-record decode, because there is no columnar batcher to push borrowed rows into here". Unlike the Union site at render.rs:1414 that the commit message motivates, there is no concat_many at TopK, so the encode is not collapsing a mixed-variant concat — it has no beneficiary. (The two context.rs sites each add one encode too, but their consumer, arrange_collection's columnar arm, reads rows borrowed, so there is no second decode there.)

The flag defaults off in production, but misc/python/materialize/mzcompose/__init__.py:129 sets enable_compute_temporal_bucketing = true for every mzcompose environment, so this is the configuration under test.

Suggested fix: leave the TopK site producing CollectionEdge::Vec until build_topk / map_topk_key can consume columnar containers without materializing a Row — the TODO on topk_result_to_columnar already names that work — or, if the uniform-producer property is worth the cost here, say so at the site so the round-trip is not read as an oversight.

[antiguru]

The cost is real and the trace is right. This PR adds the vec_to_columnar at top_k.rs:128, and the second decode it runs into, Columnar::into_owned in map_topk_key, comes from #37749. A bucketed TopK does go from one decode to decode, encode, decode, and nothing between the encode and that decode reads the columnar form.

Not taking the first suggestion, though, because leaving the site on CollectionEdge::Vec does not survive the stack. #37797 collapses the edge to a single columnar representation, so the site has to produce columnar there whatever it does here, and the Vec arm of map_topk_key goes away with it. Keeping this one site on Vec for the three intervening PRs would only move the round trip to the PR that deletes the alternative, and buying it back at that point costs a second input representation threaded through build_topk, which is exactly what the collapse removes.

Took the second suggestion instead: doc: record the bucketed TopK container round trip says at the site that the encode buys edge uniformity rather than a local consumer, and points at the push-down.

Worth noting the round trip is already fixed downstack rather than merely tracked. #38371 makes the bucketer columnar-native, so the decode and the encode around it both disappear and the comment there reads "Temporal bucketing is columnar throughout, so the bucketed input edge needs no round-trip". What outlives it is map_topk_key's own decode, which its doc comment already owns and which the TODO on topk_result_to_columnar names as the remaining work.

One correction to the details: enable_compute_temporal_bucketing alone does not reach this code. The site also requires ArrangementStrategy::TemporalBucketing on the TopK, so the mzcompose default exposes it only for plans the optimizer actually buckets, not for every mzcompose TopK.

_{Posted by Claude Code}

[def-]

QA LLM Review

1. MEDIUM -- rt_reduce names the wrong site, leaving both context.rs re-encodes unexercised

test/sqllogictest/temporal_bucketing.slt:312

The rt_reduce fixture is labelled "Hits the arrangement re-encode in context.rs", but a Temporal Filter -> GROUP BY plan buckets at reduce.rs:179, on the keyed ((key, val), T, Diff) stream, which is not a CollectionEdge and is untouched by this PR. Neither context.rs:1156 nor context.rs:1196 is reached by any of the three new runtime fixtures, so the block reads as covering the arrangement seam while asserting nothing about it.

Details

The file's own mv_firing fixture pins the plan for exactly this query shape at line 36: Reduce::Accumulable temporal_bucketing_strategy=TemporalBucketing directly over Get::Collection, no ArrangeBy. render_reduce buckets that keyed stream itself and bypasses ensure_collections, which the file's header comment at line 16 already states.

The other two fixtures are labelled correctly: rt_topk reaches top_k.rs:128, and rt_union reaches render.rs:1416 (its ArrangeBy prints no strategy=, i.e. Direct, because lowering.rs:1228 clears has_future_updates at a consolidating Union).

This matters because context.rs:1196 is the one flipped site whose new variant propagates onward: arrange_collection hands back a Columnar passthrough that becomes the bundle's self.collection at context.rs:1206 for every downstream consumer, where before the flip it was Vec. It is reachable from ordinary SQL, since MirRelationExpr::ArrangeBy over a temporal-filtered input gets strategy_from_future(true) at lowering.rs:1331 — that is what an index on a temporal-filter view builds, and the file already creates v_recent and v_topk_input that way. A dropped record or a mixed-up passthrough variant on that seam leaves the whole new block green.

Fix: repoint rt_reduce's comment at reduce.rs, and add a fixture that actually routes through ensure_collections under the enabled flag, e.g. a CREATE DEFAULT INDEX on a temporal-filtered view over rt_events plus a SELECT from it.

[antiguru]

Both halves are right, and I checked them against the code rather than the labels.

rt_reduce does not reach context.rs. render_reduce buckets the keyed ((key, val), T, Diff) stream itself, on key_val_collection.inner, which is not a CollectionEdge and is untouched by this PR. The file's own header comment already states that Reduce builds its arrangement via KeyValPlan and bypasses ensure_collections, so the label contradicts the file three hundred lines above it.

The coverage gap is the more important half, and the arranged-loop site is the one worth a fixture: arrange_collection hands back a Columnar passthrough that becomes the bundle's self.collection for every downstream consumer, where it was Vec before the flip. A dropped record or a swapped passthrough variant there leaves the whole new rt_* block green. Your suggested shape should reach it, since lowering gives ArrangeBy strategy_from_future(input_has_future_updates). The other site needs the raw collection to be formed from an arrangement, so it wants a temporal filter over an indexed input rather than the same fixture.

Taking both as a follow-up rather than amending here. This is the second layer of a ten-PR stack, so a change at this level restacks nine PRs and re-runs their builds, and neither item is a correctness defect in shipped code: one is a wrong comment in a test file, the other is absent coverage rather than failing coverage. The follow-up repoints the comment at reduce.rs and adds fixtures that route through ensure_collections, and I would rather confirm those plan shapes there than land them under a restack.

_{Posted by Claude Code}

[antiguru]

Follow-up is up as #38807, stacked on the remaining layers.

It repoints the rt_reduce comment at render_reduce and adds an index on a temporal-filtered view, which reaches the loop that builds the requested arrangements. The plan is pinned alongside the results, because the site is only reached while lowering marks the ArrangeBy with strategy=TemporalBucketing, and without that assertion the fixture could stop covering the site without failing.

The other site I could not reach. It needs the raw collection to be formed from an arrangement in the same ensure_collections call that builds one, and every shape I tried misses: a mz_now() predicate is applied at the top of an MFP chain, so it lifts above a join rather than landing in an ArrangeBy's input MFP, and an index on a temporal-filtered view splits into two dataflow objects whose first half builds no arrangement and therefore takes Direct. That site may be unreachable for bucketing, but I have not shown it, so the PR says only that it stays uncovered.

_{Posted by Claude Code}