Generalize int-proxy keys and values to Copy + Ord

differential-dataflow/src/operators/int_proxy/history.rs

@@ -1,4 +1,4 @@
 //! Time-ordered replay of proxy update histories, with meet-advancement.
 
-/// A value history suitable for integer proxy values.
-pub(in crate::operators) type IdHistory<T, R> = crate::operators::history::ValueHistory<u64, T, R>;
+/// A value history suitable for ordered proxy values.
+pub(in crate::operators) type IdHistory<T, R, V = u64> = crate::operators::history::ValueHistory<V, T, R>;

differential-dataflow/src/operators/int_proxy/join.rs

@@ -1,7 +1,7 @@
 //! The proxy join framework.
 //!
-//! A conventional differential join against `(u64, u64)` values, which are provided by
-//! and then interpreted by a backend, who is relieved of lattice-time reasoning.
+//! A conventional differential join against ordered, copyable proxies.
+//! The backend interprets proxies and is relieved of lattice-time reasoning.
 
 use std::cell::RefCell;
 use std::rc::Rc;
@@ -10,17 +10,23 @@ use timely::progress::Timestamp;
 
 use crate::difference::{Multiply, Semigroup};
 use crate::lattice::Lattice;
-use super::ProxyBridge;
+use super::{KeyPosition, ProxyBridge};
 use crate::operators::join::{Fresh, JoinTactic};
 use crate::operators::history::ValueHistory;
 
 use super::history::IdHistory;
 
-/// A type that can interpret and retire pairs of lists of batches, joined by key hashes.
+/// A type that can interpret and retire pairs of lists of batches, joined by keys.
 ///
 /// The harness repeatedly invokes [`advance`](Self::advance) to draw a block of the proxy collection,
 /// then [`cross`](Self::cross) to turn that block's matches into output containers, until `advance` reports the key space exhausted.
 pub trait ProxyJoinBackend<T, B0, B1> {
+    /// Independent groups, common to both inputs.
+    type Key: Copy + Ord;
+    /// First input value proxies, valid throughout a block.
+    type V0: Copy + Ord;
+    /// Second input value proxies, valid throughout a block.
+    type V1: Copy + Ord;
     /// Diff type of the first input.
     type R0: Semigroup + Multiply<Self::R1, Output = Self::ROut>;
     /// Diff type of the second input.
@@ -32,16 +38,17 @@ pub trait ProxyJoinBackend<T, B0, B1> {
 
     /// Populates the two bridges with all updates for all keys that match in a returned range.
     ///
-    /// The `from` indicates an inclusive lower bound on key hash, and should be updated by the implementor to an exclusive
-    /// upper bound for the range of keys it intends to return in this call. The `None` value indicates the keys are exhausted.
+    /// The `from` position is `Start` or an inclusive `At(key)` lower bound.
+    /// The backend writes the window's exclusive upper bound as `At(key)`, or `End` when exhausted.
+    /// Each call must strictly advance `from`.
     /// The returned bridges must contain all updates from both `instance` inputs for keys that are present in both inputs, and
     /// which are greater or equal to the initial `from`, and not greater or equal to its value when returned.
     fn advance(
         &mut self,
         instance: &JoinInstance<T, B0, B1>,
-        from: &mut Option<u64>,
-        bridge0: &mut ProxyBridge<T, Self::R0>,
-        bridge1: &mut ProxyBridge<T, Self::R1>,
+        from: &mut KeyPosition<Self::Key>,
+        bridge0: &mut ProxyBridge<T, Self::R0, Self::Key, Self::V0>,
+        bridge1: &mut ProxyBridge<T, Self::R1, Self::Key, Self::V1>,
     );
 
     /// Interpret matches derived from the immediately preceding [`Self::advance`] call and place
@@ -51,7 +58,7 @@ pub trait ProxyJoinBackend<T, B0, B1> {
     fn cross(
         &mut self,
         instance: &JoinInstance<T, B0, B1>,
-        matches: &mut JoinMatches<T, Self::ROut>,
+        matches: &mut JoinMatches<T, Self::ROut, Self::Key, Self::V0, Self::V1>,
         output: &mut Vec<Self::Output>,
     );
 }
@@ -71,20 +78,22 @@ pub struct JoinInstance<T, B0, B1> {
 /// Presentation of discovered join matches.
 ///
 /// The arrays have common lengths, and are in key order but may not be consolidated.
-pub struct JoinMatches<T, R> {
+pub struct JoinMatches<T, R, K = u64, V0 = u64, V1 = u64> {
     /// Triples of `(key, (val0, val1))` of matches.
-    pub ids: Vec<(u64, (u64, u64))>,
+    pub ids: Vec<(K, (V0, V1))>,
     /// Times of the updates.
     pub times: Vec<T>,
     /// Diffs of the updates.
     pub diffs: Vec<R>,
 }
 
-impl<T, R> Default for JoinMatches<T, R> {
+impl<T, R, K, V0, V1> Default for JoinMatches<T, R, K, V0, V1> {
     fn default() -> Self { Self { ids: vec![], times: vec![], diffs: vec![] } }
 }
 
-/// A proxy-space [`JoinTactic`]: matches records of the two drawn runs by `key_hash`.
+/// A proxy-space [`JoinTactic`]: matches records of the two drawn runs by `key`.
+/// Its deferred work owns the presentations, so proxy types must be `'static`.
+/// Borrowing proxies from a backend's temporary storage requires a different presentation interface.
 pub struct ProxyJoinTactic<B0, B1, Bk> {
     backend: Rc<RefCell<Bk>>,
     _marker: std::marker::PhantomData<(B0, B1)>,
@@ -104,12 +113,15 @@ where
     B1: 'static,
     Bk: ProxyJoinBackend<T, B0, B1> + 'static,
     Bk::Output: 'static,
+    Bk::Key: 'static,
+    Bk::V0: 'static,
+    Bk::V1: 'static,
 {
     fn prep(&mut self, input0: Vec<B0>, input1: Vec<B1>, _fresh: Fresh, meet: T) -> Box<dyn Iterator<Item = Bk::Output>> {
         Box::new(ProxyJoinIter {
             backend: Rc::clone(&self.backend),
             instance: JoinInstance { batches0: input0, batches1: input1, lower: meet },
-            from: Some(0),
+            from: KeyPosition::Start,
             p0: Vec::new(),
             p1: Vec::new(),
             h0: IdHistory::new(),
@@ -133,17 +145,16 @@ where
     backend: Rc<RefCell<Bk>>,
     /// The iterator's inputs, and the time at which they can consolidate as they load.
     instance: JoinInstance<T, B0, B1>,
-    /// Progress through the key space: `Some(h)` for key hashes at or above `h` remaining, `None`
-    /// once the backend reports the iteration is complete.
-    from: Option<u64>,
+    /// Progress through the key space, from `Start` through inclusive key bounds to `End`.
+    from: KeyPosition<Bk::Key>,
     /// The current block: the two runs `advance` last drew, which one `next` consumes entirely.
-    p0: ProxyBridge<T, Bk::R0>,
-    p1: ProxyBridge<T, Bk::R1>,
+    p0: ProxyBridge<T, Bk::R0, Bk::Key, Bk::V0>,
+    p1: ProxyBridge<T, Bk::R1, Bk::Key, Bk::V1>,
     /// Per-key replay histories, held across the iterator and reloaded per key when needed.
-    h0: IdHistory<T, Bk::R0>,
-    h1: IdHistory<T, Bk::R1>,
+    h0: IdHistory<T, Bk::R0, Bk::V0>,
+    h1: IdHistory<T, Bk::R1, Bk::V1>,
     /// The block's matched records, held across blocks to keep their allocations.
-    matches: JoinMatches<T, Bk::ROut>,
+    matches: JoinMatches<T, Bk::ROut, Bk::Key, Bk::V0, Bk::V1>,
     /// The last block's containers, in reverse, served from the back one `next` at a time.
     ready: Vec<Bk::Output>,
 }
@@ -157,7 +168,7 @@ where
 
     /// Serve a ready container, else draw and cross blocks until one yields any.
     fn next(&mut self) -> Option<Bk::Output> {
-        while self.ready.is_empty() && self.from.is_some() {
+        while self.ready.is_empty() && self.from != KeyPosition::End {
             self.refill();
             self.work();
             if !self.matches.ids.is_empty() { self.cross(); }
@@ -179,20 +190,19 @@ where
         self.backend.borrow_mut().advance(&self.instance, &mut self.from, &mut self.p0, &mut self.p1);
         // Without progress the iterator would never retire, so this guards liveness as well as contract.
         debug_assert!(
-            self.from.is_none() || self.from > before,
+            self.from > before,
             "advance must either strictly increase `from` or report the iteration complete",
         );
         super::debug_assert_sorted_bridge(&self.p0, "advance (bridge0)");
         super::debug_assert_sorted_bridge(&self.p1, "advance (bridge1)");
-        // A key hash outside `[before, from)` is either one an earlier block already retired, or one
-        // a later block may yet report: both split a key across blocks, which silently drops the
-        // matches that would have crossed the split.
+        // A key outside `[before, from)` belongs to an earlier or later block.
+        // Both cases split a key across blocks, silently dropping matches that would cross the split.
         debug_assert!(
             {
                 let mut keys = self.p0.iter().map(|r| r.0.0).chain(self.p1.iter().map(|r| r.0.0));
-                keys.all(|k| before.is_none_or(|b| b <= k) && self.from.is_none_or(|f| k < f))
+                keys.all(|k| before <= KeyPosition::At(k) && KeyPosition::At(k) < self.from)
             },
-            "advance must report a key hash entirely within the block that first mentions it",
+            "advance must report a key entirely within the block that first mentions it",
         );
     }
 
@@ -205,7 +215,7 @@ where
         let (mut i, mut j) = (0usize, 0usize);
         while i < p0.len() && j < p1.len() {
             let ki = p0[i].0.0;
-            debug_assert_eq!(ki, p1[j].0.0, "advance must report common keys");
+            debug_assert!(ki == p1[j].0.0, "advance must report common keys");
             let mut e0 = i;
             while e0 < p0.len() && p0[e0].0.0 == ki { e0 += 1; }
             let mut e1 = j;
@@ -237,15 +247,15 @@ where
 /// If either history is small, this performs a direct cross product.
 /// If both histories are large, this replays the histories compacting as it goes in
 /// order to (potentially) avoid quadratic blow-up.
-fn join_key<T, R0, R1, RO>(
-    kh: u64,
-    p0: &ProxyBridge<T, R0>,
+fn join_key<T, R0, R1, RO, K: Copy + Ord, V0: Copy + Ord, V1: Copy + Ord>(
+    kh: K,
+    p0: &ProxyBridge<T, R0, K, V0>,
     r0: std::ops::Range<usize>,
-    p1: &ProxyBridge<T, R1>,
+    p1: &ProxyBridge<T, R1, K, V1>,
     r1: std::ops::Range<usize>,
-    h0: &mut IdHistory<T, R0>,
-    h1: &mut IdHistory<T, R1>,
-    matches: &mut JoinMatches<T, RO>,
+    h0: &mut IdHistory<T, R0, V0>,
+    h1: &mut IdHistory<T, R1, V1>,
+    matches: &mut JoinMatches<T, RO, K, V0, V1>,
 ) where
     T: Lattice + Timestamp,
     R0: Semigroup + Multiply<R1, Output = RO> + Clone,
@@ -280,12 +290,13 @@ fn join_key<T, R0, R1, RO>(
 /// `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.
-fn bilinear_wave<V, T, R0, R1, RO>(
-    h0: &mut ValueHistory<V, T, R0>,
-    h1: &mut ValueHistory<V, T, R1>,
-    mut emit: impl FnMut(V, V, T, RO),
+fn bilinear_wave<V0, V1, T, R0, R1, RO>(
+    h0: &mut ValueHistory<V0, T, R0>,
+    h1: &mut ValueHistory<V1, T, R1>,
+    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<R1, Output = RO> + Clone,
     R1: Semigroup + Clone,

differential-dataflow/src/operators/int_proxy/mod.rs

@@ -1,17 +1,19 @@
-//! Backend-agnostic operator tactics using integer proxies.
+//! Backend-agnostic operator tactics using ordered, copyable proxies.
 //!
-//! The tactics are intended to support custom operator implementations without rebuilding
-//! the non-trivial and often non-obvious time-based logic that supports them.
+//! The tactics support custom operator implementations without rebuilding the non-trivial and often non-obvious time-based logic that supports them.
+//! They run DD's operator logic over consolidated `[((key, val), time, diff)]` lists.
+//! Keys identify independent groups; values identify data within a group.
+//! Both require only `Copy + Ord` and need not be the types presented to user logic.
+//! The tactics elicit proxies, perform time and difference computations, and return proxies to the backend for interpretation.
+//! The backend is oblivious to the navigation of time, and the operator to the backend's implementation.
 //!
-//! The tactics here run DD's operator logic over consolidated `[((u64, u64), time, diff)]`
-//! lists, the first integer a hash of the "key" and granule of independence, the second an
-//! ephemeral data identifier understood by the backend but opaque to the operator harness.
-//! The tactics first elicit proxy identifiers from the backends, perform their necessary time
-//! and difference based computations to stage integer collections, and then re-invoke the
-//! backends with those same identifiers to produce the necessary output.
+//! The default proxy types remain `u64`, with the interpretation described below.
+//! Exact key proxies do not require collision handling.
 //!
-//! The backend is oblivious to the navigation of time, and the operator to the backend's
-//! implementation.
+//! Input and output value proxies may have different types.
+//! Standalone reduce calls can use borrowed strings whose storage outlives the call and any pending keys.
+//! Deferred join work still requires `'static` proxies.
+//! Borrowing directly from backend-owned presentation storage needs a further lifetime or container abstraction.
 //!
 //! # The two integers
 //!
@@ -49,21 +51,33 @@ pub mod reduce;
 mod pending;
 pub mod vec_backend;
 
-/// Integer-only exchange medium: a consolidated collection of `[((hash, id), time, diff)]`.
+/// A consolidated collection of `[((key, val), time, diff)]` proxies.
 ///
 /// The [`debug_assert_sorted_bridge`] method is (and can be) used to validate this property.
-pub type ProxyBridge<T, R> = Vec<((u64, u64), T, R)>;
+pub type ProxyBridge<T, R, K = u64, V = u64> = Vec<((K, V), T, R)>;
 
 /// Debug check that a presented [`ProxyBridge`] is consolidated.
 ///
 /// Operator harnesses use the test to flag backend implementations that do not uphold it.
-pub(crate) fn debug_assert_sorted_bridge<T: Ord, R>(bridge: &ProxyBridge<T, R>, who: &str) {
+pub(crate) fn debug_assert_sorted_bridge<T: Ord, R, K: Copy + Ord, V: Copy + Ord>(bridge: &ProxyBridge<T, R, K, V>, who: &str) {
     debug_assert!(
         bridge.windows(2).all(|w| (w[0].0, &w[0].1) < (w[1].0, &w[1].1)),
-        "{}: a presented bridge must be sorted & consolidated by ((key_hash, value_id), time)",
+        "{}: a presented bridge must be sorted & consolidated by ((key, val), time)",
         who,
     );
 }
 
 pub use join::{JoinInstance, ProxyJoinBackend, ProxyJoinTactic};
 pub use reduce::{ProxyReduceBackend, ProxyReduceTactic, ReduceInstance, ReduceWindow};
+
+/// A position between windows of an ordered key space.
+/// No minimum key or successor operation is required.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
+pub enum KeyPosition<K> {
+    /// No keys have been visited.
+    Start,
+    /// The inclusive lower bound on keys still to visit.
+    At(K),
+    /// All keys have been visited.
+    End,
+}

differential-dataflow/src/operators/int_proxy/pending.rs

@@ -3,28 +3,28 @@ use std::ops::Range;
 use timely::progress::{Antichain, frontier::AntichainRef, Timestamp};
 
 /// Activated associations refer to a shared time table, sorted by (key, time).
-pub(super) struct Due<T> {
+pub(super) struct Due<T, K> {
     pub times: Vec<T>,
-    pub rows: Vec<(u64, usize)>,
+    pub rows: Vec<(K, usize)>,
 }
 
-struct Run<T> {
+struct Run<T, K> {
     times: Vec<T>,
-    keys: Vec<u64>,
+    keys: Vec<K>,
     ends: Vec<usize>,
     live: Vec<usize>,
     live_keys: usize,
     frontier: Antichain<T>,
 }
-impl<T: Timestamp> Run<T> {
+impl<T: Timestamp, K: Copy + Ord> Run<T, K> {
     fn empty() -> Self {
         Self { times: vec![], keys: vec![], ends: vec![], live: vec![], live_keys: 0, frontier: Antichain::new() }
     }
     fn range(&self, row: usize) -> Range<usize> {
         (if row == 0 { 0 } else { self.ends[row - 1] })..self.ends[row]
     }
     fn weight(&self) -> usize { self.live.len() + self.live_keys }
-    fn new(mut pairs: Vec<(T, u64)>) -> Self {
+    fn new(mut pairs: Vec<(T, K)>) -> Self {
         pairs.sort_unstable();
         pairs.dedup();
         let mut run = Self::empty();
@@ -83,12 +83,12 @@ impl<T: Timestamp> Run<T> {
     }
 }
 
-pub(super) struct Pending<T> { runs: Vec<Option<Run<T>>> }
-impl<T> Default for Pending<T> {
+pub(super) struct Pending<T, K = u64> { runs: Vec<Option<Run<T, K>>> }
+impl<T, K> Default for Pending<T, K> {
     fn default() -> Self { Self { runs: vec![] } }
 }
-impl<T: Timestamp> Pending<T> {
-    pub fn insert(&mut self, pairs: Vec<(T, u64)>) {
+impl<T: Timestamp, K: Copy + Ord> Pending<T, K> {
+    pub fn insert(&mut self, pairs: Vec<(T, K)>) {
         if pairs.is_empty() { return; }
         let mut run = Run::new(pairs);
         loop {
@@ -105,7 +105,7 @@ impl<T: Timestamp> Pending<T> {
         }
         frontier
     }
-    pub fn activate(&mut self, upper: AntichainRef<T>) -> Due<T> {
+    pub fn activate(&mut self, upper: AntichainRef<T>) -> Due<T, K> {
         let mut due = Due { times: vec![], rows: vec![] };
         for bin in &mut self.runs {
             let Some(run) = bin else { continue; };
@@ -133,7 +133,7 @@ impl<T: Timestamp> Pending<T> {
                 *bin = Some(old.merge(Run::empty()));
             }
         }
-        // Rank the small time table once; association sorting then compares only integers.
+        // Rank the small time table once; association sorting then compares keys and integer time ranks.
         let mut order: Vec<_> = (0..due.times.len()).collect();
         order.sort_unstable_by_key(|&row| &due.times[row]);
         let mut ranks = vec![0; order.len()];