From 70d7e288240a97c066c58f8a7abe5ed20ed42faf Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Wed, 22 Jul 2026 16:36:30 -0400 Subject: [PATCH 01/18] Generalize int-proxy tokens from u64 to Copy + Ord types The proxy tactics fixed both bridge coordinates at u64; the time-logic library beneath them (ValueHistory, bilinear_wave, discover_times) was already generic at V: Copy + Ord, so u64 was an instantiation posing as the architecture. This makes the tokens backend-chosen associated types: * `Group: Copy + Ord + 'static` on both backend traits: the granule name, commonly a u64 key hash but exactly the key for small Copy keys (collisions then impossible), wider hashes for insurance. `'static` because groups are the one token that crosses invocations (pending times are held per group). * `Token: Copy + Ord` (per input side on join: Token0/Token1): the ephemeral value token, scoped to one invocation. * `ProxyBridge`; `bilinear_wave` widened to distinct value types per side; `KeyView`, `DiscoverScratch`, `discover_times`, and `IdHistory` generalized to match. Tests are invariant under (u64, u64); no in-tree backend impls existed to migrate. Co-Authored-By: Claude Fable 5 --- differential-dataflow/src/operators/common.rs | 34 +++---- .../src/operators/int_proxy/history.rs | 4 +- .../src/operators/int_proxy/join.rs | 60 +++++++----- .../src/operators/int_proxy/mod.rs | 55 ++++++----- .../src/operators/int_proxy/reduce.rs | 95 ++++++++++--------- 5 files changed, 140 insertions(+), 108 deletions(-) diff --git a/differential-dataflow/src/operators/common.rs b/differential-dataflow/src/operators/common.rs index aea8e598b..783bd7dff 100644 --- a/differential-dataflow/src/operators/common.rs +++ b/differential-dataflow/src/operators/common.rs @@ -103,12 +103,13 @@ impl TimeHistory { /// `emit` receives every produced `(id0, id1, joined time, multiplied diff)`. Both histories /// must be pre-loaded (`load`/`load_iter`) and are fully drained. For small histories a plain /// cross product is cheaper; callers should gate on size. -pub fn bilinear_wave( - h0: &mut ValueHistory, - h1: &mut ValueHistory, - mut emit: impl FnMut(V, V, T, RO), +pub fn bilinear_wave( + h0: &mut ValueHistory, + h1: &mut ValueHistory, + mut emit: impl FnMut(V0, V1, T, RO), ) where - V: Copy + Ord, + V0: Copy + Ord, + V1: Copy + Ord, T: Ord + Clone + Lattice, R0: Semigroup + Multiply + Clone, R1: Semigroup + Clone, @@ -180,11 +181,11 @@ pub fn tile_descriptions( } /// A one-key view into an input presentation: the read-only arguments [`discover_times`] needs -/// about a single key — its slice `[i0, i1)` of the merged `(id, time, diff)` run `p_in` and +/// about a single key — its slice `[i0, i1)` of the merged `(token, time, diff)` run `p_in` and /// the carried `pending` times. -pub struct KeyView<'a, T, RIn> { - /// The presented `((key_hash, value_id), time, diff)` run the key's records live in. - pub p_in: &'a [((u64, u64), T, RIn)], +pub struct KeyView<'a, G, I, T, RIn> { + /// The presented `((group, token), time, diff)` run the key's records live in. + pub p_in: &'a [((G, I), T, RIn)], /// The key's first record. pub i0: usize, /// One past the key's last record. @@ -205,9 +206,9 @@ fn update_meet(meet: &mut Option, other: Option<&T>) { /// Reusable per-key scratch for [`discover_times`]: held once and threaded through every key, /// so replays and time buffers are cleared and refilled rather than reallocated per key. -pub struct DiscoverScratch { +pub struct DiscoverScratch { batch_replay: TimeHistory, - input_replay: ValueHistory, + input_replay: ValueHistory, output_replay: TimeHistory, synth: Vec, times_current: Vec, @@ -215,7 +216,7 @@ pub struct DiscoverScratch { meets: Vec, } -impl DiscoverScratch { +impl DiscoverScratch { /// Fresh scratch; hold one per retire and thread it through every key. pub fn new() -> Self { DiscoverScratch { @@ -230,7 +231,7 @@ impl DiscoverScratch { } } -impl Default for DiscoverScratch { +impl Default for DiscoverScratch { fn default() -> Self { Self::new() } } @@ -246,15 +247,16 @@ impl Default for DiscoverScratch /// consolidated view is unsound: compaction may advance a history record onto a novel time, /// where consolidation cancels the novel update and its interesting time is missed. #[allow(clippy::too_many_arguments)] -pub fn discover_times( - key: KeyView<'_, T, RIn>, +pub fn discover_times( + key: KeyView<'_, G, I, T, RIn>, seed_times: impl Iterator, out_times: impl Iterator, upper: &Antichain, - scratch: &mut DiscoverScratch, + scratch: &mut DiscoverScratch, moments: &mut Vec, pended: &mut Vec, ) where + I: Copy + Ord, T: Timestamp + Lattice, RIn: Semigroup + Clone, { diff --git a/differential-dataflow/src/operators/int_proxy/history.rs b/differential-dataflow/src/operators/int_proxy/history.rs index 905614ac0..d03eb3b3c 100644 --- a/differential-dataflow/src/operators/int_proxy/history.rs +++ b/differential-dataflow/src/operators/int_proxy/history.rs @@ -1,5 +1,5 @@ //! Time-ordered replay of proxy update histories, with meet-advancement. -/// A value history suitable for integer proxy values. -pub(in crate::operators) type IdHistory = crate::operators::ValueHistory; +/// A value history over proxy value tokens. +pub(in crate::operators) type IdHistory = crate::operators::ValueHistory; diff --git a/differential-dataflow/src/operators/int_proxy/join.rs b/differential-dataflow/src/operators/int_proxy/join.rs index 7a02d1b1f..4b6de2a99 100644 --- a/differential-dataflow/src/operators/int_proxy/join.rs +++ b/differential-dataflow/src/operators/int_proxy/join.rs @@ -1,6 +1,6 @@ //! The proxy join framework. //! -//! A conventional differential join against `(u64, u64)` values, which are provided by +//! A conventional differential join against `(group, token)` values, which are provided by //! and then interpreted by a backend, who is relieved of lattice-time reasoning. use timely::progress::{Antichain, Timestamp}; @@ -24,33 +24,42 @@ pub struct JoinInstance<'a, B0: BatchReader, B1: BatchReader