From ec1e7de855df6de581454f4d82b336bafa3ce7fe Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 4 Aug 2026 00:55:07 +0700 Subject: [PATCH 1/5] feat: persist native ART index backends --- Cargo.toml | 5 +- codegen/src/common/model/index.rs | 2 +- codegen/src/generators/index_backend.rs | 39 + codegen/src/generators/persist/index/info.rs | 14 +- codegen/src/generators/persist/index/mod.rs | 4 +- codegen/src/generators/persist/table/impls.rs | 37 +- codegen/src/generators/persist/table/mod.rs | 27 +- codegen/src/persist_index/generator.rs | 187 ++-- codegen/src/persist_index/space/index.rs | 62 +- codegen/src/persist_table/generator/mod.rs | 2 + codegen/src/persist_table/generator/space.rs | 8 + .../persist_table/generator/space_file/mod.rs | 53 +- .../generator/space_file/worktable_impls.rs | 6 +- codegen/src/persist_table/parser.rs | 10 + codegen/src/worktable/mod.rs | 39 +- docs/art-index-persistence-plan.md | 107 ++ docs/index-backend-dsl-proposal.md | 46 +- src/index/arctic.rs | 26 + src/index/congee.rs | 40 +- src/index/mod.rs | 2 + src/index/persistent_art.rs | 273 +++++ src/index/table_index/mod.rs | 11 +- src/lib.rs | 20 +- src/mem_stat/mod.rs | 14 +- src/persistence/mod.rs | 6 +- src/persistence/space/art_index.rs | 975 ++++++++++++++++++ src/persistence/space/mod.rs | 2 + tests/worktable/index_backends.rs | 127 +++ 28 files changed, 1965 insertions(+), 179 deletions(-) create mode 100644 docs/art-index-persistence-plan.md create mode 100644 src/index/persistent_art.rs create mode 100644 src/persistence/space/art_index.rs diff --git a/Cargo.toml b/Cargo.toml index ee23cfae..d46e93f5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -28,9 +28,10 @@ versioned-row-publication = ["worktable_codegen/versioned-row-publication"] [dependencies] async-trait = "0.1.89" -arctic-map = "=0.1.4" -congee = "=0.4.1" +arctic = { package = "arctic-wt", git = "https://github.com/pathscale/arctic-wt", rev = "e13fc7df3c040f14ae66c1cb56b1bd0a3f6da3fc" } +congee = { package = "congee-wt", git = "https://github.com/pathscale/congee-wt", rev = "005bfb1968e781800176f2d7e465e6a1af630e1a" } convert_case = "0.6.0" +crc32fast = "1.5.0" data_bucket = "=0.5.1" # data_bucket = { git = "https://github.com/pathscale/DataBucket", branch = "page_cdc_correction", version = "0.2.7" } # data_bucket = { path = "../DataBucket", version = "0.3.14" } diff --git a/codegen/src/common/model/index.rs b/codegen/src/common/model/index.rs index f3297324..b133c3b6 100644 --- a/codegen/src/common/model/index.rs +++ b/codegen/src/common/model/index.rs @@ -15,7 +15,7 @@ pub enum IndexBackend { } impl IndexBackend { - pub fn is_memory_only(self) -> bool { + pub fn requires_explicit_persistence(self) -> bool { matches!(self, Self::Congee | Self::Arctic) } diff --git a/codegen/src/generators/index_backend.rs b/codegen/src/generators/index_backend.rs index 70c56db6..ca923da6 100644 --- a/codegen/src/generators/index_backend.rs +++ b/codegen/src/generators/index_backend.rs @@ -31,6 +31,21 @@ pub(crate) fn unique_index_type( } } +/// Generates the persisted-table variant. ART backends receive a write-side +/// sequencing wrapper; memory-only tables keep the zero-overhead native type. +pub(crate) fn persistent_unique_index_type( + backend: IndexBackend, + key: &TokenStream, + value: &TokenStream, + worktables_node: Option, +) -> syn::Result { + match backend { + IndexBackend::Congee => Ok(quote! { PersistentCongeeIndex<#key, #value> }), + IndexBackend::Arctic => Ok(quote! { PersistentArcticIndex<#key, #value> }), + _ => unique_index_type(backend, key, value, worktables_node), + } +} + /// Generates the small codec needed when an ART is used for WorkTable's /// generated primary-key newtype. ART backends intentionally accept only the /// lossless, native integer shapes supported by their public APIs. @@ -67,6 +82,18 @@ pub(crate) fn primary_key_backend_impl( Self(value as #field) } } + + impl ArtPersistenceKey for #primary_key { + const WIDTH: u8 = <#field as ArtPersistenceKey>::WIDTH; + + fn encode_art_key(&self, output: &mut Vec) { + self.0.encode_art_key(output) + } + + fn decode_art_key(bytes: &[u8]) -> eyre::Result { + Ok(Self(<#field as ArtPersistenceKey>::decode_art_key(bytes)?)) + } + } }, )) } @@ -88,6 +115,18 @@ pub(crate) fn primary_key_backend_impl( Self(value) } } + + impl ArtPersistenceKey for #primary_key { + const WIDTH: u8 = <#field as ArtPersistenceKey>::WIDTH; + + fn encode_art_key(&self, output: &mut Vec) { + self.0.encode_art_key(output) + } + + fn decode_art_key(bytes: &[u8]) -> eyre::Result { + Ok(Self(<#field as ArtPersistenceKey>::decode_art_key(bytes)?)) + } + } }, )) } diff --git a/codegen/src/generators/persist/index/info.rs b/codegen/src/generators/persist/index/info.rs index 4e0394d7..38a6addf 100644 --- a/codegen/src/generators/persist/index/info.rs +++ b/codegen/src/generators/persist/index/info.rs @@ -26,15 +26,25 @@ impl PersistGenerator { let index_name_str = index_field_name.to_string(); if idx.is_unique { + let (capacity, node_count) = match idx.backend { + crate::common::model::IndexBackend::WorktablesIndex + | crate::common::model::IndexBackend::Indexset => ( + quote! { self.#index_field_name.capacity() }, + quote! { self.#index_field_name.node_count() }, + ), + crate::common::model::IndexBackend::Congee | crate::common::model::IndexBackend::Arctic => { + (quote! { self.#index_field_name.len() }, quote! { 0 }) + } + }; quote! { info.push(IndexInfo { name: #index_name_str.to_string(), index_type: IndexKind::Unique, key_count: self.#index_field_name.len(), - capacity: self.#index_field_name.capacity(), + capacity: #capacity, heap_size: self.#index_field_name.heap_size(), used_size: self.#index_field_name.used_size(), - node_count: self.#index_field_name.node_count(), + node_count: #node_count, }); } } else { diff --git a/codegen/src/generators/persist/index/mod.rs b/codegen/src/generators/persist/index/mod.rs index 0b86b239..b2068c62 100644 --- a/codegen/src/generators/persist/index/mod.rs +++ b/codegen/src/generators/persist/index/mod.rs @@ -3,7 +3,7 @@ mod info; mod usual; use crate::common::name_generator::{WorktableNameGenerator, is_float, is_unsized}; -use crate::generators::index_backend::unique_index_type; +use crate::generators::index_backend::persistent_unique_index_type; use crate::generators::persist::PersistGenerator; use convert_case::{Case, Casing}; use proc_macro2::TokenStream; @@ -57,7 +57,7 @@ impl PersistGenerator { } else { None }; - let index_type = unique_index_type(idx.backend, &t, &value_type, worktables_node)?; + let index_type = persistent_unique_index_type(idx.backend, &t, &value_type, worktables_node)?; quote! { #i: #index_type } } else { if is_unsized(&t.to_string()) { diff --git a/codegen/src/generators/persist/table/impls.rs b/codegen/src/generators/persist/table/impls.rs index 4f23eb84..c7474f6f 100644 --- a/codegen/src/generators/persist/table/impls.rs +++ b/codegen/src/generators/persist/table/impls.rs @@ -72,10 +72,6 @@ impl PersistGenerator { }) .collect::>(); let pk_types_unsized = is_unsized_vec(pk_types); - let pk_map = match self.columns.primary_index_backend { - crate::common::model::IndexBackend::Indexset => quote! { UpstreamIndexMap }, - _ => quote! { IndexMap }, - }; let index_setup = if pk_types_unsized { quote! { inner.primary_index = std::sync::Arc::new(PrimaryIndex { @@ -84,12 +80,33 @@ impl PersistGenerator { }); } } else { - quote! { - let size = get_index_page_size_from_data_length::<#pk_type>(#const_name); - inner.primary_index = std::sync::Arc::new(PrimaryIndex { - pk_map: #pk_map::<_, OffsetEqLink<#const_name>>::with_maximum_node_size(size), - reverse_pk_map: IndexMap::new(), - }); + match self.columns.primary_index_backend { + 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), + reverse_pk_map: IndexMap::new(), + }); + }, + crate::common::model::IndexBackend::Indexset => quote! { + let size = get_index_page_size_from_data_length::<#pk_type>(#const_name); + inner.primary_index = std::sync::Arc::new(PrimaryIndex { + pk_map: UpstreamIndexMap::<_, OffsetEqLink<#const_name>>::with_maximum_node_size(size), + reverse_pk_map: IndexMap::new(), + }); + }, + crate::common::model::IndexBackend::Arctic => quote! { + inner.primary_index = std::sync::Arc::new(PrimaryIndex { + pk_map: PersistentArcticIndex::<#pk_type, OffsetEqLink<#const_name>>::default(), + reverse_pk_map: IndexMap::new(), + }); + }, + crate::common::model::IndexBackend::Congee => quote! { + inner.primary_index = std::sync::Arc::new(PrimaryIndex { + pk_map: PersistentCongeeIndex::<#pk_type, OffsetEqLink<#const_name>>::default(), + reverse_pk_map: IndexMap::new(), + }); + }, } }; diff --git a/codegen/src/generators/persist/table/mod.rs b/codegen/src/generators/persist/table/mod.rs index b8930c92..339440fd 100644 --- a/codegen/src/generators/persist/table/mod.rs +++ b/codegen/src/generators/persist/table/mod.rs @@ -6,7 +6,7 @@ use proc_macro2::{Literal, TokenStream}; use quote::quote; use crate::common::name_generator::{WorktableNameGenerator, is_unsized_vec}; -use crate::generators::index_backend::unique_index_type; +use crate::generators::index_backend::persistent_unique_index_type; use crate::generators::persist::PersistGenerator; impl PersistGenerator { @@ -84,25 +84,28 @@ impl PersistGenerator { }) .collect::>(); let pk_types_unsized = is_unsized_vec(pk_types); - let pk_upstream = matches!( - self.columns.primary_index_backend, - crate::common::model::IndexBackend::Indexset - ); - - let derive = match (pk_types_unsized, pk_upstream) { - (true, true) => quote! { + let derive = match (pk_types_unsized, self.columns.primary_index_backend) { + (true, crate::common::model::IndexBackend::Indexset) => quote! { #[derive(Debug, PersistTable)] #[table(pk_unsized, pk_upstream)] }, - (true, false) => quote! { + (true, _) => quote! { #[derive(Debug, PersistTable)] #[table(pk_unsized)] }, - (false, true) => quote! { + (false, crate::common::model::IndexBackend::Indexset) => quote! { #[derive(Debug, PersistTable)] #[table(pk_upstream)] }, - (false, false) => quote! { + (false, crate::common::model::IndexBackend::Arctic) => quote! { + #[derive(Debug, PersistTable)] + #[table(pk_arctic)] + }, + (false, crate::common::model::IndexBackend::Congee) => quote! { + #[derive(Debug, PersistTable)] + #[table(pk_congee)] + }, + (false, crate::common::model::IndexBackend::WorktablesIndex) => quote! { #[derive(Debug, PersistTable)] }, }; @@ -116,7 +119,7 @@ impl PersistGenerator { } else { None }; - let node_type = unique_index_type( + let node_type = persistent_unique_index_type( self.columns.primary_index_backend, &key_type, &value_type, diff --git a/codegen/src/persist_index/generator.rs b/codegen/src/persist_index/generator.rs index 66e63646..9af18581 100644 --- a/codegen/src/persist_index/generator.rs +++ b/codegen/src/persist_index/generator.rs @@ -19,13 +19,20 @@ pub struct Generator { pub attributes: PersistIndexAttributes, } -struct IndexLayout { +pub(super) struct IndexLayout { type_ident: Ident, is_unique: bool, uses_upstream: bool, + pub(super) art_backend: Option, } -fn index_layout(field: &Field) -> syn::Result { +#[derive(Copy, Clone, Debug, Eq, PartialEq)] +pub(super) enum ArtBackend { + Arctic, + Congee, +} + +pub(super) fn index_layout(field: &Field) -> syn::Result { let syn::Type::Path(type_path) = &field.ty else { return Err(syn::Error::new_spanned( &field.ty, @@ -39,14 +46,16 @@ 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) = match type_ident.to_string().as_str() { - "IndexMap" | "TreeIndex" => (true, false), - "UpstreamIndexMap" => (true, true), - "IndexMultiMap" | "TreeMultiIndex" => (false, false), + 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)), _ => { return Err(syn::Error::new_spanned( &field.ty, - "unsupported persisted index type; use IndexMap, UpstreamIndexMap, or IndexMultiMap directly", + "unsupported persisted index type; use a WorkTable-generated index backend directly", )); } }; @@ -54,6 +63,7 @@ fn index_layout(field: &Field) -> syn::Result { type_ident, is_unique, uses_upstream, + art_backend, }) } @@ -123,24 +133,31 @@ impl Generator { let name_ident = name_generator.get_persisted_index_ident(); let fields: Vec<_> = self - .field_types + .struct_def + .fields .iter() - .map(|(i, t)| { - if is_unsized(&t.to_string()) { + .map(|field| { + let layout = index_layout(field)?; + let i = field.ident.as_ref().expect("index fields should be named"); + let t = self.field_types.get(i).expect("field type was collected"); + if layout.art_backend.is_some() { + let field_type = &field.ty; + Ok(quote! { #i: #field_type, }) + } else if is_unsized(&t.to_string()) { let const_size = name_generator.get_page_inner_size_const_ident(); - quote! { + Ok(quote! { #i: (Vec>>, Vec>>), - } + }) } else { - quote! { + Ok(quote! { #i: (Vec>>, Vec>>), - } + }) } }) - .collect(); + .collect::>>()?; Ok(quote! { - #[derive(Debug, Default, Clone)] + #[derive(Debug, Default)] pub struct #name_ident { #(#fields)* } @@ -170,35 +187,52 @@ impl Generator { fn gen_persist_fn(&self) -> TokenStream { let name_generator = WorktableNameGenerator::from_index_ident(&self.struct_def.ident); let ident = name_generator.get_work_table_ident(); + let inner_const_name = name_generator.get_page_inner_size_const_ident(); + let version_const_name = name_generator.get_version_const_ident(); let index_extension = Literal::string(WT_INDEX_EXTENSION); let persist_logic = self .struct_def .fields .iter() - .map(|f| { - f.ident - .as_ref() - .expect("index fields should always be named fields") - }) - .map(|i| { + .map(|field| { + let layout = index_layout(field)?; + let i = field.ident.as_ref().expect("index fields should be named"); + let ty = self.field_types.get(i).expect("field type was collected"); let index_name_literal = Literal::string(i.to_string().as_str()); - quote! { - { - let mut file = tokio::fs::File::create(format!("{}/{}{}", path, #index_name_literal, #index_extension)).await?; - let mut info = #ident::space_info_default(); - info.inner.page_count = self.#i.1.len() as u32 + self.#i.0.len() as u32; - persist_page(&mut info, &mut file).await?; - for mut page in &mut self.#i.0 { - persist_page(&mut page, &mut file).await?; + Ok(match layout.art_backend { + Some(ArtBackend::Arctic) => quote! { + SpaceArcticIndex::<#ty, { #inner_const_name as u32 }>::write_checkpoint( + format!("{}/{}{}", path, #index_name_literal, #index_extension), + #version_const_name, + &mut self.#i, + ).await?; + }, + Some(ArtBackend::Congee) => quote! { + SpaceCongeeIndex::<#ty, { #inner_const_name as u32 }>::write_checkpoint( + format!("{}/{}{}", path, #index_name_literal, #index_extension), + #version_const_name, + &mut self.#i, + ).await?; + }, + None => quote! { + { + let mut file = tokio::fs::File::create(format!("{}/{}{}", path, #index_name_literal, #index_extension)).await?; + let mut info = #ident::space_info_default(); + info.inner.page_count = self.#i.1.len() as u32 + self.#i.0.len() as u32; + persist_page(&mut info, &mut file).await?; + for mut page in &mut self.#i.0 { + persist_page(&mut page, &mut file).await?; + } + for mut page in &mut self.#i.1 { + persist_page(&mut page, &mut file).await?; + } } - for mut page in &mut self.#i.1 { - persist_page(&mut page, &mut file).await?; - } - } - } + }, + }) }) - .collect::>(); + .collect::>>() + .expect("generated index layouts were validated"); quote! { pub async fn persist(&mut self, path: &str) -> eyre::Result<()> @@ -214,41 +248,52 @@ impl Generator { fn gen_parse_from_file_fn(&self) -> TokenStream { let name_generator = WorktableNameGenerator::from_index_ident(&self.struct_def.ident); let page_const_name = name_generator.get_page_size_const_ident(); + let inner_const_name = name_generator.get_page_inner_size_const_ident(); + let version_const_name = name_generator.get_version_const_ident(); let index_extension = Literal::string(WT_INDEX_EXTENSION); let field_names_literals: Vec<_> = self .struct_def .fields .iter() - .map(|f| ( - Literal::string( - f.ident - .as_ref() - .expect("index fields should always be named fields") - .to_string() - .as_str() - ), - f.ident - .as_ref() - .expect("index fields should always be named fields") - )) - .map(|(l, i)| quote! { - let #i = { - let mut #i = vec![]; - let mut file = tokio::fs::File::open(format!("{}/{}{}", path, #l, #index_extension)).await?; - let info = parse_page::, { #page_const_name as u32 }>(&mut file, 0).await?; - let file_length = file.metadata().await?.len(); - let page_id = file_length / (#page_const_name as u64 + GENERAL_HEADER_SIZE as u64) + 1; - let next_page_id = std::sync::Arc::new(std::sync::atomic::AtomicU32::new(page_id as u32)); - let toc = IndexTableOfContents::<_, { #page_const_name as u32 }>::parse_from_file(&mut file, 0.into(), next_page_id.clone()).await?; - for page_id in toc.iter().map(|(_, page_id)| page_id) { - let index = parse_page::<_, { #page_const_name as u32 }>(&mut file, (*page_id).into()).await?; - #i.push(index); + .map(|field| { + let layout = index_layout(field)?; + let i = field.ident.as_ref().expect("index fields should be named"); + let ty = self.field_types.get(i).expect("field type was collected"); + let literal = Literal::string(i.to_string().as_str()); + Ok(match layout.art_backend { + Some(ArtBackend::Arctic) => quote! { + let #i = SpaceArcticIndex::<#ty, { #inner_const_name as u32 }>::load_index( + format!("{}/{}{}", path, #literal, #index_extension), + #version_const_name, + ).await?; + }, + Some(ArtBackend::Congee) => quote! { + let #i = SpaceCongeeIndex::<#ty, { #inner_const_name as u32 }>::load_index( + format!("{}/{}{}", path, #literal, #index_extension), + #version_const_name, + ).await?; + }, + None => quote! { + let #i = { + let mut #i = vec![]; + let mut file = tokio::fs::File::open(format!("{}/{}{}", path, #literal, #index_extension)).await?; + let info = parse_page::, { #page_const_name as u32 }>(&mut file, 0).await?; + let file_length = file.metadata().await?.len(); + let page_id = file_length / (#page_const_name as u64 + GENERAL_HEADER_SIZE as u64) + 1; + let next_page_id = std::sync::Arc::new(std::sync::atomic::AtomicU32::new(page_id as u32)); + let toc = IndexTableOfContents::<_, { #page_const_name as u32 }>::parse_from_file(&mut file, 0.into(), next_page_id.clone()).await?; + for page_id in toc.iter().map(|(_, page_id)| page_id) { + let index = parse_page::<_, { #page_const_name as u32 }>(&mut file, (*page_id).into()).await?; + #i.push(index); + } + (toc.pages, #i) + }; } - (toc.pages, #i) - }; + }) }) - .collect(); + .collect::>>() + .expect("generated index layouts were validated"); let idents = self .struct_def @@ -323,7 +368,15 @@ impl Generator { .field_types .get(i) .expect("should be available as constructed from same values"); - if is_unsized(&ty.to_string()) { + if layout.art_backend.is_some() { + let field_type = &field.ty; + Ok(quote! { + let #i: #field_type = Default::default(); + for (key, value) in self.#i.iter_values() { + #i.insert_value(key, value); + } + }) + } else if is_unsized(&ty.to_string()) { Ok(quote! { let mut pages = vec![]; for node in self.#i.iter_nodes() { @@ -476,7 +529,11 @@ impl Generator { } }; - if is_unsized(&ty.to_string()) { + if layout.art_backend.is_some() { + Ok(quote! { + let #i = persisted.#i; + }) + } else if is_unsized(&ty.to_string()) { if is_unique { let body = unique_reconstruct(quote! { let node = UnsizedNode::from_inner(inner, #const_name); diff --git a/codegen/src/persist_index/space/index.rs b/codegen/src/persist_index/space/index.rs index f5346ebe..d891d500 100644 --- a/codegen/src/persist_index/space/index.rs +++ b/codegen/src/persist_index/space/index.rs @@ -2,7 +2,7 @@ use proc_macro2::{Literal, TokenStream}; use quote::quote; use crate::common::name_generator::{WorktableNameGenerator, is_unsized}; -use crate::persist_index::generator::Generator; +use crate::persist_index::generator::{ArtBackend, Generator, index_layout}; impl Generator { pub fn gen_space_secondary_index_type(&self) -> TokenStream { @@ -11,20 +11,30 @@ impl Generator { let inner_const_name = name_generator.get_page_inner_size_const_ident(); let fields: Vec<_> = self - .field_types + .struct_def + .fields .iter() - .map(|(i, t)| { - if is_unsized(&t.to_string()) { - quote! { + .map(|field| { + let layout = index_layout(field)?; + let i = field.ident.as_ref().expect("index fields should be named"); + let t = self.field_types.get(i).expect("field type was collected"); + Ok(match layout.art_backend { + Some(ArtBackend::Arctic) => quote! { + #i: SpaceArcticIndex<#t, { #inner_const_name as u32}>, + }, + Some(ArtBackend::Congee) => quote! { + #i: SpaceCongeeIndex<#t, { #inner_const_name as u32}>, + }, + None if is_unsized(&t.to_string()) => quote! { #i: SpaceIndexUnsized<#t, { #inner_const_name as u32}>, - } - } else { - quote! { + }, + None => quote! { #i: SpaceIndex<#t, { #inner_const_name as u32}>, - } - } + }, + }) }) - .collect(); + .collect::>>() + .expect("generated index layouts were validated"); quote! { #[derive(Debug)] @@ -54,21 +64,31 @@ impl Generator { fn gen_space_secondary_index_from_table_files_path_fn(&self) -> TokenStream { let fields: Vec<_> = self - .field_types + .struct_def + .fields .iter() - .map(|(i, t)| { + .map(|field| { + let layout = index_layout(field)?; + let i = field.ident.as_ref().expect("index fields should be named"); + let t = self.field_types.get(i).expect("field type was collected"); let literal_name = Literal::string(i.to_string().as_str()); - if is_unsized(&t.to_string()) { - quote! { + Ok(match layout.art_backend { + Some(ArtBackend::Arctic) => quote! { + #i: SpaceArcticIndex::secondary_from_table_files_path(path, #literal_name, version).await?, + }, + Some(ArtBackend::Congee) => quote! { + #i: SpaceCongeeIndex::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?, - } - } else { - quote! { + }, + None => quote! { #i: SpaceIndex::secondary_from_table_files_path(path, #literal_name, version).await?, - } - } + }, + }) }) - .collect(); + .collect::>>() + .expect("generated index layouts were validated"); quote! { async fn from_table_files_path>(path: S, version: u32) -> eyre::Result { diff --git a/codegen/src/persist_table/generator/mod.rs b/codegen/src/persist_table/generator/mod.rs index 767a36c9..0ab25350 100644 --- a/codegen/src/persist_table/generator/mod.rs +++ b/codegen/src/persist_table/generator/mod.rs @@ -12,6 +12,8 @@ pub struct PersistTableAttributes { pub pk_unsized: bool, pub read_only: bool, pub pk_upstream: bool, + pub pk_arctic: bool, + pub pk_congee: bool, } pub struct Generator { diff --git a/codegen/src/persist_table/generator/space.rs b/codegen/src/persist_table/generator/space.rs index 70a4743b..1f81f397 100644 --- a/codegen/src/persist_table/generator/space.rs +++ b/codegen/src/persist_table/generator/space.rs @@ -40,6 +40,14 @@ impl Generator { quote! { SpaceIndexUnsized<#primary_key_type, { #inner_const_name as u32 }>, } + } else if self.attributes.pk_arctic { + quote! { + SpaceArcticIndex<#primary_key_type, { #inner_const_name as u32 }>, + } + } else if self.attributes.pk_congee { + quote! { + SpaceCongeeIndex<#primary_key_type, { #inner_const_name as u32 }>, + } } else { quote! { SpaceIndex<#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 5666b1fe..e3598653 100644 --- a/codegen/src/persist_table/generator/space_file/mod.rs +++ b/codegen/src/persist_table/generator/space_file/mod.rs @@ -32,6 +32,14 @@ impl Generator { quote! { pub primary_index: (Vec>>, Vec>>), } + } else if self.attributes.pk_arctic { + quote! { + pub primary_index: PersistentArcticIndex<#pk_type, OffsetEqLink<#inner_const_name>>, + } + } else if self.attributes.pk_congee { + quote! { + pub primary_index: PersistentCongeeIndex<#pk_type, OffsetEqLink<#inner_const_name>>, + } } else { quote! { pub primary_index: (Vec>>, Vec>>), @@ -53,6 +61,11 @@ impl Generator { let name_generator = WorktableNameGenerator::from_struct_ident(&self.struct_def.ident); let literal_name = name_generator.get_work_table_literal_name(); let version_const = name_generator.get_version_const_ident(); + let primary_page_count = if self.attributes.pk_arctic || self.attributes.pk_congee { + quote! { 1 } + } else { + quote! { self.primary_index.0.len() as u32 + self.primary_index.1.len() as u32 } + }; quote! { fn get_primary_index_info(&self) -> eyre::Result>> { @@ -82,7 +95,7 @@ impl Generator { inner } }; - info.inner.page_count = self.primary_index.0.len() as u32 + self.primary_index.1.len() as u32; + info.inner.page_count = #primary_page_count; Ok(info) } } @@ -141,6 +154,15 @@ impl Generator { } let primary_index = PrimaryIndex { pk_map, reverse_pk_map }; } + } else if self.attributes.pk_arctic || self.attributes.pk_congee { + quote! { + let pk_map = self.primary_index; + let reverse_pk_map = IndexMap::, #pk_type>::new(); + for (pk, link) in pk_map.iter_values() { + reverse_pk_map.insert(link, pk); + } + let primary_index = PrimaryIndex { pk_map, reverse_pk_map }; + } } else { let map_type = if self.pk_upstream { quote! { UpstreamIndexMap } @@ -263,6 +285,7 @@ impl Generator { let page_const_name = name_generator.get_page_size_const_ident(); let inner_const_name = name_generator.get_page_inner_size_const_ident(); let persisted_index_name = name_generator.get_persisted_index_ident(); + let version_const_name = name_generator.get_version_const_ident(); let index_extension = Literal::string(WT_INDEX_EXTENSION); let data_extension = Literal::string(WT_DATA_EXTENSION); @@ -276,9 +299,23 @@ impl Generator { } }; - quote! { - pub async fn parse_file(path: &str) -> eyre::Result { - let mut primary_index = { + let parse_primary = if self.attributes.pk_arctic { + quote! { + SpaceArcticIndex::<#pk_type, { #inner_const_name as u32 }>::load_index::<#inner_const_name>( + format!("{}/primary{}", path, #index_extension), + #version_const_name, + ).await? + } + } else if self.attributes.pk_congee { + quote! { + SpaceCongeeIndex::<#pk_type, { #inner_const_name as u32 }>::load_index::<#inner_const_name>( + format!("{}/primary{}", path, #index_extension), + #version_const_name, + ).await? + } + } else { + quote! { + { let mut primary_index = vec![]; let mut primary_file = tokio::fs::File::open(format!("{}/primary{}", path, #index_extension)).await?; let info = parse_page::, { #page_const_name as u32 }>(&mut primary_file, 0).await?; @@ -291,7 +328,13 @@ impl Generator { primary_index.push(index); } (toc.pages, primary_index) - }; + } + } + }; + + quote! { + pub async fn parse_file(path: &str) -> eyre::Result { + let primary_index = #parse_primary; let indexes = #persisted_index_name::parse_from_file(path).await?; let (data, data_info) = { diff --git a/codegen/src/persist_table/generator/space_file/worktable_impls.rs b/codegen/src/persist_table/generator/space_file/worktable_impls.rs index cacb6792..c6d95ee5 100644 --- a/codegen/src/persist_table/generator/space_file/worktable_impls.rs +++ b/codegen/src/persist_table/generator/space_file/worktable_impls.rs @@ -74,7 +74,11 @@ impl Generator { let name_generator = WorktableNameGenerator::from_struct_ident(&self.struct_def.ident); let pk_type = name_generator.get_primary_key_type_ident(); let const_name = name_generator.get_page_inner_size_const_ident(); - if self.attributes.pk_unsized { + if self.attributes.pk_arctic || self.attributes.pk_congee { + // ART durability is maintained incrementally by its native + // checkpoint/WAL file rather than materialized as DataBucket pages. + quote! {} + } else if self.attributes.pk_unsized { quote! { pub fn get_peristed_primary_key_with_toc(&self) -> (Vec>>, Vec>>) { let mut pages = vec![]; diff --git a/codegen/src/persist_table/parser.rs b/codegen/src/persist_table/parser.rs index b997c0f2..4a654ff2 100644 --- a/codegen/src/persist_table/parser.rs +++ b/codegen/src/persist_table/parser.rs @@ -30,6 +30,8 @@ impl Parser { pk_unsized: false, read_only: false, pk_upstream: false, + pk_arctic: false, + pk_congee: false, }; for attr in attrs { @@ -47,6 +49,14 @@ impl Parser { res.pk_upstream = true; return Ok(()); } + if meta.path.is_ident("pk_arctic") { + res.pk_arctic = true; + return Ok(()); + } + if meta.path.is_ident("pk_congee") { + res.pk_congee = true; + return Ok(()); + } Ok(()) }) .expect("always ok even on unrecognized attrs"); diff --git a/codegen/src/worktable/mod.rs b/codegen/src/worktable/mod.rs index 30876ef9..3725345f 100644 --- a/codegen/src/worktable/mod.rs +++ b/codegen/src/worktable/mod.rs @@ -56,7 +56,7 @@ pub fn expand(input: TokenStream) -> syn::Result { } fn validate_index_backends(columns: &Columns, persistence: Persistence) -> syn::Result<()> { - let memory_only = if columns.primary_index_backend.is_memory_only() { + let explicit_backend = if columns.primary_index_backend.requires_explicit_persistence() { Some(( columns.primary_index_backend, columns.primary_keys.first().expect("primary key exists"), @@ -66,11 +66,11 @@ fn validate_index_backends(columns: &Columns, persistence: Persistence) -> syn:: columns .indexes .values() - .find(|index| index.backend.is_memory_only()) + .find(|index| index.backend.requires_explicit_persistence()) .map(|index| (index.backend, &index.name, false)) }; - if let Some((backend, ident, is_primary)) = memory_only { + if let Some((backend, ident, is_primary)) = explicit_backend { let kind = if is_primary { "primary index" } else { "index" }; match persistence { Persistence::MemoryOnly => {} @@ -78,20 +78,12 @@ fn validate_index_backends(columns: &Columns, persistence: Persistence) -> syn:: return Err(syn::Error::new( ident.span(), format!( - "{kind} `{ident}` uses `{}`, which requires an explicitly written `persist: false`", - backend.name() - ), - )); - } - Persistence::Persisted => { - return Err(syn::Error::new( - ident.span(), - format!( - "{kind} `{ident}` uses `{}`, but persisted and S3-backed tables require `worktables_index` or `indexset`", + "{kind} `{ident}` uses `{}`, which requires an explicit `persist: true` or `persist: false`", backend.name() ), )); } + Persistence::Persisted => {} } } @@ -197,7 +189,7 @@ mod tests { } #[test] - fn memory_backend_requires_explicit_false() { + fn art_backend_requires_explicit_persistence_choice() { let error = expand(quote! { name: MissingAcknowledgement, columns: { @@ -206,25 +198,28 @@ mod tests { }) .unwrap_err(); - assert!(error.to_string().contains("explicitly written `persist: false`")); + assert!( + error + .to_string() + .contains("explicit `persist: true` or `persist: false`") + ); } #[test] - fn memory_backend_rejects_persistence() { - let error = expand(quote! { + fn art_backend_accepts_persistence() { + let output = expand(quote! { name: PersistentArctic, persist: true, columns: { - id: u64 primary_key, + id: u64 primary_key using arctic, value: u64, }, indexes: { - value_idx: value unique using arctic, + value_idx: value unique using congee, }, - }) - .unwrap_err(); + }); - assert!(error.to_string().contains("persisted and S3-backed tables")); + assert!(output.is_ok()); } #[test] diff --git a/docs/art-index-persistence-plan.md b/docs/art-index-persistence-plan.md new file mode 100644 index 00000000..91bce950 --- /dev/null +++ b/docs/art-index-persistence-plan.md @@ -0,0 +1,107 @@ +# Native ART index persistence + +**Status:** Implemented on `feat/art-native-persistence`; validation in progress. + +**Backends:** `arctic-wt` and `congee-wt`. + +## Decision + +Persisted ART indexes use a backend-native, pointer-free topology checkpoint followed by a logical Set/Remove write-ahead log. They do not serialize WorkTablesIndex pages, and WorkTablesIndex is not retained as an authoritative shadow index. + +The physical checkpoint therefore identifies the selected implementation: + +- an Arctic index stores Arctic compressed edges, exact `Node3`/`Node15`/`Node47`/`Node256` kinds, slot counts, physical branch slots, and logical WorkTable links; +- a Congee index stores Congee prefixes, exact `Node4`/`Node16`/`Node48`/`Node256` kinds, physical child slots, the `Node48` free-list order, and logical WorkTable links. + +Raw pointers, locks, atomics, allocator addresses, epoch/SMR state, and frozen/version bits are never persisted. + +## File layout + +Each existing `*.wt.idx` path contains one ART file when that access path selects an ART backend: + +1. fixed header: magic, format version, backend tag, key width, table version, snapshot length, and snapshot checksum; +2. backend-native topology snapshot; +3. zero or more checksummed logical WAL frames. + +Each WAL frame contains an event id, Set/Remove operation, fixed-width unsigned key, and WorkTable `Link` for Set. A complete frame with a bad checksum is an error. An incomplete final frame is treated as a torn tail. Opening the file truncates that tail before any new append, so later durable records cannot become hidden behind it. + +This is intentionally not the same byte layout as either crate's volatile memory. It is the same *logical physical topology*: node representation, path compression, and physical slot choices survive a clean checkpoint without persisting process-local state. + +## Mutation ordering + +WorkTable's existing persistence pipeline expects gapless, monotonic per-index event ids. A persisted ART therefore uses a persistence-only wrapper around the native index: + +- point reads still delegate directly to the native ART with no selection branch or new read lock; +- mutations take one of 64 key-hashed stripe locks; +- the native mutation commits and its logical event id is allocated before releasing that stripe; +- failed checked inserts allocate no event id. + +The stripe guarantees that non-commuting operations on the same key are replayed in their mutation order. Operations on different keys can receive event ids in a different order from their in-memory linearization, which is safe because those logical operations commute. Hash collisions only reduce write concurrency; they do not weaken correctness. + +Memory-only ART indexes do not use this wrapper and retain the original native hot path. + +## Checkpoint, recovery, and compaction + +Normal persistence appends logical frames and flushes them through WorkTable's existing asynchronous persistence task. It does not keep a second ART in memory. + +When the WAL reaches the compaction threshold: + +1. read and validate the native checkpoint and WAL; +2. reconstruct a temporary instance of the selected ART; +3. replay Set/Remove records; +4. export its exact pointer-free topology; +5. write and sync a temporary checkpoint file; +6. atomically rename it over the previous file; +7. drop the temporary ART. + +Recovery follows the same validation and replay rules but returns the reconstructed ART to the generated table. The reverse primary-key map is rebuilt from the recovered primary ART, as it is for the existing backends. + +Compaction temporarily needs memory proportional to that one index, but normal operation does not retain an authoritative shadow tree. Compaction runs in the disk persistence task rather than on point-read threads. + +## Fork boundary + +The forks add typed topology import/export and no point-operation fields or branches: + +- [`pathscale/arctic-wt`](https://github.com/pathscale/arctic-wt) exposes exclusive pointer-free topology export and exact import; +- [`pathscale/congee-wt`](https://github.com/pathscale/congee-wt) exposes the equivalent contract, including `Node48` free-list order and allocator-failure cleanup. + +WorkTable owns durable framing, checksums, table versions, WAL semantics, compaction, and WorkTable key/link codecs. The forks deliberately do not own a WorkTable-specific disk format. + +## Correctness gates + +The implementation is not ready to merge until all of these remain green: + +- fork topology round trips for every adaptive node kind and delete-created holes/free slots; +- generated Arctic-primary/Congee-secondary and Congee-primary/Arctic-secondary persist/reload/mutate/reload tests; +- checked-insert rollback and acknowledgement without event-id gaps; +- torn final WAL truncation followed by successful new appends; +- checksum, backend, key-width, topology, and table-version rejection; +- compaction from WAL to a native checkpoint and subsequent reload; +- concurrent same-key mutation ordering; +- full workspace tests, formatting, and clippy with warnings denied; +- release-mode performance comparison for memory-only ART versus persisted ART, reported separately. + +## Performance boundary + +The feature is selected only by explicit `persist: true` plus `using arctic` or `using congee`. It introduces no cost to WorkTablesIndex, vanilla IndexSet, or memory-only ART tables. + +Expected persisted-ART costs are: + +- one short striped mutex acquisition per mutation; +- one synthetic logical event allocation in the existing operation object; +- WAL encoding/checksum work in the asynchronous persistence task; +- periodic temporary reconstruction during compaction. + +Reads have no new wrapper lock. Any measurable memory-only regression is a blocker. Persisted-ART write latency, throughput, allocations, p99, WAL growth, recovery time, and compaction pause must be measured before a production recommendation. + +## Remaining production work + +- Validate the existing flush-versus-fsync durability contract against the desired crash guarantee; WAL appends currently match WorkTable's existing flush behavior, while checkpoint replacement calls `sync_data`. +- Run S3 upload/download and interrupted-sync validation. ART files use the existing index paths, but that does not substitute for an end-to-end S3 test. +- Add format migration policy before declaring a stable 1.0 disk contract. +- Measure fork maintenance cost and upstream the generic topology API if maintainers are receptive. +- Keep Congee/Arctic range-scan allocation and isolation limitations explicit; persistence does not change their scan semantics. + +## Paper claim + +The defensible claim is not “serialize an ART.” It is that WorkTable can statically specialize each generated access path while preserving a coherent durability protocol across physically different indexes: structural CDC for B-trees, and native topology checkpoints plus logical redo for lock-free/concurrent ARTs. The same typed table API selects those mechanisms at compile time without runtime backend dispatch. diff --git a/docs/index-backend-dsl-proposal.md b/docs/index-backend-dsl-proposal.md index 13b074ea..72e3f94b 100644 --- a/docs/index-backend-dsl-proposal.md +++ b/docs/index-backend-dsl-proposal.md @@ -1,6 +1,6 @@ # Per-index backends with `using` -**Status:** Experimental implementation in draft PR #187 +**Status:** PR #187 is merged; native ART persistence is implemented on `feat/art-native-persistence` and remains experimental pending validation. **Default:** `worktables_index` @@ -14,8 +14,8 @@ The generated table contains concrete map types. Selection is resolved by the ma This has two distinct uses: -- **Production migration:** persisted tables can run WorkTablesIndex and vanilla IndexSet in parallel and select either per index. Both use the existing WorkTablesIndex/DataBucket disk representation, so changing the provider does not require rebuilding table data. -- **Research and measurement:** explicitly memory-only tables can test Congee or Arctic on access paths where their ART layouts may beat a B-tree. +- **Production migration:** persisted tables can select a backend per index. WorkTablesIndex and vanilla IndexSet share the existing DataBucket page representation; Congee and Arctic use backend-native topology checkpoints plus logical WAL records. +- **Research and measurement:** the same schema can compare memory-only and persisted Congee/Arctic access paths without runtime backend dispatch. The useful paper claim is not that WorkTable bundles several maps. It is that a generated table can statically select a physical implementation per access path, keep a stable typed API, and reject incompatible persistence or key semantics at compile time. @@ -75,12 +75,12 @@ No new persistence keyword is introduced. | Declaration | Meaning | Allowed backends | |---|---|---| -| `persist` omitted | Existing non-persisted table behavior | WorkTablesIndex or vanilla IndexSet; ART use is rejected until `persist: false` is explicit | +| `persist` omitted | Existing non-persisted table behavior | WorkTablesIndex or vanilla IndexSet; ART use requires an explicit persistence choice | | `persist: false` | Explicitly memory-only | All four backends, subject to key and uniqueness constraints | -| `persist: true` | Local durable persistence plus in-memory indexes | WorkTablesIndex and vanilla IndexSet | -| S3 support | Existing S3 sync layered over local persistence | Same restrictions as `persist: true` | +| `persist: true` | Local durable persistence plus in-memory indexes | All four; ART persistence is experimental | +| S3 support | Existing S3 sync layered over local persistence | File paths are compatible; ART end-to-end S3 validation remains required | -Congee and Arctic require the literal `persist: false`. Omitting `persist` is not sufficient acknowledgement. This makes a memory-only physical choice visible during review: +Congee and Arctic require an explicit `persist: true` or `persist: false`; omitting `persist` is not sufficient acknowledgement. This makes the durability choice visible during review: ```rust worktable!( @@ -96,7 +96,7 @@ worktable!( ); ``` -The macro rejects the same schema with `persist: true`, and rejects it when `persist` is omitted. +The macro accepts the same schema with `persist: true` and selects native ART persistence. It rejects the schema when `persist` is omitted. ## Current capability matrix @@ -105,8 +105,8 @@ The macro rejects the same schema with `persist: true`, and rejects it when `per | Primary index | Yes | Yes | Yes | Yes | | Unique secondary index | Yes | Yes | Yes | Yes | | Non-unique secondary index | Yes | No | No | No | -| Persisted local disk | Yes | Yes | No | No | -| Existing S3 persistence path | Yes | Yes | No | No | +| Persisted local disk | Yes | Yes | Experimental | Experimental | +| Existing S3 persistence path | Yes | Yes | Files compatible; validation pending | Files compatible; validation pending | | Variable-sized keys | Yes | Not in this change | No | No | | Ordered point/range API | Yes | Yes | Adapter snapshot for scans | Adapter snapshot for scans | | Default when `using` is absent | Yes | No | No | No | @@ -140,6 +140,10 @@ The selected provider is therefore an in-memory implementation detail, not a new It also separately covers vanilla IndexSet persist → reload → mutate → reload. This is the technical basis for deploying the two providers in parallel without a full data rebuild. +Congee and Arctic deliberately do **not** normalize into WorkTablesIndex pages. Their `*.wt.idx` files contain a checksummed pointer-free checkpoint of the selected ART's physical topology followed by logical Set/Remove WAL frames. Compaction reconstructs a temporary native ART, applies the WAL, and atomically replaces the checkpoint; it does not retain a duplicate authoritative tree during normal operation. See [Native ART index persistence](art-index-persistence-plan.md). + +Because those physical formats differ, switching an existing index between an ART and a B-tree requires an explicit rebuild or migration. WorkTablesIndex ↔ vanilla IndexSet remains the format-compatible provider switch. + This is still a sensitive storage path. Production rollout should retain backups, verify the exact downstream schema/version, and run crash/torn-write and sustained post-reload mutation tests before changing a live table. ## Hot-path and performance details @@ -169,6 +173,7 @@ publication, and reclamation rather than index routing. - Point lookup and mutation call Congee directly. - WorkTable links do not fit in Congee's one-word payload. The adapter stores an `Arc` pointer, so inserts allocate and reads clone the `Arc` before copying the link. - Ordered reads use Congee's native range scan and materialize only the requested key interval into a `Vec`; a full iteration is therefore O(n) with one result allocation, while a narrow range no longer dumps, re-probes, and sorts the whole tree. +- With `persist: true`, mutations additionally take a key-striped sequencing lock before producing one logical WAL event. Memory-only Congee does not pay that cost. ### Arctic @@ -176,10 +181,11 @@ publication, and reclamation rather than index routing. - WorkTable links are stored in `Box` values because Arctic's inline value is limited to 64 bits. Inserts allocate; reads copy the link from the box. - Ordered reads use Arctic's native bounded traversal and materialize the requested interval into a `Vec`. - Concurrent scan behavior inherits Arctic's non-linearizable traversal contract. +- With `persist: true`, mutations use the same persistence-only sequencing wrapper as Congee. Point reads remain direct and lock-free. Generated table traversal snapshots its ordered link list once instead of restarting a range at every row. The ART adapters are still candidates for -point-heavy, explicitly memory-only paths—not automatic wins for `select_all`, +point-heavy paths—not automatic wins for `select_all`, wide ranges, iteration, or vacuum, because ordered results are materialized rather than streamed. Measurements must separate point lookup, write/allocation cost, range width, full iteration, and reclamation rather than @@ -197,26 +203,28 @@ Use allocator/RSS measurements for comparative memory results; do not treat the ## Dependency and fork status -This implementation uses the published crates directly: +This implementation pins two narrow forks for typed topology import/export: - `WorkTablesIndex 0.0.4` as the default `indexset` dependency alias already used by WorkTable; - vanilla `indexset 0.15.0` under the `vanilla_indexset` Cargo name; -- `congee 0.4.1` through its public `Congee`, `compute_or_insert`, and `new_with_drainer` APIs; -- `arctic-map 0.1.4` through its public concurrent map API. +- `congee-wt` at commit `005bfb1968e781800176f2d7e465e6a1af630e1a`; +- `arctic-wt` at commit `e13fc7df3c040f14ae66c1cb56b1bd0a3f6da3fc`. -There is no Congee fork, Cargo patch, or direct `CongeeArc` dependency in this PR. The WorkTable adapter supplies the checked-insert and pointer-lifetime behavior it needs using vanilla Congee's public API. +The forks add no point-operation fields or branches. They expose pointer-free topology values and exact reconstruction; WorkTable owns checksums, framing, WAL, recovery, and compaction. ## Correctness coverage in this PR The implementation includes: - parser/default tests for all four names; -- compile-time rejection of persisted ARTs, implicit memory-only ARTs, alternative non-unique indexes, and unsupported key shapes; +- compile-time acceptance of explicit persisted ARTs plus rejection of implicit ART persistence choices, alternative non-unique indexes, and unsupported key shapes; - shared unique-index contract and concurrent mutation-integrity tests for all four providers; - adapter contract and concurrent checked-insert tests for Congee and Arctic; - immediate disjoint insert/read/remove tests for Congee and Arctic; - a generated table using all four providers simultaneously; - generated primary-key CRUD tests for vanilla IndexSet, Congee, and Arctic; +- native Arctic-primary/Congee-secondary and Congee-primary/Arctic-secondary persist/reload/mutate/reload coverage; +- native topology codec, WAL, torn-tail truncation, and compaction tests; - vanilla IndexSet persist/reload/post-reload mutation coverage; - WorkTablesIndex → vanilla IndexSet → WorkTablesIndex disk-provider switching coverage. @@ -234,8 +242,8 @@ Minimum useful ARM campaign: 4. point-read, insert, delete, and production mixed traces measured separately; 5. range widths 1, 8, 64, and 1,024 plus full iteration; 6. p50, p99, throughput, allocations/op, RSS/entry, and post-churn reclamation; -7. persisted WorkTablesIndex versus persisted vanilla IndexSet, including reload and writes after reload; -8. memory-only Congee and Arctic only where `persist: false` is operationally valid. +7. persisted WorkTablesIndex, vanilla IndexSet, Congee, and Arctic, including reload and writes after reload; +8. memory-only versus persisted ART to isolate the sequencing/WAL overhead. Run release builds on the actual ARM deployment class. SIMD should not be treated as a reason to prefer a backend; any x86 result is a portability check, not the primary HFT decision. @@ -245,6 +253,6 @@ For the paper, the strongest controlled experiment keeps the WorkTable schema, g - **WorkTablesIndex:** production default. - **Vanilla IndexSet:** experimental provider. It preserves local/S3 persistence through the existing format boundary, but is excluded from concurrent correctness and published performance claims until upstream offers a stable structural-read primitive or the adapter gains a low-cost algorithm. -- **Congee and Arctic:** research/experimental memory-only backends in this PR. Promotion requires relevant downstream evidence, allocation/reclamation review, and a workload that does not depend on the current allocating scan path. +- **Congee and Arctic:** research/experimental backends with native local persistence. Promotion requires crash/S3 validation, relevant downstream evidence, allocation/reclamation review, and a workload that does not depend on the current allocating scan path. That boundary is deliberate: `using` exposes optional physical specialization without quietly weakening WorkTable's in-memory/on-disk coordination contract. diff --git a/src/index/arctic.rs b/src/index/arctic.rs index 6f1399df..3f124d1f 100644 --- a/src/index/arctic.rs +++ b/src/index/arctic.rs @@ -94,6 +94,32 @@ where } } +impl ArcticIndex +where + K: ArcticKey, + K::Raw: arctic::topology::Key, + V: Clone + Debug + Send + Sync + 'static, +{ + pub(crate) fn export_topology( + &mut self, + mut encode: impl FnMut(&V) -> T, + ) -> Result, arctic::topology::Error> { + self.inner.export_topology(|value| encode(value)) + } + + pub(crate) fn from_topology( + topology: arctic::topology::Topology, + mut decode: impl FnMut(T) -> V, + ) -> Result { + let inner = ConcurrentMap::from_topology(topology, |value| Box::new(decode(value)))?; + let len = inner.all().entries(Order::Ascend).count(); + Ok(Self { + inner, + len: AtomicUsize::new(len), + }) + } +} + impl UniqueIndex for ArcticIndex where K: ArcticKey, diff --git a/src/index/congee.rs b/src/index/congee.rs index 62c9887e..f2e41846 100644 --- a/src/index/congee.rs +++ b/src/index/congee.rs @@ -5,7 +5,7 @@ use std::ops::{Bound, RangeBounds}; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; -use congee::{Congee, DefaultAllocator}; +use congee::{CongeeRaw, DefaultAllocator}; use super::UniqueIndex; @@ -46,7 +46,7 @@ impl_congee_key!(u64); /// machine word while a WorkTable `Link` is wider. The adapter follows the /// reclamation pattern used by Congee's own `CongeeArc` implementation. pub struct CongeeIndex { - inner: Congee, + inner: CongeeRaw, len: AtomicUsize, marker: std::marker::PhantomData<(K, V)>, } @@ -71,7 +71,7 @@ where drop(unsafe { Self::arc_from_pointer(pointer) }); }; Self { - inner: Congee::new_with_drainer(DefaultAllocator {}, drainer), + inner: CongeeRaw::new_with_drainer(DefaultAllocator {}, drainer), len: AtomicUsize::new(0), marker: std::marker::PhantomData, } @@ -160,6 +160,40 @@ where }) .collect() } + + pub(crate) fn export_topology( + &mut self, + mut encode: impl FnMut(&V) -> T, + ) -> Result, congee::topology::Error> { + self.inner.export_topology(|pointer| { + // SAFETY: every raw payload is a live tree-owned `Arc` pointer, + // and the exclusive borrow prevents removal while it is cloned. + unsafe { encode(&*std::ptr::with_exposed_provenance::(pointer)) } + }) + } + + pub(crate) fn from_topology( + topology: congee::topology::Topology, + mut decode: impl FnMut(T) -> V, + ) -> Result { + let drainer = |_key: usize, pointer: usize| { + // SAFETY: decoded payloads below transfer exactly one `Arc` strong + // reference to the reconstructed tree. + drop(unsafe { Self::arc_from_pointer(pointer) }); + }; + let inner = CongeeRaw::from_topology_with_drainer( + topology, + DefaultAllocator {}, + |value| Arc::into_raw(Arc::new(decode(value))).expose_provenance(), + drainer, + )?; + let len = inner.keys().len(); + Ok(Self { + inner, + len: AtomicUsize::new(len), + marker: std::marker::PhantomData, + }) + } } impl UniqueIndex for CongeeIndex diff --git a/src/index/mod.rs b/src/index/mod.rs index 196f05b4..b6c4351f 100644 --- a/src/index/mod.rs +++ b/src/index/mod.rs @@ -2,6 +2,7 @@ mod arctic; mod available_index; mod congee; mod multipair; +mod persistent_art; mod primary_index; mod table_index; mod table_secondary_index; @@ -14,6 +15,7 @@ pub use congee::{CongeeIndex, CongeeKey}; 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 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_art.rs b/src/index/persistent_art.rs new file mode 100644 index 00000000..34223faa --- /dev/null +++ b/src/index/persistent_art.rs @@ -0,0 +1,273 @@ +//! Persistence-only synchronization and logical change capture for ART indexes. +//! +//! Memory-only Congee and Arctic indexes continue to use their native hot +//! paths. Persisted ART indexes use this wrapper so same-key mutations are +//! sequenced in the same order as their durable logical events. Mutations on +//! different stripes remain concurrent because their Set/Remove records +//! commute during recovery. + +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::atomic::{AtomicU64, Ordering}; + +use data_bucket::Link; +use indexset::cdc::change::{ChangeEvent, Id}; +use indexset::core::pair::Pair; +use parking_lot::Mutex; + +use crate::index::{ArcticIndex, CongeeIndex, UniqueIndex}; +use crate::util::OffsetEqLink; +use crate::{ArcticKey, CongeeKey, TableIndexCdc}; + +const MUTATION_STRIPES: usize = 64; + +/// Adds persistence sequencing to a native ART without changing its point-read +/// path or the layout of the underlying ART. +pub struct PersistentArtIndex { + inner: I, + next_event_id: AtomicU64, + mutation_stripes: [Mutex<()>; MUTATION_STRIPES], +} + +/// Persisted Arctic index selected by the generated DSL. +pub type PersistentArcticIndex = PersistentArtIndex>; + +/// Persisted Congee index selected by the generated DSL. +pub type PersistentCongeeIndex = PersistentArtIndex>; + +impl Debug for PersistentArtIndex { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("PersistentArtIndex") + .field("inner", &self.inner) + .field("next_event_id", &self.next_event_id.load(Ordering::Relaxed)) + .finish_non_exhaustive() + } +} + +impl Default for PersistentArtIndex { + fn default() -> Self { + Self::from_inner(I::default()) + } +} + +impl PersistentArtIndex { + /// Wraps a reconstructed ART. Event ids restart at zero because startup + /// recovery checkpoints and clears the preceding WAL before accepting new + /// operations. + pub fn from_inner(inner: I) -> Self { + Self { + inner, + next_event_id: AtomicU64::new(0), + mutation_stripes: array::from_fn(|_| Mutex::new(())), + } + } + + /// Returns the native ART, primarily for quiescent checkpoint encoding. + pub fn into_inner(self) -> I { + self.inner + } + + /// Exclusively borrows the native ART. + pub fn inner_mut(&mut self) -> &mut I { + &mut self.inner + } + + /// Borrows the native ART. + pub fn inner(&self) -> &I { + &self.inner + } + + fn mutation_stripe(&self, key: &K) -> &Mutex<()> { + let mut hasher = DefaultHasher::new(); + key.hash(&mut hasher); + &self.mutation_stripes[hasher.finish() as usize % MUTATION_STRIPES] + } + + fn next_event_id(&self) -> Id { + self.next_event_id.fetch_add(1, Ordering::AcqRel).into() + } +} + +impl UniqueIndex for PersistentArtIndex +where + K: Clone + Ord + Send + 'static, + V: Clone + Send + 'static, + I: UniqueIndex, +{ + #[inline] + fn get_value(&self, key: &K) -> Option { + self.inner.get_value(key) + } + + #[inline] + fn with_value(&self, key: &K, read: impl FnOnce(&V) -> R) -> Option { + self.inner.with_value(key, read) + } + + #[inline] + 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_value(key, value) + } + + #[inline] + fn insert_value_checked(&self, key: K, value: V) -> Option<()> { + self.inner.insert_value_checked(key, value) + } + + #[inline] + fn remove_value(&self, key: &K) -> Option<(K, V)> { + self.inner.remove_value(key) + } + + #[inline] + fn len(&self) -> usize { + self.inner.len() + } + + fn iter_values(&self) -> impl DoubleEndedIterator + '_ { + self.inner.iter_values() + } + + fn iter_links(&self) -> impl DoubleEndedIterator + '_ { + self.inner.iter_links() + } + + fn range_values<'a, R>(&'a self, range: R) -> impl DoubleEndedIterator + 'a + where + R: RangeBounds + 'a, + { + self.inner.range_values(range) + } + + fn range_links<'a, R>(&'a self, range: R) -> impl DoubleEndedIterator + 'a + where + R: RangeBounds + 'a, + { + self.inner.range_links(range) + } +} + +macro_rules! impl_persisted_art_cdc { + ($index:ident, $key_bound:path) => { + impl TableIndexCdc for PersistentArtIndex<$index>> + where + T: $key_bound + Eq + Hash, + { + fn insert_cdc(&self, value: T, link: Link) -> (Option, Vec>>) { + let _sequence_guard = self.mutation_stripe(&value).lock(); + let old = self + .inner + .insert_value(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).lock(); + self.inner + .insert_value_checked(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).lock(); + let Some((key, old)) = self.inner.remove_value(&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]) + } + } + }; +} + +impl_persisted_art_cdc!(ArcticIndex, ArcticKey); +impl_persisted_art_cdc!(CongeeIndex, CongeeKey); + +#[cfg(test)] +mod tests { + use std::sync::{Arc, Barrier}; + + use super::*; + + #[test] + fn failed_checked_insert_does_not_consume_an_event_id() { + let index = PersistentArcticIndex::>::default(); + let link = Link::default(); + assert!(index.insert_checked_cdc(7, link).is_some()); + assert!(index.insert_checked_cdc(7, link).is_none()); + assert_eq!(index.remove_cdc(7, link).1[0].id(), 1.into()); + } + + #[test] + fn same_key_events_follow_mutation_order() { + let index = Arc::new(PersistentArcticIndex::>::default()); + let barrier = Arc::new(Barrier::new(9)); + let mut threads = Vec::new(); + for offset in 0..8 { + let index = Arc::clone(&index); + let barrier = Arc::clone(&barrier); + threads.push(std::thread::spawn(move || { + barrier.wait(); + index + .insert_cdc( + 11, + Link { + offset, + ..Link::default() + }, + ) + .1[0] + .clone() + })); + } + barrier.wait(); + let mut events = threads + .into_iter() + .map(|thread| thread.join().unwrap()) + .collect::>(); + events.sort_by_key(ChangeEvent::id); + let last_link = match events.last().unwrap() { + ChangeEvent::InsertAt { value, .. } => value.value, + _ => unreachable!(), + }; + assert_eq!(index.get_value(&11).unwrap().0, last_link); + } +} diff --git a/src/index/table_index/mod.rs b/src/index/table_index/mod.rs index 9eb3543e..75c8b71b 100644 --- a/src/index/table_index/mod.rs +++ b/src/index/table_index/mod.rs @@ -9,7 +9,10 @@ use vanilla_indexset::core::node::NodeLike as VanillaNodeLike; use vanilla_indexset::core::pair::Pair as VanillaPair; use crate::util::OffsetEqLink; -use crate::{ArcticIndex, ArcticKey, CongeeIndex, CongeeKey, IndexMap, IndexMultiMap, UniqueIndex, UpstreamIndexMap}; +use crate::{ + ArcticIndex, ArcticKey, CongeeIndex, CongeeKey, IndexMap, IndexMultiMap, PersistentArcticIndex, + PersistentCongeeIndex, UniqueIndex, UpstreamIndexMap, +}; mod cdc; pub mod util; @@ -133,3 +136,9 @@ impl_unique_table_index!(CongeeIndex, [CongeeKey + Eq + Hash]); impl_unique_table_index!(ArcticIndex, [ ArcticKey + Eq + Hash ]); +impl_unique_table_index!(PersistentCongeeIndex, [ + CongeeKey + Eq + Hash +]); +impl_unique_table_index!(PersistentArcticIndex, [ + ArcticKey + Eq + Hash +]); diff --git a/src/lib.rs b/src/lib.rs index bdc87ca1..59f3f289 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -32,11 +32,12 @@ pub mod prelude { pub use crate::lock::{LockGuard, LockMap}; pub use crate::mem_stat::MemStat; pub use crate::persistence::{ - AcknowledgeOperation, DeleteOperation, DiskConfig, DiskPersistenceEngine, IndexTableOfContents, - InsertOperation, Operation, OperationId, PersistedWorkTable, PersistenceConfig, PersistenceEngine, - PersistenceTask, ReadOnlyPersistenceEngine, SpaceData, SpaceDataOps, SpaceIndex, SpaceIndexOps, - SpaceIndexUnsized, SpaceSecondaryIndexOps, UpdateOperation, map_index_pages_to_toc_and_general, - map_unsized_index_pages_to_toc_and_general, reconstruct_multi_index_nodes, validate_events, + AcknowledgeOperation, ArtPersistenceKey, DeleteOperation, DiskConfig, DiskPersistenceEngine, + IndexTableOfContents, InsertOperation, Operation, OperationId, PersistedWorkTable, PersistenceConfig, + PersistenceEngine, PersistenceTask, ReadOnlyPersistenceEngine, SpaceArcticIndex, SpaceCongeeIndex, SpaceData, + SpaceDataOps, SpaceIndex, SpaceIndexOps, SpaceIndexUnsized, SpaceSecondaryIndexOps, UpdateOperation, + 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}; @@ -44,10 +45,11 @@ pub mod prelude { pub use crate::util::{OffsetEqLink, OrderedF32Def, OrderedF64Def}; pub use crate::{ ArcticIndex, ArcticKey, AvailableIndex, CongeeIndex, CongeeKey, Difference, IndexError, IndexMap, - IndexMultiMap, MultiPairRecreate, PrimaryIndex, TableIndex, TableIndexCdc, TableRow, TableSecondaryIndex, - TableSecondaryIndexCdc, TableSecondaryIndexEventsOps, TableSecondaryIndexInfo, UniqueIndex, UnsizedNode, - UpstreamIndexMap, UpstreamIndexPair, WorkTable, WorkTableError, vacuum::EmptyDataVacuum, - vacuum::VacuumPersistence, vacuum::WorkTableVacuum, + 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, }; 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 f30d2e36..a933b4d6 100644 --- a/src/mem_stat/mod.rs +++ b/src/mem_stat/mod.rs @@ -20,7 +20,9 @@ use vanilla_indexset::core::pair::Pair as VanillaPair; use crate::persistence::OperationType; use crate::prelude::OperationId; use crate::util::OffsetEqLink; -use crate::{ArcticIndex, ArcticKey, CongeeIndex, CongeeKey, IndexMultiMap, UniqueIndex, UpstreamIndexMap}; +use crate::{ + ArcticIndex, ArcticKey, CongeeIndex, CongeeKey, IndexMultiMap, PersistentArtIndex, UniqueIndex, UpstreamIndexMap, +}; use crate::{IndexMap, impl_memstat_zero}; pub trait MemStat { @@ -129,6 +131,16 @@ where } } +impl MemStat for PersistentArtIndex { + 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/mod.rs b/src/persistence/mod.rs index 1eed9c04..681fbd89 100644 --- a/src/persistence/mod.rs +++ b/src/persistence/mod.rs @@ -10,9 +10,9 @@ pub use operation::{ }; pub use readonly_engine::ReadOnlyPersistenceEngine; pub use space::{ - IndexTableOfContents, 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, + 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, }; pub use task::PersistenceTask; diff --git a/src/persistence/space/art_index.rs b/src/persistence/space/art_index.rs new file mode 100644 index 00000000..96ad5c00 --- /dev/null +++ b/src/persistence/space/art_index.rs @@ -0,0 +1,975 @@ +//! Backend-native ART checkpoints with a logical Set/Remove write-ahead log. +//! +//! Checkpoints preserve the selected backend's pointer-free physical topology. +//! The WAL is logical because raw pointers, locks, SMR state, and allocator +//! addresses are process-local. Compaction reconstructs a temporary native ART, +//! applies the WAL, writes a new native checkpoint atomically, and drops the +//! temporary tree; no duplicate ART is retained during normal operation. + +use std::fmt::Debug; +use std::hash::Hash; +use std::marker::PhantomData; +use std::path::{Path, PathBuf}; + +use data_bucket::{Link, page::PageId}; +use eyre::{Context, bail, eyre}; +use indexset::cdc::change::ChangeEvent; +use indexset::core::pair::Pair; +use tokio::fs::{File, OpenOptions}; +use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt}; + +use crate::index::{ + ArcticIndex, ArcticKey, CongeeIndex, CongeeKey, PersistentArcticIndex, PersistentArtIndex, PersistentCongeeIndex, + UniqueIndex, +}; +use crate::persistence::SpaceIndexOps; +use crate::persistence::space::BatchChangeEvent; +use crate::prelude::WT_INDEX_EXTENSION; +use crate::util::OffsetEqLink; + +const FILE_MAGIC: &[u8; 8] = b"WTART001"; +const WAL_MAGIC: &[u8; 4] = b"WAL1"; +const FORMAT_VERSION: u16 = 1; +const HEADER_LEN: usize = 32; +const WAL_HEADER_LEN: usize = 12; +const COMPACT_WAL_BYTES: u64 = 4 * 1024 * 1024; + +#[derive(Copy, Clone, Debug, Eq, PartialEq)] +enum Backend { + Arctic = 1, + Congee = 2, +} + +impl Backend { + fn from_byte(byte: u8) -> eyre::Result { + match byte { + 1 => Ok(Self::Arctic), + 2 => Ok(Self::Congee), + _ => bail!("unknown ART backend tag {byte}"), + } + } +} + +/// Stable fixed-width codec used by logical ART WAL records. +/// +/// Generated single-column primary-key newtypes delegate this contract to +/// their supported unsigned integer field. +pub trait ArtPersistenceKey: Clone + Debug + Eq + Hash + Ord + Send + Sync + 'static { + /// Number of key bytes written to the WAL. + const WIDTH: u8; + + /// Appends exactly [`Self::WIDTH`] bytes in big-endian order. + fn encode_art_key(&self, output: &mut Vec); + + /// Decodes exactly [`Self::WIDTH`] bytes. + fn decode_art_key(bytes: &[u8]) -> eyre::Result; +} + +macro_rules! impl_art_persistence_key { + ($($type:ty),+ $(,)?) => { + $( + impl ArtPersistenceKey for $type { + const WIDTH: u8 = std::mem::size_of::() as u8; + + fn encode_art_key(&self, output: &mut Vec) { + output.extend_from_slice(&self.to_be_bytes()); + } + + fn decode_art_key(bytes: &[u8]) -> eyre::Result { + let bytes: [u8; std::mem::size_of::()] = bytes + .try_into() + .map_err(|_| eyre!("invalid {}-byte ART key", Self::WIDTH))?; + Ok(Self::from_be_bytes(bytes)) + } + } + )+ + }; +} + +impl_art_persistence_key!(u8, u16, u32, u64, u128, usize); + +#[derive(Clone, Debug, Eq, PartialEq)] +enum WalOp { + Set(Link), + Remove, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +struct WalRecord { + event_id: u64, + key: K, + op: WalOp, +} + +#[derive(Debug)] +struct Image { + snapshot: Vec, + wal: Vec>, + wal_bytes: u64, + durable_len: u64, +} + +#[derive(Debug)] +struct ArtFile { + path: PathBuf, + file: File, + backend: Backend, + table_version: u32, + wal_bytes: u64, + marker: PhantomData, +} + +impl ArtFile { + async fn open(path: PathBuf, backend: Backend, table_version: u32, empty_snapshot: Vec) -> eyre::Result { + if !path.exists() { + Self::write_new_file(&path, backend, table_version, &empty_snapshot).await?; + } + let image = Self::read_image(&path, backend, table_version).await?; + let mut file = OpenOptions::new().read(true).write(true).open(&path).await?; + // Remove an incomplete final frame before appending. Leaving it in + // place would make every later valid frame unreachable on recovery. + file.set_len(image.durable_len).await?; + file.seek(std::io::SeekFrom::End(0)).await?; + Ok(Self { + path, + file, + backend, + table_version, + wal_bytes: image.wal_bytes, + marker: PhantomData, + }) + } + + async fn read_image(path: &Path, backend: Backend, table_version: u32) -> eyre::Result> { + let mut file = File::open(path) + .await + .wrap_err_with(|| format!("open ART index {}", path.display()))?; + let mut bytes = Vec::new(); + file.read_to_end(&mut bytes).await?; + if bytes.len() < HEADER_LEN { + bail!("ART index {} has a truncated header", path.display()); + } + if &bytes[..8] != FILE_MAGIC { + bail!("ART index {} has an invalid magic", path.display()); + } + let format = u16::from_le_bytes(bytes[8..10].try_into().unwrap()); + if format != FORMAT_VERSION { + bail!("unsupported ART index format {format}"); + } + let found_backend = Backend::from_byte(bytes[10])?; + if found_backend != backend { + bail!("ART index backend mismatch: expected {backend:?}, found {found_backend:?}"); + } + if bytes[11] != K::WIDTH { + bail!("ART key width mismatch: expected {}, found {}", K::WIDTH, bytes[11]); + } + let found_table_version = u32::from_le_bytes(bytes[12..16].try_into().unwrap()); + if found_table_version != table_version { + bail!("ART index table version mismatch: expected {table_version}, found {found_table_version}"); + } + let snapshot_len = u64::from_le_bytes(bytes[16..24].try_into().unwrap()); + let snapshot_len = usize::try_from(snapshot_len).map_err(|_| eyre!("ART snapshot is too large"))?; + let snapshot_crc = u32::from_le_bytes(bytes[24..28].try_into().unwrap()); + let snapshot_end = HEADER_LEN + .checked_add(snapshot_len) + .ok_or_else(|| eyre!("ART snapshot length overflow"))?; + if snapshot_end > bytes.len() { + bail!("ART index {} has a truncated snapshot", path.display()); + } + let snapshot = bytes[HEADER_LEN..snapshot_end].to_vec(); + if crc32fast::hash(&snapshot) != snapshot_crc { + bail!("ART index {} snapshot checksum mismatch", path.display()); + } + + let mut wal = Vec::new(); + let mut position = snapshot_end; + let mut durable_end = snapshot_end; + while position < bytes.len() { + // A crash can leave the final frame header or payload incomplete. + // Only an incomplete final frame is ignored; complete corruption + // remains a hard error. + if bytes.len() - position < WAL_HEADER_LEN { + break; + } + if &bytes[position..position + 4] != WAL_MAGIC { + bail!("ART WAL frame at byte {position} has an invalid magic"); + } + let payload_len = u32::from_le_bytes(bytes[position + 4..position + 8].try_into().unwrap()) as usize; + let expected_len = 9usize + K::WIDTH as usize + 12; + if payload_len != expected_len { + bail!("ART WAL frame at byte {position} has invalid payload length {payload_len}"); + } + let payload_crc = u32::from_le_bytes(bytes[position + 8..position + 12].try_into().unwrap()); + let payload_start = position + WAL_HEADER_LEN; + let payload_end = payload_start + .checked_add(payload_len) + .ok_or_else(|| eyre!("ART WAL frame length overflow"))?; + if payload_end > bytes.len() { + break; + } + let payload = &bytes[payload_start..payload_end]; + if crc32fast::hash(payload) != payload_crc { + bail!("ART WAL frame at byte {position} checksum mismatch"); + } + wal.push(decode_wal_record::(payload)?); + position = payload_end; + durable_end = payload_end; + } + + Ok(Image { + snapshot, + wal, + wal_bytes: (durable_end - snapshot_end) as u64, + durable_len: durable_end as u64, + }) + } + + async fn append(&mut self, records: &[WalRecord]) -> eyre::Result<()> { + if records.is_empty() { + return Ok(()); + } + let mut bytes = Vec::with_capacity(records.len() * (WAL_HEADER_LEN + 32)); + for record in records { + let payload = encode_wal_record(record); + bytes.extend_from_slice(WAL_MAGIC); + bytes.extend_from_slice(&(payload.len() as u32).to_le_bytes()); + bytes.extend_from_slice(&crc32fast::hash(&payload).to_le_bytes()); + bytes.extend_from_slice(&payload); + } + self.file.write_all(&bytes).await?; + self.file.flush().await?; + self.wal_bytes += bytes.len() as u64; + Ok(()) + } + + fn should_compact(&self) -> bool { + self.wal_bytes >= COMPACT_WAL_BYTES + } + + async fn rewrite(&mut self, snapshot: &[u8]) -> eyre::Result<()> { + let temporary = self.path.with_extension("wt.idx.art.tmp"); + Self::write_new_file(&temporary, self.backend, self.table_version, snapshot).await?; + tokio::fs::rename(&temporary, &self.path).await?; + self.file = OpenOptions::new().read(true).write(true).open(&self.path).await?; + self.file.seek(std::io::SeekFrom::End(0)).await?; + self.wal_bytes = 0; + Ok(()) + } + + async fn write_new_file(path: &Path, backend: Backend, table_version: u32, snapshot: &[u8]) -> eyre::Result<()> { + let mut header = Vec::with_capacity(HEADER_LEN); + header.extend_from_slice(FILE_MAGIC); + header.extend_from_slice(&FORMAT_VERSION.to_le_bytes()); + header.push(backend as u8); + header.push(K::WIDTH); + header.extend_from_slice(&table_version.to_le_bytes()); + header.extend_from_slice(&(snapshot.len() as u64).to_le_bytes()); + header.extend_from_slice(&crc32fast::hash(snapshot).to_le_bytes()); + header.extend_from_slice(&0u32.to_le_bytes()); + debug_assert_eq!(header.len(), HEADER_LEN); + + let mut file = File::create(path).await?; + file.write_all(&header).await?; + file.write_all(snapshot).await?; + file.flush().await?; + file.sync_data().await?; + Ok(()) + } +} + +fn encode_wal_record(record: &WalRecord) -> Vec { + let mut bytes = Vec::with_capacity(9 + K::WIDTH as usize + 12); + bytes.extend_from_slice(&record.event_id.to_le_bytes()); + match record.op { + WalOp::Set(_) => bytes.push(1), + WalOp::Remove => bytes.push(2), + } + record.key.encode_art_key(&mut bytes); + let link = match record.op { + WalOp::Set(link) => link, + WalOp::Remove => Link::default(), + }; + let page_id: usize = link.page_id.into(); + bytes.extend_from_slice(&(page_id as u32).to_le_bytes()); + bytes.extend_from_slice(&link.offset.to_le_bytes()); + bytes.extend_from_slice(&link.length.to_le_bytes()); + bytes +} + +fn decode_wal_record(bytes: &[u8]) -> eyre::Result> { + let expected_len = 9 + K::WIDTH as usize + 12; + if bytes.len() != expected_len { + bail!("invalid ART WAL payload length {}", bytes.len()); + } + let event_id = u64::from_le_bytes(bytes[..8].try_into().unwrap()); + let operation = bytes[8]; + let key_end = 9 + K::WIDTH as usize; + let key = K::decode_art_key(&bytes[9..key_end])?; + let page_id = u32::from_le_bytes(bytes[key_end..key_end + 4].try_into().unwrap()); + let offset = u32::from_le_bytes(bytes[key_end + 4..key_end + 8].try_into().unwrap()); + let length = u32::from_le_bytes(bytes[key_end + 8..key_end + 12].try_into().unwrap()); + let op = match operation { + 1 => WalOp::Set(Link { + page_id: PageId::from(page_id), + offset, + length, + }), + 2 => WalOp::Remove, + _ => bail!("invalid ART WAL operation {operation}"), + }; + Ok(WalRecord { event_id, key, op }) +} + +fn logical_record(event: ChangeEvent>) -> eyre::Result> { + match event { + ChangeEvent::InsertAt { + event_id, + max_value, + value, + index, + } if index == 0 && max_value == value => Ok(WalRecord { + event_id: event_id.inner(), + key: value.key, + op: WalOp::Set(value.value), + }), + ChangeEvent::RemoveAt { + event_id, + max_value, + value, + index, + } if index == 0 && max_value == value => Ok(WalRecord { + event_id: event_id.inner(), + key: value.key, + op: WalOp::Remove, + }), + _ => bail!("native ART persistence received a structural WorkTablesIndex event"), + } +} + +fn apply_wal(index: &I, wal: &[WalRecord], wrap: impl Fn(Link) -> V) +where + K: ArtPersistenceKey, + V: Clone + Send + 'static, + I: UniqueIndex, +{ + for record in wal { + match record.op { + WalOp::Set(link) => { + index.insert_value(record.key.clone(), wrap(link)); + } + WalOp::Remove => { + index.remove_value(&record.key); + } + } + } +} + +/// Disk-side Arctic checkpoint and WAL state. +#[derive(Debug)] +pub struct SpaceArcticIndex { + file: ArtFile, +} + +impl SpaceArcticIndex +where + K: ArtPersistenceKey + ArcticKey, + K::Raw: arctic::topology::Key, +{ + async fn new(path: PathBuf, table_version: u32) -> eyre::Result { + let mut empty = ArcticIndex::::default(); + let snapshot = encode_arctic_topology(&empty.export_topology(|link| *link)?)?; + Ok(Self { + file: ArtFile::open(path, Backend::Arctic, table_version, snapshot).await?, + }) + } + + /// Reconstructs a persisted Arctic index from its native checkpoint and WAL. + pub async fn load_index( + path: impl AsRef, + table_version: u32, + ) -> eyre::Result>> { + let image = ArtFile::::read_image(path.as_ref(), Backend::Arctic, table_version).await?; + let topology = decode_arctic_topology(&image.snapshot)?; + let index = ArcticIndex::from_topology(topology, OffsetEqLink)?; + apply_wal(&index, &image.wal, OffsetEqLink); + Ok(PersistentArtIndex::from_inner(index)) + } + + /// Writes a complete native Arctic checkpoint with an empty WAL. + pub async fn write_checkpoint( + path: impl AsRef, + table_version: u32, + index: &mut PersistentArcticIndex>, + ) -> eyre::Result<()> { + let topology = index.inner_mut().export_topology(|link| link.0)?; + let snapshot = encode_arctic_topology(&topology)?; + ArtFile::::write_new_file(path.as_ref(), Backend::Arctic, table_version, &snapshot).await + } + + async fn compact(&mut self) -> eyre::Result<()> { + let image = ArtFile::::read_image(&self.file.path, Backend::Arctic, self.file.table_version).await?; + let topology = decode_arctic_topology(&image.snapshot)?; + let mut index = ArcticIndex::from_topology(topology, |link| link)?; + apply_wal(&index, &image.wal, |link| link); + let snapshot = encode_arctic_topology(&index.export_topology(|link| *link)?)?; + self.file.rewrite(&snapshot).await + } +} + +impl SpaceIndexOps for SpaceArcticIndex +where + K: ArtPersistenceKey + ArcticKey, + K::Raw: arctic::topology::Key, +{ + async fn primary_from_table_files_path + Send>(path: S, version: u32) -> eyre::Result { + Self::new( + PathBuf::from(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( + PathBuf::from(format!("{}/{}{}", path.as_ref(), name.as_ref(), WT_INDEX_EXTENSION)), + version, + ) + .await + } + + async fn bootstrap(_: &mut File, _: String, _: u32) -> eyre::Result<()> { + Ok(()) + } + + async fn process_change_event(&mut self, event: ChangeEvent>) -> eyre::Result<()> { + self.file.append(&[logical_record(event)?]).await?; + if self.file.should_compact() { + self.compact().await?; + } + Ok(()) + } + + async fn process_change_event_batch(&mut self, events: BatchChangeEvent) -> eyre::Result<()> { + let records = events + .into_iter() + .map(logical_record) + .collect::>>()?; + self.file.append(&records).await?; + if self.file.should_compact() { + self.compact().await?; + } + Ok(()) + } +} + +/// Disk-side Congee checkpoint and WAL state. +#[derive(Debug)] +pub struct SpaceCongeeIndex { + file: ArtFile, +} + +impl SpaceCongeeIndex +where + K: ArtPersistenceKey + CongeeKey, +{ + async fn new(path: PathBuf, table_version: u32) -> eyre::Result { + let mut empty = CongeeIndex::::default(); + let snapshot = encode_congee_topology(&empty.export_topology(|link| *link)?)?; + Ok(Self { + file: ArtFile::open(path, Backend::Congee, table_version, snapshot).await?, + }) + } + + /// Reconstructs a persisted Congee index from its native checkpoint and WAL. + pub async fn load_index( + path: impl AsRef, + table_version: u32, + ) -> eyre::Result>> { + let image = ArtFile::::read_image(path.as_ref(), Backend::Congee, table_version).await?; + let topology = decode_congee_topology(&image.snapshot)?; + let index = CongeeIndex::from_topology(topology, OffsetEqLink)?; + apply_wal(&index, &image.wal, OffsetEqLink); + Ok(PersistentArtIndex::from_inner(index)) + } + + /// Writes a complete native Congee checkpoint with an empty WAL. + pub async fn write_checkpoint( + path: impl AsRef, + table_version: u32, + index: &mut PersistentCongeeIndex>, + ) -> eyre::Result<()> { + let topology = index.inner_mut().export_topology(|link| link.0)?; + let snapshot = encode_congee_topology(&topology)?; + ArtFile::::write_new_file(path.as_ref(), Backend::Congee, table_version, &snapshot).await + } + + async fn compact(&mut self) -> eyre::Result<()> { + let image = ArtFile::::read_image(&self.file.path, Backend::Congee, self.file.table_version).await?; + let topology = decode_congee_topology(&image.snapshot)?; + let mut index = CongeeIndex::from_topology(topology, |link| link)?; + apply_wal(&index, &image.wal, |link| link); + let snapshot = encode_congee_topology(&index.export_topology(|link| *link)?)?; + self.file.rewrite(&snapshot).await + } +} + +impl SpaceIndexOps for SpaceCongeeIndex +where + K: ArtPersistenceKey + CongeeKey, +{ + async fn primary_from_table_files_path + Send>(path: S, version: u32) -> eyre::Result { + Self::new( + PathBuf::from(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( + PathBuf::from(format!("{}/{}{}", path.as_ref(), name.as_ref(), WT_INDEX_EXTENSION)), + version, + ) + .await + } + + async fn bootstrap(_: &mut File, _: String, _: u32) -> eyre::Result<()> { + Ok(()) + } + + async fn process_change_event(&mut self, event: ChangeEvent>) -> eyre::Result<()> { + self.file.append(&[logical_record(event)?]).await?; + if self.file.should_compact() { + self.compact().await?; + } + Ok(()) + } + + async fn process_change_event_batch(&mut self, events: BatchChangeEvent) -> eyre::Result<()> { + let records = events + .into_iter() + .map(logical_record) + .collect::>>()?; + self.file.append(&records).await?; + if self.file.should_compact() { + self.compact().await?; + } + Ok(()) + } +} + +fn encode_link(link: Link, output: &mut Vec) { + let page_id: usize = link.page_id.into(); + output.extend_from_slice(&(page_id as u32).to_le_bytes()); + output.extend_from_slice(&link.offset.to_le_bytes()); + output.extend_from_slice(&link.length.to_le_bytes()); +} + +struct Decoder<'a> { + bytes: &'a [u8], + position: usize, +} + +impl<'a> Decoder<'a> { + fn new(bytes: &'a [u8]) -> Self { + Self { bytes, position: 0 } + } + + fn take(&mut self, count: usize) -> eyre::Result<&'a [u8]> { + let end = self + .position + .checked_add(count) + .ok_or_else(|| eyre!("ART topology length overflow"))?; + let value = self + .bytes + .get(self.position..end) + .ok_or_else(|| eyre!("truncated ART topology at byte {}", self.position))?; + self.position = end; + Ok(value) + } + + fn u8(&mut self) -> eyre::Result { + Ok(self.take(1)?[0]) + } + + fn u16(&mut self) -> eyre::Result { + Ok(u16::from_le_bytes(self.take(2)?.try_into().unwrap())) + } + + fn u64(&mut self) -> eyre::Result { + Ok(u64::from_le_bytes(self.take(8)?.try_into().unwrap())) + } + + fn link(&mut self) -> eyre::Result { + let page_id = u32::from_le_bytes(self.take(4)?.try_into().unwrap()); + let offset = u32::from_le_bytes(self.take(4)?.try_into().unwrap()); + let length = u32::from_le_bytes(self.take(4)?.try_into().unwrap()); + Ok(Link { + page_id: PageId::from(page_id), + offset, + length, + }) + } + + fn finish(self) -> eyre::Result<()> { + if self.position != self.bytes.len() { + bail!( + "ART topology contains {} trailing bytes", + self.bytes.len() - self.position + ); + } + Ok(()) + } +} + +fn encode_arctic_topology(topology: &arctic::topology::Topology) -> eyre::Result> { + let mut output = Vec::new(); + output.extend_from_slice(&topology.version.to_le_bytes()); + match &topology.root { + Some(root) => { + output.push(1); + encode_arctic_edge(root, &mut output)?; + } + None => output.push(0), + } + Ok(output) +} + +fn encode_arctic_edge(edge: &arctic::topology::Edge, output: &mut Vec) -> eyre::Result<()> { + output.extend_from_slice(&edge.metadata.to_le_bytes()); + match &edge.child { + arctic::topology::Child::Value(link) => { + output.push(0); + encode_link(*link, output); + } + arctic::topology::Child::Node(node) => { + output.push(1); + encode_arctic_node(node, output)?; + } + } + Ok(()) +} + +fn encode_arctic_node(node: &arctic::topology::Node, output: &mut Vec) -> eyre::Result<()> { + let kind = match node.kind { + arctic::topology::NodeKind::Node3 => 3, + arctic::topology::NodeKind::Node15 => 15, + arctic::topology::NodeKind::Node47 => 47, + arctic::topology::NodeKind::Node256 => 255, + }; + output.push(kind); + output.extend_from_slice(&node.slot_count.to_le_bytes()); + let branch_count = u16::try_from(node.branches.len()).map_err(|_| eyre!("too many Arctic branches"))?; + output.extend_from_slice(&branch_count.to_le_bytes()); + for branch in &node.branches { + output.push(branch.key); + output.extend_from_slice(&branch.slot.to_le_bytes()); + encode_arctic_edge(&branch.edge, output)?; + } + Ok(()) +} + +fn decode_arctic_topology(bytes: &[u8]) -> eyre::Result> { + let mut decoder = Decoder::new(bytes); + let version = decoder.u16()?; + let root = match decoder.u8()? { + 0 => None, + 1 => Some(decode_arctic_edge(&mut decoder, 0)?), + tag => bail!("invalid Arctic root tag {tag}"), + }; + decoder.finish()?; + Ok(arctic::topology::Topology { version, root }) +} + +fn decode_arctic_edge(decoder: &mut Decoder<'_>, depth: usize) -> eyre::Result> { + if depth > 16 { + bail!("Arctic topology exceeds maximum key depth"); + } + let metadata = decoder.u64()?; + let child = match decoder.u8()? { + 0 => arctic::topology::Child::Value(decoder.link()?), + 1 => arctic::topology::Child::Node(decode_arctic_node(decoder, depth + 1)?), + tag => bail!("invalid Arctic child tag {tag}"), + }; + Ok(arctic::topology::Edge { metadata, child }) +} + +fn decode_arctic_node(decoder: &mut Decoder<'_>, depth: usize) -> eyre::Result> { + let (kind, capacity) = match decoder.u8()? { + 3 => (arctic::topology::NodeKind::Node3, 3), + 15 => (arctic::topology::NodeKind::Node15, 15), + 47 => (arctic::topology::NodeKind::Node47, 47), + 255 => (arctic::topology::NodeKind::Node256, 256), + tag => bail!("invalid Arctic node kind {tag}"), + }; + let slot_count = decoder.u16()?; + let count = decoder.u16()? as usize; + if count > capacity { + bail!("Arctic node has {count} branches but capacity is {capacity}"); + } + let mut branches = Vec::with_capacity(count); + for _ in 0..count { + branches.push(arctic::topology::Branch { + key: decoder.u8()?, + slot: decoder.u16()?, + edge: decode_arctic_edge(decoder, depth)?, + }); + } + Ok(arctic::topology::Node { + kind, + slot_count, + branches, + }) +} + +fn encode_congee_topology(topology: &congee::topology::Topology) -> eyre::Result> { + let mut output = Vec::new(); + output.extend_from_slice(&topology.version.to_le_bytes()); + encode_congee_node(&topology.root, &mut output)?; + Ok(output) +} + +fn encode_congee_node(node: &congee::topology::Node, output: &mut Vec) -> eyre::Result<()> { + let kind = match node.kind { + congee::topology::NodeKind::N4 => 4, + congee::topology::NodeKind::N16 => 16, + congee::topology::NodeKind::N48 => 48, + congee::topology::NodeKind::N256 => 255, + }; + output.push(kind); + let prefix_len = u8::try_from(node.prefix.len()).map_err(|_| eyre!("Congee prefix is too long"))?; + output.push(prefix_len); + output.extend_from_slice(&node.prefix); + let branch_count = u16::try_from(node.branches.len()).map_err(|_| eyre!("too many Congee branches"))?; + output.extend_from_slice(&branch_count.to_le_bytes()); + let free_count = u16::try_from(node.free_slots.len()).map_err(|_| eyre!("too many Congee free slots"))?; + output.extend_from_slice(&free_count.to_le_bytes()); + output.extend_from_slice(&node.free_slots); + for branch in &node.branches { + output.push(branch.key); + output.extend_from_slice(&branch.slot.to_le_bytes()); + match &branch.child { + congee::topology::Child::Value(link) => { + output.push(0); + encode_link(*link, output); + } + congee::topology::Child::Node(child) => { + output.push(1); + encode_congee_node(child, output)?; + } + } + } + Ok(()) +} + +fn decode_congee_topology(bytes: &[u8]) -> eyre::Result> { + let mut decoder = Decoder::new(bytes); + let version = decoder.u16()?; + let root = decode_congee_node(&mut decoder, 0)?; + decoder.finish()?; + Ok(congee::topology::Topology { version, root }) +} + +fn decode_congee_node(decoder: &mut Decoder<'_>, depth: usize) -> eyre::Result> { + if depth > 8 { + bail!("Congee topology exceeds its eight-byte key depth"); + } + let (kind, capacity) = match decoder.u8()? { + 4 => (congee::topology::NodeKind::N4, 4), + 16 => (congee::topology::NodeKind::N16, 16), + 48 => (congee::topology::NodeKind::N48, 48), + 255 => (congee::topology::NodeKind::N256, 256), + tag => bail!("invalid Congee node kind {tag}"), + }; + let prefix_len = decoder.u8()? as usize; + if prefix_len > 8 { + bail!("Congee prefix length {prefix_len} exceeds eight bytes"); + } + let prefix = decoder.take(prefix_len)?.to_vec(); + let branch_count = decoder.u16()? as usize; + if branch_count > capacity { + bail!("Congee node has {branch_count} branches but capacity is {capacity}"); + } + let free_count = decoder.u16()? as usize; + if free_count > 48 { + bail!("Congee node has {free_count} free slots"); + } + let free_slots = decoder.take(free_count)?.to_vec(); + let mut branches = Vec::with_capacity(branch_count); + for _ in 0..branch_count { + let key = decoder.u8()?; + let slot = decoder.u16()?; + let child = match decoder.u8()? { + 0 => congee::topology::Child::Value(decoder.link()?), + 1 => congee::topology::Child::Node(decode_congee_node(decoder, depth + 1)?), + tag => bail!("invalid Congee child tag {tag}"), + }; + branches.push(congee::topology::Branch { key, slot, child }); + } + Ok(congee::topology::Node { + kind, + prefix, + branches, + free_slots, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn link(value: u32) -> Link { + Link { + page_id: value.into(), + offset: value * 2, + length: value * 3, + } + } + + fn set_event(id: u64, key: u64, value: Link) -> ChangeEvent> { + let pair = Pair { key, value }; + ChangeEvent::InsertAt { + event_id: id.into(), + max_value: pair.clone(), + value: pair, + index: 0, + } + } + + fn congee_contains_kind(node: &congee::topology::Node, expected: congee::topology::NodeKind) -> bool { + node.kind == expected + || node.branches.iter().any(|branch| match &branch.child { + congee::topology::Child::Value(_) => false, + congee::topology::Child::Node(child) => congee_contains_kind(child, expected), + }) + } + + #[test] + fn wal_round_trip() { + for record in [ + WalRecord { + event_id: 17, + key: 42u64, + op: WalOp::Set(link(3)), + }, + WalRecord { + event_id: 18, + key: 42u64, + op: WalOp::Remove, + }, + ] { + assert_eq!(decode_wal_record(&encode_wal_record(&record)).unwrap(), record); + } + } + + #[test] + fn arctic_topology_codec_preserves_physical_shape() { + let mut index = ArcticIndex::::default(); + for key in 0..80 { + index.insert_value(key, link(key as u32 + 1)); + } + let topology = index.export_topology(|link| *link).unwrap(); + let bytes = encode_arctic_topology(&topology).unwrap(); + let decoded = decode_arctic_topology(&bytes).unwrap(); + assert_eq!(decoded, topology); + } + + #[test] + fn congee_topology_codec_preserves_physical_shape() { + let mut index = CongeeIndex::::default(); + for key in 0..80 { + index.insert_value(key, link(key as u32 + 1)); + } + let topology = index.export_topology(|link| *link).unwrap(); + let bytes = encode_congee_topology(&topology).unwrap(); + let decoded = decode_congee_topology(&bytes).unwrap(); + assert_eq!(decoded, topology); + } + + #[tokio::test] + async fn torn_wal_tail_is_truncated_before_new_appends() { + let path = std::env::temp_dir().join(format!("worktable-art-torn-{}.wt.idx", uuid::Uuid::new_v4())); + let mut space = SpaceArcticIndex::::new(path.clone(), 1).await.unwrap(); + space.process_change_event(set_event(0, 7, link(7))).await.unwrap(); + drop(space); + + let durable_len = tokio::fs::metadata(&path).await.unwrap().len(); + let mut file = OpenOptions::new().append(true).open(&path).await.unwrap(); + file.write_all(&WAL_MAGIC[..2]).await.unwrap(); + file.flush().await.unwrap(); + drop(file); + assert_eq!(tokio::fs::metadata(&path).await.unwrap().len(), durable_len + 2); + + let mut space = SpaceArcticIndex::::new(path.clone(), 1).await.unwrap(); + assert_eq!(tokio::fs::metadata(&path).await.unwrap().len(), durable_len); + space.process_change_event(set_event(1, 8, link(8))).await.unwrap(); + drop(space); + + let index = SpaceArcticIndex::::load_index::<4096>(&path, 1) + .await + .unwrap(); + assert_eq!(index.get_value(&7).unwrap().0, link(7)); + assert_eq!(index.get_value(&8).unwrap().0, link(8)); + tokio::fs::remove_file(path).await.unwrap(); + } + + #[tokio::test] + async fn compaction_replaces_wal_with_native_checkpoint() { + let path = std::env::temp_dir().join(format!("worktable-art-compact-{}.wt.idx", uuid::Uuid::new_v4())); + let mut space = SpaceCongeeIndex::::new(path.clone(), 3).await.unwrap(); + let events = (0..128).map(|key| set_event(key, key, link(key as u32 + 1))).collect(); + space.process_change_event_batch(events).await.unwrap(); + assert!( + !ArtFile::::read_image(&path, Backend::Congee, 3) + .await + .unwrap() + .wal + .is_empty() + ); + space.compact().await.unwrap(); + let image = ArtFile::::read_image(&path, Backend::Congee, 3).await.unwrap(); + assert!(image.wal.is_empty()); + let topology = decode_congee_topology(&image.snapshot).unwrap(); + assert!(congee_contains_kind(&topology.root, congee::topology::NodeKind::N256)); + drop(space); + + let index = SpaceCongeeIndex::::load_index::<4096>(&path, 3) + .await + .unwrap(); + assert_eq!(index.len(), 128); + assert_eq!(index.get_value(&91).unwrap().0, link(92)); + tokio::fs::remove_file(path).await.unwrap(); + } + + #[tokio::test] + async fn complete_corruption_and_header_mismatches_are_rejected() { + let path = std::env::temp_dir().join(format!("worktable-art-corrupt-{}.wt.idx", uuid::Uuid::new_v4())); + let mut space = SpaceArcticIndex::::new(path.clone(), 7).await.unwrap(); + space.process_change_event(set_event(0, 9, link(9))).await.unwrap(); + drop(space); + + assert!(ArtFile::::read_image(&path, Backend::Arctic, 8).await.is_err()); + assert!(ArtFile::::read_image(&path, Backend::Congee, 7).await.is_err()); + assert!(ArtFile::::read_image(&path, Backend::Arctic, 7).await.is_err()); + + let mut file = OpenOptions::new().read(true).write(true).open(&path).await.unwrap(); + file.seek(std::io::SeekFrom::End(-1)).await.unwrap(); + let mut last = [0u8; 1]; + file.read_exact(&mut last).await.unwrap(); + file.seek(std::io::SeekFrom::End(-1)).await.unwrap(); + file.write_all(&[last[0] ^ 0x80]).await.unwrap(); + file.flush().await.unwrap(); + drop(file); + + assert!(ArtFile::::read_image(&path, Backend::Arctic, 7).await.is_err()); + tokio::fs::remove_file(path).await.unwrap(); + } +} diff --git a/src/persistence/space/mod.rs b/src/persistence/space/mod.rs index 54ef46fb..dee296f3 100644 --- a/src/persistence/space/mod.rs +++ b/src/persistence/space/mod.rs @@ -1,3 +1,4 @@ +mod art_index; mod data; mod index; @@ -11,6 +12,7 @@ use indexset::cdc::change::ChangeEvent; use indexset::core::pair::Pair; use tokio::fs::{File, OpenOptions}; +pub use art_index::{ArtPersistenceKey, SpaceArcticIndex, SpaceCongeeIndex}; pub use data::SpaceData; pub use index::{ IndexTableOfContents, SpaceIndex, SpaceIndexUnsized, map_index_pages_to_toc_and_general, diff --git a/tests/worktable/index_backends.rs b/tests/worktable/index_backends.rs index dec823a1..636e03d6 100644 --- a/tests/worktable/index_backends.rs +++ b/tests/worktable/index_backends.rs @@ -94,6 +94,30 @@ worktable! { }, } +worktable! { + name: PersistedArctic, + persist: true, + columns: { + id: u64 primary_key autoincrement using arctic, + congee_key: u64, + }, + indexes: { + congee_idx: congee_key unique using congee, + }, +} + +worktable! { + name: PersistedCongee, + persist: true, + columns: { + id: u64 primary_key autoincrement using congee, + arctic_key: u64, + }, + indexes: { + arctic_idx: arctic_key unique using arctic, + }, +} + #[tokio::test] async fn all_unique_backends_support_crud_ranges_and_conflict_rollback() { let table = MixedBackendWorkTable::default(); @@ -265,6 +289,109 @@ async fn upstream_indexset_survives_persist_reload_and_more_writes() { remove_dir_if_exists(ROOT.to_string()).await; } +#[tokio::test] +async fn native_art_backends_survive_wal_reload_and_further_mutation() { + const ARCTIC_ROOT: &str = "tests/data/index_backend_arctic_persistence"; + const CONGEE_ROOT: &str = "tests/data/index_backend_congee_persistence"; + remove_dir_if_exists(ARCTIC_ROOT.to_string()).await; + remove_dir_if_exists(CONGEE_ROOT.to_string()).await; + + let arctic_config = DiskConfig::new_with_table_name( + ARCTIC_ROOT, + PersistedArcticWorkTable::name_snake_case(), + PersistedArcticWorkTable::version(), + ); + let engine = PersistedArcticPersistenceEngine::new(arctic_config.clone()) + .await + .unwrap(); + let table = PersistedArcticWorkTable::load(engine).await.unwrap(); + for congee_key in 0..256 { + table + .insert(PersistedArcticRow { + id: table.get_next_pk().into(), + congee_key, + }) + .unwrap(); + } + let rejected_id = table.get_next_pk().0; + assert!( + table + .insert(PersistedArcticRow { + id: rejected_id, + congee_key: 77, + }) + .is_err() + ); + let accepted_id = table.get_next_pk().0; + table + .insert(PersistedArcticRow { + id: accepted_id, + congee_key: 300, + }) + .unwrap(); + table.wait_for_ops().await; + drop(table); + + let engine = PersistedArcticPersistenceEngine::new(arctic_config.clone()) + .await + .unwrap(); + let table = PersistedArcticWorkTable::load(engine).await.unwrap(); + assert_eq!(table.count(), 257); + assert!(table.select(rejected_id).is_none()); + assert_eq!(table.select(accepted_id).unwrap().congee_key, 300); + assert_eq!(table.select_by_congee_key(77).unwrap().congee_key, 77); + table.delete(77).await.unwrap(); + table.wait_for_ops().await; + drop(table); + + let engine = PersistedArcticPersistenceEngine::new(arctic_config).await.unwrap(); + let table = PersistedArcticWorkTable::load(engine).await.unwrap(); + assert!(table.select(77).is_none()); + assert!(table.select_by_congee_key(77).is_none()); + table.wait_for_ops().await; + drop(table); + + let congee_config = DiskConfig::new_with_table_name( + CONGEE_ROOT, + PersistedCongeeWorkTable::name_snake_case(), + PersistedCongeeWorkTable::version(), + ); + let engine = PersistedCongeePersistenceEngine::new(congee_config.clone()) + .await + .unwrap(); + let table = PersistedCongeeWorkTable::load(engine).await.unwrap(); + for arctic_key in 0..256 { + table + .insert(PersistedCongeeRow { + id: table.get_next_pk().into(), + arctic_key, + }) + .unwrap(); + } + table.wait_for_ops().await; + drop(table); + + let engine = PersistedCongeePersistenceEngine::new(congee_config.clone()) + .await + .unwrap(); + let table = PersistedCongeeWorkTable::load(engine).await.unwrap(); + assert_eq!(table.count(), 256); + assert_eq!(table.select_by_arctic_key(199).unwrap().arctic_key, 199); + table.delete(199).await.unwrap(); + table.wait_for_ops().await; + drop(table); + + let engine = PersistedCongeePersistenceEngine::new(congee_config).await.unwrap(); + let table = PersistedCongeeWorkTable::load(engine).await.unwrap(); + assert!(table.select(199).is_none()); + assert!(table.select_by_arctic_key(199).is_none()); + table.wait_for_ops().await; + drop(table); + + remove_dir_if_exists(ARCTIC_ROOT.to_string()).await; + remove_dir_if_exists(CONGEE_ROOT.to_string()).await; +} + #[tokio::test] async fn persisted_tables_can_switch_between_wti_and_upstream_without_rebuild() { use provider_switch_upstream as upstream; From 19b6ef247eed3692eae8aee54e7648ae62080530 Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 4 Aug 2026 01:17:01 +0700 Subject: [PATCH 2/5] test: validate ART persistence concurrency and cost --- examples/art_backend_probe.rs | 123 ++++++++++++++++ examples/art_persistence_overhead_probe.rs | 154 +++++++++++++++++++++ examples/wti_persistence_overhead_probe.rs | 92 ++++++++++++ src/index/persistent_art.rs | 7 +- tests/worktable/index_backends.rs | 63 +++++++++ 5 files changed, 436 insertions(+), 3 deletions(-) create mode 100644 examples/art_backend_probe.rs create mode 100644 examples/art_persistence_overhead_probe.rs create mode 100644 examples/wti_persistence_overhead_probe.rs diff --git a/examples/art_backend_probe.rs b/examples/art_backend_probe.rs new file mode 100644 index 00000000..73b3a1ce --- /dev/null +++ b/examples/art_backend_probe.rs @@ -0,0 +1,123 @@ +//! Small repeatable probe for ART point-operation regressions. +//! +//! This is deliberately separate from publication benchmarks. Run it several +//! times on the same quiet machine and compare revisions with identical build +//! flags: +//! +//! `cargo run --release --example art_backend_probe` + +use std::hint::black_box; +use std::time::{Duration, Instant}; + +use worktable::prelude::{ArcticIndex, CongeeIndex, UniqueIndex}; + +const POPULATION: u64 = 65_536; +const READ_OPERATIONS: u64 = 4_000_000; +const MUTATION_PAIRS: u64 = 250_000; +const TRIALS: usize = 10; + +fn measure(operation: impl Fn()) -> Duration { + let started = Instant::now(); + operation(); + started.elapsed() +} + +fn nanoseconds_per(duration: Duration, operations: u64) -> f64 { + duration.as_secs_f64() * 1_000_000_000.0 / operations as f64 +} + +fn summarize(name: &str, mut values: Vec) { + values.sort_by(f64::total_cmp); + let median = (values[TRIALS / 2 - 1] + values[TRIALS / 2]) / 2.0; + println!("{name:24} median {median:8.3} ns/op trials {values:?}"); +} + +fn probe_congee() { + let index = CongeeIndex::::default(); + for key in 0..POPULATION { + index.insert_value(key, key ^ 0x5a5a_5a5a); + } + + // Warm instruction/data caches before collecting whole-run averages. + for key in 0..POPULATION { + black_box(index.get_value(black_box(&key))); + } + + let reads = (0..TRIALS) + .map(|trial| { + nanoseconds_per( + measure(|| { + for operation in 0..READ_OPERATIONS { + let key = operation.wrapping_mul(1_103_515_245).wrapping_add(trial as u64) & (POPULATION - 1); + black_box(index.get_value(black_box(&key))); + } + }), + READ_OPERATIONS, + ) + }) + .collect(); + summarize("congee point read", reads); + + let mutations = (0..TRIALS) + .map(|trial| { + nanoseconds_per( + measure(|| { + for operation in 0..MUTATION_PAIRS { + let key = POPULATION + operation + trial as u64 * MUTATION_PAIRS; + black_box(index.insert_value(black_box(key), black_box(key))); + black_box(index.remove_value(black_box(&key))); + } + }), + MUTATION_PAIRS * 2, + ) + }) + .collect(); + summarize("congee insert/remove", mutations); +} + +fn probe_arctic() { + let index = ArcticIndex::::default(); + for key in 0..POPULATION { + index.insert_value(key, key ^ 0x5a5a_5a5a); + } + + for key in 0..POPULATION { + black_box(index.get_value(black_box(&key))); + } + + let reads = (0..TRIALS) + .map(|trial| { + nanoseconds_per( + measure(|| { + for operation in 0..READ_OPERATIONS { + let key = operation.wrapping_mul(1_103_515_245).wrapping_add(trial as u64) & (POPULATION - 1); + black_box(index.get_value(black_box(&key))); + } + }), + READ_OPERATIONS, + ) + }) + .collect(); + summarize("arctic point read", reads); + + let mutations = (0..TRIALS) + .map(|trial| { + nanoseconds_per( + measure(|| { + for operation in 0..MUTATION_PAIRS { + let key = POPULATION + operation + trial as u64 * MUTATION_PAIRS; + black_box(index.insert_value(black_box(key), black_box(key))); + black_box(index.remove_value(black_box(&key))); + } + }), + MUTATION_PAIRS * 2, + ) + }) + .collect(); + summarize("arctic insert/remove", mutations); +} + +fn main() { + probe_congee(); + probe_arctic(); +} diff --git a/examples/art_persistence_overhead_probe.rs b/examples/art_persistence_overhead_probe.rs new file mode 100644 index 00000000..3863bbaf --- /dev/null +++ b/examples/art_persistence_overhead_probe.rs @@ -0,0 +1,154 @@ +//! Measures the in-process sequencing/event cost of persisted ART indexes. +//! +//! Disk encoding and I/O happen in WorkTable's persistence task and are not +//! included here. This probe isolates the synchronous mutation-front-end cost: +//! `cargo run --release --example art_persistence_overhead_probe`. + +use std::hint::black_box; +use std::time::{Duration, Instant}; + +use worktable::prelude::{ + ArcticIndex, CongeeIndex, Link, OffsetEqLink, PersistentArcticIndex, PersistentCongeeIndex, TableIndexCdc, + UniqueIndex, +}; + +const POPULATION: u64 = 65_536; +const READ_OPERATIONS: u64 = 4_000_000; +const MUTATION_PAIRS: u64 = 250_000; +const TRIALS: usize = 10; +const DATA_LENGTH: usize = 4_096; + +fn measure(operation: impl Fn()) -> Duration { + let started = Instant::now(); + operation(); + started.elapsed() +} + +fn nanoseconds_per(duration: Duration, operations: u64) -> f64 { + duration.as_secs_f64() * 1_000_000_000.0 / operations as f64 +} + +fn summarize(name: &str, mut values: Vec) { + values.sort_by(f64::total_cmp); + let median = (values[TRIALS / 2 - 1] + values[TRIALS / 2]) / 2.0; + println!("{name:30} median {median:8.3} ns/op trials {values:?}"); +} + +fn link(operation: u64) -> Link { + Link { + page_id: ((operation / DATA_LENGTH as u64) as u32).into(), + offset: (operation % DATA_LENGTH as u64) as u32, + length: 8, + } +} + +fn point_read_trial(index: &I, trial: usize) -> f64 +where + I: UniqueIndex>, +{ + nanoseconds_per( + measure(|| { + for operation in 0..READ_OPERATIONS { + let key = operation.wrapping_mul(1_103_515_245).wrapping_add(trial as u64) & (POPULATION - 1); + black_box(index.get_value(black_box(&key))); + } + }), + READ_OPERATIONS, + ) +} + +fn cdc_mutation_trial(index: &I, trial: usize) -> f64 +where + I: TableIndexCdc, +{ + nanoseconds_per( + measure(|| { + for operation in 0..MUTATION_PAIRS { + let key = POPULATION + operation + trial as u64 * MUTATION_PAIRS; + let value = link(operation); + black_box(TableIndexCdc::insert_cdc(index, black_box(key), black_box(value))); + black_box(TableIndexCdc::remove_cdc(index, black_box(key), black_box(value))); + } + }), + MUTATION_PAIRS * 2, + ) +} + +fn paired_reads(native: &I, persisted: &P) -> (Vec, Vec) +where + I: UniqueIndex>, + P: UniqueIndex>, +{ + let mut native_results = Vec::with_capacity(TRIALS); + let mut persisted_results = Vec::with_capacity(TRIALS); + for trial in 0..TRIALS { + if trial % 2 == 0 { + native_results.push(point_read_trial(native, trial)); + persisted_results.push(point_read_trial(persisted, trial)); + } else { + persisted_results.push(point_read_trial(persisted, trial)); + native_results.push(point_read_trial(native, trial)); + } + } + (native_results, persisted_results) +} + +fn paired_mutations(native: &I, persisted: &P) -> (Vec, Vec) +where + I: TableIndexCdc, + P: TableIndexCdc, +{ + let mut native_results = Vec::with_capacity(TRIALS); + let mut persisted_results = Vec::with_capacity(TRIALS); + for trial in 0..TRIALS { + if trial % 2 == 0 { + native_results.push(cdc_mutation_trial(native, trial)); + persisted_results.push(cdc_mutation_trial(persisted, trial)); + } else { + persisted_results.push(cdc_mutation_trial(persisted, trial)); + native_results.push(cdc_mutation_trial(native, trial)); + } + } + (native_results, persisted_results) +} + +fn congee() { + let native = CongeeIndex::>::default(); + let persisted = PersistentCongeeIndex::>::default(); + for key in 0..POPULATION { + let value = OffsetEqLink(link(key)); + native.insert_value(key, value); + persisted.insert_value(key, value); + } + + let (native_reads, persisted_reads) = paired_reads(&native, &persisted); + summarize("congee native point read", native_reads); + summarize("congee persisted point read", persisted_reads); + + let (native_mutations, persisted_mutations) = paired_mutations(&native, &persisted); + summarize("congee native CDC mutation", native_mutations); + summarize("congee persisted CDC mutation", persisted_mutations); +} + +fn arctic() { + let native = ArcticIndex::>::default(); + let persisted = PersistentArcticIndex::>::default(); + for key in 0..POPULATION { + let value = OffsetEqLink(link(key)); + native.insert_value(key, value); + persisted.insert_value(key, value); + } + + let (native_reads, persisted_reads) = paired_reads(&native, &persisted); + summarize("arctic native point read", native_reads); + summarize("arctic persisted point read", persisted_reads); + + let (native_mutations, persisted_mutations) = paired_mutations(&native, &persisted); + summarize("arctic native CDC mutation", native_mutations); + summarize("arctic persisted CDC mutation", persisted_mutations); +} + +fn main() { + congee(); + arctic(); +} diff --git a/examples/wti_persistence_overhead_probe.rs b/examples/wti_persistence_overhead_probe.rs new file mode 100644 index 00000000..e65f3454 --- /dev/null +++ b/examples/wti_persistence_overhead_probe.rs @@ -0,0 +1,92 @@ +//! Measures WorkTablesIndex's synchronous structural-CDC mutation cost. +//! +//! This is the WTI control for the ART persistence probe. It excludes the +//! asynchronous data/index file writes: +//! `cargo run --release --example wti_persistence_overhead_probe`. + +use std::hint::black_box; +use std::time::{Duration, Instant}; + +use worktable::prelude::{IndexMap, Link, OffsetEqLink, TableIndexCdc, UniqueIndex}; + +const POPULATION: u64 = 65_536; +const MUTATION_PAIRS: u64 = 250_000; +const TRIALS: usize = 10; +const DATA_LENGTH: usize = 4_096; + +fn link(operation: u64) -> Link { + Link { + page_id: ((operation / DATA_LENGTH as u64) as u32).into(), + offset: (operation % DATA_LENGTH as u64) as u32, + length: 8, + } +} + +fn measure(operation: impl Fn()) -> Duration { + let started = Instant::now(); + operation(); + started.elapsed() +} + +fn nanoseconds_per(duration: Duration, operations: u64) -> f64 { + duration.as_secs_f64() * 1_000_000_000.0 / operations as f64 +} + +fn summarize(name: &str, mut values: Vec) { + values.sort_by(f64::total_cmp); + let median = (values[TRIALS / 2 - 1] + values[TRIALS / 2]) / 2.0; + println!("{name:30} median {median:8.3} ns/op trials {values:?}"); +} + +fn direct_trial(index: &IndexMap>, trial: usize) -> f64 { + nanoseconds_per( + measure(|| { + for operation in 0..MUTATION_PAIRS { + let key = POPULATION + operation + trial as u64 * MUTATION_PAIRS; + let value = OffsetEqLink(link(operation)); + black_box(index.insert_value(black_box(key), black_box(value))); + black_box(index.remove_value(black_box(&key))); + } + }), + MUTATION_PAIRS * 2, + ) +} + +fn cdc_trial(index: &IndexMap>, trial: usize) -> f64 { + nanoseconds_per( + measure(|| { + for operation in 0..MUTATION_PAIRS { + let key = POPULATION + operation + trial as u64 * MUTATION_PAIRS; + let value = link(operation); + black_box(TableIndexCdc::insert_cdc(index, black_box(key), black_box(value))); + black_box(TableIndexCdc::remove_cdc(index, black_box(key), black_box(value))); + } + }), + MUTATION_PAIRS * 2, + ) +} + +fn main() { + let direct = IndexMap::>::default(); + let cdc = IndexMap::>::default(); + for key in 0..POPULATION { + let value = OffsetEqLink(link(key)); + direct.insert_value(key, value); + cdc.insert_value(key, value); + } + + let mut direct_results = Vec::with_capacity(TRIALS); + let mut cdc_results = Vec::with_capacity(TRIALS); + for trial in 0..TRIALS { + if trial % 2 == 0 { + direct_results.push(direct_trial(&direct, trial)); + cdc_results.push(cdc_trial(&cdc, trial)); + } else { + cdc_results.push(cdc_trial(&cdc, trial)); + direct_results.push(direct_trial(&direct, trial)); + } + } + + summarize("WTI direct mutation", direct_results); + summarize("WTI structural CDC mutation", cdc_results); +} diff --git a/src/index/persistent_art.rs b/src/index/persistent_art.rs index 34223faa..4d74eca9 100644 --- a/src/index/persistent_art.rs +++ b/src/index/persistent_art.rs @@ -55,9 +55,10 @@ impl Default for PersistentArtIndex { } impl PersistentArtIndex { - /// Wraps a reconstructed ART. Event ids restart at zero because startup - /// recovery checkpoints and clears the preceding WAL before accepting new - /// operations. + /// Wraps a reconstructed ART. Event ids are session-local and restart at + /// zero with the persistence analyzer. The durable WAL is replayed in file + /// append order, so records from earlier sessions do not need their event + /// ids renumbered. pub fn from_inner(inner: I) -> Self { Self { inner, diff --git a/tests/worktable/index_backends.rs b/tests/worktable/index_backends.rs index 636e03d6..2465d787 100644 --- a/tests/worktable/index_backends.rs +++ b/tests/worktable/index_backends.rs @@ -392,6 +392,69 @@ async fn native_art_backends_survive_wal_reload_and_further_mutation() { remove_dir_if_exists(CONGEE_ROOT.to_string()).await; } +#[tokio::test] +async fn native_art_backends_recover_concurrent_same_row_updates() { + use std::sync::Arc; + + use tokio::sync::Barrier; + + const ROOT: &str = "tests/data/index_backend_art_concurrent_persistence"; + 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, + PersistedArcticWorkTable::name_snake_case(), + PersistedArcticWorkTable::version(), + ); + let engine = PersistedArcticPersistenceEngine::new(config.clone()).await.unwrap(); + let table = Arc::new(PersistedArcticWorkTable::load(engine).await.unwrap()); + let id = table.get_next_pk().0; + table.insert(PersistedArcticRow { id, congee_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(PersistedArcticRow { + id, + congee_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; + drop(table); + + let engine = PersistedArcticPersistenceEngine::new(config).await.unwrap(); + let table = PersistedArcticWorkTable::load(engine).await.unwrap(); + assert_eq!(table.select(id), Some(expected.clone())); + assert_eq!(table.select_by_congee_key(expected.congee_key), Some(expected.clone())); + for key in 10_000..10_000 + WORKERS * UPDATES_PER_WORKER { + if key != expected.congee_key { + assert!(table.select_by_congee_key(key).is_none()); + } + } + table.wait_for_ops().await; + 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; From bf37e2cd38104a621a8c1b01a30cc6c7817f1933 Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 4 Aug 2026 01:17:07 +0700 Subject: [PATCH 3/5] release: prepare 1.0.0-beta.2 --- Cargo.toml | 4 ++-- README.md | 6 +++--- codegen/Cargo.toml | 2 +- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index d46e93f5..8e81fcea 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ members = ["codegen", "examples", "performance_measurement", "performance_measur [package] name = "worktable" -version = "1.0.0-beta.1" +version = "1.0.0-beta.2" edition = "2024" authors = ["Handy-caT"] license = "MIT" @@ -59,7 +59,7 @@ tracing = "0.1" url = { version = "2", optional = true } uuid = { version = "1.24.0", features = ["v4", "v7"] } walkdir = { version = "2", optional = true } -worktable_codegen = { path = "codegen", version = "=1.0.0-beta.1" } +worktable_codegen = { path = "codegen", version = "=1.0.0-beta.2" } [dev-dependencies] chrono = "0.4.43" diff --git a/README.md b/README.md index 430957d3..023fb7c2 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,7 @@ from a macro, and that persisting it is one feature flag away. ## Install ```sh -cargo add worktable@1.0.0-beta.1 +cargo add worktable@1.0.0-beta.2 ``` ## What you get @@ -44,7 +44,7 @@ S3 support layers *on top of* the disk engine rather than replacing it. ```toml [dependencies] -worktable = { version = "=1.0.0-beta.1", features = ["s3-support"] } # S3 sync, optional +worktable = { version = "=1.0.0-beta.2", features = ["s3-support"] } # S3 sync, optional ``` Persisted indexes default to WorkTablesIndex. Vanilla IndexSet can be selected explicitly with `using indexset` while retaining the existing disk/S3 representation. Congee and Arctic are explicitly memory-only and require `persist: false`. The full syntax and capability matrix are documented in [Per-index backends with `using`](docs/index-backend-dsl-proposal.md). @@ -60,7 +60,7 @@ opt into immutable row-version publication: ```toml [dependencies] -worktable = { version = "=1.0.0-beta.1", features = ["versioned-row-publication"] } +worktable = { version = "=1.0.0-beta.2", features = ["versioned-row-publication"] } ``` Generated point lookups use a strict backend-specific visibility contract by diff --git a/codegen/Cargo.toml b/codegen/Cargo.toml index 26b8e18b..1add3e47 100644 --- a/codegen/Cargo.toml +++ b/codegen/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "worktable_codegen" -version = "1.0.0-beta.1" +version = "1.0.0-beta.2" edition = "2024" license = "MIT" description = "Proc-macro companion crate for worktable: the worktable! macro and its derives." From 2812dcb291509f84c89c0d230cc8efc3ed119da6 Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 4 Aug 2026 01:17:10 +0700 Subject: [PATCH 4/5] docs: plan dirty-generation WTI persistence --- docs/wti-dirty-generation-persistence-plan.md | 186 ++++++++++++++++++ 1 file changed, 186 insertions(+) create mode 100644 docs/wti-dirty-generation-persistence-plan.md diff --git a/docs/wti-dirty-generation-persistence-plan.md b/docs/wti-dirty-generation-persistence-plan.md new file mode 100644 index 00000000..c881cac0 --- /dev/null +++ b/docs/wti-dirty-generation-persistence-plan.md @@ -0,0 +1,186 @@ +# WTI dirty-generation persistence + +**Status:** immediate follow-up design; not implemented. + +**Scope:** reduce caller-thread persistence overhead for WorkTablesIndex (WTI) without changing point-read behavior or weakening recovery. + +## Why investigate this + +WTI currently emits structural change events while mutating the in-memory tree. Encoding, batching, and file I/O already run in WorkTable's asynchronous persistence task, but the mutation thread must still construct the exact split/insert/remove event stream. + +A focused local ARM probe measured: + +| WTI mutation path | Median | +|---|---:| +| Direct in-memory insert/remove | 125–126 ns/op | +| Structural CDC insert/remove | 160–161 ns/op | +| Synchronous persistence-specific delta | approximately 35 ns/op (27.7%) | + +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. + +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 + +A Boolean has a lost-update race: + +1. a writer sets `dirty = true`; +2. the flusher snapshots the node; +3. a second writer mutates the node while `dirty` is already true; +4. the flusher writes its older snapshot and clears `dirty`; +5. the second mutation is no longer represented by either the disk image or the dirty bit. + +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 + +Separate crash authority from physical checkpoint maintenance: + +```text +caller mutation + ├─ commit in-memory WTI mutation + ├─ append/enqueue a small logical redo record + └─ advance dirty generation for affected topology + +background persistence task + ├─ persist logical redo in operation order + ├─ snapshot dirty topology while pinned + ├─ write native WTI checkpoint pages + └─ retire redo only after checkpoint durability +``` + +The logical redo log guarantees recoverability while physical WTI pages catch up. Dirty generations make physical mirroring coalescing and asynchronous: ten mutations to one node can become one checkpoint write. + +### Dirty-generation state + +The conceptual state is: + +```text +Clean(g) -> Dirty(g + 1) -> Flushing(g + 1) + ├─ unchanged -> Clean(g + 1) + └─ mutated -> Dirty(g + n) +``` + +This can be represented by one atomic generation plus state bits, or by separate generation and queued/flushing flags. The required properties are: + +- every committed mutation advances a generation; +- only the clean-to-dirty transition needs to enqueue work; +- a writer during `Flushing(g)` advances the generation and leaves the node dirty; +- the flusher records the generation before copying and rechecks it after copying; +- the flusher clears dirty state only with a compare/exchange against the generation it copied; +- failed compare/exchange requeues or retains the existing queued marker. + +Memory ordering must publish node bytes before the dirty generation becomes visible to the flusher and must prevent a successful clean transition from moving before snapshot completion. + +### Logical redo record + +The minimum WTI record is backend-native logical state, not a structural IndexSet event: + +```text +Set { sequence, key, link } +Remove { sequence, key } +``` + +Rollback/acknowledgement must remain gapless in the persistence analyzer. A complete frame has a checksum; an incomplete final frame is a torn tail. Records are encoded and written by the existing background task. + +A logical log does not promise that replay recreates the exact intermediate split history. It promises the correct WTI key/link state. A successful physical checkpoint still stores an exact native WTI topology for its checkpoint generation. + +If exact post-checkpoint topology must also be reproduced solely from redo, then the mutation linearization order must be captured globally. That cost must be measured before making it a requirement; logical correctness does not need it for commuting keys. + +## Structural changes are groups, not isolated nodes + +A split, merge, root replacement, or table-of-contents change can affect a parent, old node, new node, and root metadata. Incremental flushing must not publish an arbitrary mixture as one durable checkpoint. + +Two implementation levels are possible: + +### Level 1: logical WAL plus whole-index native checkpoints + +- Main thread creates only the logical record and advances one index generation. +- Background task periodically snapshots/reconstructs the complete WTI index. +- Atomic checkpoint replacement establishes one generation boundary. +- No per-node durable grouping is required. + +This is the recommended first implementation. It has the smallest correctness surface and directly tests how much of the approximately 35 ns structural-CDC delta is recoverable. + +### Level 2: incremental dirty-node checkpoints + +- Every structural mutation reports the complete affected-node set. +- Nodes are pinned against reclamation while copied. +- A checkpoint transaction or generation manifest commits the affected pages and root/TOC metadata together. +- Old pages remain reachable until the new generation manifest is durable. + +This can reduce checkpoint bandwidth for large indexes, but it is substantially more complex. It should follow, not precede, Level 1 evidence. + +## Reclamation and identity + +The background task must never follow a node pointer that can be freed or reused. Acceptable designs include: + +- queue an owning `Arc`/pin rather than a raw pointer; +- retain the node through WTI's existing epoch/reclamation mechanism until the flush generation finishes; +- queue a stable node ID and validate its generation before copying; +- use a quiescent whole-index snapshot for Level 1. + +A stable ID without a generation is insufficient because an allocator may reuse it for another node. + +## Durability and recovery ordering + +The intended recovery protocol is: + +1. validate and load the last complete native WTI checkpoint; +2. replay complete logical redo frames after that checkpoint generation; +3. reject complete checksum corruption and ignore only an incomplete final frame; +4. rebuild from authoritative data pages if the checkpoint/log contract cannot be validated; +5. truncate or rotate redo only after the replacement checkpoint and generation manifest are durable. + +The data/index ordering rule remains essential: a durable index record must not point to row bytes that were never made durable. If WorkTable continues to acknowledge operations before fsync, documentation and `wait_for_ops()` semantics must state that boundary precisely. + +S3 synchronization must copy a self-consistent checkpoint generation plus its required redo tail. Uploading files independently without a generation manifest can create a valid-but-mismatched restore set. + +## Hot-path target + +The mutation thread should do only: + +- the existing WTI mutation; +- one small logical record in an inline/stack-backed event container; +- one sequence allocation needed by rollback/ordering; +- a generation transition and at most one queue publication per clean-to-dirty transition. + +It should not: + +- serialize a node; +- allocate one heap `Vec` for every single event when an inline event suffices; +- wait for file I/O or checkpoint reconstruction; +- take a new lock on point reads; +- scan or diff a node after releasing the mutation's synchronization. + +## 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. +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. + +## Correctness gates + +- A writer racing a flusher cannot lose its dirty state. +- Multiple writes while queued coalesce without losing the latest generation. +- Split/merge/root transitions recover to a valid WTI with exactly the expected key/link set. +- Failed unique inserts and rolled-back multi-index operations leave no durable logical mutation. +- Delete/reinsert and vacuum link relocation replay correctly. +- Torn final redo is recoverable; complete corruption is a hard error or explicit rebuild path. +- Node reclamation cannot race a background snapshot. +- Restart, replay, further mutation, and a second restart remain correct. +- Memory-only WTI and WTI point reads have no new field access, branch, lock, or measurable regression. + +## Performance gates + +- Report absolute nanoseconds and percentages; the absolute caller cost is the engineering target. +- Compare direct WTI, current structural CDC, logical redo, and full persisted WorkTable operations. +- Run at least ten interleaved trials on quiet ARM hardware. +- Include 1 thread and representative contention, allocations/op, throughput, p50, p99, checkpoint bandwidth, recovery time, and WAL growth. +- Feature-gate the new protocol if any production-relevant workload regresses measurably. + +The first success criterion is to recover a meaningful portion of the approximately 35 ns synchronous structural-CDC delta without moving cost onto reads or weakening crash recovery. Incremental dirty-node flushing is justified only if whole-index background checkpoints then become the limiting cost. From f2e6b96cfb8a8dcb548ea68092433eac19f6aff0 Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 4 Aug 2026 01:39:05 +0700 Subject: [PATCH 5/5] build: use published ART backend crates --- Cargo.toml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 8e81fcea..36903ce7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -28,8 +28,8 @@ versioned-row-publication = ["worktable_codegen/versioned-row-publication"] [dependencies] async-trait = "0.1.89" -arctic = { package = "arctic-wt", git = "https://github.com/pathscale/arctic-wt", rev = "e13fc7df3c040f14ae66c1cb56b1bd0a3f6da3fc" } -congee = { package = "congee-wt", git = "https://github.com/pathscale/congee-wt", rev = "005bfb1968e781800176f2d7e465e6a1af630e1a" } +arctic = { package = "arctic-wt", version = "0.1.4" } +congee = { package = "congee-wt", version = "0.4.1" } convert_case = "0.6.0" crc32fast = "1.5.0" data_bucket = "=0.5.1"