From cb16928c5d752c00a73f1b87f3e663a8627ecb71 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 14 Aug 2026 09:56:51 -0400 Subject: [PATCH 01/14] Arrangements present an integer key, and are sorted by it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Corgi must hand the proxy seam an integer key, and the seam works properly only when the arrangement is already sorted by that integer. Until now corgi supplied the integer but not the order: `ids()` derived it per retire and threw it away, so for anything but a primitive-integer key the arrangement was in structural key order and the identifier was in some other order. Everything downstream paid for that mismatch — the identifier was re-hashed every retire, and `changed` could not be turned into needles, so the accumulated history was scanned in full. The rule is now applied once, at ingest (`CorgiChunker::flush`): a key that is already a primitive integer is used as it stands; any other key shape is hashed and the hash PREPENDED, making the key `Prod([hash, key])`. `from_columns` sorts lexicographically over lanes, so that is hash order with the real key as tie-break, and the identifier is lane 0 either way. The real key stays in the column, so colliding keys land adjacent and sub-sorted rather than indistinguishable, and reads recover it by dropping a lane (an `Arc` bump). The hash is computed once and thereafter moves as data: merge, advance and settle permute key columns with `gather_lanes`, so no transducer recomputes it. This is the distinction the earlier hash-ordering attempt missed — a coordinate is carried, a sidecar is recomputed. What this deletes: * `collect_present`'s scan branch, and with it `seek_needles`' `Option`, the seek-vs-scan heuristic and `SEEK_ADVANTAGE`. Every key shape now seeks, and seeks by `find_ranges` over a `u64` leaf — corgi's fast path. * `ids()` on keys. It reads lane 0. Egress strips the lane in the two places a key leaves an arrangement: the container built by `as_collection`, and the key handed to the join's projection, which is written against the key the program declared. Values are untouched — value order is load-bearing for Min/Collect, and value identifiers are a separate question. 53 tests, plus the 4 heavy release ones. `pair_keys` caught the join egress. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01YH7iq9JoXmf7ATpq1gZaQS --- interactive/src/backend/corgi.rs | 6 +- interactive/src/corgi/chunk.rs | 56 ++++++++++++++++- interactive/src/corgi/join.rs | 7 ++- interactive/src/corgi/reduce.rs | 103 +++++++------------------------ 4 files changed, 88 insertions(+), 84 deletions(-) diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs index 6eec6d6fd..8b1d3efd7 100644 --- a/interactive/src/backend/corgi.rs +++ b/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(), diff --git a/interactive/src/corgi/chunk.rs b/interactive/src/corgi/chunk.rs index 4b33b21a3..b9c81e454 100644 --- a/interactive/src/corgi/chunk.rs +++ b/interactive/src/corgi/chunk.rs @@ -472,6 +472,60 @@ impl Default for CorgiChunker { } } +/// 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 { + 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 u64 leaf, so `find_ranges` over it +/// takes corgi's u64 fast path whatever the underlying key shape. +pub fn key_lane(keys: &CValue) -> &CValue { + match keys { + CValue::Prod(cols) if corgi::arrange::leaf_slice(keys).is_none() => &cols[0], + _ => keys, + } +} + +/// 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 +549,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(); diff --git a/interactive/src/corgi/join.rs b/interactive/src/corgi/join.rs index 82e514075..2d11d118e 100644 --- a/interactive/src/corgi/join.rs +++ b/interactive/src/corgi/join.rs @@ -40,7 +40,7 @@ use differential_dataflow::trace::chunk::{Chunk, ChunkBatch}; use corgi::arrange::{compare_at, find_ranges, gather, gather_lanes}; use corgi::{shape_of_value, Shape, Value as CValue}; -use crate::corgi::chunk::CorgiChunk; +use crate::corgi::chunk::{recover_key, CorgiChunk}; use crate::corgi::col_times::ColTime; use crate::corgi::container::CorgiContainer; use crate::corgi::logic::compile_join_projection; @@ -119,7 +119,10 @@ impl ProxyJoinBackend, CBatch> for CorgiJoinBackend tag1.push((c1 >> COORD_BITS) as usize); off1.push((c1 & ((1 << COORD_BITS) - 1)) as usize); } - let kc = gather_lanes(&keys0, &tag0, &off0); + // The join's projection is written against the key the program declared, so drop + // the arrangement's leading identifier lane before evaluating it. The output goes to + // an arrange, which re-derives the identifier for the new key. + let kc = recover_key(&gather_lanes(&keys0, &tag0, &off0)); let v0 = gather_lanes(&vals0, &tag0, &off0); let v1 = gather_lanes(&vals1, &tag1, &off1); let proj = compile_join_projection(&self.key, &self.val, &shape_of_value(&kc), &shape_of_value(&v0), &shape_of_value(&v1)); diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index 12b01738f..56ea596a2 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -38,10 +38,10 @@ use differential_dataflow::operators::int_proxy::ProxyBridge; use differential_dataflow::operators::int_proxy::reduce::{ProxyReduceBackend, ReduceInstance, ReduceWindow}; use corgi::arrange::{find_ranges, gather, gather_lanes, sort_blocks}; -use corgi::{Bounds, Shape, Value as CValue}; +use corgi::{Bounds, Value as CValue}; use crate::corgi::col_times::ColTime; -use crate::corgi::chunk::{columns_to_batch, CorgiChunk}; +use crate::corgi::chunk::{columns_to_batch, key_ids, key_lane, CorgiChunk}; use crate::ir::Diff; use crate::parse::Reducer; @@ -157,7 +157,7 @@ fn concat_columns(blocks: &[CValue]) -> CValue { } } -/// Id column for a key/value column. For a PRIMITIVE column — a bare 64-bit `Prim`, or a 1-field +/// Id column for a VALUE column. For a PRIMITIVE column — a bare 64-bit `Prim`, or a 1-field /// `Prod([Prim(64)])` — the value itself is already a collision-free id (`i64 as u64` is a bijection), /// so pass it straight through and skip the content hash. Compound shapes (Unit / List / Sum / /// multi-field `Prod`) hash via the CANONICAL native `corgi::hash` (the designed boundary-id fold, @@ -178,93 +178,38 @@ fn ids(col: &CValue) -> Vec { corgi::hash(col).into_u64("ids") } -/// The `changed` set as a needle column in the chunks' own key shape — possible exactly -/// when `ids` uses key VALUES (a bare `u64` leaf, or a 1-tuple of one); the hashed ids of -/// structural keys cannot be inverted into needles. -fn seek_needles(sample: &CValue, changed: &[u64]) -> Option { - match corgi::shape_of_value(sample) { - Shape::Prim(64) => Some(CValue::u64(changed.to_vec())), - Shape::Prod(ref fs) if fs.len() == 1 && matches!(fs[0], Shape::Prim(64)) => { - Some(CValue::Prod(vec![CValue::u64(changed.to_vec())])) - } - _ => None, - } -} - /// Concatenate the records of the `changed` keys across a run of chunks into parallel /// `(keys_col, vals_col)` corgi columns plus per-record `(key_hash, time, diff)`. `changed` is the -/// ASCENDING set of changed key hashes; a row is kept iff its key hash is in it. -/// -/// Seek-vs-scan, decided per retire, now that the sizes are known: seeking the changed keys -/// (`find_ranges`, O(|changed|·log rows) per chunk, no key hashing at all) wins when the -/// changed set is narrow — the steady incremental case; the full scan (O(rows) per chunk, -/// plus each chunk's key hashes re-derived) wins for broad churn — loads and label-cascade -/// retires, where most keys change and a gallop per key only adds overhead. Seeking requires -/// ids that ARE key values (single-leaf keys, `ids`' fast paths): hashed ids of structural -/// keys cannot be inverted into needles, so those always scan. +/// ASCENDING set of changed key ids; a row is kept iff its key id is in it. /// -/// TODO: the scan's per-row work can still batch: `ids` re-derives (and copies) each chunk's -/// key hashes every retire (memoize per chunk, or a stored hash column), and each hit -/// materializes an owned time (`times().get`); kept RANGES could move via `push_range`. +/// Always a seek. An arrangement's key leads with its identifier and is sorted by it +/// ([`present_key`](crate::corgi::chunk::present_key)), so the changed set is a `u64` needle column +/// over [`key_lane`] whatever the key's real shape — `find_ranges` on a `u64` leaf, corgi's fast +/// path. This is what the stored identifier bought: the alternative branch used to be a full +/// O(rows) scan per chunk with that chunk's key hashes re-derived, taken by every structural key +/// because a hash computed on the fly cannot be inverted into a needle. fn collect_present(chunks: &[&CorgiChunk], changed: &[u64]) -> (CValue, CValue, Vec, Vec, Vec) where T: ColTime, { - /// Seek only when the changed set is at least this many times narrower than the - /// presented rows: a `find_ranges` probe is a structurally-dispatched binary search - /// (~log(rows) compares, each far costlier than the scan's flat membership test), so - /// marginal seeks LOSE to the scan — measured, not modeled; 16 regressed load-shaped - /// retires before this was widened. - const SEEK_ADVANTAGE: usize = 64; - let key_srcs: Vec> = chunks.iter().map(|c| Some(c.keys())).collect(); let val_srcs: Vec> = chunks.iter().map(|c| Some(c.vals())).collect(); let (mut tags, mut offs) = (Vec::new(), Vec::new()); let (mut khs, mut times, mut diffs) = (Vec::new(), Vec::new(), Vec::new()); - let total: usize = chunks.iter().map(|c| c.diffs().len()).sum(); - let needles = if changed.len().saturating_mul(SEEK_ADVANTAGE) < total { - chunks.iter().find(|c| c.diffs().len() > 0).and_then(|c| seek_needles(c.keys(), changed)) - } else { - None - }; - if let Some(needles) = needles { - // Narrow changed set over seekable keys: gallop each chunk once per changed key. - // Chunks are key-ordered and `changed` ascends, so emission order matches the scan's. - for (ci, ch) in chunks.iter().enumerate() { - if ch.diffs().is_empty() { - continue; - } - let (lo, hi) = find_ranges(&needles, ch.keys()); - for (j, (&l, &h)) in lo.iter().zip(hi.iter()).enumerate() { - for i in l..h { - tags.push(ci); - offs.push(i); - khs.push(changed[j]); - times.push(ch.times().get(i)); - diffs.push(ch.diffs()[i]); - } - } + let needles = CValue::u64(changed.to_vec()); + // Chunks are id-ordered and `changed` ascends, so emission order is the merged order. + for (ci, ch) in chunks.iter().enumerate() { + if ch.diffs().is_empty() { + continue; } - } else { - for (ci, ch) in chunks.iter().enumerate() { - // Borrow the key leaf when there is one (`ids`' value-as-id fast paths); only - // structural keys need the hash, and only they pay a materialization. A shared - // column's `Arc` cannot be unwrapped, so `ids` would copy the whole key column - // here, once per chunk per retire, to read values it never mutates. - let hashed: Option> = corgi::arrange::leaf_slice(ch.keys()).is_none().then(|| ids(ch.keys())); - let kh: &[u64] = match (&hashed, corgi::arrange::leaf_slice(ch.keys())) { - (Some(v), _) => &v[..], - (None, Some(sl)) => sl, - (None, None) => unreachable!("leaf_slice absent implies hashed present"), - }; - for i in 0..kh.len() { - if changed.binary_search(&kh[i]).is_ok() { - tags.push(ci); - offs.push(i); - khs.push(kh[i]); - times.push(ch.times().get(i)); - diffs.push(ch.diffs()[i]); - } + let (lo, hi) = find_ranges(&needles, key_lane(ch.keys())); + for (j, (&l, &h)) in lo.iter().zip(hi.iter()).enumerate() { + for i in l..h { + tags.push(ci); + offs.push(i); + khs.push(changed[j]); + times.push(ch.times().get(i)); + diffs.push(ch.diffs()[i]); } } } @@ -494,7 +439,7 @@ where // key hashes come from the scan the key list needs anyway. let mut seeds: Vec<(u64, T)> = Vec::new(); for ch in novel_chunks.iter() { - let khs = ids(ch.keys()); + let khs = key_ids(ch.keys()); let times = ch.times(); for (i, kh) in khs.iter().enumerate() { seeds.push((*kh, times.get(i))); From 488dc4a328a0fb52767bce020371c33c28f75342 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 14 Aug 2026 10:05:10 -0400 Subject: [PATCH 02/14] Keep the seek-vs-scan choice; the identifier lane makes both branches cheap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deleting the scan branch cost the primitive-integer programs: scc +1.2%, ident +3.4%, both of which this change should not have touched at all. The heuristic was earning its keep — broad-churn retires (loads, label cascades) present most of the key space, and a gallop per changed key loses to a flat membership test there. Restored, with both branches reading the identifier lane. That is strictly better than the version this branch started from: the seek is `find_ranges` over a `u64` leaf for EVERY key shape, and the scan BORROWS the lane rather than materializing it — a structural key used to be forced onto the scan (its on-the-fly hash could not be inverted into a needle) and forced to allocate a hash vector per chunk per retire to take it. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01YH7iq9JoXmf7ATpq1gZaQS --- interactive/src/corgi/reduce.rs | 59 ++++++++++++++++++++++++--------- 1 file changed, 44 insertions(+), 15 deletions(-) diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index 56ea596a2..5c8919859 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -182,34 +182,63 @@ fn ids(col: &CValue) -> Vec { /// `(keys_col, vals_col)` corgi columns plus per-record `(key_hash, time, diff)`. `changed` is the /// ASCENDING set of changed key ids; a row is kept iff its key id is in it. /// -/// Always a seek. An arrangement's key leads with its identifier and is sorted by it -/// ([`present_key`](crate::corgi::chunk::present_key)), so the changed set is a `u64` needle column -/// over [`key_lane`] whatever the key's real shape — `find_ranges` on a `u64` leaf, corgi's fast -/// path. This is what the stored identifier bought: the alternative branch used to be a full -/// O(rows) scan per chunk with that chunk's key hashes re-derived, taken by every structural key -/// because a hash computed on the fly cannot be inverted into a needle. +/// Seek or scan, decided per retire now that the sizes are known: seeking the changed keys wins +/// when the changed set is narrow (the steady incremental case), and a flat membership scan wins +/// for broad churn — loads and label-cascade retires, where most keys change and a gallop per key +/// only adds overhead. +/// +/// What the stored identifier changed is that BOTH branches are now available, and cheap, for every +/// key shape. An arrangement's key leads with its identifier and is sorted by it +/// ([`present_key`](crate::corgi::chunk::present_key)), so [`key_lane`] is a sorted `u64` leaf: the +/// seek is `find_ranges` over it (corgi's `u64` fast path) and the scan borrows it outright, with no +/// hashing and no allocation on either side. Previously a structural key could do neither — its +/// identifier was hashed per chunk per retire, and a hash derived on the fly cannot be inverted into +/// a needle, so those keys were forced onto the scan and forced to materialize to take it. fn collect_present(chunks: &[&CorgiChunk], changed: &[u64]) -> (CValue, CValue, Vec, Vec, Vec) where T: ColTime, { + /// Seek only when the changed set is at least this many times narrower than the presented + /// rows: a `find_ranges` probe costs ~log(rows) compares against the scan's flat membership + /// test, so marginal seeks LOSE — measured, not modeled; 16 regressed load-shaped retires + /// before this was widened. + const SEEK_ADVANTAGE: usize = 64; + let key_srcs: Vec> = chunks.iter().map(|c| Some(c.keys())).collect(); let val_srcs: Vec> = chunks.iter().map(|c| Some(c.vals())).collect(); let (mut tags, mut offs) = (Vec::new(), Vec::new()); let (mut khs, mut times, mut diffs) = (Vec::new(), Vec::new(), Vec::new()); + let total: usize = chunks.iter().map(|c| c.diffs().len()).sum(); + let seek = changed.len().saturating_mul(SEEK_ADVANTAGE) < total; let needles = CValue::u64(changed.to_vec()); - // Chunks are id-ordered and `changed` ascends, so emission order is the merged order. + // Chunks are id-ordered and `changed` ascends, so either branch emits in merged order. for (ci, ch) in chunks.iter().enumerate() { if ch.diffs().is_empty() { continue; } - let (lo, hi) = find_ranges(&needles, key_lane(ch.keys())); - for (j, (&l, &h)) in lo.iter().zip(hi.iter()).enumerate() { - for i in l..h { - tags.push(ci); - offs.push(i); - khs.push(changed[j]); - times.push(ch.times().get(i)); - diffs.push(ch.diffs()[i]); + let lane = key_lane(ch.keys()); + if seek { + let (lo, hi) = find_ranges(&needles, lane); + for (j, (&l, &h)) in lo.iter().zip(hi.iter()).enumerate() { + for i in l..h { + tags.push(ci); + offs.push(i); + khs.push(changed[j]); + times.push(ch.times().get(i)); + diffs.push(ch.diffs()[i]); + } + } + } else { + // Borrowed, never materialized: the identifier lane is a `u64` leaf whatever the key. + let kh = corgi::arrange::leaf_slice(lane).expect("the identifier lane is a u64 leaf"); + for i in 0..kh.len() { + if changed.binary_search(&kh[i]).is_ok() { + tags.push(ci); + offs.push(i); + khs.push(kh[i]); + times.push(ch.times().get(i)); + diffs.push(ch.diffs()[i]); + } } } } From 41a5f4a38de31fa9fc68dc33fab46f7b71b2985b Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 14 Aug 2026 13:24:15 -0400 Subject: [PATCH 03/14] The join seeks the identifier lane, so every key shape blocks and resumes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `advance_leaf` was the only regime that blocks: it seeks `from`, ends at a key boundary, and writes back where to resume. It was reachable only by keys that were a single u64 lane. `advance_lanes` and `advance_structured` opened with `*from = None` and staged the whole intersection in one call, which is the join half of the windowing problem. Now that every arrangement leads with its identifier and is sorted by it, the leaf walk applies to every key shape — it seeks and resumes on that lane. The dispatch tries it first and the two whole-key walks become the fallback. Three sites moved off whole-key-shaped needles onto the lane: the block seek, the batched probe, and `LeafView`'s pull. `needle_like` goes with them — building a needle in the key's own shape was only ever a way to search a column that had no integer to search by. The one exposure is a hash collision, which would cross-product two distinct keys sharing an identifier. `one_key` checks per matched key, not per row: runs are contiguous and sub-sorted by the real key, so a run holds one key exactly when its first and last rows agree, and the same pass over both sides' runs confirms they matched on the key rather than the hash. On failure the call is redone by the whole-key walks, which compare the real key and cannot be fooled. That path cannot be provoked with a real hash, so it is mutation-tested both ways: forcing the fallback for every hashed key leaves all 53 tests passing (it computes the same answer), and panicking in the fallback also leaves them passing (it is never taken, so the leaf walk really is handling these keys). Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01YH7iq9JoXmf7ATpq1gZaQS --- interactive/src/corgi/chunk.rs | 17 ++++++-- interactive/src/corgi/join.rs | 80 ++++++++++++++++++++++++---------- 2 files changed, 70 insertions(+), 27 deletions(-) diff --git a/interactive/src/corgi/chunk.rs b/interactive/src/corgi/chunk.rs index b9c81e454..403b85fd5 100644 --- a/interactive/src/corgi/chunk.rs +++ b/interactive/src/corgi/chunk.rs @@ -508,15 +508,26 @@ pub fn key_ids(keys: &CValue) -> 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 u64 leaf, so `find_ranges` over it -/// takes corgi's u64 fast path whatever the underlying key shape. +/// 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) if corgi::arrange::leaf_slice(keys).is_none() => &cols[0], + 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 { diff --git a/interactive/src/corgi/join.rs b/interactive/src/corgi/join.rs index 2d11d118e..8a28d3f11 100644 --- a/interactive/src/corgi/join.rs +++ b/interactive/src/corgi/join.rs @@ -40,7 +40,7 @@ use differential_dataflow::trace::chunk::{Chunk, ChunkBatch}; use corgi::arrange::{compare_at, find_ranges, gather, gather_lanes}; use corgi::{shape_of_value, Shape, Value as CValue}; -use crate::corgi::chunk::{recover_key, CorgiChunk}; +use crate::corgi::chunk::{key_is_hashed, key_lane, recover_key, CorgiChunk}; use crate::corgi::col_times::ColTime; use crate::corgi::container::CorgiContainer; use crate::corgi::logic::compile_join_projection; @@ -87,8 +87,22 @@ impl ProxyJoinBackend, CBatch> for CorgiJoinBackend *from = None; return; } + // Every arrangement key leads with its identifier and is sorted by it, so the leaf walk + // applies to every key shape: it seeks and resumes `from` on that lane. It is the only + // regime that blocks — the other two stage the whole intersection in one call. + // + // Its one exposure is a hash collision, which would cross-product two distinct keys that + // share an identifier. `advance_leaf` checks for that per matched key and reports it rather + // than staging it, and the call is redone by the whole-key walks, which cannot be fooled + // (they compare the real key). Astronomically rare, and correct by reuse when it happens. + let entry = *from; + if advance_leaf(&chunks0, &chunks1, &instance.lower, from, bridge0, bridge1) { + return; + } + bridge0.clear(); + bridge1.clear(); + *from = entry; match (leaf_key_lanes(&chunks0), leaf_key_lanes(&chunks1)) { - (Some(1), Some(1)) => advance_leaf(&chunks0, &chunks1, &instance.lower, from, bridge0, bridge1), (Some(_), Some(_)) => advance_lanes(&chunks0, &chunks1, &instance.lower, from, bridge0, bridge1), _ => advance_structured(&chunks0, &chunks1, &instance.lower, from, bridge0, bridge1), } @@ -188,16 +202,6 @@ fn leaf_valued(chunks: &[&CorgiChunk]) -> bool { chunks.iter().filter(|c| c.len() > 0).all(|c| leaf_lanes(c.vals()).is_some()) } -/// A needle column of `keys`, shaped like `col` (a single-leaf-lane key column). -fn needle_like(col: &CValue, keys: &[u64]) -> CValue { - match col { - CValue::Prim(_) => CValue::u64(keys.to_vec()), - CValue::Prod(fields) => CValue::Prod(fields.iter().map(|f| needle_like(f, keys)).collect()), - CValue::Unit(_) => CValue::Unit(keys.len()), - _ => unreachable!("needle_like: single-leaf key columns only"), - } -} - /// Pull rows `idx` of the column's leaf lanes as `u64` buffers. fn pull_lanes(col: &CValue, idx: &[usize]) -> Vec> { leaf_lanes(col).expect("pull_lanes: leaf-laned column") @@ -431,8 +435,8 @@ impl<'a, T: ColTime> LeafView<'a, T> { let e = (s + more).min(self.chunk.len()); if s == e { return; } let idx: Vec = (s..e).collect(); - let key_lane = leaf_lanes(self.chunk.keys()).expect("single-lane keys")[0]; - self.keys.extend(gather(key_lane, &idx).into_u64("corgi join key pull")); + let lane = key_lane(self.chunk.keys()); + self.keys.extend(gather(lane, &idx).into_u64("corgi join key pull")); if let Some(vals) = self.vals.as_mut() { let pulled = pull_lanes(self.chunk.vals(), &idx); if vals.is_empty() { @@ -480,7 +484,7 @@ struct Probe<'a, T: ColTime> { impl<'a, T: ColTime> Probe<'a, T> { fn new(chunk: &'a CorgiChunk, cid: usize, needles: &CValue, leaf_vals: bool) -> Self { - let (lo, hi) = find_ranges(needles, chunk.keys()); + let (lo, hi) = find_ranges(needles, key_lane(chunk.keys())); let mut off = Vec::with_capacity(lo.len() + 1); let mut idx: Vec = Vec::new(); off.push(0); @@ -509,6 +513,23 @@ impl<'a, T: ColTime> Probe<'a, T> { } } +/// Whether every row covered by `refs` carries the same KEY, not merely the same identifier. +/// +/// The identifier is injective for a primitive-integer key, but a hashed key can in principle put +/// two distinct keys under one identifier, and the leaf path would then cross-product them. Runs +/// are contiguous and sub-sorted by the real key ([`present_key`](crate::corgi::chunk::present_key) +/// keeps the key in the column, after the hash), so a run holds one key exactly when its first and +/// last rows agree — a handful of `compare_at` per matched key, never per row. `refs` spans both +/// sides, so one pass also confirms the two sides matched on the key and not just the hash. +fn one_key(a: &[RunRef<'_, T>], b: &[RunRef<'_, T>]) -> bool { + let Some(first) = a.first().or_else(|| b.first()) else { return true }; + let (rk, ri) = (first.chunk.keys(), first.s); + a.iter().chain(b).all(|r| { + compare_at(r.chunk.keys(), r.s, rk, ri) == Ordering::Equal + && compare_at(r.chunk.keys(), r.e - 1, rk, ri) == Ordering::Equal + }) +} + /// Blockwise `advance` for leaf-keyed inputs: group token = the key's own `u64` (chunk order /// IS `u64` order), so blocks resume by seeking `from` and end at key boundaries. /// @@ -524,10 +545,12 @@ fn advance_leaf( from: &mut Option, bridge0: &mut ProxyBridge, bridge1: &mut ProxyBridge, -) { +) -> bool { let start = from.expect("advance called on an exhausted unit"); + let hashed = chunks0.iter().chain(chunks1).find(|c| c.len() > 0).is_some_and(|c| key_is_hashed(c.keys())); let seek = |chunks: &[&CorgiChunk]| -> Vec { - chunks.iter().map(|c| if c.len() == 0 { 0 } else { find_ranges(&needle_like(c.keys(), &[start]), c.keys()).0[0] }).collect() + let needle = CValue::u64(vec![start]); + chunks.iter().map(|c| if c.len() == 0 { 0 } else { find_ranges(&needle, key_lane(c.keys())).0[0] }).collect() }; let start0 = seek(chunks0); let start1 = seek(chunks1); @@ -546,9 +569,9 @@ fn advance_leaf( let drive0 = r0 <= r1; let (dviews, pchunks) = if drive0 { (views(chunks0, &start0), chunks1) } else { (views(chunks1, &start1), chunks0) }; let (bd, bp) = if drive0 { (bridge0, bridge1) } else { (bridge1, bridge0) }; - leaf_probe(dviews, pchunks, lower, start, from, bd, bp); + leaf_probe(dviews, pchunks, lower, start, from, bd, bp, hashed) } else { - leaf_merge(views(chunks0, &start0), views(chunks1, &start1), lower, start, from, bridge0, bridge1); + leaf_merge(views(chunks0, &start0), views(chunks1, &start1), lower, start, from, bridge0, bridge1, hashed) } } @@ -581,7 +604,8 @@ fn leaf_probe<'a, T: ColTime>( from: &mut Option, bridge_d: &mut ProxyBridge, bridge_p: &mut ProxyBridge, -) { + hashed: bool, +) -> bool { let h = pull_horizon(&mut dviews, &mut [], start); // Merge the driver block's keys (strictly below the horizon) into the distinct key @@ -602,14 +626,14 @@ fn leaf_probe<'a, T: ColTime>( if keyset.is_empty() { // Nothing below the horizon: the driver is spent (or the block was empty). *from = h; - return; + return true; } // One batched probe of every probee chunk at the driver's keys. let pvleaf = leaf_valued(pchunks); let probes: Vec> = pchunks.iter().enumerate() .filter(|(_, c)| c.len() > 0) - .map(|(cid, c)| Probe::new(c, cid, &needle_like(c.keys(), &keyset), pvleaf)) + .map(|(cid, c)| Probe::new(c, cid, &CValue::u64(keyset.clone()), pvleaf)) .collect(); // Walk the keys in order, staging both sides and emitting the survivors. @@ -628,6 +652,9 @@ fn leaf_probe<'a, T: ColTime>( if refs.len() == dref_count { continue; // key absent from the probee: no matches } + if hashed && !one_key(&refs, &[]) { + return false; + } let (drefs, prefs) = refs.split_at(dref_count); sd.stage_runs(drefs, lower); sp.stage_runs(prefs, lower); @@ -640,6 +667,7 @@ fn leaf_probe<'a, T: ColTime>( // The block ran to the horizon; resume there (`None` = the driver is exhausted, and // with it the intersection). *from = h; + true } /// Comparable-sides regime: both sides pulled and merged symmetrically on the `u64` @@ -652,7 +680,8 @@ fn leaf_merge<'a, T: ColTime>( from: &mut Option, bridge0: &mut ProxyBridge, bridge1: &mut ProxyBridge, -) { + hashed: bool, +) -> bool { let h = pull_horizon(&mut views0, &mut views1, start); let (mut s0, mut s1) = (SideScratch::new(), SideScratch::new()); let (mut refs0, mut refs1): (Vec<(usize, usize, usize)>, Vec<(usize, usize, usize)>) = (Vec::new(), Vec::new()); @@ -660,7 +689,7 @@ fn leaf_merge<'a, T: ColTime>( let k = views0.iter().chain(&views1).filter_map(LeafView::cur_key).min(); let Some(k) = k.filter(|k| h.map_or(true, |h| *k < h)) else { *from = h; - return; + return true; }; refs0.clear(); refs0.extend(views0.iter_mut().enumerate().filter_map(|(vi, v)| v.take_run(k).map(|(s, e)| (vi, s, e)))); @@ -671,6 +700,9 @@ fn leaf_merge<'a, T: ColTime>( } let r0: Vec> = refs0.iter().map(|&(vi, s, e)| views0[vi].run_ref(s, e)).collect(); let r1: Vec> = refs1.iter().map(|&(vi, s, e)| views1[vi].run_ref(s, e)).collect(); + if hashed && !one_key(&r0, &r1) { + return false; + } s0.stage_runs(&r0, lower); s1.stage_runs(&r1, lower); if s0.entries.is_empty() || s1.entries.is_empty() { From 0d6607c3e14f6c0b0ffb1ef9a06d111fd308825a Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 14 Aug 2026 14:42:10 -0400 Subject: [PATCH 04/14] Decide the block's extent before reading it, then read exactly that MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The block bound used to be discovered from what had already been read: pull a fixed number of rows per chunk, then take the least last-pulled identifier, since anything at or beyond it might have more rows outside somebody's buffer. Whatever sat past that bound was discarded and read again by the next block. The waste is structural rather than incidental — a chunk whose keys are sparse spans a wide identifier range with its rows and a dense one spans a narrow range, and the bound is the minimum, so the sparse chunks re-read most of what they touched, every block. A totally ordered identifier makes the other order possible. `block_horizon` reads ONE value per chunk — the identifier a budget's worth of rows in, bumped past its own run so a key is never split — and takes the least. `block_ends` then binary-searches that bound in each chunk. Each chunk is read once, over exactly the rows the block needs, and nothing is read twice. This is why it could not have been written this way before: choosing a bound before reading requires a single ordered value you can both index by position and seek by value, and until the arrangement led with its identifier, a non-scalar key had neither. Falling out of it: * `pull_horizon` and its extend loop are gone. The degenerate case they existed for — a run longer than the budget denying all progress — cannot arise, because the bound is taken past the run at the budget, so the chunk that set it always contributes at least a budget of rows. Progress is by construction, not by retry. * `LeafView` borrows its identifiers straight from the chunk's lane instead of gathering them into a `Vec`. The block reads keys; it no longer copies them. * Seeking the resume point is a `partition_point` on that slice rather than a column probe. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01YH7iq9JoXmf7ATpq1gZaQS --- interactive/src/corgi/join.rs | 135 ++++++++++++++++++---------------- 1 file changed, 70 insertions(+), 65 deletions(-) diff --git a/interactive/src/corgi/join.rs b/interactive/src/corgi/join.rs index 8a28d3f11..6416c95bf 100644 --- a/interactive/src/corgi/join.rs +++ b/interactive/src/corgi/join.rs @@ -399,54 +399,71 @@ fn min_key<'a, T: ColTime>( best } -/// Pulled-buffer view over one leaf-keyed chunk of the DRIVER side: the block's key rows -/// extracted once per `advance` call as `u64` buffers, so the key merge runs on slices. +/// The identifier column of a chunk's keys, as a sorted `u64` slice. Every arrangement key +/// leads with its identifier ([`present_key`](crate::corgi::chunk::present_key)), so this is +/// always available and always ordered — no gather, no copy. +fn ident<'a, T: ColTime>(chunk: &'a CorgiChunk) -> &'a [u64] { + corgi::arrange::leaf_slice(key_lane(chunk.keys())).expect("the identifier lane is a u64 leaf") +} + +/// The block's exclusive identifier bound, chosen so that no chunk contributes more than +/// `budget` rows: for each chunk with more than `budget` rows left, the identifier `budget` +/// rows in (bumped past its own run, so a key is never split); the least of those. `None` +/// when no chunk is over budget — the block runs to the end and the walk is exhausted. +/// +/// This is the whole point of a totally ordered identifier: the block's extent is decided by +/// reading ONE value per chunk, before anything is read in bulk, so each chunk can then be +/// read exactly once over exactly the rows the block needs. Deciding it the other way round — +/// read a fixed number of rows, then discover how far the block can reach — leaves whatever +/// was read past the bound to be discarded and read again next block, and the further apart +/// the chunks' key densities are, the more that is. +fn block_horizon(chunks: &[&CorgiChunk], starts: &[usize], budget: usize) -> Option { + let mut horizon: Option = None; + for (c, &s) in chunks.iter().zip(starts) { + let lane = ident(c); + if lane.len() - s <= budget { + continue; + } + // Past the whole run of the identifier at the budget, so the block ends at a key + // boundary. `None` on overflow means nothing can exceed it: run to the end. + let Some(bound) = lane[s + budget].checked_add(1) else { continue }; + horizon = Some(horizon.map_or(bound, |h: u64| h.min(bound))); + } + horizon +} + +/// Where each chunk's block ends: the first row at or past `horizon`, or its end. +fn block_ends(chunks: &[&CorgiChunk], horizon: Option) -> Vec { + chunks.iter().map(|c| match horizon { + Some(h) => ident(c).partition_point(|&x| x < h), + None => c.len(), + }).collect() +} + +/// View over one leaf-keyed chunk's rows for THIS block, `[base, end)`. The identifiers are +/// borrowed from the chunk's own lane — the block reads them, it does not copy them — and the +/// vals are gathered once, over exactly those rows. struct LeafView<'a, T: ColTime> { chunk: &'a CorgiChunk, cid: usize, /// Absolute row of `keys[0]`. base: usize, - keys: Vec, + keys: &'a [u64], /// Leaf-laned vals over the same rows; `None` when vals are structured. vals: Option>>, /// Cursor within `keys`. cur: usize, } -/// Rows pulled per driver chunk per `advance` call. Unprocessed pulled rows are re-pulled -/// by the next call (bounded per-call waste, in exchange for pull-once simplicity). +/// Rows per chunk per block. Bounds the block's working set; nothing outside it is read. const PULL: usize = 1 << 14; impl<'a, T: ColTime> LeafView<'a, T> { - fn new(chunk: &'a CorgiChunk, cid: usize, start: usize, leaf_vals: bool) -> Self { - let mut view = LeafView { chunk, cid, base: start, keys: Vec::new(), vals: if leaf_vals { Some(Vec::new()) } else { None }, cur: 0 }; - view.extend_pull(PULL); - view - } - fn abs_end(&self) -> usize { - self.base + self.keys.len() - } - fn fully_pulled(&self) -> bool { - self.abs_end() == self.chunk.len() - } - /// Append up to `more` further rows to the buffers. - fn extend_pull(&mut self, more: usize) { - let s = self.abs_end(); - let e = (s + more).min(self.chunk.len()); - if s == e { return; } - let idx: Vec = (s..e).collect(); - let lane = key_lane(self.chunk.keys()); - self.keys.extend(gather(lane, &idx).into_u64("corgi join key pull")); - if let Some(vals) = self.vals.as_mut() { - let pulled = pull_lanes(self.chunk.vals(), &idx); - if vals.is_empty() { - *vals = pulled; - } else { - for (lane, more) in vals.iter_mut().zip(pulled) { lane.extend(more); } - } - } + fn new(chunk: &'a CorgiChunk, cid: usize, start: usize, end: usize, leaf_vals: bool) -> Self { + let vals = (leaf_vals && end > start).then(|| pull_lanes(chunk.vals(), &(start..end).collect::>())); + LeafView { chunk, cid, base: start, keys: &ident(chunk)[start..end], vals, cur: 0 } } - /// The key under the cursor, if any remains in the buffer. + /// The key under the cursor, if any remains in this block. fn cur_key(&self) -> Option { self.keys.get(self.cur).copied() } @@ -548,9 +565,10 @@ fn advance_leaf( ) -> bool { let start = from.expect("advance called on an exhausted unit"); let hashed = chunks0.iter().chain(chunks1).find(|c| c.len() > 0).is_some_and(|c| key_is_hashed(c.keys())); + // Resume: the first row of each chunk at or past `start`, by binary search on its identifier + // lane. The lane is a sorted `u64` slice, so this is a slice operation, not a column probe. let seek = |chunks: &[&CorgiChunk]| -> Vec { - let needle = CValue::u64(vec![start]); - chunks.iter().map(|c| if c.len() == 0 { 0 } else { find_ranges(&needle, key_lane(c.keys())).0[0] }).collect() + chunks.iter().map(|c| ident(c).partition_point(|&x| x < start)).collect() }; let start0 = seek(chunks0); let start1 = seek(chunks1); @@ -558,40 +576,30 @@ fn advance_leaf( chunks.iter().zip(starts).map(|(c, &s)| c.len() - s).sum() }; let (r0, r1) = (remaining(chunks0, &start0), remaining(chunks1, &start1)); - fn views<'a, T: ColTime>(chunks: &[&'a CorgiChunk], starts: &[usize]) -> Vec> { + fn views<'a, T: ColTime>(chunks: &[&'a CorgiChunk], starts: &[usize], ends: &[usize]) -> Vec> { let leaf_vals = leaf_valued(chunks); chunks.iter().enumerate() - .filter(|(_, c)| c.len() > 0) - .map(|(cid, c)| LeafView::new(c, cid, starts[cid], leaf_vals)) + .filter(|(cid, _)| ends[*cid] > starts[*cid]) + .map(|(cid, c)| LeafView::new(c, cid, starts[cid], ends[cid], leaf_vals)) .collect() } if r0.max(r1) >= 2 * r0.min(r1) { + // Lopsided: only the driver is read in bulk, so only the driver bounds the block. let drive0 = r0 <= r1; - let (dviews, pchunks) = if drive0 { (views(chunks0, &start0), chunks1) } else { (views(chunks1, &start1), chunks0) }; + let (dchunks, dstarts, pchunks) = if drive0 { (chunks0, &start0, chunks1) } else { (chunks1, &start1, chunks0) }; + let h = block_horizon(dchunks, dstarts, PULL); + let dviews = views(dchunks, dstarts, &block_ends(dchunks, h)); let (bd, bp) = if drive0 { (bridge0, bridge1) } else { (bridge1, bridge0) }; - leaf_probe(dviews, pchunks, lower, start, from, bd, bp, hashed) + leaf_probe(dviews, pchunks, lower, h, from, bd, bp, hashed) } else { - leaf_merge(views(chunks0, &start0), views(chunks1, &start1), lower, start, from, bridge0, bridge1, hashed) - } -} - -/// The pull horizon: keys at or beyond the least last-pulled key of a partially-pulled -/// view may have rows outside its buffer, so the block stops there. Extended pulls push -/// it out when it would deny all progress (a run longer than the pull). -fn pull_horizon<'a, T: ColTime>(views0: &mut [LeafView<'a, T>], views1: &mut [LeafView<'a, T>], start: u64) -> Option { - let horizon = |a: &[LeafView<'a, T>], b: &[LeafView<'a, T>]| -> Option { - a.iter().chain(b).filter(|v| !v.fully_pulled()).map(|v| *v.keys.last().unwrap()).min() - }; - let mut h = horizon(views0, views1); - while h == Some(start) { - for v in views0.iter_mut().chain(views1.iter_mut()) { - if !v.fully_pulled() && *v.keys.last().unwrap() == start { - v.extend_pull(v.keys.len()); - } - } - h = horizon(views0, views1); + // Symmetric: both sides are read, so both bound the block. + let h = block_horizon(chunks0, &start0, PULL).into_iter() + .chain(block_horizon(chunks1, &start1, PULL)) + .min(); + let views0 = views(chunks0, &start0, &block_ends(chunks0, h)); + let views1 = views(chunks1, &start1, &block_ends(chunks1, h)); + leaf_merge(views0, views1, lower, h, from, bridge0, bridge1, hashed) } - h } /// Lopsided regime: the driver views are walked; the probee is presented only at the @@ -600,14 +608,12 @@ fn leaf_probe<'a, T: ColTime>( mut dviews: Vec>, pchunks: &[&CorgiChunk], lower: &T, - start: u64, + h: Option, from: &mut Option, bridge_d: &mut ProxyBridge, bridge_p: &mut ProxyBridge, hashed: bool, ) -> bool { - let h = pull_horizon(&mut dviews, &mut [], start); - // Merge the driver block's keys (strictly below the horizon) into the distinct key // list and each key's runs. let mut keyset: Vec = Vec::new(); @@ -676,13 +682,12 @@ fn leaf_merge<'a, T: ColTime>( mut views0: Vec>, mut views1: Vec>, lower: &T, - start: u64, + h: Option, from: &mut Option, bridge0: &mut ProxyBridge, bridge1: &mut ProxyBridge, hashed: bool, ) -> bool { - let h = pull_horizon(&mut views0, &mut views1, start); let (mut s0, mut s1) = (SideScratch::new(), SideScratch::new()); let (mut refs0, mut refs1): (Vec<(usize, usize, usize)>, Vec<(usize, usize, usize)>) = (Vec::new(), Vec::new()); loop { From 03710d438d6341b032d1b7df74af5c4dc17c63fc Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sat, 15 Aug 2026 15:34:15 -0400 Subject: [PATCH 05/14] Reduce proxy pending-state rebuild overhead --- .../src/operators/int_proxy/reduce.rs | 142 +++++++++++++++--- 1 file changed, 118 insertions(+), 24 deletions(-) diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs index 1e46d2c4f..52108087d 100644 --- a/differential-dataflow/src/operators/int_proxy/reduce.rs +++ b/differential-dataflow/src/operators/int_proxy/reduce.rs @@ -3,7 +3,10 @@ //! A conventional differential reduce against `(u64, u64)`, where the backend supplies the //! implementation of the interpretation of the integers. +use std::cell::RefCell; use std::collections::BTreeMap; +use std::sync::OnceLock; +use std::time::{Duration, Instant}; use timely::PartialOrder; use timely::progress::{Antichain, Timestamp}; @@ -16,6 +19,72 @@ use super::ProxyBridge; use crate::operators::reduce::{sort_dedup, ReduceTactic}; use crate::operators::ValueHistory; +/// One opt-in phase measurement from the generic proxy-reduce driver. +#[derive(Clone, Debug)] +pub struct ProxyReducePhaseStat { + /// Stable phase label. + pub phase: &'static str, + /// Number of timed invocations. + pub calls: u64, + /// Phase-specific row or slot count, useful only within the named phase. + pub work: u64, + /// Total wall-clock time spent in the phase. + pub elapsed: Duration, +} + +#[derive(Default)] +struct ProxyReducePhaseAccum { + calls: u64, + work: u64, + elapsed: Duration, +} + +thread_local! { + static PROXY_REDUCE_PROFILE: RefCell> = RefCell::new(BTreeMap::new()); +} + +fn proxy_profile_enabled() -> bool { + static ENABLED: OnceLock = OnceLock::new(); + *ENABLED.get_or_init(|| std::env::var("PROXY_REDUCE_PROFILE").is_ok_and(|x| x != "0")) +} + +#[inline] +fn phase_start() -> Option { + proxy_profile_enabled().then(Instant::now) +} + +#[inline] +fn phase_finish(phase: &'static str, start: Option, work: usize) { + if let Some(start) = start { + PROXY_REDUCE_PROFILE.with(|profile| { + let mut profile = profile.borrow_mut(); + let stat = profile.entry(phase).or_default(); + stat.calls += 1; + stat.work += work as u64; + stat.elapsed += start.elapsed(); + }); + } +} + +/// Clear proxy-reduce phase counters for the current worker thread. +pub fn reset_phase_profile() { + PROXY_REDUCE_PROFILE.with(|profile| profile.borrow_mut().clear()); +} + +/// Snapshot current-thread proxy-reduce phase counters, slowest phase first. +pub fn phase_profile() -> Vec { + PROXY_REDUCE_PROFILE.with(|profile| { + let mut stats: Vec<_> = profile.borrow().iter().map(|(&phase, stat)| ProxyReducePhaseStat { + phase, + calls: stat.calls, + work: stat.work, + elapsed: stat.elapsed, + }).collect(); + stats.sort_unstable_by_key(|stat| std::cmp::Reverse(stat.elapsed)); + stats + }) +} + /// A unit of proxied reduce work, presented to the backend. pub struct ReduceInstance<'a, B1: BatchReader, B2: BatchReader