Skip to content

compute: Emit linear and delta join outputs as the columnar edge - #37784

Open
antiguru wants to merge 1 commit into
columnar-p6-topk-outputfrom
columnar-p7-join-output
Open

compute: Emit linear and delta join outputs as the columnar edge#37784
antiguru wants to merge 1 commit into
columnar-p6-topk-outputfrom
columnar-p7-join-output

Conversation

@antiguru

@antiguru antiguru commented Jul 21, 2026

Copy link
Copy Markdown
Member

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.

@linear-code

linear-code Bot commented Jul 22, 2026

Copy link
Copy Markdown

CPU-51

@antiguru
antiguru force-pushed the columnar-p7-join-output branch from 90ef7ab to 4ab77c6 Compare July 22, 2026 16:25
@antiguru
antiguru force-pushed the columnar-p7-join-output branch from 4ab77c6 to 4767931 Compare July 22, 2026 17:51
@antiguru
antiguru force-pushed the columnar-p7-join-output branch 2 times, most recently from 6e4854a to 495b483 Compare August 19, 2026 13:17
Comment on lines +249 to +254
// 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.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 a Collection<(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_err demux, the time-unpair map, 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,

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Rewiring mz_join_core to use container builders is an option, and should do as part of this PR stack.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@antiguru
antiguru force-pushed the columnar-p7-join-output branch from 495b483 to cea3e9d Compare August 20, 2026 08:49
@antiguru
antiguru force-pushed the columnar-p7-join-output branch from cea3e9d to 3b89a2b Compare August 20, 2026 09:13
@antiguru
antiguru force-pushed the columnar-p7-join-output branch from 3b89a2b to e28ea6e Compare September 6, 2026 18:23
@antiguru
antiguru force-pushed the columnar-p7-join-output branch from e28ea6e to 64bc5f3 Compare September 10, 2026 11:33
@antiguru
antiguru marked this pull request as ready for review September 10, 2026 11:33
@antiguru
antiguru requested a review from a team as a code owner September 10, 2026 11:33
@antiguru
antiguru force-pushed the columnar-p7-join-output branch from 64bc5f3 to 38c1f3a Compare September 10, 2026 12:04
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>
@antiguru
antiguru force-pushed the columnar-p7-join-output branch from 38c1f3a to a4df7f3 Compare September 10, 2026 12:09
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant