From f9980737ad0c92175c7abf5f15e4b1ce41c720f9 Mon Sep 17 00:00:00 2001 From: Igor Malovitsa Date: Tue, 7 Jul 2026 20:15:05 +0000 Subject: [PATCH] Add ArenaCompactTree::merge_zipper_into_file: append-only merge into an ACT file Merges any zipper (Zipper + ZipperMoving + ZipperValues) into an existing on-disk ACT by appending only the new data and rewriting the root offset; existing bytes are never modified. Backward relative offsets let appended nodes reference old subtrees directly, so untouched subtrees are shared byte-for-byte. Because siblings must be stored contiguously, a node with any changed child gets its sibling run re-emitted as shallow copies whose children still point into the old arena; line nodes are split as needed, reusing the old line data when a whole segment survives. On value conflicts the zipper's value wins, and a merge that adds nothing leaves the file untouched. Works with PathMap read zippers and ACT zippers alike (ACT -> ACT merge), enabling incremental accumulation into one ACT file without rebuilding it. Also fixes get_val_at reading past a branch's last sibling when asked for a byte absent from the child mask (missing test_bit check; subtract-overflow panic in debug builds). Co-Authored-By: Claude Fable 5 --- src/arena_compact.rs | 596 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 596 insertions(+) diff --git a/src/arena_compact.rs b/src/arena_compact.rs index 69be596..50e9ff6 100644 --- a/src/arena_compact.rs +++ b/src/arena_compact.rs @@ -732,6 +732,9 @@ where Storage: AsRef<[u8]> if path.is_empty() { return node.value; } + if !node.bytemask.test_bit(path[0]) { + return None; + } let first_child = node.first_child?; let idx = node.bytemask.index_of(path[0]) as usize; cur_node = self.nth_node(first_child, idx).0; @@ -1201,6 +1204,439 @@ fn dump_arena_tree( Ok(arena) } +impl ArenaCompactTree { + /// Merge a zipper's trie into the existing ACT file at `path`, appending + /// only the new data and updating the root pointer. + /// + /// The old file contents are never rewritten: since all node references + /// are backward relative offsets, appended nodes can point into the + /// existing arena, and subtrees the zipper does not touch are shared + /// byte-for-byte. Only the nodes along changed paths (plus their sibling + /// runs, which the format requires to be contiguous) are appended, and + /// the root offset at byte 8 is rewritten to the new root. + /// + /// Where both tries hold a value on the same path, the zipper's value + /// wins. If the zipper adds nothing, the file is left untouched. + /// + /// Returns the updated tree, memory-mapped from the merged file. + /// + /// # Examples + /// ``` + /// use pathmap::{PathMap, arena_compact::ArenaCompactTree}; + /// # fn main() -> std::io::Result<()> { + /// let dir = tempfile::tempdir()?; + /// let file = dir.path().join("merge.act"); + /// let base = PathMap::from_iter([("apple", 1u64), ("banana", 2)]); + /// let act = ArenaCompactTree::from_zipper(base.read_zipper(), |&v| v); + /// std::fs::write(&file, act.get_data())?; + /// + /// let update = PathMap::from_iter([("apricot", 3u64), ("banana", 20)]); + /// let merged = ArenaCompactTree::merge_zipper_into_file( + /// &file, update.read_zipper(), |&v| v)?; + /// assert_eq!(merged.get_val_at("apple"), Some(1)); + /// assert_eq!(merged.get_val_at("apricot"), Some(3)); + /// assert_eq!(merged.get_val_at("banana"), Some(20)); // zipper value wins + /// # Ok(()) + /// # } + /// ``` + pub fn merge_zipper_into_file( + path: P, zipper: Z, map_val: F, + ) -> Result + where + Z: Zipper + ZipperMoving + ZipperValues, + F: Fn(&V) -> u64, + P: AsRef, + { + let file = OpenOptions::new().read(true).write(true).open(&path)?; + let old_map = unsafe { Mmap::map(&file) }?; + let old = old_map.as_ref(); + if old.len() < MAGIC_LENGTH + U64_SIZE + MAX_VARINT_SIZE + || &old[..MAGIC_LENGTH] != &COMPACT_TREE_MAGIC + { + return Err(std::io::Error::other("Invalid file magic")); + } + let root_buf: [u8; U64_SIZE] = old[MAGIC_LENGTH..][..U64_SIZE].try_into().unwrap(); + let root_id = NodeId(u64::from_le_bytes(root_buf)); + + let mut out = BufWriter::with_capacity(DUMPER_BUFFER_SIZE, file); + out.seek(SeekFrom::End(0))?; + let mut merger = ZipperMerger { + old, + out, + position: old.len() as u64, + zipper, + map_val, + counters: Counters::default(), + _marker: PhantomData, + }; + let (merged, changed) = merger.merge_node(root_id)?; + if changed { + let new_root = merger.push_merged(merged)?; + // Maintain the trailing-padding invariant for the appended data + merger.out.write_all(&[0; MAX_VARINT_SIZE - 1])?; + merger.out.seek(SeekFrom::Start(MAGIC_LENGTH as u64))?; + merger.out.write_all(&new_root.0.to_le_bytes())?; + } + let ZipperMerger { out, counters, .. } = merger; + let file = out.into_inner()?; + drop(old_map); + let memmap = unsafe { Mmap::map(&file) }?; + Ok(Self { + position: memmap.as_ref().len() as u64, + storage: memmap, + line_map: Default::default(), + lines: Default::default(), + hasher: Default::default(), + value: Cell::new(0), + counters, + }) + } +} + +/// A merged subtree: either an existing node reused as-is, or a freshly +/// built node (whose descendants have already been appended). +enum Merged { + /// Reuse the old subtree at this id. When placed into a sibling run the + /// node itself is shallow-copied (the format requires siblings to be + /// contiguous), but everything below it stays shared with the old file. + Reuse(NodeId), + /// A new node to append; references old and new data by absolute id. + Fresh(Node), +} + +/// State for [ArenaCompactTree::merge_zipper_into_file]: reads the old arena +/// through `old`, appends through `out`, and walks `zipper` in lockstep with +/// the old trie. Every merge method returns the zipper to its entry position. +struct ZipperMerger<'a, V, Z, F> { + old: &'a [u8], + out: BufWriter, + /// Append position: the absolute offset the next object will get + position: u64, + zipper: Z, + map_val: F, + counters: Counters, + _marker: PhantomData u64>, +} + +impl<'a, V, Z, F> ZipperMerger<'a, V, Z, F> +where + Z: Zipper + ZipperMoving + ZipperValues, + F: Fn(&V) -> u64, +{ + fn old_node(&self, id: NodeId) -> (Node, usize) { + read_node(&self.old[id.0 as usize..], id) + } + + fn old_line(&self, id: LineId) -> &'a [u8] { + let old: &'a [u8] = self.old; + let start = &old[id.0 as usize..]; + let (len, off) = read_varint_u64(start); + &start[off..off + len as usize] + } + + /// The zipper's value at its current focus, mapped to `u64` + fn z_val(&self) -> Option { + self.zipper.val().map(|v| (self.map_val)(v)) + } + + /// Append a node; all ids it references must be below `self.position` + fn push_fresh(&mut self, node: &Node) -> Result { + let node_id = NodeId(self.position); + let mut cursor = std::io::Cursor::new([0; MAX_BRANCH_NODE_SIZE]); + match node { + Node::Branch(branch) => { + ArenaCompactTree::>::write_node( + &mut cursor, branch, node_id, &mut self.counters)?; + } + Node::Line(line) => { + ArenaCompactTree::>::write_line( + &mut cursor, line, node_id, &mut self.counters)?; + } + } + let len = cursor.position(); + self.out.write_all(&cursor.get_ref()[..len as usize])?; + self.position += len; + Ok(node_id) + } + + /// Append a merged child as part of a sibling run. `Reuse` becomes a + /// shallow copy: same value/mask/line, child pointers into the old file. + fn push_merged(&mut self, merged: Merged) -> Result { + match merged { + Merged::Fresh(node) => self.push_fresh(&node), + Merged::Reuse(id) => { + let (node, _) = self.old_node(id); + self.push_fresh(&node) + } + } + } + + /// Append line data, returning its id + fn add_line_data(&mut self, data: &[u8]) -> Result { + debug_assert!(!data.is_empty()); + let line_id = LineId(self.position); + let lenlen = push_varint_u64(&mut self.out, data.len() as u64)?; + self.out.write_all(data)?; + self.position += (lenlen + data.len()) as u64; + Ok(line_id) + } + + /// Merge the zipper's current position with the old node `id`. + /// Returns the merged subtree and whether anything changed; when nothing + /// changed, nothing has been appended and the old node is reused. + fn merge_node(&mut self, id: NodeId) -> Result<(Merged, bool), std::io::Error> { + match self.old_node(id).0 { + Node::Branch(branch) => self.merge_branch(id, branch), + Node::Line(line) => match self.merge_line_from(&line, 0)? { + Some(merged) => Ok((merged, true)), + None => Ok((Merged::Reuse(id), false)), + }, + } + } + + fn merge_branch( + &mut self, id: NodeId, branch: NodeBranch, + ) -> Result<(Merged, bool), std::io::Error> { + let z_mask = self.zipper.child_mask(); + let value = self.z_val().or(branch.value); + let mut changed = value != branch.value; + + // Ids of the old children (siblings are stored sequentially) + let mut old_kids = Vec::with_capacity(branch.bytemask.count_bits()); + if let Some(first) = branch.first_child { + let mut cur = first; + for _ in 0..branch.bytemask.count_bits() { + old_kids.push(cur); + let (_, len) = self.old_node(cur); + cur = NodeId(cur.0 + len as u64); + } + } + + let union = branch.bytemask.or(&z_mask); + let mut children: Vec = Vec::with_capacity(union.count_bits()); + let mut old_idx = 0; + for byte in union.iter() { + let in_old = branch.bytemask.test_bit(byte); + let in_new = z_mask.test_bit(byte); + if in_old { + let child_id = old_kids[old_idx]; + old_idx += 1; + if in_new { + self.zipper.descend_to_byte(byte); + let (merged, child_changed) = self.merge_node(child_id)?; + self.zipper.ascend_byte(); + changed |= child_changed; + children.push(merged); + } else { + children.push(Merged::Reuse(child_id)); + } + } else { + changed = true; + self.zipper.descend_to_byte(byte); + let node = self.fresh_subtree()?; + self.zipper.ascend_byte(); + children.push(Merged::Fresh(node)); + } + } + if !changed { + return Ok((Merged::Reuse(id), false)); + } + let mut first_child = None; + for child in children { + let child_id = self.push_merged(child)?; + first_child = first_child.or(Some(child_id)); + } + let node = NodeBranch { bytemask: union, first_child, value }; + Ok((Merged::Fresh(Node::Branch(node)), true)) + } + + /// Merge the zipper (positioned `k` bytes into `line`'s segment) with the + /// remainder of the line. Returns `None` when the zipper adds nothing + /// (in which case nothing has been appended). + fn merge_line_from( + &mut self, line: &NodeLine, k: usize, + ) -> Result, std::io::Error> { + let data = self.old_line(line.path); + let len = data.len(); + // Scan forward while the zipper follows the segment exactly + let mut j = k; + while j < len && self.z_val().is_none() && { + let z_mask = self.zipper.child_mask(); + z_mask.count_bits() == 1 && z_mask.test_bit(data[j]) + } { + self.zipper.descend_to_byte(data[j]); + j += 1; + } + + let inner: Option = if j == len { + // Reached the end of the segment + if let Some(child) = line.child { + let (merged, child_changed) = self.merge_node(child)?; + child_changed.then_some(merged) + } else { + // Old leaf; the zipper may update the value or extend below + let z_mask = self.zipper.child_mask(); + let value = self.z_val().or(line.value); + if value == line.value && z_mask.is_empty_mask() { + None + } else { + let mut fresh = Vec::with_capacity(z_mask.count_bits()); + for byte in z_mask.iter() { + self.zipper.descend_to_byte(byte); + fresh.push(self.fresh_subtree()?); + self.zipper.ascend_byte(); + } + let mut first_child = None; + for node in &fresh { + let child_id = self.push_fresh(node)?; + first_child = first_child.or(Some(child_id)); + } + let node = NodeBranch { bytemask: z_mask, first_child, value }; + Some(Merged::Fresh(Node::Branch(node))) + } + } + } else { + // The zipper diverges at segment offset `j` + let b_old = data[j]; + let z_mask = self.zipper.child_mask(); + let z_val = self.z_val(); + let matched = z_mask.test_bit(b_old); + let cont: Option = if matched { + self.zipper.descend_to_byte(b_old); + let merged = self.merge_line_from(line, j + 1)?; + self.zipper.ascend_byte(); + merged + } else { + None + }; + let extras = z_mask.count_bits() - (matched as usize); + if z_val.is_none() && extras == 0 && cont.is_none() { + None + } else { + // Build a branch at offset `j`: the old segment continuation + // plus whatever the zipper adds here + let mut cont = Some(match cont { + Some(merged) => merged, + None => self.tail_child(line, data, j + 1)?, + }); + let mut mask = z_mask; + mask.set_bit(b_old); + let mut children: Vec = Vec::with_capacity(mask.count_bits()); + for byte in mask.iter() { + if byte == b_old { + children.push(cont.take().unwrap()); + } else { + self.zipper.descend_to_byte(byte); + let node = self.fresh_subtree()?; + self.zipper.ascend_byte(); + children.push(Merged::Fresh(node)); + } + } + let mut first_child = None; + for child in children { + let child_id = self.push_merged(child)?; + first_child = first_child.or(Some(child_id)); + } + let node = NodeBranch { bytemask: mask, first_child, value: z_val }; + Some(Merged::Fresh(Node::Branch(node))) + } + }; + + self.zipper.ascend(j - k); + let Some(inner) = inner else { return Ok(None) }; + if j == k { + return Ok(Some(inner)); + } + // Wrap the merged node in a line for the matched prefix data[k..j]; + // reuse the old line data when the whole segment matched. + let path_id = if k == 0 && j == len { + line.path + } else { + self.add_line_data(&data[k..j])? + }; + let node = match inner { + Merged::Fresh(Node::Branch(branch)) if branch.bytemask.is_empty_mask() => { + Node::Line(NodeLine { path: path_id, value: branch.value, child: None }) + } + inner => { + let child_id = self.push_merged(inner)?; + Node::Line(NodeLine { path: path_id, value: None, child: Some(child_id) }) + } + }; + Ok(Some(Merged::Fresh(node))) + } + + /// The old, unmerged continuation of `line` from segment offset `k`, + /// packaged so it can sit in a new sibling run + fn tail_child( + &mut self, line: &NodeLine, data: &[u8], k: usize, + ) -> Result { + if k == data.len() { + Ok(match line.child { + Some(child) => Merged::Reuse(child), + None => Merged::Fresh(Node::Branch(NodeBranch { + bytemask: ByteMask::EMPTY, + first_child: None, + value: line.value, + })), + }) + } else { + let path_id = self.add_line_data(&data[k..])?; + Ok(Merged::Fresh(Node::Line(NodeLine { + path: path_id, + value: if line.child.is_none() { line.value } else { None }, + child: line.child, + }))) + } + } + + /// Serialize the zipper's current subtree (absent from the old trie), + /// compressing single-child chains into line nodes. Descendants are + /// appended; the returned node is pushed by the caller's sibling run. + fn fresh_subtree(&mut self) -> Result { + let mut segment: Vec = Vec::new(); + loop { + if self.z_val().is_some() { + break; + } + let mask = self.zipper.child_mask(); + if mask.count_bits() != 1 { + break; + } + let byte = mask.iter().next().unwrap(); + segment.push(byte); + self.zipper.descend_to_byte(byte); + } + let value = self.z_val(); + let mask = self.zipper.child_mask(); + let mut children = Vec::with_capacity(mask.count_bits()); + for byte in mask.iter() { + self.zipper.descend_to_byte(byte); + children.push(self.fresh_subtree()?); + self.zipper.ascend_byte(); + } + let mut first_child = None; + for child in &children { + let child_id = self.push_fresh(child)?; + first_child = first_child.or(Some(child_id)); + } + let branch = NodeBranch { bytemask: mask, first_child, value }; + let node = if segment.is_empty() { + Node::Branch(branch) + } else { + let path_id = self.add_line_data(&segment)?; + if mask.is_empty_mask() { + Node::Line(NodeLine { path: path_id, value, child: None }) + } else { + let child_id = self.push_fresh(&Node::Branch(branch))?; + Node::Line(NodeLine { path: path_id, value: None, child: Some(child_id) }) + } + }; + self.zipper.ascend(segment.len()); + Ok(node) + } +} + /* fn tree_to_btm(tree: &ArenaCompactTree) -> PathMap<()> { struct PathIdx(NodeId, usize) @@ -2257,6 +2693,166 @@ mod tests { assert_eq!(btm_value, act_value); } + fn build_act_file(path: &std::path::Path, items: &[(&str, u64)]) { + let btm = PathMap::from_iter(items.iter().map(|&(k, v)| (k, v))); + let act = ArenaCompactTree::from_zipper(btm.read_zipper(), |&v| v); + std::fs::write(path, act.get_data()).unwrap(); + } + + /// Assert `act` holds exactly `items` (no duplicates in `items`), + /// both by point lookups and by an ordered walk. + fn assert_act_content(act: &super::ACTMmap, items: &[(&str, u64)]) { + for &(k, v) in items { + assert_eq!(act.get_val_at(k), Some(v), "key {k}"); + } + let btm = PathMap::from_iter(items.iter().map(|&(k, v)| (k, v))); + let mut bz = btm.read_zipper(); + let mut az = act.read_zipper_u64(); + loop { + let more_b = bz.to_next_val(); + let more_a = az.to_next_val(); + assert_eq!(more_b, more_a, "walks end together"); + assert_eq!(bz.path(), az.path()); + assert_eq!(bz.val().copied(), az.val().copied()); + if !more_a { + break; + } + } + } + + #[test] + fn test_act_merge_zipper_into_file() { + use super::MAGIC_LENGTH; + let dir = tempfile::tempdir().unwrap(); + let file = dir.path().join("merge.act"); + let base: &[(&str, u64)] = &[ + ("arrow", 1), ("bow", 2), ("roman", 3), ("romane", 4), + ("rubicon", 5), ("aaaaaaaaaaaaaaaaaaaaaaaaaaaaab", 6), + ]; + let add: &[(&str, u64)] = &[ + ("bow", 20), // value conflict -> zipper wins + ("rom", 7), // value inside a line segment + ("romanus", 8), // splits the line under "roman" + ("rub", 9), ("rubble", 10), // diverges inside the "rubicon" line + ("zebra", 11), // fresh subtree at the root + ("aaaaaaaaaaaaaaaaaaaaaaaaaaaaab", 6), // identical entry (unchanged subtree) + ("arrowhead", 12), // extends an existing leaf + ]; + build_act_file(&file, base); + let before = std::fs::read(&file).unwrap(); + + let add_map = PathMap::from_iter(add.iter().map(|&(k, v)| (k, v))); + let merged = ArenaCompactTree::merge_zipper_into_file( + &file, add_map.read_zipper(), |&v| v).unwrap(); + + // Append-only: everything except the root pointer is byte-identical + let after = std::fs::read(&file).unwrap(); + assert!(after.len() > before.len(), "merge must append"); + assert_eq!(&after[..MAGIC_LENGTH], &before[..MAGIC_LENGTH]); + assert_eq!(&after[MAGIC_LENGTH + 8..before.len()], &before[MAGIC_LENGTH + 8..]); + + assert_act_content(&merged, &[ + ("arrow", 1), ("arrowhead", 12), ("bow", 20), ("rom", 7), + ("roman", 3), ("romane", 4), ("romanus", 8), ("rub", 9), + ("rubble", 10), ("rubicon", 5), ("zebra", 11), + ("aaaaaaaaaaaaaaaaaaaaaaaaaaaaab", 6), + ]); + for absent in ["row", "arrowh", "arrowheads", "zebr", "zebras", "romanu"] { + assert_eq!(merged.get_val_at(absent), None, "absent {absent}"); + } + } + + #[test] + fn test_act_merge_noop() { + let dir = tempfile::tempdir().unwrap(); + let file = dir.path().join("noop.act"); + let base: &[(&str, u64)] = &[ + ("arrow", 1), ("bow", 2), ("roman", 3), ("romane", 4), ("rubicon", 5), + ]; + build_act_file(&file, base); + let before = std::fs::read(&file).unwrap(); + + // A subset with identical values adds nothing + let subset = PathMap::from_iter([("bow", 2u64), ("romane", 4)]); + let merged = ArenaCompactTree::merge_zipper_into_file( + &file, subset.read_zipper(), |&v| v).unwrap(); + assert_eq!(std::fs::read(&file).unwrap(), before, "no-op merge must not touch the file"); + assert_act_content(&merged, base); + } + + #[test] + fn test_act_merge_wide_branch() { + // Merging into a >=32-child branch exercises the mask encoding + let dir = tempfile::tempdir().unwrap(); + let file = dir.path().join("wide.act"); + let evens: Vec> = (0..=254u16).step_by(2).map(|b| vec![b as u8]).collect(); + let odds: Vec> = (1..=255u16).step_by(2).map(|b| vec![b as u8]).collect(); + let base_map = PathMap::from_iter(evens.iter().map(|k| (k, 1u64))); + let act = ArenaCompactTree::from_zipper(base_map.read_zipper(), |&v| v); + std::fs::write(&file, act.get_data()).unwrap(); + + let add_map = PathMap::from_iter(odds.iter().map(|k| (k, 2u64))); + let merged = ArenaCompactTree::merge_zipper_into_file( + &file, add_map.read_zipper(), |&v| v).unwrap(); + for b in 0..=255u8 { + assert_eq!(merged.get_val_at([b]), Some(1 + (b & 1) as u64), "byte {b}"); + } + assert_eq!(merged.get_val_at([0, 0]), None); + } + + #[test] + fn test_act_merge_repeated_and_act_source() { + // Several merge waves over an LCG key soup; the last wave merges from + // an ACT zipper instead of a PathMap zipper. + let dir = tempfile::tempdir().unwrap(); + let file = dir.path().join("waves.act"); + let mut state: u64 = 0x1234_5678_9abc_def0; + let mut next = move || { + state = state.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407); + state >> 33 + }; + let mut make_wave = |n: usize| -> Vec<(String, u64)> { + (0..n).map(|_| { + let r = next(); + // small alphabet + variable length -> shared prefixes and line splits + let len = 1 + (r % 12) as usize; + let key: String = (0..len) + .map(|i| (b'a' + ((r >> (i * 2)) & 0x3) as u8) as char) + .collect(); + (key, r % 1000) + }).collect() + }; + + let wave0 = make_wave(200); + let base_map: PathMap = wave0.iter().map(|(k, v)| (k, *v)).collect(); + let act = ArenaCompactTree::from_zipper(base_map.read_zipper(), |&v| v); + std::fs::write(&file, act.get_data()).unwrap(); + + let mut expect: std::collections::HashMap = + wave0.into_iter().collect(); + for wave_idx in 0..3 { + let wave = make_wave(300); + let wave_map: PathMap = wave.iter().map(|(k, v)| (k, *v)).collect(); + let before = std::fs::read(&file).unwrap(); + let merged = if wave_idx < 2 { + ArenaCompactTree::merge_zipper_into_file( + &file, wave_map.read_zipper(), |&v| v).unwrap() + } else { + // final wave: merge from an ACT zipper (ACT -> ACT merge) + let wave_act = ArenaCompactTree::from_zipper(wave_map.read_zipper(), |&v| v); + ArenaCompactTree::merge_zipper_into_file( + &file, wave_act.read_zipper_u64(), |&v| v).unwrap() + }; + // PathMap::from_iter and HashMap::extend agree: later entries win + expect.extend(wave.into_iter()); + let after = std::fs::read(&file).unwrap(); + assert_eq!(&after[16..before.len()], &before[16..], "append-only violated"); + + let items: Vec<(&str, u64)> = expect.iter().map(|(k, v)| (k.as_str(), *v)).collect(); + assert_act_content(&merged, &items); + } + } + #[test] fn test_act_mmap() -> Result<(), std::io::Error> { use tempfile::NamedTempFile;