diff --git a/core-relations/src/action/mod.rs b/core-relations/src/action/mod.rs index 0e35804c4..3e0f96b6a 100644 --- a/core-relations/src/action/mod.rs +++ b/core-relations/src/action/mod.rs @@ -760,7 +760,7 @@ impl<'a> ExecutionState<'a> { self.db.containers } - /// Predict and stage a sequence-backed container row. + /// Predict and stage a container row. /// /// The row is `key` followed by one fresh identity allocated from /// `counter`. Repeated keys within this execution reuse the first identity @@ -792,7 +792,7 @@ impl<'a> ExecutionState<'a> { identity } - /// Return a sequence-backed container row predicted by this execution. + /// Return a container row predicted by this execution. pub(crate) fn predicted_container_row( &self, container: ContainerValueId, diff --git a/core-relations/src/containers/mod.rs b/core-relations/src/containers/mod.rs index 40ecda60e..3a8288574 100644 --- a/core-relations/src/containers/mod.rs +++ b/core-relations/src/containers/mod.rs @@ -1,30 +1,22 @@ //! Support for containers //! -//! Containers behave a lot like base values. They are implemented differently because -//! their ids share a space with other Ids in the egraph and as a result, their ids need to be -//! sparse. -//! -//! This is a relatively "eagler" implementation of containers, reflecting egglog's current -//! semantics. One could imagine a variant of containers in which they behave more like egglog -//! functions than base values. +//! Each container type owns a variable-arity table keyed by its canonical flat +//! sequence. The table's one non-key column is the container identity, which +//! shares the ordinary e-graph id space and is therefore sparse. A type-erased +//! registry coordinates publication, rebuilding, and dependency scheduling +//! across those tables. use std::{ any::{Any, TypeId}, - hash::{Hash, Hasher}, + hash::Hash, ops::Deref, }; -use crate::numeric_id::{DenseIdMap, IdVec, NumericId, define_id}; -use crossbeam_queue::SegQueue; -use dashmap::SharedValue; -use rustc_hash::FxHasher; +use crate::numeric_id::{DenseIdMap, NumericId, define_id}; use crate::{ - ColumnId, CounterId, ExecutionState, Offset, Subset, SubsetRef, TableId, TaggedRowBuffer, - Value, WrappedTable, - common::{DashMap, HashMap, IndexSet, SubsetTracker}, - parallel, - parallel_heuristics::{parallelize_inter_container_op, parallelize_intra_container_op}, + BaseValues, CounterId, ExecutionState, Subset, SubsetRef, TableId, Value, WrappedTable, + common::{HashMap, IndexSet, SubsetTracker}, table_spec::{MaintenanceTable, Rebuilder, ValueRebuilder}, }; @@ -34,7 +26,6 @@ mod sequence; mod tests; use sequence::SequenceContainerEnv; -pub use sequence::SequenceContainerValue; define_id!( pub ContainerValueId, @@ -91,29 +82,20 @@ pub struct ContainerValues { data: DenseIdMap>, } -enum ContainerValueRefInner<'a, C> { - Legacy(Box + 'a>), - Sequence(C), -} - -/// A borrowed legacy container or an owned value decoded from a sequence. +/// An owned container decoded from its canonical sequence. /// -/// Sequence-backed lookups deliberately reconstruct the Rust container: this -/// is the slow compatibility path. Performance-sensitive primitives should +/// This private `Deref` facade preserves the existing public lookup API even +/// though lookups no longer borrow from shared container storage. +/// Performance-sensitive primitives should /// use [`ExecutionState::container_sequence`] and operate on the serialized /// values directly. -struct ContainerValueRef<'a, C> { - inner: ContainerValueRefInner<'a, C>, -} +struct ContainerValueRef(C); -impl Deref for ContainerValueRef<'_, C> { +impl Deref for ContainerValueRef { type Target = C; fn deref(&self) -> &Self::Target { - match &self.inner { - ContainerValueRefInner::Legacy(value) => value, - ContainerValueRefInner::Sequence(value) => value, - } + &self.0 } } @@ -132,7 +114,7 @@ impl Deref for ContainerValueRef<'_, C> { /// changing the `Vec` id. The row is now newly matchable, but seminaive will /// miss it unless the parent row is retimestamped. #[derive(Clone, Default)] -pub struct ContainerRebuildSummary { +pub(crate) struct ContainerRebuildSummary { changed: bool, // Container ids whose semantics changed in a way that may not produce a // fresh parent-row delta during ordinary table rebuild. @@ -140,11 +122,11 @@ pub struct ContainerRebuildSummary { } impl ContainerRebuildSummary { - pub fn changed(&self) -> bool { + pub(crate) fn changed(&self) -> bool { self.changed } - pub fn dirty_ids(&self) -> &IndexSet { + pub(crate) fn dirty_ids(&self) -> &IndexSet { &self.dirty_ids } @@ -163,22 +145,10 @@ impl ContainerRebuildSummary { } } -/// The storage implementation behind a type-erased container environment. -/// -/// Sequence environments are eligible to become independently scheduled -/// database maintenance targets. Legacy environments remain in the registry's -/// compatibility rebuild pass. -#[derive(Clone, Copy, Debug, Eq, PartialEq)] -pub(crate) enum ContainerBackend { - Legacy, - Sequence, -} - /// The shared union-find rebuild inputs for one container rebuild cycle. /// -/// Preparing this once lets the common sequence-table traversal and the -/// legacy-only compatibility pass share the same incremental delta without -/// advancing the subset tracker twice. +/// Preparing this once lets every container type share the same incremental +/// union-find delta without advancing the subset tracker more than once. pub(crate) struct ContainerRebuildContext<'a> { table: &'a WrappedTable, rebuilder: Box, @@ -228,38 +198,25 @@ impl ContainerValues { Some((id, &**self.data.get(id)?)) } - fn get_sequence_env(&self) -> Option<&SequenceContainerEnv> { + fn get_typed_env(&self) -> Option<&SequenceContainerEnv> { self.get_env::()? .1 .as_any() .downcast_ref::>() } - /// Return the globally allocated storage id for an already registered - /// legacy container type. - pub(crate) fn registered_type(&self) -> Option { - let (id, env) = self.get_env::()?; - assert!( - env.as_any().downcast_ref::>().is_some(), - "container type was already registered with a different backend" - ); - Some(id) - } - - /// Return the id of an already registered sequence environment. + /// Return the id of an already registered container environment. /// /// This check does not mutate the registry, so the database can distinguish /// an idempotent registration from a first registration before installing /// the corresponding maintenance-scheduler participant. - pub(crate) fn registered_sequence_type( - &self, - ) -> Option { + pub(crate) fn registered_type(&self) -> Option { let (id, env) = self.get_env::()?; assert!( env.as_any() .downcast_ref::>() .is_some(), - "container type was already registered with a different backend" + "container environment has the wrong concrete value type" ); Some(id) } @@ -269,47 +226,21 @@ impl ContainerValues { let Some((_, env)) = self.get_env::() else { return; }; - if let Some(env) = env.as_any().downcast_ref::>() { - for ent in env.to_id.iter() { - f(ent.key(), *ent.value()); - } - } else if let Some(env) = env.as_any().downcast_ref::>() { - env.for_each(&mut f); - } else { - unreachable!("container environment has the wrong concrete value type"); - } + env.as_any() + .downcast_ref::>() + .expect("container environment has the wrong concrete value type") + .for_each(&mut f); } /// Get the container associated with the value `val` in the database. The caller must know the /// type of the container. /// - /// The return type of this function may contain lock guards. Attempts to modify the contents - /// of the containers database may deadlock if the given guard has not been dropped. + /// This is the slow compatibility path: the returned facade owns a Rust + /// container reconstructed from its canonical sequence. pub fn get_val(&self, val: Value) -> Option + '_> { - let (_, env) = self.get_env::()?; - if let Some(env) = env.as_any().downcast_ref::>() { - return Some(ContainerValueRef { - inner: ContainerValueRefInner::Legacy(Box::new(env.get_container(val)?)), - }); - } - let env = env - .as_any() - .downcast_ref::>() - .expect("container environment has the wrong concrete value type"); - Some(ContainerValueRef { - inner: ContainerValueRefInner::Sequence(env.get_container(val)?), - }) - } - - fn get_owned(&self, value: Value) -> Option { - let (_, env) = self.get_env::()?; - if let Some(env) = env.as_any().downcast_ref::>() { - return Some(env.get_container(value)?.deref().clone()); - } - env.as_any() - .downcast_ref::>() - .expect("container environment has the wrong concrete value type") - .get_container(value) + Some(ContainerValueRef( + self.get_typed_env::()?.get_container(val)?, + )) } pub fn register_val( @@ -317,24 +248,15 @@ impl ContainerValues { container: C, exec_state: &mut ExecutionState, ) -> Value { - let (_, env) = self - .get_env::() - .expect("must register container type before registering a value"); - if let Some(env) = env.as_any().downcast_ref::>() { - env.get_or_insert(&container, exec_state) - } else { - env.as_any() - .downcast_ref::>() - .expect("container environment has the wrong concrete value type") - .get_or_insert(&container, exec_state) - } + self.get_typed_env::() + .expect("must register container type before registering a value") + .get_or_insert(&container, exec_state) } - /// Return the committed flat value sequence for a sequence-backed - /// container. This is the fast read path and performs no Rust-container - /// reconstruction. + /// Return the committed flat value sequence for a container. This is the + /// fast read path and performs no Rust-container reconstruction. pub fn get_sequence(&self, value: Value) -> Option<&[Value]> { - self.get_sequence_env::()?.get_values(value) + self.get_typed_env::()?.get_values(value) } /// Rebuild a single container value by remapping each contained value @@ -376,59 +298,6 @@ impl ContainerValues { Some(ContainerRebuildPlan { to_scan }) } - /// Rebuild only environments that still use the legacy DashMap storage. - /// - /// Sequence-backed environments are ordinary maintenance participants and - /// are rebuilt by the common dependency traversal. Keeping this pass - /// legacy-only means each migrated container immediately leaves the old - /// container-wide rebuild loop. - pub(crate) fn rebuild_legacy_pass( - &mut self, - context: &ContainerRebuildContext<'_>, - exec_state: &mut ExecutionState, - ) -> ContainerRebuildSummary { - let selected = self - .data - .iter() - .filter(|(_, env)| env.backend() == ContainerBackend::Legacy) - .count(); - if selected == 0 { - return ContainerRebuildSummary::default(); - } - - if parallelize_inter_container_op(selected) { - parallel::map_dense_id_map_mut(&mut self.data, |_, env| { - if env.backend() != ContainerBackend::Legacy { - return ContainerRebuildSummary::default(); - } - let mut exec_state = exec_state.clone(); - context.apply(&mut **env, &mut exec_state) - }) - .into_iter() - .fold(ContainerRebuildSummary::default(), |mut acc, summary| { - acc.extend(summary); - acc - }) - } else { - let mut summary = ContainerRebuildSummary::default(); - for (_, env) in self.data.iter_mut() { - if env.backend() == ContainerBackend::Legacy { - summary.extend(context.apply(&mut **env, exec_state)); - } - } - summary - } - } - - /// Finish a split container rebuild by propagating stable semantic changes - /// through all containing containers. - /// - /// Call this once, after every backend-specific pass has completed and all - /// temporarily removed environments have been restored. - pub(crate) fn finish_rebuild(&self, summary: &mut ContainerRebuildSummary) { - self.expand_dirty_id_closure(summary); - } - /// Add ancestor containers to the dirty-id set until it is transitively closed. /// /// A rebuild can change a container's semantics in place without changing @@ -459,38 +328,17 @@ impl ContainerValues { } } - /// Add a new container type to the given [`ContainerValue`] instance. - /// - /// Container types need a meaans of generating fresh ids (`id_counter`) along with a means of - /// merging conflicting ids (`merge_fn`). - pub fn register_type( - &mut self, - id: ContainerValueId, - id_counter: CounterId, - merge_fn: impl MergeFn + 'static, - ) -> ContainerValueId { - let id = self.container_ids.insert(TypeId::of::(), id); - self.data.get_or_insert(id, || { - Box::new(ContainerEnv::::new(Box::new(merge_fn), id_counter)) - }); - assert!( - self.data[id] - .as_any() - .downcast_ref::>() - .is_some(), - "container type was already registered with a different backend" - ); - id - } - /// Register a container type whose canonical representation is a /// variable-length sequence followed by one non-key identity value. - pub(crate) fn register_sequence_type( + /// + /// Database-level registration wraps this method so the environment also + /// becomes a maintenance-scheduler participant. + pub(crate) fn register_type( &mut self, id: ContainerValueId, id_counter: CounterId, merge_fn: impl MergeFn + 'static, - base_values: crate::BaseValues, + base_values: BaseValues, ) -> ContainerValueId { let id = self.container_ids.insert(TypeId::of::(), id); self.data.get_or_insert(id, || { @@ -506,44 +354,33 @@ impl ContainerValues { .as_any() .downcast_ref::>() .is_some(), - "container type was already registered with a different backend" + "container environment has the wrong concrete value type" ); id } - /// Return the storage backend for a registered type-erased environment. - #[cfg(test)] - pub(crate) fn env_backend(&self, id: ContainerValueId) -> ContainerBackend { - self.data[id].backend() - } - /// Return the current number of values in a registered environment. #[cfg(test)] pub(crate) fn env_len(&self, id: ContainerValueId) -> usize { self.data[id].len() } - /// Return the combined number of committed sequence-backed rows. - pub(crate) fn sequence_len(&self) -> usize { - self.data - .iter() - .filter(|(_, env)| env.backend() == ContainerBackend::Sequence) - .map(|(_, env)| env.len()) - .sum() - } - - /// Snapshot the ids of all sequence-backed environments. - pub(crate) fn sequence_env_ids(&self) -> Vec { - self.data - .iter() - .filter_map(|(id, env)| (env.backend() == ContainerBackend::Sequence).then_some(id)) - .collect() + /// Return the combined number of committed container rows. + pub(crate) fn container_len(&self) -> usize { + self.data.iter().map(|(_, env)| env.len()).sum() } /// Return the common maintenance surface for a container's backing /// sequence table. pub(crate) fn maintenance_table(&self, id: ContainerValueId) -> Option<&dyn MaintenanceTable> { - self.data.get(id)?.maintenance_table() + self.data + .get(id) + .map(|environment| &**environment as &dyn MaintenanceTable) + } + + /// Snapshot the ids of all registered container environments. + pub(crate) fn env_ids(&self) -> Vec { + self.data.iter().map(|(id, _)| id).collect() } pub(crate) fn take_env( @@ -570,26 +407,25 @@ impl ExecutionState<'_> { /// predicted earlier by the same execution. /// /// This is the slow compatibility path for primitive implementations. - /// Sequence-backed containers are reconstructed from their flat key. + /// Containers are reconstructed from their flat key. pub fn get_container(&self, value: Value) -> Option { - if let Some(env) = self.db.containers.get_sequence_env::() { - env.get_container_with_predictions(self, value) - } else { - self.db.containers.get_owned::(value) - } + self.db + .containers + .get_typed_env::()? + .get_container_with_predictions(self, value) } - /// Borrow the fast serialized payload for a sequence-backed container - /// visible to this execution. No Rust container is constructed. The - /// payload may include type-specific metadata, such as compact counts. + /// Borrow the fast serialized payload for a container visible to this + /// execution. No Rust container is constructed. The payload may include + /// type-specific metadata, such as compact counts. pub fn container_sequence(&self, value: Value) -> Option<&[Value]> { self.db .containers - .get_sequence_env::()? + .get_typed_env::()? .get_values_with_predictions(self, value) } - /// Intern an already serialized key for a sequence-backed container. + /// Intern an already serialized container key. /// /// This is the fast write path. Callers avoid constructing a Rust /// container but remain responsible for producing that type's canonical @@ -597,49 +433,62 @@ impl ExecutionState<'_> { pub fn register_container_sequence(&mut self, key: &[Value]) -> Value { let containers = self.db.containers; let env = containers - .get_sequence_env::() - .expect("container type is not registered with the sequence backend"); + .get_typed_env::() + .expect("container type is not registered"); env.get_or_insert_key(key, self) } } -/// A trait implemented by container types. +/// A container with a canonical flat sequence representation. +/// +/// The encoded sequence is the container's table key: two containers that are +/// semantically equal must encode identically, and decoding an encoded key +/// must recover the same container. The registry stores one non-key identity +/// value alongside that key. /// -/// Containers behave a lot like base values, but they include extra trait methods to support -/// rebuilding of container contents and merging containers that become equal after a rebuild pass -/// has taken place. +/// [`BaseValues`] is available to codecs that keep type-erased descriptors in +/// the base-value pool. Most containers do not need it. pub trait ContainerValue: Hash + Eq + Clone + Send + Sync + 'static { - /// Rebuild an additional container in place according the the given [`ValueRebuilder`]. - /// - /// If this method returns `false` then the container must not have been modified (i.e. it must - /// hash to the same value, and compare equal to a copy of itself before the call). - fn rebuild_contents(&mut self, rebuilder: &dyn ValueRebuilder) -> bool; + /// Append this container's canonical table key to `out`. + fn encode_sequence(&self, base_values: &BaseValues, out: &mut Vec); - /// Iterate over the contents of the container. - /// - /// Note that containers can be more structured than just a sequence of values. This iterator - /// is used to populate an index that in turn is used to speed up rebuilds. If a value in the - /// container is eligible for a rebuild and it is not mentioned by this iterator, the outer - /// container registry may skip rebuilding this container. - fn iter(&self) -> impl Iterator + '_; -} + /// Reconstruct the Rust value used by slow primitives and external APIs. + fn decode_sequence(sequence: &[Value], base_values: &BaseValues) -> Self; -pub(crate) trait DynamicContainerEnv: Any + dyn_clone::DynClone + Send + Sync { - fn as_any(&self) -> &dyn Any; - fn backend(&self) -> ContainerBackend; - fn len(&self) -> usize; + /// Return the fast primitive view of a serialized key. + /// + /// The view may include compact metadata needed by fast primitives. Use + /// [`ContainerValue::visit_sequence_values`] to identify the child values + /// that actually participate in congruence closure. + fn sequence_values(sequence: &[Value]) -> &[Value]; - /// Return the common maintenance interface when this environment is - /// sequence-backed. Legacy environments are intentionally absent from the - /// scheduler and return `None`. - fn maintenance_table(&self) -> Option<&dyn MaintenanceTable> { - None + /// Visit semantic child values used by the occurrence index and transitive + /// dirty-container discovery. + /// + /// Implementations with metadata or non-rebuildable lanes must override + /// this method so only values eligible for rebuilding are visited. + fn visit_sequence_values(sequence: &[Value], visitor: &mut dyn FnMut(Value)) { + for value in Self::sequence_values(sequence) { + visitor(*value); + } } - fn maintenance_table_mut(&mut self) -> Option<&mut dyn MaintenanceTable> { - None - } + /// Rebuild a serialized key into the initially empty `out` buffer. + /// + /// Returning `false` requires leaving `out` empty. Returning `true` + /// requires writing a canonical key, including any unchanged metadata. + fn rebuild_sequence( + sequence: &[Value], + base_values: &BaseValues, + rebuilder: &dyn ValueRebuilder, + out: &mut Vec, + ) -> bool; +} +pub(crate) trait DynamicContainerEnv: + Any + dyn_clone::DynClone + MaintenanceTable + Send + Sync +{ + fn as_any(&self) -> &dyn Any; fn apply_rebuild( &mut self, table: &WrappedTable, @@ -649,9 +498,8 @@ pub(crate) trait DynamicContainerEnv: Any + dyn_clone::DynClone + Send + Sync { ) -> ContainerRebuildSummary; /// Add ids for containers in this environment that contain any `values`. /// - /// This uses the container content index populated from - /// [`ContainerValue::iter`] and lets callers climb from dirty child ids to - /// all directly containing parent container ids. + /// This uses the container table's occurrence index and lets callers climb + /// from dirty child ids to all directly containing parent container ids. fn extend_containers_containing(&self, values: &IndexSet, out: &mut IndexSet); /// Rebuild the single container `value` by remapping each contained value /// through `remap`, returning the (possibly new) interned value, or `None` @@ -667,454 +515,6 @@ pub(crate) trait DynamicContainerEnv: Any + dyn_clone::DynClone + Send + Sync { // Implements `Clone` for `Box`. dyn_clone::clone_trait_object!(DynamicContainerEnv); -fn hash_container(container: &impl ContainerValue) -> u64 { - let mut hasher = FxHasher::default(); - container.hash(&mut hasher); - hasher.finish() -} - -#[derive(Clone)] -struct ContainerEnv { - merge_fn: Box, - counter: CounterId, - to_id: DashMap, - to_container: DashMap, - /// Map from a Value to the set of ids of containers that contain that value. - val_index: DashMap>, -} - -impl DynamicContainerEnv for ContainerEnv { - fn as_any(&self) -> &dyn Any { - self - } - - fn backend(&self) -> ContainerBackend { - ContainerBackend::Legacy - } - - fn len(&self) -> usize { - self.to_id.len() - } - - fn apply_rebuild( - &mut self, - table: &WrappedTable, - rebuilder: &dyn Rebuilder, - subset: Option, - exec_state: &mut ExecutionState, - ) -> ContainerRebuildSummary { - if let Some(subset) = subset - && incremental_rebuild( - subset.size(), - self.to_id.len(), - parallelize_intra_container_op(self.to_id.len()), - ) - { - return self.apply_rebuild_incremental( - table, - rebuilder, - exec_state, - subset, - rebuilder.hint_col().unwrap(), - ); - } - self.apply_rebuild_nonincremental(rebuilder, exec_state) - } - - fn extend_containers_containing(&self, values: &IndexSet, out: &mut IndexSet) { - for value in values { - if let Some(containers) = self.val_index.get(value) { - out.extend(containers.iter().copied()); - } - } - } - - fn rebuild_val_with( - &self, - value: Value, - exec_state: &mut ExecutionState, - remap: &(dyn Fn(Value) -> Value + Send + Sync), - ) -> Option { - // Clone out of the guard before re-interning to avoid deadlocking on - // the underlying map. - let mut container = self.get_container(value)?.clone(); - container.rebuild_contents(&ClosureRebuilder { remap }); - Some(self.get_or_insert(&container, exec_state)) - } -} - -impl ContainerEnv { - pub fn new(merge_fn: Box, counter: CounterId) -> Self { - Self { - merge_fn, - counter, - to_id: DashMap::default(), - to_container: DashMap::default(), - val_index: DashMap::default(), - } - } - - fn get_or_insert(&self, container: &C, exec_state: &mut ExecutionState) -> Value { - if let Some(value) = self.to_id.get(container) { - return *value; - } - - // Time to insert a new mapping. First, insert into `to_container`: the moment that we - // insert a new value into `to_id`, someone else can return it from another call to - // `get_or_insert` and then feed that value to `get_container`. - - let value = Value::from_usize(exec_state.inc_counter(self.counter)); - let target_map = self.to_id.determine_map(container); - // This assertion is here because in parallel rebuilding we use `to_container` to - // compute the intended shard for to_id, because we have a mutable borrow of - // `to_container` that means we cannot call `determine_map` on `to_id`. - debug_assert_eq!( - target_map, - self.to_container - .determine_shard(hash_container(container) as usize) - ); - self.to_container - .insert(value, (hash_container(container) as usize, target_map)); - - // Now insert into `to_id`, handling the case where a different thread is doing the same - // thing. - match self.to_id.entry(container.clone()) { - dashmap::Entry::Vacant(vac) => { - // Common case: insert the mapping in to_id and update the index. - vac.insert(value); - for val in container.iter() { - self.val_index.entry(val).or_default().insert(value); - } - value - } - dashmap::Entry::Occupied(occ) => { - // Someone inserted `container` into the mapping since we looked it up. Remove the - // mapping that we inserted into `to_container` (we won't use it), and instead - // return the "winning" value. - let res = *occ.get(); - std::mem::drop(occ); // drop the lock. - self.to_container.remove(&value); - res - } - } - } - - fn insert_owned(&self, container: C, value: Value, exec_state: &mut ExecutionState) -> Value { - let hc = hash_container(&container); - let target_map = self.to_id.determine_map(&container); - match self.to_id.entry(container) { - dashmap::Entry::Occupied(mut occ) => { - let result = (self.merge_fn)(exec_state, *occ.get(), value); - let old_val = *occ.get(); - if result != old_val { - self.to_container.remove(&old_val); - self.to_container.insert(result, (hc as usize, target_map)); - *occ.get_mut() = result; - for val in occ.key().iter() { - let mut index = self.val_index.entry(val).or_default(); - index.swap_remove(&old_val); - index.insert(result); - } - } - result - } - dashmap::Entry::Vacant(vacant_entry) => { - self.to_container.insert(value, (hc as usize, target_map)); - for val in vacant_entry.key().iter() { - self.val_index.entry(val).or_default().insert(value); - } - vacant_entry.insert(value); - value - } - } - } - - fn reinsert_incremental( - &self, - container: C, - old_id: Value, - rebuilt_id: Value, - container_changed: bool, - exec_state: &mut ExecutionState, - summary: &mut ContainerRebuildSummary, - ) { - if container_changed || rebuilt_id != old_id { - summary.note_change(); - } - if rebuilt_id != old_id { - // Parent rows will get a real delta from ordinary table rebuild, so - // we only need an explicit refresh when the outer id stayed stable. - self.to_container.remove(&old_id); - } - let actual = self.insert_owned(container, rebuilt_id, exec_state); - if container_changed && rebuilt_id == old_id && actual == old_id { - summary.note_dirty_id(old_id); - } - } - - fn apply_rebuild_incremental( - &mut self, - table: &WrappedTable, - rebuilder: &dyn Rebuilder, - exec_state: &mut ExecutionState, - to_scan: SubsetRef, - search_col: ColumnId, - ) -> ContainerRebuildSummary { - // NB: there is no parallel implementation as of now. - // - // Implementing one should be straightforward, but we should wait for a real benchmark that - // requires it. It's possible that incremental rebuilding will only be profitable when the - // total number of ids to rebuild is small, in which case the overhead of parallelism may - // not be worth it in the first place. - let mut summary = ContainerRebuildSummary::default(); - let mut buf = TaggedRowBuffer::new(1); - table.scan_project( - to_scan, - &[search_col], - Offset::new(0), - usize::MAX, - &[], - &mut buf, - ); - // For each value in the buffer, rebuild all containers that mention it. - let mut to_rebuild = IndexSet::::default(); - for (_, row) in buf.iter() { - to_rebuild.insert(row[0]); - let Some(ids) = self.val_index.get(&row[0]) else { - continue; - }; - to_rebuild.extend(&*ids); - } - for id in to_rebuild { - let Some((hc, target_map)) = self.to_container.get(&id).map(|x| *x) else { - continue; - }; - let shard_mut = self.to_id.shards_mut()[target_map].get_mut(); - let Some((mut container, _)) = - shard_mut.remove_entry(hc as u64, |(_, v)| *v.get() == id) - else { - continue; - }; - let rebuilt_id = rebuilder.rebuild_val(id); - let container_changed = container.rebuild_contents(rebuilder); - self.reinsert_incremental( - container, - id, - rebuilt_id, - container_changed, - exec_state, - &mut summary, - ); - } - summary - } - - fn apply_rebuild_nonincremental( - &mut self, - rebuilder: &dyn Rebuilder, - exec_state: &mut ExecutionState, - ) -> ContainerRebuildSummary { - if parallelize_inter_container_op(self.to_id.len()) { - return self.apply_rebuild_nonincremental_parallel(rebuilder, exec_state); - } - let mut summary = ContainerRebuildSummary::default(); - let mut to_reinsert = Vec::new(); - let shards = self.to_id.shards_mut(); - for shard in shards.iter_mut() { - let shard = shard.get_mut(); - // SAFETY: the iterator does not outlive `shard`. - for bucket in unsafe { shard.iter() } { - // SAFETY: the bucket is valid; we just got it from the iterator. - let (container, val) = unsafe { bucket.as_mut() }; - let old_val = *val.get(); - let new_val = rebuilder.rebuild_val(old_val); - let container_changed = container.rebuild_contents(rebuilder); - if !container_changed && new_val == old_val { - // Nothing changed about this entry. Leave it in place. - continue; - } - summary.note_change(); - if container_changed { - // The container changed. Remove both map entries then reinsert. - // SAFETY: This is a valid bucket. Furthermore, iterators remain valid if - // buckets they have already yielded have been removed. - let ((container, _), _) = unsafe { shard.remove(bucket) }; - self.to_container.remove(&old_val); - to_reinsert.push((container, new_val, new_val == old_val)); - } else { - // Just the value changed. Leave the container in place. - *val.get_mut() = new_val; - let prev = self.to_container.remove(&old_val).unwrap().1; - self.to_container.insert(new_val, prev); - } - } - } - for (container, val, stable_id) in to_reinsert { - let actual = self.insert_owned(container, val, exec_state); - // Refresh only when rebuild changed container semantics in place. - // If the outer id changed, ordinary table rebuild already creates a - // fresh parent-row delta for seminaive to follow. - if stable_id && actual == val { - summary.note_dirty_id(val); - } - } - summary - } - - fn apply_rebuild_nonincremental_parallel( - &mut self, - rebuilder: &dyn Rebuilder, - exec_state: &mut ExecutionState, - ) -> ContainerRebuildSummary { - // This is very similar to the serial variant. The main difference is that - // `to_reinsert` isn't a flat vector. It's instead a vector of queues - one per - // destination map shard. This lets us do a bulk insertion in parallel without having - // to grab a lock per container. - let mut to_reinsert = - IdVec::>::default(); - to_reinsert.resize_with(self.to_id.shards().len(), Default::default); - - let shards = self.to_id.shards_mut(); - let changed = parallel::map_mut(shards, |_, shard| { - let mut changed = false; - let shard = shard.get_mut(); - // SAFETY: the iterator does not outlive `shard`. - for bucket in unsafe { shard.iter() } { - // SAFETY: the bucket is valid; we just got it from the iterator. - let (container, val) = unsafe { bucket.as_mut() }; - let old_val = *val.get(); - let new_val = rebuilder.rebuild_val(old_val); - let container_changed = container.rebuild_contents(rebuilder); - if !container_changed && new_val == old_val { - // Nothing changed about this entry. Leave it in place. - continue; - } - changed = true; - if container_changed { - // The container changed. Remove both map entries then reinsert. - // SAFETY: This is a valid bucket. Furthermore, iterators remain valid if - // buckets they have already yielded have been removed. - let ((container, _), _) = unsafe { shard.remove(bucket) }; - self.to_container.remove(&old_val); - // Spooky: we're using `to_container` to determine the shard for - // `to_id`. We are assuming that the # shards determination is - // deterministic here. There is a debug assertion in `get_or_insert` - // that attempts to verify this. - let shard = self - .to_container - .determine_shard(hash_container(&container) as usize); - to_reinsert[shard].push((container, new_val, new_val == old_val)); - } else { - // Just the value changed. Leave the container in place. - *val.get_mut() = new_val; - let prev = self.to_container.remove(&old_val).unwrap().1; - self.to_container.insert(new_val, prev); - } - } - changed - }) - .into_iter() - .any(|changed| changed); - - let dirty_ids = SegQueue::new(); - parallel::for_each_mut(shards, |shard_id, shard| { - let mut exec_state = exec_state.clone(); - // This bit is a real slog. Once Dashmap updates from RawTable to HashTable for - // the underlying shard, this will get a little better. - // - // NB: We are probably leaving some paralellism on the floor with these calls - // to `to_container` and `val_index`. - let shard = shard.get_mut(); - let queue = &to_reinsert[shard_id]; - while let Some((container, val, stable_id)) = queue.pop() { - let hc = hash_container(&container); - let target_map = self.to_container.determine_shard(hc as usize); - match shard.find_or_find_insert_slot( - hc, - |(c, _)| c == &container, - |(c, _)| hash_container(c), - ) { - Ok(bucket) => { - // SAFETY: the bucket is valid; we just got it from the shard and - // we have not done any operations that can invalidate the bucket. - let (container, val_slot) = unsafe { bucket.as_mut() }; - let old_val = *val_slot.get(); - let result = (self.merge_fn)(&mut exec_state, old_val, val); - if result != old_val { - self.to_container.remove(&old_val); - self.to_container.insert(result, (hc as usize, target_map)); - *val_slot.get_mut() = result; - for val in container.iter() { - let mut index = self.val_index.entry(val).or_default(); - index.swap_remove(&old_val); - index.insert(result); - } - } - // As in the serial path, only same-id semantic - // changes need an explicit parent-row refresh. - if stable_id && result == val { - dirty_ids.push(val); - } - } - Err(slot) => { - self.to_container.insert(val, (hc as usize, target_map)); - for v in container.iter() { - self.val_index.entry(v).or_default().insert(val); - } - // SAFETY: We just got this slot from `find_or_find_insert_slot` - // and we have not mutated the map at all since then. - unsafe { - shard.insert_in_slot(hc, slot, (container, SharedValue::new(val))); - } - if stable_id { - dirty_ids.push(val); - } - } - } - } - }); - let mut summary = ContainerRebuildSummary::default(); - if changed { - summary.note_change(); - } - while let Some(value) = dirty_ids.pop() { - summary.note_dirty_id(value); - } - summary - } - - fn get_container(&self, value: Value) -> Option + '_> { - let (hc, target_map) = *self.to_container.get(&value)?; - let shard = &self.to_id.shards()[target_map]; - let read_guard = shard.read(); - let val_ptr: *const (C, _) = shard - .read() - .find(hc as u64, |(_, v)| *v.get() == value)? - .as_ptr(); - struct ValueDeref<'a, T, Guard> { - _guard: Guard, - data: &'a T, - } - - impl Deref for ValueDeref<'_, T, Guard> { - type Target = T; - - fn deref(&self) -> &T { - self.data - } - } - - Some(ValueDeref { - _guard: read_guard, - // SAFETY: the value will remain valid for as long as `read_guard` is in scope. - data: unsafe { - let unwrapped: &(C, _) = &*val_ptr; - &unwrapped.0 - }, - }) - } -} - fn incremental_rebuild(uf_size: usize, table_size: usize, parallel: bool) -> bool { if parallel { table_size > 1000 && uf_size * 512 <= table_size @@ -1125,7 +525,7 @@ fn incremental_rebuild(uf_size: usize, table_size: usize, parallel: bool) -> boo /// A [`ValueRebuilder`] that remaps individual values through a caller-supplied /// closure. Used by [`ContainerValues::rebuild_val_with`] to rebuild a single -/// container against an explicit value mapping rather than a backend union-find. +/// container against an explicit value mapping rather than the database union-find. struct ClosureRebuilder<'a> { remap: &'a (dyn Fn(Value) -> Value + Send + Sync), } diff --git a/core-relations/src/containers/sequence.rs b/core-relations/src/containers/sequence.rs index 950a06efc..acef1ed29 100644 --- a/core-relations/src/containers/sequence.rs +++ b/core-relations/src/containers/sequence.rs @@ -24,62 +24,10 @@ use crate::{ }; use super::{ - ClosureRebuilder, ContainerBackend, ContainerRebuildSummary, ContainerValue, ContainerValueId, + ClosureRebuilder, ContainerRebuildSummary, ContainerValue, ContainerValueId, DynamicContainerEnv, MergeFn, incremental_rebuild, }; -/// A container with a canonical flat sequence representation. -/// -/// The default rebuild path is intentionally the slow compatibility path: it -/// deserializes the sequence, invokes [`ContainerValue::rebuild_contents`], -/// and serializes the result. Implementations may override -/// [`SequenceContainerValue::rebuild_sequence`] to transform the flat values -/// directly. -pub trait SequenceContainerValue: ContainerValue { - /// Append the canonical serialized key to `out`. - fn encode_sequence(&self, base_values: &BaseValues, out: &mut Vec); - - /// Reconstruct the Rust container used by slow primitives and external - /// APIs. - fn decode_sequence(sequence: &[Value], base_values: &BaseValues) -> Self; - - /// Return the fast primitive view of the serialized key. - /// - /// Metadata used only to distinguish container modes belongs outside this - /// slice. Most containers return their contained values directly; compact - /// encodings may return an encoded payload and override - /// [`SequenceContainerValue::visit_sequence_values`] for dependency - /// discovery. - fn sequence_values(sequence: &[Value]) -> &[Value]; - - /// Visit semantic child values used for transitive dirty-container - /// discovery. The default visits the fast primitive view directly. - fn visit_sequence_values(sequence: &[Value], visitor: &mut dyn FnMut(Value)) { - for value in Self::sequence_values(sequence) { - visitor(*value); - } - } - - /// Rebuild a serialized key into the initially empty `out` buffer. - /// - /// Returning `false` requires leaving `out` empty. The default implements - /// the slow deserialize/rebuild/serialize round trip. - fn rebuild_sequence( - sequence: &[Value], - base_values: &BaseValues, - rebuilder: &dyn ValueRebuilder, - out: &mut Vec, - ) -> bool { - let mut container = Self::decode_sequence(sequence, base_values); - if container.rebuild_contents(rebuilder) { - container.encode_sequence(base_values, out); - true - } else { - false - } - } -} - #[derive(Clone)] struct SequenceCodec { base_values: BaseValues, @@ -90,7 +38,7 @@ struct SequenceCodec { marker: PhantomData C>, } -impl SequenceCodec { +impl SequenceCodec { fn new(base_values: BaseValues) -> Self { Self { base_values, @@ -154,7 +102,7 @@ pub(super) struct SequenceContainerEnv { codec: SequenceCodec, } -impl SequenceContainerEnv { +impl SequenceContainerEnv { pub(super) fn new( id: ContainerValueId, merge: Box, @@ -367,22 +315,6 @@ impl DynamicContainerEnv for SequenceContainerEnv { self } - fn backend(&self) -> ContainerBackend { - ContainerBackend::Sequence - } - - fn len(&self) -> usize { - self.table.len() - } - - fn maintenance_table(&self) -> Option<&dyn MaintenanceTable> { - Some(self) - } - - fn maintenance_table_mut(&mut self) -> Option<&mut dyn MaintenanceTable> { - Some(self) - } - fn apply_rebuild( &mut self, table: &crate::WrappedTable, @@ -486,22 +418,12 @@ mod tests { table_spec::WrappedTableRef, }; - use super::{DynamicContainerEnv, SequenceContainerEnv, SequenceContainerValue}; + use super::{DynamicContainerEnv, SequenceContainerEnv}; #[derive(Clone, Debug, Eq, Hash, PartialEq)] struct TestSequence(Vec); impl ContainerValue for TestSequence { - fn rebuild_contents(&mut self, rebuilder: &dyn ValueRebuilder) -> bool { - rebuilder.rebuild_slice(&mut self.0) - } - - fn iter(&self) -> impl Iterator + '_ { - self.0.iter().copied() - } - } - - impl SequenceContainerValue for TestSequence { fn encode_sequence(&self, _base_values: &crate::BaseValues, out: &mut Vec) { out.extend_from_slice(&self.0); } diff --git a/core-relations/src/containers/tests.rs b/core-relations/src/containers/tests.rs index 0ec1c8515..6c96e69d1 100644 --- a/core-relations/src/containers/tests.rs +++ b/core-relations/src/containers/tests.rs @@ -12,48 +12,19 @@ use std::{ }; use crate::numeric_id::NumericId; -use egglog_concurrency::Notification; use crate::{ - ColumnId, Database, DisplacedTable, ExecutionState, Rebuilder, RowId, SequenceContainerValue, - SortedWritesTable, Value, ValueRebuilder, row_buffer::RowBuffer, table_spec::WrappedTableRef, -}; - -use super::{ - ContainerBackend, ContainerEnv, ContainerRebuildSummary, ContainerValue, hash_container, + ColumnId, ContainerValue, Database, DisplacedTable, SortedWritesTable, Value, ValueRebuilder, }; #[derive(Hash, PartialEq, Eq, Clone, Debug)] struct VecContainer(Vec); -#[derive(Hash, PartialEq, Eq, Clone, Debug)] -struct LegacyVecContainer(Vec); - fn cont(values: [usize; N]) -> VecContainer { VecContainer(values.iter().map(|&v| Value::from_usize(v)).collect()) } impl ContainerValue for VecContainer { - fn rebuild_contents(&mut self, rebuilder: &dyn ValueRebuilder) -> bool { - rebuilder.rebuild_slice(&mut self.0) - } - - fn iter(&self) -> impl Iterator + '_ { - self.0.iter().copied() - } -} - -impl ContainerValue for LegacyVecContainer { - fn rebuild_contents(&mut self, rebuilder: &dyn ValueRebuilder) -> bool { - rebuilder.rebuild_slice(&mut self.0) - } - - fn iter(&self) -> impl Iterator + '_ { - self.0.iter().copied() - } -} - -impl SequenceContainerValue for VecContainer { fn encode_sequence(&self, _base_values: &crate::BaseValues, out: &mut Vec) { out.extend_from_slice(&self.0); } @@ -82,233 +53,28 @@ impl SequenceContainerValue for VecContainer { } } -/// A tiny rebuilder used to isolate outer-id canonicalization from inner -/// container rewrites in the unit tests below. -struct FakeRebuilder { - old_outer_id: Option, - new_outer_id: Option, - old_inner_val: Option, - new_inner_val: Option, -} - -impl ValueRebuilder for FakeRebuilder { - fn rebuild_val(&self, val: Value) -> Value { - match (self.old_outer_id, self.new_outer_id) { - (Some(old), Some(new)) if val == old => new, - _ => val, - } - } - - fn rebuild_slice(&self, vals: &mut [Value]) -> bool { - let mut changed = false; - for val in vals { - if let (Some(old), Some(new)) = (self.old_inner_val, self.new_inner_val) - && *val == old - { - *val = new; - changed = true; - } - } - changed - } -} - -// Also exercised via the table-level rebuild path, so it implements the full -// `Rebuilder`; that path only calls the value-level methods for containers. -impl Rebuilder for FakeRebuilder { - fn hint_col(&self) -> Option { - None - } - - fn rebuild_buf( - &self, - _buf: &RowBuffer, - _start: RowId, - _end: RowId, - _out: &mut crate::TaggedRowBuffer, - _exec_state: &mut ExecutionState, - ) { - unreachable!("FakeRebuilder does not support rebuild_buf") - } - - fn rebuild_subset( - &self, - _other: WrappedTableRef, - _subset: crate::SubsetRef, - _out: &mut crate::TaggedRowBuffer, - _exec_state: &mut ExecutionState, - ) { - unreachable!("FakeRebuilder does not support rebuild_subset") - } -} - -#[test] -fn racing_inserts() { - let mut db = Database::new(); - let counter = db.add_counter(); - let db = Arc::new(db); - let start = Arc::new(Notification::new()); - let env = Arc::new(ContainerEnv::::new( - Box::new(|_state, v1, v2| { - assert_eq!(v1, v2, "this test shouldn't merge anything"); - v1 - }), - counter, - )); - let threads = (0..10) - .map(|_| { - let start = start.clone(); - let env = env.clone(); - let db = db.clone(); - std::thread::spawn(move || { - db.with_execution_state(|es| { - start.wait(); - env.get_or_insert(&cont([1, 2, 3]), es) - }) - }) - }) - .collect::>(); - start.notify(); - let results = Vec::from_iter(threads.into_iter().map(|t| t.join().unwrap())); - - for result in &results { - assert_eq!( - &*env.get_container(*result).unwrap_or_else(|| { - panic!("container {result:?} not found"); - }), - &cont([1, 2, 3]) - ); - } - assert!( - results.windows(2).all(|w| w[0] == w[1]), - "all containers should be the same, got {results:?}" - ); -} - #[test] -fn incremental_reinsert_canonicalizes_displaced_outer_id() { +fn type_erased_registry_reports_container_lengths() { let mut db = Database::new(); - let counter = db.add_counter(); - let mut env = ContainerEnv::::new( - Box::new(|_state, v1, v2| { - assert_eq!(v1, v2, "this test shouldn't merge anything"); - v1 - }), + let counter = db.add_reservable_counter(8); + let container_id = db.register_container_type::( counter, - ); - let container = cont([1, 2, 3]); - - db.with_execution_state(|es| { - let old_id = env.get_or_insert(&container, es); - let new_id = Value::from_usize(old_id.index() + 1000); - let hc = hash_container(&container); - let target_map = env.to_id.determine_map(&container); - let shard_mut = env.to_id.shards_mut()[target_map].get_mut(); - let (container, _) = shard_mut - .remove_entry(hc as u64, |(_, v)| *v.get() == old_id) - .expect("container should be present before reinsertion"); - - let mut summary = ContainerRebuildSummary::default(); - env.reinsert_incremental(container, old_id, new_id, false, es, &mut summary); - - assert!(summary.changed()); - assert!(summary.dirty_ids().is_empty()); - assert!(env.get_container(old_id).is_none()); - assert_eq!(&*env.get_container(new_id).unwrap(), &cont([1, 2, 3])); - }); -} - -#[test] -fn nonincremental_dirty_ids_only_include_stable_ids() { - let mut db = Database::new(); - let counter = db.add_counter(); - let old_inner = Value::from_usize(1); - let new_inner = Value::from_usize(2); - - let run_case = |outer_id_changes: bool| { - let mut env = ContainerEnv::::new( - Box::new(|_state, v1, v2| { - assert_eq!(v1, v2, "this test shouldn't merge anything"); - v1 - }), - counter, - ); - db.with_execution_state(|es| { - let old_id = env.get_or_insert(&VecContainer(vec![old_inner]), es); - let new_id = if outer_id_changes { - Value::from_usize(old_id.index() + 1000) - } else { - old_id - }; - let rebuilder = FakeRebuilder { - old_outer_id: outer_id_changes.then_some(old_id), - new_outer_id: outer_id_changes.then_some(new_id), - old_inner_val: Some(old_inner), - new_inner_val: Some(new_inner), - }; - - let summary = env.apply_rebuild_nonincremental(&rebuilder, es); - assert!(summary.changed()); - if outer_id_changes { - assert!(summary.dirty_ids().is_empty()); - assert!(env.get_container(old_id).is_none()); - assert_eq!( - &*env.get_container(new_id).unwrap(), - &VecContainer(vec![new_inner]) - ); - } else { - assert_eq!( - summary.dirty_ids().iter().copied().collect::>(), - vec![old_id] - ); - assert_eq!( - &*env.get_container(old_id).unwrap(), - &VecContainer(vec![new_inner]) - ); - } - }); - }; - - run_case(false); - run_case(true); -} - -#[test] -fn type_erased_backend_metadata_reports_sequence_lengths() { - let mut db = Database::new(); - let legacy_counter = db.add_counter(); - let sequence_counter = db.add_reservable_counter(8); - let legacy_id = db - .register_container_type::(legacy_counter, |_state, left, right| { - left.min(right) - }); - let sequence_id = db.register_sequence_container_type::( - sequence_counter, |_state, left, right| left.min(right), iter::empty(), iter::empty(), ); - assert_eq!( - db.container_values().env_backend(legacy_id), - ContainerBackend::Legacy - ); - assert_eq!( - db.container_values().env_backend(sequence_id), - ContainerBackend::Sequence - ); - assert_eq!(db.container_values().sequence_env_ids(), vec![sequence_id]); + assert_eq!(db.container_values().env_ids(), vec![container_id]); db.with_execution_state(|state| { let containers = state.container_values(); - containers.register_val(LegacyVecContainer(vec![Value::from_usize(1)]), state); containers.register_val(VecContainer(vec![Value::from_usize(2)]), state); }); - assert_eq!(db.container_values().env_len(legacy_id), 1); - assert_eq!(db.container_values().env_len(sequence_id), 0); + assert_eq!(db.container_values().env_len(container_id), 0); assert!(db.merge_all()); - assert_eq!(db.container_values().env_len(sequence_id), 1); + assert_eq!(db.container_values().env_len(container_id), 1); + assert_eq!(db.container_values().container_len(), 1); } #[test] @@ -320,8 +86,12 @@ fn relations_and_containers_share_one_table_id_namespace() { iter::empty(), ); let counter = db.add_reservable_counter(8); - let container = - db.register_container_type::(counter, |_state, left, right| left.min(right)); + let container = db.register_container_type::( + counter, + |_state, left, right| left.min(right), + iter::empty(), + iter::empty(), + ); let second_relation = db.add_table( SortedWritesTable::new(1, 1, None, vec![], Box::new(|_, _, _, _| false)), iter::empty(), @@ -332,10 +102,7 @@ fn relations_and_containers_share_one_table_id_namespace() { assert_eq!(crate::TableId::from(container).index(), 1); assert_eq!(second_relation.index(), 2); assert_eq!(db.next_table_id().index(), 3); - assert_eq!( - db.container_values().env_backend(container), - ContainerBackend::Legacy - ); + assert_eq!(db.container_values().env_ids(), [container]); } #[test] @@ -345,7 +112,7 @@ fn shared_rebuild_defers_relations_until_nested_container_collisions_converge() let collision_merges = Arc::new(AtomicUsize::new(0)); let collision_merges_for_callback = Arc::clone(&collision_merges); let counter = db.add_reservable_counter(8); - db.register_sequence_container_type::( + db.register_container_type::( counter, move |state, left, right| { assert!( @@ -446,7 +213,7 @@ fn shared_rebuild_defers_relations_until_nested_container_collisions_converge() fn containers_round_trip_predictions_through_database_merge() { let mut db = Database::new(); let counter = db.add_reservable_counter(8); - db.register_sequence_container_type::( + db.register_container_type::( counter, |_state, left, right| left.min(right), iter::empty(), @@ -492,10 +259,10 @@ fn containers_round_trip_predictions_through_database_merge() { } #[test] -fn cloned_databases_have_independent_sequence_notifications() { +fn cloned_databases_have_independent_container_notifications() { let mut original = Database::new(); let counter = original.add_reservable_counter(8); - original.register_sequence_container_type::( + original.register_container_type::( counter, |_state, left, right| left.min(right), iter::empty(), @@ -527,7 +294,7 @@ fn cloned_databases_have_independent_sequence_notifications() { } #[test] -fn sequence_merge_writes_participate_in_simple_fixed_point() { +fn container_merge_writes_participate_in_simple_fixed_point() { let mut db = Database::new(); let sink = db.add_table( SortedWritesTable::new( @@ -549,13 +316,13 @@ fn sequence_merge_writes_participate_in_simple_fixed_point() { iter::empty(), ); let counter = db.add_reservable_counter(8); - db.register_sequence_container_type::( + db.register_container_type::( counter, move |state, left, right| { assert_eq!( state.get_table(gate).len(), 1, - "the simple scheduler must honor sequence read dependencies" + "the simple scheduler must honor container read dependencies" ); state.stage_insert(sink, &[left, right]); left.min(right) @@ -582,7 +349,7 @@ fn sequence_merge_writes_participate_in_simple_fixed_point() { } #[test] -fn sequence_merge_writes_participate_in_dependency_aware_fixed_point() { +fn container_merge_writes_participate_in_dependency_aware_fixed_point() { let mut db = Database::new(); let sink = db.add_table( SortedWritesTable::new( @@ -609,13 +376,13 @@ fn sequence_merge_writes_participate_in_dependency_aware_fixed_point() { let first_bystander = add_flag_table(); let second_bystander = add_flag_table(); let counter = db.add_reservable_counter(8); - db.register_sequence_container_type::( + db.register_container_type::( counter, move |state, left, right| { assert_eq!( state.get_table(gate).len(), 1, - "the sequence merge must run after its read dependency" + "the container merge must run after its read dependency" ); state.stage_insert(sink, &[left, right]); left.min(right) @@ -640,16 +407,16 @@ fn sequence_merge_writes_participate_in_dependency_aware_fixed_point() { }); assert_ne!(first, second); - // The sequence environment plus the three ordinary tables above force + // The container environment plus the three ordinary tables above force // `merge_all` through its dependency-aware (four-or-more participant) - // path. The sequence merge then notifies the previously inactive sink, + // path. The container merge then notifies the previously inactive sink, // which must be consumed by the same call's fixed-point loop. assert!(db.merge_all()); assert_eq!(db.get_table(sink).len(), 1); } #[test] -fn relation_merge_can_depend_on_nonzero_level_sequence_container() { +fn relation_merge_can_depend_on_nonzero_level_container() { let mut db = Database::new(); let gate = db.add_table( SortedWritesTable::new(1, 1, None, vec![], Box::new(|_, _, _, _| false)), @@ -662,7 +429,7 @@ fn relation_merge_can_depend_on_nonzero_level_sequence_container() { iter::empty(), ); let counter = db.add_reservable_counter(8); - let container = db.register_sequence_container_type::( + let container = db.register_container_type::( counter, |_state, left, right| left.min(right), [gate], @@ -682,7 +449,7 @@ fn relation_merge_can_depend_on_nonzero_level_sequence_container() { assert_eq!( state.container_sequence::(winner), Some(expected.0.as_slice()), - "the relation merge must run after its sequence dependency" + "the relation merge must run after its container dependency" ); if current[1] == winner { false @@ -723,7 +490,7 @@ fn relation_merge_can_depend_on_nonzero_level_sequence_container() { fn same_stratum_relation_merge_can_read_active_container() { let mut db = Database::new(); let counter = db.add_reservable_counter(8); - db.register_sequence_container_type::( + db.register_container_type::( counter, |_state, left, right| left.min(right), iter::empty(), @@ -813,12 +580,12 @@ fn failed_table_dependency_registration_does_not_mutate_database() { } #[test] -fn failed_sequence_dependency_registration_can_be_retried() { +fn failed_container_dependency_registration_can_be_retried() { let mut db = Database::new(); let counter = db.add_reservable_counter(8); let missing_table = db.next_table_id(); let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - db.register_sequence_container_type::( + db.register_container_type::( counter, |_state, left, right| left.min(right), [missing_table], @@ -828,7 +595,7 @@ fn failed_sequence_dependency_registration_can_be_retried() { assert!(result.is_err()); let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - db.register_sequence_container_type::( + db.register_container_type::( counter, |_state, left, right| left.min(right), iter::empty(), @@ -837,13 +604,13 @@ fn failed_sequence_dependency_registration_can_be_retried() { })); assert!(result.is_err()); - let container = db.register_sequence_container_type::( + let container = db.register_container_type::( counter, |_state, left, right| left.min(right), iter::empty(), iter::empty(), ); - let duplicate = db.register_sequence_container_type::( + let duplicate = db.register_container_type::( counter, |_state, left, right| left.max(right), [missing_table], diff --git a/core-relations/src/dependency_graph.rs b/core-relations/src/dependency_graph.rs index ddab3c7eb..174fad4d9 100644 --- a/core-relations/src/dependency_graph.rs +++ b/core-relations/src/dependency_graph.rs @@ -43,6 +43,12 @@ impl DependencyGraph { self.to_level.insert(participant, level); } + /// Return the globally unique id that will be assigned to the next + /// relation or container table registered with the database. + pub(crate) fn next_id(&self) -> TableId { + self.to_level.next_id() + } + pub(crate) fn contains(&self, participant: TableId) -> bool { self.to_level.contains_key(participant) } @@ -85,16 +91,13 @@ mod tests { } #[test] - fn participant_ids_may_have_legacy_storage_holes() { + fn participant_ids_are_allocated_from_the_dependency_graph() { let mut graph = DependencyGraph::default(); - graph.add_participant(TableId::new(3), [], []); - graph.add_participant(TableId::new(7), [TableId::new(3)], []); + assert_eq!(graph.next_id(), TableId::new(0)); - assert!(!graph.contains(TableId::new(0))); - assert!(graph.contains(TableId::new(3))); - assert_eq!( - graph.participants().collect::>(), - [TableId::new(3), TableId::new(7)] - ); + graph.add_participant(graph.next_id(), [], []); + assert_eq!(graph.next_id(), TableId::new(1)); + assert!(graph.contains(TableId::new(0))); + assert_eq!(graph.participants().collect::>(), [TableId::new(0)]); } } diff --git a/core-relations/src/free_join/mod.rs b/core-relations/src/free_join/mod.rs index 100c58ed3..f0a3e7014 100644 --- a/core-relations/src/free_join/mod.rs +++ b/core-relations/src/free_join/mod.rs @@ -23,7 +23,7 @@ use crate::{ }, containers::{ ContainerRebuildContext, ContainerRebuildPlan, ContainerRebuildSummary, ContainerValue, - DynamicContainerEnv, SequenceContainerValue, + DynamicContainerEnv, }, dependency_graph::DependencyGraph, hash_index::{ColumnIndex, Index, IndexBase}, @@ -387,10 +387,6 @@ pub struct Database { pub(crate) counters: Counters, pub(crate) external_functions: ExternalFunctions, container_values: ContainerValues, - /// Next id in the storage namespace shared by relations and both - /// container backends. Legacy containers deliberately consume an id - /// without becoming dependency-graph participants. - next_storage_id: usize, /// Participants modified since the last call to [`Database::merge_all`]. notification_list: NotificationList, // Tracks relative dependencies between maintenance participants. @@ -441,9 +437,7 @@ impl OwnedParticipant { fn maintenance_table_mut(&mut self) -> &mut dyn MaintenanceTable { match self { OwnedParticipant::Relation { info, .. } => &mut info.table, - OwnedParticipant::Container { env, .. } => env - .maintenance_table_mut() - .expect("scheduled containers must use sequence storage"), + OwnedParticipant::Container { env, .. } => &mut **env, } } @@ -498,7 +492,6 @@ impl Clone for Database { counters: self.counters.clone(), external_functions: self.external_functions.clone(), container_values: self.container_values.clone(), - next_storage_id: self.next_storage_id, notification_list, deps: self.deps.clone(), base_values: self.base_values.clone(), @@ -550,31 +543,7 @@ impl Database { &mut self.container_values } - /// Register a legacy container in the database-wide storage namespace. - /// - /// Legacy environments keep their existing compatibility rebuild loop and - /// are not maintenance-scheduler participants. Allocating their ids here - /// nevertheless prevents a later relation or sequence container from - /// reusing the same physical slot while both backends coexist. - pub fn register_container_type( - &mut self, - id_counter: CounterId, - merge_fn: impl Fn(&mut ExecutionState, Value, Value) -> Value + Clone + Send + Sync + 'static, - ) -> crate::ContainerValueId { - if let Some(container) = self.container_values.registered_type::() { - return container; - } - - let participant = self.allocate_storage_id(); - let container = crate::ContainerValueId::from_table_id(participant); - let registered = self - .container_values - .register_type::(container, id_counter, merge_fn); - assert_eq!(registered, container); - container - } - - /// Register a sequence-backed container as a normal database maintenance + /// Register a container as a normal database maintenance /// participant. Its packed rows remain outside the fixed-schema query /// table interface, but merge notifications and dependency ordering share /// the same scheduler namespace as relation tables. @@ -584,15 +553,15 @@ impl Database { /// return the existing [`crate::ContainerValueId`]. /// /// Every read and write dependency must already be registered in this - /// database. Container read dependencies must use the sequence backend. - pub fn register_sequence_container_type( + /// database. + pub fn register_container_type( &mut self, id_counter: CounterId, merge_fn: impl Fn(&mut ExecutionState, Value, Value) -> Value + Clone + Send + Sync + 'static, read_deps: impl IntoIterator, write_deps: impl IntoIterator, ) -> crate::ContainerValueId { - if let Some(container) = self.container_values.registered_sequence_type::() { + if let Some(container) = self.container_values.registered_type::() { return container; } @@ -601,15 +570,12 @@ impl Database { self.validate_dependencies(&read_deps, "read"); self.validate_dependencies(&write_deps, "write"); - let base_values = self.base_values.clone(); - let participant = self.allocate_storage_id(); + let participant = self.deps.next_id(); let container = crate::ContainerValueId::from_table_id(participant); - let registered = self.container_values.register_sequence_type::( - container, - id_counter, - merge_fn, - base_values, - ); + let base_values = self.base_values.clone(); + let registered = + self.container_values + .register_type::(container, id_counter, merge_fn, base_values); assert_eq!(registered, container); self.deps .add_participant(participant, read_deps, write_deps); @@ -625,29 +591,22 @@ impl Database { } } - fn allocate_storage_id(&mut self) -> TableId { - let id = TableId::from_usize(self.next_storage_id); - self.next_storage_id = self - .next_storage_id - .checked_add(1) - .expect("database storage id space exhausted"); - assert!(!id.is_dummy(), "database storage id space exhausted"); - id - } - /// Apply one database-maintenance round using the value-level rebuild /// encoded by `func_id`. /// - /// Sequence-backed containers use the same notification/dependency - /// scheduler and rebuild traversal as relation tables. Unmigrated legacy - /// environments retain their compatibility pass, but consume the same - /// prepared union-find delta. Both kinds of container rebuild before the - /// publication barrier because canonicalizing their keys can create new - /// identities in the rebuild source. + /// Fixed- and variable-arity tables share the same dependency-ordered + /// participant traversal. Variable-arity container tables form the first + /// batch because canonicalizing their keys can create new identities in the + /// rebuild source. Publishing that batch before rebuilding fixed-schema + /// relations is a correctness barrier: a relation whose keys collapse must + /// see any simultaneously-collapsing container values before invoking its + /// merge policy. If publication changes the source, relation rebuilding is + /// deferred to the caller's next outer `apply_rebuild` round. This handles + /// nested-container collision chains without adding a container-only inner + /// fixpoint; both batches use the common participant implementation. /// - /// If publication changes the source, relation rebuilding is deferred to - /// the caller's next outer round. Stable container identities additionally - /// require the post-publication dirty-parent refresh below. + /// Stable container identities additionally require the post-publication + /// dirty-parent refresh below. pub fn apply_rebuild( &mut self, func_id: TableId, @@ -656,7 +615,7 @@ impl Database { ) -> bool { let containers = self .container_values - .sequence_env_ids() + .env_ids() .into_iter() .map(TableId::from) .collect::>(); @@ -672,26 +631,6 @@ impl Database { let mut rebuild = self.rebuild_batch(containers, func_id, next_ts, container_plan.as_ref()); - // Legacy environments are intentionally outside the common scheduler, - // but they must observe exactly the same incremental UF subset as the - // sequence participants. Keep their old pass until the final legacy - // backend is migrated. - if let Some(plan) = container_plan.as_ref() { - let mut container_values = mem::take(&mut self.container_values); - let (legacy, staged) = { - let source = &self.tables[func_id].table; - let context = plan - .context(source) - .expect("a prepared container rebuild plan must remain applicable"); - self.with_execution_state_tracked(|state| { - container_values.rebuild_legacy_pass(&context, state) - }) - }; - self.container_values = container_values; - rebuild.changed |= legacy.changed() || staged; - rebuild.containers.extend(legacy); - } - // A container collision can add an identity union to `func_id`. // Publish it before constructing relation rebuilders or mutation // buffers; otherwise a relation can collapse its key while retaining @@ -722,7 +661,7 @@ impl Database { rebuild: &mut ParticipantRebuild, ) { self.container_values - .finish_rebuild(&mut rebuild.containers); + .expand_dirty_id_closure(&mut rebuild.containers); let dirty_ids = rebuild .containers .dirty_ids() @@ -1019,7 +958,7 @@ impl Database { }); size_estimate += info.table.len(); } - size_estimate += self.container_values.sequence_len(); + size_estimate += self.container_values.container_len(); self.total_size_estimate = size_estimate; ever_changed } @@ -1189,7 +1128,7 @@ impl Database { /// This can be useful for "knot tying", when tables need to reference their /// own id. pub fn next_table_id(&self) -> TableId { - TableId::from_usize(self.next_storage_id) + self.deps.next_id() } /// Add a table with the given schema to the database. @@ -1231,7 +1170,7 @@ impl Database { let spec = table.spec(); let table = WrappedTable::new(table); - let res = self.allocate_storage_id(); + let res = self.deps.next_id(); let old = self.tables.insert( res, TableInfo { diff --git a/core-relations/src/lib.rs b/core-relations/src/lib.rs index 2b861d954..95e12b981 100644 --- a/core-relations/src/lib.rs +++ b/core-relations/src/lib.rs @@ -26,10 +26,7 @@ mod tests; pub use action::{ExecutionState, MergeVal, QueryEntry, WriteVal}; pub use base_values::{BaseValue, BaseValueId, BaseValuePrinter, BaseValues, Boxed}; pub use common::Value; -pub use containers::{ - ContainerRebuildSummary, ContainerValue, ContainerValueId, ContainerValues, - SequenceContainerValue, -}; +pub use containers::{ContainerValue, ContainerValueId, ContainerValues}; pub use free_join::{ AtomId, CounterId, Database, ExternalFunction, ExternalFunctionId, TableId, Variable, make_external_func, plan::PlanStrategy, diff --git a/core-relations/src/parallel.rs b/core-relations/src/parallel.rs index 85741fed3..8a32542f8 100644 --- a/core-relations/src/parallel.rs +++ b/core-relations/src/parallel.rs @@ -2,7 +2,7 @@ use std::sync::OnceLock; -use crate::numeric_id::{DenseIdMap, IdVec, NumericId}; +use crate::numeric_id::{IdVec, NumericId}; const DEFAULT_TASKS_PER_THREAD: usize = 1; const TASKS_PER_THREAD_ENV: &str = "EGGLOG_PARALLEL_TASKS_PER_THREAD"; @@ -143,21 +143,6 @@ where collect_result_slots(results) } -pub(crate) fn map_dense_id_map_mut(map: &mut DenseIdMap, f: F) -> Vec -where - K: NumericId, - V: Send, - R: Send, - F: Fn(K, &mut V) -> R + Sync, -{ - map_mut(map.raw_mut(), |index, slot| { - slot.as_mut().map(|value| f(K::from_usize(index), value)) - }) - .into_iter() - .flatten() - .collect() -} - pub(crate) fn for_each_id_vec_mut(map: &mut IdVec, f: F) where K: NumericId, diff --git a/core-relations/src/parallel_heuristics.rs b/core-relations/src/parallel_heuristics.rs index 061c490f1..dc2148681 100644 --- a/core-relations/src/parallel_heuristics.rs +++ b/core-relations/src/parallel_heuristics.rs @@ -10,7 +10,6 @@ const DEFAULT_INDEX_CONSTRUCTION_CUTOFF: usize = 400_000; const DEFAULT_REBUILD_CUTOFF: usize = 400_000; const DEFAULT_INCREMENTAL_REBUILD_CUTOFF: usize = 16 * 1024; const DEFAULT_INTRA_CONTAINER_CUTOFF: usize = 10_000; -const DEFAULT_INTER_CONTAINER_CUTOFF: usize = 8; const DEFAULT_TABLE_OP_CUTOFF: usize = 400_000; const DEFAULT_FREE_JOIN_FORK_DEPTH: usize = 2; const DEFAULT_ACTION_BATCH_SIZE: usize = 8 * 1024; @@ -27,7 +26,6 @@ struct Cutoffs { rebuild: usize, incremental_rebuild: usize, intra_container: usize, - inter_container: usize, table_op: usize, free_join_fork_depth: usize, action_batch_size: usize, @@ -65,12 +63,6 @@ pub(crate) fn parallelize_intra_container_op(num_containers: usize) -> bool { should_parallelize(num_containers, cutoffs().intra_container) } -/// Whether or not to perform an operation in parallel across a set of different container memo -/// tables. -pub(crate) fn parallelize_inter_container_op(num_containers: usize) -> bool { - should_parallelize(num_containers, cutoffs().inter_container) -} - #[track_caller] pub(crate) fn parallelize_table_op(table_size: usize) -> bool { should_parallelize(table_size, cutoffs().table_op) @@ -109,10 +101,6 @@ fn cutoffs() -> &'static Cutoffs { "EGGLOG_PARALLEL_INTRA_CONTAINER_CUTOFF", DEFAULT_INTRA_CONTAINER_CUTOFF, ), - inter_container: cutoff( - "EGGLOG_PARALLEL_INTER_CONTAINER_CUTOFF", - DEFAULT_INTER_CONTAINER_CUTOFF, - ), table_op: cutoff("EGGLOG_PARALLEL_TABLE_OP_CUTOFF", DEFAULT_TABLE_OP_CUTOFF), free_join_fork_depth: cutoff( "EGGLOG_PARALLEL_FREE_JOIN_FORK_DEPTH", diff --git a/core-relations/src/table_spec.rs b/core-relations/src/table_spec.rs index 22d11c05f..48a9611b1 100644 --- a/core-relations/src/table_spec.rs +++ b/core-relations/src/table_spec.rs @@ -100,8 +100,8 @@ pub enum Constraint { } /// Remap individual values (e.g. to their union-find leaders) — the value-level -/// half of rebuilding, enough to rebuild a single container's contents (see -/// [`crate::ContainerValue::rebuild_contents`]). +/// half of rebuilding, enough to rebuild a serialized container key (see +/// [`crate::ContainerValue::rebuild_sequence`]). pub trait ValueRebuilder: Send + Sync { /// Rebuild a single value. fn rebuild_val(&self, val: Value) -> Value; diff --git a/egglog-bridge/src/lib.rs b/egglog-bridge/src/lib.rs index bfbea4b87..5377740a6 100644 --- a/egglog-bridge/src/lib.rs +++ b/egglog-bridge/src/lib.rs @@ -19,8 +19,8 @@ use std::{ use crate::core_relations::{ BaseValue, BaseValueId, BaseValues, ColumnId, Constraint, ContainerValue, ContainerValues, CounterId, Database, DisplacedTable, ExecutionState, ExternalFunction, ExternalFunctionId, - MergeVal, Offset, PlanStrategy, SequenceContainerValue, SortedWritesTable, TableId, - TaggedRowBuffer, Value, WrappedTable, + MergeVal, Offset, PlanStrategy, SortedWritesTable, TableId, TaggedRowBuffer, Value, + WrappedTable, }; use crate::numeric_id::{DenseIdMap, DenseIdMapWithReuse, NumericId, define_id}; use egglog_concurrency::ThreadPool; @@ -313,30 +313,15 @@ impl EGraph { value } - /// Register the given [`ContainerValue`] type with this EGraph. + /// Register a serialized container type with this EGraph. /// - /// The given container will use the EGraph's union-find to manage rebuilding and the merging - /// of containers with a common id. + /// Containers use the variable-arity table backend and participate in the + /// normal maintenance scheduler. The EGraph's union-find manages rebuilding + /// and merges identities whose serialized keys become equal. pub fn register_container_ty(&mut self) { let uf_table = self.uf_table; let ts_counter = self.timestamp_counter; - self.db - .register_container_type::(self.id_counter, move |state, old, new| { - if old != new { - let next_ts = Value::from_usize(state.read_counter(ts_counter)); - state.stage_insert(uf_table, &[old, new, next_ts]); - std::cmp::min(old, new) - } else { - old - } - }); - } - - /// Register a container using the variable-arity sequence-table backend. - pub fn register_sequence_container_ty(&mut self) { - let uf_table = self.uf_table; - let ts_counter = self.timestamp_counter; - self.db.register_sequence_container_type::( + self.db.register_container_type::( self.id_counter, move |state, old, new| { if old != new { diff --git a/egglog-bridge/src/tests.rs b/egglog-bridge/src/tests.rs index 54df39f03..627d057ca 100644 --- a/egglog-bridge/src/tests.rs +++ b/egglog-bridge/src/tests.rs @@ -11,7 +11,7 @@ use std::{ use crate::core_relations; use crate::core_relations::{ - ContainerValue, ExternalFunctionId, Value, ValueRebuilder, make_external_func, + BaseValues, ContainerValue, ExternalFunctionId, Value, ValueRebuilder, make_external_func, }; use crate::numeric_id::NumericId; use log::debug; @@ -449,11 +449,31 @@ fn math_test(mut egraph: EGraph, can_subsume: bool) { #[derive(Clone, Debug, Hash, Eq, PartialEq)] struct VecContainer(Vec); impl ContainerValue for VecContainer { - fn rebuild_contents(&mut self, rebuilder: &dyn ValueRebuilder) -> bool { - rebuilder.rebuild_slice(&mut self.0) + fn encode_sequence(&self, _base_values: &BaseValues, out: &mut Vec) { + out.extend_from_slice(&self.0); } - fn iter(&self) -> impl Iterator + '_ { - self.0.iter().copied() + + fn decode_sequence(sequence: &[Value], _base_values: &BaseValues) -> Self { + Self(sequence.to_vec()) + } + + fn sequence_values(sequence: &[Value]) -> &[Value] { + sequence + } + + fn rebuild_sequence( + sequence: &[Value], + _base_values: &BaseValues, + rebuilder: &dyn ValueRebuilder, + out: &mut Vec, + ) -> bool { + out.extend_from_slice(sequence); + if rebuilder.rebuild_slice(out) { + true + } else { + out.clear(); + false + } } } diff --git a/src/exec_state.rs b/src/exec_state.rs index 3583446a8..72aea82ce 100644 --- a/src/exec_state.rs +++ b/src/exec_state.rs @@ -203,13 +203,13 @@ pub trait Core<'a, 'db: 'a>: Internal<'a, 'db> { } /// Reconstruct a container visible to this execution, including a local - /// prediction. This is the explicit slow sequence-container path. + /// prediction. This is the explicit slow serialized-container path. fn value_to_owned_container(&self, x: Value) -> Option { self.es().get_container::(x) } - /// Run `f` over the fast serialized payload of a sequence-backed container - /// without reconstructing its Rust container. + /// Run `f` over a container's serialized payload without reconstructing its + /// Rust representation. fn with_container_sequence( &self, x: Value, @@ -218,7 +218,7 @@ pub trait Core<'a, 'db: 'a>: Internal<'a, 'db> { self.es().container_sequence::(x).map(f) } - /// Intern an already serialized sequence-container key. + /// Intern an already serialized container key. fn register_container_sequence(&mut self, key: &[Value]) -> Value { self.es_mut().register_container_sequence::(key) } diff --git a/src/lib.rs b/src/lib.rs index 5fb9cdab8..5ec3088a7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -2252,12 +2252,12 @@ impl EGraph { self.backend.base_values().get::(x) } - /// Convert from an egglog value to a reference of a Rust container type. + /// Decode an egglog value into a Rust container. /// /// Returns `None` if the value cannot be converted to the requested container type. /// - /// Warning: The return type of this function may contain lock guards. - /// Attempts to modify the contents of the containers database may deadlock if the given guard has not been dropped. + /// The returned `Deref` facade owns the decoded container; it does not hold + /// a lock on the container registry. pub fn value_to_container( &self, x: Value, diff --git a/src/sort/fn.rs b/src/sort/fn.rs index 1465ef4dd..96fa6349a 100644 --- a/src/sort/fn.rs +++ b/src/sort/fn.rs @@ -198,23 +198,6 @@ impl Hash for FunctionContainer { } impl ContainerValue for FunctionContainer { - fn rebuild_contents(&mut self, rebuilder: &dyn ValueRebuilder) -> bool { - let mut changed = false; - for (s, old) in &mut self.1 { - if s.is_eq_sort() || s.is_eq_container_sort() { - let new = rebuilder.rebuild_val(*old); - changed |= *old != new; - *old = new; - } - } - changed - } - fn iter(&self) -> impl Iterator + '_ { - self.1.iter().map(|(_, v)| v).copied() - } -} - -impl SequenceContainerValue for FunctionContainer { fn encode_sequence(&self, base_values: &BaseValues, out: &mut Vec) { let partial_arcsorts = self .1 @@ -411,7 +394,7 @@ impl Sort for FunctionSort { backend .base_values_mut() .register_type::(); - backend.register_sequence_container_ty::(); + backend.register_container_ty::(); } fn as_arc_any(self: Arc) -> Arc { diff --git a/src/sort/map.rs b/src/sort/map.rs index 8690d9f95..0857e6101 100644 --- a/src/sort/map.rs +++ b/src/sort/map.rs @@ -13,34 +13,6 @@ pub struct MapContainer { } impl ContainerValue for MapContainer { - fn rebuild_contents(&mut self, rebuilder: &dyn ValueRebuilder) -> bool { - let mut changed = false; - if self.do_rebuild_keys { - self.data = self - .data - .iter() - .map(|(old, v)| { - let new = rebuilder.rebuild_val(*old); - changed |= *old != new; - (new, *v) - }) - .collect(); - } - if self.do_rebuild_vals { - for old in self.data.values_mut() { - let new = rebuilder.rebuild_val(*old); - changed |= *old != new; - *old = new; - } - } - changed - } - fn iter(&self) -> impl Iterator + '_ { - self.data.iter().flat_map(|(k, v)| [k, v]).copied() - } -} - -impl SequenceContainerValue for MapContainer { fn encode_sequence(&self, _base_values: &BaseValues, out: &mut Vec) { out.push(map_rebuild_mask(self.do_rebuild_keys, self.do_rebuild_vals)); out.extend(self.data.iter().flat_map(|(key, value)| [*key, *value])); @@ -114,10 +86,10 @@ impl SequenceContainerValue for MapContainer { return false; } - // Rebuild keys before values, matching `MapContainer::rebuild_contents`. - // Iteration is in ascending old-key order, so inserting rebuilt keys in - // that order preserves BTreeMap's existing collision rule: when two - // old keys canonicalize together, the later old key's value wins. + // Rebuild keys before values. Iteration is in ascending old-key order, + // so inserting rebuilt keys in that order preserves BTreeMap's existing + // collision rule: when two old keys canonicalize together, the later + // old key's value wins. let mut rebuilt = BTreeMap::new(); for pair in data.chunks_exact(2) { let key = if mask & REBUILD_KEYS != 0 { @@ -302,10 +274,6 @@ impl ContainerSort for MapSort { &self.name } - fn register_type(&self, backend: &mut egglog_bridge::EGraph) { - backend.register_sequence_container_ty::(); - } - fn inner_sorts(&self) -> Vec { vec![self.key.clone(), self.value.clone()] } @@ -531,8 +499,8 @@ enum MapReadOp { NotContains, } -/// Map reads use binary search over the borrowed canonical sequence and only -/// reconstruct a `MapContainer` for a legacy backend. +/// Map reads use binary search over the borrowed canonical sequence, with an +/// explicit decode-based slow fallback. #[derive(Clone)] struct MapRead { name: String, @@ -623,8 +591,8 @@ enum MapEditOp { Remove, } -/// Map updates splice the borrowed alternating sequence directly and only -/// round-trip through `MapContainer` for a legacy backend. +/// Map updates splice the borrowed alternating sequence directly, with an +/// explicit decode/edit/encode slow fallback. #[derive(Clone)] struct MapEdit { name: String, @@ -699,9 +667,8 @@ impl PurePrim for MapEdit { return Some(state.register_container_sequence::(&key?)); } - // Compatibility path for a legacy environment: reconstruct the Rust - // map, perform the operation, and let normal container registration - // serialize it when the environment is sequence-backed. + // Slow compatibility path: reconstruct the Rust map, perform the + // operation, and let normal container registration serialize it again. let mut map = state.value_to_owned_container::(map_id)?; match self.op { MapEditOp::Insert => { @@ -872,7 +839,7 @@ mod tests { } #[test] - fn sequence_rebuild_matches_legacy_key_collision_semantics() { + fn sequence_rebuild_preserves_last_old_key_on_collision() { let original = map(true, true, &[(2, 20), (4, 40), (6, 60)]); let rebuilder = Remap(vec![ (value(2), value(1)), @@ -882,10 +849,6 @@ mod tests { (value(60), value(61)), ]); - let mut legacy = original.clone(); - assert!(legacy.rebuild_contents(&rebuilder)); - assert_eq!(legacy, map(true, true, &[(1, 41), (3, 61)])); - let mut encoded = Vec::new(); original.encode_sequence(&BaseValues::default(), &mut encoded); let mut rebuilt = Vec::new(); @@ -897,7 +860,7 @@ mod tests { )); assert_eq!( MapContainer::decode_sequence(&rebuilt, &BaseValues::default()), - legacy + map(true, true, &[(1, 41), (3, 61)]) ); } diff --git a/src/sort/mod.rs b/src/sort/mod.rs index 906e27f3a..7abf6afff 100644 --- a/src/sort/mod.rs +++ b/src/sort/mod.rs @@ -8,8 +8,7 @@ use std::{any::Any, sync::Arc}; use crate::core_relations; pub use core_relations::{ - BaseValues, Boxed, ContainerValue, ContainerValues, ExecutionState, SequenceContainerValue, - ValueRebuilder, + BaseValues, Boxed, ContainerValue, ContainerValues, ExecutionState, ValueRebuilder, }; pub use egglog_bridge::ColumnTy; diff --git a/src/sort/multiset.rs b/src/sort/multiset.rs index fd68d50e0..679fb27f8 100644 --- a/src/sort/multiset.rs +++ b/src/sort/multiset.rs @@ -88,24 +88,6 @@ fn multiset_term_children(termdag: &TermDag, term: TermId) -> Option } impl ContainerValue for MultiSetContainer { - fn rebuild_contents(&mut self, rebuilder: &dyn ValueRebuilder) -> bool { - // If the contents are an eq-sort then we want to rebuild - if self.do_rebuild { - let mut xs: Vec<_> = self.data.iter().copied().collect(); - let changed = rebuilder.rebuild_slice(&mut xs); - self.data = xs.into_iter().collect(); - changed - // if the contents are just a primitive then don't need to do anything. - } else { - false - } - } - fn iter(&self) -> impl Iterator + '_ { - self.data.iter().copied() - } -} - -impl SequenceContainerValue for MultiSetContainer { fn encode_sequence(&self, _base_values: &BaseValues, out: &mut Vec) { assert!( i64::try_from(self.data.len()).is_ok(), @@ -287,10 +269,6 @@ impl ContainerSort for MultiSetSort { &self.name } - fn register_type(&self, backend: &mut egglog_bridge::EGraph) { - backend.register_sequence_container_ty::(); - } - fn inner_sorts(&self) -> Vec { vec![self.element.clone()] } diff --git a/src/sort/pair.rs b/src/sort/pair.rs index b2a92b1f2..1887322ba 100644 --- a/src/sort/pair.rs +++ b/src/sort/pair.rs @@ -11,26 +11,6 @@ pub struct PairContainer { } impl ContainerValue for PairContainer { - fn rebuild_contents(&mut self, rebuilder: &dyn ValueRebuilder) -> bool { - let mut changed = false; - if self.do_rebuild_first { - let new = rebuilder.rebuild_val(self.first); - changed |= self.first != new; - self.first = new; - } - if self.do_rebuild_second { - let new = rebuilder.rebuild_val(self.second); - changed |= self.second != new; - self.second = new; - } - changed - } - fn iter(&self) -> impl Iterator + '_ { - [self.first, self.second].into_iter() - } -} - -impl SequenceContainerValue for PairContainer { fn encode_sequence(&self, _base_values: &BaseValues, out: &mut Vec) { let rebuild_mask = self.do_rebuild_first as usize | ((self.do_rebuild_second as usize) << 1); @@ -196,10 +176,6 @@ impl ContainerSort for PairSort { &self.name } - fn register_type(&self, backend: &mut egglog_bridge::EGraph) { - backend.register_sequence_container_ty::(); - } - fn inner_sorts(&self) -> Vec { vec![self.first.clone(), self.second.clone()] } diff --git a/src/sort/set.rs b/src/sort/set.rs index 77d617795..b331618ab 100644 --- a/src/sort/set.rs +++ b/src/sort/set.rs @@ -9,22 +9,6 @@ pub struct SetContainer { } impl ContainerValue for SetContainer { - fn rebuild_contents(&mut self, rebuilder: &dyn ValueRebuilder) -> bool { - if self.do_rebuild { - let mut xs: Vec<_> = self.data.iter().copied().collect(); - let changed = rebuilder.rebuild_slice(&mut xs); - self.data = xs.into_iter().collect(); - changed - } else { - false - } - } - fn iter(&self) -> impl Iterator + '_ { - self.data.iter().copied() - } -} - -impl SequenceContainerValue for SetContainer { fn encode_sequence(&self, _base_values: &BaseValues, out: &mut Vec) { out.push(Value::from_usize(self.do_rebuild as usize)); out.extend(self.data.iter().copied()); @@ -186,10 +170,6 @@ impl ContainerSort for SetSort { &self.name } - fn register_type(&self, backend: &mut egglog_bridge::EGraph) { - backend.register_sequence_container_ty::(); - } - fn inner_sorts(&self) -> Vec { vec![self.element.clone()] } diff --git a/src/sort/vec.rs b/src/sort/vec.rs index f454fbd44..574fbdc1e 100644 --- a/src/sort/vec.rs +++ b/src/sort/vec.rs @@ -13,19 +13,6 @@ pub struct VecContainer { } impl ContainerValue for VecContainer { - fn rebuild_contents(&mut self, rebuilder: &dyn ValueRebuilder) -> bool { - if self.do_rebuild { - rebuilder.rebuild_slice(&mut self.data) - } else { - false - } - } - fn iter(&self) -> impl Iterator + '_ { - self.data.iter().copied() - } -} - -impl SequenceContainerValue for VecContainer { fn encode_sequence(&self, _base_values: &BaseValues, out: &mut Vec) { out.push(Value::from_usize(self.do_rebuild as usize)); out.extend_from_slice(&self.data); @@ -178,10 +165,6 @@ impl ContainerSort for VecSort { &self.name } - fn register_type(&self, backend: &mut egglog_bridge::EGraph) { - backend.register_sequence_container_ty::(); - } - fn inner_sorts(&self) -> Vec { vec![self.element.clone()] } diff --git a/tests/container-before-table-rebuild.egg b/tests/container-before-table-rebuild.egg index 13d90aa8e..df5ce980e 100644 --- a/tests/container-before-table-rebuild.egg +++ b/tests/container-before-table-rebuild.egg @@ -1,4 +1,4 @@ -;; A sequence-backed container table must rebuild before an ordinary table +;; A container table must rebuild before an ordinary table ;; whose keys and values both refer to the affected e-classes. Otherwise the ;; two `image` rows collide while their Vec values are still distinct, and the ;; `:no-merge` function reports an illegal merge. diff --git a/tests/container_rebuild.rs b/tests/container_rebuild.rs index 2ab017943..67a4ca697 100644 --- a/tests/container_rebuild.rs +++ b/tests/container_rebuild.rs @@ -428,7 +428,7 @@ fn nested_set_rebuild_term_only() { .unwrap(); } -/// A sequence-backed `UnstableFn` canonicalizes its rebuildable partial args. +/// A serialized `UnstableFn` canonicalizes its rebuildable partial args. /// `UnstableFn` deliberately has no proof validator, so this runs in ordinary /// mode rather than term/proof encoding. #[test]