From 39c3198112195f7583c1dd1387c70b2c81e36469 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 9 Oct 2026 21:36:59 -0400 Subject: [PATCH 1/5] Bump corgi to WIP 5b2b272: one Int Corgi now has one integer type, `Int`, exact within i64, and DDIR's `Int` is it. The sign-bit swizzles (`ToSigned`, `signed_order_view`) are gone, and division and ordered compares are corgi's own. Corgi orders integers by signed value, so arrangement identifiers are `i64` throughout join, reduce, and search. The join walk starts at `i64::MIN`, not 0, so negative keys are not skipped. Corgi chooses how to hold an integer column, and a mask or tag may be held as bytes. `present_key` widens such a key to `i64`, so it is its own identifier rather than hashed, and arranged values are widened the same way so that every chunk's lanes can be borrowed. `&&` and `or` read each operand as 0/1 (nonzero is true) before `min`/`max`. `hash` shifts with `shr_b64`, and a bound that is not positive reduces to the identity. F64 ops reach corgi's `Float` through host kernels on the newtype payload, until DDIR has a Float of its own. Co-Authored-By: Claude Opus 5.5 --- interactive/Cargo.toml | 2 +- interactive/src/backend/corgi.rs | 15 +- interactive/src/corgi/bytes.rs | 4 +- interactive/src/corgi/chunk.rs | 130 ++++++--- interactive/src/corgi/join.rs | 179 +++++++------ interactive/src/corgi/logic.rs | 357 ++++++++++++------------- interactive/src/corgi/reduce.rs | 119 ++++----- interactive/src/corgi/search.rs | 20 +- interactive/src/explain/mod.rs | 6 +- interactive/src/ir.rs | 2 +- interactive/src/parse/pipe.rs | 2 +- interactive/src/server.rs | 6 +- interactive/src/shapes.rs | 18 +- interactive/tests/corgi_backend.rs | 12 +- interactive/tests/corgi_chunk_merge.rs | 4 +- interactive/tests/explain.rs | 2 +- interactive/tests/server_shapes.rs | 2 +- 17 files changed, 457 insertions(+), 423 deletions(-) diff --git a/interactive/Cargo.toml b/interactive/Cargo.toml index 37e8347c9..2e49dcca9 100644 --- a/interactive/Cargo.toml +++ b/interactive/Cargo.toml @@ -14,7 +14,7 @@ workspace = true [dependencies] columnar = { workspace = true } # The columnar kernels for the interpreted backend, pinned by git rev. -corgi = { git = "https://github.com/frankmcsherry/WIP", rev = "4b1d23bfc4ffc8a4ee51e2230216426a17f880a4", features = ["serde"] } +corgi = { git = "https://github.com/frankmcsherry/WIP", rev = "5b2b2721515790bc291e827ff5b52f561bf21002", features = ["serde"] } differential-dataflow = { workspace = true } serde = { version = "1.0", features = ["derive"] } smallvec = "1.15.1" diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs index 61180c136..b80eea9b7 100644 --- a/interactive/src/backend/corgi.rs +++ b/interactive/src/backend/corgi.rs @@ -109,7 +109,8 @@ fn apply_ops(mut c: CC, ops: &[LinearOp], level: usize, plans: &mut [Plan]) -> C } LinearOp::Filter(_) => { let g = plan.graph(op, kshape, vshape); - let mask = corgi::eval_graph(g, CValue::Prod(vec![c.keys.clone(), c.vals.clone()])).into_u64("filter mask").unwrap(); + let mask = corgi::eval_graph(g, CValue::Prod(vec![c.keys.clone(), c.vals.clone()])); + let mask = mask.as_i64("filter mask").unwrap(); let keep: Vec = (0..mask.len()).filter(|&i| mask[i] != 0).collect(); let keys = gather(&c.keys, &keep); let vals = gather(&c.vals, &keep); @@ -126,15 +127,15 @@ fn apply_ops(mut c: CC, ops: &[LinearOp], level: usize, plans: &mut [Plan]) -> C LinearOp::EnterAt(_) => { let g = plan.graph(op, kshape, vshape); // The key and val columns are IDENTITY here — only times change. Evaluate the - // delay field to a `U64` column and join it into each time in place. Joining + // delay field to an `Int` column and join it into each time in place. Joining // `Product(0, PointStamp([0,..,0, delay]))` is, coordinate-wise, `max` at index // `level-1` and identity everywhere else (u64's minimum is 0), so the delta // never has to be built. The epoch is lane 0, so PointStamp index `level-1` is // lane `level`, and the join is one lane-wise max. let raw = corgi::eval_graph(g, CValue::Prod(vec![c.keys.clone(), c.vals.clone()])) - .into_u64("enter_at delay") + .into_i64("enter_at delay") .unwrap(); - let delays: Vec = raw.iter().map(|r| 256 * (64 - r.leading_zeros() as u64)).collect(); + let delays: Vec = raw.iter().map(|&r| 256 * (64 - (r as u64).leading_zeros() as u64)).collect(); c.times.lane_max(level.saturating_sub(1) + 1, &delays); c } @@ -150,7 +151,7 @@ fn apply_ops(mut c: CC, ops: &[LinearOp], level: usize, plans: &mut [Plan]) -> C // PointStamp index `level-1` is lane `level`; at the root there is no iteration // coordinate and the value is zero. let iters: Vec = if level == 0 { vec![0; c.times.len()] } else { c.times.lane(level) }; - let lane = CValue::u64(iters); + let lane = CValue::i64(iters.into_iter().map(|i| i as i64).collect()); let vals = match c.vals { CValue::Prod(mut fields) => { fields.push(lane); CValue::Prod(fields) } CValue::Unit(_) => CValue::Prod(vec![lane]), @@ -173,13 +174,13 @@ fn apply_ops(mut c: CC, ops: &[LinearOp], level: usize, plans: &mut [Plan]) -> C for (row, end) in ends.into_iter().enumerate() { for p in 0..(end - start) { reps.push(row); - pos.push(p as u64); + pos.push(p as i64); } start = end; } CorgiContainer { keys: gather(&c.keys, &reps), - vals: CValue::Prod(vec![CValue::u64(pos), elems]), + vals: CValue::Prod(vec![CValue::i64(pos), elems]), times: c.times.gather(&reps), diffs: reps.iter().map(|&r| c.diffs[r]).collect(), } diff --git a/interactive/src/corgi/bytes.rs b/interactive/src/corgi/bytes.rs index 24b54f5f0..8c1abf8b1 100644 --- a/interactive/src/corgi/bytes.rs +++ b/interactive/src/corgi/bytes.rs @@ -197,8 +197,8 @@ mod test { /// The declared shapes of each family — what a program's schema would pin (a variant carries /// only its tag, so a family with sums cannot be pinned from a row). fn shapes_of(name: &str) -> (corgi::Shape, corgi::Shape) { - use corgi::Shape::{List, Prim, Prod, Sum}; - let u = || Prim(64); + use corgi::Shape::{Int, List, Prod, Sum}; + let u = || Int; let pair = || Prod(vec![u(), u()]); match name { "scalars" => (u(), u()), diff --git a/interactive/src/corgi/chunk.rs b/interactive/src/corgi/chunk.rs index 3b4ec7fba..98b3a15e5 100644 --- a/interactive/src/corgi/chunk.rs +++ b/interactive/src/corgi/chunk.rs @@ -455,11 +455,14 @@ where /// One sorted+consolidated chunk from columns already in corgi form (the column-native arrange /// ingest — no transcode). pub fn from_columns(keys: CValue, vals: CValue, times: ColTimes, diffs: Vec) -> Self { + // An integer value is held as `i64`s, as an integer key is, so that readers may borrow it + // whichever chunk of the arrangement it is in. + let vals = hold_wide(vals); let (keys, vals, times, diffs) = sort_consolidate(keys, vals, times, diffs); debug_assert!({ - let lane = corgi::arrange::leaf_slice(key_lane(&keys)); + let lane = int_lane(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"); + }, "arrangement key must lead with a sorted i64 identifier lane"); Self::from_parts(keys, vals, times, diffs) } @@ -517,7 +520,7 @@ 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 +/// A key that is already an integer (an `Int` leaf, 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 @@ -530,31 +533,67 @@ impl Default for CorgiChunker { /// (`leaf_slice` succeeds on exactly the un-prepended one) because no compound key reaches an /// arrangement un-prepended. /// +/// The identifier lane is held as `i64`s, so readers borrow it ([`ident`]). Corgi chooses how to +/// hold an integer column (a mask or a sum tag may be held as bytes); a key held narrower is +/// widened here, once. +/// /// 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; + match corgi::arrange::leaf_slice(&keys) { + Some(_) => hold_wide(keys), + None => { + let hashes = corgi::hash(&keys).into_iter().map(|h| h as i64).collect(); + CValue::Prod(vec![CValue::i64(hashes), keys]) + } } - let hashes = corgi::hash(&keys); - 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(); +/// An integer column (an `Int` leaf, or a 1-field `Prod` of one) held as `i64`s; any other +/// column as it is. Corgi chooses how to hold an integer column, and may hold one narrower. +pub fn hold_wide(col: CValue) -> CValue { + match corgi::arrange::leaf_slice(&col) { + Some(std::borrow::Cow::Owned(ids)) => rewrap(&col, ids), + _ => col, + } +} + +/// An integer column's values held as `i64`s, wrapped in `template`'s single-field products. +pub fn rewrap(template: &CValue, ids: Vec) -> CValue { + match template { + CValue::Prod(fields) => CValue::Prod(vec![rewrap(&fields[0], ids)]), + _ => CValue::i64(ids), } - 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. +/// An integer column's values, borrowed: `Some` exactly when the column is an `Int` leaf (or a +/// 1-field `Prod` of one) held as `i64`s. An integer column held narrower is `None`, and a +/// reader that can take one falls back to its structural path. +pub fn int_lane(col: &CValue) -> Option<&[i64]> { + match corgi::arrange::leaf_slice(col)? { + std::borrow::Cow::Borrowed(ids) => Some(ids), + std::borrow::Cow::Owned(_) => None, + } +} + +/// The identifier lane of a [`present_key`] column, borrowed: the key's own values when it is an +/// integer, and the prepended hash lane otherwise. `present_key` holds it as `i64`s. +pub fn ident(keys: &CValue) -> &[i64] { + int_lane(key_lane(keys)).expect("an arrangement's identifier lane is held as i64s") +} + +/// The integer identifier of each row of a [`present_key`] column: the key's own values when it is +/// an integer, and the prepended hash lane otherwise. Never re-hashes. +pub fn key_ids(keys: &CValue) -> Vec { + ident(keys).to_vec() +} + +/// The single column an arrangement is sorted by, for seeking: the key itself when it is an +/// integer, else the prepended hash lane. Always an `i64` leaf, so `find_ranges` over it takes +/// corgi's integer 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])`. +/// `Int`, the 1-field `Prod` that also counts as an integer, or a prepended `Prod([hash, key])`. /// The leading field is the identifier in all three. pub fn key_lane(keys: &CValue) -> &CValue { match keys { @@ -677,6 +716,12 @@ mod test { fn xorshift(s: &mut u64) -> u64 { *s ^= *s << 13; *s ^= *s >> 7; *s ^= *s << 17; *s } + /// An integer column of these values. + fn ints(xs: Vec) -> CValue { CValue::i64(xs.into_iter().map(|x| x as i64).collect()) } + + /// An integer column's values. + fn lane(col: &CValue) -> Vec { col.as_i64("lane").unwrap().iter().map(|&x| x as u64).collect() } + #[test] fn ingest_nested_payload_matches_scalar_consolidation() { use differential_dataflow::dynamic::pointstamp::PointStamp; @@ -686,14 +731,14 @@ mod test { let ends: Vec = (0..32).scan(0, |end, i| { *end += i % 5; Some(*end) }).collect(); let elements = *ends.last().unwrap(); let values = CValue::List(ends.into(), Box::new(CValue::Prod(vec![ - CValue::u64((0..elements).map(|i| (i % 7) as u64).collect()), - CValue::u64((0..elements).map(|i| (i % 3) as u64).collect()), + ints((0..elements).map(|i| (i % 7) as u64).collect()), + ints((0..elements).map(|i| (i % 3) as u64).collect()), ]))); for n in [0, 1, 64, 1024] { for cancel_all in [false, true] { let mut seed = 719; let rows: Vec<_> = (0..n).map(|_| xorshift(&mut seed) as usize % 32).collect(); - let keys = CValue::u64(rows.iter().map(|&r| (r % 4) as u64).collect()); + let keys = ints(rows.iter().map(|&r| (r % 4) as u64).collect()); let vals = gather(&values, &rows); let kv = CValue::Prod(vec![keys.clone(), vals.clone()]); let times: ColTimes = rows.iter().map(|&r| T::new((r % 2) as u64, @@ -755,7 +800,7 @@ mod test { source.diffs.push(diff); *expected.entry((key, time)).or_insert(0i64) += diff; } - source.keys = CValue::u64(keys); + source.keys = ints(keys); source.vals = CValue::Unit(64); let donated = source.diffs.as_ptr(); builder.push_into(&mut source); @@ -764,7 +809,7 @@ mod test { } expected.retain(|_, diff| *diff != 0); let result = builder.finish().unwrap(); - let keys = corgi::arrange::leaf_slice(result.keys()).unwrap(); + let keys = lane(result.keys()); let actual: BTreeMap<_, _> = (0..result.len_()).map(|i| ((keys[i], result.times().get(i)), result.diffs()[i])).collect(); assert_eq!(actual.len(), result.len_(), "duplicate output triples"); @@ -773,12 +818,31 @@ mod test { } } + /// A key corgi holds as bytes (here a comparison mask) is an integer key like any other: it is + /// its own identifier, held as `i64`s, and not hashed. + #[test] + fn byte_held_integer_keys_are_presented_as_i64() { + let xs = CValue::i64(vec![-3, 0, 5, 9]); + let mask = corgi::eval_graph(&{ + let mut b = corgi::Builder::::default(); + let input = b.input(); + let out = b.add(corgi::CmpOp::RelImm(corgi::Pred::Gt, corgi::Scalar::Int(0)), vec![input]); + b.finish(out) + }, xs); + assert_eq!(int_lane(&mask), None, "corgi holds a mask as bytes"); + for keys in [mask.clone(), CValue::Prod(vec![mask])] { + let presented = present_key(keys); + assert!(!key_is_hashed(&presented)); + assert_eq!(ident(&presented), &[0, 0, 1, 1]); + } + } + #[test] fn cancelled_merge_does_not_retain_input_sized_diff_storage() { let mut retained_capacity = None; for rows in [16, 256, 4096] { let make = |diffs| CorgiChunk::from_columns( - CValue::u64((0..rows).collect()), CValue::Unit(rows as usize), + ints((0..rows).collect()), CValue::Unit(rows as usize), (0..rows).map(|_| 0u64).collect(), diffs, ); let mut retractions = vec![-1i64; rows as usize]; @@ -790,7 +854,7 @@ mod test { assert!(left.is_empty() && right.is_empty()); assert_eq!(output.len(), 1); assert_eq!(output[0].diffs(), &[-1]); - assert_eq!(corgi::arrange::leaf_slice(output[0].keys()).unwrap(), &[rows - 1]); + assert_eq!(lane(output[0].keys()), &[rows - 1]); let capacity = output[0].0.diffs.capacity(); assert_eq!(*retained_capacity.get_or_insert(capacity), capacity, "one surviving row should not retain storage proportional to cancelled input"); @@ -820,8 +884,8 @@ mod test { for size in [1, 3, rows.len()] { for shared in [false, true] { let chunks: Vec<_> = rows.chunks(size).map(|rows| CorgiChunk::from_columns( - CValue::u64(rows.iter().map(|r| r.0.0).collect()), - CValue::u64(rows.iter().map(|r| r.0.1).collect()), + ints(rows.iter().map(|r| r.0.0).collect()), + ints(rows.iter().map(|r| r.0.1).collect()), rows.iter().map(|r| r.1.clone()).collect(), rows.iter().map(|r| r.2).collect(), )).collect(); let retained = if shared { chunks.clone() } else { Vec::new() }; @@ -835,8 +899,8 @@ mod test { let mut actual = BTreeMap::new(); let mut previous = None; for chunk in output { - let keys = corgi::arrange::leaf_slice(chunk.keys()).unwrap(); - let vals = corgi::arrange::leaf_slice(chunk.vals()).unwrap(); + let keys = lane(chunk.keys()); + let vals = lane(chunk.vals()); for i in 0..chunk.len_() { let key = ((keys[i], vals[i]), chunk.times().get(i)); assert!(previous.as_ref().is_none_or(|p| p < &key)); @@ -865,7 +929,7 @@ mod test { for frontier in &frontiers { for size in [1, 2, times.len()] { let chunks: Vec<_> = (0..times.len()).collect::>().chunks(size).map(|rows| CorgiChunk::from_columns( - CValue::u64(rows.iter().map(|&r| r as u64).collect()), CValue::u64(vec![0; rows.len()]), + ints(rows.iter().map(|&r| r as u64).collect()), ints(vec![0; rows.len()]), rows.iter().map(|&r| times[r].clone()).collect(), vec![1i64; rows.len()], )).collect(); let mut residual = Antichain::from_elem(time(5, &[5])); @@ -892,8 +956,8 @@ mod test { let mut m: BTreeMap<((u64, u64), u64), i64> = BTreeMap::new(); for &(kv, t, d) in rows { *m.entry((kv, t)).or_insert(0) += d; } m.retain(|_, d| *d != 0); - let keys = CValue::u64(m.keys().map(|((k, _), _)| *k).collect()); - let vals = CValue::u64(m.keys().map(|((_, v), _)| *v).collect()); + let keys = ints(m.keys().map(|((k, _), _)| *k).collect()); + let vals = ints(m.keys().map(|((_, v), _)| *v).collect()); let times: ColTimes = m.keys().map(|(_, t)| *t).collect(); let diffs = m.values().copied().collect(); CorgiChunk::from_parts(keys, vals, times, diffs) @@ -902,8 +966,8 @@ mod test { fn read_batch(b: &ChunkBatch>) -> BTreeMap<((u64, u64), u64), i64> { let mut m = BTreeMap::new(); for ch in &b.chunks { - let ks = ch.keys().clone().into_u64("k").unwrap(); - let vs = ch.vals().clone().into_u64("v").unwrap(); + let ks = lane(ch.keys()); + let vs = lane(ch.vals()); for i in 0..ch.len_() { *m.entry(((ks[i], vs[i]), ch.times().get(i))).or_insert(0) += ch.diffs()[i]; } } m.retain(|_, d| *d != 0); diff --git a/interactive/src/corgi/join.rs b/interactive/src/corgi/join.rs index 62c98166e..cb6f0ed99 100644 --- a/interactive/src/corgi/join.rs +++ b/interactive/src/corgi/join.rs @@ -37,11 +37,11 @@ use differential_dataflow::operators::int_proxy::{KeyPosition, JoinInstance, Pro use differential_dataflow::operators::int_proxy::join::JoinMatches; use differential_dataflow::trace::chunk::{Chunk, ChunkBatch}; -use corgi::arrange::{compare_at, gather_lanes, leaf_slice}; +use corgi::arrange::{compare_at, gather_lanes}; use crate::corgi::search::matching_ranges; use corgi::{shape_of_value, Graph, NumOp, Shape, Value as CValue}; -use crate::corgi::chunk::{key_is_hashed, key_lane, recover_key_ref, CorgiChunk}; +use crate::corgi::chunk::{int_lane, key_is_hashed, recover_key_ref, rewrap, CorgiChunk}; use crate::corgi::col_times::ColTime; use crate::corgi::container::CorgiContainer; use crate::corgi::logic::compile_join_projection; @@ -62,7 +62,7 @@ pub struct CorgiJoinBackend { /// Identifier tokens in the current block that cover more than one real key. `advance` writes /// this and the immediately following `cross` reads it before the next `advance`; only matches /// under these astronomically rare tokens need a real-key comparison. - colliding: Vec, + colliding: Vec, /// The projection, compiled once at the shapes of the key and both values it reads. compiled: Option<([Shape; 3], Graph)>, _t: PhantomData, @@ -94,7 +94,7 @@ fn projection<'a>(compiled: &'a mut Option<([Shape; 3], Graph)>, key: &Te } impl ProxyJoinBackend, CBatch> for CorgiJoinBackend { - type Key = u64; + type Key = i64; type V0 = u64; type V1 = u64; type R0 = Diff; @@ -105,12 +105,12 @@ impl ProxyJoinBackend, CBatch> for CorgiJoinBackend< fn advance( &mut self, instance: &JoinInstance, CBatch>, - from: &mut KeyPosition, - bridge0: &mut ProxyBridge, - bridge1: &mut ProxyBridge, + from: &mut KeyPosition, + bridge0: &mut ProxyBridge, + bridge1: &mut ProxyBridge, ) { let mut next = match *from { - KeyPosition::Start => Some(0), + KeyPosition::Start => Some(i64::MIN), KeyPosition::At(key) => Some(key), KeyPosition::End => return, }; @@ -150,7 +150,7 @@ impl ProxyJoinBackend, CBatch> for CorgiJoinBackend< fn cross( &mut self, instance: &JoinInstance, CBatch>, - matches: &mut JoinMatches, + matches: &mut JoinMatches, output: &mut Vec>, ) { let chunks0 = side_chunks(&instance.batches0); @@ -212,13 +212,13 @@ impl ProxyJoinBackend, CBatch> for CorgiJoinBackend< // the identifier for the projected key. let ids = &matches.ids[start..end]; let kc = if primitive_keys { - from_ids(chunks0[0].keys(), ids.iter().map(|x| x.0).collect()) + rewrap(chunks0[0].keys(), ids.iter().map(|x| x.0).collect()) } else { gather_lanes(&keys0, &tag0, &off0) }; let v0 = if primitive0 { - from_ids(chunks0[0].vals(), ids.iter().map(|x| x.1.0).collect()) + rewrap(chunks0[0].vals(), ids.iter().map(|x| x.1.0 as i64).collect()) } else { gather_lanes(&vals0, &tag0, &off0) }; let v1 = if primitive1 { - from_ids(chunks1[0].vals(), ids.iter().map(|x| x.1.1).collect()) + rewrap(chunks1[0].vals(), ids.iter().map(|x| x.1.1 as i64).collect()) } else { gather_lanes(&vals1, &tag1, &off1) }; let shapes = [shape_of_value(&kc), shape_of_value(&v0), shape_of_value(&v1)]; let proj = projection(&mut self.compiled, &self.key, &self.val, shapes); @@ -245,15 +245,7 @@ impl ProxyJoinBackend, CBatch> for CorgiJoinBackend< /// Coordinates remain necessary for structured keys, including collision validation. fn primitive_values(chunk: &CorgiChunk) -> bool { - !key_is_hashed(chunk.keys()) && leaf_slice(chunk.vals()).is_some() -} - -/// Rebuild a primitive column with its original singleton-product wrappers. -fn from_ids(template: &CValue, ids: Vec) -> CValue { - match template { - CValue::Prod(fields) => CValue::Prod(vec![from_ids(&fields[0], ids)]), - _ => CValue::u64(ids), - } + !key_is_hashed(chunk.keys()) && int_lane(chunk.vals()).is_some() } /// The instance's chunks on one side, in the deterministic order coordinates index @@ -270,14 +262,14 @@ fn side_chunks(batches: &[CBatch]) -> Vec<&CorgiChunk> { chunks } -/// The column flattened to `u64` leaf lanes: `Some(lanes)` when it is a (possibly nested) -/// product of 64-bit leaves — the shape DDIR tuples transcode to — so row order is the -/// lexicographic order of the lane tuples. `Sum`/`List`/narrow leaves give `None` -/// (structural compares). -fn leaf_lanes(col: &CValue) -> Option> { - fn walk<'a>(col: &'a CValue, out: &mut Vec<&'a [u64]>) -> bool { +/// The column flattened to `i64` leaf lanes: `Some(lanes)` when it is a (possibly nested) +/// product of integer leaves held as `i64`s — the shape DDIR tuples transcode to — so row order +/// is the lexicographic order of the lane tuples. `Sum`/`List`/float leaves, and integers held +/// narrower, give `None` (structural compares). +fn leaf_lanes(col: &CValue) -> Option> { + fn walk<'a>(col: &'a CValue, out: &mut Vec<&'a [i64]>) -> bool { match col { - CValue::Prim(_) => match leaf_slice(col) { Some(lane) => { out.push(lane); true }, None => false }, + CValue::Prim(_) => match int_lane(col) { Some(lane) => { out.push(lane); true }, None => false }, CValue::Prod(fields) => fields.iter().all(|f| walk(f, out)), CValue::Unit(_) => true, _ => false, @@ -301,7 +293,7 @@ struct RunRef<'a, T: ColTime> { cid: usize, s: usize, e: usize, - vals: Option<&'a [&'a [u64]]>, + vals: Option<&'a [&'a [i64]]>, } impl<'a, T: ColTime> RunRef<'a, T> { @@ -353,9 +345,9 @@ impl SideScratch { self.entries.clear(); if runs.first().is_some_and(|run| primitive_values(run.chunk)) { for run in runs { - let ids = leaf_slice(run.chunk.vals()).unwrap(); + let ids = int_lane(run.chunk.vals()).unwrap(); self.entries.extend((run.s..run.e).map(|row| { - (ids[row], run.chunk.times().get(row).join(lower), run.chunk.diffs()[row]) + (ids[row] as u64, run.chunk.times().get(row).join(lower), run.chunk.diffs()[row]) })); } differential_dataflow::consolidation::consolidate_updates(&mut self.entries); @@ -421,18 +413,18 @@ impl SideScratch { } /// Move the staged entries into `bridge` under group token `k`. - fn emit(&mut self, k: u64, bridge: &mut ProxyBridge) -> usize { + fn emit(&mut self, k: i64, bridge: &mut ProxyBridge) -> usize { let n = self.entries.len(); bridge.extend(self.entries.drain(..).map(|(coord, t, d)| ((k, coord), t, d))); n } } -/// The identifier column of a chunk's keys, as a sorted `u64` slice. Every arrangement key +/// The identifier column of a chunk's keys, as a sorted `i64` 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") +fn ident<'a, T: ColTime>(chunk: &'a CorgiChunk) -> &'a [i64] { + crate::corgi::chunk::ident(chunk.keys()) } /// The block's exclusive identifier bound: for each chunk with more than `budget` rows left, @@ -447,8 +439,8 @@ fn ident<'a, T: ColTime>(chunk: &'a CorgiChunk) -> &'a [u64] { /// 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; +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 { @@ -457,13 +449,13 @@ fn block_horizon(chunks: &[&CorgiChunk], starts: &[usize], // 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 = Some(horizon.map_or(bound, |h: i64| 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 { +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(), @@ -478,9 +470,9 @@ struct LeafView<'a, T: ColTime> { cid: usize, /// Absolute row of `keys[0]`. base: usize, - keys: &'a [u64], + keys: &'a [i64], /// The chunk's full borrowed value lanes; `None` when vals are structured. - vals: Option>, + vals: Option>, /// Cursor within `keys`. cur: usize, } @@ -494,11 +486,11 @@ impl<'a, T: ColTime> LeafView<'a, T> { LeafView { chunk, cid, base: start, keys: &ident(chunk)[start..end], vals, cur: 0 } } /// The key under the cursor, if any remains in this block. - fn cur_key(&self) -> Option { + fn cur_key(&self) -> Option { self.keys.get(self.cur).copied() } /// Take the run of `k` at the cursor, if that is the cursor's key: absolute `(start, end)`. - fn take_run(&mut self, k: u64) -> Option<(usize, usize)> { + fn take_run(&mut self, k: i64) -> Option<(usize, usize)> { if self.cur_key() != Some(k) { return None; } let s = self.cur; let e = s + self.keys[s..].partition_point(|&x| x == k); @@ -525,12 +517,12 @@ struct Probe<'a, T: ColTime> { cid: usize, lo: Vec, hi: Vec, - vals: Option>, + vals: Option>, } impl<'a, T: ColTime> Probe<'a, T> { - fn new(chunk: &'a CorgiChunk, cid: usize, needles: &[u64], leaf_vals: bool) -> Self { - let keys = corgi::arrange::leaf_slice(key_lane(chunk.keys())).expect("identifier lane is a u64 leaf"); + fn new(chunk: &'a CorgiChunk, cid: usize, needles: &[i64], leaf_vals: bool) -> Self { + let keys = ident(chunk); let (mut lo, mut hi) = (vec![0; needles.len()], vec![0; needles.len()]); let mut found = Vec::new(); matching_ranges(needles, keys, &mut Vec::new(), &mut found); @@ -578,8 +570,8 @@ fn one_key(a: &[RunRef<'_, T>], b: &[RunRef<'_, T>]) -> bool { fn stage_collision( runs: &[RunRef<'_, T>], lower: &T, - token: u64, - bridge: &mut ProxyBridge, + token: i64, + bridge: &mut ProxyBridge, ) { let mut positions: Vec = runs.iter().map(|run| run.s).collect(); let mut scratch = SideScratch::new(); @@ -634,15 +626,15 @@ fn advance_leaf( chunks0: &[&CorgiChunk], chunks1: &[&CorgiChunk], lower: &T, - from: &mut Option, - bridge0: &mut ProxyBridge, - bridge1: &mut ProxyBridge, - colliding: &mut Vec, + from: &mut Option, + bridge0: &mut ProxyBridge, + bridge1: &mut ProxyBridge, + colliding: &mut Vec, ) { let start = from.expect("advance called on an exhausted unit"); let hashed = key_is_hashed(chunks0[0].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. + // lane. The lane is a sorted `i64` slice, so this is a slice operation, not a column probe. let seek = |chunks: &[&CorgiChunk]| -> Vec { chunks.iter().map(|c| ident(c).partition_point(|&x| x < start)).collect() }; @@ -684,16 +676,16 @@ fn leaf_probe<'a, T: ColTime>( mut dviews: Vec>, pchunks: &[&CorgiChunk], lower: &T, - h: Option, - from: &mut Option, - bridge_d: &mut ProxyBridge, - bridge_p: &mut ProxyBridge, + h: Option, + from: &mut Option, + bridge_d: &mut ProxyBridge, + bridge_p: &mut ProxyBridge, hashed: bool, - colliding: &mut Vec, + colliding: &mut Vec, ) { // 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(); + let mut keyset: Vec = Vec::new(); let mut druns: Vec<(usize, usize, usize, usize)> = Vec::new(); // (key idx, view idx, s, e) loop { let k = dviews.iter().filter_map(LeafView::cur_key).min(); @@ -763,18 +755,18 @@ fn leaf_probe<'a, T: ColTime>( *from = h; } -/// Comparable-sides regime: both sides pulled and merged symmetrically on the `u64` +/// Comparable-sides regime: both sides pulled and merged symmetrically on the `i64` /// buffers. fn leaf_merge<'a, T: ColTime>( mut views0: Vec>, mut views1: Vec>, lower: &T, - h: Option, - from: &mut Option, - bridge0: &mut ProxyBridge, - bridge1: &mut ProxyBridge, + h: Option, + from: &mut Option, + bridge0: &mut ProxyBridge, + bridge1: &mut ProxyBridge, hashed: bool, - colliding: &mut Vec, + colliding: &mut Vec, ) { 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()); @@ -821,17 +813,20 @@ fn leaf_merge<'a, T: ColTime>( mod tests { use super::*; + /// An integer column of these values. + fn ints(xs: Vec) -> CValue { CValue::i64(xs.into_iter().map(|x| x as i64).collect()) } + fn batch(rows: &[(u64, u64, u64)]) -> CBatch { let keys = CValue::Prod(vec![ - CValue::u64(rows.iter().map(|row| row.0).collect()), + ints(rows.iter().map(|row| row.0).collect()), CValue::Prod(vec![ - CValue::u64(rows.iter().map(|row| row.1).collect()), - CValue::u64(rows.iter().map(|row| row.2).collect()), + ints(rows.iter().map(|row| row.1).collect()), + ints(rows.iter().map(|row| row.2).collect()), ]), ]); // Deliberately equal across different real keys: collision staging must not consolidate // these together before `cross` has a chance to compare their keys. - let vals = CValue::u64(vec![0; rows.len()]); + let vals = ints(vec![0; rows.len()]); let chunk = CorgiChunk::from_columns(keys, vals, (0..rows.len()).map(|_| 0).collect(), vec![1; rows.len()]); Rc::new(ChunkBatch::new(vec![chunk])) } @@ -847,8 +842,8 @@ mod tests { fn cross_bridges( backend: &mut CorgiJoinBackend, instance: &JoinInstance, CBatch>, - left: &ProxyBridge, - right: &ProxyBridge, + left: &ProxyBridge, + right: &ProxyBridge, ) -> usize { let mut matches = JoinMatches::default(); for a in left { @@ -876,13 +871,13 @@ mod tests { }; let values = |ids: Vec, compound| { let n = ids.len(); - let col = wrap(CValue::u64(ids)); - if compound { CValue::Prod(vec![col, CValue::u64(vec![9; n])]) } else { col } + let col = wrap(ints(ids)); + if compound { CValue::Prod(vec![col, ints(vec![9; n])]) } else { col } }; let make = |compound, ids: Vec, diffs: Vec, time| { let n = ids.len(); Rc::new(ChunkBatch::new(vec![CorgiChunk::from_columns( - wrap(CValue::u64(vec![7; n])), values(ids, compound), + wrap(ints(vec![7; n])), values(ids, compound), (0..n).map(|_| time).collect(), diffs, )])) }; @@ -893,7 +888,7 @@ mod tests { let mut instance = JoinInstance { batches0: side(compound0), batches1: side(compound1), lower: 5 }; if let Some(right) = padded_side { let extra = CorgiChunk::from_columns( - wrap(CValue::u64(vec![8; 20])), values(vec![0; 20], if right { compound1 } else { compound0 }), + wrap(ints(vec![8; 20])), values(vec![0; 20], if right { compound1 } else { compound0 }), (0..20).collect(), vec![1; 20], ); if right { &mut instance.batches1 } else { &mut instance.batches0 } @@ -910,7 +905,7 @@ mod tests { let mut output = Vec::new(); backend.cross(&instance, &mut matches, &mut output); assert_eq!(output.len(), 1); - assert_eq!(output[0].keys, wrap(CValue::u64(vec![7]))); + assert_eq!(output[0].keys, wrap(ints(vec![7]))); assert_eq!(output[0].vals, CValue::Prod(vec![values(vec![u64::MAX], compound0), values(vec![u64::MAX], compound1)])); assert_eq!(output[0].diffs, vec![4]); assert_eq!(output[0].times.get(0), 5); @@ -923,15 +918,31 @@ mod tests { } } + /// Keys are ordered as signed integers, so a walk from the start reaches the negative ones. + #[test] + fn negative_keys_join() { + let make = |keys: Vec| Rc::new(ChunkBatch::new(vec![CorgiChunk::from_columns( + CValue::i64(keys.clone()), CValue::i64(keys.clone()), keys.iter().map(|_| 0u64).collect(), vec![1; keys.len()], + )])); + let instance = JoinInstance { batches0: vec![make(vec![i64::MIN, -5, 3])], batches1: vec![make(vec![-5, -1, 3])], lower: 0 }; + let mut backend = backend(); + let (mut from, mut left, mut right) = (KeyPosition::Start, Vec::new(), Vec::new()); + while from != KeyPosition::End { + backend.advance(&instance, &mut from, &mut left, &mut right); + } + let keys = |bridge: &ProxyBridge| bridge.iter().map(|r| r.0.0).collect::>(); + assert_eq!((keys(&left), keys(&right)), (vec![-5, 3], vec![-5, 3])); + } + #[test] fn borrowed_comparisons_match_structural_rows_after_skips() { let make = |keys: Vec, salt: u64| { let n = keys.len(); CorgiChunk::from_columns( - CValue::u64(keys), + ints(keys), CValue::Prod(vec![ - CValue::u64((0..n).map(|i| (i as u64 ^ salt) % 3).collect()), - CValue::Prod(vec![CValue::Unit(n), CValue::u64((0..n).map(|i| u64::MAX - i as u64).collect())]), + ints((0..n).map(|i| (i as u64 ^ salt) % 3).collect()), + CValue::Prod(vec![CValue::Unit(n), ints((0..n).map(|i| u64::MAX - i as u64).collect())]), ]), (0..n).map(|_| 0u64).collect(), vec![1; n], ) @@ -965,7 +976,7 @@ mod tests { fn primitive_tokens_consolidate_after_product_time_reordering() { use timely::order::Product; let chunk = CorgiChunk::from_columns( - CValue::u64(vec![7; 3]), CValue::u64(vec![11; 3]), + ints(vec![7; 3]), ints(vec![11; 3]), [Product::new(0, 2), Product::new(1, 0), Product::new(1, 2)].into_iter().collect(), vec![1, 1, -1], ); @@ -981,8 +992,8 @@ mod tests { fn compound_tokens_consolidate_after_product_time_reordering() { use timely::order::Product; let chunk = CorgiChunk::from_columns( - CValue::u64(vec![7; 3]), - CValue::Prod(vec![CValue::u64(vec![11; 3]), CValue::u64(vec![12; 3])]), + ints(vec![7; 3]), + CValue::Prod(vec![ints(vec![11; 3]), ints(vec![12; 3])]), [Product::new(0, 2), Product::new(1, 0), Product::new(1, 2)].into_iter().collect(), vec![1, 1, -1], ); @@ -1015,7 +1026,7 @@ mod tests { left.clear(); right.clear(); backend.advance(&instance, &mut from, &mut left, &mut right); - assert_eq!(backend.colliding, vec![collision]); + assert_eq!(backend.colliding, vec![collision as i64]); assert_eq!(cross_bridges(&mut backend, &instance, &left, &right), 2); } @@ -1035,7 +1046,7 @@ mod tests { let (mut left, mut right) = (Vec::new(), Vec::new()); backend.advance(&instance, &mut from, &mut left, &mut right); - assert_eq!(backend.colliding, vec![collision]); + assert_eq!(backend.colliding, vec![collision as i64]); assert_eq!(cross_bridges(&mut backend, &instance, &left, &right), 2); } } diff --git a/interactive/src/corgi/logic.rs b/interactive/src/corgi/logic.rs index 2045f7f43..57e46d63d 100644 --- a/interactive/src/corgi/logic.rs +++ b/interactive/src/corgi/logic.rs @@ -11,12 +11,13 @@ //! `corgi::shape_of` (its evaluator on zero rows) for the result shape, so every shape rule — //! which lanes a `case` sees, what an `if` may blend, whether two operands compare — is the one //! the kernels enforce, and a program that lowers is a program that runs. `Err` is a type error, -//! reported with corgi's message. Ordered compares are signed-correct (`ToSigned`); `hash` is corgi's structural -//! `Op::Hash`, the same function `ir::eval` folds row-wise. +//! reported with corgi's message. DDIR's `Int` is corgi's `Int`, so arithmetic and ordered +//! compares are corgi's own; `hash` is corgi's structural `Op::Hash`, the same function +//! `ir::eval` folds row-wise. use crate::ir::{BinOp, F64Fn, SumTy, Term, UnOp, Value as DValue}; -use corgi::{ArithOp, BinOp as CBinOp, Builder, CmpOp, Graph, Kind, NumOp, Op, Pred, Shape, Value as CValue}; +use corgi::{ArithOp, BinOp as CBinOp, Builder, CmpOp, Graph, NumOp, Op, Pred, Scalar, Shape, ShiftOp, Value as CValue}; type Res = Result; @@ -26,7 +27,7 @@ type Res = Result; /// input's schema declared. pub fn shape_of_row(row: &DValue) -> Res { match row { - DValue::Int(_) => Ok(Shape::Prim(64)), + DValue::Int(_) => Ok(Shape::Int), DValue::Tuple(xs) if xs.is_empty() => Ok(Shape::Unit), DValue::Tuple(xs) => Ok(Shape::Prod(xs.iter().map(shape_of_row).collect::>()?)), DValue::List(xs) => match xs.first() { @@ -47,7 +48,8 @@ pub fn transcode(rows: &[DValue], shape: &Shape) -> CValue { /// As [`transcode`], consuming the rows: each field moves into its column, nothing is cloned. pub fn transcode_owned(rows: Vec, shape: &Shape) -> CValue { match shape { - Shape::Prim(_) => CValue::u64(rows.iter().map(|r| r.as_int() as u64).collect()), + Shape::Int => CValue::i64(rows.iter().map(DValue::as_int).collect()), + Shape::Float => panic!("transcode: DDIR has no Float values"), Shape::Unit => CValue::Unit(rows.len()), Shape::Prod(fs) => { // Transpose by moving: one Vec per field, filled by draining each tuple. @@ -109,7 +111,8 @@ pub fn transcode_owned(rows: Vec, shape: &Shape) -> CValue { /// moves into its row; nothing is cloned. pub fn untranscode(col: CValue, shape: &Shape) -> Vec { match shape { - Shape::Prim(_) => col.into_u64("untranscode").unwrap().into_iter().map(|x| DValue::Int(x as i64)).collect(), + Shape::Int => col.into_i64("untranscode").unwrap().into_iter().map(DValue::Int).collect(), + Shape::Float => panic!("untranscode: DDIR has no Float values"), Shape::Unit => vec![DValue::unit(); col.len()], Shape::Prod(fs) => { let cols = col.into_prod("untranscode").unwrap(); @@ -278,7 +281,7 @@ pub fn compile( Term::Bound(k) => { env.len().checked_sub(1 + *k).map(|i| env[i]).ok_or_else(|| format!("binder `^{k}` is not in scope here")) } - Term::Int(n) => Ok(b.add(Op::Lit(CValue::u64(vec![*n as u64])), vec![anchor])), + Term::Int(n) => Ok(int_lit(b, anchor, *n)), Term::Tuple(fields) => { // A `Spread(t)` child splices `t`'s `Prod` fields in place; any other value is one field. let mut ids: Vec = Vec::new(); @@ -313,7 +316,7 @@ pub fn compile( let id = compile(t, b, env, env_shapes, anchor, None)?; match shape_of_term(t, env_shapes, None)? { Shape::List(_) => { - let idx = b.add(Op::Lit(CValue::u64(vec![*i as u64])), vec![anchor]); + let idx = int_lit(b, anchor, *i as i64); let pair = b.tuple(vec![idx, id]); Ok(b.add(Op::Gather, vec![pair])) } @@ -325,34 +328,24 @@ pub fn compile( let rid = compile(r, b, env, env_shapes, anchor, None)?; let pair = |b: &mut Builder, x, y| b.tuple(vec![x, y]); Ok(match op { - BinOp::Add => { let p = pair(b, lid, rid); b.add(ArithOp::Bin(CBinOp::Add, Kind::U, 64), vec![p]) } - BinOp::Sub => { let p = pair(b, lid, rid); b.add(ArithOp::Bin(CBinOp::Sub, Kind::U, 64), vec![p]) } - BinOp::Mul => { let p = pair(b, lid, rid); b.add(ArithOp::Bin(CBinOp::Mul, Kind::U, 64), vec![p]) } - BinOp::Div => { - let l = b.add(ArithOp::ToSigned, vec![lid]); - let r = b.add(ArithOp::ToSigned, vec![rid]); - let p = pair(b, l, r); - let q = b.add(ArithOp::Bin(CBinOp::Div, Kind::I, 64), vec![p]); - b.add(ArithOp::ToSigned, vec![q]) - } + BinOp::Add => { let p = pair(b, lid, rid); b.add(ArithOp::Bin(CBinOp::Add), vec![p]) } + BinOp::Sub => { let p = pair(b, lid, rid); b.add(ArithOp::Bin(CBinOp::Sub), vec![p]) } + BinOp::Mul => { let p = pair(b, lid, rid); b.add(ArithOp::Bin(CBinOp::Mul), vec![p]) } + BinOp::Div => { let p = pair(b, lid, rid); b.add(ArithOp::Bin(CBinOp::Div), vec![p]) } BinOp::Append => { let p = pair(b, lid, rid); b.add(Op::Append, vec![p]) } BinOp::F64Add | BinOp::F64Sub | BinOp::F64Mul | BinOp::F64Div => { - let f64_shape = Shape::Sum(vec![Shape::Prim(64)]); + let f64_shape = Shape::Sum(vec![Shape::Int]); if shape_of_term(l, env_shapes, None)? != f64_shape || shape_of_term(r, env_shapes, None)? != f64_shape { return Err("floating arithmetic expects two F64 newtypes; use float(int)".into()); } - let l = b.add(Op::Unwrap, vec![lid]); - let r = b.add(Op::Unwrap, vec![rid]); - let l = b.add(ArithOp::ToSigned, vec![l]); - let r = b.add(ArithOp::ToSigned, vec![r]); - let p = pair(b, l, r); + let (x, y) = (float_leaf(b, lid), float_leaf(b, rid)); + let p = pair(b, x, y); let op = match op { BinOp::F64Add => CBinOp::Add, BinOp::F64Sub => CBinOp::Sub, BinOp::F64Mul => CBinOp::Mul, _ => CBinOp::Div }; - let f = b.add(ArithOp::Bin(op, Kind::F, 64), vec![p]); - let payload = b.add(ArithOp::ToSigned, vec![f]); - b.add(Op::Inject(0, vec![Shape::Prim(64)]), vec![payload]) + let f = b.add(ArithOp::Bin(op), vec![p]); + float_newtype(b, f) } BinOp::F64Min | BinOp::F64Max | BinOp::F64Eq | BinOp::F64Ne | BinOp::F64Lt | BinOp::F64Le | BinOp::F64Gt | BinOp::F64Ge => { - let f64_shape = Shape::Sum(vec![Shape::Prim(64)]); + let f64_shape = Shape::Sum(vec![Shape::Int]); if shape_of_term(l, env_shapes, None)? != f64_shape || shape_of_term(r, env_shapes, None)? != f64_shape { return Err(format!("{op:?} expects two F64 newtypes; use float(int)")); } @@ -370,8 +363,7 @@ pub fn compile( let x_nan = is_nan(b, x); let choices_x = b.tuple(vec![x_nan, y, unless_y]); let f = b.add(Op::Select, vec![choices_x]); - let payload = b.add(ArithOp::ToSigned, vec![f]); - b.add(Op::Inject(0, vec![Shape::Prim(64)]), vec![payload]) + float_newtype(b, f) } _ => { // IEEE: the total order once `-0.0` is folded onto `0.0`, and false @@ -388,14 +380,11 @@ pub fn compile( let (x_nan, y_nan) = (is_nan(b, x), is_nan(b, y)); let nans = pair(b, x_nan, y_nan); let any_nan = b.add(CmpOp::Max, vec![nans]); - let zero = b.add(Op::Lit(CValue::u64(vec![0])), vec![anchor]); - let no_nan = pair(b, any_nan, zero); - let ordered = b.add(CmpOp::Rel(Pred::Eq), vec![no_nan]); + let ordered = b.add(CmpOp::RelImm(Pred::Eq, Scalar::Int(0)), vec![any_nan]); let both = pair(b, rel, ordered); let holds = b.add(CmpOp::Min, vec![both]); if matches!(op, BinOp::F64Ne) { - let negate = pair(b, holds, zero); - b.add(CmpOp::Rel(Pred::Eq), vec![negate]) + b.add(CmpOp::RelImm(Pred::Eq, Scalar::Int(0)), vec![holds]) } else { holds } @@ -403,8 +392,8 @@ pub fn compile( } } BinOp::F64Pow | BinOp::F64PowI => { - let f64_shape = Shape::Sum(vec![Shape::Prim(64)]); - let exponent = if matches!(op, BinOp::F64Pow) { f64_shape.clone() } else { Shape::Prim(64) }; + let f64_shape = Shape::Sum(vec![Shape::Int]); + let exponent = if matches!(op, BinOp::F64Pow) { f64_shape.clone() } else { Shape::Int }; if shape_of_term(l, env_shapes, None)? != f64_shape || shape_of_term(r, env_shapes, None)? != exponent { return Err(format!("{op:?} expects an F64 newtype and an {}", if matches!(op, BinOp::F64Pow) { "F64" } else { "Int" })); } @@ -416,36 +405,33 @@ pub fn compile( }; let args = b.tuple(vec![x, y]); let z = b.add(NumOp::Host(float_kernels::op(kind)), vec![args]); - let payload = b.add(ArithOp::ToSigned, vec![z]); - b.add(Op::Inject(0, vec![Shape::Prim(64)]), vec![payload]) + float_newtype(b, z) } BinOp::Eq | BinOp::Ne => { // Cross-shape structural compare folds to a constant (Eq→0, Ne→1) over `anchor`; // same-shape emits a real corgi `Rel`. if shape_of_term(l, env_shapes, None)? != shape_of_term(r, env_shapes, None)? { - let v = if matches!(op, BinOp::Ne) { 1u64 } else { 0u64 }; - b.add(Op::Lit(CValue::u64(vec![v])), vec![anchor]) + int_lit(b, anchor, matches!(op, BinOp::Ne) as i64) } else { let pred = if matches!(op, BinOp::Eq) { Pred::Eq } else { Pred::Ne }; let p = pair(b, lid, rid); b.add(CmpOp::Rel(pred), vec![p]) } } - // Ordered compares go through `ToSigned` (XOR the sign bit: the order-preserving - // signed encoding), so they agree with `ir::eval`'s signed semantics for negative - // ints too. `Eq`/`Ne` are bit-equality — sign-safe as raw bits. + // Ordered compares are corgi's structural order, which is `ir::eval`'s: integers + // by signed value, and an F64 newtype by its payload, the total order. BinOp::Lt | BinOp::Le | BinOp::Gt | BinOp::Ge => { - let float_shape = Shape::Sum(vec![Shape::Prim(64)]); - let (lid, rid) = if shape_of_term(l, env_shapes, None)? == float_shape && shape_of_term(r, env_shapes, None)? == float_shape { - (b.add(Op::Unwrap, vec![lid]), b.add(Op::Unwrap, vec![rid])) - } else { (lid, rid) }; - let ls = b.add(ArithOp::ToSigned, vec![lid]); - let rs = b.add(ArithOp::ToSigned, vec![rid]); - let p = if matches!(op, BinOp::Gt | BinOp::Ge) { pair(b, rs, ls) } else { pair(b, ls, rs) }; + let p = if matches!(op, BinOp::Gt | BinOp::Ge) { pair(b, rid, lid) } else { pair(b, lid, rid) }; b.add(CmpOp::Rel(if matches!(op, BinOp::Le | BinOp::Ge) { Pred::Le } else { Pred::Lt }), vec![p]) } - BinOp::And => { let p = pair(b, lid, rid); b.add(CmpOp::Min, vec![p]) } - BinOp::Or => { let p = pair(b, lid, rid); b.add(CmpOp::Max, vec![p]) } + // `min`/`max` are AND/OR only on 0/1 masks, so each operand is first read as one: + // nonzero `Int` is true, and anything else is false (as `ir::eval`'s `truthy`). + BinOp::And | BinOp::Or => { + let lm = truth(b, l, lid, env_shapes, anchor)?; + let rm = truth(b, r, rid, env_shapes, anchor)?; + let p = pair(b, lm, rm); + b.add(if matches!(op, BinOp::And) { CmpOp::Min } else { CmpOp::Max }, vec![p]) + } }) } Term::If { cond, then, els } => { @@ -583,49 +569,37 @@ pub fn compile( let id = compile(inner, b, env, env_shapes, anchor, None)?; let shape = shape_of_term(inner, env_shapes, None)?; Ok(match op { - // Wrapping negate on the raw two's-complement bits — exactly `-as_int()`. - UnOp::Neg => b.add(ArithOp::Neg(Kind::U, 64), vec![id]), + // Wrapping negate, exactly `-as_int()`. + UnOp::Neg => b.add(ArithOp::Neg, vec![id]), UnOp::ToF64 => { - if shape != Shape::Prim(64) { return Err("float expects an Int".into()); } - let sign = b.add(ArithOp::Shr(63), vec![id]); - let negative = b.add(ArithOp::Neg(Kind::U, 64), vec![id]); - let choices = b.tuple(vec![sign, negative, id]); - let magnitude = b.add(Op::Select, vec![choices]); - let positive = b.add(ArithOp::ToFloat(64), vec![magnitude]); - let negative = b.add(ArithOp::Neg(Kind::F, 64), vec![positive]); - let choices = b.tuple(vec![sign, negative, positive]); - let f = b.add(Op::Select, vec![choices]); - let payload = b.add(ArithOp::ToSigned, vec![f]); - b.add(Op::Inject(0, vec![Shape::Prim(64)]), vec![payload]) + if shape != Shape::Int { return Err("float expects an Int".into()); } + let f = b.add(ArithOp::ToFloat, vec![id]); + float_newtype(b, f) } UnOp::F64Neg => { - if shape != Shape::Sum(vec![Shape::Prim(64)]) { return Err("fneg expects an F64 newtype".into()); } - let payload = b.add(Op::Unwrap, vec![id]); - let f = b.add(ArithOp::ToSigned, vec![payload]); - let negative = b.add(ArithOp::Neg(Kind::F, 64), vec![f]); - let payload = b.add(ArithOp::ToSigned, vec![negative]); - b.add(Op::Inject(0, vec![Shape::Prim(64)]), vec![payload]) + if shape != Shape::Sum(vec![Shape::Int]) { return Err("fneg expects an F64 newtype".into()); } + let f = float_leaf(b, id); + let negative = b.add(ArithOp::Neg, vec![f]); + float_newtype(b, negative) } // |x| is the larger of x and -x in the total order: that clears the sign bit, // NaN included, exactly as `f64::abs`. UnOp::F64Fn(F64Fn::Abs) => { - if shape != Shape::Sum(vec![Shape::Prim(64)]) { return Err("fabs expects an F64 newtype".into()); } + if shape != Shape::Sum(vec![Shape::Int]) { return Err("fabs expects an F64 newtype".into()); } let f = float_leaf(b, id); let abs = float_abs(b, f); - let payload = b.add(ArithOp::ToSigned, vec![abs]); - b.add(Op::Inject(0, vec![Shape::Prim(64)]), vec![payload]) + float_newtype(b, abs) } // The rest (sqrt, exp, ln, rounding, trig) and `fint` are host kernels over the - // float leaf: decode each key, apply the function, encode the result. + // float leaf. UnOp::F64Fn(f) => { - if shape != Shape::Sum(vec![Shape::Prim(64)]) { return Err(format!("{op:?} expects an F64 newtype")); } + if shape != Shape::Sum(vec![Shape::Int]) { return Err(format!("{op:?} expects an F64 newtype")); } let x = float_leaf(b, id); let y = b.add(NumOp::Host(float_kernels::op(float_kernels::FloatOp::Fn(*f))), vec![x]); - let payload = b.add(ArithOp::ToSigned, vec![y]); - b.add(Op::Inject(0, vec![Shape::Prim(64)]), vec![payload]) + float_newtype(b, y) } UnOp::F64ToInt => { - if shape != Shape::Sum(vec![Shape::Prim(64)]) { return Err("fint expects an F64 newtype".into()); } + if shape != Shape::Sum(vec![Shape::Int]) { return Err("fint expects an F64 newtype".into()); } let x = float_leaf(b, id); b.add(NumOp::Host(float_kernels::op(float_kernels::FloatOp::ToInt)), vec![x]) } @@ -633,34 +607,19 @@ pub fn compile( // are never truthy, so their `not` folds to the constant 1 (the cross-shape // `Eq` fold's precedent). UnOp::Not => match shape { - Shape::Prim(_) => { - let zero = b.add(Op::Lit(CValue::u64(vec![0])), vec![anchor]); - let p = b.tuple(vec![id, zero]); - b.add(CmpOp::Rel(Pred::Eq), vec![p]) - } - _ => b.add(Op::Lit(CValue::u64(vec![1])), vec![anchor]), + Shape::Int => b.add(CmpOp::RelImm(Pred::Eq, Scalar::Int(0)), vec![id]), + _ => int_lit(b, anchor, 1), }, - // Tuple arity is static (a shape fact); list length folds `acc + 1` along - // each row's list; anything else is the program error `eval` reports. + // Tuple arity is static (a shape fact); a list's length is read off its bounds; + // anything else is the program error `eval` reports. UnOp::Len => match shape { - Shape::Prod(fs) => b.add(Op::Lit(CValue::u64(vec![fs.len() as u64])), vec![anchor]), - Shape::Unit => b.add(Op::Lit(CValue::u64(vec![0])), vec![anchor]), - Shape::List(_) => { - let zero = b.add(Op::Lit(CValue::u64(vec![0])), vec![anchor]); - let seed = b.tuple(vec![zero, id]); - let body = { - let mut bb = Builder::::default(); - let inp = bb.input(); - let acc = bb.add(Op::Field(0), vec![inp]); - let out = bb.add(ArithOp::BinImm(CBinOp::Add, Kind::U, 64, 1), vec![acc]); - bb.finish(out) - }; - b.add(Op::Fold(Box::new(body)), vec![seed]) - } + Shape::Prod(fs) => int_lit(b, anchor, fs.len() as i64), + Shape::Unit => int_lit(b, anchor, 0), + Shape::List(_) => b.add(Op::Len, vec![id]), other => return Err(format!("len of a {other}")), }, // On a sum, every lane maps to its constant answer and the result unwraps - // (lanes are homogeneous `U64`); on any other shape, `istag` is constantly 0 + // (lanes are homogeneous `Int`); on any other shape, `istag` is constantly 0 // (matching `eval`'s "non-Variant is never the tag"). UnOp::IsTag(t) => match shape { Shape::Sum(lanes) => { @@ -668,15 +627,14 @@ pub fn compile( .map(|i| { let mut bb = Builder::::default(); let inp = bb.input(); - let v = (i as u32 == *t) as u64; - let out = bb.add(Op::Lit(CValue::u64(vec![v])), vec![inp]); + let out = int_lit(&mut bb, inp, (i as u32 == *t) as i64); (i, bb.finish(out)) }) .collect(); let mapped = b.add(Op::MapSum(arms), vec![id]); b.add(Op::Unwrap, vec![mapped]) } - _ => b.add(Op::Lit(CValue::u64(vec![0])), vec![anchor]), + _ => int_lit(b, anchor, 0), }, }) } @@ -698,16 +656,10 @@ pub fn compile( let e = compile(f, b, env, env_shapes, anchor, None)?; lanes.push(b.add(Op::Enlist, vec![e])); } - let count = b.add(Op::Lit(CValue::u64(vec![fields.len() as u64])), vec![anchor]); - // Weave reads its tags as bytes; `iota` counts in `U64`, so each tag is narrowed. - let positions = b.add(Op::Iota, vec![count]); - let narrow = { - let mut bb = Builder::::default(); - let inp = bb.input(); - let out = bb.add(Op::Cast(8), vec![inp]); - bb.finish(out) - }; - let tags = b.add(Op::MapList(Box::new(narrow)), vec![positions]); + // Weave reads its tags as bytes, so they are one byte literal, `[0, 1, .., k-1]` per row. + let k = fields.len(); + let positions = CValue::List(corgi::Bounds::Stride(k, 1), Box::new(CValue::u8((0..k as u8).collect()))); + let tags = b.add(Op::Lit(positions), vec![anchor]); let mut weave_in = vec![tags]; weave_in.extend(lanes); let woven_in = b.tuple(weave_in); @@ -723,10 +675,8 @@ pub fn compile( // DDIR's `hash` IS corgi's `Op::Hash` (`ir::structural_hash` is the row-wise twin): hash // the arguments as one tuple, shift out the sign bit, reduce by the bound. // - // The bound guard is pure arithmetic — no `Select`. `Rem`'s total `x % 0 = x` gives - // `bound == 0` the identity, and a NEGATIVE bound reads as a `u64` at or above 2^63, - // which is larger than the shifted hash, so it reduces to the identity too. Both are - // exactly what `ir::eval`'s `if bound > 0` produces. + // `ir::eval` reduces only by a positive bound. A bound that is not positive is replaced by + // zero, and `Rem`'s total `x % 0 = x` then gives the identity. Term::Hash(args) => { let (bound, rest) = args.split_first().ok_or("hash needs a bound")?; let bid = compile(bound, b, env, env_shapes, anchor, None)?; @@ -740,46 +690,74 @@ pub fn compile( b.tuple(ids) }; let h = b.add(Op::Hash, vec![payload]); - let shifted = b.add(ArithOp::Shr(1), vec![h]); - let pair = b.tuple(vec![shifted, bid]); - Ok(b.add(ArithOp::Bin(CBinOp::Rem, Kind::U, 64), vec![pair])) + let shifted = b.add(ArithOp::Shift(ShiftOp::ShrB64, 1), vec![h]); + let positive = b.add(CmpOp::RelImm(Pred::Gt, Scalar::Int(0)), vec![bid]); + let zero = int_lit(b, anchor, 0); + let choices = b.tuple(vec![positive, bid, zero]); + let bound = b.add(Op::Select, vec![choices]); + let pair = b.tuple(vec![shifted, bound]); + Ok(b.add(ArithOp::Bin(CBinOp::Rem), vec![pair])) } } } -/// Corgi's float encoding of an F64 constant: the total-order key, as a `U64` leaf holds it. +/// An `Int` literal, one row per row of `anchor`. +fn int_lit(b: &mut Builder, anchor: usize, n: i64) -> usize { + b.add(Op::Lit(CValue::i64(vec![n])), vec![anchor]) +} + +/// `term` (compiled at `id`) as a 0/1 mask, as `ir::eval`'s `truthy` reads it: a nonzero `Int` is +/// true and any other value is false. A comparison or logical term is a mask already. +fn truth(b: &mut Builder, term: &Term, id: usize, env_shapes: &[Shape], anchor: usize) -> Res { + let mask = matches!(term, + Term::Unary(UnOp::Not | UnOp::IsTag(_), _) + | Term::Binary(BinOp::Eq | BinOp::Ne | BinOp::Lt | BinOp::Le | BinOp::Gt | BinOp::Ge | BinOp::And | BinOp::Or, _, _) + | Term::Binary(BinOp::F64Eq | BinOp::F64Ne | BinOp::F64Lt | BinOp::F64Le | BinOp::F64Gt | BinOp::F64Ge, _, _)); + Ok(match shape_of_term(term, env_shapes, None)? { + Shape::Int if mask => id, + Shape::Int => b.add(CmpOp::RelImm(Pred::Ne, Scalar::Int(0)), vec![id]), + _ => int_lit(b, anchor, 0), + }) +} + +/// The total-order key of an `f64`, as corgi's `Scalar::Float` holds a float constant. fn float_key(f: f64) -> u64 { let bits = f.to_bits(); if bits >> 63 == 1 { !bits } else { bits ^ (1 << 63) } } -/// An F64 newtype column -> its corgi float leaf (total-order key). The DDIR payload is the -/// signed form of that key, and `ToSigned` is the involution between the two. +/// An F64 newtype column -> a corgi `Float` leaf. fn float_leaf(b: &mut Builder, newtype: usize) -> usize { let payload = b.add(Op::Unwrap, vec![newtype]); - b.add(ArithOp::ToSigned, vec![payload]) + b.add(NumOp::Host(float_kernels::op(float_kernels::FloatOp::FromPayload)), vec![payload]) +} + +/// A corgi `Float` leaf -> an F64 newtype column. +fn float_newtype(b: &mut Builder, float: usize) -> usize { + let payload = b.add(NumOp::Host(float_kernels::op(float_kernels::FloatOp::ToPayload)), vec![float]); + b.add(Op::Inject(0, vec![Shape::Int]), vec![payload]) } /// `|x|` on a float leaf: the larger of `x` and `-x` in the total order. fn float_abs(b: &mut Builder, x: usize) -> usize { - let negated = b.add(ArithOp::Neg(Kind::F, 64), vec![x]); + let negated = b.add(ArithOp::Neg, vec![x]); let p = b.tuple(vec![x, negated]); b.add(CmpOp::Max, vec![p]) } -/// A 0/1 mask: is the float leaf NaN? The NaNs are exactly the keys whose magnitude exceeds +inf. +/// A 0/1 mask: is the float leaf NaN? The NaNs are exactly the values whose magnitude exceeds +/// +inf in the total order. fn is_nan(b: &mut Builder, x: usize) -> usize { let abs = float_abs(b, x); - b.add(CmpOp::RelImm(Pred::Gt, 64, float_key(f64::INFINITY)), vec![abs]) + b.add(CmpOp::RelImm(Pred::Gt, Scalar::Float(float_key(f64::INFINITY))), vec![abs]) } -/// Replace `-0.0` by `0.0` in a float leaf. Their keys are adjacent, so this adds the mask. +/// Replace `-0.0` by `0.0` in a float leaf. fn fold_negative_zero(b: &mut Builder, x: usize, anchor: usize) -> usize { - let negative_zero = b.add(Op::Lit(CValue::u64(vec![float_key(-0.0)])), vec![anchor]); - let probe = b.tuple(vec![x, negative_zero]); - let is_negative_zero = b.add(CmpOp::Rel(Pred::Eq), vec![probe]); - let sum = b.tuple(vec![x, is_negative_zero]); - b.add(ArithOp::Bin(CBinOp::Add, Kind::U, 64), vec![sum]) + let is_negative_zero = b.add(CmpOp::RelImm(Pred::Eq, Scalar::Float(float_key(-0.0))), vec![x]); + let zero = b.add(Op::Lit(CValue::f64(vec![0.0])), vec![anchor]); + let choices = b.tuple(vec![is_negative_zero, zero, x]); + b.add(Op::Select, vec![choices]) } /// Compile a `Fold` step into a closed corgi sub-graph. Without capture the body's input is @@ -835,11 +813,11 @@ pub fn compile_flatmap(list_term: &Term, kshape: &Shape, vshape: &Shape) -> Res< } } -/// Compile a scalar term (`EnterAt`'s delay field) → a `U64` column; a non-integer term is the +/// Compile a scalar term (`EnterAt`'s delay field) → an `Int` column; a non-integer term is the /// type error (the delay is read as one integer per row). pub fn compile_scalar(term: &Term, kshape: &Shape, vshape: &Shape) -> Res> { match lower(&[term], &[kshape.clone(), vshape.clone()])? { - (k, Shape::Prim(_)) => Ok(k), + (k, Shape::Int) => Ok(k), (_, other) => Err(format!("enter_at delay is not an integer: {other}")), } } @@ -848,7 +826,7 @@ pub fn compile_scalar(term: &Term, kshape: &Shape, vshape: &Shape) -> Res Res> { match lower(&[cond], &[kshape.clone(), vshape.clone()])? { - (k, Shape::Prim(_)) => Ok(k), + (k, Shape::Int) => Ok(k), (_, other) => Err(format!("a filter predicate must be an Int, got {other}")), } } @@ -865,21 +843,17 @@ pub fn compile_projection(key: &Term, val: &Term, kshape: &Shape, vshape: &Shape Ok(lower(&[key, val], &[kshape.clone(), vshape.clone()])?.0) } -/// The F64 functions without a composition of corgi ops, as host kernels over the float leaf -/// (corgi's total-order key): decode each key, apply the function as `ir::eval` does, encode. -/// One kernel per function, shared by every call site, so CSE merges equal calls. +/// The F64 functions without a composition of corgi ops, as host kernels over the float leaf: +/// apply the function as `ir::eval` does. `FromPayload` and `ToPayload` convert between an F64 +/// newtype's `Int` payload ([`DValue::f64_value`]) and a float leaf. One kernel per function, +/// shared by every call site, so CSE merges equal calls. mod float_kernels { use std::sync::{Arc, OnceLock}; use corgi::{HostKernel, HostOp, Shape, Value}; - use crate::ir::F64Fn; + use crate::ir::{F64Fn, Value as DValue}; #[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)] - pub enum FloatOp { Fn(F64Fn), ToInt, Pow, PowI } - - fn decode(key: u64) -> f64 { - f64::from_bits(if key >> 63 == 1 { key ^ (1 << 63) } else { !key }) - } - fn encode(x: f64) -> u64 { super::float_key(x) } + pub enum FloatOp { Fn(F64Fn), ToInt, Pow, PowI, FromPayload, ToPayload } struct Kernel { name: String, op: FloatOp, input: Shape, output: Shape } impl HostKernel for Kernel { @@ -887,23 +861,26 @@ mod float_kernels { fn input(&self) -> &Shape { &self.input } fn output(&self) -> &Shape { &self.output } fn eval(&self, input: Value) -> Result { - let out: Vec = match self.op { - FloatOp::Fn(f) => input.as_u64(&self.name)?.iter().map(|&k| encode(f.apply(decode(k)))).collect(), - FloatOp::ToInt => input.as_u64(&self.name)?.iter().map(|&k| decode(k) as i64 as u64).collect(), + let payload = |x: f64| match DValue::f64_value(x) { DValue::Variant(_, p) => p.as_int(), _ => unreachable!() }; + Ok(match self.op { + FloatOp::Fn(f) => Value::f64(input.as_f64(&self.name)?.into_iter().map(|x| f.apply(x)).collect()), + FloatOp::ToInt => Value::i64(input.as_f64(&self.name)?.into_iter().map(|x| x as i64).collect()), FloatOp::Pow | FloatOp::PowI => { let args = input.into_prod(&self.name)?; - let (x, y) = (args[0].as_u64(&self.name)?, args[1].as_u64(&self.name)?); - x.iter().zip(y).map(|(&a, &b)| { - let a = decode(a); - encode(if self.op == FloatOp::Pow { - a.powf(decode(b)) - } else { - a.powi((b as i64).clamp(i32::MIN as i64, i32::MAX as i64) as i32) - }) - }).collect() + let x = args[0].as_f64(&self.name)?; + Value::f64(if self.op == FloatOp::Pow { + x.into_iter().zip(args[1].as_f64(&self.name)?).map(|(a, b)| a.powf(b)).collect() + } else { + x.into_iter().zip(args[1].as_i64(&self.name)?.iter()).map(|(a, &b)| { + a.powi(b.clamp(i32::MIN as i64, i32::MAX as i64) as i32) + }).collect() + }) } - }; - Ok(Value::u64(out)) + FloatOp::FromPayload => Value::f64(input.as_i64(&self.name)?.iter().map(|&p| { + DValue::Variant(0, Box::new(DValue::Int(p))).as_f64() + }).collect()), + FloatOp::ToPayload => Value::i64(input.as_f64(&self.name)?.into_iter().map(payload).collect()), + }) } } @@ -912,14 +889,16 @@ mod float_kernels { static KERNELS: OnceLock>>> = OnceLock::new(); let mut map = KERNELS.get_or_init(Default::default).lock().unwrap(); let k = map.entry(op).or_insert_with(|| { - let (p, pair) = (Shape::Prim(64), Shape::Prod(vec![Shape::Prim(64), Shape::Prim(64)])); - let (name, input) = match op { - FloatOp::Fn(f) => (format!("{f:?}").to_lowercase(), p.clone()), - FloatOp::ToInt => ("fint".into(), p.clone()), - FloatOp::Pow => ("fpow".into(), pair.clone()), - FloatOp::PowI => ("fpowi".into(), pair), + let (f, i) = (Shape::Float, Shape::Int); + let (name, input, output) = match op { + FloatOp::Fn(g) => (format!("{g:?}").to_lowercase(), f.clone(), f), + FloatOp::ToInt => ("fint".into(), f, i), + FloatOp::Pow => ("fpow".into(), Shape::Prod(vec![f.clone(), f.clone()]), f), + FloatOp::PowI => ("fpowi".into(), Shape::Prod(vec![f.clone(), i]), f), + FloatOp::FromPayload => ("from_payload".into(), i, f), + FloatOp::ToPayload => ("to_payload".into(), f, i), }; - Arc::new(Kernel { name: format!("f64:{name}"), op, input, output: p }) + Arc::new(Kernel { name: format!("f64:{name}"), op, input, output }) }); HostOp(Arc::clone(k)) } @@ -930,7 +909,7 @@ mod tests { use super::*; use crate::ir::Value as V; - fn u64s() -> Shape { Shape::Prim(64) } + fn ints() -> Shape { Shape::Int } fn sum(lanes: Vec) -> Shape { Shape::Sum(lanes) } fn agrees_with_rows(term: &Term, shapes: &[Shape], rows: &[Vec]) { @@ -957,9 +936,13 @@ mod tests { "fadd(float($0), float($1))", "fsub(float($0), float($1))", "fmul(float($0), float($1))", "fdiv(float($0), float($1))", "float($0) < float($1)", "float($0) >= float($1)", - "append(list($0), list($1, $0))"] { + "append(list($0), list($1, $0))", + // Truthiness is "nonzero", so a negative operand is true and the result is 0/1. + "$0 && $1", "or($0, $1)", "not($0)", "or($0 < $1, $1)", "if($0, $1, 3)", + // `hash` reduces by a positive bound only. + "hash($1, $0)", "hash($0, $1, tuple($0, $1))"] { let term = crate::parse::pipe::parse_term(source); - agrees_with_rows(&term, &[u64s(), u64s()], &rows); + agrees_with_rows(&term, &[ints(), ints()], &rows); } } @@ -973,7 +956,7 @@ mod tests { /// and the ones that are host kernels (`float_kernels`), alone and inside larger terms. #[test] fn columnar_float_math_agrees() { - let f = sum(vec![u64s()]); + let f = sum(vec![ints()]); for source in ["fabs($0)", "fmin($0, $1)", "fmax($0, $1)", "feq($0, $1)", "fne($0, $1)", "flt($0, $1)", "fle($0, $1)", "fgt($0, $1)", "fge($0, $1)", "fsqrt($0)", "fexp($0)", "fln($0)", "ffloor($0)", "fceil($0)", "fround($0)", @@ -989,7 +972,7 @@ mod tests { agrees_with_rows(&term, &[f.clone(), f.clone()], &finite); // The kernels are typed: their operands must be F64 newtypes (and `fpowi`'s exponent an Int). for bad in ["fsqrt($0)", "fint($0)", "fpow($0, $0)"] { - assert!(lower(&[&crate::parse::pipe::parse_term(bad)], &[u64s()]).is_err(), "{bad}"); + assert!(lower(&[&crate::parse::pipe::parse_term(bad)], &[ints()]).is_err(), "{bad}"); } assert!(lower(&[&crate::parse::pipe::parse_term("fpowi($0, $0)")], &[f.clone()]).is_err()); } @@ -1019,9 +1002,9 @@ mod tests { fn declared_constructor_types_an_empty_list() { let term = Term::Inject { tag: Box::new(Term::Int(0)), payload: Box::new(Term::List(vec![])), - sum: SumTy::Declared(vec![Shape::List(Box::new(Shape::List(Box::new(u64s()))))]), + sum: SumTy::Declared(vec![Shape::List(Box::new(Shape::List(Box::new(ints()))))]), }; - agrees_with_rows(&term, &[u64s()], &[vec![V::Int(0)], vec![V::Int(1)]]); + agrees_with_rows(&term, &[ints()], &[vec![V::Int(0)], vec![V::Int(1)]]); } /// The pin on DDIR's `hash`: `ir::structural_hash` is a row-at-a-time transcription of @@ -1037,7 +1020,7 @@ mod tests { #[test] fn hash_matches_corgi_on_scalars() { - hash_agrees(vec![V::Int(0), V::Int(1), V::Int(-1), V::Int(i64::MIN), V::Int(i64::MAX)], u64s()); + hash_agrees(vec![V::Int(0), V::Int(1), V::Int(-1), V::Int(i64::MIN), V::Int(i64::MAX)], ints()); } #[test] @@ -1057,7 +1040,7 @@ mod tests { fn hash_matches_corgi_on_lists() { hash_agrees( vec![V::List(vec![V::Int(1), V::Int(2), V::Int(3)]), V::List(vec![]), V::List(vec![V::Int(3), V::Int(2), V::Int(1)])], - Shape::List(Box::new(u64s())), + Shape::List(Box::new(ints())), ); } @@ -1065,19 +1048,19 @@ mod tests { fn hash_matches_corgi_on_variants() { hash_agrees( vec![V::Variant(0, Box::new(V::Int(5))), V::Variant(1, Box::new(V::Int(5))), V::Variant(0, Box::new(V::Int(6)))], - sum(vec![u64s(), u64s()]), + sum(vec![ints(), ints()]), ); } #[test] fn hash_matches_corgi_on_nesting() { - let pair = Shape::Prod(vec![u64s(), u64s()]); + let pair = Shape::Prod(vec![ints(), ints()]); hash_agrees( vec![ V::Tuple(vec![V::List(vec![V::Int(1)]), V::Variant(0, Box::new(V::Tuple(vec![V::Int(2), V::Int(3)])))]), V::Tuple(vec![V::List(vec![V::Int(1), V::Int(1)]), V::Variant(0, Box::new(V::Tuple(vec![V::Int(2), V::Int(4)])))]), ], - Shape::Prod(vec![Shape::List(Box::new(u64s())), sum(vec![pair, Shape::Unit])]), + Shape::Prod(vec![Shape::List(Box::new(ints())), sum(vec![pair, Shape::Unit])]), ); } @@ -1098,7 +1081,7 @@ mod tests { V::Variant(0, Box::new(V::List(vec![V::Int(3)]))), V::Variant(0, Box::new(V::List(vec![]))), ], - sum(vec![Shape::List(Box::new(u64s()))]), + sum(vec![Shape::List(Box::new(ints()))]), ); } @@ -1113,7 +1096,7 @@ mod tests { V::Variant(1, Box::new(V::Tuple(vec![V::Int(3), V::Int(4)]))), V::Variant(0, Box::new(V::Int(30))), ], - sum(vec![u64s(), Shape::Prod(vec![u64s(), u64s()])]), + sum(vec![ints(), Shape::Prod(vec![ints(), ints()])]), ); } @@ -1122,7 +1105,7 @@ mod tests { // tags {0, 2} present, arm 1 absent: its lane is an empty column of the declared shape. roundtrip( vec![V::Variant(0, Box::new(V::Int(1))), V::Variant(2, Box::new(V::Int(2))), V::Variant(0, Box::new(V::Int(3)))], - sum(vec![u64s(), Shape::List(Box::new(u64s())), u64s()]), + sum(vec![ints(), Shape::List(Box::new(ints())), ints()]), ); } @@ -1133,13 +1116,13 @@ mod tests { V::Tuple(vec![V::Int(1), V::Variant(0, Box::new(V::Int(7)))]), V::Tuple(vec![V::Int(2), V::Variant(1, Box::new(V::unit()))]), ], - Shape::Prod(vec![u64s(), sum(vec![u64s(), Shape::Unit])]), + Shape::Prod(vec![ints(), sum(vec![ints(), Shape::Unit])]), ); } #[test] fn shape_of_row_pins_what_a_row_can_say() { - assert_eq!(shape_of_row(&V::Tuple(vec![V::Int(1), V::List(vec![V::Int(2)])])).unwrap(), Shape::Prod(vec![u64s(), Shape::List(Box::new(u64s()))])); + assert_eq!(shape_of_row(&V::Tuple(vec![V::Int(1), V::List(vec![V::Int(2)])])).unwrap(), Shape::Prod(vec![ints(), Shape::List(Box::new(ints()))])); assert!(shape_of_row(&V::List(vec![])).is_err()); assert!(shape_of_row(&V::Variant(0, Box::new(V::Int(1)))).is_err()); } diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index 68a2864a4..7f27357ee 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -19,10 +19,8 @@ //! * materialize — gather the emitted keys and values from the pools' columns and seal a //! `CorgiChunk` batch column-natively. //! -//! Min/Collect values use a segmented structural sort over an order-only columnar view: signed -//! integer leaves are swizzled, and lists keep their structure, as Corgi orders them -//! lexicographically just as DDIR does. The winning rows are still gathered from the original -//! columns. +//! Min/Collect values use a segmented structural sort over the candidate columns: Corgi orders +//! integers as signed values and lists lexicographically, just as DDIR does. //! //! The changed-key restriction is honored by presenting only the changed keys: novel batches are //! read whole (delta-sized), the accumulated history is scanned and filtered to the changed hashes @@ -38,40 +36,15 @@ use differential_dataflow::operators::int_proxy::{KeyPosition, ProxyBridge}; use differential_dataflow::operators::int_proxy::reduce::{ProxyReduceBackend, ReduceInstance, ReduceWindow}; use corgi::arrange::{compare_at, gather_lanes, sort_blocks}; -use corgi::{ArithOp, Bounds, NumOp, OpLike, Value as CValue}; +use corgi::{Bounds, Value as CValue}; use crate::corgi::col_times::{ColTime, ColTimes}; use crate::corgi::search::matching_ranges; -use crate::corgi::chunk::{columns_to_batch, key_ids, key_is_hashed, key_lane, CorgiChunk}; +use crate::corgi::chunk::{columns_to_batch, ident, key_ids, key_is_hashed, CorgiChunk}; use crate::ir::{Diff, Reducer}; type CBatch = Rc>>; -/// Build a sortable view matching DDIR's signed leaves. -/// -/// DDIR's leaf scalar is `Int`, transcoded into a Corgi primitive as its raw -/// bits. Corgi's radix sort is unsigned, so XORing each payload leaf's sign bit -/// turns signed order into unsigned order. Sum discriminants remain untouched; -/// only their payload lanes recurse. This consumes freshly gathered candidate -/// columns, allowing Corgi to swizzle their buffers in place when unshared. -fn signed_order_view(value: CValue) -> CValue { - match value { - value @ CValue::Prim(_) => NumOp::from(ArithOp::ToSigned).eval(value).expect("ToSigned on a leaf"), - CValue::Prod(fields) => { - CValue::Prod(fields.into_iter().map(signed_order_view).collect()) - } - CValue::Sum(tags, variants) => { - // the lane assignment is untouched — only the payload lanes are swizzled. - CValue::Sum(tags, variants.into_iter().map(signed_order_view).collect()) - } - // Corgi orders lists lexicographically, as DDIR does, so only their elements are swizzled. - CValue::List(bounds, values) => CValue::List(bounds, Box::new(signed_order_view(*values))), - CValue::Unit(len) => CValue::Unit(len), - // References order as the rows they name. - CValue::Ref(arena, rows) => signed_order_view(corgi::arrange::gather(&arena, &rows)), - } -} - /// Values named by id: id `i` is row `refs[i].1` of `columns[refs[i].0]`. #[derive(Default)] struct Pool { @@ -125,7 +98,7 @@ pub struct CorgiReduceBackend { #[derive(Default)] struct Retire { /// The retire's keys, ascending. - keys: Vec, + keys: Vec, /// The input records each key holds, novel and prior. held: Vec, /// Where the input chunks, prior then novel, hold the keys. @@ -133,7 +106,7 @@ struct Retire { /// The current window, as a range of `keys`. window: Range, /// The window's identifiers that hold more than one real key, ascending. - colliding: Vec, + colliding: Vec, } impl Retire { @@ -187,13 +160,12 @@ impl CorgiReduceBackend { /// /// Both the keys and stored identifier lane are sorted; [`matching_ranges`] chooses between /// merging them and searching the keys in lockstep. -fn search(chunks: &[&CorgiChunk], keys: &[u64]) -> Matches { +fn search(chunks: &[&CorgiChunk], keys: &[i64]) -> Matches { let mut scratch = Vec::new(); chunks.iter().map(|chunk| { let mut found = Vec::new(); if chunk.diffs().is_empty() { return found; } - let lane = corgi::arrange::leaf_slice(key_lane(chunk.keys())).expect("the identifier lane is a u64 leaf"); - matching_ranges(keys, lane, &mut scratch, &mut found); + matching_ranges(keys, ident(chunk.keys()), &mut scratch, &mut found); found }).collect() } @@ -208,11 +180,11 @@ fn search(chunks: &[&CorgiChunk], keys: &[u64]) -> Matches /// at holds that key. fn present( chunks: &[&CorgiChunk], - keys: &[u64], + keys: &[i64], matches: &Matches, colliding: &[bool], refs: &mut Vec<(usize, usize)>, - bridge: &mut ProxyBridge, + bridge: &mut ProxyBridge, ) { // The matched `(chunk, row)`s, key by key, and each one's key index. let mut runs: Vec<_> = matches.iter().enumerate() @@ -285,7 +257,7 @@ fn split_real_keys( /// side, carry more than one real key. Chunks sort each identifier's rows by real key, so a run holds /// one key exactly when its first and last rows agree: two comparisons per run, never per row. /// Comparing across sides also catches an identifier whose input and output hold different keys. -fn colliding_keys(keys: &[u64], sides: &[(&[&CorgiChunk], &Matches)]) -> Vec { +fn colliding_keys(keys: &[i64], sides: &[(&[&CorgiChunk], &Matches)]) -> Vec { let mut colliding = vec![false; keys.len()]; if !sides.iter().flat_map(|side| side.0).any(|chunk| key_is_hashed(chunk.keys())) { return colliding; @@ -315,8 +287,8 @@ where /// A retire's keys, ascending: those the novel batches touch, and the `changed` set the harness /// supplies. -fn retire_keys(novel_chunks: &[&CorgiChunk], changed: &[u64]) -> Vec { - let mut keys: Vec = novel_chunks.iter().flat_map(|chunk| key_ids(chunk.keys())).chain(changed.iter().copied()).collect(); +fn retire_keys(novel_chunks: &[&CorgiChunk], changed: &[i64]) -> Vec { + let mut keys: Vec = novel_chunks.iter().flat_map(|chunk| key_ids(chunk.keys())).chain(changed.iter().copied()).collect(); keys.sort_unstable(); keys.dedup(); keys @@ -333,18 +305,18 @@ where let mut out_ends: Vec = Vec::with_capacity(ends.len()); match self.reducer { Reducer::Count => { - // Per-bracket sum of diffs; survivors become a `Tuple([Int(sum)])` = corgi `Prod([u64])`. - let mut sums: Vec = Vec::new(); + // Per-bracket sum of diffs; survivors become a `Tuple([Int(sum)])` = corgi `Prod([Int])`. + let mut sums: Vec = Vec::new(); let mut start = 0; for &end in ends { let c: Diff = input[start..end].iter().map(|&(_, d)| d).sum(); if c > 0 { - sums.push(c as u64); + sums.push(c); } out_ends.push(sums.len()); start = end; } - (CValue::Prod(vec![CValue::u64(sums)]), out_ends) + (CValue::Prod(vec![CValue::i64(sums)]), out_ends) } Reducer::Distinct => { // Present iff any value has NON-ZERO net -- the sign does not matter. DD's `reduce` @@ -393,14 +365,14 @@ where return (CValue::Unit(0), out_ends); } let candidates = self.input.gather(&cand_reps); - // Integer values need no sort: scan each block for its least as signed. + // Integer values need no sort: scan each block for its least. let min_reps: Vec = if let Some(values) = corgi::arrange::leaf_slice(&candidates) { let block_ends = block_starts.iter().skip(1).copied().chain([cand_reps.len()]); block_starts.iter().zip(block_ends).map(|(&lo, hi)| { - cand_reps[(lo..hi).min_by_key(|&k| values[k] as i64).expect("blocks are non-empty")] + cand_reps[(lo..hi).min_by_key(|&k| values[k]).expect("blocks are non-empty")] }).collect() } else { - let (perm, _) = sort_blocks(&labels, &signed_order_view(candidates)); + let (perm, _) = sort_blocks(&labels, &candidates); block_starts.iter().map(|&lo| cand_reps[perm[lo]]).collect() }; (self.input.gather(&min_reps), out_ends) @@ -434,7 +406,7 @@ where if blocks.is_empty() { return (CValue::Unit(0), out_ends); } - let perm = sort_blocks(&labels, &signed_order_view(self.input.gather(&entry_reps))).0; + let perm = sort_blocks(&labels, &self.input.gather(&entry_reps)).0; // Expand each bracket's sorted entries by their diff (max(0, ·) copies). let mut elem_reps: Vec = Vec::new(); let mut bracket_ends: Vec = Vec::with_capacity(ends.len()); @@ -461,7 +433,7 @@ impl ProxyReduceBackend, CBatch> for CorgiReduceBackend where T: ColTime + Ord, { - type Key = u64; + type Key = i64; type VIn = u64; type VOut = u64; type RIn = Diff; @@ -471,7 +443,7 @@ where self.rows = (Vec::new(), Vec::new(), ColTimes::default(), Vec::new()); } - fn next_window(&mut self, instance: &ReduceInstance<'_, T, CBatch, CBatch>, changed: &[u64], from: &mut KeyPosition, window: &mut ReduceWindow) { + fn next_window(&mut self, instance: &ReduceInstance<'_, T, CBatch, CBatch>, changed: &[i64], from: &mut KeyPosition, window: &mut ReduceWindow) { if *from == KeyPosition::End { return; } @@ -534,7 +506,7 @@ where *from = retire.keys.get(retire.window.end).map_or(KeyPosition::End, |key| KeyPosition::At(*key)); } - fn reduce_corrections(&mut self, keys: &[u64], in_ends: &[usize], input: &[(u64, Diff)], out_ends: &[usize], output: &[(u64, Diff)]) -> (Vec<(u64, Diff)>, Vec) { + fn reduce_corrections(&mut self, keys: &[i64], in_ends: &[usize], input: &[(u64, Diff)], out_ends: &[usize], output: &[(u64, Diff)]) -> (Vec<(u64, Diff)>, Vec) { // Each key wants its rows of `desired`, once each, and has `(id, diff)`s of output. The // correction nets the two by value: a wanted row equal to an output id's value counts // toward that id, and one equal to none gets a new id. Keys have a handful of each. @@ -604,7 +576,7 @@ where (corr, corr_ends) } - fn emit(&mut self, records: &[((u64, u64), T, Diff)]) { + fn emit(&mut self, records: &[((i64, u64), T, Diff)]) { // An output id names its key's row, as two keys may share an identifier. let (key_rows, ids, times, diffs) = &mut self.rows; for ((_, id), time, diff) in records { @@ -631,16 +603,19 @@ mod tests { use crate::corgi::chunk::present_key; use crate::ir::Time; - /// The rows of a column of `u64` leaves, products of them, or units, as vectors. + /// The rows of a column of `Int` leaves, products of them, or units, as vectors. fn rows_of(col: &CValue) -> Vec> { - let fields: Vec<&[u64]> = match col { + let fields: Vec> = match col { CValue::Unit(_) => Vec::new(), - CValue::Prod(fields) => fields.iter().map(|field| corgi::arrange::leaf_slice(field).unwrap()).collect(), - leaf => vec![corgi::arrange::leaf_slice(leaf).unwrap()], + CValue::Prod(fields) => fields.iter().map(|field| field.as_i64("rows_of").unwrap().into_owned()).collect(), + leaf => vec![leaf.as_i64("rows_of").unwrap().into_owned()], }; - (0..col.len()).map(|i| fields.iter().map(|field| field[i]).collect()).collect() + (0..col.len()).map(|i| fields.iter().map(|field| field[i] as u64).collect()).collect() } + /// An integer column of these values. + fn ints(xs: Vec) -> CValue { CValue::i64(xs.into_iter().map(|x| x as i64).collect()) } + /// Random `(key, value, time, diff)` rows over few keys and values, so that they collide. fn random_rows(seed: u64, count: usize, keys: u64) -> Vec<(u64, (u64, u64), Time, Diff)> { let stamp = |outer, coords: &[u64]| Time::new(outer, PointStamp::new(coords.iter().copied().collect())); @@ -654,12 +629,12 @@ mod tests { } const KEY_SHAPES: [fn(&[u64]) -> CValue; 2] = [ - |k| CValue::u64(k.to_vec()), - |k| present_key(CValue::Prod(vec![CValue::u64(k.to_vec()), CValue::u64(k.iter().map(|k| k * 7).collect())])), + |k| ints(k.to_vec()), + |k| present_key(CValue::Prod(vec![ints(k.to_vec()), ints(k.iter().map(|k| k * 7).collect())])), ]; const VAL_SHAPES: [fn(&[(u64, u64)]) -> CValue; 3] = [ - |v| CValue::u64(v.iter().map(|v| v.0).collect()), - |v| CValue::Prod(vec![CValue::u64(v.iter().map(|v| v.0).collect()), CValue::u64(v.iter().map(|v| v.1).collect())]), + |v| ints(v.iter().map(|v| v.0).collect()), + |v| CValue::Prod(vec![ints(v.iter().map(|v| v.0).collect()), ints(v.iter().map(|v| v.1).collect())]), |v| CValue::Unit(v.len()), ]; @@ -674,7 +649,7 @@ mod tests { } /// The netted records of `keys` in `chunks`, by key id and value. - fn netted(chunks: &[&CorgiChunk], keys: &BTreeSet) -> BTreeMap<(u64, Vec, Time), Diff> { + fn netted(chunks: &[&CorgiChunk], keys: &BTreeSet) -> BTreeMap<(i64, Vec, Time), Diff> { let mut netted = BTreeMap::new(); for chunk in chunks { for (i, (key, value)) in key_ids(chunk.keys()).into_iter().zip(rows_of(chunk.vals())).enumerate() { @@ -718,10 +693,10 @@ mod tests { #[test] fn corrections_net_by_value() { let mut backend = CorgiReduceBackend::