qualia_core_db/q42/volume/
postings.rs

1//! Compact per-block S/P/C membership for range pruning.
2//!
3//! Field min/max ranges are a weak first reject: any hash inside a wide interval
4//! still forces a SuperBlock decode. These postings are exact unique hashes,
5//! delta-varint encoded, so a bound subject/predicate/context either is present
6//! or the block is skipped. No false positives, no false negatives.
7//!
8//! A 256-byte Bloom is used only when unique hashes would be *larger* than that
9//! Bloom (measured per field). Blooms have no false negatives; measured FPR is
10//! reported by tests.
11
12use std::io;
13
14use crate::NQuin;
15
16pub const FIELD_POSTINGS_MAGIC: [u8; 4] = *b"PIDX";
17pub const FIELD_POSTINGS_HEADER_BYTES: usize = 16;
18pub const BLOOM_BYTES: usize = 256;
19pub const BLOOM_BITS: usize = BLOOM_BYTES * 8;
20pub const BLOOM_HASHES: usize = 4;
21
22const KIND_POSTINGS: u8 = 0;
23const KIND_BLOOM: u8 = 1;
24
25fn invalid(message: impl Into<String>) -> io::Error {
26    io::Error::new(io::ErrorKind::InvalidData, message.into())
27}
28
29/// One SuperBlock's compact S/P/C membership.
30#[derive(Clone, Debug, Default, PartialEq, Eq)]
31pub struct BlockFieldPostings {
32    pub subjects: Vec<u64>,
33    pub predicates: Vec<u64>,
34    pub contexts: Vec<u64>,
35}
36
37impl BlockFieldPostings {
38    pub fn from_quins(quins: &[NQuin]) -> Self {
39        let mut subjects = Vec::with_capacity(quins.len());
40        let mut predicates = Vec::with_capacity(quins.len());
41        let mut contexts = Vec::with_capacity(quins.len());
42        for quin in quins {
43            subjects.push(quin.subject);
44            predicates.push(quin.predicate);
45            contexts.push(quin.context);
46        }
47        sort_unique(&mut subjects);
48        sort_unique(&mut predicates);
49        sort_unique(&mut contexts);
50        Self {
51            subjects,
52            predicates,
53            contexts,
54        }
55    }
56
57    pub fn contains_subject(&self, value: u64) -> bool {
58        self.subjects.binary_search(&value).is_ok()
59    }
60    pub fn contains_predicate(&self, value: u64) -> bool {
61        self.predicates.binary_search(&value).is_ok()
62    }
63    pub fn contains_context(&self, value: u64) -> bool {
64        self.contexts.binary_search(&value).is_ok()
65    }
66
67    /// Encoded size if this field used exact postings.
68    pub fn posting_bytes(values: &[u64]) -> usize {
69        2 + encoded_delta_len(values)
70    }
71}
72
73/// Encode one SuperBlock's three fields. Chooses Bloom when it is strictly
74/// smaller than the exact posting list.
75pub fn encode_block_postings(postings: &BlockFieldPostings) -> Vec<u8> {
76    let mut out = Vec::new();
77    encode_field(&mut out, &postings.subjects);
78    encode_field(&mut out, &postings.predicates);
79    encode_field(&mut out, &postings.contexts);
80    out
81}
82
83fn encode_field(out: &mut Vec<u8>, values: &[u64]) {
84    let posting_len = BlockFieldPostings::posting_bytes(values);
85    if values.len() > 8 && posting_len > BLOOM_BYTES + 1 {
86        out.push(KIND_BLOOM);
87        let mut bloom = [0u8; BLOOM_BYTES];
88        for value in values {
89            bloom_insert(&mut bloom, *value);
90        }
91        out.extend_from_slice(&bloom);
92        return;
93    }
94    out.push(KIND_POSTINGS);
95    out.extend_from_slice(&(values.len() as u16).to_le_bytes());
96    encode_deltas(out, values);
97}
98
99/// Decode one SuperBlock payload into caller-owned hash scratch, then test
100/// membership without retaining the payload.
101pub fn field_may_contain(encoded: &[u8], field: usize, value: u64) -> io::Result<bool> {
102    let mut offset = 0usize;
103    for current in 0..3 {
104        let (next, present) = decode_field_contains(encoded, offset, value)?;
105        if current == field {
106            return Ok(present);
107        }
108        offset = next;
109    }
110    Err(invalid("field postings have fewer than three fields"))
111}
112
113fn decode_field_contains(encoded: &[u8], offset: usize, value: u64) -> io::Result<(usize, bool)> {
114    let kind = *encoded
115        .get(offset)
116        .ok_or_else(|| invalid("truncated field postings"))?;
117    match kind {
118        KIND_POSTINGS => {
119            if offset + 3 > encoded.len() {
120                return Err(invalid("truncated posting count"));
121            }
122            let count = u16::from_le_bytes([encoded[offset + 1], encoded[offset + 2]]) as usize;
123            let mut cursor = offset + 3;
124            let mut previous = 0u64;
125            let mut found = false;
126            for _ in 0..count {
127                let (delta, used) = decode_varint(&encoded[cursor..])?;
128                previous = previous
129                    .checked_add(delta)
130                    .ok_or_else(|| invalid("posting delta overflow"))?;
131                cursor = cursor
132                    .checked_add(used)
133                    .ok_or_else(|| invalid("posting cursor overflow"))?;
134                if previous == value {
135                    found = true;
136                }
137            }
138            Ok((cursor, found))
139        }
140        KIND_BLOOM => {
141            let end = offset
142                .checked_add(1 + BLOOM_BYTES)
143                .ok_or_else(|| invalid("bloom overflow"))?;
144            if end > encoded.len() {
145                return Err(invalid("truncated bloom filter"));
146            }
147            let bloom = &encoded[offset + 1..end];
148            Ok((end, bloom_may_contain(bloom, value)))
149        }
150        _ => Err(invalid("unknown field-postings kind")),
151    }
152}
153
154pub fn encode_postings_section(blocks: &[BlockFieldPostings]) -> Vec<u8> {
155    let mut payloads = Vec::with_capacity(blocks.len());
156    let mut offsets = Vec::with_capacity(blocks.len() + 1);
157    let mut cursor = 0u32;
158    offsets.push(0);
159    for block in blocks {
160        let payload = encode_block_postings(block);
161        cursor = cursor
162            .checked_add(payload.len() as u32)
163            .expect("field postings payload exceeds u32");
164        offsets.push(cursor);
165        payloads.push(payload);
166    }
167    let mut out = Vec::new();
168    out.extend_from_slice(&FIELD_POSTINGS_MAGIC);
169    out.extend_from_slice(&1u32.to_le_bytes());
170    out.extend_from_slice(&(blocks.len() as u32).to_le_bytes());
171    out.extend_from_slice(&0u32.to_le_bytes());
172    for offset in &offsets {
173        out.extend_from_slice(&offset.to_le_bytes());
174    }
175    for payload in payloads {
176        out.extend_from_slice(&payload);
177    }
178    out
179}
180
181pub fn validate_postings_section(bytes: &[u8], expected_blocks: usize) -> io::Result<()> {
182    if bytes.len() < FIELD_POSTINGS_HEADER_BYTES {
183        return Err(invalid("field postings shorter than header"));
184    }
185    if bytes[0..4] != FIELD_POSTINGS_MAGIC {
186        return Err(invalid("field postings have bad magic"));
187    }
188    if u32::from_le_bytes(bytes[4..8].try_into().unwrap()) != 1 {
189        return Err(invalid("unsupported field postings version"));
190    }
191    let block_count = u32::from_le_bytes(bytes[8..12].try_into().unwrap()) as usize;
192    if block_count != expected_blocks {
193        return Err(invalid("field postings block count mismatch"));
194    }
195    let table_bytes = (block_count + 1)
196        .checked_mul(4)
197        .ok_or_else(|| invalid("field postings table overflow"))?;
198    let header_end = FIELD_POSTINGS_HEADER_BYTES + table_bytes;
199    if bytes.len() < header_end {
200        return Err(invalid("field postings offset table truncated"));
201    }
202    let last = u32::from_le_bytes(
203        bytes[header_end - 4..header_end]
204            .try_into()
205            .unwrap(),
206    ) as usize;
207    if bytes.len() != header_end + last {
208        return Err(invalid("field postings payload length mismatch"));
209    }
210    for block_index in 0..block_count {
211        let _ = block_payload(bytes, block_index)?;
212    }
213    Ok(())
214}
215
216/// Absolute byte interval of one SuperBlock's encoded PIDX payload, relative to
217/// the start of a complete PIDX section (header + offset table + payloads).
218pub(crate) fn block_payload_interval(
219    block_count: usize,
220    block_index: usize,
221    table_start: u32,
222    table_end: u32,
223) -> io::Result<(usize, usize)> {
224    if block_index >= block_count {
225        return Err(invalid("field postings block index out of range"));
226    }
227    if table_end < table_start {
228        return Err(invalid("field postings interval inverted"));
229    }
230    let table_bytes = (block_count + 1)
231        .checked_mul(4)
232        .ok_or_else(|| invalid("field postings table overflow"))?;
233    let payload_base = FIELD_POSTINGS_HEADER_BYTES
234        .checked_add(table_bytes)
235        .ok_or_else(|| invalid("field postings payload base overflow"))?;
236    let from = payload_base
237        .checked_add(table_start as usize)
238        .ok_or_else(|| invalid("posting payload start overflow"))?;
239    let to = payload_base
240        .checked_add(table_end as usize)
241        .ok_or_else(|| invalid("posting payload end overflow"))?;
242    Ok((from, to))
243}
244
245/// Slice one SuperBlock's encoded S/P/C membership out of an in-memory PIDX section.
246pub fn block_payload<'a>(bytes: &'a [u8], block_index: usize) -> io::Result<&'a [u8]> {
247    if bytes.len() < FIELD_POSTINGS_HEADER_BYTES {
248        return Err(invalid("field postings shorter than header"));
249    }
250    let block_count = u32::from_le_bytes(bytes[8..12].try_into().unwrap()) as usize;
251    let table = FIELD_POSTINGS_HEADER_BYTES
252        .checked_add(
253            block_index
254                .checked_mul(4)
255                .ok_or_else(|| invalid("field postings table overflow"))?,
256        )
257        .ok_or_else(|| invalid("field postings table overflow"))?;
258    if bytes.len() < table + 8 {
259        return Err(invalid("field postings offset table truncated"));
260    }
261    let start = u32::from_le_bytes(bytes[table..table + 4].try_into().unwrap());
262    let end = u32::from_le_bytes(bytes[table + 4..table + 8].try_into().unwrap());
263    let (from, to) = block_payload_interval(block_count, block_index, start, end)?;
264    bytes
265        .get(from..to)
266        .ok_or_else(|| invalid("field postings payload slice out of range"))
267}
268
269fn sort_unique(values: &mut Vec<u64>) {
270    values.sort_unstable();
271    values.dedup();
272}
273
274fn encoded_delta_len(values: &[u64]) -> usize {
275    let mut previous = 0u64;
276    let mut len = 0usize;
277    for value in values {
278        len += varint_len(value - previous);
279        previous = *value;
280    }
281    len
282}
283
284fn encode_deltas(out: &mut Vec<u8>, values: &[u64]) {
285    let mut previous = 0u64;
286    for value in values {
287        encode_varint(out, value - previous);
288        previous = *value;
289    }
290}
291
292fn encode_varint(out: &mut Vec<u8>, mut value: u64) {
293    loop {
294        let mut byte = (value & 0x7f) as u8;
295        value >>= 7;
296        if value != 0 {
297            byte |= 0x80;
298        }
299        out.push(byte);
300        if value == 0 {
301            return;
302        }
303    }
304}
305
306fn varint_len(mut value: u64) -> usize {
307    let mut len = 1;
308    while value > 0x7f {
309        value >>= 7;
310        len += 1;
311    }
312    len
313}
314
315fn decode_varint(bytes: &[u8]) -> io::Result<(u64, usize)> {
316    let mut value = 0u64;
317    let mut shift = 0u32;
318    for (index, byte) in bytes.iter().copied().enumerate() {
319        if shift >= 64 {
320            return Err(invalid("varint shift overflow"));
321        }
322        value |= u64::from(byte & 0x7f) << shift;
323        if byte & 0x80 == 0 {
324            return Ok((value, index + 1));
325        }
326        shift += 7;
327    }
328    Err(invalid("truncated varint"))
329}
330
331fn bloom_insert(bloom: &mut [u8], value: u64) {
332    for hash in bloom_hashes(value) {
333        let bit = (hash as usize) % BLOOM_BITS;
334        bloom[bit / 8] |= 1 << (bit % 8);
335    }
336}
337
338fn bloom_may_contain(bloom: &[u8], value: u64) -> bool {
339    bloom_hashes(value).into_iter().all(|hash| {
340        let bit = (hash as usize) % BLOOM_BITS;
341        bloom[bit / 8] & (1 << (bit % 8)) != 0
342    })
343}
344
345fn bloom_hashes(value: u64) -> [u64; BLOOM_HASHES] {
346    let mixed = value.wrapping_mul(0x9E37_79B9_7F4A_7C15);
347    let alt = (!value).wrapping_mul(0xBF58_476D_1CE4_E5B9);
348    [
349        mixed,
350        alt,
351        mixed.wrapping_add(alt),
352        mixed ^ alt.rotate_left(17),
353    ]
354}
355
356/// Measured false-positive rate of the Bloom encoding on a synthetic field.
357pub fn measure_bloom_false_positives(values: &[u64], probes: &[u64]) -> (usize, usize) {
358    let mut bloom = [0u8; BLOOM_BYTES];
359    for value in values {
360        bloom_insert(&mut bloom, *value);
361    }
362    let mut false_positives = 0usize;
363    let mut eligible = 0usize;
364    for probe in probes {
365        if values.binary_search(probe).is_ok() {
366            continue;
367        }
368        eligible += 1;
369        if bloom_may_contain(&bloom, *probe) {
370            false_positives += 1;
371        }
372    }
373    (false_positives, eligible)
374}
375
376#[cfg(test)]
377mod tests {
378    use super::*;
379
380    #[test]
381    fn exact_postings_have_no_false_positives() {
382        let quins = [
383            quin(10, 1, 100),
384            quin(20, 1, 101),
385            quin(10, 2, 102),
386        ];
387        let postings = BlockFieldPostings::from_quins(&quins);
388        let encoded = encode_block_postings(&postings);
389        assert!(field_may_contain(&encoded, 0, 10).unwrap());
390        assert!(field_may_contain(&encoded, 0, 20).unwrap());
391        assert!(!field_may_contain(&encoded, 0, 99).unwrap());
392        assert!(field_may_contain(&encoded, 1, 2).unwrap());
393        assert!(!field_may_contain(&encoded, 1, 9).unwrap());
394    }
395
396    #[test]
397    fn bloom_is_chosen_when_it_is_smaller_and_has_no_false_negatives() {
398        let values: Vec<u64> = (0..400).map(|i| i * 1_000_003 + 17).collect();
399        let mut encoded = Vec::new();
400        encode_field(&mut encoded, &values);
401        assert_eq!(encoded[0], KIND_BLOOM);
402        for value in &values {
403            assert!(field_may_contain(&encoded, 0, *value).unwrap());
404        }
405        let probes: Vec<u64> = (0..2_000).map(|i| i * 97 + 3).collect();
406        let (fp, eligible) = measure_bloom_false_positives(&values, &probes);
407        assert!(eligible > 0);
408        // 256-byte / 4-hash Bloom on 400 items is compact; FPR must stay usable.
409        assert!(
410            (fp as f64) / (eligible as f64) < 0.35,
411            "bloom FPR too high: {fp}/{eligible}"
412        );
413    }
414
415    #[test]
416    fn section_round_trip_and_validation() {
417        let blocks = vec![
418            BlockFieldPostings::from_quins(&[quin(1, 2, 3)]),
419            BlockFieldPostings::from_quins(&[quin(4, 5, 6), quin(7, 5, 8)]),
420        ];
421        let section = encode_postings_section(&blocks);
422        validate_postings_section(&section, 2).unwrap();
423        let payload = block_payload(&section, 1).unwrap();
424        assert!(field_may_contain(payload, 0, 7).unwrap());
425        assert!(!field_may_contain(payload, 0, 1).unwrap());
426        assert!(block_payload(&section, 2).is_err());
427    }
428
429    #[test]
430    fn inverted_table_interval_is_rejected() {
431        let mut section = encode_postings_section(&[
432            BlockFieldPostings::from_quins(&[quin(1, 2, 3)]),
433            BlockFieldPostings::from_quins(&[quin(4, 5, 6)]),
434        ]);
435        let table = FIELD_POSTINGS_HEADER_BYTES;
436        section[table..table + 4].copy_from_slice(&8u32.to_le_bytes());
437        section[table + 4..table + 8].copy_from_slice(&0u32.to_le_bytes());
438        assert!(validate_postings_section(&section, 2).is_err());
439        assert!(block_payload(&section, 0).is_err());
440    }
441
442    fn quin(subject: u64, predicate: u64, context: u64) -> NQuin {
443        NQuin {
444            subject,
445            predicate,
446            object: 0,
447            context,
448            metadata: 0,
449            parity: 0,
450        }
451    }
452}
qualia_core_db/q42/volume/postings.rs

qualia_core_db/q42/volume/
postings.rs
