DDIR: half_join, on the corgi backend

interactive/Cargo.toml

@@ -16,6 +16,7 @@ columnar = { workspace = true }
 # The columnar kernels for the interpreted backend, pinned by git rev.
 corgi = { git = "https://github.com/frankmcsherry/WIP", rev = "41e873124a850772a9c4f7df88be94e3163512ef", features = ["serde"] }
 differential-dataflow = { workspace = true }
+differential-dogs3 = { path = "../dogsdogsdogs", default-features = false }
 serde = { version = "1.0", features = ["derive"] }
 smallvec = "1.15.1"
 timely = { workspace = true }

interactive/examples/triangles_bench.rs

@@ -0,0 +1,133 @@
+//! Triangles in a random graph, by binary joins or by a delta query of half joins. The binary
+//! joins arrange the two-paths; the delta query arranges only the edges, by source, by
+//! destination, and by themselves. Reports the time and memory to load the graph, and then the
+//! time per tick as edges are replaced.
+//!
+//! ```text
+//! cargo run --release --example triangles_bench -- [join|delta] [edges] [nodes] [churn] [ticks] [workers]
+//! ```
+
+use std::alloc::{GlobalAlloc, Layout, System};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::time::Instant;
+
+/// Counts live heap bytes and their peak, to tell the heap's size from the process's.
+struct Counting;
+static LIVE: AtomicUsize = AtomicUsize::new(0);
+static PEAK: AtomicUsize = AtomicUsize::new(0);
+unsafe impl GlobalAlloc for Counting {
+    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
+        let live = LIVE.fetch_add(layout.size(), Ordering::Relaxed) + layout.size();
+        PEAK.fetch_max(live, Ordering::Relaxed);
+        unsafe { System.alloc(layout) }
+    }
+    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
+        LIVE.fetch_sub(layout.size(), Ordering::Relaxed);
+        unsafe { System.dealloc(ptr, layout) }
+    }
+}
+#[global_allocator]
+static ALLOC: Counting = Counting;
+
+fn heap() -> String {
+    format!("heap {:.1} MiB (peak {:.1} MiB)", LIVE.load(Ordering::Relaxed) as f64 / 1048576.0, PEAK.load(Ordering::Relaxed) as f64 / 1048576.0)
+}
+
+use interactive::server::{RenderBackend, Server};
+use interactive::{lower, parse};
+
+const EDGES: &str = r#"
+    let e = input 0 : ((int, int) ; ()) | key(((x, y), ()) -> (x, y));
+    let r = e | key((x, y) -> (y, x));
+    let s = e | key((x, y) -> ((x, y), ()));
+"#;
+
+/// Each triangle `(a, b, c)` has edges `(a, b)`, `(b, c)`, and `(a, c)`.
+const JOIN: &str = r#"
+    let paths = r | join(e, (b, a, c) -> ((a, c), b));
+    let tri = paths | join(s, ((a, c), b, ()) -> ((a, b, c), ()));
+    export "count" = tri | key((_, _) -> ((), ())) | count;
+"#;
+
+/// One path of half joins per edge of the triangle, each meeting the earlier edges' changes at or
+/// before its own time and the later edges' strictly before.
+const DELTA: &str = r#"
+    let d1 = r | lift | half_join(e, 1, (b, a, c) -> ((a, c), b)) | half_join(s, 1, ((a, c), b, ()) -> ((a, b, c), ()));
+    let d2 = e | lift | half_join(r, 0, (b, c, a) -> ((a, c), b)) | half_join(s, 1, ((a, c), b, ()) -> ((a, b, c), ()));
+    let d3 = e | lift | half_join(e, 0, (a, c, b) -> ((b, c), a)) | half_join(s, 0, ((b, c), a, ()) -> ((a, b, c), ()));
+    let tri = (d1 + d2 + d3) | delay((_, (_, t)) -> t) | key((abc, (v, _)) -> (abc, v));
+    export "count" = tri | key((_, _) -> ((), ())) | count;
+"#;
+
+/// This process's memory in MiB: on macOS its `footprint`, elsewhere its RSS.
+fn memory_mib() -> Option<f64> {
+    let out = std::process::Command::new("footprint").args(["-p", &std::process::id().to_string()]).output();
+    if let Ok(out) = out {
+        let text = String::from_utf8_lossy(&out.stdout).into_owned();
+        let mut words = text.split("Footprint:").nth(1)?.split_whitespace();
+        let n: f64 = words.next()?.parse().ok()?;
+        return match words.next()? {
+            "B" => Some(n / (1024.0 * 1024.0)),
+            "KB" => Some(n / 1024.0),
+            "MB" => Some(n),
+            "GB" => Some(n * 1024.0),
+            _ => None,
+        };
+    }
+    let status = std::fs::read_to_string("/proc/self/status").ok()?;
+    let kib: f64 = status.lines().find(|l| l.starts_with("VmRSS:"))?.split_whitespace().nth(1)?.parse().ok()?;
+    Some(kib / 1024.0)
+}
+
+fn mem() -> String {
+    memory_mib().map_or_else(|| "n/a".to_string(), |m| format!("{m:>7.1} MiB"))
+}
+
+fn main() {
+    let mut args = std::env::args().skip(1);
+    let method = args.next().unwrap_or("delta".into());
+    let nums: Vec<u64> = args.map(|a| a.parse().expect("integer argument")).collect();
+    let edges = nums.first().copied().unwrap_or(1_000_000);
+    let nodes = nums.get(1).copied().unwrap_or(100_000);
+    let churn = nums.get(2).copied().unwrap_or(1_000);
+    let ticks = nums.get(3).copied().unwrap_or(20);
+    let workers = nums.get(4).copied().unwrap_or(1) as usize;
+    let query = match method.as_str() {
+        "join" => JOIN,
+        "delta" => DELTA,
+        other => panic!("unknown method {other:?}: join or delta"),
+    };
+    let source = format!("{EDGES}{query}");
+
+    timely::execute(timely::Config::process(workers), move |worker| {
+        let index = worker.index();
+        let mut server = Server::with_backend(RenderBackend::Corgi);
+        let mut program = lower::lower_tree(parse::pipe::parse(&source));
+        program.optimize();
+        server.install(worker, "p", &program).unwrap();
+        server.load(worker, "p", 0, &format!("random:nodes={nodes},edges={edges},churn={churn}")).unwrap();
+        let loaded = Instant::now();
+        server.tick(worker);
+        let count = |server: &mut Server, worker: &mut timely::worker::Worker| -> i64 {
+            server.snapshot(worker, "count").unwrap().iter().map(|(_, v, d)| match v { interactive::ir::Value::Int(n) => n * d, _ => 0 }).sum()
+        };
+        // A snapshot is a dataflow, so every worker takes part; worker 0 receives the rows.
+        let elapsed = loaded.elapsed();
+        let n = count(&mut server, worker);
+        if index == 0 {
+            println!("{method}: {edges} edges over {nodes} nodes, {workers} workers");
+            println!("load:  {:>9.1} ms  mem {}  {}  count {n}", elapsed.as_secs_f64() * 1e3, mem(), heap());
+        }
+        let churned = Instant::now();
+        for _ in 0..ticks { server.tick(worker); }
+        let elapsed = churned.elapsed();
+        let n = count(&mut server, worker);
+        if index == 0 {
+            println!(
+                "churn: {:>9.2} ms/tick over {ticks} ticks ({churn} edges replaced per tick)  mem {}  count {n}",
+                elapsed.as_secs_f64() * 1e3 / ticks as f64, mem(),
+            );
+        }
+    })
+    .unwrap();
+}

interactive/src/backend/corgi.rs

@@ -32,7 +32,9 @@ use crate::corgi::chunk::{recover_key, CorgiChunk, CorgiChunker};
 use crate::corgi::col_times::ColTime;
 use crate::corgi::container::CorgiContainer;
 use crate::corgi::exchange::CorgiPact;
+use crate::corgi::half_join::{CorgiHalfJoin, HalfJoinBatcher};
 use crate::corgi::join::CorgiJoinBackend;
+use differential_dogs3::operators::half_join::{half_join_with_tactic, HalfJoinTactic};
 use crate::corgi::reduce::CorgiReduceBackend;
 use differential_dataflow::operators::int_proxy::{ProxyJoinTactic, ProxyReduceTactic};
 use crate::corgi::logic::{compile_flatmap, compile_join_function, compile_predicate, compile_projection, compile_scalar, shape_of_row};
@@ -275,6 +277,23 @@ impl<B0: Native, B1: Native, J: JoinTactic<Time, NativeBatch, NativeBatch, CC>>
     }
 }
 
+/// A half-join tactic over native batches, accepting an imported arrangement.
+struct RetimeHalfJoin<J>(J);
+impl<B: Native, M, J: HalfJoinTactic<Time, NativeBatch, M, CC>> HalfJoinTactic<Time, B, M, CC> for RetimeHalfJoin<J> {
+    fn prep(&mut self, chunks: Vec<M>, batches: Vec<B>, lower: Antichain<Time>) -> Box<dyn Iterator<Item = (CC, Time)>> {
+        self.0.prep(chunks, batches.into_iter().map(Native::native).collect(), lower)
+    }
+}
+
+/// The time each coordinate of `t` less one, at least zero: the half join's compaction of its
+/// arrangement. Every later stream time is at or beyond some such frontier element, and advancing
+/// arranged times to it keeps their order against those stream times, strict and not.
+fn step_back(t: &Time, out: &mut Antichain<Time>) {
+    use differential_dataflow::dynamic::pointstamp::PointStamp;
+    let inner = PointStamp::new(t.inner.iter().map(|x| x.saturating_sub(1)).collect());
+    out.insert(timely::order::Product::new(t.outer.saturating_sub(1), inner));
+}
+
 /// A reduce tactic over native batches, accepting an imported input. (Its output is its own.)
 struct RetimeReduce<R>(R);
 impl<B1: Native, R: ReduceTactic<Time, NativeBatch, NativeBatch>> ReduceTactic<Time, B1, NativeBatch> for RetimeReduce<R> {
@@ -404,6 +423,17 @@ impl Backend for CorgiBackend {
         }
     }
 
+    fn half_join<'s>(stream: Collection<'s, Time, CC>, arrangement: Self::Arr<'s>, strict: bool, projection: &Projection, level: usize, _shapes: Option<(st::RowShape, st::RowShape)>) -> Collection<'s, Time, CC> {
+        // dogs3's driver owns progress and yielding; corgi supplies when an update is unblocked
+        // (`HalfJoinBatcher`) and how released updates meet the arrangement (`CorgiHalfJoin`).
+        // `CorgiPact` places each update on the worker holding its key, as `arrange` placed the rows.
+        let tactic = || RetimeHalfJoin(CorgiHalfJoin::new(projection.key.clone(), projection.val.clone(), strict, level + 1));
+        let yielding = |_timer: std::time::Instant, count: usize| count > 1_000_000;
+        match arrangement {
+            Arr::Local(a) => half_join_with_tactic(stream.inner, a, CorgiPact, step_back, yielding, HalfJoinBatcher::new(strict, level + 1), tactic()).as_collection(),
+            Arr::Imported(a) => half_join_with_tactic(stream.inner, a, CorgiPact, step_back, yielding, HalfJoinBatcher::new(strict, level + 1), tactic()).as_collection(),
+        }
+    }
     fn reduce<'s>(a: Self::Arr<'s>, reducer: &Reducer) -> Self::Arr<'s> {
         // Amortize columnar corrections while reusing sweep scratch across wide presentations.
         let tactic = RetimeReduce(ProxyReduceTactic::new(CorgiReduceBackend::new(reducer.clone())).with_key_batch_size(256));

interactive/src/backend/mod.rs

@@ -49,6 +49,9 @@ pub trait Backend {
     fn as_collection<'s>(a: Self::Arr<'s>) -> Collection<'s, Time, Self::Container>;
     /// `shapes` are the two inputs' row shapes, where they are known before any data arrives.
     fn join<'s>(l: Self::Arr<'s>, r: Self::Arr<'s>, projection: &Projection, shapes: Option<(st::RowShape, st::RowShape)>) -> Collection<'s, Time, Self::Container>;
+    /// See `scope_ir::Node::HalfJoin`. `level` is the scope depth, which fixes the width of a time
+    /// as data; `shapes` are the stream's and the arrangement's row shapes, where known.
+    fn half_join<'s>(stream: Collection<'s, Time, Self::Container>, arrangement: Self::Arr<'s>, strict: bool, projection: &Projection, level: usize, shapes: Option<(st::RowShape, st::RowShape)>) -> Collection<'s, Time, Self::Container>;
     fn reduce<'s>(a: Self::Arr<'s>, reducer: &Reducer) -> Self::Arr<'s>;
     fn inspect<'s>(c: Collection<'s, Time, Self::Container>, label: String) -> Collection<'s, Time, Self::Container>;
     fn leave_dynamic<'s>(c: Collection<'s, Time, Self::Container>, depth: usize) -> Collection<'s, Time, Self::Container>;
@@ -133,6 +136,7 @@ fn reads_rows(s: &st::Scope, k: usize) -> bool {
     let is_k = |r: &st::Ref| matches!(r, st::Ref::Import(i) if *i == k);
     s.items.iter().any(|item| match item {
         st::Item::Op(st::Node::Linear { input, .. } | st::Node::Inspect { input, .. }) => is_k(input),
+        st::Item::Op(st::Node::HalfJoin { stream, .. }) => is_k(stream),
         st::Item::Op(st::Node::Concat(refs)) => refs.iter().any(is_k),
         st::Item::Op(st::Node::Arrange(_) | st::Node::Join { .. } | st::Node::Reduce { .. }) => false,
         st::Item::Sub(_) => false,
@@ -189,6 +193,11 @@ pub fn render_tree<'s, B: Backend>(
                         let r = resolve(&items, &imports, &var_cols, right).arrange();
                         Rendered::Collection(B::join(l, r, projection, shape_of(left).zip(shape_of(right))))
                     },
+                    st::Node::HalfJoin { stream, arrangement, strict, projection } => {
+                        let c = resolve(&items, &imports, &var_cols, stream).collection();
+                        let a = resolve(&items, &imports, &var_cols, arrangement).arrange();
+                        Rendered::Collection(B::half_join(c, a, *strict, projection, depth, shape_of(stream).zip(shape_of(arrangement))))
+                    },
                     st::Node::Reduce { input, reducer } => {
                         let a = resolve(&items, &imports, &var_cols, input).arrange();
                         Rendered::Arrangement(B::reduce(a, reducer))

interactive/src/backend/vec.rs

@@ -207,6 +207,9 @@ impl Backend for VecBackend {
             (Arr::Imported(l), Arr::Imported(r)) => l.join_core(r, move |k, v1, v2| f(k, v1, v2)),
         }
     }
+    fn half_join<'s>(_stream: Collection<'s, Time, Self::Container>, _arrangement: Self::Arr<'s>, _strict: bool, _projection: &Projection, _level: usize, _shapes: Option<(st::RowShape, st::RowShape)>) -> Collection<'s, Time, Self::Container> {
+        panic!("half_join runs on the corgi backend only")
+    }
     fn reduce<'s>(a: Self::Arr<'s>, reducer: &Reducer) -> Self::Arr<'s> {
         let f: Arc<dyn Fn(&Row, &[(&Row, Diff)], &mut Vec<(Row, Diff)>) + Send + Sync> = match reducer {
             Reducer::Min => Arc::new(|_key, vals, output| { if let Some(min) = vals.iter().map(|(v, _)| (*v).clone()).min() { output.push((min, 1)); } }),

interactive/src/corgi/chunk.rs

@@ -640,7 +640,7 @@ pub fn recover_key_ref(keys: &CValue) -> &CValue {
 }
 
 /// Concatenate column blocks into one column (multi-source `gather_lanes`, no sort).
-fn concat_blocks(blocks: &[CValue]) -> CValue {
+pub(crate) fn concat_blocks(blocks: &[CValue]) -> CValue {
     if blocks.len() == 1 {
         return blocks[0].clone();
     }

interactive/src/corgi/col_times.rs

@@ -209,6 +209,11 @@ impl<T> ColTimes<T> {
         Ordering::Equal
     }
 
+    /// Whether this column's row `i` is less than or equal to `other`'s row `j` in the partial order.
+    pub fn less_equal_cross(&self, i: usize, other: &ColTimes<T>, j: usize) -> bool {
+        (0..self.lanes.len().max(other.lanes.len())).all(|k| self.at(k, i) <= other.at(k, j))
+    }
+
     /// Whether row `i` is less than or equal to row `j` in the partial order.
     fn less_equal(&self, i: usize, j: usize) -> bool {
         self.lanes.iter().all(|lane| lane[i] <= lane[j])