From 885d94c1a398b25cdc51c250faf6bbfad1e5a11b Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 3 Jul 2026 04:01:50 -0400 Subject: [PATCH 01/68] Integer-proxy chunks: backend-agnostic join and reduce over (key_hash, value_id) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A boundary where only integers cross: a storage backend presents each record as ((key_hash, value_id), time, diff) — integer proxies for data it keeps in its own layout — the operators own all the lattice/time logic over those integers, and the backend supplies value semantics via callbacks. Any columnar (or otherwise opaque-to-DD) value store can then reuse join and reduce without materializing values. - trace/chunk/int_proxy: ProxyChunk, a cursor-less Chunk of proxy columns, with from_unsorted (integer sort+consolidate with representative provenance) as the presentation-building helper. - operators/int_proxy: ProxyJoinTactic / ProxyReduceTactic for the join_with_tactic and reduce_with_tactic seams (made pub here), and the backend traits: present-as-proxies (read), value callback with hash-minted output ids (write), materialize (egress). Reduce output ids are content hashes, so an output arrangement re-presents with the same ids downstream with no registry; pending interesting times are keyed by the stable key_hash across retires. The module doc carries the boundary contract and design notes (why value_id is not order-preserving; collision risk); each tactic and the in-memory reference backend (VecChunk arrangements, fnv hashes) sit in their own file under the module. - Tests: join and count/distinct/min reduces against the row operators over multi-round retracting inputs, and a scripted Product-time retire sequence exercising synthetic corrections and pending. Co-Authored-By: Claude Fable 5 --- .../src/operators/int_proxy/join.rs | 136 +++++ .../src/operators/int_proxy/mod.rs | 96 ++++ .../src/operators/int_proxy/reduce.rs | 306 +++++++++++ .../src/operators/int_proxy/reference.rs | 258 +++++++++ differential-dataflow/src/operators/join.rs | 6 +- differential-dataflow/src/operators/mod.rs | 1 + differential-dataflow/src/operators/reduce.rs | 4 +- .../src/trace/chunk/int_proxy.rs | 505 ++++++++++++++++++ differential-dataflow/src/trace/chunk/mod.rs | 1 + differential-dataflow/tests/int_proxy.rs | 341 ++++++++++++ 10 files changed, 1649 insertions(+), 5 deletions(-) create mode 100644 differential-dataflow/src/operators/int_proxy/join.rs create mode 100644 differential-dataflow/src/operators/int_proxy/mod.rs create mode 100644 differential-dataflow/src/operators/int_proxy/reduce.rs create mode 100644 differential-dataflow/src/operators/int_proxy/reference.rs create mode 100644 differential-dataflow/src/trace/chunk/int_proxy.rs create mode 100644 differential-dataflow/tests/int_proxy.rs diff --git a/differential-dataflow/src/operators/int_proxy/join.rs b/differential-dataflow/src/operators/int_proxy/join.rs new file mode 100644 index 000000000..386eadc1a --- /dev/null +++ b/differential-dataflow/src/operators/int_proxy/join.rs @@ -0,0 +1,136 @@ +//! The proxy join: equi-match by `key_hash`, value work by matched index lists. + +use std::collections::VecDeque; + +use timely::ContainerBuilder; +use timely::dataflow::operators::Capability; +use timely::dataflow::operators::generic::OutputBuilderSession; + +use crate::difference::{Multiply, Semigroup}; +use crate::lattice::Lattice; +use crate::trace::BatchReader; +use crate::trace::chunk::int_proxy::ProxyChunk; +use crate::operators::join::{EffortBuilder, Fresh, JoinTactic}; + +/// The join backend: value semantics for a proxy-space equijoin. +/// +/// Protocol (per deferred work unit, driven by [`ProxyJoinTactic`]): one `present0`, one +/// `present1`, then at most one `cross` whose indices refer to those two presentations. +/// The backend must keep each presentation's alignment (run index → real record) until +/// the `cross` call. +pub trait ProxyJoinBackend> { + /// Diff type presented for the first input. + type R0: Semigroup + Multiply; + /// Diff type presented for the second input. + type R1: Semigroup; + /// Diff type of matched records (`R0 * R1`), computed by the tactic. + type ROut: Semigroup; + /// The output container built from matched index lists. + type Output; + + /// (read) Flatten `batches` into one sorted, consolidated proxy run. + fn present0(&mut self, batches: &[B0]) -> ProxyChunk; + /// (read) As `present0`, for the second input. + fn present1(&mut self, batches: &[B1]) -> ProxyChunk; + /// (value work) The projection. `left[i]`/`right[i]` index the current presentations; + /// each pair matched on `key_hash`. `times[i]`/`diffs[i]` are the DD-computed lattice + /// join of the pair's times and product of its diffs, to carry through per record. + fn cross(&mut self, left: &[usize], right: &[usize], times: Vec, diffs: Vec) -> Self::Output; +} + +/// A proxy-space [`JoinTactic`]: matches records of the two presented runs by `key_hash`, +/// cross-products matched records with DD-computed times (lattice join) and diffs +/// (product), and hands the backend the matched index lists for the value work. +/// +/// Work units are queued by arrival direction (as in the cursor tactic) and currently +/// drained eagerly; fuel budgeting is a later refinement. +pub struct ProxyJoinTactic, Bk> { + backend: Bk, + todo0: VecDeque>, + todo1: VecDeque>, +} + +/// A deferred bilinear join unit: a fresh batch list against the accumulated other side. +struct JoinUnit> { + left: Vec, + right: Vec, + capability: Capability, +} + +impl, Bk> ProxyJoinTactic { + /// A tactic deferring all value semantics to `backend`. + pub fn new(backend: Bk) -> Self { + ProxyJoinTactic { backend, todo0: VecDeque::new(), todo1: VecDeque::new() } + } +} + +impl JoinTactic for ProxyJoinTactic +where + B0: BatchReader, + B1: BatchReader