Corgi hash key lane

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

@@ -45,7 +45,10 @@
         bridge1: &mut ProxyBridge<B0::Time, Self::R1>,
     );
 
-    /// Interpret a list of matching identifiers, translate them to outputs, and place them in `output`.
+    /// Interpret matches derived from the immediately preceding [`Self::advance`] call and place
+    /// them in `output`. The iterator calls `cross` before another `advance`, so a backend may keep
+    /// block-local interpretation state between the two calls. `cross` may be skipped when the
+    /// block produced no matches, in which case the next `advance` may overwrite that state.
     fn cross(
         &mut self,
         instance: &JoinInstance<B0, B1>,
@@ -238,7 +241,7 @@
 /// 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>,
     r0: std::ops::Range<usize>,

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

@@ -147,7 +147,7 @@ impl<T, Bk> ProxyReduceTactic<T, Bk> {
     }
 }
 
-fn assert_pending_frontier<T: PartialOrder + Clone>(pending: &BTreeMap<u64, Vec<T>>, maintained: &Antichain<T>) {
+fn debug_assert_pending_frontier<T: PartialOrder + Clone>(pending: &BTreeMap<u64, Vec<T>>, maintained: &Antichain<T>) {
     debug_assert!({
         let mut expected = Antichain::new();
         for time in pending.values().flatten() { expected.insert_ref(time); }
@@ -214,7 +214,7 @@ where
         // beyond `upper` can remain when nothing is due, and releasing their capabilities would
         // strand them (see the frontier clause of the `ReduceTactic::retire` contract).
         if changed.is_empty() && instance.input_batches.iter().all(|b| b.is_empty()) {
-            assert_pending_frontier(&self.pending, &pending_frontier);
+            debug_assert_pending_frontier(&self.pending, &pending_frontier);
             return (Vec::new(), pending_frontier);
         }
 
@@ -397,7 +397,7 @@ where
         }
 
         let produced: Vec<(B1::Time, B2)> = tile_held.into_iter().zip(self.backend.finish()).collect();
-        assert_pending_frontier(&self.pending, &pending_frontier);
+        debug_assert_pending_frontier(&self.pending, &pending_frontier);
         (produced, pending_frontier)
     }
 }

interactive/src/backend/corgi.rs

@@ -24,7 +24,7 @@ use corgi::arrange::gather;
 use corgi::Value as CValue;
 
 use crate::backend::Backend;
-use crate::corgi::chunk::{CorgiChunk, CorgiChunker};
+use crate::corgi::chunk::{recover_key, CorgiChunk, CorgiChunker};
 use crate::corgi::container::CorgiContainer;
 use crate::corgi::join::CorgiJoinBackend;
 use crate::corgi::reduce::CorgiReduceBackend;
@@ -315,7 +315,9 @@ impl Backend for CorgiBackend {
                         for batch in data.iter() {
                             for ch in batch.chunks.iter().filter(|c| c.len() > 0) {
                                 let mut c = CorgiContainer {
-                                    keys: ch.keys().clone(),
+                                    // Drop the arrangement's leading identifier lane: edges carry
+                                    // the key the program wrote, so `$0` indexes what it always did.
+                                    keys: recover_key(ch.keys()),
                                     vals: ch.vals().clone(),
                                     times: ch.times().to_vec(),
                                     diffs: ch.diffs().to_vec(),

interactive/src/corgi/chunk.rs

@@ -416,6 +416,10 @@ where
     /// ingest — no transcode).
     pub fn from_columns(keys: CValue, vals: CValue, times: Vec<T>, diffs: Vec<R>) -> Self {
         let (keys, vals, times, diffs) = sort_consolidate(keys, vals, times, diffs);
+        debug_assert!({
+            let lane = corgi::arrange::leaf_slice(key_lane(&keys));
+            lane.is_some_and(|ids| ids.windows(2).all(|pair| pair[0] <= pair[1]))
+        }, "arrangement key must lead with a sorted u64 identifier lane");
         Self::from_parts(keys, vals, ColTimes::from_iter(times), diffs)
     }
 
@@ -472,6 +476,71 @@ impl<T: Columnar, R> Default for CorgiChunker<T, R> {
     }
 }
 
+/// An arrangement's key column, in the form every consumer of a `CorgiChunk` can rely on:
+/// **it leads with an integer, and the chunk is sorted by that integer.**
+///
+/// A key that is already a primitive integer (a bare 64-bit `Prim`, or the 1-field `Prod` that
+/// [`corgi::arrange::leaf_slice`] also reads through) is used as it stands — the value IS the
+/// identifier, injectively, and a hash lane would cost 8 bytes a row to say the same thing.
+/// Any other key shape — multi-field `Prod`, `List`, `Sum`, `Unit` — is hashed and the hash is
+/// PREPENDED, so the key becomes `Prod([hash, key])`. `CorgiChunk::from_columns` then sorts
+/// lexicographically over lanes, which is hash order with the real key as tie-break.
+///
+/// The original key stays in the column, which is what makes the hash safe: colliding keys land
+/// adjacent and sub-sorted, so they are told apart by comparison rather than by luck, and reads
+/// recover the real key by [`recover_key`]. The two forms are distinguishable after the fact
+/// (`leaf_slice` succeeds on exactly the un-prepended one) because no compound key reaches an
+/// arrangement un-prepended.
+///
+/// The hash is computed ONCE here, at ingest, and thereafter moves as data: `merge`, `advance`
+/// and `settle` permute key columns with `gather_lanes`, so no transducer recomputes it.
+pub fn present_key(keys: CValue) -> CValue {
+    if corgi::arrange::leaf_slice(&keys).is_some() {
+        return keys;
+    }
+    let hashes = corgi::hash(&keys).into_u64("present_key");
+    CValue::Prod(vec![CValue::u64(hashes), keys])
+}
+
+/// The integer identifier of each row of a [`present_key`] column: the key's own values when it is
+/// a primitive integer, and the prepended hash lane otherwise. Never re-hashes.
+pub fn key_ids(keys: &CValue) -> Vec<u64> {
+    if let Some(sl) = corgi::arrange::leaf_slice(keys) {
+        return sl.to_vec();
+    }
+    corgi::arrange::leaf_slice(key_lane(keys)).expect("a prepended hash lane is a u64 leaf").to_vec()
+}
+
+/// The single column an arrangement is sorted by, for seeking: the key itself when it is a
+/// primitive integer, else the prepended hash lane. Always a bare `u64` leaf, so `find_ranges`
+/// over it takes corgi's `u64` fast path whatever the underlying key shape.
+///
+/// One rule covers both forms, because [`present_key`] leaves exactly three possibilities: a bare
+/// `Prim`, the 1-field `Prod` that also counts as primitive, or a prepended `Prod([hash, key])`.
+/// The leading field is the identifier in all three.
+pub fn key_lane(keys: &CValue) -> &CValue {
+    match keys {
+        CValue::Prod(cols) => &cols[0],
+        _ => keys,
+    }
+}
+
+/// Whether [`present_key`] prepended a hash to this key — i.e. whether rows sharing an identifier
+/// may hold DIFFERENT keys. False for primitive-integer keys, whose identifier is injective, so
+/// readers can skip the checks that guard against collisions entirely.
+pub fn key_is_hashed(keys: &CValue) -> bool {
+    corgi::arrange::leaf_slice(keys).is_none()
+}
+
+/// Undo [`present_key`]: the key as the rest of the system knows it. A corgi clone is an `Arc`
+/// bump, so dropping the hash lane costs nothing.
+pub fn recover_key(keys: &CValue) -> CValue {
+    match keys {
+        CValue::Prod(cols) if corgi::arrange::leaf_slice(keys).is_none() => cols[1].clone(),
+        _ => keys.clone(),
+    }
+}
+
 /// Concatenate column blocks into one column (multi-source `gather_lanes`, no sort).
 fn concat_blocks(blocks: &[CValue]) -> CValue {
     if blocks.len() == 1 {
@@ -495,7 +564,7 @@ where
         if self.times.is_empty() {
             return;
         }
-        let keys = concat_blocks(&self.k_blocks);
+        let keys = present_key(concat_blocks(&self.k_blocks));
         let vals = concat_blocks(&self.v_blocks);
         self.k_blocks.clear();
         self.v_blocks.clear();