Skip to content

compute: concatenate row bytes for identity join closures - #38670

Open
frankmcsherry wants to merge 2 commits into
MaterializeInc:mainfrom
frankmcsherry:join-identity-concat
Open

compute: concatenate row bytes for identity join closures#38670
frankmcsherry wants to merge 2 commits into
MaterializeInc:mainfrom
frankmcsherry:join-identity-concat

Conversation

@frankmcsherry

@frankmcsherry frankmcsherry commented Sep 4, 2026

Copy link
Copy Markdown
Contributor

When a linear join stage's closure is the identity, its output row is the key followed by the stream value and the lookup value, in that order. The renderer still built each output row by decoding all three into datums and packing them again. This change concatenates the three row encodings directly when all three are available as row-encoded bytes, and falls back to the datum path otherwise (a dictionary-coded arrangement does not expose contiguous row bytes).

ExtendDatums gains as_row_ref, defaulting to None, implemented for Row, references, and the row spine's DatumSeq when no codec is present. The join uses it in a new is_identity branch ahead of the could_error split; an identity closure cannot error, so that branch returns no error collection.

Hydration of SELECT t.k, t.v, d.w FROM fact t JOIN dim d ON t.k = d.k indexed by k, 10M fact rows, 1M dim rows, one worker, three runs each, median:

before after
hydration wall 6.19s 5.78s
Join operator 1.50s 1.05s

The other operators in the dataflow are unchanged. Measured on a local build with the other hydration changes in flight (#38660, #38663, #38665, #38668, #38669) applied, which is why the baseline is below main; the join operator's share is independent of those.

Existing join coverage in sqllogictest and testdrive exercises both branches; no new tests, since the output is byte-identical by construction.

Second commit: no output consolidation for identity closures

mz_join_core consolidates each chunk of produced results before emitting it, which pays off when the join closure can map distinct input pairs to equal results, as a projection that drops columns does. An identity closure cannot: distinct (key, left value, right value) triples concatenate to distinct rows, so the sort behind that consolidation, about ten row comparisons per output row, recovers nothing.

The second commit threads a consolidate_output flag from the linear join renderer into the join core. The identity-closure path passes false; every other path keeps consolidating. The differential join implementation never consolidated and is unchanged.

Hydration of SELECT t.k, t.v, d.w FROM fact t JOIN dim d ON t.k = d.k indexed by k, 10M fact rows and 1M dimension rows, one worker, median of three, on a local build with the other hydration PRs in flight applied:

before after
hydration wall 5.14s 4.59s
Join operator 1007ms 440ms

joins.slt and join-identity-elision.slt pass; the output is the same multiset either way, since the downstream arrangement consolidates.

When a linear join stage's closure is the identity, the output row is the
key, the stream value and the lookup value in that order. Build it by
concatenating the three row encodings instead of decoding them into datums
and packing them again. ExtendDatums gains as_row_ref so the join can ask
each side for its row bytes, falling back to the datum path when a side is
not stored as one row encoding.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
@frankmcsherry
frankmcsherry requested a review from a team as a code owner September 4, 2026 20:43
mz_join_core consolidates each chunk of results before emitting it, which
pays off when the closure can map distinct input pairs to equal results. An
identity closure cannot, so let the renderer tell the join core whether to
consolidate, and pass false on the identity path.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
@def-

def- commented Sep 5, 2026

Copy link
Copy Markdown
Contributor

QA LLM Review

1. MEDIUM -- Skipping join output consolidation lets the join core's own cancelling matches escape

src/compute/src/render/join/linear_join.rs:509

The identity path passes consolidate_output: false on the premise that distinct input pairs give distinct outputs, but both of mz_join_core's strategies deliberately emit matches at times where one of the values was not present, and those only cancel through the consolidation that this commit removes. With a churning value on one side and a high fan-out on the other, the operator now ships O(fan-out × edits) records that previously cancelled to nothing inside it, moving the sort and the peak memory into the downstream arrangement batcher.

Details

join_key_simple (src/compute/src/render/join/mz_join_core.rs:763) emits one record per (v1,t1,r1) × (v2,t2,r2) combination at t1.join(t2). The Joiner doc comment at src/compute/src/render/join/mz_join_core.rs:681 says so explicitly: it "may produce matches for times in which one of the values wasn't present. These matches cancel each other out, so the result ends up correct." Distinct input pairs do give distinct rows, but the output times collapse, so the (row, time) pairs are not distinct.

Concrete case: for key k the stream side holds value v inserted at t=1 and retracted at t=2 inside one batch (2 edits, so the < 10 test picks the simple strategy no matter how large the other side is), and the lookup side holds 1M rows for k at t=3. The operator produces 2M records: (v,w_j) at t=3 with +1 and (v,w_j) at t=3 with -1, for every j. All of them are produced inside a single join_key_simple call with no yield point, so before this change consolidate_updates in Work::process always saw both halves and cancelled all 2M in the operator's own buffer. Now all 2M flow into the downstream mz_arrange, which has to buffer, sort and cancel them there. The linear-scan strategy has the same shape: work_history1 emits at t1.join(t2), and successive t1 values below t2 land on the same output time.

Results stay correct, since downstream arrangements consolidate; the cost is throughput and peak memory on exactly the wide fan-out joins this change targets.

A dynamic guard would keep the measured win without giving up the cancellation: during hydration EditList::load advances every time to meet and consolidates per value, so each value carries a single edit at a single time and the sort provably recovers nothing. Deciding per drain (consolidate only when the inputs for a key span more than one distinct time) covers the benchmarked case and still cancels when the inputs actually churn, whereas a static per-closure flag cannot tell the two apart.

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.

2 participants