Skip to content

Half joins read imported arrangements as of a frontier - #1025

Merged
frankmcsherry merged 3 commits into
TimelyDataflow:master-nextfrom
frankmcsherry:backend-frontier-half-join
Oct 24, 2026
Merged

frankmcsherry merged 3 commits into
TimelyDataflow:master-nextfrom
frankmcsherry:backend-frontier-half-join

Conversation

@frankmcsherry

Copy link
Copy Markdown
Member

The follow-up to #1023 and #1024 for the half join, now that it is a proxy tactic (#1022) and DDIR's half join is a backend under it (#1021).

differential: what an outside wrapper needed

Two additions to int_proxy::frontier, the only things dogs3 needed in order to implement its own backend trait for BackendFrontier:

  • BackendFrontier::inner_mut(), so the wrapper can drive the inner backend;
  • read_as_of, now public and re-exported from int_proxy, so an outside wrapper reads a bridge as of since the same way the join and reduce wrappers do. It advances only times not already beyond since, and consolidates only if any moved.

dogs3: BackendFrontier as a proxy half-join backend

operators/half_join/frontier.rs implements ProxyHalfJoinBackend for BackendFrontier<Bk, T>:

  • stage and cross pass through, so cross sees the advanced times;
  • advance calls the inner backend, then reads the arranged side as of since;
  • the stream side is untouched;
  • no key intersection is needed, because the tactic skips keys that only one side holds.

Unlike the join and reduce wrappers, it has no unit-level shortcut. The half join compares times in the total order, and advancing an arranged time can change that comparison even against stream times beyond since: stored (0, 5) precedes (1, 0), but advanced to {(1, 0)} it is (1, 5) and does not. The check is per row instead. It is one comparison, and at a program time it moves nothing.

Tests in tests/proxy_half_join.rs:

  • that flip;
  • a random oracle over Product times, with arranged rows on both sides of since;
  • every one- or two-element frontier of the pentagon lattice N5, which is not distributive.

DDIR

CorgiBackend::half_join wraps its backend with the arrangement's since, as join and reduce do. Every half join is wrapped, so each call site compiles to one operator.

In a delta join, an import's rows arrive twice: once arranged, and once lifted as a stream read as of since. Each pair meets exactly once only if both paths compare the same times. So imports feeding half joins no longer rely on published traces carrying root times, and the comment arguing they could be read as stored is replaced.

server_late_import.rs gains a delta join of an import with itself, installed early and late, checked against the join's result. It passes without the wrapper too, and has to: published traces carry root times today, where stored and as-of readings agree. The dogs3 tests are the ones that tell the wrapper from no wrapper.

Checks

  • All test binaries of differential, dogs3, interactive and ddir-server pass.
  • cargo check --all-targets: no warnings. Clippy: nothing new in the files touched.
  • Release difftest, 1,000 cases with all nine oracles: all passed. Late-install sweep, 2,000 seeds of split,bind,reinstall: all passed.
  • Cost at since = {minimum}, measured with a half-join probe kept out of the tree: wrapped and unwrapped were within run-to-run spread, in both directions, for integer and hashed keys.

🤖 Generated with Claude Code

https://claude.ai/code/session_019XsakwFGwtESwZ1gSYwG4o

frankmcsherry and others added 3 commits October 24, 2026 01:35
A wrapper implementing a backend trait defined outside this crate, such as
dogs3's half-join backend, needs to drive the inner backend and to read a
bridge as of `since` the one way. The two items an outside implementor
needed from the frontier wrapper; nothing else changes.

Co-Authored-By: Claude Fable 5.1 <[email protected]>
Claude-Session: https://claude.ai/code/session_019XsakwFGwtESwZ1gSYwG4o
Reads the arranged side as of `since`: after the inner backend's `advance`,
every arranged time not already beyond `since` is advanced by it and the
bridge consolidated if any moved; `stage` and `cross` pass through, and
`cross` sees the advanced times. The stream side is untouched, and no key
intersection is needed since the tactic skips keys on one side alone.

Every unit reads as of `since`, unlike the join and reduce wrappers: the
half join compares in the total order, and advancing changes that
comparison even against stream times beyond `since` (stored (0, 5)
precedes (1, 0); advanced to {(1, 0)} it is (1, 5) and does not). The
shortcut is per row, one comparison, which moves nothing at a program
time.

Tests: the flip above, a random oracle over product times with arranged
rows on both sides of `since`, and every one- or two-element frontier of
the pentagon N5.

Co-Authored-By: Claude Fable 5.1 <[email protected]>
Claude-Session: https://claude.ai/code/session_019XsakwFGwtESwZ1gSYwG4o
`CorgiBackend::half_join` wraps its backend in `BackendFrontier` with the
arrangement's `since`, as join and reduce do. The import's rows reach a
delta join once arranged and once lifted as a stream read as of `since`,
and each pair meets exactly once only if both paths compare the same
times. Imports feeding half joins no longer rely on published traces
carrying root times; the comment arguing they could be read as stored is
replaced. Every half join is wrapped, so each call site compiles to one
operator; at the minimum the wrapper costs one comparison per arranged
row, which the half-join probe could not tell from no wrapper.

`server_late_import.rs` gains a delta join of an import with itself,
installed early and late, against the join's result.

Co-Authored-By: Claude Fable 5.1 <[email protected]>
Claude-Session: https://claude.ai/code/session_019XsakwFGwtESwZ1gSYwG4o
@frankmcsherry
frankmcsherry merged commit 4a20f0c into TimelyDataflow:master-next Oct 24, 2026
7 checks passed
@frankmcsherry
frankmcsherry deleted the backend-frontier-half-join branch October 24, 2026 05:44
frankmcsherry added a commit to frankmcsherry/differential-dataflow that referenced this pull request Oct 24, 2026
…ejoin the results table

Since TimelyDataflow#1024 and TimelyDataflow#1025 a one-off reads every import as of the tick it starts, so the
README no longer calls the one-off runs experimental. The 0.0025 run, rerun on
master-next 7eb7da8, passes again (1,049 ops/s); 0.002 still fails. The one-off row
was split from the results table by a blank line.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01GVgz9LXpYfNWPmu352Yt1u
frankmcsherry added a commit that referenced this pull request Oct 24, 2026
#1010)

* LDBC: an SNB Interactive v2 driver for DDIR, with DuckDB as a baseline

interactive_v2/driver.py reimplements the official Java driver's workload from its
source: per-operation updates on the simulation clock, complex reads on their
frequency grid with the paramgen's parameters, the short-read walk, TCR, and the
per-type 95% on-time audit. systems.py runs it against DDIR (group-commit ticks)
and DuckDB (concurrent connections). snb/operations.py turns the BI CSV inserts and
deletes into single operations with indexed cascades; factors.py rebuilds the
paramgen's inputs so the official paramgen can run. README.md lists what is
reproduced, where it departs, and first SF1 results.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01GVgz9LXpYfNWPmu352Yt1u

* LDBC Interactive v2: per-tick log and server/process CPU time for attribution

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01GVgz9LXpYfNWPmu352Yt1u

* LDBC: IC9 ranks per friend before per request; IC14 iterates costs, then recovers the path; driver --batch-ms

IC9's final top 20 is keyed by request, so every candidate went to one worker; each
slow-path friend now keeps their own 20 first (4 workers: 223 -> 116 ms answer plus
retract). IC14 runs Bellman-Ford on costs alone, keeps every least-cost predecessor,
and walks the smallest next node from the source over the nodes on a cheapest path
(121 -> 74 ms). Answers equal the previous plans on 62 SF1 bindings and the fixture.
The driver can hold DDIR ticks so they are fuller; the client passes DDIR_DIAGNOSTICS.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01GVgz9LXpYfNWPmu352Yt1u

* LDBC Interactive v2: replay and per-read cost tools; README results and attribution

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01GVgz9LXpYfNWPmu352Yt1u

* LDBC Interactive v2: one-off complex reads (--oneoff), from split plans

oneoff/split.py splits IC1, IC9, IC10, IC12, IC13 and IC14 into a standing program
(their binding-independent statements, published) and a one-off program that each
read installs, answers once and drops at the next tick instead of retracting.
Experimental: a one-off installed late reads published traces without #982's
consistent-frontier import.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01GVgz9LXpYfNWPmu352Yt1u

* LDBC Interactive v2: one-off splits for all 14 complex reads; replay --oneoff and --only

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01GVgz9LXpYfNWPmu352Yt1u

* LDBC Interactive v2: delete cascades in the graph program; the driver streams its inputs

graph.py's program takes inserted rows and explicit delete targets and derives the
live relations by snb.updates.State's rules, ahead of the unchanged graph program.
Over the SF1 stream they equal the BI daily journal (three checkpoints); 20,000
operations answer identically either way; test_cascade.py checks the rules in CI.
stream.py reads the snapshot in chunks and the operations a day at a time, and
complex reads are generated lazily: the driver peaks at 181 MB at SF1 (was 8 GB);
the server at 7.5 GB (was 4.7). --cascade client keeps the old path; DuckDB uses it.
Also --sf 0.1 (SF1's mix on SF0.1 data) and factors.py reading plain CSV partitions.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01GVgz9LXpYfNWPmu352Yt1u

* LDBC Interactive v2: sort each relation once when feeding the client-side snapshot

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01GVgz9LXpYfNWPmu352Yt1u

* LDBC Interactive v2 README: one-offs read imports as of their tick; rejoin the results table

Since #1024 and #1025 a one-off reads every import as of the tick it starts, so the
README no longer calls the one-off runs experimental. The 0.0025 run, rerun on
master-next 7eb7da8, passes again (1,049 ops/s); 0.002 still fails. The one-off row
was split from the results table by a blank line.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01GVgz9LXpYfNWPmu352Yt1u

---------

Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]>
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