diff --git a/Cargo.toml b/Cargo.toml index 59b8c5f..b30b580 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,6 +18,10 @@ categories = ["database-implementations", "data-structures", "caching"] default = ["wti-predictable-search"] perf_measurements = ["dep:performance_measurement", "dep:performance_measurement_codegen"] s3-support = ["dep:rusty-s3", "dep:url", "dep:reqwest", "dep:walkdir", "worktable_codegen/s3-support"] +# Moves unique WorkTablesIndex structural CDC work out of the table mutation +# path and into the background persistence worker. The persisted page format +# is unchanged, so stores remain readable with or without this feature. +logical-index-persistence = ["worktable_codegen/logical-index-persistence"] wti-hybrid-search = ["indexset/wt-slice-binary-search"] wti-predictable-search = ["indexset/custom-binary-search"] wti-std-search = ["indexset/std-binary-search"] diff --git a/codegen/Cargo.toml b/codegen/Cargo.toml index c7a5562..3871b64 100644 --- a/codegen/Cargo.toml +++ b/codegen/Cargo.toml @@ -8,6 +8,7 @@ repository = "https://github.com/pathscale/WorkTable" [features] s3-support = [] +logical-index-persistence = [] # Compatibility no-op retained for downstream manifests. versioned-row-publication = [] diff --git a/codegen/src/generators/index_backend.rs b/codegen/src/generators/index_backend.rs index ca923da..defcaf6 100644 --- a/codegen/src/generators/index_backend.rs +++ b/codegen/src/generators/index_backend.rs @@ -39,7 +39,16 @@ pub(crate) fn persistent_unique_index_type( value: &TokenStream, worktables_node: Option, ) -> syn::Result { + // This intentionally evaluates the proc-macro crate's feature. WorkTable's + // public feature forwards to worktable_codegen in Cargo.toml, so the + // runtime types and emitted types are selected together. Emitting a cfg in + // the expansion would instead test the consuming package's unrelated + // feature namespace, which may rename or omit the dependency feature. match backend { + IndexBackend::WorktablesIndex if cfg!(feature = "logical-index-persistence") => Ok(match worktables_node { + Some(node) => quote! { PersistentWtiIndex<#key, #value, #node> }, + None => quote! { PersistentWtiIndex<#key, #value> }, + }), IndexBackend::Congee => Ok(quote! { PersistentCongeeIndex<#key, #value> }), IndexBackend::Arctic => Ok(quote! { PersistentArcticIndex<#key, #value> }), _ => unique_index_type(backend, key, value, worktables_node), diff --git a/codegen/src/generators/persist/index/mod.rs b/codegen/src/generators/persist/index/mod.rs index b2068c6..476fe72 100644 --- a/codegen/src/generators/persist/index/mod.rs +++ b/codegen/src/generators/persist/index/mod.rs @@ -109,11 +109,16 @@ impl PersistGenerator { let res = if idx.is_unique { match idx.backend { crate::common::model::IndexBackend::WorktablesIndex => { + let map = if cfg!(feature = "logical-index-persistence") { + quote! { PersistentWtiIndex } + } else { + quote! { IndexMap } + }; if is_unsized(&t.to_string()) { - quote! { #i: IndexMap::with_maximum_node_size(#const_name), } + quote! { #i: #map::with_maximum_node_size(#const_name), } } else { quote! { - #i: IndexMap::with_maximum_node_size( + #i: #map::with_maximum_node_size( get_index_page_size_from_data_length::<#t>(#const_name) ), } diff --git a/codegen/src/generators/persist/table/impls.rs b/codegen/src/generators/persist/table/impls.rs index 12edae5..02f8f82 100644 --- a/codegen/src/generators/persist/table/impls.rs +++ b/codegen/src/generators/persist/table/impls.rs @@ -237,10 +237,15 @@ impl PersistGenerator { }) .collect::>(); let pk_types_unsized = is_unsized_vec(pk_types); + let wti_map = if cfg!(feature = "logical-index-persistence") { + quote! { PersistentWtiIndex } + } else { + quote! { IndexMap } + }; let index_setup = if pk_types_unsized { quote! { inner.primary_index = std::sync::Arc::new(PrimaryIndex { - pk_map: IndexMap::<#pk_type, OffsetEqLink<#const_name>, UnsizedNode<_>>::with_maximum_node_size(#const_name), + pk_map: #wti_map::<#pk_type, OffsetEqLink<#const_name>, UnsizedNode<_>>::with_maximum_node_size(#const_name), reverse_pk_map: IndexMap::new(), }); } @@ -249,7 +254,7 @@ impl PersistGenerator { crate::common::model::IndexBackend::WorktablesIndex => quote! { let size = get_index_page_size_from_data_length::<#pk_type>(#const_name); inner.primary_index = std::sync::Arc::new(PrimaryIndex { - pk_map: IndexMap::<_, OffsetEqLink<#const_name>>::with_maximum_node_size(size), + pk_map: #wti_map::<_, OffsetEqLink<#const_name>>::with_maximum_node_size(size), reverse_pk_map: IndexMap::new(), }); }, diff --git a/codegen/src/generators/persist/table/mod.rs b/codegen/src/generators/persist/table/mod.rs index 272fcb1..1d339ed 100644 --- a/codegen/src/generators/persist/table/mod.rs +++ b/codegen/src/generators/persist/table/mod.rs @@ -89,6 +89,14 @@ impl PersistGenerator { #[derive(Debug, PersistTable)] #[table(pk_unsized, pk_upstream)] }, + (true, crate::common::model::IndexBackend::WorktablesIndex) + if cfg!(feature = "logical-index-persistence") => + { + quote! { + #[derive(Debug, PersistTable)] + #[table(pk_unsized, pk_wti_logical)] + } + } (true, _) => quote! { #[derive(Debug, PersistTable)] #[table(pk_unsized)] @@ -105,6 +113,14 @@ impl PersistGenerator { #[derive(Debug, PersistTable)] #[table(pk_congee)] }, + (false, crate::common::model::IndexBackend::WorktablesIndex) + if cfg!(feature = "logical-index-persistence") => + { + quote! { + #[derive(Debug, PersistTable)] + #[table(pk_wti_logical)] + } + } (false, crate::common::model::IndexBackend::WorktablesIndex) => quote! { #[derive(Debug, PersistTable)] }, diff --git a/codegen/src/persist_index/generator.rs b/codegen/src/persist_index/generator.rs index 9af1858..9c53628 100644 --- a/codegen/src/persist_index/generator.rs +++ b/codegen/src/persist_index/generator.rs @@ -24,6 +24,7 @@ pub(super) struct IndexLayout { is_unique: bool, uses_upstream: bool, pub(super) art_backend: Option, + pub(super) logical_wti: bool, } #[derive(Copy, Clone, Debug, Eq, PartialEq)] @@ -46,12 +47,13 @@ pub(super) fn index_layout(field: &Field) -> syn::Result { .ok_or_else(|| syn::Error::new_spanned(&field.ty, "index type path cannot be empty"))? .ident .clone(); - let (is_unique, uses_upstream, art_backend) = match type_ident.to_string().as_str() { - "IndexMap" | "TreeIndex" => (true, false, None), - "UpstreamIndexMap" => (true, true, None), - "IndexMultiMap" | "TreeMultiIndex" => (false, false, None), - "PersistentArcticIndex" => (true, false, Some(ArtBackend::Arctic)), - "PersistentCongeeIndex" => (true, false, Some(ArtBackend::Congee)), + let (is_unique, uses_upstream, art_backend, logical_wti) = match type_ident.to_string().as_str() { + "IndexMap" | "TreeIndex" => (true, false, None, false), + "PersistentWtiIndex" => (true, false, None, true), + "UpstreamIndexMap" => (true, true, None, false), + "IndexMultiMap" | "TreeMultiIndex" => (false, false, None, false), + "PersistentArcticIndex" => (true, false, Some(ArtBackend::Arctic), false), + "PersistentCongeeIndex" => (true, false, Some(ArtBackend::Congee), false), _ => { return Err(syn::Error::new_spanned( &field.ty, @@ -64,6 +66,7 @@ pub(super) fn index_layout(field: &Field) -> syn::Result { is_unique, uses_upstream, art_backend, + logical_wti, }) } diff --git a/codegen/src/persist_index/space/index.rs b/codegen/src/persist_index/space/index.rs index d891d50..eb308ee 100644 --- a/codegen/src/persist_index/space/index.rs +++ b/codegen/src/persist_index/space/index.rs @@ -25,6 +25,12 @@ impl Generator { Some(ArtBackend::Congee) => quote! { #i: SpaceCongeeIndex<#t, { #inner_const_name as u32}>, }, + None if layout.logical_wti && is_unsized(&t.to_string()) => quote! { + #i: SpaceLogicalIndexUnsized<#t, { #inner_const_name as u32}>, + }, + None if layout.logical_wti => quote! { + #i: SpaceLogicalIndex<#t, { #inner_const_name as u32}>, + }, None if is_unsized(&t.to_string()) => quote! { #i: SpaceIndexUnsized<#t, { #inner_const_name as u32}>, }, @@ -79,6 +85,12 @@ impl Generator { Some(ArtBackend::Congee) => quote! { #i: SpaceCongeeIndex::secondary_from_table_files_path(path, #literal_name, version).await?, }, + None if layout.logical_wti && is_unsized(&t.to_string()) => quote! { + #i: SpaceLogicalIndexUnsized::secondary_from_table_files_path(path, #literal_name, version).await?, + }, + None if layout.logical_wti => quote! { + #i: SpaceLogicalIndex::secondary_from_table_files_path(path, #literal_name, version).await?, + }, None if is_unsized(&t.to_string()) => quote! { #i: SpaceIndexUnsized::secondary_from_table_files_path(path, #literal_name, version).await?, }, diff --git a/codegen/src/persist_table/generator/mod.rs b/codegen/src/persist_table/generator/mod.rs index 7baf87b..c93e091 100644 --- a/codegen/src/persist_table/generator/mod.rs +++ b/codegen/src/persist_table/generator/mod.rs @@ -14,6 +14,7 @@ pub struct PersistTableAttributes { pub pk_upstream: bool, pub pk_arctic: bool, pub pk_congee: bool, + pub pk_wti_logical: bool, pub row_schema: Vec<(String, String)>, pub primary_key_fields: Vec, pub secondary_index_types: Vec<(String, String)>, diff --git a/codegen/src/persist_table/generator/space.rs b/codegen/src/persist_table/generator/space.rs index 1f81f39..dc2d279 100644 --- a/codegen/src/persist_table/generator/space.rs +++ b/codegen/src/persist_table/generator/space.rs @@ -36,10 +36,18 @@ impl Generator { let space_secondary_indexes = name_generator.get_space_secondary_index_ident(); let space_secondary_indexes_events = name_generator.get_space_secondary_index_events_ident(); let avt_index_ident = name_generator.get_available_indexes_ident(); - let space_index_type = if self.attributes.pk_unsized { + let space_index_type = if self.attributes.pk_unsized && self.attributes.pk_wti_logical { + quote! { + SpaceLogicalIndexUnsized<#primary_key_type, { #inner_const_name as u32 }>, + } + } else if self.attributes.pk_unsized { quote! { SpaceIndexUnsized<#primary_key_type, { #inner_const_name as u32 }>, } + } else if self.attributes.pk_wti_logical { + quote! { + SpaceLogicalIndex<#primary_key_type, { #inner_const_name as u32 }>, + } } else if self.attributes.pk_arctic { quote! { SpaceArcticIndex<#primary_key_type, { #inner_const_name as u32 }>, diff --git a/codegen/src/persist_table/generator/space_file/mod.rs b/codegen/src/persist_table/generator/space_file/mod.rs index af79d7a..cef7ba3 100644 --- a/codegen/src/persist_table/generator/space_file/mod.rs +++ b/codegen/src/persist_table/generator/space_file/mod.rs @@ -143,8 +143,13 @@ impl Generator { let primary_index_init = if self.attributes.pk_unsized { let pk_ident = &self.pk_ident; + let map_type = if self.attributes.pk_wti_logical { + quote! { PersistentWtiIndex } + } else { + quote! { IndexMap } + }; quote! { - let pk_map = IndexMap::<#pk_ident, OffsetEqLink<#const_name>, UnsizedNode<_>>::with_maximum_node_size(#const_name); + let pk_map = #map_type::<#pk_ident, OffsetEqLink<#const_name>, UnsizedNode<_>>::with_maximum_node_size(#const_name); for page in self.primary_index.1 { let node = page .inner @@ -175,7 +180,9 @@ impl Generator { let primary_index = PrimaryIndex { pk_map, reverse_pk_map }; } } else { - let map_type = if self.pk_upstream { + let map_type = if self.attributes.pk_wti_logical { + quote! { PersistentWtiIndex } + } else if self.pk_upstream { quote! { UpstreamIndexMap } } else { quote! { IndexMap } diff --git a/codegen/src/persist_table/parser.rs b/codegen/src/persist_table/parser.rs index 26773d9..5803beb 100644 --- a/codegen/src/persist_table/parser.rs +++ b/codegen/src/persist_table/parser.rs @@ -32,6 +32,7 @@ impl Parser { pk_upstream: false, pk_arctic: false, pk_congee: false, + pk_wti_logical: false, row_schema: vec![], primary_key_fields: vec![], secondary_index_types: vec![], @@ -60,6 +61,10 @@ impl Parser { res.pk_congee = true; return Ok(()); } + if meta.path.is_ident("pk_wti_logical") { + res.pk_wti_logical = true; + return Ok(()); + } if meta.path.is_ident("row_schema") { meta.parse_nested_meta(|field| { let name = field diff --git a/codegen/src/worktable/mod.rs b/codegen/src/worktable/mod.rs index fe99341..e9334af 100644 --- a/codegen/src/worktable/mod.rs +++ b/codegen/src/worktable/mod.rs @@ -206,9 +206,42 @@ mod tests { indexes: { value_idx: value unique, }, - }); + }) + .unwrap() + .to_string(); + + if cfg!(feature = "logical-index-persistence") { + assert!(output.contains("PersistentWtiIndex")); + } else { + assert!(output.contains("IndexMap")); + assert!(!output.contains("PersistentWtiIndex")); + } + } - assert!(output.is_ok()); + #[cfg(feature = "logical-index-persistence")] + #[test] + fn logical_persistence_wraps_only_default_wti_backends() { + let output = expand(quote! { + name: LogicalDefaultBackend, + persist: true, + columns: { + id: u64 primary_key autoincrement, + wti_value: u64, + congee_value: u64, + arctic_value: u64, + }, + indexes: { + wti_idx: wti_value unique, + congee_idx: congee_value unique using congee, + arctic_idx: arctic_value unique using arctic, + }, + }) + .unwrap() + .to_string(); + + assert!(output.contains("PersistentWtiIndex")); + assert!(output.contains("PersistentCongeeIndex")); + assert!(output.contains("PersistentArcticIndex")); } #[test] diff --git a/docs/wti-dirty-generation-persistence-plan.md b/docs/wti-dirty-generation-persistence-plan.md index c881cac..dd81f51 100644 --- a/docs/wti-dirty-generation-persistence-plan.md +++ b/docs/wti-dirty-generation-persistence-plan.md @@ -1,6 +1,6 @@ # WTI dirty-generation persistence -**Status:** immediate follow-up design; not implemented. +**Status:** first stage implemented behind `logical-index-persistence`; dirty-generation checkpoints remain follow-up work. **Scope:** reduce caller-thread persistence overhead for WorkTablesIndex (WTI) without changing point-read behavior or weakening recovery. @@ -18,6 +18,22 @@ A focused local ARM probe measured: These are two interleaved ten-trial microprobe runs, not publication results. They exclude row work and disk I/O. They establish that caller-side CDC is large enough to optimize. +The first implementation was then measured with another ten-trial interleaved +ARM probe over 2,000,000 existing-key mutations and 8,000,000 point reads per +trial: + +| Path | Median | Delta | +|---|---:|---:| +| Structural CDC update | 80.916 ns/op | baseline | +| Logical foreground update | 71.440 ns/op | 11.71% faster | +| Raw WTI point read | 105.096 ns/op | baseline | +| Wrapped WTI point read | 103.068 ns/op | 1.93% faster (noise; no observed regression) | + +These repeat-run results were tight across the ten trials, but they remain local +engineering probes rather than paper numbers. Generated-table select +measurements were noisy enough to require a dedicated quiet-window run before +drawing a sub-percent conclusion. The feature therefore remains opt-in. + WorkTable already uses lifecycle flags (`GHOSTED`, `DELETED`, and `VACUUMED`) and optional immutable row versions to prevent readers from observing partially published rows. Those are visibility states, not persistence dirty states, but the staged-publication pattern is relevant. ## Why one dirty bit is insufficient @@ -32,7 +48,69 @@ A Boolean has a lost-update race: The persistence marker must carry a generation or an explicit redirty state. Clearing is conditional: the flusher may mark a node clean only if no writer advanced its generation after the snapshot began. -## Proposed architecture +## Implemented first stage: background structural translation + +The first implementation deliberately avoids a new WTI disk format. With the +`logical-index-persistence` Cargo feature enabled: + +1. each persisted primary or unique-secondary WTI is wrapped by `PersistentWtiIndex`; +2. point reads delegate directly to WTI, with no runtime feature branch or new + read-side lock; +3. each successful foreground mutation emits one logical Set/Remove event; +4. the existing persistence worker owns a private shadow WTI reconstructed from + the current `.wt.idx` pages; +5. that shadow translates logical events into native structural CDC; and +6. the existing `SpaceIndex`/`SpaceIndexUnsized` writer applies those events to + the unchanged DataBucket page format. + +Foreground stripes establish same-key mutation order only. They use +`DefaultHasher`, so stripe identity has no relationship to key or range order. +The queue analyzer sorts each per-index stream by its global event ID before +dispatch, and the logical shadow sorts again at its boundary as defense in +depth. Reversed-delivery regression coverage proves that a Set/Remove pair is +applied in event-ID order. + +Each logical WTI contains 64 inline `parking_lot::Mutex<()>` stripes. Their +target-dependent fixed footprint is paid once per persisted primary or unique +WTI; point reads never access them. This is a deliberate write-concurrency +tradeoff and should be included in schema-level memory measurements. + +This is an intentionally smaller step than the checkpoint/WAL design below. It +moves the measured structural-CDC work off the caller thread while retaining +format compatibility in both directions: a store written with the feature can +be reopened without it, and a pre-feature store can be opened with it. The +existing WorkTable persistence queue remains the recovery authority, including +its documented best-effort crash-durability boundary; this stage does not add a +new durable logical WAL. + +Shadow divergence is reported as `PersistenceIndexCorruption`, not a generic +worker error. It quarantines the table's whole persistence engine, rejects later +persistence submissions, and is surfaced by `wait_for_ops` and `close`. +Continuing row or sibling-index writes after one index diverges would knowingly +create an inconsistent store, so the current architecture cannot safely +quarantine only that index. + +The code generator selects this path using the forwarded +`worktable_codegen/logical-index-persistence` feature. That check intentionally +runs in the proc-macro crate: emitting a downstream `#[cfg]` would inspect the +consumer package's feature namespace rather than WorkTable's dependency +feature. Feature-off and feature-on expansion tests cover both selections. + +The DSL contract is unchanged: + +- omitting `using` selects WorkTablesIndex; +- `using congee` selects `congee-wt`; +- `using arctic` selects `arctic-wt`. + +Congee and Arctic already use their native topology checkpoint plus logical WAL +implementations. Their `export_topology`/`from_topology` APIs are sufficient, so +this first stage requires no source change in either fork. + +Non-unique WTI secondary indexes continue to emit structural multimap CDC in +this stage. Their logical record must identify both key and row link, and should +be added only with the same rollback, duplicate-ordering, and reload coverage. + +## Longer-term architecture Separate crash authority from physical checkpoint maintenance: @@ -155,11 +233,11 @@ It should not: ## Implementation stages -1. **Measure and instrument.** Keep the direct-versus-structural-CDC probe, add allocations/op and p50/p99, and measure full generated WTI table operations. -2. **Define recovery authority.** Make index rebuild from authoritative data pages an explicit, tested fallback and version the new WTI format. -3. **Add feature-gated logical WTI redo.** Preserve the existing structural-CDC path as default until validation is complete. -4. **Add whole-index background checkpointing.** Use atomic replacement and a checkpoint generation; prove redo truncation ordering with crash injection. -5. **Evaluate the result.** Continue only if the caller-thread savings survive end-to-end WorkTable benchmarks. +1. **Background structural translation (implemented, feature-gated).** Preserve the existing disk format and move structural CDC onto a worker-owned shadow WTI. +2. **Measure and instrument.** Keep the direct-versus-structural-CDC probe, add allocations/op and p50/p99, and measure full generated WTI table operations. +3. **Evaluate the result.** Continue only if the caller-thread savings survive end-to-end WorkTable benchmarks. +4. **Define stronger recovery authority.** Make index rebuild from authoritative data pages an explicit, tested fallback and version any new WTI format. +5. **Add durable logical WTI redo and whole-index checkpoints.** Use atomic replacement and a checkpoint generation; prove redo truncation ordering with crash injection. 6. **Optionally add incremental dirty nodes.** Introduce the generation state machine, pinning, structural groups, and generation manifest. 7. **Validate local disk and S3.** Test interrupted append, checkpoint, rename, upload, download, and writes after recovery. diff --git a/src/index/mod.rs b/src/index/mod.rs index b6c4351..8ed7f27 100644 --- a/src/index/mod.rs +++ b/src/index/mod.rs @@ -3,6 +3,7 @@ mod available_index; mod congee; mod multipair; mod persistent_art; +mod persistent_wti; mod primary_index; mod table_index; mod table_secondary_index; @@ -16,6 +17,7 @@ pub use indexset::concurrent::map::BTreeMap as IndexMap; pub use indexset::concurrent::multimap::BTreeMultiMap as IndexMultiMap; pub use multipair::MultiPairRecreate; pub use persistent_art::{PersistentArcticIndex, PersistentArtIndex, PersistentCongeeIndex}; +pub use persistent_wti::PersistentWtiIndex; pub use primary_index::PrimaryIndex; pub use table_index::{TableIndex, TableIndexCdc, convert_change_events, convert_upstream_change_events}; pub use table_secondary_index::{ diff --git a/src/index/persistent_wti.rs b/src/index/persistent_wti.rs new file mode 100644 index 0000000..0853ffd --- /dev/null +++ b/src/index/persistent_wti.rs @@ -0,0 +1,308 @@ +//! Logical change capture for persisted unique WorkTablesIndex instances. +//! +//! The normal persisted path asks the live index to produce structural CDC +//! events. With `logical-index-persistence`, this wrapper instead emits one +//! logical Set/Remove event. The background disk-side shadow index translates +//! it back into structural events, preserving the existing on-disk format +//! while removing structural CDC bookkeeping from foreground mutations. +//! `index = 0` and `max_value == value` are an intentionally synthetic marker, +//! not claims about the live WTI node position or maximum; the shadow validates +//! that marker and derives the real structural metadata itself. + +use std::array; +use std::collections::hash_map::DefaultHasher; +use std::fmt::{self, Debug}; +use std::hash::{Hash, Hasher}; +use std::ops::RangeBounds; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; + +use data_bucket::Link; +use indexset::cdc::change::{ChangeEvent, Id}; +use indexset::core::node::NodeLike; +use indexset::core::pair::Pair; +use parking_lot::{Mutex, MutexGuard}; + +use crate::index::UniqueIndex; +use crate::util::OffsetEqLink; +use crate::{IndexMap, TableIndexCdc}; + +// A stripe provides per-key exclusion only: DefaultHasher has no relationship +// to key order, so callers must not infer range or cross-key ordering from the +// selected mutex. Reads never touch this fixed inline table. Logical batches +// are ordered independently by event id in the persistence worker. +const MUTATION_STRIPES: usize = 64; + +/// A persisted WorkTablesIndex whose foreground mutations emit logical CDC. +/// +/// Point reads delegate directly to the native index. There is no runtime +/// feature check, read lock, or consistency branch on the select path. +pub struct PersistentWtiIndex>> +where + K: Send + Ord + Clone + 'static, + V: Send + Clone + 'static, + Node: NodeLike>, +{ + inner: IndexMap, + next_event_id: AtomicU64, + mutation_stripes: [Mutex<()>; MUTATION_STRIPES], +} + +impl Debug for PersistentWtiIndex +where + K: Send + Ord + Clone + 'static, + V: Send + Clone + 'static, + Node: NodeLike>, +{ + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("PersistentWtiIndex") + .field("next_event_id", &self.next_event_id.load(Ordering::Relaxed)) + .finish_non_exhaustive() + } +} + +impl Default for PersistentWtiIndex +where + K: Send + Ord + Clone + 'static, + V: Send + Clone + 'static, + Node: NodeLike>, + IndexMap: Default, +{ + fn default() -> Self { + Self::from_inner(IndexMap::default()) + } +} + +impl PersistentWtiIndex +where + K: Send + Ord + Clone + 'static, + V: Send + Clone + 'static, + Node: NodeLike>, +{ + pub fn from_inner(inner: IndexMap) -> Self { + Self { + inner, + next_event_id: AtomicU64::new(0), + mutation_stripes: array::from_fn(|_| Mutex::new(())), + } + } + + pub fn into_inner(self) -> IndexMap { + self.inner + } + + pub fn inner(&self) -> &IndexMap { + &self.inner + } + + fn mutation_stripe(&self, key: &Q) -> MutexGuard<'_, ()> { + let mut hasher = DefaultHasher::new(); + key.hash(&mut hasher); + self.mutation_stripes[hasher.finish() as usize % MUTATION_STRIPES].lock() + } + + fn next_event_id(&self) -> Id { + self.next_event_id.fetch_add(1, Ordering::AcqRel).into() + } +} + +impl PersistentWtiIndex +where + K: Debug + Send + Ord + Clone + 'static, + V: Debug + Send + Clone + 'static, + Node: NodeLike> + Send + 'static, +{ + pub fn with_maximum_node_size(node_capacity: usize) -> Self { + Self::from_inner(IndexMap::with_maximum_node_size(node_capacity)) + } + + pub fn attach_node(&self, node: Node) { + self.inner.attach_node(node); + } + + pub fn iter_nodes(&self) -> impl Iterator>> + '_ { + self.inner.iter_nodes() + } + + pub fn iter(&self) -> impl DoubleEndedIterator + '_ { + self.inner.iter() + } + + pub fn capacity(&self) -> usize { + self.inner.capacity() + } + + pub fn node_count(&self) -> usize { + self.inner.node_count() + } +} + +impl UniqueIndex for PersistentWtiIndex +where + K: Debug + Eq + Hash + Clone + Send + Ord + 'static, + V: Debug + Clone + Send + Ord + 'static, + Node: NodeLike> + Send + 'static, +{ + #[inline(always)] + fn get_value(&self, key: &K) -> Option { + self.inner.lookup_for_select(key) + } + + #[inline(always)] + fn lookup_for_select(&self, key: &K) -> Option { + self.inner.lookup_for_select(key) + } + + #[inline(always)] + fn with_value(&self, key: &K, read: impl FnOnce(&V) -> R) -> Option { + self.inner.lookup_for_select(key).as_ref().map(read) + } + + #[inline(always)] + fn contains_key(&self, key: &K) -> bool { + self.inner.contains_key(key) + } + + #[inline] + fn insert_value(&self, key: K, value: V) -> Option { + self.inner.insert(key, value) + } + + #[inline] + fn insert_value_checked(&self, key: K, value: V) -> Option<()> { + self.inner.checked_insert(key, value) + } + + #[inline] + fn remove_value(&self, key: &K) -> Option<(K, V)> { + self.inner.remove(key) + } + + #[inline] + fn len(&self) -> usize { + self.inner.len() + } + + fn iter_values(&self) -> impl DoubleEndedIterator + '_ { + self.inner.iter().map(|(key, value)| (key.clone(), value.clone())) + } + + fn iter_links(&self) -> impl DoubleEndedIterator + '_ { + self.inner.iter().map(|(_, value)| value.clone()) + } + + fn range_values<'a, R>(&'a self, range: R) -> impl DoubleEndedIterator + 'a + where + R: RangeBounds + 'a, + { + self.inner.range(range).map(|(key, value)| (key.clone(), value.clone())) + } + + fn range_links<'a, R>(&'a self, range: R) -> impl DoubleEndedIterator + 'a + where + R: RangeBounds + 'a, + { + self.inner.range(range).map(|(_, value)| value.clone()) + } +} + +impl TableIndexCdc for PersistentWtiIndex, Node> +where + T: Debug + Eq + Hash + Clone + Send + Ord + 'static, + Node: NodeLike>> + Send + 'static, +{ + fn insert_cdc(&self, value: T, link: Link) -> (Option, Vec>>) { + let _sequence_guard = self.mutation_stripe(&value); + let old = self + .inner + .insert(value.clone(), OffsetEqLink(link)) + .map(|value| value.0); + let pair = Pair { + key: value, + value: link, + }; + let event = ChangeEvent::InsertAt { + event_id: self.next_event_id(), + max_value: pair.clone(), + value: pair, + index: 0, + }; + (old, vec![event]) + } + + fn insert_checked_cdc(&self, value: T, link: Link) -> Option>>> { + let _sequence_guard = self.mutation_stripe(&value); + self.inner.checked_insert(value.clone(), OffsetEqLink(link))?; + let pair = Pair { + key: value, + value: link, + }; + Some(vec![ChangeEvent::InsertAt { + event_id: self.next_event_id(), + max_value: pair.clone(), + value: pair, + index: 0, + }]) + } + + fn remove_cdc(&self, value: T, _: Link) -> (Option<(T, Link)>, Vec>>) { + let _sequence_guard = self.mutation_stripe(&value); + let Some((key, old)) = self.inner.remove(&value) else { + return (None, Vec::new()); + }; + let pair = Pair { + key: key.clone(), + value: old.0, + }; + let event = ChangeEvent::RemoveAt { + event_id: self.next_event_id(), + max_value: pair.clone(), + value: pair, + index: 0, + }; + (Some((key, old.0)), vec![event]) + } +} + +#[cfg(test)] +mod tests { + use data_bucket::page::PageId; + + use super::*; + + fn link(offset: u32) -> Link { + Link { + page_id: PageId::from(1), + offset, + length: 8, + } + } + + #[test] + fn reads_delegate_and_mutations_emit_one_logical_event() { + let index = PersistentWtiIndex::>::default(); + let (old, events) = index.insert_cdc(7, link(7)); + assert_eq!(old, None); + assert_eq!(events.len(), 1); + assert_eq!(index.get_value(&7), Some(OffsetEqLink(link(7)))); + + let (removed, events) = index.remove_cdc(7, link(7)); + assert_eq!(removed, Some((7, link(7)))); + assert_eq!(events.len(), 1); + assert_eq!(index.get_value(&7), None); + } + + #[test] + fn rejected_checked_insert_does_not_consume_an_event_id() { + let index = PersistentWtiIndex::>::default(); + assert!(index.insert_checked_cdc(7, link(7)).is_some()); + assert_eq!(index.next_event_id.load(Ordering::Relaxed), 1); + + assert!(index.insert_checked_cdc(7, link(8)).is_none()); + assert_eq!(index.next_event_id.load(Ordering::Relaxed), 1); + + assert!(index.insert_checked_cdc(8, link(8)).is_some()); + assert_eq!(index.next_event_id.load(Ordering::Relaxed), 2); + } +} diff --git a/src/index/table_index/mod.rs b/src/index/table_index/mod.rs index 75c8b71..4affdae 100644 --- a/src/index/table_index/mod.rs +++ b/src/index/table_index/mod.rs @@ -11,7 +11,7 @@ use vanilla_indexset::core::pair::Pair as VanillaPair; use crate::util::OffsetEqLink; use crate::{ ArcticIndex, ArcticKey, CongeeIndex, CongeeKey, IndexMap, IndexMultiMap, PersistentArcticIndex, - PersistentCongeeIndex, UniqueIndex, UpstreamIndexMap, + PersistentCongeeIndex, PersistentWtiIndex, UniqueIndex, UpstreamIndexMap, }; mod cdc; @@ -132,6 +132,24 @@ where } } +impl TableIndex for PersistentWtiIndex +where + T: Debug + Eq + Hash + Clone + Send + Ord + 'static, + Node: NodeLike> + Send + 'static, +{ + fn insert(&self, value: T, link: Link) -> Option { + unique_insert(self, value, link) + } + + fn insert_checked(&self, value: T, link: Link) -> Option<()> { + unique_insert_checked(self, value, link) + } + + fn remove(&self, value: &T, _: Link) -> Option<(T, Link)> { + unique_remove(self, value) + } +} + impl_unique_table_index!(CongeeIndex, [CongeeKey + Eq + Hash]); impl_unique_table_index!(ArcticIndex, [ ArcticKey + Eq + Hash diff --git a/src/lib.rs b/src/lib.rs index 96624a2..9e28c54 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -36,11 +36,11 @@ pub mod prelude { pub use crate::persistence::{ AcknowledgeOperation, ArtPersistenceKey, DeleteOperation, DiskConfig, DiskPersistenceEngine, IndexTableOfContents, InsertOperation, LoadMode, Operation, OperationId, PersistedWorkTable, PersistenceConfig, - PersistenceEngine, PersistenceError, PersistenceLoadError, PersistenceResult, PersistenceState, - PersistenceTask, ReadOnlyPersistenceEngine, SpaceArcticIndex, SpaceCongeeIndex, SpaceData, SpaceDataOps, - SpaceIndex, SpaceIndexOps, SpaceIndexUnsized, SpaceSecondaryIndexOps, UpdateOperation, load_persisted_state, - map_index_pages_to_toc_and_general, map_unsized_index_pages_to_toc_and_general, reconstruct_multi_index_nodes, - validate_events, + PersistenceEngine, PersistenceError, PersistenceIndexCorruption, PersistenceLoadError, PersistenceResult, + PersistenceState, PersistenceTask, ReadOnlyPersistenceEngine, SpaceArcticIndex, SpaceCongeeIndex, SpaceData, + SpaceDataOps, SpaceIndex, SpaceIndexOps, SpaceIndexUnsized, SpaceLogicalIndex, SpaceLogicalIndexUnsized, + SpaceSecondaryIndexOps, UpdateOperation, load_persisted_state, map_index_pages_to_toc_and_general, + map_unsized_index_pages_to_toc_and_general, reconstruct_multi_index_nodes, validate_events, }; pub use crate::primary_key::{PrimaryKeyGenerator, PrimaryKeyGeneratorState, TablePrimaryKey}; pub use crate::table::select::{Order, QueryParams, SelectQueryBuilder, SelectQueryExecutor}; @@ -49,10 +49,10 @@ pub mod prelude { pub use crate::{ ArcticIndex, ArcticKey, AvailableIndex, CongeeIndex, CongeeKey, Difference, IndexError, IndexMap, IndexMultiMap, MultiPairRecreate, PersistentArcticIndex, PersistentArtIndex, PersistentCongeeIndex, - PrimaryIndex, TableIndex, TableIndexCdc, TableRow, TableSecondaryIndex, TableSecondaryIndexCdc, - TableSecondaryIndexEventsOps, TableSecondaryIndexInfo, UniqueIndex, UnsizedNode, UpstreamIndexMap, - UpstreamIndexPair, WorkTable, WorkTableError, vacuum::EmptyDataVacuum, vacuum::VacuumPersistence, - vacuum::WorkTableVacuum, + PersistentWtiIndex, PrimaryIndex, TableIndex, TableIndexCdc, TableRow, TableSecondaryIndex, + TableSecondaryIndexCdc, TableSecondaryIndexEventsOps, TableSecondaryIndexInfo, UniqueIndex, UnsizedNode, + UpstreamIndexMap, UpstreamIndexPair, WorkTable, WorkTableError, vacuum::EmptyDataVacuum, + vacuum::VacuumPersistence, vacuum::WorkTableVacuum, }; pub use data_bucket::{ DATA_VERSION, DataPage, GENERAL_HEADER_SIZE, GeneralHeader, GeneralPage, INNER_PAGE_SIZE, IndexPage, Interval, diff --git a/src/mem_stat/mod.rs b/src/mem_stat/mod.rs index a933b4d..666d61d 100644 --- a/src/mem_stat/mod.rs +++ b/src/mem_stat/mod.rs @@ -21,7 +21,8 @@ use crate::persistence::OperationType; use crate::prelude::OperationId; use crate::util::OffsetEqLink; use crate::{ - ArcticIndex, ArcticKey, CongeeIndex, CongeeKey, IndexMultiMap, PersistentArtIndex, UniqueIndex, UpstreamIndexMap, + ArcticIndex, ArcticKey, CongeeIndex, CongeeKey, IndexMultiMap, PersistentArtIndex, PersistentWtiIndex, UniqueIndex, + UpstreamIndexMap, }; use crate::{IndexMap, impl_memstat_zero}; @@ -141,6 +142,21 @@ impl MemStat for PersistentArtIndex { } } +impl MemStat for PersistentWtiIndex +where + K: Debug + Ord + Clone + 'static + MemStat + Send, + V: Debug + Clone + 'static + MemStat + Send, + Node: NodeLike> + Send + 'static, +{ + fn heap_size(&self) -> usize { + self.inner().heap_size() + } + + fn used_size(&self) -> usize { + self.inner().used_size() + } +} + impl MemStat for IndexMultiMap where K: Debug + Ord + Clone + 'static + MemStat + Send, diff --git a/src/persistence/error.rs b/src/persistence/error.rs index ed077fc..fbfac49 100644 --- a/src/persistence/error.rs +++ b/src/persistence/error.rs @@ -79,6 +79,50 @@ where } } +/// A persisted index no longer agrees with the logical mutation stream. +/// +/// The safe quarantine boundary is the table's entire persistence engine: a +/// worker must not continue writing row data or sibling indexes after one +/// index diverges, because that would knowingly create a store that cannot be +/// recovered consistently. The in-memory table remains inspectable, while +/// `wait_for_ops`, `close`, and all later persistence submissions return this +/// terminal error through [`PersistenceError::IndexCorruption`]. +#[derive(Debug)] +pub struct PersistenceIndexCorruption { + path: PathBuf, + reason: String, +} + +impl PersistenceIndexCorruption { + pub fn new(path: impl AsRef, reason: impl Display) -> Self { + Self { + path: path.as_ref().to_path_buf(), + reason: reason.to_string(), + } + } + + pub fn path(&self) -> &Path { + &self.path + } + + pub fn reason(&self) -> &str { + &self.reason + } +} + +impl Display for PersistenceIndexCorruption { + fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result { + write!( + formatter, + "persisted index at {} was quarantined: {}", + self.path.display(), + self.reason + ) + } +} + +impl Error for PersistenceIndexCorruption {} + /// Terminal and lifecycle errors reported by a persistence task. #[derive(Debug)] pub enum PersistenceError { @@ -86,6 +130,8 @@ pub enum PersistenceError { Closing, /// New work was submitted after graceful shutdown completed. Closed, + /// An index diverged from its validated logical mutation stream. + IndexCorruption(PersistenceIndexCorruption), /// The persistence engine or its queue analyzer failed permanently. Engine(eyre::Report), } @@ -95,6 +141,7 @@ impl Display for PersistenceError { match self { Self::Closing => formatter.write_str("persistence task is closing"), Self::Closed => formatter.write_str("persistence task is closed"), + Self::IndexCorruption(error) => Display::fmt(error, formatter), Self::Engine(error) => write!(formatter, "persistence engine failed: {error:#}"), } } diff --git a/src/persistence/mod.rs b/src/persistence/mod.rs index 9aa2376..121e27a 100644 --- a/src/persistence/mod.rs +++ b/src/persistence/mod.rs @@ -6,7 +6,10 @@ use crate::persistence::operation::BatchOperation; pub use engine::DiskConfig; pub use engine::DiskPersistenceEngine; -pub use error::{PersistenceError, PersistenceLoadError, PersistenceResult, PersistenceState, load_persisted_state}; +pub use error::{ + PersistenceError, PersistenceIndexCorruption, PersistenceLoadError, PersistenceResult, PersistenceState, + load_persisted_state, +}; pub use operation::{ AcknowledgeOperation, DeleteOperation, InsertOperation, Operation, OperationId, OperationType, UpdateOperation, validate_events, @@ -14,8 +17,8 @@ pub use operation::{ pub use readonly_engine::ReadOnlyPersistenceEngine; pub use space::{ ArtPersistenceKey, IndexTableOfContents, SpaceArcticIndex, SpaceCongeeIndex, SpaceData, SpaceDataOps, SpaceIndex, - SpaceIndexOps, SpaceIndexUnsized, SpaceSecondaryIndexOps, map_index_pages_to_toc_and_general, - map_unsized_index_pages_to_toc_and_general, reconstruct_multi_index_nodes, + SpaceIndexOps, SpaceIndexUnsized, SpaceLogicalIndex, SpaceLogicalIndexUnsized, SpaceSecondaryIndexOps, + map_index_pages_to_toc_and_general, map_unsized_index_pages_to_toc_and_general, reconstruct_multi_index_nodes, }; pub use task::PersistenceTask; diff --git a/src/persistence/space/logical_index.rs b/src/persistence/space/logical_index.rs new file mode 100644 index 0000000..f5d637d --- /dev/null +++ b/src/persistence/space/logical_index.rs @@ -0,0 +1,425 @@ +//! Background logical-to-structural CDC translation for WorkTablesIndex. +//! +//! The disk shadow is reconstructed from the existing DataBucket index pages. +//! Foreground tables can therefore emit compact logical Set/Remove events while +//! this persistence-worker-owned index derives the structural events required +//! by the unchanged WTI disk format. + +use std::fmt::Debug; +use std::hash::Hash; +use std::path::{Path, PathBuf}; + +use data_bucket::{Link, SizeMeasurable, SpaceId, VariableSizeMeasurable}; +use indexset::cdc::change::ChangeEvent; +use indexset::concurrent::map::BTreeMap; +use indexset::core::node::NodeLike; +use indexset::core::pair::Pair; +use rkyv::de::Pool; +use rkyv::rancor::Strategy; +use rkyv::ser::Serializer; +use rkyv::ser::allocator::ArenaHandle; +use rkyv::ser::sharing::Share; +use rkyv::util::AlignedVec; +use rkyv::{Archive, Deserialize, Serialize, rancor}; +use tokio::fs::File; + +use crate::UnsizedNode; +use crate::persistence::space::BatchChangeEvent; +use crate::persistence::{PersistenceIndexCorruption, SpaceIndex, SpaceIndexOps, SpaceIndexUnsized}; +use crate::prelude::WT_INDEX_EXTENSION; + +fn translate_logical_event( + index_path: &Path, + shadow: &BTreeMap, + event: ChangeEvent>, +) -> Result>>, PersistenceIndexCorruption> +where + T: Debug + Eq + Hash + Clone + Send + Ord + 'static, + Node: NodeLike> + Send + 'static, +{ + match event { + ChangeEvent::InsertAt { + max_value, + value, + index: 0, + .. + } if max_value.key == value.key && max_value.value == value.value => { + let (_, events) = shadow.insert_cdc(value.key, value.value); + Ok(events) + } + ChangeEvent::RemoveAt { + max_value, + value, + index: 0, + .. + } if max_value.key == value.key && max_value.value == value.value => { + let found = shadow.get(&value.key).map(|entry| entry.get().value); + if found != Some(value.value) { + return Err(PersistenceIndexCorruption::new( + index_path, + format!( + "logical WTI shadow diverged while removing key {:?}: expected {:?}, found {:?}", + value.key, value.value, found, + ), + )); + } + let (_, events) = shadow.remove_cdc(&value.key); + Ok(events) + } + _ => Err(PersistenceIndexCorruption::new( + index_path, + "logical WTI persistence received a structural or malformed event", + )), + } +} + +fn translate_logical_batch( + index_path: &Path, + shadow: &BTreeMap, + mut events: BatchChangeEvent, +) -> Result, PersistenceIndexCorruption> +where + T: Debug + Eq + Hash + Clone + Send + Ord + 'static, + Node: NodeLike> + Send + 'static, +{ + // BatchOperation already sorts every per-index stream by event id before + // dispatch. Sort again at this logical/structural boundary so direct + // SpaceIndexOps callers and future batching changes cannot reorder a + // same-key Set/Remove pair after the foreground stripe guard is released. + events.sort_by_key(ChangeEvent::id); + let mut structural = Vec::new(); + for event in events { + structural.extend(translate_logical_event(index_path, shadow, event)?); + } + Ok(structural) +} + +/// Sized-key WTI persistence with foreground logical CDC and a background +/// structural shadow. The wrapped `SpaceIndex` retains the existing file +/// layout byte-for-byte. +pub struct SpaceLogicalIndex +where + T: Send + Ord + Eq + Clone + 'static, +{ + index_path: PathBuf, + shadow: BTreeMap, + disk: SpaceIndex, +} + +impl Debug for SpaceLogicalIndex +where + T: Send + Ord + Eq + Clone + 'static, +{ + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.debug_struct("SpaceLogicalIndex").finish_non_exhaustive() + } +} + +impl SpaceLogicalIndex +where + T: Archive + + Ord + + Eq + + Hash + + Clone + + Default + + Debug + + SizeMeasurable + + for<'a> Serialize, Share>, rancor::Error>> + + Send + + Sync + + 'static, + ::Archived: Deserialize> + + Ord + + Eq + + Debug + + for<'a> rkyv::bytecheck::CheckBytes>, +{ + async fn new(path: String, version: u32) -> eyre::Result { + let index_path = PathBuf::from(&path); + let mut disk = SpaceIndex::new(path, SpaceId::from(0), version).await?; + let shadow = disk.parse_indexset().await?; + Ok(Self { + index_path, + shadow, + disk, + }) + } +} + +impl SpaceIndexOps for SpaceLogicalIndex +where + T: Archive + + Ord + + Eq + + Hash + + Clone + + Default + + Debug + + SizeMeasurable + + for<'a> Serialize, Share>, rancor::Error>> + + Send + + Sync + + 'static, + ::Archived: Deserialize> + + Ord + + Eq + + Debug + + for<'a> rkyv::bytecheck::CheckBytes>, +{ + async fn primary_from_table_files_path + Send>(path: S, version: u32) -> eyre::Result { + Self::new(format!("{}/primary{}", path.as_ref(), WT_INDEX_EXTENSION), version).await + } + + async fn secondary_from_table_files_path + Send, S2: AsRef + Send>( + path: S1, + name: S2, + version: u32, + ) -> eyre::Result { + Self::new( + format!("{}/{}{}", path.as_ref(), name.as_ref(), WT_INDEX_EXTENSION), + version, + ) + .await + } + + async fn bootstrap(file: &mut File, table_name: String, version: u32) -> eyre::Result<()> { + SpaceIndex::::bootstrap(file, table_name, version).await + } + + async fn process_change_event(&mut self, event: ChangeEvent>) -> eyre::Result<()> { + let events = translate_logical_event(&self.index_path, &self.shadow, event)?; + self.disk.process_change_event_batch(events).await + } + + async fn process_change_event_batch(&mut self, events: BatchChangeEvent) -> eyre::Result<()> { + let events = translate_logical_batch(&self.index_path, &self.shadow, events)?; + self.disk.process_change_event_batch(events).await + } +} + +/// Variable-sized-key counterpart to [`SpaceLogicalIndex`]. +pub struct SpaceLogicalIndexUnsized +where + T: Send + Ord + Eq + Clone + Default + Debug + SizeMeasurable + VariableSizeMeasurable + 'static, +{ + index_path: PathBuf, + shadow: BTreeMap>>, + disk: SpaceIndexUnsized, +} + +impl Debug for SpaceLogicalIndexUnsized +where + T: Send + Ord + Eq + Clone + Default + Debug + SizeMeasurable + VariableSizeMeasurable + 'static, +{ + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("SpaceLogicalIndexUnsized") + .finish_non_exhaustive() + } +} + +impl SpaceLogicalIndexUnsized +where + T: Archive + + Ord + + Eq + + Hash + + Clone + + Default + + Debug + + SizeMeasurable + + VariableSizeMeasurable + + for<'a> Serialize, Share>, rancor::Error>> + + Send + + Sync + + 'static, + ::Archived: Deserialize> + + Ord + + Eq + + Debug + + for<'a> rkyv::bytecheck::CheckBytes>, +{ + async fn new(path: String, version: u32) -> eyre::Result { + let index_path = PathBuf::from(&path); + let mut disk = SpaceIndexUnsized::new(path, SpaceId::from(0), version).await?; + let shadow = disk.parse_indexset().await?; + Ok(Self { + index_path, + shadow, + disk, + }) + } +} + +impl SpaceIndexOps for SpaceLogicalIndexUnsized +where + T: Archive + + Ord + + Eq + + Hash + + Clone + + Default + + Debug + + SizeMeasurable + + VariableSizeMeasurable + + for<'a> Serialize, Share>, rancor::Error>> + + Send + + Sync + + 'static, + ::Archived: Deserialize> + + Ord + + Eq + + Debug + + for<'a> rkyv::bytecheck::CheckBytes>, +{ + async fn primary_from_table_files_path + Send>(path: S, version: u32) -> eyre::Result { + Self::new(format!("{}/primary{}", path.as_ref(), WT_INDEX_EXTENSION), version).await + } + + async fn secondary_from_table_files_path + Send, S2: AsRef + Send>( + path: S1, + name: S2, + version: u32, + ) -> eyre::Result { + Self::new( + format!("{}/{}{}", path.as_ref(), name.as_ref(), WT_INDEX_EXTENSION), + version, + ) + .await + } + + async fn bootstrap(file: &mut File, table_name: String, version: u32) -> eyre::Result<()> { + SpaceIndexUnsized::::bootstrap(file, table_name, version).await + } + + async fn process_change_event(&mut self, event: ChangeEvent>) -> eyre::Result<()> { + let events = translate_logical_event(&self.index_path, &self.shadow, event)?; + self.disk.process_change_event_batch(events).await + } + + async fn process_change_event_batch(&mut self, events: BatchChangeEvent) -> eyre::Result<()> { + let events = translate_logical_batch(&self.index_path, &self.shadow, events)?; + self.disk.process_change_event_batch(events).await + } +} + +#[cfg(test)] +mod tests { + use data_bucket::page::PageId; + + use super::*; + + fn link(offset: u32) -> Link { + Link { + page_id: PageId::from(1), + offset, + length: 8, + } + } + + #[test] + fn logical_events_rebuild_structural_events_on_the_shadow() { + let shadow = BTreeMap::::default(); + let pair = Pair { key: 7, value: link(7) }; + let events = translate_logical_event( + Path::new("test.wt.idx"), + &shadow, + ChangeEvent::InsertAt { + event_id: 0.into(), + max_value: pair.clone(), + value: pair, + index: 0, + }, + ) + .unwrap(); + assert!(!events.is_empty()); + assert_eq!(shadow.get(&7).map(|entry| entry.get().value), Some(link(7))); + } + + #[test] + fn logical_event_requires_an_exact_key_and_link_sentinel() { + let shadow = BTreeMap::::default(); + let result = translate_logical_event( + Path::new("test.wt.idx"), + &shadow, + ChangeEvent::InsertAt { + event_id: 0.into(), + max_value: Pair { key: 7, value: link(8) }, + value: Pair { key: 7, value: link(7) }, + index: 0, + }, + ); + + assert!(result.is_err()); + assert!(shadow.get(&7).is_none()); + } + + #[test] + fn logical_batch_restores_event_id_order_after_reversed_delivery() { + let shadow = BTreeMap::::default(); + let pair = Pair { key: 7, value: link(7) }; + let insert = ChangeEvent::InsertAt { + event_id: 0.into(), + max_value: pair.clone(), + value: pair.clone(), + index: 0, + }; + let remove = ChangeEvent::RemoveAt { + event_id: 1.into(), + max_value: pair.clone(), + value: pair, + index: 0, + }; + + let structural = translate_logical_batch(Path::new("test.wt.idx"), &shadow, vec![remove, insert]).unwrap(); + + assert!(!structural.is_empty()); + assert!(shadow.get(&7).is_none()); + } + + #[test] + fn divergence_is_typed_and_does_not_mutate_the_shadow() { + let shadow = BTreeMap::::default(); + shadow.insert(7, link(7)); + let pair = Pair { key: 7, value: link(8) }; + + let error = translate_logical_event( + Path::new("test.wt.idx"), + &shadow, + ChangeEvent::RemoveAt { + event_id: 1.into(), + max_value: pair.clone(), + value: pair, + index: 0, + }, + ) + .unwrap_err(); + + assert_eq!(error.path(), Path::new("test.wt.idx")); + assert!(error.reason().contains("shadow diverged")); + assert_eq!(shadow.get(&7).map(|entry| entry.get().value), Some(link(7))); + } + + #[test] + fn logical_set_replacement_derives_the_new_structural_link() { + let shadow = BTreeMap::::default(); + shadow.insert(7, link(7)); + let replacement = Pair { key: 7, value: link(8) }; + + let structural = translate_logical_event( + Path::new("test.wt.idx"), + &shadow, + ChangeEvent::InsertAt { + event_id: 1.into(), + max_value: replacement.clone(), + value: replacement, + index: 0, + }, + ) + .unwrap(); + + assert!(!structural.is_empty()); + assert_eq!(shadow.get(&7).map(|entry| entry.get().value), Some(link(8))); + } +} diff --git a/src/persistence/space/mod.rs b/src/persistence/space/mod.rs index 12ca9c3..843ca8b 100644 --- a/src/persistence/space/mod.rs +++ b/src/persistence/space/mod.rs @@ -1,6 +1,7 @@ mod art_index; mod data; mod index; +mod logical_index; use std::collections::HashMap; use std::future::Future; @@ -18,6 +19,7 @@ pub use index::{ IndexTableOfContents, SpaceIndex, SpaceIndexUnsized, map_index_pages_to_toc_and_general, map_unsized_index_pages_to_toc_and_general, reconstruct_multi_index_nodes, }; +pub use logical_index::{SpaceLogicalIndex, SpaceLogicalIndexUnsized}; pub type BatchData = HashMap)>>; diff --git a/src/persistence/task.rs b/src/persistence/task.rs index f639341..5e4b51e 100644 --- a/src/persistence/task.rs +++ b/src/persistence/task.rs @@ -13,7 +13,9 @@ use tokio::task::JoinHandle; use worktable_codegen::worktable; use crate::persistence::operation::{BatchInnerRow, BatchInnerWorkTable, BatchOperation, OperationId, PosByOpIdQuery}; -use crate::persistence::{PersistenceEngine, PersistenceError, PersistenceResult, PersistenceState}; +use crate::persistence::{ + PersistenceEngine, PersistenceError, PersistenceIndexCorruption, PersistenceResult, PersistenceState, +}; use crate::prelude::*; use crate::util::OptimizedVec; use crate::vacuum::VacuumPersistence; @@ -81,7 +83,10 @@ impl PersistenceLifecycle { let error = match &*state { PersistenceState::Failed(error) => error.clone(), _ => { - let error = Arc::new(PersistenceError::Engine(report)); + let error = Arc::new(match report.downcast::() { + Ok(corruption) => PersistenceError::IndexCorruption(corruption), + Err(report) => PersistenceError::Engine(report), + }); *state = PersistenceState::Failed(error.clone()); error } @@ -507,6 +512,25 @@ mod lifecycle_tests { assert_eq!(&*events.lock(), &["batch", "reclaim", "batch"]); } + + #[test] + fn typed_index_corruption_quarantines_the_persistence_lifecycle() { + let lifecycle = PersistenceLifecycle::new(); + let error = lifecycle.fail( + PersistenceIndexCorruption::new("table/primary.wt.idx", "shadow diverged from logical stream").into(), + ); + + match error.as_ref() { + PersistenceError::IndexCorruption(corruption) => { + assert_eq!(corruption.path(), std::path::Path::new("table/primary.wt.idx")); + assert!(corruption.reason().contains("shadow diverged")); + } + other => panic!("expected typed index corruption, got {other:?}"), + } + + let intake_error = lifecycle.ensure_running().unwrap_err(); + assert!(Arc::ptr_eq(&error, &intake_error)); + } } #[derive(Debug)] diff --git a/tests/worktable/index_backends.rs b/tests/worktable/index_backends.rs index 0924722..f789ae3 100644 --- a/tests/worktable/index_backends.rs +++ b/tests/worktable/index_backends.rs @@ -455,6 +455,71 @@ async fn native_art_backends_recover_concurrent_same_row_updates() { remove_dir_if_exists(ROOT.to_string()).await; } +#[cfg(feature = "logical-index-persistence")] +#[tokio::test] +async fn logical_wti_recovers_concurrent_same_row_updates() { + use std::sync::Arc; + + use provider_switch_wti as wti; + use tokio::sync::Barrier; + + const ROOT: &str = "tests/data/index_backend_logical_wti_concurrent"; + const WORKERS: u64 = 8; + const UPDATES_PER_WORKER: u64 = 100; + remove_dir_if_exists(ROOT.to_string()).await; + + let config = DiskConfig::new_with_table_name( + ROOT, + wti::ProviderSwitchWorkTable::name_snake_case(), + wti::ProviderSwitchWorkTable::version(), + ); + let engine = wti::ProviderSwitchPersistenceEngine::new(config.clone()).await.unwrap(); + let table = Arc::new(wti::ProviderSwitchWorkTable::load(engine).await.unwrap()); + let id: u64 = table.get_next_pk().into(); + table.insert(wti::ProviderSwitchRow { id, unique_key: 1 }).unwrap(); + + let barrier = Arc::new(Barrier::new(WORKERS as usize + 1)); + let mut workers = Vec::new(); + for worker in 0..WORKERS { + let table = Arc::clone(&table); + let barrier = Arc::clone(&barrier); + workers.push(tokio::spawn(async move { + barrier.wait().await; + for update in 0..UPDATES_PER_WORKER { + table + .update(wti::ProviderSwitchRow { + id, + unique_key: 10_000 + worker * UPDATES_PER_WORKER + update, + }) + .await + .unwrap(); + } + })); + } + barrier.wait().await; + for worker in workers { + worker.await.unwrap(); + } + + let expected = table.select(id).unwrap(); + table.wait_for_ops().await.unwrap(); + drop(table); + + let engine = wti::ProviderSwitchPersistenceEngine::new(config).await.unwrap(); + let table = wti::ProviderSwitchWorkTable::load(engine).await.unwrap(); + assert_eq!(table.select(id), Some(expected.clone())); + assert_eq!(table.select_by_unique_key(expected.unique_key), Some(expected.clone())); + for key in 10_000..10_000 + WORKERS * UPDATES_PER_WORKER { + if key != expected.unique_key { + assert!(table.select_by_unique_key(key).is_none()); + } + } + table.wait_for_ops().await.unwrap(); + drop(table); + + remove_dir_if_exists(ROOT.to_string()).await; +} + #[tokio::test] async fn persisted_tables_can_switch_between_wti_and_upstream_without_rebuild() { use provider_switch_upstream as upstream;