qualia_core_db/q42/volume/
stream_writer.rs

1//! Cold, bounded-memory unified-Q42 writer for large sorted streams.
2
3use std::collections::HashMap;
4use std::fs::{File, OpenOptions};
5use std::io::{self, BufWriter, Read, Write};
6use std::path::{Path, PathBuf};
7
8use tempfile::TempDir;
9
10use super::super::{
11    encode_lex, encode_superblock, header_to_bytes, BlockDirectoryEntry, Q42VolumeHeader,
12    FLAG_BLOCKS_LZ4, FLAG_FIELD_POSTINGS, FLAG_FIELD_RANGES, FLAG_OBJECT_SORTED,
13    FLAG_PERMISSIVE_COMMONS, FLAG_SANCTUARY, HEADER_SIZE, Q42_VERSION_V3, QUINS_PER_BLOCK,
14    SUPERBLOCK_SIZE,
15};
16use super::postings::{encode_block_postings, BlockFieldPostings, FIELD_POSTINGS_MAGIC};
17use super::publication::quin_requires_sanctuary;
18use crate::NQuin;
19
20fn invalid(message: impl Into<String>) -> io::Error {
21    io::Error::new(io::ErrorKind::InvalidInput, message.into())
22}
23
24/// Writes a sorted Q42 segment without retaining its compressed payload or
25/// per-block metadata in heap collections. Temporary streams are RAII-owned.
26pub struct StreamingQ42VolumeWriter {
27    _temp: TempDir,
28    bidx_path: PathBuf,
29    field_ranges_path: PathBuf,
30    postings_path: PathBuf,
31    directory_path: PathBuf,
32    data_path: PathBuf,
33    bidx: BufWriter<File>,
34    field_ranges: BufWriter<File>,
35    postings: BufWriter<File>,
36    directory: BufWriter<File>,
37    data: BufWriter<File>,
38    posting_offsets: Vec<u32>,
39    lex_bytes: Vec<u8>,
40    block_count: u64,
41    data_length: u64,
42    last_object_hash: Option<u64>,
43    publication_commons: bool,
44    publication_sanctuary: bool,
45    sanctuary_quin_count: u64,
46    merkle_root: [u8; 32],
47}
48
49impl StreamingQ42VolumeWriter {
50    pub fn new(lexicon: &HashMap<u64, String>) -> io::Result<Self> {
51        let temp = TempDir::new()?;
52        let bidx_path = temp.path().join("bidx.entries");
53        let field_ranges_path = temp.path().join("field-ranges.entries");
54        let postings_path = temp.path().join("field.postings");
55        let directory_path = temp.path().join("block.directory");
56        let data_path = temp.path().join("blocks.lz4");
57        let open = |path: &Path| OpenOptions::new().create_new(true).write(true).open(path);
58        Ok(Self {
59            bidx: BufWriter::new(open(&bidx_path)?),
60            field_ranges: BufWriter::new(open(&field_ranges_path)?),
61            postings: BufWriter::new(open(&postings_path)?),
62            directory: BufWriter::new(open(&directory_path)?),
63            data: BufWriter::new(open(&data_path)?),
64            _temp: temp,
65            bidx_path,
66            field_ranges_path,
67            postings_path,
68            directory_path,
69            data_path,
70            posting_offsets: vec![0],
71            lex_bytes: encode_lex(lexicon)
72                .map_err(|error| invalid(format!("invalid Q42LEX: {error:?}")))?,
73            block_count: 0,
74            data_length: 0,
75            last_object_hash: None,
76            publication_commons: false,
77            publication_sanctuary: false,
78            sanctuary_quin_count: 0,
79            merkle_root: [0; 32],
80        })
81    }
82
83    /// Mark this segment as a Permissive Commons catalog. Ignored if any
84    /// pushed Quin requires Sanctuary.
85    pub fn declare_permissive_commons(&mut self) {
86        self.publication_commons = true;
87    }
88
89    /// Mark this segment as Sanctuary / Selfhood. Public hash addressing is denied.
90    pub fn declare_sanctuary(&mut self) {
91        self.publication_sanctuary = true;
92    }
93
94    pub fn push_block(&mut self, seq_id: u64, quins: &[NQuin]) -> io::Result<()> {
95        let Some(first) = quins.first() else {
96            return Err(invalid("Q42 SuperBlock must not be empty"));
97        };
98        if quins.len() > QUINS_PER_BLOCK {
99            return Err(invalid("Q42 SuperBlock exceeds Quin capacity"));
100        }
101        let mut previous = first.object;
102        for quin in &quins[1..] {
103            if quin.object < previous {
104                return Err(invalid("Q42 SuperBlock is not object-sorted"));
105            }
106            previous = quin.object;
107        }
108        if self
109            .last_object_hash
110            .is_some_and(|last| first.object < last)
111        {
112            return Err(invalid("Q42 blocks are not globally object-sorted"));
113        }
114        let compressed = lz4_flex::compress_prepend_size(&encode_superblock(seq_id, quins));
115        let compressed_len = u32::try_from(compressed.len())
116            .map_err(|_| invalid("compressed Q42 block exceeds u32"))?;
117        self.bidx.write_all(&first.object.to_le_bytes())?;
118        self.bidx.write_all(&previous.to_le_bytes())?;
119        let mut subject_min = first.subject;
120        let mut subject_max = first.subject;
121        let mut predicate_min = first.predicate;
122        let mut predicate_max = first.predicate;
123        let mut context_min = first.context;
124        let mut context_max = first.context;
125        for quin in &quins[1..] {
126            subject_min = subject_min.min(quin.subject);
127            subject_max = subject_max.max(quin.subject);
128            predicate_min = predicate_min.min(quin.predicate);
129            predicate_max = predicate_max.max(quin.predicate);
130            context_min = context_min.min(quin.context);
131            context_max = context_max.max(quin.context);
132        }
133        for value in [
134            subject_min,
135            subject_max,
136            predicate_min,
137            predicate_max,
138            context_min,
139            context_max,
140        ] {
141            self.field_ranges.write_all(&value.to_le_bytes())?;
142        }
143        let encoded = encode_block_postings(&BlockFieldPostings::from_quins(quins));
144        self.postings.write_all(&encoded)?;
145        let next = self
146            .posting_offsets
147            .last()
148            .copied()
149            .unwrap_or(0)
150            .checked_add(encoded.len() as u32)
151            .ok_or_else(|| invalid("Q42 field postings overflow u32"))?;
152        self.posting_offsets.push(next);
153        BlockDirectoryEntry {
154            rel_offset: self.data_length,
155            comp_len: compressed_len,
156            uncomp_len: SUPERBLOCK_SIZE as u32,
157        }
158        .write_to(&mut self.directory)?;
159        self.data.write_all(&compressed)?;
160        self.data_length = self
161            .data_length
162            .checked_add(compressed.len() as u64)
163            .ok_or_else(|| invalid("Q42 data length overflow"))?;
164        self.block_count += 1;
165        self.last_object_hash = Some(previous);
166        self.sanctuary_quin_count += quins
167            .iter()
168            .filter(|quin| quin_requires_sanctuary(quin))
169            .count() as u64;
170        {
171            use sha2::{Digest, Sha256};
172            let mut hasher = Sha256::new();
173            hasher.update(self.merkle_root);
174            hasher.update(&compressed);
175            self.merkle_root = hasher.finalize().into();
176        }
177        Ok(())
178    }
179
180    pub fn block_count(&self) -> u64 {
181        self.block_count
182    }
183
184    /// The exact final length if the writer were finished now. It includes the
185    /// front matter and fixed BIDX/directory records, not just compressed data.
186    pub fn estimated_final_length(&self) -> io::Result<u64> {
187        let bidx_length = 16u64
188            .checked_add(
189                self.block_count
190                    .checked_mul(16)
191                    .ok_or_else(|| invalid("Q42 BIDX length overflow"))?,
192            )
193            .ok_or_else(|| invalid("Q42 BIDX length overflow"))?;
194        let field_ranges_length = 16u64
195            .checked_add(
196                self.block_count
197                    .checked_mul(48)
198                    .ok_or_else(|| invalid("Q42 field-range index length overflow"))?,
199            )
200            .ok_or_else(|| invalid("Q42 field-range index length overflow"))?;
201        let postings_length = self.encoded_postings_section_len()?;
202        let directory_length = self
203            .block_count
204            .checked_mul(BlockDirectoryEntry::SIZE as u64)
205            .ok_or_else(|| invalid("Q42 directory length overflow"))?;
206        (HEADER_SIZE as u64)
207            .checked_add(self.lex_bytes.len() as u64)
208            .and_then(|value| value.checked_add(bidx_length))
209            .and_then(|value| value.checked_add(field_ranges_length))
210            .and_then(|value| value.checked_add(postings_length))
211            .and_then(|value| value.checked_add(directory_length))
212            .and_then(|value| value.checked_add(self.data_length))
213            .ok_or_else(|| invalid("Q42 final length overflow"))
214    }
215
216    /// A safe upper bound for the final length after one additional block.
217    /// It lets volume publishers split before a block crosses their byte cap
218    /// without retaining compressed payloads.
219    pub fn maximum_final_length_after_next_block(&self) -> io::Result<u64> {
220        self.estimated_final_length()?
221            .checked_add(32) // one BIDX interval plus one directory entry
222            .and_then(|value| {
223                value.checked_add(crate::q42_volume::MAX_COMPRESSED_SUPERBLOCK_SIZE as u64)
224            })
225            .ok_or_else(|| invalid("Q42 final length overflow"))
226    }
227
228    fn encoded_postings_section_len(&self) -> io::Result<u64> {
229        let table = (self.block_count + 1)
230            .checked_mul(4)
231            .ok_or_else(|| invalid("Q42 postings table overflow"))?;
232        let payload = u64::from(*self.posting_offsets.last().unwrap_or(&0));
233        16u64
234            .checked_add(table)
235            .and_then(|value| value.checked_add(payload))
236            .ok_or_else(|| invalid("Q42 postings section overflow"))
237    }
238
239    pub fn finish(mut self, path: &Path) -> io::Result<()> {
240        self.bidx.flush()?;
241        self.field_ranges.flush()?;
242        self.postings.flush()?;
243        self.directory.flush()?;
244        self.data.flush()?;
245        let bidx_length = 16
246            + self
247                .block_count
248                .checked_mul(16)
249                .ok_or_else(|| invalid("Q42 BIDX length overflow"))?;
250        let directory_length = self
251            .block_count
252            .checked_mul(BlockDirectoryEntry::SIZE as u64)
253            .ok_or_else(|| invalid("Q42 directory length overflow"))?;
254        let field_ranges_length = 16
255            + self
256                .block_count
257                .checked_mul(48)
258                .ok_or_else(|| invalid("Q42 field-range index length overflow"))?;
259        let postings_length = self.encoded_postings_section_len()?;
260        let lex_offset = HEADER_SIZE as u64;
261        let bidx_offset = lex_offset + self.lex_bytes.len() as u64;
262        let field_ranges_offset = bidx_offset + bidx_length;
263        let postings_offset = field_ranges_offset + field_ranges_length;
264        let directory_offset = postings_offset + postings_length;
265        let data_offset = directory_offset + directory_length;
266        let mut flags = FLAG_BLOCKS_LZ4
267            | FLAG_OBJECT_SORTED
268            | FLAG_FIELD_RANGES
269            | FLAG_FIELD_POSTINGS;
270        if self.publication_sanctuary || self.sanctuary_quin_count > 0 {
271            flags |= FLAG_SANCTUARY;
272        } else if self.publication_commons {
273            flags |= FLAG_PERMISSIVE_COMMONS;
274        }
275        let header = Q42VolumeHeader {
276            magic: super::super::Q42_MAGIC,
277            version: Q42_VERSION_V3,
278            flags,
279            lex_offset,
280            lex_length: self.lex_bytes.len() as u64,
281            bidx_offset,
282            bidx_length,
283            block_dir_offset: directory_offset,
284            block_dir_length: directory_length,
285            data_offset,
286            data_length: self.data_length,
287            block_count: self.block_count,
288            block_size: SUPERBLOCK_SIZE as u32,
289            quins_per_block: QUINS_PER_BLOCK as u32,
290            temporal_index_offset: 0,
291            temporal_index_length: 0,
292            merkle_root: self.merkle_root,
293            assertion_timestamp: 0,
294            dag_root_offset: 0,
295            dag_root_length: 0,
296            natural_person_did_offset: 0,
297            software_agent_did_offset: 0,
298            _reserved: {
299                let mut reserved = [0; 80];
300                reserved[16..24].copy_from_slice(&field_ranges_offset.to_le_bytes());
301                reserved[24..32].copy_from_slice(&(field_ranges_length as u64).to_le_bytes());
302                reserved[32..40].copy_from_slice(&postings_offset.to_le_bytes());
303                reserved[40..48].copy_from_slice(&postings_length.to_le_bytes());
304                reserved
305            },
306        };
307        let mut output = BufWriter::new(
308            OpenOptions::new()
309                .create(true)
310                .write(true)
311                .truncate(true)
312                .open(path)?,
313        );
314        output.write_all(&header_to_bytes(&header))?;
315        output.write_all(&self.lex_bytes)?;
316        output.write_all(&super::super::BIDX_MAGIC)?;
317        output.write_all(&1u32.to_le_bytes())?;
318        output.write_all(&(self.block_count as u32).to_le_bytes())?;
319        output.write_all(&0u32.to_le_bytes())?;
320        copy_file(&self.bidx_path, &mut output)?;
321        output.write_all(&super::super::FIELD_RANGE_INDEX_MAGIC)?;
322        output.write_all(&1u32.to_le_bytes())?;
323        output.write_all(&(self.block_count as u32).to_le_bytes())?;
324        output.write_all(&0u32.to_le_bytes())?;
325        copy_file(&self.field_ranges_path, &mut output)?;
326        output.write_all(&FIELD_POSTINGS_MAGIC)?;
327        output.write_all(&1u32.to_le_bytes())?;
328        output.write_all(&(self.block_count as u32).to_le_bytes())?;
329        output.write_all(&0u32.to_le_bytes())?;
330        for offset in &self.posting_offsets {
331            output.write_all(&offset.to_le_bytes())?;
332        }
333        copy_file(&self.postings_path, &mut output)?;
334        copy_file(&self.directory_path, &mut output)?;
335        copy_file(&self.data_path, &mut output)?;
336        output.flush()
337    }
338}
339
340fn copy_file(path: &Path, output: &mut BufWriter<File>) -> io::Result<()> {
341    let mut input = File::open(path)?;
342    let mut buffer = [0u8; 64 * 1024];
343    loop {
344        let count = input.read(&mut buffer)?;
345        if count == 0 {
346            return Ok(());
347        }
348        output.write_all(&buffer[..count])?;
349    }
350}
351
352#[cfg(test)]
353mod tests {
354    use super::*;
355    use crate::q42_volume::Q42Volume;
356    use tempfile::NamedTempFile;
357
358    #[test]
359    fn streams_a_readable_volume_without_payload_accumulation() {
360        let mut lex = HashMap::new();
361        lex.insert(1, "s".to_string());
362        lex.insert(2, "p".to_string());
363        lex.insert(3, "o".to_string());
364        let mut writer = StreamingQ42VolumeWriter::new(&lex).unwrap();
365        writer
366            .push_block(
367                0,
368                &[NQuin {
369                    subject: 1,
370                    predicate: 2,
371                    object: 3,
372                    context: 0,
373                    metadata: 0,
374                    parity: 0,
375                }],
376            )
377            .unwrap();
378        let output = NamedTempFile::new().unwrap();
379        writer.finish(output.path()).unwrap();
380        let volume = Q42Volume::open(output.path()).unwrap();
381        assert_eq!(volume.block_count(), 1);
382        assert_eq!(volume.object_hash_bounds(), Some((3, 3)));
383    }
384}
qualia_core_db/q42/volume/stream_writer.rs

qualia_core_db/q42/volume/
stream_writer.rs
