diff --git a/core-relations/src/action/mod.rs b/core-relations/src/action/mod.rs index a64255410..0e35804c4 100644 --- a/core-relations/src/action/mod.rs +++ b/core-relations/src/action/mod.rs @@ -21,7 +21,6 @@ use rustc_hash::FxHasher; use crate::{ BaseValues, ContainerValueId, ContainerValues, ExternalFunctionId, WrappedTable, common::Value, - dependency_graph::MaintenanceId, free_join::{ CounterId, CounterReservation, Counters, ExternalFunctions, TableId, TableInfo, Variable, }, @@ -292,45 +291,12 @@ pub(crate) struct ExtractedBinding { pub(crate) vals: Pooled>, } -/// Namespace-qualified owner of a predicted row. -/// -/// Tables and sequence-backed container types use independent dense `u32` id -/// spaces. The high bit distinguishes them without widening [`PredictedEntry`]. -#[derive(Clone, Copy, Debug, Eq, PartialEq)] -#[repr(transparent)] -struct PredictionOwner(u32); - -impl PredictionOwner { - const CONTAINER_TAG: u32 = 1 << 31; - - fn table(table: TableId) -> Self { - Self::new(table.rep(), false) - } - - fn container(container: ContainerValueId) -> Self { - Self::new(container.rep(), true) - } - - fn new(rep: u32, is_container: bool) -> Self { - assert_eq!( - rep & Self::CONTAINER_TAG, - 0, - "prediction owner id exceeds its 31-bit namespace" - ); - Self(rep | if is_container { Self::CONTAINER_TAG } else { 0 }) - } - - fn rep(self) -> u32 { - self.0 - } -} - #[derive(Clone, Copy)] struct PredictedEntry { hash: u64, index: u32, row_end: u32, - owner: PredictionOwner, + owner: TableId, key_arity: u32, } @@ -339,7 +305,7 @@ struct PredictedValueEntry { hash: u64, index: u32, row_end: u32, - owner: PredictionOwner, + owner: TableId, } #[derive(Default)] @@ -368,10 +334,10 @@ impl Clear for PredictedVals { impl PredictedVals { #[cfg(test)] fn hash(table: TableId, key: &[Value]) -> u64 { - Self::owner_hash(PredictionOwner::table(table), key) + Self::owner_hash(table, key) } - fn owner_hash(owner: PredictionOwner, key: &[Value]) -> u64 { + fn owner_hash(owner: TableId, key: &[Value]) -> u64 { let mut hasher = FxHasher::default(); hasher.write_u32(owner.rep()); hasher.write_usize(key.len()); @@ -381,7 +347,7 @@ impl PredictedVals { hasher.finish() } - fn value_hash(owner: PredictionOwner, value: Value) -> u64 { + fn value_hash(owner: TableId, value: Value) -> u64 { let mut hasher = FxHasher::default(); hasher.write_u32(owner.rep()); hasher.write_u32(value.rep()); @@ -396,13 +362,7 @@ impl PredictedVals { &self.values[entry.index as usize..entry.row_end as usize] } - fn matches( - &self, - entry: &PredictedEntry, - hash: u64, - owner: PredictionOwner, - key: &[Value], - ) -> bool { + fn matches(&self, entry: &PredictedEntry, hash: u64, owner: TableId, key: &[Value]) -> bool { if entry.hash != hash || entry.owner != owner || entry.key_arity as usize != key.len() { return false; } @@ -413,7 +373,7 @@ impl PredictedVals { &self, entry: &PredictedValueEntry, hash: u64, - owner: PredictionOwner, + owner: TableId, value: Value, ) -> bool { entry.hash == hash @@ -433,7 +393,7 @@ impl PredictedVals { container: ContainerValueId, value: Value, ) -> Option<&[Value]> { - let owner = PredictionOwner::container(container); + let owner = TableId::from(container); let hash = Self::value_hash(owner, value); let entry = self .by_value @@ -481,13 +441,7 @@ impl PredictedVals { row_arity: usize, default: impl FnOnce(&mut Vec, usize), ) -> (&[Value], bool) { - self.get_or_insert_impl( - PredictionOwner::table(table), - key, - row_arity, - false, - default, - ) + self.get_or_insert_impl(table, key, row_arity, false, default) } /// Predict a row with exactly one non-key value and index it in both @@ -506,18 +460,12 @@ impl PredictedVals { .len() .checked_add(1) .expect("predicted container row arity overflow"); - self.get_or_insert_impl( - PredictionOwner::container(container), - key, - row_arity, - true, - default, - ) + self.get_or_insert_impl(TableId::from(container), key, row_arity, true, default) } fn get_or_insert_impl( &mut self, - owner: PredictionOwner, + owner: TableId, key: &[Value], row_arity: usize, index_value: bool, @@ -573,9 +521,26 @@ pub(crate) struct DbView<'a> { pub(crate) external_funcs: &'a ExternalFunctions, pub(crate) bases: &'a BaseValues, pub(crate) containers: &'a ContainerValues, - pub(crate) notification_list: &'a NotificationList, - pub(crate) table_maintenance: &'a DenseIdMap, - pub(crate) container_maintenance: &'a DenseIdMap, + pub(crate) notification_list: &'a NotificationList, +} + +impl DbView<'_> { + /// Create a mutation buffer for any installed maintenance participant. + /// + /// Relations and sequence-backed containers share the [`TableId`] + /// namespace, so action execution should not need to know which owner map + /// contains a write destination. A participant temporarily detached for + /// merge or rebuild is intentionally absent from this view; callers seed + /// buffers for those destinations before detaching the participant group. + fn new_buffer(&self, table: TableId) -> Box { + if let Some(info) = self.table_info.get(table) { + return info.table.new_buffer(); + } + self.containers + .maintenance_table(ContainerValueId::from_table_id(table)) + .unwrap_or_else(|| panic!("write destination {table:?} has no maintenance storage")) + .new_buffer() + } } /// A handle on a database that may be in the process of running a rule. @@ -610,7 +575,6 @@ pub struct ExecutionState<'a> { pub(crate) db: DbView<'a>, counter_reservations: CounterReservations, buffers: MutationBuffers<'a>, - container_buffers: DenseIdMap>, /// Whether any mutations have been staged via this ExecutionState. pub(crate) changed: bool, /// Atomic flag for early stopping of rule execution. @@ -646,12 +610,7 @@ impl<'db> ExecutionStateSeed<'db, '_> { predicted: Default::default(), db: self.db, counter_reservations: Default::default(), - buffers: MutationBuffers::new( - self.db.notification_list, - self.db.table_maintenance, - Default::default(), - ), - container_buffers: Default::default(), + buffers: MutationBuffers::new(self.db.notification_list, Default::default()), changed: false, stop_match: Arc::clone(self.stop_match), } @@ -665,15 +624,13 @@ impl<'db> ExecutionStateSeed<'db, '_> { /// A basic wrapper around an map from table id to a mutation buffer for that table that also /// tracks if a table has been modified. struct MutationBuffers<'a> { - notify_list: &'a NotificationList, - table_maintenance: &'a DenseIdMap, + notify_list: &'a NotificationList, buffers: DenseIdMap>, } impl Clone for MutationBuffers<'_> { fn clone(&self) -> Self { - let mut res = - MutationBuffers::new(self.notify_list, self.table_maintenance, Default::default()); + let mut res = MutationBuffers::new(self.notify_list, Default::default()); for (id, buf) in self.buffers.iter() { res.buffers.insert(id, buf.fresh_handle()); } @@ -681,25 +638,13 @@ impl Clone for MutationBuffers<'_> { } } -fn fresh_container_buffers( - buffers: &DenseIdMap>, -) -> DenseIdMap> { - let mut result = DenseIdMap::new(); - for (id, buffer) in buffers.iter() { - result.insert(id, buffer.fresh_handle()); - } - result -} - impl<'a> MutationBuffers<'a> { fn new( - notify_list: &'a NotificationList, - table_maintenance: &'a DenseIdMap, + notify_list: &'a NotificationList, buffers: DenseIdMap>, ) -> MutationBuffers<'a> { MutationBuffers { notify_list, - table_maintenance, buffers, } } @@ -708,12 +653,12 @@ impl<'a> MutationBuffers<'a> { } fn stage_insert(&mut self, table_id: TableId, row: &[Value]) { self.buffers[table_id].stage_insert(row); - self.notify_list.notify(self.table_maintenance[table_id]); + self.notify_list.notify(table_id); } fn stage_remove(&mut self, table_id: TableId, key: &[Value]) { self.buffers[table_id].stage_remove(key); - self.notify_list.notify(self.table_maintenance[table_id]); + self.notify_list.notify(table_id); } } @@ -724,7 +669,6 @@ impl Clone for ExecutionState<'_> { db: self.db, counter_reservations: Default::default(), buffers: self.buffers.clone(), - container_buffers: fresh_container_buffers(&self.container_buffers), changed: false, stop_match: Arc::clone(&self.stop_match), } @@ -740,8 +684,7 @@ impl<'a> ExecutionState<'a> { predicted: Default::default(), db, counter_reservations: Default::default(), - buffers: MutationBuffers::new(db.notification_list, db.table_maintenance, buffers), - container_buffers: Default::default(), + buffers: MutationBuffers::new(db.notification_list, buffers), changed: false, stop_match: Arc::new(AtomicBool::new(false)), } @@ -762,8 +705,7 @@ impl<'a> ExecutionState<'a> { /// /// If you are using `egglog`, consider using `egglog_bridge::TableAction`. pub fn stage_insert(&mut self, table: TableId, row: &[Value]) { - self.buffers - .lazy_init(table, || self.db.table_info[table].table.new_buffer()); + self.buffers.lazy_init(table, || self.db.new_buffer(table)); self.buffers.stage_insert(table, row); self.changed = true; } @@ -772,8 +714,7 @@ impl<'a> ExecutionState<'a> { /// /// If you are using `egglog`, consider using `egglog_bridge::TableAction`. pub fn stage_remove(&mut self, table: TableId, key: &[Value]) { - self.buffers - .lazy_init(table, || self.db.table_info[table].table.new_buffer()); + self.buffers.lazy_init(table, || self.db.new_buffer(table)); self.buffers.stage_remove(table, key); self.changed = true; } @@ -823,8 +764,8 @@ impl<'a> ExecutionState<'a> { /// /// The row is `key` followed by one fresh identity allocated from /// `counter`. Repeated keys within this execution reuse the first identity - /// and are staged only once. The buffer cache is separate from ordinary - /// table buffers because its namespace is [`ContainerValueId`]. + /// and are staged only once. Container and relation tables share one + /// backing-table namespace and mutation-buffer cache. pub(crate) fn predict_container_value( &mut self, container: ContainerValueId, @@ -832,6 +773,7 @@ impl<'a> ExecutionState<'a> { counter: CounterId, new_buffer: impl FnOnce() -> Box, ) -> Value { + let table = TableId::from(container); let counters = self.db.counters; let counter_reservations = &mut self.counter_reservations; let (row, inserted) = @@ -843,14 +785,8 @@ impl<'a> ExecutionState<'a> { }); let identity = row[row.len() - 1]; if inserted { - self.container_buffers - .get_or_insert(container, new_buffer) - .stage_insert(row); - // Standalone SequenceContainerEnv tests do not register a - // database scheduler target; database-owned environments do. - if let Some(maintenance) = self.db.container_maintenance.get(container) { - self.db.notification_list.notify(*maintenance); - } + self.buffers.lazy_init(table, new_buffer); + self.buffers.stage_insert(table, row); self.changed = true; } identity diff --git a/core-relations/src/action/tests.rs b/core-relations/src/action/tests.rs index 0f23e840e..668338a7c 100644 --- a/core-relations/src/action/tests.rs +++ b/core-relations/src/action/tests.rs @@ -12,7 +12,7 @@ use crate::{ }; use super::{ - PredictedEntry, PredictedVals, PredictedValueEntry, PredictionOwner, + PredictedEntry, PredictedVals, PredictedValueEntry, mask::{Mask, MaskIter}, }; @@ -119,7 +119,7 @@ fn predicted_vals_resolve_hash_collisions_with_the_backing_rows() { hash, index: 0, row_end: collision_row.len() as u32, - owner: PredictionOwner::table(table), + owner: table, key_arity: collision_key.len() as u32, }, |entry| entry.hash, @@ -196,10 +196,10 @@ fn predicted_vals_reverse_lookup_rejects_a_conflicting_identity() { } #[test] -fn predicted_vals_keep_table_and_container_namespaces_distinct() { +fn predicted_vals_keep_backing_table_namespaces_distinct() { let mut predicted = PredictedVals::default(); let table = TableId::from_usize(3); - let container = ContainerValueId::from_usize(3); + let container = ContainerValueId::from_usize(4); let key = [Value::from_usize(10)]; let table_row = [key[0], Value::from_usize(80)]; let container_row = [key[0], Value::from_usize(90)]; @@ -244,7 +244,7 @@ fn predicted_container_rows_require_exactly_one_identity_value() { fn predicted_vals_reverse_lookup_resolves_raw_hash_collisions() { let mut predicted = PredictedVals::default(); let container = ContainerValueId::from_usize(3); - let owner = PredictionOwner::container(container); + let owner = TableId::from(container); let collision_row = [Value::from_usize(10), Value::from_usize(80)]; let key = [Value::from_usize(20), Value::from_usize(21)]; let identity = Value::from_usize(90); diff --git a/core-relations/src/containers/mod.rs b/core-relations/src/containers/mod.rs index dd032a8f4..d56658c14 100644 --- a/core-relations/src/containers/mod.rs +++ b/core-relations/src/containers/mod.rs @@ -22,13 +22,10 @@ use rustc_hash::FxHasher; use crate::{ ColumnId, CounterId, ExecutionState, Offset, Subset, SubsetRef, TableId, TaggedRowBuffer, Value, WrappedTable, - common::{DashMap, IndexSet, SubsetTracker}, + common::{DashMap, HashMap, IndexSet, SubsetTracker}, parallel, - parallel_heuristics::{ - parallelize_db_level_op, parallelize_inter_container_op, parallelize_intra_container_op, - parallelize_rebuild, - }, - table_spec::{Rebuilder, ValueRebuilder}, + parallel_heuristics::{parallelize_inter_container_op, parallelize_intra_container_op}, + table_spec::{MaintenanceTable, Rebuilder, ValueRebuilder}, }; mod sequence; @@ -39,7 +36,27 @@ mod tests; use sequence::SequenceContainerEnv; pub use sequence::SequenceContainerValue; -define_id!(pub ContainerValueId, u32, "an identifier for containers"); +define_id!( + pub ContainerValueId, + u32, + "a typed identifier for a container's variable-arity table" +); + +// Container tables participate in the same dense storage namespace as +// fixed-arity relation tables. `ContainerValueId` remains a distinct public +// type so APIs cannot accidentally use a relation as a container, but the +// representation is the backing table's `TableId`. +impl From for TableId { + fn from(id: ContainerValueId) -> Self { + TableId::from_usize(id.index()) + } +} + +impl ContainerValueId { + pub(crate) fn from_table_id(id: TableId) -> Self { + ContainerValueId::from_usize(id.index()) + } +} pub trait MergeFn: Fn(&mut ExecutionState, Value, Value) -> Value + dyn_clone::DynClone + Send + Sync @@ -52,22 +69,16 @@ dyn_clone::clone_trait_object!(MergeFn); #[derive(Clone, Default)] struct ContainerIds { - ids: IndexSet, + ids: HashMap, } impl ContainerIds { - fn insert(&mut self, ty: TypeId) -> ContainerValueId { - if let Some(idx) = self.ids.get_index_of(&ty) { - ContainerValueId::from_usize(idx) - } else { - let idx = self.ids.len(); - self.ids.insert(ty); - ContainerValueId::from_usize(idx) - } + fn insert(&mut self, ty: TypeId, id: ContainerValueId) -> ContainerValueId { + *self.ids.entry(ty).or_insert(id) } fn get(&self, ty: &TypeId) -> Option { - self.ids.get_index_of(ty).map(ContainerValueId::from_usize) + self.ids.get(ty).copied() } } @@ -75,6 +86,8 @@ impl ContainerIds { pub struct ContainerValues { subset_tracker: SubsetTracker, container_ids: ContainerIds, + // ContainerValueId shares the global TableId representation, so relation + // registrations appear as holes in this container-only map. data: DenseIdMap>, } @@ -127,12 +140,10 @@ pub struct ContainerRebuildSummary { } impl ContainerRebuildSummary { - /// Returns whether any container entry changed during rebuild. pub fn changed(&self) -> bool { self.changed } - /// Returns the container ids whose parent rows may need retimestamping. pub fn dirty_ids(&self) -> &IndexSet { &self.dirty_ids } @@ -146,7 +157,7 @@ impl ContainerRebuildSummary { self.dirty_ids.insert(value); } - fn extend(&mut self, other: Self) { + pub(crate) fn extend(&mut self, other: Self) { self.changed |= other.changed; self.dirty_ids.extend(other.dirty_ids); } @@ -165,15 +176,46 @@ pub(crate) enum ContainerBackend { /// The shared union-find rebuild inputs for one container rebuild cycle. /// -/// Preparing this once is important for incremental rebuilds: consulting the -/// subset tracker separately for the sequence and legacy passes would advance -/// its watermark twice and make the second pass miss the relevant UF delta. -struct ContainerRebuildContext<'a> { +/// 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. +pub(crate) struct ContainerRebuildContext<'a> { table: &'a WrappedTable, rebuilder: Box, + to_scan: Option>, +} + +/// Owned portion of a container rebuild context. Keeping the union-find subset +/// separate lets the database detach and restore one dependency stratum at a +/// time while recreating the short-lived rebuilder borrow from the still- +/// installed source table. +pub(crate) struct ContainerRebuildPlan { to_scan: Option, } +impl ContainerRebuildPlan { + pub(crate) fn context<'a>( + &'a self, + table: &'a WrappedTable, + ) -> Option> { + Some(ContainerRebuildContext { + table, + rebuilder: table.rebuilder(&[])?, + to_scan: self.to_scan.as_ref().map(Subset::as_ref), + }) + } +} + +impl ContainerRebuildContext<'_> { + pub(crate) fn apply( + &self, + environment: &mut (dyn DynamicContainerEnv + Send + Sync), + exec_state: &mut ExecutionState, + ) -> ContainerRebuildSummary { + environment.apply_rebuild(self.table, &*self.rebuilder, self.to_scan, exec_state) + } +} + impl ContainerValues { pub fn new() -> Self { Default::default() @@ -193,6 +235,17 @@ impl ContainerValues { .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. /// /// This check does not mutate the registry, so the database can distinguish @@ -289,9 +342,8 @@ impl ContainerValues { /// unchanged if it is not a registered container of the type behind /// `type_id`. /// - /// Unlike [`ContainerValues::rebuild_all`], which drives rebuilds off the - /// backend union-find, the caller supplies the remapping explicitly and - /// identifies the container type dynamically by its [`TypeId`]. + /// The caller supplies the remapping explicitly and identifies the + /// container type dynamically by its [`TypeId`]. pub fn rebuild_val_with( &self, type_id: TypeId, @@ -309,104 +361,48 @@ impl ContainerValues { .unwrap_or(value) } - /// Apply the given rebuild to the contents of each container. - pub fn rebuild_all( + /// Prepare the source-table delta once without retaining a borrow of the + /// source table. A short-lived [`ContainerRebuildContext`] can then be + /// reconstructed for each mixed participant stratum. + pub(crate) fn prepare_rebuild_plan( &mut self, table_id: TableId, table: &WrappedTable, - exec_state: &mut ExecutionState, - ) -> ContainerRebuildSummary { - let Some(context) = self.prepare_rebuild(table_id, table) else { - return Default::default(); - }; - let mut summary = self.rebuild_sequence_pass(&context, exec_state); - summary.extend(self.rebuild_legacy_pass(&context, exec_state)); - self.finish_rebuild(&mut summary); - summary - } - - /// Prepare the immutable UF inputs shared by the backend-specific passes - /// in one container rebuild cycle. - fn prepare_rebuild<'a>( - &mut self, - table_id: TableId, - table: &'a WrappedTable, - ) -> Option> { + ) -> Option { let rebuilder = table.rebuilder(&[])?; - let to_scan = rebuilder.hint_col().map(|_| { - // We may attempt an incremental rebuild. This must be computed - // exactly once and shared by both backend passes. - self.subset_tracker.recent_updates(table_id, table) - }); - Some(ContainerRebuildContext { - table, - rebuilder, - to_scan, - }) - } - - /// Rebuild only sequence-backed container environments. - fn rebuild_sequence_pass( - &mut self, - context: &ContainerRebuildContext<'_>, - exec_state: &mut ExecutionState, - ) -> ContainerRebuildSummary { - self.rebuild_backend_pass(ContainerBackend::Sequence, context, exec_state) - } - - /// Rebuild only legacy container environments. - fn rebuild_legacy_pass( - &mut self, - context: &ContainerRebuildContext<'_>, - exec_state: &mut ExecutionState, - ) -> ContainerRebuildSummary { - self.rebuild_backend_pass(ContainerBackend::Legacy, context, exec_state) + let to_scan = rebuilder + .hint_col() + .map(|_| self.subset_tracker.recent_updates(table_id, table)); + Some(ContainerRebuildPlan { to_scan }) } - fn rebuild_backend_pass( + /// 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, - backend: ContainerBackend, context: &ContainerRebuildContext<'_>, exec_state: &mut ExecutionState, ) -> ContainerRebuildSummary { - let (selected, selected_rows, largest_env) = self + let selected = self .data .iter() - .filter(|(_, env)| env.backend() == backend) - .fold( - (0usize, 0usize, 0usize), - |(count, rows, largest), (_, env)| { - let len = env.len(); - (count + 1, rows.saturating_add(len), largest.max(len)) - }, - ); + .filter(|(_, env)| env.backend() == ContainerBackend::Legacy) + .count(); if selected == 0 { return ContainerRebuildSummary::default(); } - // Sequence environments are themselves tables, so a few large types - // are enough useful work to parallelize even when the legacy - // environment-count heuristic does not fire. Keep the aggregate-size - // alternative sequence-only, and use it only while every individual - // sequence scan remains serial. This avoids nesting outer environment - // tasks around tables that already use all workers internally; legacy - // environments likewise retain their existing nested rebuild policy. - let parallelize_large_sequence_batch = backend == ContainerBackend::Sequence - && selected > 1 - && !parallelize_rebuild(largest_env) - && parallelize_db_level_op(selected_rows); - if parallelize_inter_container_op(selected) || parallelize_large_sequence_batch { + if parallelize_inter_container_op(selected) { parallel::map_dense_id_map_mut(&mut self.data, |_, env| { - if env.backend() != backend { + if env.backend() != ContainerBackend::Legacy { return ContainerRebuildSummary::default(); } let mut exec_state = exec_state.clone(); - env.apply_rebuild( - context.table, - &*context.rebuilder, - context.to_scan.as_ref().map(|subset| subset.as_ref()), - &mut exec_state, - ) + context.apply(&mut **env, &mut exec_state) }) .into_iter() .fold(ContainerRebuildSummary::default(), |mut acc, summary| { @@ -416,15 +412,9 @@ impl ContainerValues { } else { let mut summary = ContainerRebuildSummary::default(); for (_, env) in self.data.iter_mut() { - if env.backend() != backend { - continue; + if env.backend() == ContainerBackend::Legacy { + summary.extend(context.apply(&mut **env, exec_state)); } - summary.extend(env.apply_rebuild( - context.table, - &*context.rebuilder, - context.to_scan.as_ref().map(|subset| subset.as_ref()), - exec_state, - )); } summary } @@ -435,7 +425,7 @@ impl ContainerValues { /// /// Call this once, after every backend-specific pass has completed and all /// temporarily removed environments have been restored. - fn finish_rebuild(&self, summary: &mut ContainerRebuildSummary) { + pub(crate) fn finish_rebuild(&self, summary: &mut ContainerRebuildSummary) { self.expand_dirty_id_closure(summary); } @@ -450,7 +440,7 @@ impl ContainerValues { /// inner `Vec` id is dirty; the outer `Vec` row is not retimestamped, so a /// later rule like `(rewrite (p (vec-of (vec-of (b)))) (b))` can miss the /// newly matchable parent row. - fn expand_dirty_id_closure(&self, summary: &mut ContainerRebuildSummary) { + pub(crate) fn expand_dirty_id_closure(&self, summary: &mut ContainerRebuildSummary) { let mut frontier = summary.dirty_ids.clone(); let mut seen = frontier.iter().copied().collect::>(); @@ -475,13 +465,21 @@ impl ContainerValues { /// 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::()); + 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 } @@ -489,10 +487,11 @@ impl ContainerValues { /// variable-length sequence followed by one non-key identity value. pub(crate) fn register_sequence_type( &mut self, + id: ContainerValueId, id_counter: CounterId, merge_fn: impl MergeFn + 'static, ) -> ContainerValueId { - let id = self.container_ids.insert(TypeId::of::()); + let id = self.container_ids.insert(TypeId::of::(), id); self.data.get_or_insert(id, || { Box::new(SequenceContainerEnv::::new( id, @@ -532,7 +531,6 @@ impl ContainerValues { } /// Snapshot the ids of all sequence-backed environments. - #[cfg(test)] pub(crate) fn sequence_env_ids(&self) -> Vec { self.data .iter() @@ -540,6 +538,12 @@ impl ContainerValues { .collect() } + /// 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() + } + pub(crate) fn take_env( &mut self, id: ContainerValueId, @@ -622,10 +626,18 @@ pub(crate) trait DynamicContainerEnv: Any + dyn_clone::DynClone + Send + Sync { fn as_any(&self) -> &dyn Any; fn backend(&self) -> ContainerBackend; fn len(&self) -> usize; - /// Publish one epoch of staged sequence-table mutations. - fn merge_pending(&mut self, _exec_state: &mut ExecutionState) -> bool { - false + + /// 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 + } + + fn maintenance_table_mut(&mut self) -> Option<&mut dyn MaintenanceTable> { + None } + fn apply_rebuild( &mut self, table: &WrappedTable, diff --git a/core-relations/src/containers/sequence.rs b/core-relations/src/containers/sequence.rs index 325066168..904387656 100644 --- a/core-relations/src/containers/sequence.rs +++ b/core-relations/src/containers/sequence.rs @@ -14,12 +14,13 @@ use crossbeam_queue::SegQueue; use rustc_hash::FxHasher; use crate::{ - ExecutionState, Offset, RowId, SequenceTable, Subset, TableVersion, TaggedRowBuffer, Value, + ExecutionState, Offset, RowId, SequenceTable, Subset, TableChange, TableVersion, + TaggedRowBuffer, Value, common::{HashMap, IndexSet, ShardData, ShardId}, numeric_id::{DenseIdMap, NumericId}, offsets::Offsets, parallel_heuristics::parallelize_intra_container_op, - table_spec::{Rebuilder, ValueRebuilder}, + table_spec::{MaintenanceTable, Rebuilder, ValueRebuilder}, }; use super::{ @@ -279,11 +280,11 @@ impl SequenceContainerEnv { }); } - fn merge_table(&mut self, exec_state: &mut ExecutionState) -> bool { + fn merge_table(&mut self, exec_state: &mut ExecutionState) -> TableChange { let previous = self.table.version(); let changed = self.table.merge_with_state(exec_state); self.refresh_reverse(&previous); - changed.added || changed.removed + changed } /// Resolve the union-find delta through the sequence occurrence index. @@ -339,6 +340,20 @@ impl SequenceContainerEnv { } } +impl MaintenanceTable for SequenceContainerEnv { + fn len(&self) -> usize { + self.table.len() + } + + fn new_buffer(&self) -> Box { + self.table.new_buffer() + } + + fn merge(&mut self, exec_state: &mut ExecutionState) -> TableChange { + self.merge_table(exec_state) + } +} + impl DynamicContainerEnv for SequenceContainerEnv { fn as_any(&self) -> &dyn std::any::Any { self @@ -352,8 +367,12 @@ impl DynamicContainerEnv for SequenceContainerEnv { self.table.len() } - fn merge_pending(&mut self, exec_state: &mut ExecutionState) -> bool { - self.merge_table(exec_state) + fn maintenance_table(&self) -> Option<&dyn MaintenanceTable> { + Some(self) + } + + fn maintenance_table_mut(&mut self) -> Option<&mut dyn MaintenanceTable> { + Some(self) } fn apply_rebuild( @@ -537,7 +556,10 @@ mod tests { let second = db.with_execution_state(|state| env.get_or_insert(&key, state)); assert_ne!(first, second); - db.with_execution_state(|state| assert!(env.merge_table(state))); + db.with_execution_state(|state| { + let change = env.merge_table(state); + assert!(change.added || change.removed); + }); let winner = first.min(second); assert_eq!(env.get_container(winner), Some(key)); assert_eq!(env.get_container(first.max(second)), None); @@ -552,14 +574,20 @@ mod tests { .map(|index| env.get_or_insert_key(&[value(100 + index)], state)) .collect::>() }); - db.with_execution_state(|state| assert!(env.merge_table(state))); + db.with_execution_state(|state| { + let change = env.merge_table(state); + assert!(change.added || change.removed); + }); let mut removals = env.table.new_buffer(); for index in 0..25 { removals.stage_remove(&[value(100 + index)]); } drop(removals); - db.with_execution_state(|state| assert!(env.merge_table(state))); + db.with_execution_state(|state| { + let change = env.merge_table(state); + assert!(change.added || change.removed); + }); for (index, id) in ids.into_iter().enumerate() { if index < 25 { @@ -573,6 +601,7 @@ mod tests { struct HintRebuilder { from: Value, to: Value, + hint: Option, visits: AtomicUsize, } @@ -585,7 +614,7 @@ mod tests { impl Rebuilder for HintRebuilder { fn hint_col(&self) -> Option { - Some(ColumnId::new(0)) + self.hint } fn rebuild_buf( @@ -610,6 +639,74 @@ mod tests { } } + fn empty_rebuild_table() -> WrappedTable { + WrappedTable::new(SortedWritesTable::new( + 1, + 1, + None, + vec![], + Box::new(|_, _, _, _| false), + )) + } + + #[test] + fn full_rebuild_marks_a_stable_identity_dirty() { + let from = value(10); + let to = value(20); + let mut db = Database::new(); + let mut env = test_env(&mut db); + let id = db.with_execution_state(|state| env.get_or_insert_key(&[from], state)); + db.with_execution_state(|state| { + let change = env.merge_table(state); + assert!(change.added || change.removed); + }); + + let rebuilder = HintRebuilder { + from, + to, + hint: None, + visits: AtomicUsize::new(0), + }; + let summary = db.with_execution_state(|state| { + env.apply_rebuild(&empty_rebuild_table(), &rebuilder, None, state) + }); + + assert!(summary.changed()); + assert_eq!( + summary.dirty_ids().iter().copied().collect::>(), + [id] + ); + assert_eq!(env.get_key(id), Some([to].as_slice())); + } + + #[test] + fn full_rebuild_remaps_an_outer_identity_without_marking_it_dirty() { + let child = value(30); + let mut db = Database::new(); + let mut env = test_env(&mut db); + let old_id = db.with_execution_state(|state| env.get_or_insert_key(&[child], state)); + db.with_execution_state(|state| { + let change = env.merge_table(state); + assert!(change.added || change.removed); + }); + let new_id = value(old_id.index() + 100_000); + + let rebuilder = HintRebuilder { + from: old_id, + to: new_id, + hint: None, + visits: AtomicUsize::new(0), + }; + let summary = db.with_execution_state(|state| { + env.apply_rebuild(&empty_rebuild_table(), &rebuilder, None, state) + }); + + assert!(summary.changed()); + assert!(summary.dirty_ids().is_empty()); + assert_eq!(env.get_key(old_id), None); + assert_eq!(env.get_key(new_id), Some([child].as_slice())); + } + #[test] fn incremental_rebuild_uses_value_index_candidates() { const CONTAINERS: usize = 1_002; @@ -629,7 +726,10 @@ mod tests { }) .collect::>() }); - db.with_execution_state(|state| assert!(env.merge_table(state))); + db.with_execution_state(|state| { + let change = env.merge_table(state); + assert!(change.added || change.removed); + }); let mut dirty = SortedWritesTable::new(1, 1, None, vec![], Box::new(|_, _, _, _| false)); dirty.new_buffer().stage_insert(&[from]); @@ -641,6 +741,7 @@ mod tests { let rebuilder = HintRebuilder { from, to, + hint: Some(ColumnId::new(0)), visits: AtomicUsize::new(0), }; let summary = db.with_execution_state(|state| { diff --git a/core-relations/src/containers/tests.rs b/core-relations/src/containers/tests.rs index 1e216c764..94263d5f6 100644 --- a/core-relations/src/containers/tests.rs +++ b/core-relations/src/containers/tests.rs @@ -3,13 +3,19 @@ //! This module has tests that verify specific behavior in a multithreaded setting that are harder //! to exercise deterministically when testing end to end. -use std::{iter, sync::Arc}; +use std::{ + iter, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, +}; use crate::numeric_id::NumericId; use egglog_concurrency::Notification; use crate::{ - ColumnId, Database, ExecutionState, Rebuilder, RowId, SequenceContainerValue, + ColumnId, Database, DisplacedTable, ExecutionState, Rebuilder, RowId, SequenceContainerValue, SortedWritesTable, Value, ValueRebuilder, row_buffer::RowBuffer, table_spec::WrappedTableRef, }; @@ -272,8 +278,9 @@ fn type_erased_backend_metadata_reports_sequence_lengths() { let legacy_counter = db.add_counter(); let sequence_counter = db.add_reservable_counter(8); let legacy_id = db - .container_values_mut() - .register_type::(legacy_counter, |_state, left, right| left.min(right)); + .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), @@ -304,7 +311,138 @@ fn type_erased_backend_metadata_reports_sequence_lengths() { } #[test] -fn sequence_backend_round_trips_predictions_through_database_merge() { +fn relations_and_containers_share_one_table_id_namespace() { + let mut db = Database::new(); + let first_relation = db.add_table( + SortedWritesTable::new(1, 1, None, vec![], Box::new(|_, _, _, _| false)), + iter::empty(), + iter::empty(), + ); + let counter = db.add_reservable_counter(8); + let container = + db.register_container_type::(counter, |_state, left, right| left.min(right)); + let second_relation = db.add_table( + SortedWritesTable::new(1, 1, None, vec![], Box::new(|_, _, _, _| false)), + iter::empty(), + iter::empty(), + ); + + assert_eq!(first_relation.index(), 0); + 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 + ); +} + +#[test] +fn shared_rebuild_defers_relations_until_nested_container_collisions_converge() { + let mut db = Database::new(); + let uf = db.add_table(DisplacedTable::default(), iter::empty(), iter::empty()); + 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::( + counter, + move |state, left, right| { + assert!( + state.get_table(uf).rebuilder(&[]).is_some(), + "the rebuild source must remain readable by container collision callbacks" + ); + collision_merges_for_callback.fetch_add(1, Ordering::Relaxed); + state.stage_insert(uf, &[left, right, Value::from_usize(2)]); + left.min(right) + }, + [uf], + [uf], + ); + let parents = db.add_table( + SortedWritesTable::new( + 1, + 3, + Some(ColumnId::new(2)), + vec![ColumnId::new(0), ColumnId::new(1)], + Box::new(|_, current, incoming, _| { + assert_eq!(current[1], incoming[1]); + false + }), + ), + [uf], + iter::empty(), + ); + + let child_a = Value::from_usize(10_000); + let child_b = Value::from_usize(20_000); + let tag_a = Value::from_usize(30_000); + let tag_b = Value::from_usize(40_000); + let (inner_a, inner_b, outer_a, outer_b) = db.with_execution_state(|state| { + let containers = state.container_values(); + let inner_a = containers.register_val(VecContainer(vec![child_a]), state); + let inner_b = containers.register_val(VecContainer(vec![child_b]), state); + let outer_a = containers.register_val(VecContainer(vec![inner_a]), state); + let outer_b = containers.register_val(VecContainer(vec![inner_b]), state); + state.stage_insert(parents, &[tag_a, outer_a, Value::from_usize(0)]); + state.stage_insert(parents, &[tag_b, outer_b, Value::from_usize(0)]); + state.stage_insert(uf, &[child_a, child_b, Value::from_usize(1)]); + state.stage_insert(uf, &[tag_a, tag_b, Value::from_usize(1)]); + (inner_a, inner_b, outer_a, outer_b) + }); + assert!(db.merge_all()); + + // The first container round merges the inner Vec identities. The relation + // must remain untouched because its keys would collapse while the outer + // Vec identities are still distinct. + assert!(db.apply_rebuild(uf, &[parents], Value::from_usize(2))); + assert_eq!(collision_merges.load(Ordering::Relaxed), 1); + assert!(db.get_table(parents).get_row(&[tag_a]).is_some()); + assert!(db.get_table(parents).get_row(&[tag_b]).is_some()); + + let rebuilder = db.get_table(uf).rebuilder(&[]).unwrap(); + assert_eq!( + rebuilder.rebuild_val(inner_a), + rebuilder.rebuild_val(inner_b) + ); + assert_ne!( + rebuilder.rebuild_val(outer_a), + rebuilder.rebuild_val(outer_b) + ); + drop(rebuilder); + + // The next outer round can now collide the outer Vec keys. It likewise + // defers the strict relation until that new identity union is published. + assert!(db.apply_rebuild(uf, &[parents], Value::from_usize(3))); + assert_eq!(collision_merges.load(Ordering::Relaxed), 2); + assert!(db.get_table(parents).get_row(&[tag_a]).is_some()); + assert!(db.get_table(parents).get_row(&[tag_b]).is_some()); + + // With the container/source chain stable, the third round rebuilds the + // relation from one coherent snapshot. Both key and value canonicalize + // together, so the :no-merge-style assertion above is satisfied. + assert!(db.apply_rebuild(uf, &[parents], Value::from_usize(4))); + + let (canonical_outer_a, canonical_outer_b, canonical_tag_a, canonical_tag_b) = { + let rebuilder = db.get_table(uf).rebuilder(&[]).unwrap(); + ( + rebuilder.rebuild_val(outer_a), + rebuilder.rebuild_val(outer_b), + rebuilder.rebuild_val(tag_a), + rebuilder.rebuild_val(tag_b), + ) + }; + assert_eq!(canonical_outer_a, canonical_outer_b); + assert_eq!(canonical_tag_a, canonical_tag_b); + let row = db + .get_table(parents) + .get_row(&[canonical_tag_a]) + .expect("canonical relation row must be published after source convergence"); + assert_eq!(row.vals[1], canonical_outer_a); + assert_eq!(db.get_table(parents).len(), 1); +} + +#[test] +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::( @@ -421,7 +559,7 @@ fn sequence_merge_writes_participate_in_simple_fixed_point() { state.stage_insert(sink, &[left, right]); left.min(right) }, - [gate.into()], + [gate], [sink], ); db.with_execution_state(|state| { @@ -481,7 +619,7 @@ fn sequence_merge_writes_participate_in_dependency_aware_fixed_point() { state.stage_insert(sink, &[left, right]); left.min(right) }, - [gate.into()], + [gate], [sink], ); @@ -526,11 +664,11 @@ fn relation_merge_can_depend_on_nonzero_level_sequence_container() { let container = db.register_sequence_container_type::( counter, |_state, left, right| left.min(right), - [gate.into()], + [gate], iter::empty(), ); let expected = cont([11, 12]); - let relation = db.add_table_with_maintenance_deps( + let relation = db.add_table( SortedWritesTable::new( 1, 2, @@ -539,16 +677,16 @@ fn relation_merge_can_depend_on_nonzero_level_sequence_container() { Box::new({ let expected = expected.clone(); move |state, current, incoming, out| { - let winner = current[0].min(incoming[0]); + let winner = current[1].min(incoming[1]); assert_eq!( state.container_sequence::(winner), Some(expected.0.as_slice()), "the relation merge must run after its sequence dependency" ); - if current[0] == winner { + if current[1] == winner { false } else { - out.push(winner); + out.extend_from_slice(&[current[0], winner]); true } } @@ -580,6 +718,73 @@ fn relation_merge_can_depend_on_nonzero_level_sequence_container() { assert_eq!(db.get_table(relation).len(), 1); } +#[test] +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::( + counter, + |_state, left, right| left.min(right), + iter::empty(), + iter::empty(), + ); + let relation = db.add_table( + SortedWritesTable::new( + 1, + 2, + None, + vec![], + Box::new(|state, current, incoming, _| { + assert!( + state + .container_sequence::(current[1]) + .is_some(), + "a relation merge must be able to read an active container" + ); + assert!( + state + .container_sequence::(incoming[1]) + .is_some(), + "a relation merge must be able to read an active container" + ); + false + }), + ), + iter::empty(), + iter::empty(), + ); + let mut add_bystander = || { + db.add_table( + SortedWritesTable::new(1, 1, None, vec![], Box::new(|_, _, _, _| false)), + iter::empty(), + iter::empty(), + ) + }; + let first_bystander = add_bystander(); + let second_bystander = add_bystander(); + let first = db.with_execution_state(|state| { + let containers = state.container_values(); + containers.register_val(cont([21]), state) + }); + let second = db.with_execution_state(|state| { + let containers = state.container_values(); + containers.register_val(cont([22]), state) + }); + db.with_execution_state(|state| { + state.stage_insert(relation, &[Value::from_usize(0), first]); + state.stage_insert(relation, &[Value::from_usize(0), second]); + state.stage_insert(first_bystander, &[Value::from_usize(1)]); + state.stage_insert(second_bystander, &[Value::from_usize(2)]); + }); + + // Four active same-level participants force the stratum path. Container + // reads by arbitrary primitive callbacks are ambient rather than declared + // in the dependency graph, so the environment must remain installed while + // the relation merge runs. + assert!(db.merge_all()); + assert_eq!(db.get_table(relation).len(), 1); +} + #[test] fn failed_table_dependency_registration_does_not_mutate_database() { let mut db = Database::new(); @@ -615,7 +820,7 @@ fn failed_sequence_dependency_registration_can_be_retried() { db.register_sequence_container_type::( counter, |_state, left, right| left.min(right), - [missing_table.into()], + [missing_table], iter::empty(), ); })); @@ -640,7 +845,7 @@ fn failed_sequence_dependency_registration_can_be_retried() { let duplicate = db.register_sequence_container_type::( counter, |_state, left, right| left.max(right), - [missing_table.into()], + [missing_table], [missing_table], ); assert_eq!(duplicate, container, "the first registration owns the deps"); diff --git a/core-relations/src/dependency_graph.rs b/core-relations/src/dependency_graph.rs index bd67a2ef6..ddab3c7eb 100644 --- a/core-relations/src/dependency_graph.rs +++ b/core-relations/src/dependency_graph.rs @@ -1,20 +1,14 @@ //! A simple data-structure for tracking the dependencies of the merge functions //! from different tables on one another. -use crate::numeric_id::{DenseIdMap, define_id}; +use crate::numeric_id::DenseIdMap; use crate::{TableId, common::IndexSet}; -define_id!( - pub(crate) MaintenanceId, - u32, - "an identifier for a database maintenance participant" -); - #[derive(Clone, Default)] pub(crate) struct DependencyGraph { - to_level: DenseIdMap, - write_deps: DenseIdMap>, + to_level: DenseIdMap, + write_deps: DenseIdMap>, } impl DependencyGraph { @@ -23,16 +17,16 @@ impl DependencyGraph { /// Participants can have two kinds of dependencies: /// 1. Read dependencies are participants whose storage must remain readable /// during this participant's merge function. - /// 2. Write dependencies are fixed-arity tables that must merely be - /// writable during this participant's merge function. + /// 2. Write dependencies are participants that must merely be writable + /// during this participant's merge function. /// /// Write dependencies are generally weaker than read dependencies. Two tables with write /// dependencies on one another can run their merge operations in parallel. The same is not /// true for read dependencies. pub(crate) fn add_participant( &mut self, - participant: MaintenanceId, - read_deps: impl IntoIterator, + participant: TableId, + read_deps: impl IntoIterator, write_deps: impl IntoIterator, ) { self.write_deps @@ -49,14 +43,19 @@ impl DependencyGraph { self.to_level.insert(participant, level); } - pub(crate) fn level(&self, participant: MaintenanceId) -> usize { + pub(crate) fn contains(&self, participant: TableId) -> bool { + self.to_level.contains_key(participant) + } + + pub(crate) fn participants(&self) -> impl Iterator + '_ { + self.to_level.iter().map(|(participant, _)| participant) + } + + pub(crate) fn level(&self, participant: TableId) -> usize { self.to_level[participant] } - pub(crate) fn write_deps( - &self, - participant: MaintenanceId, - ) -> impl Iterator + '_ { + pub(crate) fn write_deps(&self, participant: TableId) -> impl Iterator + '_ { self.write_deps .get(participant) .into_iter() @@ -71,8 +70,8 @@ mod tests { #[test] fn maintenance_levels_keep_table_write_dependencies() { - let first = MaintenanceId::new(0); - let second = MaintenanceId::new(1); + let first = TableId::new(0); + let second = TableId::new(1); let sink = TableId::new(7); let mut graph = DependencyGraph::default(); @@ -84,4 +83,18 @@ mod tests { assert_eq!(graph.write_deps(first).collect::>(), vec![sink]); assert!(graph.write_deps(second).next().is_none()); } + + #[test] + fn participant_ids_may_have_legacy_storage_holes() { + let mut graph = DependencyGraph::default(); + graph.add_participant(TableId::new(3), [], []); + graph.add_participant(TableId::new(7), [TableId::new(3)], []); + + assert!(!graph.contains(TableId::new(0))); + assert!(graph.contains(TableId::new(3))); + assert_eq!( + graph.participants().collect::>(), + [TableId::new(3), TableId::new(7)] + ); + } } diff --git a/core-relations/src/free_join/mod.rs b/core-relations/src/free_join/mod.rs index ac1b901ba..7b9f8ac2b 100644 --- a/core-relations/src/free_join/mod.rs +++ b/core-relations/src/free_join/mod.rs @@ -22,16 +22,20 @@ use crate::{ Bindings, DbView, mask::{Mask, MaskIter, ValueSource}, }, - containers::SequenceContainerValue, - dependency_graph::{DependencyGraph, MaintenanceId}, + containers::{ + ContainerRebuildContext, ContainerRebuildPlan, ContainerRebuildSummary, ContainerValue, + DynamicContainerEnv, SequenceContainerValue, + }, + dependency_graph::DependencyGraph, hash_index::{ColumnIndex, Index, IndexBase}, offsets::Subset, parallel, - parallel_heuristics::parallelize_db_level_op, + parallel_heuristics::{parallelize_db_level_op, parallelize_rebuild}, pool::{Pool, Pooled, with_pool_set}, query::{Query, RuleSetBuilder}, table_spec::{ - ColumnId, Constraint, MutationBuffer, Table, TableSpec, WrappedTable, WrappedTableRef, + ColumnId, Constraint, MaintenanceTable, MutationBuffer, Table, TableSpec, WrappedTable, + WrappedTableRef, }, }; @@ -58,7 +62,11 @@ impl Variable { } } -define_id!(pub TableId, u32, "a table in the database"); +define_id!( + pub TableId, + u32, + "a fixed- or variable-arity table in the database" +); impl TableId { pub fn dummy() -> TableId { @@ -70,37 +78,6 @@ impl TableId { } } -/// A storage participant that must be readable before another participant's -/// merge callback runs. -/// -/// Write dependencies remain relation-table ids because they are used to -/// preallocate [`MutationBuffer`]s while a relation stratum is temporarily -/// removed from the database. Sequence-container buffers are initialized -/// lazily and their environments remain installed during relation merges. -/// -/// Dependencies must refer to participants that were previously registered in -/// the same [`Database`]. This registration order makes the dependency graph -/// acyclic. A [`MaintenanceReadDependency::SequenceContainer`] must name a -/// sequence-backed container; legacy container environments are not scheduler -/// participants. -#[derive(Clone, Copy, Debug, Eq, PartialEq)] -pub enum MaintenanceReadDependency { - Table(TableId), - SequenceContainer(crate::ContainerValueId), -} - -impl From for MaintenanceReadDependency { - fn from(table: TableId) -> Self { - Self::Table(table) - } -} - -impl From for MaintenanceReadDependency { - fn from(container: crate::ContainerValueId) -> Self { - Self::SequenceContainer(container) - } -} - define_id!(pub(crate) ActionId, u32, "an identifier picking out the RHS of a rule"); #[derive(Debug)] @@ -402,18 +379,21 @@ impl Counters { pub struct Database { // NB: some fields are pub(crate) to allow some internal modules to avoid // borrowing the whole table. + // + // TableId is shared with container tables, so this relation-only map may + // contain holes occupied by `ContainerValues`. pub(crate) tables: DenseIdMap, // Reservable counters amortize shared atomic increments across an // ExecutionState. Exact counters retain one atomic increment per value. pub(crate) counters: Counters, pub(crate) external_functions: ExternalFunctions, container_values: ContainerValues, - /// Maps the common maintenance namespace to its concrete storage owner. - maintenance_targets: DenseIdMap, - table_maintenance: DenseIdMap, - container_maintenance: DenseIdMap, + /// 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, + notification_list: NotificationList, // Tracks relative dependencies between maintenance participants. deps: DependencyGraph, base_values: BaseValues, @@ -424,10 +404,81 @@ pub struct Database { total_size_estimate: usize, } -#[derive(Clone, Copy, Debug)] -enum MaintenanceTarget { - Relation(TableId), - SequenceContainer(crate::ContainerValueId), +/// A maintenance participant temporarily removed from its typed owner map. +/// +/// This is only an ownership adapter for constructing a read-only database +/// view while a mixed relation/container stratum is merged. Persistent +/// identity, notifications, buffers, and dependencies all use [`TableId`]. +enum OwnedParticipant { + Relation { + id: TableId, + info: TableInfo, + }, + Container { + id: crate::ContainerValueId, + env: Box, + }, +} + +struct ParticipantWork { + participant: OwnedParticipant, + buffers: DenseIdMap>, +} + +#[derive(Default)] +struct ParticipantRebuild { + changed: bool, + containers: ContainerRebuildSummary, +} + +impl ParticipantRebuild { + fn extend(&mut self, other: Self) { + self.changed |= other.changed; + self.containers.extend(other.containers); + } +} + +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"), + } + } + + fn merge(&mut self, exec_state: &mut ExecutionState<'_>) -> bool { + let change = self.maintenance_table_mut().merge(exec_state); + change.added || change.removed || exec_state.changed + } + + fn apply_rebuild( + &mut self, + table_id: TableId, + table: &WrappedTable, + next_ts: Value, + container_context: Option<&ContainerRebuildContext<'_>>, + exec_state: &mut ExecutionState<'_>, + ) -> ParticipantRebuild { + match self { + OwnedParticipant::Relation { info, .. } => ParticipantRebuild { + changed: info + .table + .apply_rebuild(table_id, table, next_ts, exec_state), + containers: ContainerRebuildSummary::default(), + }, + OwnedParticipant::Container { env, .. } => { + let containers = container_context + .map(|context| context.apply(&mut **env, exec_state)) + .unwrap_or_default(); + ParticipantRebuild { + changed: containers.changed(), + containers, + } + } + } + } } impl Clone for Database { @@ -440,7 +491,7 @@ impl Clone for Database { // the source queue, and harmlessly performs one empty merge for // participants that had no work. let notification_list = NotificationList::default(); - for (participant, _) in self.maintenance_targets.iter() { + for participant in self.deps.participants() { notification_list.notify(participant); } Self { @@ -448,9 +499,7 @@ impl Clone for Database { counters: self.counters.clone(), external_functions: self.external_functions.clone(), container_values: self.container_values.clone(), - maintenance_targets: self.maintenance_targets.clone(), - table_maintenance: self.table_maintenance.clone(), - container_maintenance: self.container_maintenance.clone(), + next_storage_id: self.next_storage_id, notification_list, deps: self.deps.clone(), base_values: self.base_values.clone(), @@ -502,6 +551,30 @@ 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 /// participant. Its packed rows remain outside the fixed-schema query /// table interface, but merge notifications and dependency ordering share @@ -517,126 +590,247 @@ impl Database { &mut self, id_counter: CounterId, merge_fn: impl Fn(&mut ExecutionState, Value, Value) -> Value + Clone + Send + Sync + 'static, - read_deps: impl IntoIterator, + read_deps: impl IntoIterator, write_deps: impl IntoIterator, ) -> crate::ContainerValueId { if let Some(container) = self.container_values.registered_sequence_type::() { - let existing = self - .container_maintenance - .get(container) - .expect("registered sequence container must have a maintenance target"); - debug_assert!(matches!( - self.maintenance_targets[*existing], - MaintenanceTarget::SequenceContainer(id) if id == container - )); return container; } let read_deps = read_deps.into_iter().collect::>(); let write_deps = write_deps.into_iter().collect::>(); - let read_deps = read_deps - .into_iter() - .map(|dependency| self.resolve_maintenance_dependency(dependency)) - .collect::>(); - self.validate_write_dependencies(&write_deps); + self.validate_dependencies(&read_deps, "read"); + self.validate_dependencies(&write_deps, "write"); - let container = self + let participant = self.allocate_storage_id(); + let container = crate::ContainerValueId::from_table_id(participant); + let registered = self .container_values - .register_sequence_type::(id_counter, merge_fn); - let maintenance = self - .maintenance_targets - .push(MaintenanceTarget::SequenceContainer(container)); - self.container_maintenance.insert(container, maintenance); + .register_sequence_type::(container, id_counter, merge_fn); + assert_eq!(registered, container); self.deps - .add_participant(maintenance, read_deps, write_deps); + .add_participant(participant, read_deps, write_deps); container } - fn resolve_maintenance_dependency( - &self, - dependency: MaintenanceReadDependency, - ) -> MaintenanceId { - match dependency { - MaintenanceReadDependency::Table(table) => *self - .table_maintenance - .get(table) - .unwrap_or_else(|| panic!("table dependency {table:?} is not registered")), - MaintenanceReadDependency::SequenceContainer(container) => *self - .container_maintenance - .get(container) - .unwrap_or_else(|| { - panic!("sequence-container dependency {container:?} is not registered") - }), - } - } - - fn validate_write_dependencies(&self, dependencies: &[TableId]) { - for table in dependencies { + fn validate_dependencies(&self, dependencies: &[TableId], kind: &str) { + for participant in dependencies { assert!( - self.table_maintenance.get(*table).is_some(), - "table write dependency {table:?} is not registered" + self.deps.contains(*participant), + "maintenance {kind} dependency {participant:?} is not registered" ); } } - /// Apply one ordered database-maintenance cycle using the value-level - /// rebuild encoded by `func_id`. + 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 as relation tables; legacy environments remain a rebuild-only - /// compatibility pass. Container canonicalization runs first because key - /// collisions can stage ordinary union-find writes. The common - /// [`Database::merge_all`] fixed point publishes those writes before - /// dependent fixed-schema relations rebuild. Stable container identities - /// are then propagated through the temporary opaque-read refresh path. + /// 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. + /// + /// 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. pub fn apply_rebuild( &mut self, func_id: TableId, to_rebuild: &[TableId], next_ts: Value, ) -> bool { - let container_rebuild = { - let mut containers = mem::take(&mut self.container_values); - let table = &self.tables[func_id].table; - let result = - self.with_execution_state(|state| containers.rebuild_all(func_id, table, state)); - self.container_values = containers; - result - }; - - let mut changed = container_rebuild.changed(); - // This is a correctness barrier, not a separate fixed point: relation - // rebuilds must observe identity unions produced by container-key - // collisions (notably for :no-merge functions). - changed |= self.merge_all(); - - let func = self.tables.take(func_id).unwrap(); - self.run_on_tables(to_rebuild, |_, info, view| { - info.table.apply_rebuild( - func_id, - &func.table, - next_ts, - &mut ExecutionState::new(*view, Default::default()), - ) - }); - self.tables.insert(func_id, func); - // Ordinary table rebuilds stage their replacements. Publish those - // rows before dirty-id refresh scans the tables, otherwise the refresh - // can re-stage an obsolete pre-rebuild row. - changed |= self.merge_all(); + let containers = self + .container_values + .sequence_env_ids() + .into_iter() + .map(TableId::from) + .collect::>(); + let relations = to_rebuild.to_vec(); + assert!( + !containers.contains(&func_id) && !relations.contains(&func_id), + "the rebuild source cannot also be a mutable rebuild participant" + ); + let container_plan = self + .container_values + .prepare_rebuild_plan(func_id, &self.tables[func_id].table); + let source_before = self.tables[func_id].table.version(); + + 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 + // two still-distinct container values and spuriously invoke :no-merge. + rebuild.changed |= self.merge_all(); + if self.tables[func_id].table.version() != source_before { + // A nested container can expose another collision only after this + // source delta is visible. Preserve stable-id notifications from + // this round, then let the existing outer rebuild loop run the + // container batch again before any fixed-table observer proceeds. + self.refresh_dirty_container_parents(to_rebuild, next_ts, &mut rebuild); + return true; + } + rebuild.extend(self.rebuild_batch(relations, func_id, next_ts, None)); - let dirty_ids = container_rebuild + // Publish every rebuilt participant before dirty-id refresh scans the + // relation tables; otherwise it can re-stage an obsolete pre-rebuild + // row. + rebuild.changed |= self.merge_all(); + self.refresh_dirty_container_parents(to_rebuild, next_ts, &mut rebuild); + rebuild.changed + } + + fn refresh_dirty_container_parents( + &mut self, + to_rebuild: &[TableId], + next_ts: Value, + rebuild: &mut ParticipantRebuild, + ) { + self.container_values + .finish_rebuild(&mut rebuild.containers); + let dirty_ids = rebuild + .containers .dirty_ids() .iter() .copied() .collect::>(); - if !dirty_ids.is_empty() { - self.run_on_tables(to_rebuild, |_, info, _| { - info.table.refresh_rows_for_values(&dirty_ids, next_ts) - }); - changed |= self.merge_all(); + if dirty_ids.is_empty() { + return; } - changed + self.run_on_tables(to_rebuild, |_, info, _| { + info.table.refresh_rows_for_values(&dirty_ids, next_ts) + }); + rebuild.changed |= self.merge_all(); + } + + /// Order one homogeneous rebuild batch and run it through the common + /// fixed-/variable-arity participant implementation. + /// + /// Destination buffers are constructed after any preceding publication + /// barrier, so none can retain a detached pending epoch across that barrier. + fn rebuild_batch( + &mut self, + mut participants: Vec, + func_id: TableId, + next_ts: Value, + container_plan: Option<&ContainerRebuildPlan>, + ) -> ParticipantRebuild { + participants.sort_unstable(); + participants.dedup(); + participants.sort_unstable_by_key(|participant| self.deps.level(*participant)); + + self.rebuild_participants(&participants, func_id, next_ts, container_plan) + } + + /// Rebuild a mixed set of fixed- and variable-arity participants through + /// one dependency-ordered traversal. + /// + /// Only one read-dependency stratum is detached at a time. Consequently a + /// callback can still read every declared predecessor, including the + /// rebuild source itself, while same-level participants remain eligible + /// for outer parallelism. + fn rebuild_participants( + &mut self, + scheduled: &[TableId], + func_id: TableId, + next_ts: Value, + container_plan: Option<&ContainerRebuildPlan>, + ) -> ParticipantRebuild { + let mut result = ParticipantRebuild::default(); + let mut stratum_start = 0usize; + while stratum_start < scheduled.len() { + let level = self.deps.level(scheduled[stratum_start]); + let stratum_end = scheduled[stratum_start..] + .iter() + .position(|participant| self.deps.level(*participant) != level) + .map_or(scheduled.len(), |offset| stratum_start + offset); + let stratum = &scheduled[stratum_start..stratum_end]; + + let (rows, largest) = + stratum + .iter() + .fold((0usize, 0usize), |(rows, largest), participant| { + let len = self.maintenance_table(*participant).len(); + (rows.saturating_add(len), largest.max(len)) + }); + let do_parallel = + stratum.len() > 1 && !parallelize_rebuild(largest) && parallelize_db_level_op(rows); + + let rebuild = |db: DbView<'_>, work: &mut ParticipantWork| { + let source = &db.table_info[func_id].table; + let container_context = container_plan.and_then(|plan| plan.context(source)); + let mut exec_state = ExecutionState::new(db, mem::take(&mut work.buffers)); + let mut outcome = work.participant.apply_rebuild( + func_id, + source, + next_ts, + container_context.as_ref(), + &mut exec_state, + ); + outcome.changed |= exec_state.changed; + outcome + }; + let outcomes = if do_parallel { + self.with_participants(stratum, |db, rebuilding| { + parallel::map_mut(rebuilding, |_, work| rebuild(db, work)) + }) + } else { + // A serial rebuild does not need to detach an entire stratum. + // Keeping installed destinations in the DbView lets their + // mutation buffers remain lazy, matching ordinary rule + // execution and avoiding per-stratum setup proportional to the + // number of declared write dependencies. + stratum + .iter() + .map(|participant| { + self.with_participants(std::slice::from_ref(participant), |db, work| { + rebuild(db, &mut work[0]) + }) + }) + .collect::>() + }; + + for (participant, outcome) in stratum.iter().copied().zip(outcomes) { + if outcome.changed { + self.notification_list.notify(participant); + } + result.extend(outcome); + } + stratum_start = stratum_end; + } + result } fn run_on_tables( @@ -652,7 +846,7 @@ impl Database { let view = self.read_only_view(); parallel::for_each_mut(&mut tables, |_, (id, info)| { if run(*id, info, &view) { - self.notification_list.notify(self.table_maintenance[*id]); + self.notification_list.notify(*id); } }); for (id, info) in tables { @@ -666,7 +860,7 @@ impl Database { run(*id, &mut info, &view) }; if changed { - self.notification_list.notify(self.table_maintenance[*id]); + self.notification_list.notify(*id); } self.tables.insert(*id, info); } @@ -699,8 +893,6 @@ impl Database { bases: &self.base_values, containers: &self.container_values, notification_list: &self.notification_list, - table_maintenance: &self.table_maintenance, - container_maintenance: &self.container_maintenance, } } @@ -760,7 +952,7 @@ impl Database { } /// A helper for merging all pending updates. Used to write to the database after updates have - /// been staged. Returns true if any tuples were added. + /// been staged. Returns true if any participant changed or staged downstream work. /// /// Exposed for testing purposes. /// @@ -775,12 +967,10 @@ impl Database { loop { let mut active = self.notification_list.reset(); touched.extend( - active.iter().filter_map(|participant| { - match self.maintenance_targets[*participant] { - MaintenanceTarget::Relation(table) => Some(table), - MaintenanceTarget::SequenceContainer(_) => None, - } - }), + active + .iter() + .copied() + .filter(|participant| self.tables.contains_key(*participant)), ); if active.len() < 4 { ever_changed |= self.merge_simple(active, &mut touched); @@ -790,11 +980,6 @@ impl Database { let mut changed = false; let mut stratum_start = 0; - let mut tables_merging = Vec::<( - TableId, - Option, - DenseIdMap>, - )>::new(); while stratum_start < active.len() { let level = self.deps.level(active[stratum_start]); let stratum_end = active[stratum_start..] @@ -803,77 +988,28 @@ impl Database { .map_or(active.len(), |offset| stratum_start + offset); let stratum = &active[stratum_start..stratum_end]; - // Keep sequence environments locally owned and merge them one - // at a time. This leaves every other environment visible to a - // merge callback while still using the common scheduler and - // notification fixed point. - let sequence_participants = stratum + // Primitive merge callbacks can read containers without those + // reads appearing in the dependency graph. Preserve that + // existing API contract by publishing each active container + // before removing any same-stratum relations. Containers run + // one at a time so every other container also remains visible. + // This is only a visibility phase: both kinds still use the + // same ids, dependency graph, buffers, notifications, and + // participant merge implementation. + let containers = stratum .iter() .copied() - .filter_map(|participant| match self.maintenance_targets[participant] { - MaintenanceTarget::SequenceContainer(container) => { - Some((participant, container)) - } - MaintenanceTarget::Relation(_) => None, - }) + .filter(|participant| !self.tables.contains_key(*participant)) .collect::>(); - for (participant, container) in sequence_participants { - let mut bufs = DenseIdMap::default(); - for dep in self.deps.write_deps(participant) { - if let Some(info) = self.tables.get(dep) { - bufs.insert(dep, info.table.new_buffer()); - } - } - changed |= self.merge_sequence_container(container, bufs); - } - let relation_participants = stratum + let relations = stratum .iter() .copied() - .filter_map(|participant| match self.maintenance_targets[participant] { - MaintenanceTarget::Relation(table) => Some((participant, table)), - MaintenanceTarget::SequenceContainer(_) => None, - }) + .filter(|participant| self.tables.contains_key(*participant)) .collect::>(); - tables_merging.reserve(relation_participants.len()); - for (participant, table) in &relation_participants { - let mut bufs = DenseIdMap::default(); - for dep in self.deps.write_deps(*participant) { - if let Some(info) = self.tables.get(dep) { - bufs.insert(dep, info.table.new_buffer()); - } - } - tables_merging.push((*table, None, bufs)); - } - for (table, info, _) in &mut tables_merging { - let value = self.tables.unwrap_val(*table); - self.total_size_estimate = - self.total_size_estimate.wrapping_sub(value.table.len()); - *info = Some(value); - } - let db = self.read_only_view(); - changed |= if do_parallel { - parallel::map_mut(&mut tables_merging, |_, (_, info, buffers)| { - let mut es = ExecutionState::new(db, mem::take(buffers)); - info.as_mut().unwrap().table.merge(&mut es).added || es.changed - }) - .into_iter() - .any(|changed| changed) - } else { - tables_merging - .iter_mut() - .map(|(_, info, buffers)| { - let mut es = ExecutionState::new(db, mem::take(buffers)); - info.as_mut().unwrap().table.merge(&mut es).added || es.changed - }) - .max() - .unwrap_or(false) - }; - for (id, table, _) in tables_merging.drain(..) { - let value = table.unwrap(); - self.total_size_estimate = - self.total_size_estimate.wrapping_add(value.table.len()); - self.tables.insert(id, value); + for participant in containers { + changed |= self.merge_participants(std::slice::from_ref(&participant), false); } + changed |= self.merge_participants(&relations, do_parallel); stratum_start = stratum_end; } ever_changed |= changed; @@ -904,69 +1040,156 @@ impl Database { ever_changed } - fn merge_sequence_container( + fn maintenance_table(&self, participant: TableId) -> &dyn MaintenanceTable { + if let Some(info) = self.tables.get(participant) { + return &info.table; + } + self.container_values + .maintenance_table(crate::ContainerValueId::from_table_id(participant)) + .unwrap_or_else(|| panic!("maintenance participant {participant:?} has no storage")) + } + + fn write_buffers( + &self, + participant: TableId, + detached: Option<&DenseIdMap>, + ) -> DenseIdMap> { + let mut buffers = DenseIdMap::default(); + for dependency in self.deps.write_deps(participant) { + let is_detached = detached.map_or(dependency == participant, |participants| { + participants.contains_key(dependency) + }); + if is_detached { + buffers.insert(dependency, self.maintenance_table(dependency).new_buffer()); + } + } + buffers + } + + /// Temporarily own a participant group while retaining a read-only view of + /// every other table. + /// + /// Merge and rebuild both use this ownership protocol. Only destinations + /// in the detached group need buffers created up front; installed relation + /// and container destinations are initialized lazily through [`DbView`]. + /// This keeps same-group writes attached to the destination's current + /// pending epoch without paying for declared writes that never occur. + fn with_participants( &mut self, - container: crate::ContainerValueId, - buffers: DenseIdMap>, - ) -> bool { - let mut env = self.container_values.take_env(container); - let changed = { - let mut exec_state = ExecutionState::new(self.read_only_view(), buffers); - env.merge_pending(&mut exec_state) || exec_state.changed - }; - self.container_values.put_env(container, env); - changed + participants: &[TableId], + run: impl for<'a> FnOnce(DbView<'a>, &mut [ParticipantWork]) -> R, + ) -> R { + if let [participant] = participants { + let buffers = self.write_buffers(*participant, None); + let mut work = ParticipantWork { + participant: self.take_participant(*participant), + buffers, + }; + let result = run(self.read_only_view(), std::slice::from_mut(&mut work)); + self.put_participant(work.participant); + return result; + } + + let mut detached = DenseIdMap::with_capacity(participants.len()); + for participant in participants { + detached.insert(*participant, ()); + } + let buffers = participants + .iter() + .map(|participant| self.write_buffers(*participant, Some(&detached))) + .collect::>(); + let mut work = participants + .iter() + .copied() + .zip(buffers) + .map(|(participant, buffers)| ParticipantWork { + participant: self.take_participant(participant), + buffers, + }) + .collect::>(); + let result = run(self.read_only_view(), &mut work); + for work in work { + self.put_participant(work.participant); + } + result + } + + /// Merge one visibility-compatible group through the common participant + /// interface, optionally in parallel. + /// + /// Declared destination buffers are created before any participant is + /// removed. This retains access to a same-group destination's pending + /// state while its storage is temporarily owned by another worker. + fn merge_participants(&mut self, participants: &[TableId], do_parallel: bool) -> bool { + self.with_participants(participants, |db, merging| { + let merge_one = |work: &mut ParticipantWork| { + let mut es = ExecutionState::new(db, mem::take(&mut work.buffers)); + work.participant.merge(&mut es) + }; + if do_parallel { + parallel::map_mut(merging, |_, work| merge_one(work)) + .into_iter() + .any(|changed| changed) + } else { + merging.iter_mut().map(merge_one).max().unwrap_or(false) + } + }) + } + + fn take_participant(&mut self, participant: TableId) -> OwnedParticipant { + if let Some(info) = self.tables.take(participant) { + self.total_size_estimate = self.total_size_estimate.wrapping_sub(info.table.len()); + return OwnedParticipant::Relation { + id: participant, + info, + }; + } + let id = crate::ContainerValueId::from_table_id(participant); + OwnedParticipant::Container { + id, + env: self.container_values.take_env(id), + } + } + + fn put_participant(&mut self, participant: OwnedParticipant) { + match participant { + OwnedParticipant::Relation { id, info } => { + self.total_size_estimate = self.total_size_estimate.wrapping_add(info.table.len()); + let old = self.tables.insert(id, info); + assert!(old.is_none(), "relation maintenance storage inserted twice"); + } + OwnedParticipant::Container { id, env } => { + self.container_values.put_env(id, env); + } + } } /// A serial fast path for a small number of notified participants. /// /// It follows read-dependency strata, but avoids taking a whole stratum out - /// of the database and preallocating its declared write buffers. With only - /// a few participants, leaving each destination table installed lets - /// [`ExecutionState`] initialize those buffers lazily. + /// of the database. Declared write buffers are still initialized before + /// detaching the participant, which also supports self-directed writes. fn merge_simple( &mut self, - mut to_merge: SmallVec<[MaintenanceId; 4]>, + mut to_merge: SmallVec<[TableId; 4]>, touched: &mut IndexSet, ) -> bool { let mut changed = false; while !to_merge.is_empty() { - // Even this small serial path must respect read dependencies: - // sequence merge callbacks are allowed to observe an ordinary - // table declared as a predecessor. Within one stratum, run - // sequence storage first so relation merges can consume writes it - // publishes without waiting for another fixed-point turn. to_merge.sort_unstable_by_key(|participant| { - let target_order = match self.maintenance_targets[*participant] { - MaintenanceTarget::SequenceContainer(_) => 0usize, - MaintenanceTarget::Relation(_) => 1usize, - }; - (self.deps.level(*participant), target_order) + let relation_order = usize::from(self.tables.contains_key(*participant)); + (self.deps.level(*participant), relation_order) }); - for participant in to_merge.iter().copied() { - match self.maintenance_targets[participant] { - MaintenanceTarget::SequenceContainer(container) => { - changed |= self.merge_sequence_container(container, Default::default()); - } - MaintenanceTarget::Relation(table) => { - let mut info = self.tables.unwrap_val(table); - self.total_size_estimate = - self.total_size_estimate.wrapping_sub(info.table.len()); - let mut es = ExecutionState::new(self.read_only_view(), Default::default()); - changed |= info.table.merge(&mut es).added || es.changed; - self.total_size_estimate = - self.total_size_estimate.wrapping_add(info.table.len()); - self.tables.insert(table, info); - } - } + for participant in &to_merge { + changed |= self.merge_participants(std::slice::from_ref(participant), false); } to_merge = self.notification_list.reset(); - touched.extend(to_merge.iter().filter_map( - |participant| match self.maintenance_targets[*participant] { - MaintenanceTarget::Relation(table) => Some(table), - MaintenanceTarget::SequenceContainer(_) => None, - }, - )); + touched.extend( + to_merge + .iter() + .copied() + .filter(|participant| self.tables.contains_key(*participant)), + ); } changed } @@ -994,39 +1217,19 @@ impl Database { /// This can be useful for "knot tying", when tables need to reference their /// own id. pub fn next_table_id(&self) -> TableId { - self.tables.next_id() + TableId::from_usize(self.next_storage_id) } /// Add a table with the given schema to the database. /// /// The table must have a compatible spec with `types` (e.g. same number of - /// columns). Every read and write dependency must identify a table already - /// registered in this database. + /// columns). Every read and write dependency must identify a maintenance + /// participant already registered in this database. pub fn add_table( &mut self, table: T, read_deps: impl IntoIterator, write_deps: impl IntoIterator, - ) -> TableId { - self.add_table_impl( - table, - None, - read_deps.into_iter().map(MaintenanceReadDependency::from), - write_deps, - ) - } - - /// Add a table whose merge callback may read relation tables, sequence - /// containers, or both. - /// - /// Every read and write dependency must already be registered in this - /// database. This ordering keeps the graph acyclic, and container read - /// dependencies must use the sequence backend. - pub fn add_table_with_maintenance_deps( - &mut self, - table: T, - read_deps: impl IntoIterator, - write_deps: impl IntoIterator, ) -> TableId { self.add_table_impl(table, None, read_deps, write_deps) } @@ -1038,22 +1241,6 @@ impl Database { name: Arc, read_deps: impl IntoIterator, write_deps: impl IntoIterator, - ) -> TableId { - self.add_table_impl( - table, - Some(name), - read_deps.into_iter().map(MaintenanceReadDependency::from), - write_deps, - ) - } - - /// Named variant of [`Database::add_table_with_maintenance_deps`]. - pub fn add_table_named_with_maintenance_deps( - &mut self, - table: T, - name: Arc, - read_deps: impl IntoIterator, - write_deps: impl IntoIterator, ) -> TableId { self.add_table_impl(table, Some(name), read_deps, write_deps) } @@ -1062,32 +1249,32 @@ impl Database { &mut self, table: T, name: Option>, - read_deps: impl IntoIterator, + read_deps: impl IntoIterator, write_deps: impl IntoIterator, ) -> TableId { let read_deps = read_deps.into_iter().collect::>(); let write_deps = write_deps.into_iter().collect::>(); - let read_deps = read_deps - .into_iter() - .map(|dependency| self.resolve_maintenance_dependency(dependency)) - .collect::>(); - self.validate_write_dependencies(&write_deps); + self.validate_dependencies(&read_deps, "read"); + self.validate_dependencies(&write_deps, "write"); let spec = table.spec(); let table = WrappedTable::new(table); - let res = self.tables.push(TableInfo { - name, - spec, - table, - indexes: IndexCatalog::new(), - column_indexes: IndexCatalog::new(), - }); - let maintenance = self - .maintenance_targets - .push(MaintenanceTarget::Relation(res)); - self.table_maintenance.insert(res, maintenance); - self.deps - .add_participant(maintenance, read_deps, write_deps); + let res = self.allocate_storage_id(); + let old = self.tables.insert( + res, + TableInfo { + name, + spec, + table, + indexes: IndexCatalog::new(), + column_indexes: IndexCatalog::new(), + }, + ); + assert!( + old.is_none(), + "global maintenance id already owns relation storage" + ); + self.deps.add_participant(res, read_deps, write_deps); res } @@ -1129,7 +1316,7 @@ impl Database { /// Unlike calling [`Table::new_buffer`] on a table returned from a getter, this method also /// triggers change notification metadata that is read by [`Database::merge_all`]. pub fn new_buffer(&self, id: TableId) -> Box { - self.notification_list.notify(self.table_maintenance[id]); + self.notification_list.notify(id); self.get_table(id).new_buffer() } diff --git a/core-relations/src/lib.rs b/core-relations/src/lib.rs index 022b680bd..2b861d954 100644 --- a/core-relations/src/lib.rs +++ b/core-relations/src/lib.rs @@ -31,8 +31,8 @@ pub use containers::{ SequenceContainerValue, }; pub use free_join::{ - AtomId, CounterId, Database, ExternalFunction, ExternalFunctionId, MaintenanceReadDependency, - TableId, Variable, make_external_func, plan::PlanStrategy, + AtomId, CounterId, Database, ExternalFunction, ExternalFunctionId, TableId, Variable, + make_external_func, plan::PlanStrategy, }; pub use hash_index::TupleIndex; #[doc(hidden)] diff --git a/core-relations/src/table/sequence.rs b/core-relations/src/table/sequence.rs index 0c7c3b377..c6274f0a4 100644 --- a/core-relations/src/table/sequence.rs +++ b/core-relations/src/table/sequence.rs @@ -40,7 +40,9 @@ use crate::{ parallelize_index_construction, parallelize_rebuild, parallelize_table_op, }, pool::{Pooled, with_pool_set}, - table_spec::{Generation, MutationBuffer, Offset, Row, TableVersion, ValueRebuilder}, + table_spec::{ + Generation, MaintenanceTable, MutationBuffer, Offset, Row, TableVersion, ValueRebuilder, + }, }; use super::{ @@ -919,6 +921,20 @@ impl Default for SequenceTable { } } +impl MaintenanceTable for SequenceTable { + fn len(&self) -> usize { + SequenceTable::len(self) + } + + fn new_buffer(&self) -> Box { + SequenceTable::new_buffer(self) + } + + fn merge(&mut self, exec_state: &mut ExecutionState) -> TableChange { + self.merge_with_state(exec_state) + } +} + impl Clone for SequenceTable { fn clone(&self) -> Self { Self { diff --git a/core-relations/src/table_spec.rs b/core-relations/src/table_spec.rs index 6dfb1b3bf..22d11c05f 100644 --- a/core-relations/src/table_spec.rs +++ b/core-relations/src/table_spec.rs @@ -163,6 +163,17 @@ pub struct Row { pub vals: Pooled>, } +/// The arity-agnostic surface shared by every database storage participant. +/// +/// Fixed-schema relations expose the larger [`Table`] interface for querying, +/// while variable-arity sequence tables only need this protocol to share +/// mutation buffers, notifications, dependency ordering, and merge scheduling. +pub(crate) trait MaintenanceTable: Send + Sync { + fn len(&self) -> usize; + fn new_buffer(&self) -> Box; + fn merge(&mut self, exec_state: &mut ExecutionState) -> TableChange; +} + /// An interface for a table. pub trait Table: Any + Send + Sync { /// A variant of clone that returns a boxed trait object; this trait object @@ -201,7 +212,8 @@ pub trait Table: Any + Send + Sync { /// parent-row delta even though the row's key columns do not otherwise /// change. /// - /// One source of such ids is [`crate::ContainerRebuildSummary::dirty_ids`]. + /// Container maintenance uses this when a canonicalized container keeps + /// its identity but changes the values represented by that identity. /// /// Tables that do not maintain rebuildable id columns can use the default /// no-op implementation. @@ -666,6 +678,20 @@ impl WrappedTable { } } +impl MaintenanceTable for WrappedTable { + fn len(&self) -> usize { + WrappedTable::len(self) + } + + fn new_buffer(&self) -> Box { + self.inner.new_buffer() + } + + fn merge(&mut self, exec_state: &mut ExecutionState) -> TableChange { + self.inner.merge(exec_state) + } +} + impl Deref for WrappedTable { type Target = dyn Table; diff --git a/core-relations/src/uf/mod.rs b/core-relations/src/uf/mod.rs index 60ab8e00e..0214f88e7 100644 --- a/core-relations/src/uf/mod.rs +++ b/core-relations/src/uf/mod.rs @@ -250,7 +250,9 @@ impl Drop for UfBuffer { // This avoids creating a fresh row buffer via `mem::take` or `mem::swap` and // dropping it immediately. let to_insert = unsafe { ManuallyDrop::take(&mut self.to_insert) }; - buffered_writes.push(to_insert); + if to_insert.len() != 0 { + buffered_writes.push(to_insert); + } } } diff --git a/core-relations/src/uf/tests.rs b/core-relations/src/uf/tests.rs index 5d163ed71..227db077c 100644 --- a/core-relations/src/uf/tests.rs +++ b/core-relations/src/uf/tests.rs @@ -11,6 +11,13 @@ fn v(x: usize) -> Value { Value::from_usize(x) } +#[test] +fn empty_buffer_does_not_enqueue_work() { + let d = DisplacedTable::default(); + drop(d.new_buffer()); + assert!(d.buffered_writes.is_empty()); +} + #[test] fn displaced() { empty_execution_state!(e); diff --git a/egglog-bridge/src/lib.rs b/egglog-bridge/src/lib.rs index 7ac4cbe69..bfbea4b87 100644 --- a/egglog-bridge/src/lib.rs +++ b/egglog-bridge/src/lib.rs @@ -320,9 +320,8 @@ impl EGraph { pub fn register_container_ty(&mut self) { let uf_table = self.uf_table; let ts_counter = self.timestamp_counter; - self.db.container_values_mut().register_type::( - self.id_counter, - move |state, old, new| { + 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]); @@ -330,8 +329,7 @@ impl EGraph { } else { old } - }, - ); + }); } /// Register a container using the variable-arity sequence-table backend.