From ae6807f8e3147b44012b8da3919537ef124c5dd0 Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Fri, 21 Aug 2026 18:21:38 +0000 Subject: [PATCH] Add content hashing and equality glue to record layouts --- node-graph/libraries/core-types/src/record.rs | 249 +++++++++++++++++- node-graph/nodes/gcore/src/record.rs | 4 + 2 files changed, 251 insertions(+), 2 deletions(-) diff --git a/node-graph/libraries/core-types/src/record.rs b/node-graph/libraries/core-types/src/record.rs index 037473ee1d..a216dd3075 100644 --- a/node-graph/libraries/core-types/src/record.rs +++ b/node-graph/libraries/core-types/src/record.rs @@ -21,10 +21,25 @@ pub struct FieldWrite { pub align: usize, pub read_erased: unsafe fn(*const u8) -> Box, pub repark: Option Option<()>>, + /// Hashes the field's content. `None` means the stored bytes are the + /// content, which holds for every unparked value. + pub content_hash: Option, + /// Compares two fields' content. `None` as for `content_hash`. + pub content_eq: Option bool>, } impl FieldWrite { - pub fn of(level: u8) -> Self { + pub fn of(level: u8) -> Self + where + A::Value<'static>: graphene_hash::CacheHash + PartialEq, + { + unsafe fn content_hash(ptr: *const u8, state: &mut dyn core::hash::Hasher) { + let mut state = state; + unsafe { &*ptr.cast::() }.cache_hash(&mut state); + } + unsafe fn content_eq(a: *const u8, b: *const u8) -> bool { + unsafe { &*a.cast::() == &*b.cast::() } + } Self { name: A::NAME, level, @@ -32,6 +47,8 @@ impl FieldWrite { align: align_of::>(), read_erased: A::read_erased, repark: A::REPARK, + content_hash: Some(content_hash::>), + content_eq: Some(content_eq::>), } } } @@ -50,6 +67,11 @@ pub struct FieldDesc { pub align: usize, pub read_erased: unsafe fn(*const u8) -> Box, pub repark: Option Option<()>>, + /// Hashes the field's content. `None` means the stored bytes are the + /// content, which holds for every unparked value. + pub content_hash: Option, + /// Compares two fields' content. `None` as for `content_hash`. + pub content_eq: Option bool>, } impl PartialEq for FieldDesc { @@ -71,6 +93,11 @@ pub struct ElementWrite { pub parked: bool, pub clone_out: unsafe fn(*const u8) -> Box, pub repark: unsafe fn(&(dyn std::any::Any + Send + Sync), *mut u8, &crate::arena::Arena) -> Option<()>, + /// Hashes the element's content. `None` means the stored bytes are the + /// content, which holds for every unparked element. + pub content_hash: Option, + /// Compares two elements' content. `None` as for `content_hash`. + pub content_eq: Option bool>, } impl PartialEq for ElementWrite { @@ -95,6 +122,8 @@ impl Default for ElementWrite { parked: false, clone_out, repark, + content_hash: None, + content_eq: None, } } } @@ -155,6 +184,8 @@ impl Layout { align: field.align, read_erased: field.read_erased, repark: field.repark, + content_hash: field.content_hash, + content_eq: field.content_eq, }) .collect(); for &write in writes { @@ -179,6 +210,8 @@ impl Layout { align: write.align, read_erased: write.read_erased, repark: write.repark, + content_hash: write.content_hash, + content_eq: write.content_eq, }; offset += write.size; desc @@ -207,6 +240,8 @@ impl Layout { align: field.align, read_erased: field.read_erased, repark: field.repark, + content_hash: field.content_hash, + content_eq: field.content_eq, }) .collect(); Layout::default().with_writes(self.depth, self.element, &retained) @@ -229,6 +264,8 @@ impl Layout { align: field.align, read_erased: field.read_erased, repark: field.repark, + content_hash: field.content_hash, + content_eq: field.content_eq, }) .collect(); union = union.with_writes(union.depth, union.element, &writes); @@ -1129,6 +1166,51 @@ pub fn element_write() -> ElementWrite { parked: element_parked::(), clone_out: clone_out::, repark: repark::, + content_hash: None, + content_eq: None, + } +} + +/// [`element_write`] plus the content hashing and equality glue, for element +/// types that support them. +pub fn element_write_hashed() -> ElementWrite { + unsafe fn content_hash(ptr: *const u8, state: &mut dyn core::hash::Hasher) { + let mut state = state; + unsafe { borrow_element::(Rec::new(ptr)) }.cache_hash(&mut state); + } + unsafe fn content_eq(a: *const u8, b: *const u8) -> bool { + unsafe { borrow_element::(Rec::new(a)) == borrow_element::(Rec::new(b)) } + } + ElementWrite { + content_hash: Some(content_hash::), + content_eq: Some(content_eq::), + ..element_write::() + } +} + +/// Selects [`element_write_hashed`] when the element type supports the +/// content glue and [`element_write`] otherwise, by autoref method +/// resolution: call `(&ElementWritePick::(..)).element_write()` with both +/// traits in scope. +pub struct ElementWritePick(pub std::marker::PhantomData); + +pub trait ElementWritePickHashed { + fn element_write(&self) -> ElementWrite; +} + +impl ElementWritePickHashed for ElementWritePick { + fn element_write(&self) -> ElementWrite { + element_write_hashed::() + } +} + +pub trait ElementWritePickPlain { + fn element_write(&self) -> ElementWrite; +} + +impl ElementWritePickPlain for &ElementWritePick { + fn element_write(&self) -> ElementWrite { + element_write::() } } @@ -1503,9 +1585,15 @@ unsafe impl Sync for GroupItem {} impl GroupItem { /// Copies the batch's lanes into the arena and clones its layout. - /// Returns `None` when the arena is exhausted. + /// Returns `None` when the arena is exhausted. Parked regions must carry + /// the content glue, so equality and hashing never fall back to pointer + /// bytes. pub fn adopt(batch: crate::node::RecordBatch<'_>, arena: &crate::arena::Arena) -> Option { let layout = batch.layout().clone(); + assert!(!layout.element.parked || layout.element.content_hash.is_some(), "a parked element adopts only with content glue"); + for field in &layout.fields { + assert!(field.repark.is_none() || field.content_hash.is_some(), "a parked field adopts only with content glue"); + } let stride = layout.lane_stride(); let scratch = arena.alloc_scratch::((batch.len() * stride).div_ceil(8))?; let frames = scratch.as_mut_ptr().cast::(); @@ -1556,6 +1644,114 @@ pub struct Group { pub content: GroupContent, } +/// Compares one record region of `layout` by content. Regions without glue +/// compare as bytes, which is the content for unparked values. +unsafe fn record_content_eq(layout: &Layout, a: *const u8, b: *const u8) -> bool { + let bytes_eq = |offset: usize, size: usize| unsafe { std::slice::from_raw_parts(a.add(offset), size) == std::slice::from_raw_parts(b.add(offset), size) }; + let element = match layout.element.content_eq { + Some(eq) => unsafe { eq(a, b) }, + None => bytes_eq(0, layout.element.size), + }; + element + && layout.fields.iter().all(|field| match field.content_eq { + Some(eq) => unsafe { eq(a.add(field.offset), b.add(field.offset)) }, + None => bytes_eq(field.offset, field.size), + }) +} + +/// Hashes one record region of `layout` by content, with the byte fallback +/// of [`record_content_eq`]. +unsafe fn record_content_hash(layout: &Layout, ptr: *const u8, state: &mut dyn core::hash::Hasher) { + match layout.element.content_hash { + Some(hash) => unsafe { hash(ptr, state) }, + None => state.write(unsafe { std::slice::from_raw_parts(ptr, layout.element.size) }), + } + for field in &layout.fields { + match field.content_hash { + Some(hash) => unsafe { hash(ptr.add(field.offset), state) }, + None => state.write(unsafe { std::slice::from_raw_parts(ptr.add(field.offset), field.size) }), + } + } +} + +fn layout_shape_hash(layout: &Layout, state: &mut dyn core::hash::Hasher) { + state.write_u8(layout.depth); + state.write_usize(layout.fields.len()); + for field in &layout.fields { + state.write(field.name.as_bytes()); + state.write_u8(field.level); + state.write_usize(field.offset); + state.write_usize(field.size); + } +} + +impl PartialEq for GroupItem { + fn eq(&self, other: &Self) -> bool { + if self.layout != other.layout || self.len != other.len { + return false; + } + let stride = self.layout.lane_stride(); + // SAFETY: both sides hold `len` lanes of the shared layout. + (0..self.len).all(|lane| unsafe { record_content_eq(&self.layout, self.frames.add(lane * stride), other.frames.add(lane * stride)) }) + } +} + +impl graphene_hash::CacheHash for GroupItem { + fn cache_hash(&self, state: &mut H) { + layout_shape_hash(&self.layout, state); + state.write_usize(self.len); + let stride = self.layout.lane_stride(); + for lane in 0..self.len { + // SAFETY: `adopt` filled `len` lanes of `layout`. + unsafe { record_content_hash(&self.layout, self.frames.add(lane * stride), state) }; + } + } +} + +impl PartialEq for GroupContent { + fn eq(&self, other: &Self) -> bool { + match (self, other) { + (GroupContent::Run(a), GroupContent::Run(b)) => a == b, + (GroupContent::Stack(a), GroupContent::Stack(b)) => a == b, + _ => false, + } + } +} + +impl graphene_hash::CacheHash for GroupContent { + fn cache_hash(&self, state: &mut H) { + match self { + GroupContent::Run(item) => { + state.write_u8(0); + item.cache_hash(state); + } + GroupContent::Stack(children) => { + state.write_u8(1); + state.write_usize(children.len()); + for child in children { + child.cache_hash(state); + } + } + } + } +} + +impl PartialEq for Group { + fn eq(&self, other: &Self) -> bool { + self.row == other.row && self.content == other.content + } +} + +impl graphene_hash::CacheHash for Group { + fn cache_hash(&self, state: &mut H) { + state.write_u8(self.row.is_some() as u8); + if let Some(row) = &self.row { + row.cache_hash(state); + } + self.content.cache_hash(state); + } +} + #[cfg(test)] mod tests { use super::*; @@ -1572,6 +1768,8 @@ mod tests { align, read_erased: unread, repark: None, + content_hash: None, + content_eq: None, } } @@ -1635,6 +1833,53 @@ mod tests { } } + #[test] + fn the_element_write_pick_selects_the_content_glue_by_type() { + use super::{ElementWritePickHashed as _, ElementWritePickPlain as _}; + + #[derive(Clone)] + struct Opaque; + + let hashed = (&ElementWritePick::(std::marker::PhantomData)).element_write(); + assert!(hashed.content_hash.is_some() && hashed.content_eq.is_some()); + let plain = (&ElementWritePick::(std::marker::PhantomData)).element_write(); + assert!(plain.content_hash.is_none() && plain.content_eq.is_none()); + } + + #[test] + fn group_content_equality_and_hashing_read_through_the_park() { + use graphene_hash::CacheHash; + + let arena = crate::arena::Arena::new(4096).unwrap(); + let layout = Layout::default().with_writes(1, element_write_hashed::(), &[]); + let stride = layout.lane_stride(); + + let build = |labels: &[&str]| { + let mut buffer = vec![0u64; (labels.len() * stride).div_ceil(8)]; + let base = buffer.as_mut_ptr().cast::(); + for (lane, label) in labels.iter().enumerate() { + unsafe { write_element(base.add(lane * stride), label.to_string(), &arena) }.unwrap(); + } + let batch = unsafe { crate::node::RecordBatch::new(base.cast_const(), labels.len(), &layout) }; + GroupItem::adopt(batch, &arena).unwrap() + }; + let digest = |item: &GroupItem| { + let mut state = std::collections::hash_map::DefaultHasher::new(); + item.cache_hash(&mut state); + std::hash::Hasher::finish(&state) + }; + + // Each build parks its strings at fresh arena addresses, so equality + // and hashing must read the parked content, not the pointer bytes. + let a = build(&["x", "y"]); + let b = build(&["x", "y"]); + let c = build(&["x", "z"]); + assert_eq!(a, b); + assert_eq!(digest(&a), digest(&b)); + assert_ne!(a, c); + assert_ne!(digest(&a), digest(&c)); + } + #[test] fn translation_moves_fields_and_fills_census_defaults() { let source = Layout::default().with_writes(0, element_write::(), &[f64_field("length")]); diff --git a/node-graph/nodes/gcore/src/record.rs b/node-graph/nodes/gcore/src/record.rs index 345ff5ced6..4c33c26045 100644 --- a/node-graph/nodes/gcore/src/record.rs +++ b/node-graph/nodes/gcore/src/record.rs @@ -553,6 +553,8 @@ mod tests { align: 8, read_erased: ::read_erased, repark: None, + content_hash: None, + content_eq: None, }) .collect(); Layout::default().with_writes(0, core_types::record::element_write::(), &writes) @@ -568,6 +570,8 @@ mod tests { align: 8, read_erased: ::read_erased, repark: None, + content_hash: None, + content_eq: None, }) .collect(); Layout::default().with_writes(1, core_types::record::element_write::(), &writes)