compute: bucket temporally on the columnar edge

[antiguru]

Makes temporal bucketing columnar throughout, removing the four Vec round-trips its call sites paid. The operator's state was already columnar, since its bucket chain wraps ColumnMergeBatcher, so the decode and re-encode existed only at its boundary.

Three per-record costs go with the round-trips. The reveal path handed the output builder owned tuples reconstituted from the chain's Column chunks, and now gives each chunk as a container. MergeBatcherWrapper::push_container staged its Vec argument into a Column one record at a time before the chunker could read it, and now takes a Column. Bucket::split round-tripped the sealed chunks through a Vec, cloning every record, and now re-chunks the containers.

On the input side, records are addressed by index through a permutation rather than moved, so neither the pass-through nor the bucket routing materializes an owned record. The partition happens before the sort, because only retained records need ordering and in steady state the pass-through share is the larger one.

TemporalBucketing::bucket now returns Self, which lets a Vec stream and a columnar stream both implement it. The reduce key-value path stays on the Vec implementation, which stages through the columnar operator so there is one bucketing implementation rather than two. It feeds an arrangement, and a consumer that re-encodes what it reads is cheaper to hand moved allocations than copied bytes.

Worth knowing for review: enable_compute_temporal_bucketing is compiled in as false but served on in production, so this is a live path rather than a dormant one. CPU-213 tracks reconciling the default. Verified with the flag defaulted on, since the compiled-in default otherwise leaves the operator untested outside temporal_bucketing.slt.

_{Posted by Claude Code}

[linear-code]

CPU-51

[antiguru]

I think the trick is to have a one-element container for upper, and compare the Ref type directly. Check if that's possible.

[antiguru]

Not taken yet, and I want to check it properly rather than guess. upper is an Antichain<T>, and less_equal wants a &T, so comparing a Ref directly needs PartialOrder on the Ref type, which columnar does not provide for a Product<Timestamp, PointStamp<u64>>. A one-element container gives us Ref on both sides, but the comparison still has to come from somewhere.

Took the cheaper half of the win in the meantime: the allocation, not the comparison. time_buf.copy_from(update.1) reuses one T for both the upper test and the bucket placement, so the loop no longer allocates a PointStamp per record. What is left is the copy itself, a few words.

Will look at whether a Ref-level less_equal is expressible, most likely as a small trait on our side implemented for the timestamp types the bucketer actually sees, and report back rather than holding this PR for it.

_{Posted by Claude Code}

[antiguru]

Try to use Columnar::copy_from instead of creating a new owned object.

[antiguru]

Done. One time_buf now serves both reads of a record's time, the upper test and the bucket placement, so the pass allocates nothing per record.

_{Posted by Claude Code}

[DAlperin]

don't need to comment this anymore

[antiguru]

Dropped, along with the two sibling comments that said the same thing at the call sites in render.rs and top_k.rs. Those now say only when bucketing fires. The implementation's doc paragraph went too: the permutation it described is already noted where the permutation is built, which is where it belongs.

Pushed as doc: drop the comments the columnar bucketing made vacuous.

Posted by Claude Code.

[DAlperin]

??

[antiguru]

Test-only. The bucketing tests capture their output for comparison, and capture() wants a stream of owned records, so columnar_to_vec is how the assertion reads a columnar edge back. No decode is added to a rendered dataflow by this PR.

There is a note on one of those tests saying why the decode is not allowed to hide a bug: agreeing contents alone would not rule out a silent decode on the ok path, so the tests assert on the edge's container type as well.

Posted by Claude Code.

[DAlperin]

dont need to say that

[antiguru]

Gone. Both this one and the matching line in render.rs now say only when bucketing fires, not what its containers are.

Posted by Claude Code.

[DAlperin]

||

[antiguru]

Gone. Both this one and the matching line in render.rs now say only when bucketing fires, not what its containers are.

Posted by Claude Code.

[def-]

QA LLM Review

1. MEDIUM -- the reduce path's _vec bucketing now copies every record four times where it previously moved them

src/compute/src/extensions/temporal_bucket.rs:212

The StreamVec implementation no longer has an operator body: it encodes every record into a Column (BucketStage), runs the columnar operator, then decodes every record back into owned Rows (BucketUnstage). The old Vec-native operator moved pass-through records end to end. Since render_reduce is the only caller and temporal bucketing is served on in production, a pass-through record on the temporal-filter → reduce path goes from zero copies to three byte-copies plus a decode with two Row allocations.

Details

Per pass-through record, before: Exchange pact (ExchangeCore<CapacityContainerBuilder<Vec<_>>, _>, timely pact.rs:57) moves the owned tuple, extract_if moves it, give_iterator moves it into the output Vec. Intra-process exchange does not serialize, so nothing is copied.

After: BucketStage copies the record into a column and drops the owned tuple (two Row frees per record for D = (Row, Row)); the columnar ExchangeCore<ColumnBuilder<_>, _> re-encodes it per destination; session.give(update) at temporal_bucket.rs:117 copies it into the output column; BucketUnstage decodes it back with into_owned, allocating two Rows. Retained records pay two extra copies on top of what they already paid.

Cross-process exchange is a partial offset (Vec containers bincode-serialize, columns do not), but single-process replicas are common and there the extra work is unconditional.

The doc on this impl says "The staging copy is the price of a Vec caller, paid here rather than at the call site", and the PR body says a re-encoding consumer "is cheaper to hand moved allocations than copied bytes". Neither holds: the call site paid no such price before, because the Vec-native operator existed, and the allocations BucketUnstage hands the arrangement were just decoded from a column this change introduced.

Fix: keep a Vec-native operator body for this impl (two implementations, as before), or move the reduce call site onto the columnar operator and decode once at the arrangement boundary. Note that dropping BucketStage by having the reduce key/val operator emit a column directly is not free either -- it currently uses ConsolidatingContainerBuilder (reduce.rs:111), which consolidates in the container, and a ColumnBuilder does not.

2. LOW -- split's comment claims it "visits no record", but re-chunking visits and sorts every record

src/compute/src/extensions/temporal_bucket.rs:336

push_container hands each chunk to ColumnChunker::push_into, which materializes a Vec of one Ref per record, sorts it, and re-pushes every record into a fresh target column (src/timely-util/src/columnar/batcher.rs:167). The per-record fuel charge on the line below the comment is there precisely because the work is per-record. What the change removed is the owned (D, T, R) round-trip, not the visit; the comment's own first sentence ("Re-chunks the sealed chunks") already says so.

[antiguru]

Both findings are real. Fixed in compute: keep a Vec-native operator for the Vec bucketing caller.

1, the _vec path. Confirmed, including the premise: the repo default for enable_compute_temporal_bucketing is false, but LaunchDarkly has it launched and on in production and staging, so the reduce path is live and this was a real regression on it.

The Vec stream gets its own operator body back. Records near the frontier pass through as moves, and only records that enter the chain are encoded, which they were before this PR as well since the chain's batcher is columnar. The columnar implementation stays for the columnar callers.

You also caught something I should have caught myself. The doc justified the staging with "a Vec is the cheaper intermediate for consumers that re-encode", which is a rule I retracted on #38368 after reading the consumers it was written about, and I then carried it into this file anyway. The doc now says only that a caller whose consumer wants owned records keeps a Vec-native operator, and that this implementation goes away once reduce's consumer reads columns.

Not taking the second option you offer, moving the reduce call site onto the columnar operator. Its bucketed output feeds render_reduce_plan, the whole reduce tree rather than a single arrange, so that is a layer of its own rather than part of this PR. One correction to the note attached to it: ColumnBuilder indeed does not consolidate, but ConsolidatingColumnBuilder does, and it is what the rest of the stack uses, so in-container consolidation is not the obstacle there.

2, the split comment. Right, and the per-record fuel charge on the next line contradicted it in place. It now says the chunker sorts and re-pushes every record, and that what the columnar chain removed is the owned-tuple round trip rather than the visit.

temporal_bucketing.slt 24/24, explain/default.slt 139/139, explain/physical_plan_as_json.slt 46/46, clippy clean.

Posted by Claude Code.