differential_dataflow/operators/int_proxy/
join.rs

1//! The proxy join framework.
2//!
3//! A conventional differential join against `(u64, u64)` values, which are provided by
4//! and then interpreted by a backend, who is relieved of lattice-time reasoning.
5
6use timely::progress::{Antichain, Timestamp};
7use timely::progress::frontier::AntichainRef;
8
9use crate::difference::{Multiply, Semigroup};
10use crate::lattice::Lattice;
11use crate::trace::BatchReader;
12use super::ProxyBridge;
13use crate::operators::join::{Fresh, JoinTactic};
14
15use super::history::IdHistory;
16
17/// A unit of proxied join work, presented to the backend.
18pub struct JoinInstance<'a, B0: BatchReader, B1: BatchReader<Time = B0::Time>> {
19    /// The first input's batches.
20    pub batches0: &'a [B0],
21    /// The second input's batches.
22    pub batches1: &'a [B1],
23    /// The compaction frontier for loading (the unit's capability time).
24    pub lower: AntichainRef<'a, B0::Time>,
25}
26
27/// A type that can interpret and retire pairs of batches, joined by key hashes.
28///
29/// The protocol invokes the `present*` methods with the instance to produce the proxy collection,
30/// and then returns with any number (including zero) calls to `cross` to produce outputs.
31pub trait ProxyJoinBackend<B0: BatchReader, B1: BatchReader<Time = B0::Time>> {
32    /// Diff type presented for the first input.
33    type R0: Semigroup + Multiply<Self::R1, Output = Self::ROut>;
34    /// Diff type presented for the second input.
35    type R1: Semigroup;
36    /// Diff type of matched records (`R0 * R1`), computed by the tactic.
37    type ROut: Semigroup;
38    /// The output container built from matched value ids.
39    type Output;
40
41    /// Prepare a proxy bridge from the instance's first input, optionally key-hash restricted.
42    ///
43    /// The returned bridge **must be sorted and consolidated** by `((key_hash, value_id), time)`.
44    fn present0(&mut self, instance: &JoinInstance<'_, B0, B1>, filter: Option<&[u64]>) -> ProxyBridge<B0::Time, Self::R0>;
45    /// Prepare a proxy bridge from the instance's second input, optionally key-hash restricted.
46    ///
47    /// The returned bridge **must be sorted and consolidated** by `((key_hash, value_id), time)`.
48    fn present1(&mut self, instance: &JoinInstance<'_, B0, B1>, filter: Option<&[u64]>) -> ProxyBridge<B0::Time, Self::R1>;
49    /// From a list of left and right identifiers, and corresponding times and diffs, the output.
50    fn cross(&mut self, instance: &JoinInstance<'_, B0, B1>, left: &[(u64, u64)], right: &[(u64, u64)], times: Vec<B0::Time>, diffs: Vec<Self::ROut>) -> Self::Output;
51}
52
53/// A proxy-space [`JoinTactic`]: matches records of the two presented runs by `key_hash`,
54pub struct ProxyJoinTactic<B0, B1, Bk> {
55    backend: Bk,
56    _marker: std::marker::PhantomData<(B0, B1)>,
57}
58
59impl<B0, B1, Bk> ProxyJoinTactic<B0, B1, Bk> {
60    /// A join tactic deferring all value semantics to `backend`.
61    pub fn new(backend: Bk) -> Self {
62        ProxyJoinTactic { backend, _marker: std::marker::PhantomData }
63    }
64}
65
66impl<B0, B1, Bk> JoinTactic<B0, B1, Bk::Output> for ProxyJoinTactic<B0, B1, Bk>
67where
68    B0: BatchReader,
69    B1: BatchReader<Time = B0::Time>,
70    Bk: ProxyJoinBackend<B0, B1>,
71    Bk::Output: 'static,
72{
73    fn prep(&mut self, input0: Vec<B0>, input1: Vec<B1>, fresh: Fresh, meet: B0::Time) -> Box<dyn Iterator<Item = Bk::Output>> {
74        Box::new(join_prep(&mut self.backend, input0, input1, fresh, meet).into_iter())
75    }
76}
77
78/// Update count threshold used to return to the backend with join output to instantiate.
79const JOIN_CHUNK: usize = 1 << 20;
80
81/// Prepare one join unit: present both sides, merge-match key runs over the sorted hashes, and
82/// cross-product matched records into a sequence of output containers — one per ~[`JOIN_CHUNK`]
83/// matches, flushed wherever the batch fills (including mid-key; parity: their concatenation is the
84/// single-container output). `meet` is a lower bound on the fresh side's times, so the accumulated
85/// side loads compacted (see [`JoinInstance`]); every output is produced under the capability at
86/// `meet`, so advancing loaded times by it leaves the output unchanged.
87fn join_prep<B0, B1, Bk>(backend: &mut Bk, input0: Vec<B0>, input1: Vec<B1>, fresh: Fresh, meet: B0::Time) -> Vec<Bk::Output>
88where
89    B0: BatchReader,
90    B1: BatchReader<Time = B0::Time>,
91    Bk: ProxyJoinBackend<B0, B1>,
92{
93    let lower = Antichain::from_elem(meet);
94    let instance = JoinInstance { batches0: &input0, batches1: &input1, lower: lower.borrow() };
95
96    // Prepare work using the keys of the just-received side.
97    // FIXME: Use the smaller of the two instead.
98    let (p0, p1) = match fresh {
99        Fresh::Input0 => {
100            let p0 = backend.present0(&instance, None);
101            if p0.is_empty() { return Vec::new(); }
102            let mut keys: Vec<u64> = p0.iter().map(|r| r.0.0).collect();
103            keys.dedup();
104            let p1 = backend.present1(&instance, Some(&keys));
105            (p0, p1)
106        }
107        Fresh::Input1 => {
108            let p1 = backend.present1(&instance, None);
109            if p1.is_empty() { return Vec::new(); }
110            let mut keys: Vec<u64> = p1.iter().map(|r| r.0.0).collect();
111            keys.dedup();
112            let p0 = backend.present0(&instance, Some(&keys));
113            (p0, p1)
114        }
115    };
116    if p0.is_empty() || p1.is_empty() { return Vec::new(); }
117    super::debug_assert_sorted_bridge(&p0, "present0");
118    super::debug_assert_sorted_bridge(&p1, "present1");
119
120    // Merge-join the two presented runs on `key_hash` (both sorted) and cross matched pairs. `flush`
121    // ships the accumulated batch as a container and resets the buffers; `join_key` calls it at each
122    // `JOIN_CHUNK` boundary — including *within* a high-fanout key's wave — so no single key
123    // materializes more than a chunk of its cross product. The closure scope releases its
124    // `backend`/`out` borrow before `out` is returned.
125    let mut out: Vec<Bk::Output> = Vec::new();
126    {
127        let mut flush = |li: &mut Vec<(u64, u64)>, ri: &mut Vec<(u64, u64)>, ot: &mut Vec<B0::Time>, od: &mut Vec<Bk::ROut>| {
128            out.push(backend.cross(&instance, li.as_slice(), ri.as_slice(), std::mem::take(ot), std::mem::take(od)));
129            li.clear();
130            ri.clear();
131        };
132
133        let (mut li, mut ri) = (Vec::new(), Vec::new());
134        let (mut ot, mut od) = (Vec::new(), Vec::new());
135        // Per-key replay histories, held across the unit and reloaded per key (only the >=16/>=16
136        // replay-wave path touches them), so a high-fanout join pays no per-key `IdHistory` alloc.
137        let mut h0 = IdHistory::new();
138        let mut h1 = IdHistory::new();
139        let (mut i, mut j) = (0usize, 0usize);
140        while i < p0.len() && j < p1.len() {
141            let (ki, kj) = (p0[i].0.0, p1[j].0.0);
142            if ki < kj {
143                i += 1;
144            } else if kj < ki {
145                j += 1;
146            } else {
147                let mut e0 = i;
148                while e0 < p0.len() && p0[e0].0.0 == ki { e0 += 1; }
149                let mut e1 = j;
150                while e1 < p1.len() && p1[e1].0.0 == ki { e1 += 1; }
151                join_key(ki, &p0, i..e0, &p1, j..e1, &mut h0, &mut h1, &mut li, &mut ri, &mut ot, &mut od, &mut flush);
152                i = e0;
153                j = e1;
154            }
155        }
156        if !li.is_empty() {
157            flush(&mut li, &mut ri, &mut ot, &mut od);
158        }
159    }
160    out
161}
162
163/// Match one key's records across the two presented runs.
164///
165/// If either history is small, this performs a simple cross product.
166/// If both histories are large, this replays the histories compacting as it goes in
167/// order to (potentially) avoid quadratic blow-up.
168#[allow(clippy::too_many_arguments)]
169fn join_key<T, R0, R1, RO, F>(
170    kh: u64,
171    p0: &ProxyBridge<T, R0>,
172    r0: std::ops::Range<usize>,
173    p1: &ProxyBridge<T, R1>,
174    r1: std::ops::Range<usize>,
175    h0: &mut IdHistory<T, R0>,
176    h1: &mut IdHistory<T, R1>,
177    li: &mut Vec<(u64, u64)>,
178    ri: &mut Vec<(u64, u64)>,
179    ot: &mut Vec<T>,
180    od: &mut Vec<RO>,
181    flush: &mut F,
182) where
183    T: Lattice + Timestamp,
184    R0: Semigroup + Multiply<R1, Output = RO> + Clone,
185    R1: Semigroup + Clone,
186    F: FnMut(&mut Vec<(u64, u64)>, &mut Vec<(u64, u64)>, &mut Vec<T>, &mut Vec<RO>),
187{
188    // `flush` ships the accumulated batch once it reaches `JOIN_CHUNK` and resets the buffers. It's
189    // checked after every produced pair — *including inside the replay wave below* — so even a key
190    // whose fanout dwarfs `JOIN_CHUNK` never materializes more than a chunk of its cross product.
191    if r0.len() < 16 || r1.len() < 16 {
192        for a in r0 {
193            for b in r1.clone() {
194                li.push((kh, p0[a].0.1));
195                ri.push((kh, p1[b].0.1));
196                ot.push(p0[a].1.join(&p1[b].1));
197                od.push(p0[a].2.clone().multiply(&p1[b].2));
198                if li.len() >= JOIN_CHUNK { flush(li, ri, ot, od); }
199            }
200        }
201        return;
202    }
203
204    // Reusable replay scratch, reloaded per key (`load_iter` clears + rebuilds, keeping capacity);
205    // the caller holds `h0`/`h1` across the unit so a high-fanout join allocates no per-key history.
206    h0.load_iter(r0.map(|i| (p0[i].0.1, p0[i].1.clone(), p0[i].2.clone())), None);
207    h1.load_iter(r1.map(|i| (p1[i].0.1, p1[i].1.clone(), p1[i].2.clone())), None);
208
209    crate::operators::common::bilinear_wave(h0, h1, |v0, v1, t, d| {
210        li.push((kh, v0));
211        ri.push((kh, v1));
212        ot.push(t);
213        od.push(d);
214        if li.len() >= JOIN_CHUNK {
215            flush(li, ri, ot, od);
216        }
217    });
218}
differential_dataflow/operators/int_proxy/join.rs

differential_dataflow/operators/int_proxy/
join.rs
