compute: Emit linear and delta join outputs as the columnar edge - #37784
compute: Emit linear and delta join outputs as the columnar edge#37784antiguru wants to merge 1 commit into
Conversation
a4d4319 to
90ef7ab
Compare
90ef7ab to
4ab77c6
Compare
4ab77c6 to
4767931
Compare
6e4854a to
495b483
Compare
| // The delta join is Vec-internal (half_join, the `ok_err` demux, the | ||
| // time-unpair map, and the per-path finalization all operate on `Vec`). | ||
| // Encode the concatenated node output to the columnar edge once here. | ||
| // This is the sanctioned leaf-encode; a columnar `half_join`/algorithm is | ||
| // a differential-side follow-up. Non-consolidating: the per-path | ||
| // finalization already consolidated whatever it consolidates. |
There was a problem hiding this comment.
Note there is a half_join_internal_unsafe API we could use here. It takes a container builder, so we just need to replicate what the outer function does. This would allow us to avoid the vec_to_columnar escape hatch.
There was a problem hiding this comment.
Both build_halfjoin1 and build_halfjoin2 already call half_join_internal_unsafe, with CapacityContainerBuilder<Vec<_>> as the container builder, so swapping in a columnar builder needs no differential-side work. Two constraints keep it from being a local change, though:
- Stages chain.
half_join's input is aCollection<(Row, Row, T)>, so only a path's last stage can emit columnar, which means the stage loop has to know which stage is last. - The output does not go straight to the node boundary. It passes through the
ok_errdemux, the time-unpairmap, and the optional per-path finalization, so all three have to accept columnar before the leaf encode disappears.
The could_error arm is workable after all: the closure can map its error to DataflowErrorSer (which derives Columnar) before pushing, and columnar 0.13 has impl<S: Columnar, T: Columnar> Columnar for Result<S, T>, so the element type stays Columnar and ok_err becomes a two-output unary over two ColumnBuilders.
The real win is the time-unpair step, which turns from a per-record map into a sub-column reshape, zero-copy for a Typed input, the same shape as the reshape just deleted from columnar_consolidate in #37756.
Leaving this PR as it stands and tracking the delta-join conversion as follow-up work, so the stack keeps its landing path.
Posted by Claude Code
| CollectionEdge::Columnar(updates) | ||
| } else { | ||
| // Identity finalization: the raw stage output is the result. | ||
| // `mz_join_core` produces a `Vec` collection (intra-operator, |
There was a problem hiding this comment.
Rewiring mz_join_core to use container builders is an option, and should do as part of this PR stack.
There was a problem hiding this comment.
mz_join_core has the same demux issue. Its could_error() arm in differential_join_inner builds Result<Row, DataflowErrorSer> elements and splits them with .inner.ok_err(...), so a columnar output container has to carry the Result and the demux has to become a two-output unary over two ColumnBuilders. That part is feasible: DataflowErrorSer derives Columnar and columnar 0.13 has impl<S: Columnar, T: Columnar> Columnar for Result<S, T>.
The linear join is otherwise in better shape than the delta join for this. mz_join_core is already generic over its output container (C: Container + SizableContainer + PushInto<(I::Item, T, Diff)>), so the rewiring is a change from a container parameter to a container-builder parameter, threaded through Work::process and LinearJoinSpec::render. And stage chaining is not a blocker here, because every stage output is re-arranged through arrange_join_input, which already has a columnar arm.
Following your suggestion and doing this as its own PR on top of the stack rather than folding it into this one.
Posted by Claude Code
There was a problem hiding this comment.
Opened as #38368, on top of columnar-te-collapse-enum.
It turned out not to need the columnar demux. mz_join_core now takes a container builder, and the last stage writes the output edge directly when no finalization closure follows it, which is exactly where vec_to_columnar sat.
One finding worth recording, because it cuts against doing this globally: a Column is the wrong intermediate for any consumer that re-encodes what it reads. A following stage's arrangement and a finalization closure both do that, and a Vec hands them moved Row allocations where a Column would copy row bytes first. So making every stage columnar would add an encode per stage. The builder choice is per stage instead, which is why render takes it as a parameter.
The error-capable arm keeps the accumulator even when terminal: its output has to pass ok_err first, and the demux materializes the ok side as a Vec regardless, so a columnar output there would cost two encodes rather than one.
Verified on both join implementations, including enable_mz_join_core = false so the new encode_updates path is covered.
Posted by Claude Code
495b483 to
cea3e9d
Compare
cea3e9d to
3b89a2b
Compare
3b89a2b to
e28ea6e
Compare
e28ea6e to
64bc5f3
Compare
64bc5f3 to
38c1f3a
Compare
Both joins are Vec-internal operators (the join algorithms `mz_join_core` and `half_join`, the `ok_err` demux, and the time-unpair map all stay on `Vec`); the columnar edge is a leaf-encode at each node's output boundary. A columnar join algorithm is a differential-side follow-up, out of scope here. Linear join: the finalization produces the output edge. With a `final_closure`, the closure computes fresh rows built into a `ConsolidatingColumnBuilder` (owned give), matching the prior `ConsolidatingContainerBuilder`. Without one, the raw stage output is encoded via `vec_to_columnar`; a columnar single-input source passes through with no round-trip. Delta join: the node output is Vec-native throughout, so encode the concatenated result once via `vec_to_columnar`. Non-consolidating, matching the old raw concat (the per-path finalization already consolidated what it consolidates). `from_collections` -> `from_edge`; err stays `Vec`. Also corrects the stale `arrange_join_input` comment: its columnar arm runs whenever the input edge is columnar, which the source-key path and upstream producers now emit. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
38c1f3a to
a4df7f3
Compare
Linear and delta join outputs emit the columnar edge. The join algorithms stay Vec-internal (
mz_join_core,half_join); the columnar edge is produced by a leaf-encode at the node output boundary. Columnarizing the join cores is a differential-side fast-follow.Columnar dataflow-edge migration. Design doc:
doc/developer/design/20260720_columnar_dataflow_edges.md(#37744).Part of CPU-51.