diff --git a/src/index/unsized_node.rs b/src/index/unsized_node.rs index b569b5ab..cb993e03 100644 --- a/src/index/unsized_node.rs +++ b/src/index/unsized_node.rs @@ -113,7 +113,27 @@ where } fn need_to_split(&self, _: usize, value: &T) -> bool { - let final_length = self.length + value.aligned_size(); + let value_size = value.aligned_size(); + let current_node_id_size = self.max().map(SizeMeasurable::aligned_size).unwrap_or(0); + let next_node_id_size = self + .max() + .map(|current_max| { + if value > current_max { + value_size + } else { + current_max.aligned_size() + } + }) + .unwrap_or(value_size); + // `length` deliberately retains removed bytes until a rebuild. That + // makes this estimate conservative before reload; persisted pages are + // compacted separately when physical fragmentation survives reload. + let final_length = self + .length + .saturating_sub(current_node_id_size) + .saturating_add(next_node_id_size) + .saturating_add(value_size) + .saturating_add(UnsizedIndexPageUtility::::slots_value_size()); final_length >= self.length_capacity && self.inner.len() > 1 } diff --git a/src/persistence/space/index/unsized_.rs b/src/persistence/space/index/unsized_.rs index 99a2b134..42070a4b 100644 --- a/src/persistence/space/index/unsized_.rs +++ b/src/persistence/space/index/unsized_.rs @@ -7,7 +7,7 @@ use std::sync::atomic::{AtomicU32, Ordering}; use data_bucket::page::PageId; use data_bucket::{ GeneralHeader, GeneralPage, IndexPageUtility, IndexValue, Link, PageType, SizeMeasurable, SpaceId, SpaceInfoPage, - UnsizedIndexPage, VariableSizeMeasurable, parse_page, persist_page, persist_pages_batch, + UnsizedIndexPage, UnsizedIndexPageUtility, VariableSizeMeasurable, parse_page, persist_page, persist_pages_batch, }; use eyre::eyre; use indexset::cdc::change::ChangeEvent; @@ -59,6 +59,32 @@ where + Debug + for<'a> rkyv::bytecheck::CheckBytes>, { + fn compact_page_if_needed(page: &mut UnsizedIndexPage) -> eyre::Result<()> { + let persisted_size = + UnsizedIndexPageUtility::::persisted_size(page.slots_size as usize, page.node_id_size as usize) + + page.last_value_offset as usize; + if persisted_size <= DATA_LENGTH as usize { + return Ok(()); + } + + // Removed variable-width entries leave holes at the page tail. After + // reload the in-memory node contains only live values, so it cannot + // see that physical fragmentation. Compact before the growing slot + // directory can overlap the next tail value. + // An empty page cannot pass the occupancy guard above, which also + // protects `UnsizedIndexPage::rebuild` from its non-empty assumption. + page.rebuild(); + let compacted_size = + UnsizedIndexPageUtility::::persisted_size(page.slots_size as usize, page.node_id_size as usize) + + page.last_value_offset as usize; + if compacted_size > DATA_LENGTH as usize { + return Err(eyre!( + "unsized index page requires {compacted_size} bytes after compaction, but its capacity is {DATA_LENGTH}" + )); + } + Ok(()) + } + pub async fn new>(index_file_path: S, space_id: SpaceId, version: u32) -> eyre::Result { let space_index = SpaceIndex::::new(index_file_path, space_id, version).await?; Ok(Self { @@ -138,6 +164,32 @@ where key: value.key.clone(), link: value.value, }; + let value_size = index_value.aligned_size(); + let future_node_id_size = if node_id.key < value.key { + value_size + } else { + utility.node_id_size as usize + }; + let future_utility_size = + UnsizedIndexPageUtility::::persisted_size(utility.slots_size as usize + 1, future_node_id_size); + if future_utility_size + utility.last_value_offset as usize + value_size > DATA_LENGTH as usize { + let mut page = + parse_page::, DATA_LENGTH>(&mut self.index_file, page_id.into()) + .await?; + page.inner.apply_change_event(ChangeEvent::InsertAt { + // The page mutation ignores event ids; this synthetic event + // exists only to reuse the same insertion accounting. + event_id: 0.into(), + max_value: node_id, + value, + index, + })?; + Self::compact_page_if_needed(&mut page.inner)?; + let changed_node_id = + (page.inner.node_id.key != utility.node_id.key).then(|| Pair::from(page.inner.node_id.clone())); + persist_page(&mut page, &mut self.index_file).await?; + return Ok(changed_node_id); + } let previous_offset = utility.last_value_offset; let value_offset = UnsizedIndexPage::::persist_value( &mut self.index_file, @@ -449,6 +501,10 @@ where } } + for page in pages.values_mut() { + Self::compact_page_if_needed(&mut page.inner)?; + } + self.table_of_contents.persist(&mut self.index_file).await?; persist_pages_batch(pages.values().cloned().collect(), &mut self.index_file).await?; // The batch's last page write is a buffered `write_all`; flush so the diff --git a/tests/persistence/sync/string_secondary_index.rs b/tests/persistence/sync/string_secondary_index.rs index ff2bb1e0..e7266298 100644 --- a/tests/persistence/sync/string_secondary_index.rs +++ b/tests/persistence/sync/string_secondary_index.rs @@ -29,6 +29,101 @@ worktable! ( } ); +worktable! ( + name: FragmentedStringSecondary, + persist: true, + columns: { + id: u64 primary_key autoincrement, + project_id: String, + }, + indexes: { + project_idx: project_id, + }, +); + +#[test] +fn fragmented_string_index_compacts_after_restart_before_appending() { + let path = "tests/data/unsized_secondary_sync/fragmented_restart"; + let config = DiskConfig::new_with_table_name( + path, + FragmentedStringSecondaryWorkTable::name_snake_case(), + FragmentedStringSecondaryWorkTable::version(), + ); + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_io() + .enable_time() + .build() + .unwrap(); + + runtime.block_on(async { + remove_dir_if_exists(path.to_string()).await; + let project_id = "proj-fdfb706b-72bc-4b0c-a96e-6b6235be6fb4".to_string(); + let mut inserted_ids = Vec::new(); + + { + let engine = FragmentedStringSecondaryPersistenceEngine::new(config.clone()) + .await + .unwrap(); + let table = FragmentedStringSecondaryWorkTable::load(engine).await.unwrap(); + for _ in 0..211 { + let row = FragmentedStringSecondaryRow { + id: table.get_next_pk().0, + project_id: project_id.clone(), + }; + inserted_ids.push(row.id); + table.insert(row).unwrap(); + } + table.wait_for_ops().await.unwrap(); + for id in inserted_ids.iter().take(37).copied() { + table.delete(id).await.unwrap(); + } + table.wait_for_ops().await.unwrap(); + } + + { + let engine = FragmentedStringSecondaryPersistenceEngine::new(config.clone()) + .await + .unwrap(); + let table = FragmentedStringSecondaryWorkTable::load(engine).await.unwrap(); + for _ in 0..20 { + let row = FragmentedStringSecondaryRow { + id: table.get_next_pk().0, + project_id: project_id.clone(), + }; + table.insert(row).unwrap(); + } + table.wait_for_ops().await.unwrap(); + } + + { + let engine = FragmentedStringSecondaryPersistenceEngine::new(config).await.unwrap(); + let table = FragmentedStringSecondaryWorkTable::load(engine).await.unwrap(); + assert_eq!(table.select_all().execute().unwrap().len(), 194); + assert_eq!(table.select_by_project_id(project_id).execute().unwrap().len(), 194); + } + + let index_path = format!("{path}/fragmented_string_secondary/project_idx.wt.idx"); + let mut index_file = tokio::fs::File::open(index_path).await.unwrap(); + let page = parse_page::, { INNER_PAGE_SIZE as u32 }>( + &mut index_file, + 2, + ) + .await + .unwrap(); + let utility_size = worktable::data_bucket::UnsizedIndexPageUtility::::persisted_size( + page.inner.slots_size as usize, + page.inner.node_id_size as usize, + ); + assert!( + utility_size + page.inner.last_value_offset as usize <= INNER_PAGE_SIZE, + "slot directory must not overlap values stored at the page tail" + ); + + remove_dir_if_exists(path.to_string()).await; + }); +} + #[test] fn test_space_insert_sync() { let config = DiskConfig::new_with_table_name(