qualia_core_db/q42/volume/
publish.rs

1//! Segment rollover and append-as-new-segment for logical Q42 volumes.
2
3use std::collections::HashMap;
4use std::io;
5use std::path::{Path, PathBuf};
6
7use super::super::{
8    write_volume_root, Q42Volume, Q42VolumeManifest, StreamingQ42VolumeWriter,
9};
10use crate::NQuin;
11
12/// Default child-segment cap (512 MiB). Includes header, lexicon, indexes, data.
13pub const DEFAULT_SEGMENT_MAX_BYTES: u64 = 512 * 1024 * 1024;
14
15/// Writes object-sorted SuperBlocks into successive `.q42` children and
16/// publishes an atomic root catalog when finished.
17pub struct Q42RolloverPublisher {
18    dir: PathBuf,
19    stem: String,
20    lex: HashMap<u64, String>,
21    max_bytes: u64,
22    writer: StreamingQ42VolumeWriter,
23    child_paths: Vec<PathBuf>,
24    next_index: u32,
25}
26
27impl Q42RolloverPublisher {
28    pub fn new(dir: impl Into<PathBuf>, stem: impl Into<String>, lex: HashMap<u64, String>) -> io::Result<Self> {
29        Self::with_limit(dir, stem, lex, DEFAULT_SEGMENT_MAX_BYTES)
30    }
31
32    pub fn with_limit(
33        dir: impl Into<PathBuf>,
34        stem: impl Into<String>,
35        lex: HashMap<u64, String>,
36        max_bytes: u64,
37    ) -> io::Result<Self> {
38        if max_bytes < 8 * 1024 {
39            return Err(io::Error::new(
40                io::ErrorKind::InvalidInput,
41                "Q42 segment cap is too small",
42            ));
43        }
44        let dir = dir.into();
45        std::fs::create_dir_all(&dir)?;
46        Ok(Self {
47            writer: StreamingQ42VolumeWriter::new(&lex)?,
48            lex,
49            dir,
50            stem: stem.into(),
51            max_bytes,
52            child_paths: Vec::new(),
53            next_index: 0,
54        })
55    }
56
57    pub fn push_block(&mut self, seq_id: u64, quins: &[NQuin]) -> io::Result<()> {
58        let projected = self.writer.maximum_final_length_after_next_block()?;
59        if self.writer.block_count() > 0 && projected > self.max_bytes {
60            self.roll()?;
61        }
62        self.writer.push_block(seq_id, quins)
63    }
64
65    fn roll(&mut self) -> io::Result<()> {
66        let path = self.child_path(self.next_index);
67        let writer = std::mem::replace(
68            &mut self.writer,
69            StreamingQ42VolumeWriter::new(&self.lex)?,
70        );
71        writer.finish(&path)?;
72        self.child_paths.push(path);
73        self.next_index += 1;
74        Ok(())
75    }
76
77    fn child_path(&self, index: u32) -> PathBuf {
78        self.dir.join(format!("{}-{:04}.q42", self.stem, index))
79    }
80
81    /// Finish the open child and write `stem-root.q42` naming every child.
82    pub fn finish(mut self) -> io::Result<PathBuf> {
83        if self.writer.block_count() > 0 || self.child_paths.is_empty() {
84            let path = self.child_path(self.next_index);
85            self.writer.finish(&path)?;
86            self.child_paths.push(path);
87        }
88        let mut segments = Vec::new();
89        for path in &self.child_paths {
90            let name = path
91                .file_name()
92                .and_then(|n| n.to_str())
93                .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "Q42 child name"))?
94                .to_string();
95            segments.push(Q42VolumeManifest::segment_from_file(path, name)?);
96        }
97        let root = self.dir.join(format!("{}-root.q42", self.stem));
98        write_volume_root(
99            &root,
100            &Q42VolumeManifest {
101                generation: 1,
102                segments,
103                lexicon_segments: Vec::new(),
104            },
105        )?;
106        Ok(root)
107    }
108}
109
110/// Append a finished child segment to an existing root by publishing a new
111/// generation (`stem-root.q42` overwritten atomically via temp + rename).
112pub fn append_segment_to_root(root_path: &Path, new_child: &Path) -> io::Result<()> {
113    let parent = root_path
114        .parent()
115        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "Q42 root has no parent"))?;
116    let volume = Q42Volume::open(root_path)?;
117    let mut manifest = volume.volume_manifest()?.ok_or_else(|| {
118        io::Error::new(
119            io::ErrorKind::InvalidInput,
120            "append requires a volume-root catalog",
121        )
122    })?;
123    let name = new_child
124        .file_name()
125        .and_then(|n| n.to_str())
126        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "child name"))?
127        .to_string();
128    manifest
129        .segments
130        .push(Q42VolumeManifest::segment_from_file(new_child, name)?);
131    manifest.generation = manifest.generation.saturating_add(1);
132    let tmp = parent.join(format!(
133        ".{}.tmp",
134        root_path.file_name().and_then(|n| n.to_str()).unwrap_or("root.q42")
135    ));
136    write_volume_root(&tmp, &manifest)?;
137    std::fs::rename(&tmp, root_path)?;
138    Ok(())
139}
140
141#[cfg(test)]
142mod tests {
143    use super::*;
144    use crate::NQuin;
145
146    fn quin(object: u64) -> NQuin {
147        NQuin {
148            subject: object,
149            predicate: 1,
150            object,
151            context: 0,
152            metadata: 0,
153            parity: object ^ 1 ^ object,
154        }
155    }
156
157    #[test]
158    fn rolls_when_the_next_block_would_exceed_the_cap() {
159        let dir = tempfile::TempDir::new().unwrap();
160        let mut lex = HashMap::new();
161        lex.insert(1, "p".into());
162        let mut pubr = Q42RolloverPublisher::with_limit(dir.path(), "set", lex, 16 * 1024).unwrap();
163        for i in 0..8u64 {
164            pubr.push_block(i, &[quin(i + 1)]).unwrap();
165        }
166        let root = pubr.finish().unwrap();
167        let volume = Q42Volume::open(&root).unwrap();
168        let manifest = volume.volume_manifest().unwrap().unwrap();
169        assert!(
170            manifest.segments.len() >= 2,
171            "expected rollover, got {} children",
172            manifest.segments.len()
173        );
174    }
175
176    #[test]
177    fn append_publishes_a_new_generation() {
178        let dir = tempfile::TempDir::new().unwrap();
179        let mut lex = HashMap::new();
180        lex.insert(1, "p".into());
181        let mut pubr = Q42RolloverPublisher::with_limit(dir.path(), "gen", lex.clone(), 1024 * 1024)
182            .unwrap();
183        pubr.push_block(0, &[quin(1)]).unwrap();
184        let root = pubr.finish().unwrap();
185        let extra = dir.path().join("extra.q42");
186        let mut w = StreamingQ42VolumeWriter::new(&lex).unwrap();
187        w.push_block(1, &[quin(2)]).unwrap();
188        w.finish(&extra).unwrap();
189        append_segment_to_root(&root, &extra).unwrap();
190        let volume = Q42Volume::open(&root).unwrap();
191        let manifest = volume.volume_manifest().unwrap().unwrap();
192        assert_eq!(manifest.generation, 2);
193        assert_eq!(manifest.segments.len(), 2);
194    }
195}
qualia_core_db/q42/volume/publish.rs

qualia_core_db/q42/volume/
publish.rs
