qualia_core_db/q42/volume/
compact.rs

1//! Compact a logical Q42 volume set (or a single-file volume) into a new generation.
2//!
3//! SuperBlocks are streamed in object order through a caller-owned 40960-byte
4//! buffer. The hot per-Quin loop only copies into a stack `[NQuin; QUINS_PER_BLOCK]`.
5//! Integrator wires `mod compact;` in `volume/mod.rs`.
6
7use std::collections::HashMap;
8use std::io;
9use std::path::{Path, PathBuf};
10
11use super::super::{
12    write_volume_root_with_lex, Q42Volume, Q42VolumeManifest, StreamingQ42VolumeWriter,
13    FLAG_PERMISSIVE_COMMONS, FLAG_SANCTUARY, QUIN_SIZE, SUPERBLOCK_HEADER, SUPERBLOCK_SIZE,
14};
15use super::manifest::Q42VolumeSet;
16use super::publication::quin_requires_sanctuary;
17use super::publish::{Q42RolloverPublisher, DEFAULT_SEGMENT_MAX_BYTES};
18use crate::{NQuin, QUINS_PER_BLOCK};
19
20fn invalid(message: impl Into<String>) -> io::Error {
21    io::Error::new(io::ErrorKind::InvalidInput, message.into())
22}
23
24/// Compact `root` into `out_dir`.
25///
26/// A logical volume-set root is rewritten through [`Q42RolloverPublisher`]
27/// (or [`StreamingQ42VolumeWriter`] when every input may remain Permissive
28/// Commons). A single-file volume is rewritten through the streaming writer
29/// so the result carries PIDX + FIDX + BIDX. The new catalog is published
30/// with the append temp+rename pattern.
31pub fn compact_volume_set(root: &Path, out_dir: &Path) -> io::Result<PathBuf> {
32    std::fs::create_dir_all(out_dir)?;
33    let volume = Q42Volume::open(root)?;
34    if volume.volume_manifest()?.is_some() {
35        compact_logical_set(root, out_dir)
36    } else {
37        compact_single_volume(&volume, root, out_dir)
38    }
39}
40
41fn compact_logical_set(root: &Path, out_dir: &Path) -> io::Result<PathBuf> {
42    let set = Q42VolumeSet::open_root(root)?;
43    if set.segments().is_empty() {
44        return Err(invalid("Q42 volume set has no data segments"));
45    }
46
47    let mut lex = HashMap::new();
48    merge_lexicon(set.root(), &mut lex)?;
49    for segment in set.segments() {
50        merge_lexicon(segment, &mut lex)?;
51    }
52    for segment in set.lexicon_segments() {
53        merge_lexicon(segment, &mut lex)?;
54    }
55
56    let all_commons = set.segments().iter().all(volume_declares_commons);
57    let saw_sanctuary = scan_set_for_sanctuary(&set)?;
58    let declare_commons = all_commons && !saw_sanctuary;
59    let generation = set.manifest().generation.saturating_add(1);
60    let stem = output_stem(root);
61
62    let child_paths = if declare_commons {
63        rewrite_set_through_streaming_writer(&set, out_dir, &stem, &lex, true)?
64    } else {
65        rewrite_set_through_publisher(&set, out_dir, &stem, lex.clone())?
66    };
67
68    let mut segments = Vec::with_capacity(child_paths.len());
69    for path in &child_paths {
70        let name = path
71            .file_name()
72            .and_then(|n| n.to_str())
73            .ok_or_else(|| invalid("Q42 compacted child name"))?
74            .to_string();
75        segments.push(Q42VolumeManifest::segment_from_file(path, name)?);
76    }
77    let manifest = Q42VolumeManifest {
78        generation,
79        segments,
80        lexicon_segments: Vec::new(),
81    };
82    let root_path = out_dir.join(format!("{stem}-root.q42"));
83    publish_root_atomically(&root_path, &lex, &manifest)?;
84    Ok(root_path)
85}
86
87fn compact_single_volume(volume: &Q42Volume, root: &Path, out_dir: &Path) -> io::Result<PathBuf> {
88    if volume.block_count() == 0 {
89        return Err(invalid("Q42 volume has no SuperBlocks to compact"));
90    }
91    let mut lex = HashMap::new();
92    merge_lexicon(volume, &mut lex)?;
93    let declare_commons = volume_declares_commons(volume) && !scan_volume_for_sanctuary(volume)?;
94
95    let mut writer = StreamingQ42VolumeWriter::new(&lex)?;
96    if declare_commons {
97        writer.declare_permissive_commons();
98    }
99    stream_volume_blocks(volume, |seq, quins| writer.push_block(seq, quins))?;
100
101    let out = out_dir.join(format!("{}.q42", output_stem(root)));
102    writer.finish(&out)?;
103    Ok(out)
104}
105
106fn rewrite_set_through_publisher(
107    set: &Q42VolumeSet,
108    out_dir: &Path,
109    stem: &str,
110    lex: HashMap<u64, String>,
111) -> io::Result<Vec<PathBuf>> {
112    let mut publisher = Q42RolloverPublisher::new(out_dir, stem, lex)?;
113    stream_set_blocks(set, |seq, quins| publisher.push_block(seq, quins))?;
114    let produced = publisher.finish()?;
115    let catalog = Q42Volume::open(&produced)?;
116    let manifest = catalog.volume_manifest()?.ok_or_else(|| {
117        io::Error::new(
118            io::ErrorKind::InvalidData,
119            "Q42 rollover publisher produced no root catalog",
120        )
121    })?;
122    let parent = produced.parent().unwrap_or(out_dir);
123    let children = manifest
124        .segments
125        .iter()
126        .map(|segment| parent.join(&segment.locator))
127        .collect();
128    // The publisher already wrote a generation-1 catalog; the caller replaces
129    // it atomically with the incremented generation.
130    let _ = std::fs::remove_file(&produced);
131    Ok(children)
132}
133
134fn rewrite_set_through_streaming_writer(
135    set: &Q42VolumeSet,
136    out_dir: &Path,
137    stem: &str,
138    lex: &HashMap<u64, String>,
139    declare_commons: bool,
140) -> io::Result<Vec<PathBuf>> {
141    let mut writer = CompactRollover::new(out_dir, stem, lex, declare_commons)?;
142    stream_set_blocks(set, |seq, quins| writer.push_block(seq, quins))?;
143    writer.finish_children()
144}
145
146/// Same rollover contract as [`Q42RolloverPublisher`], with publication flags.
147struct CompactRollover {
148    dir: PathBuf,
149    stem: String,
150    lex: HashMap<u64, String>,
151    max_bytes: u64,
152    writer: StreamingQ42VolumeWriter,
153    child_paths: Vec<PathBuf>,
154    next_index: u32,
155    declare_commons: bool,
156}
157
158impl CompactRollover {
159    fn new(
160        dir: impl Into<PathBuf>,
161        stem: impl Into<String>,
162        lex: &HashMap<u64, String>,
163        declare_commons: bool,
164    ) -> io::Result<Self> {
165        let dir = dir.into();
166        std::fs::create_dir_all(&dir)?;
167        Ok(Self {
168            writer: new_streaming_writer(lex, declare_commons)?,
169            lex: lex.clone(),
170            dir,
171            stem: stem.into(),
172            max_bytes: DEFAULT_SEGMENT_MAX_BYTES,
173            child_paths: Vec::new(),
174            next_index: 0,
175            declare_commons,
176        })
177    }
178
179    fn push_block(&mut self, seq_id: u64, quins: &[NQuin]) -> io::Result<()> {
180        let projected = self.writer.maximum_final_length_after_next_block()?;
181        if self.writer.block_count() > 0 && projected > self.max_bytes {
182            self.roll()?;
183        }
184        self.writer.push_block(seq_id, quins)
185    }
186
187    fn roll(&mut self) -> io::Result<()> {
188        let path = self.child_path(self.next_index);
189        let writer = std::mem::replace(
190            &mut self.writer,
191            new_streaming_writer(&self.lex, self.declare_commons)?,
192        );
193        writer.finish(&path)?;
194        self.child_paths.push(path);
195        self.next_index += 1;
196        Ok(())
197    }
198
199    fn child_path(&self, index: u32) -> PathBuf {
200        self.dir.join(format!("{}-{:04}.q42", self.stem, index))
201    }
202
203    fn finish_children(mut self) -> io::Result<Vec<PathBuf>> {
204        if self.writer.block_count() > 0 || self.child_paths.is_empty() {
205            let path = self.child_path(self.next_index);
206            self.writer.finish(&path)?;
207            self.child_paths.push(path);
208        }
209        Ok(self.child_paths)
210    }
211}
212
213fn new_streaming_writer(
214    lex: &HashMap<u64, String>,
215    declare_commons: bool,
216) -> io::Result<StreamingQ42VolumeWriter> {
217    let mut writer = StreamingQ42VolumeWriter::new(lex)?;
218    if declare_commons {
219        writer.declare_permissive_commons();
220    }
221    Ok(writer)
222}
223
224fn publish_root_atomically(
225    root_path: &Path,
226    lex: &HashMap<u64, String>,
227    manifest: &Q42VolumeManifest,
228) -> io::Result<()> {
229    let parent = root_path
230        .parent()
231        .ok_or_else(|| invalid("Q42 compact root has no parent"))?;
232    let tmp = parent.join(format!(
233        ".{}.tmp",
234        root_path
235            .file_name()
236            .and_then(|n| n.to_str())
237            .unwrap_or("root.q42")
238    ));
239    write_volume_root_with_lex(&tmp, lex, manifest)?;
240    std::fs::rename(&tmp, root_path)
241}
242
243fn stream_set_blocks(
244    set: &Q42VolumeSet,
245    mut push: impl FnMut(u64, &[NQuin]) -> io::Result<()>,
246) -> io::Result<()> {
247    let mut seq = 0u64;
248    for segment in set.segments() {
249        stream_volume_blocks(segment, |_, quins| {
250            let result = push(seq, quins);
251            seq += 1;
252            result
253        })?;
254    }
255    if seq == 0 {
256        return Err(invalid("Q42 volume set has no SuperBlocks to compact"));
257    }
258    Ok(())
259}
260
261fn stream_volume_blocks(
262    volume: &Q42Volume,
263    mut push: impl FnMut(u64, &[NQuin]) -> io::Result<()>,
264) -> io::Result<()> {
265    let mut decoded = [0u8; SUPERBLOCK_SIZE];
266    let mut block = [NQuin::default(); QUINS_PER_BLOCK];
267    for index in 0..volume.block_count() as usize {
268        volume.read_superblock_into(index, &mut decoded)?;
269        let live = decode_live_quins(&decoded, &mut block)?;
270        push(index as u64, &block[..live])?;
271    }
272    Ok(())
273}
274
275fn decode_live_quins(
276    decoded: &[u8; SUPERBLOCK_SIZE],
277    out: &mut [NQuin; QUINS_PER_BLOCK],
278) -> io::Result<usize> {
279    let live = u64::from_le_bytes(decoded[16..24].try_into().unwrap()) as usize;
280    if live == 0 || live > QUINS_PER_BLOCK {
281        return Err(io::Error::new(
282            io::ErrorKind::InvalidData,
283            "Q42 SuperBlock has invalid live Quin count",
284        ));
285    }
286    for i in 0..live {
287        let offset = SUPERBLOCK_HEADER + i * QUIN_SIZE;
288        out[i] = bytemuck::pod_read_unaligned(&decoded[offset..offset + QUIN_SIZE]);
289    }
290    Ok(live)
291}
292
293fn merge_lexicon(volume: &Q42Volume, dest: &mut HashMap<u64, String>) -> io::Result<()> {
294    let view = volume.lex_view().map_err(|error| {
295        io::Error::new(
296            io::ErrorKind::InvalidData,
297            format!("invalid Q42LEX: {error:?}"),
298        )
299    })?;
300    for i in 0..view.entry_count() {
301        let Some(hash) = view.hash_at(i) else {
302            continue;
303        };
304        if let Some(text) = view.string_at(i) {
305            dest.entry(hash).or_insert_with(|| text.to_owned());
306        }
307    }
308    Ok(())
309}
310
311fn volume_declares_commons(volume: &Q42Volume) -> bool {
312    volume.header().flags & FLAG_PERMISSIVE_COMMONS != 0
313}
314
315fn scan_set_for_sanctuary(set: &Q42VolumeSet) -> io::Result<bool> {
316    for segment in set.segments() {
317        if segment.header().flags & FLAG_SANCTUARY != 0 || scan_volume_for_sanctuary(segment)? {
318            return Ok(true);
319        }
320    }
321    Ok(false)
322}
323
324fn scan_volume_for_sanctuary(volume: &Q42Volume) -> io::Result<bool> {
325    let mut decoded = [0u8; SUPERBLOCK_SIZE];
326    let mut block = [NQuin::default(); QUINS_PER_BLOCK];
327    for index in 0..volume.block_count() as usize {
328        volume.read_superblock_into(index, &mut decoded)?;
329        let live = decode_live_quins(&decoded, &mut block)?;
330        if block[..live].iter().any(quin_requires_sanctuary) {
331            return Ok(true);
332        }
333    }
334    Ok(false)
335}
336
337fn output_stem(root: &Path) -> String {
338    let stem = root
339        .file_stem()
340        .and_then(|s| s.to_str())
341        .unwrap_or("volume");
342    stem.strip_suffix("-root").unwrap_or(stem).to_string()
343}
344
345#[cfg(test)]
346mod tests {
347    use super::*;
348    use super::super::super::{
349        write_volume_root, FLAG_FIELD_POSTINGS, FLAG_FIELD_RANGES, FLAG_OBJECT_SORTED,
350    };
351    use crate::q42_volume::write_unified_volume;
352
353    fn quin(object: u64) -> NQuin {
354        NQuin {
355            subject: object,
356            predicate: 1,
357            object,
358            context: 0,
359            metadata: 0,
360            parity: object ^ 1 ^ object,
361        }
362    }
363
364    fn medical_quin(object: u64) -> NQuin {
365        let mut q = quin(object);
366        q.set_sensitivity_byte(NQuin::SENSITIVITY_CLASSIFIED);
367        q.set_sensitivity_tier(NQuin::SENSITIVITY_TIER_MEDICAL);
368        q.parity = q.subject ^ q.predicate ^ q.object ^ q.context;
369        q
370    }
371
372    fn lex_for(entries: &[(u64, &str)]) -> HashMap<u64, String> {
373        entries
374            .iter()
375            .map(|(hash, text)| (*hash, (*text).to_string()))
376            .collect()
377    }
378
379    fn write_child(path: &Path, quins: &[NQuin], lex: &HashMap<u64, String>, commons: bool) {
380        if commons {
381            let mut writer = StreamingQ42VolumeWriter::new(lex).unwrap();
382            writer.declare_permissive_commons();
383            writer.push_block(0, quins).unwrap();
384            writer.finish(path).unwrap();
385        } else {
386            let first = quins[0].object;
387            let last = quins[quins.len() - 1].object;
388            write_unified_volume(path, lex, &[(first, last)], &[quins.to_vec()]).unwrap();
389        }
390    }
391
392    fn write_two_child_root(
393        dir: &Path,
394        first: &[NQuin],
395        first_lex: &HashMap<u64, String>,
396        first_commons: bool,
397        second: &[NQuin],
398        second_lex: &HashMap<u64, String>,
399        second_commons: bool,
400    ) -> PathBuf {
401        let a = dir.join("segment-000.q42");
402        let b = dir.join("segment-001.q42");
403        write_child(&a, first, first_lex, first_commons);
404        write_child(&b, second, second_lex, second_commons);
405        let root = dir.join("set-root.q42");
406        let manifest = Q42VolumeManifest {
407            generation: 3,
408            segments: vec![
409                Q42VolumeManifest::segment_from_file(&a, "segment-000.q42".into()).unwrap(),
410                Q42VolumeManifest::segment_from_file(&b, "segment-001.q42".into()).unwrap(),
411            ],
412            lexicon_segments: Vec::new(),
413        };
414        write_volume_root(&root, &manifest).unwrap();
415        root
416    }
417
418    fn compacted_quins(root: &Path) -> Vec<NQuin> {
419        let set = Q42VolumeSet::open_root(root).unwrap();
420        let mut out = Vec::new();
421        for segment in set.segments() {
422            out.extend(segment.read_all_quins().unwrap());
423        }
424        out
425    }
426
427    #[test]
428    fn two_children_compact_to_a_readable_root() {
429        let src = tempfile::TempDir::new().unwrap();
430        let out = tempfile::TempDir::new().unwrap();
431        let first_lex = lex_for(&[(1, "p"), (10, "o-a")]);
432        let second_lex = lex_for(&[(1, "p"), (20, "o-b")]);
433        let root = write_two_child_root(
434            src.path(),
435            &[quin(10)],
436            &first_lex,
437            false,
438            &[quin(20)],
439            &second_lex,
440            false,
441        );
442
443        let compacted = compact_volume_set(&root, out.path()).unwrap();
444        let set = Q42VolumeSet::open_root(&compacted).unwrap();
445        assert_eq!(set.manifest().generation, 4);
446        assert!(!set.segments().is_empty());
447        let quins = compacted_quins(&compacted);
448        assert_eq!(quins.len(), 2);
449        assert!(quins.iter().any(|q| q.object == 10));
450        assert!(quins.iter().any(|q| q.object == 20));
451        assert_eq!(set.lookup_hash(10), Some("o-a"));
452        assert_eq!(set.lookup_hash(20), Some("o-b"));
453    }
454
455    #[test]
456    fn object_order_preserved() {
457        let src = tempfile::TempDir::new().unwrap();
458        let out = tempfile::TempDir::new().unwrap();
459        let lex = lex_for(&[(1, "p")]);
460        let root = write_two_child_root(
461            src.path(),
462            &[quin(2), quin(4)],
463            &lex,
464            false,
465            &[quin(6), quin(8)],
466            &lex,
467            false,
468        );
469
470        let compacted = compact_volume_set(&root, out.path()).unwrap();
471        let objects: Vec<u64> = compacted_quins(&compacted).iter().map(|q| q.object).collect();
472        assert_eq!(objects, vec![2, 4, 6, 8]);
473        assert!(objects.windows(2).all(|pair| pair[0] <= pair[1]));
474    }
475
476    #[test]
477    fn sanctuary_child_cannot_become_commons() {
478        let src = tempfile::TempDir::new().unwrap();
479        let out = tempfile::TempDir::new().unwrap();
480        let lex = lex_for(&[(1, "p")]);
481        let root = write_two_child_root(
482            src.path(),
483            &[quin(3)],
484            &lex,
485            true,
486            &[medical_quin(9)],
487            &lex,
488            false,
489        );
490
491        let compacted = compact_volume_set(&root, out.path()).unwrap();
492        let set = Q42VolumeSet::open_root(&compacted).unwrap();
493        for segment in set.segments() {
494            let flags = segment.header().flags;
495            assert_eq!(
496                flags & FLAG_PERMISSIVE_COMMONS,
497                0,
498                "Sanctuary input must not mint FLAG_PERMISSIVE_COMMONS"
499            );
500            assert_ne!(flags & FLAG_SANCTUARY, 0);
501        }
502        assert!(compacted_quins(&compacted)
503            .iter()
504            .any(quin_requires_sanctuary));
505    }
506
507    #[test]
508    fn single_file_rewrites_with_field_indexes() {
509        let src = tempfile::TempDir::new().unwrap();
510        let out = tempfile::TempDir::new().unwrap();
511        let path = src.path().join("plain.q42");
512        let lex = lex_for(&[(1, "s"), (2, "p"), (3, "o")]);
513        write_unified_volume(&path, &lex, &[(3, 3)], &[vec![quin(3)]]).unwrap();
514
515        let compacted = compact_volume_set(&path, out.path()).unwrap();
516        let volume = Q42Volume::open(&compacted).unwrap();
517        let flags = volume.header().flags;
518        assert_ne!(flags & FLAG_OBJECT_SORTED, 0);
519        assert_ne!(flags & FLAG_FIELD_RANGES, 0);
520        assert_ne!(flags & FLAG_FIELD_POSTINGS, 0);
521        assert_eq!(volume.read_all_quins().unwrap()[0].object, 3);
522        assert_eq!(volume.lex_view().unwrap().lookup_hash(3), Some("o"));
523    }
524}
qualia_core_db/q42/volume/compact.rs

qualia_core_db/q42/volume/
compact.rs
