qualia_core_db/query/
query_engine.rs

1use crate::NQuin;
2use std::fs::File;
3use std::io::{Read, Seek, SeekFrom};
4use std::path::Path;
5use std::time::Instant;
6
7/// Bounded sample of a `.q42` graph — unified v3 first, then flat packed NQuins.
8pub fn mmap_sample_quins(
9    file_path: &str,
10    max_quins: usize,
11) -> Result<Vec<NQuin>, Box<dyn std::error::Error>> {
12    #[cfg(not(target_arch = "wasm32"))]
13    {
14        use crate::q42_volume::{Q42Volume, Q42VolumeSet};
15
16        if max_quins == 0 {
17            return Ok(Vec::new());
18        }
19        let path = Path::new(file_path);
20        if crate::q42_volume::is_unified_volume(path)? {
21            let volume = Q42Volume::open(path)?;
22            if volume.volume_manifest()?.is_some() {
23                let set = Q42VolumeSet::open_root(path)?;
24                let mut out = Vec::new();
25                for segment in set.segments() {
26                    sample_volume_blocks(&segment, max_quins.saturating_sub(out.len()), &mut out)?;
27                    if out.len() >= max_quins {
28                        break;
29                    }
30                }
31                return Ok(out);
32            }
33            let mut out = Vec::new();
34            sample_volume_blocks(&volume, max_quins, &mut out)?;
35            return Ok(out);
36        }
37
38        if let Ok(quins) = crate::q42_reader::read_q42_quins(path) {
39            return Ok(stride_sample(&quins, max_quins));
40        }
41
42        use memmap2::MmapOptions;
43        const QUIN_SIZE: usize = std::mem::size_of::<NQuin>();
44        let file = File::open(path)?;
45        let mmap = unsafe { MmapOptions::new().map(&file)? };
46        let len = mmap.len();
47        if len % QUIN_SIZE != 0 {
48            return Err(format!(
49                "File size {} is not a multiple of NQuin ({} bytes)",
50                len, QUIN_SIZE
51            )
52            .into());
53        }
54        let count = len / QUIN_SIZE;
55        let quins: &[NQuin] =
56            unsafe { std::slice::from_raw_parts(mmap.as_ptr() as *const NQuin, count) };
57        Ok(stride_sample(quins, max_quins))
58    }
59    #[cfg(target_arch = "wasm32")]
60    {
61        let _ = file_path;
62        let _ = max_quins;
63        Err("mmap_sample_quins is not available on wasm32".into())
64    }
65}
66
67#[cfg(not(target_arch = "wasm32"))]
68fn sample_volume_blocks(
69    volume: &crate::q42_volume::Q42Volume,
70    max_quins: usize,
71    out: &mut Vec<NQuin>,
72) -> Result<(), Box<dyn std::error::Error>> {
73    use crate::q42_volume::{decode_superblock_quins, SUPERBLOCK_SIZE};
74    let blocks = volume.block_count() as usize;
75    if blocks == 0 || max_quins == 0 {
76        return Ok(());
77    }
78    let mut decoded = [0u8; SUPERBLOCK_SIZE];
79    let stride = (blocks / max_quins).max(1);
80    let mut index = 0usize;
81    while out.len() < max_quins && index < blocks {
82        volume.read_superblock_into(index, &mut decoded)?;
83        for quin in decode_superblock_quins(&decoded)? {
84            if quin.subject != 0 || quin.predicate != 0 {
85                out.push(quin);
86                if out.len() >= max_quins {
87                    break;
88                }
89            }
90        }
91        index = index.saturating_add(stride);
92    }
93    Ok(())
94}
95
96#[cfg(not(target_arch = "wasm32"))]
97fn collect_subject(
98    volume: &crate::q42_volume::Q42Volume,
99    subject_id: u64,
100    out: &mut Vec<NQuin>,
101) -> Result<(), Box<dyn std::error::Error>> {
102    use crate::q42_volume::{decode_superblock_quins, SUPERBLOCK_SIZE};
103    let mut decoded = [0u8; SUPERBLOCK_SIZE];
104    for index in 0..volume.block_count() as usize {
105        volume.read_superblock_into(index, &mut decoded)?;
106        for quin in decode_superblock_quins(&decoded)? {
107            if quin.subject == subject_id {
108                out.push(quin);
109            }
110        }
111    }
112    Ok(())
113}
114
115fn stride_sample(quins: &[NQuin], max_quins: usize) -> Vec<NQuin> {
116    if max_quins == 0 || quins.is_empty() {
117        return Vec::new();
118    }
119    let cap = max_quins.min(quins.len());
120    let stride = (quins.len() / cap).max(1);
121    let mut out = Vec::with_capacity(cap);
122    let mut idx = 0usize;
123    while out.len() < cap && idx < quins.len() {
124        let quin = quins[idx];
125        if quin.subject != 0 || quin.predicate != 0 {
126            out.push(quin);
127        }
128        idx += stride;
129    }
130    out
131}
132
133/// Query a `.q42` for Quins whose `subject` matches `subject_id`.
134/// Unified v3 volumes are decoded; flat packed files remain a fallback.
135pub fn mmap_query_subject(
136    file_path: &str,
137    subject_id: u64,
138) -> Result<Vec<NQuin>, Box<dyn std::error::Error>> {
139    #[cfg(not(target_arch = "wasm32"))]
140    {
141        use crate::q42_volume::{Q42Volume, Q42VolumeSet};
142
143        let path = Path::new(file_path);
144        if crate::q42_volume::is_unified_volume(path)? {
145            let volume = Q42Volume::open(path)?;
146            if volume.volume_manifest()?.is_some() {
147                let set = Q42VolumeSet::open_root(path)?;
148                let mut out = Vec::new();
149                for segment in set.segments() {
150                    collect_subject(&segment, subject_id, &mut out)?;
151                }
152                return Ok(out);
153            }
154            let mut out = Vec::new();
155            collect_subject(&volume, subject_id, &mut out)?;
156            return Ok(out);
157        }
158
159        if let Ok(quins) = crate::q42_reader::read_q42_quins(path) {
160            return Ok(quins
161                .into_iter()
162                .filter(|q| q.subject == subject_id)
163                .collect());
164        }
165
166        use memmap2::MmapOptions;
167        const QUIN_SIZE: usize = std::mem::size_of::<NQuin>();
168        let file = File::open(path)?;
169        let mmap = unsafe { MmapOptions::new().map(&file)? };
170        let len = mmap.len();
171        if len % QUIN_SIZE != 0 {
172            return Err(format!(
173                "File size {} is not a multiple of NQuin ({} bytes)",
174                len, QUIN_SIZE
175            )
176            .into());
177        }
178        let count = len / QUIN_SIZE;
179        let quins: &[NQuin] =
180            unsafe { std::slice::from_raw_parts(mmap.as_ptr() as *const NQuin, count) };
181        Ok(quins
182            .iter()
183            .filter(|q| q.subject == subject_id)
184            .copied()
185            .collect())
186    }
187    #[cfg(target_arch = "wasm32")]
188    {
189        let _ = file_path;
190        let _ = subject_id;
191        Err("mmap_query_subject is not available on wasm32".into())
192    }
193}
194
195/// Telemetry counters for `lazy_superblock_query`.
196pub struct TelemetryHook {
197    pub blocks_loaded: usize,
198    pub bytes_decompressed: usize,
199    /// Reserved for future WebRTC P2P streaming telemetry.
200    pub remote_blocks_streamed: usize,
201}
202
203/// Reads a SuperBlock file lazily: unified v3 volumes decode selected blocks;
204/// legacy 16-byte framed transport remains a fallback (O(1) seek on skip).
205pub fn lazy_superblock_query(
206    file_path: &str,
207    target_percent: u8,
208) -> Result<TelemetryHook, Box<dyn std::error::Error>> {
209    let start_time = Instant::now();
210    let path = Path::new(file_path);
211    let mut telemetry = TelemetryHook {
212        blocks_loaded: 0,
213        bytes_decompressed: 0,
214        remote_blocks_streamed: 0,
215    };
216
217    #[cfg(not(target_arch = "wasm32"))]
218    if crate::q42_volume::is_unified_volume(path).ok() == Some(true) {
219        use crate::q42_volume::{Q42Volume, SUPERBLOCK_SIZE};
220        let volume = Q42Volume::open(path)?;
221        let mut decoded = [0u8; SUPERBLOCK_SIZE];
222        for index in 0..volume.block_count() {
223            let is_relevant = (index % 100) < target_percent as u64;
224            if !is_relevant {
225                continue;
226            }
227            let n = volume.read_superblock_into(index as usize, &mut decoded)?;
228            telemetry.blocks_loaded += 1;
229            telemetry.bytes_decompressed += n;
230        }
231        let _duration = start_time.elapsed();
232        return Ok(telemetry);
233    }
234
235    let mut file = File::open(path)?;
236    let file_len = file.metadata()?.len();
237    let mut offset = 0u64;
238    let mut block_index = 0u64;
239
240    while offset < file_len {
241        let mut header = [0u8; 16];
242        if file.read_exact(&mut header).is_err() {
243            break;
244        }
245        offset += 16;
246
247        let _block_id = u64::from_le_bytes(header[0..8].try_into().unwrap());
248        let compressed_len = u32::from_le_bytes(header[8..12].try_into().unwrap()) as usize;
249        let uncompressed_len = u32::from_le_bytes(header[12..16].try_into().unwrap()) as usize;
250
251        let is_relevant = (block_index % 100) < target_percent as u64;
252
253        if is_relevant {
254            let mut compressed_buf = vec![0u8; compressed_len];
255            file.read_exact(&mut compressed_buf)?;
256            telemetry.blocks_loaded += 1;
257            let _uncompressed = lz4_flex::decompress_size_prepended(&compressed_buf)?;
258            telemetry.bytes_decompressed += uncompressed_len;
259        } else {
260            file.seek(SeekFrom::Current(compressed_len as i64))?;
261        }
262
263        offset += compressed_len as u64;
264        block_index += 1;
265    }
266
267    let _duration = start_time.elapsed();
268
269    Ok(telemetry)
270}
271
272/// Filter a slice of NQuin by context hash
273pub fn filter_by_context(quins: &[NQuin], context_hash: u64) -> Vec<NQuin> {
274    if context_hash == 0 {
275        return quins.to_vec();
276    }
277    quins
278        .iter()
279        .filter(|q| q.context == context_hash)
280        .copied()
281        .collect()
282}
283
284/// Filter a slice of NQuin by multiple context hashes
285pub fn filter_by_contexts(quins: &[NQuin], context_hashes: &[u64]) -> Vec<NQuin> {
286    if context_hashes.is_empty() {
287        return quins.to_vec();
288    }
289    let context_set: std::collections::HashSet<u64> = context_hashes.iter().copied().collect();
290    quins
291        .iter()
292        .filter(|q| context_set.contains(&q.context))
293        .copied()
294        .collect()
295}
296
297/// Count Quins per context hash
298pub fn count_by_context(quins: &[NQuin]) -> std::collections::HashMap<u64, usize> {
299    let mut counts = std::collections::HashMap::new();
300    for quin in quins {
301        *counts.entry(quin.context).or_insert(0) += 1;
302    }
303    counts
304}
305
306/// Get unique context hashes from a slice of NQuin
307pub fn unique_contexts(quins: &[NQuin]) -> Vec<u64> {
308    let mut contexts = std::collections::HashSet::new();
309    for quin in quins {
310        contexts.insert(quin.context);
311    }
312    contexts.into_iter().collect()
313}
314
315/// Filter Quins by context and subject
316pub fn filter_by_context_and_subject(
317    quins: &[NQuin],
318    context_hash: u64,
319    subject: u64,
320) -> Vec<NQuin> {
321    quins
322        .iter()
323        .filter(|q| (context_hash == 0 || q.context == context_hash) && q.subject == subject)
324        .copied()
325        .collect()
326}
327
328/// Filter Quins by context and predicate
329pub fn filter_by_context_and_predicate(
330    quins: &[NQuin],
331    context_hash: u64,
332    predicate: u64,
333) -> Vec<NQuin> {
334    quins
335        .iter()
336        .filter(|q| (context_hash == 0 || q.context == context_hash) && q.predicate == predicate)
337        .copied()
338        .collect()
339}
340
341/// Filter Quins by context and object
342pub fn filter_by_context_and_object(quins: &[NQuin], context_hash: u64, object: u64) -> Vec<NQuin> {
343    quins
344        .iter()
345        .filter(|q| (context_hash == 0 || q.context == context_hash) && q.object == object)
346        .copied()
347        .collect()
348}
349
350#[cfg(test)]
351mod volume_query_tests {
352    use super::*;
353    use crate::q42_volume::write_sorted_quins_volume;
354
355    fn quin(subject: u64, object: u64) -> NQuin {
356        NQuin {
357            subject,
358            predicate: 2,
359            object,
360            context: 0,
361            metadata: 0,
362            parity: NQuin::calculate_parity(subject, 2, object, 0, 0),
363        }
364    }
365
366    #[test]
367    fn sample_and_subject_query_read_unified_v3() {
368        let dir = tempfile::tempdir().unwrap();
369        let path = dir.path().join("graph.q42");
370        write_sorted_quins_volume(&path, &[quin(7, 1), quin(9, 2), quin(7, 3)]).unwrap();
371        let sampled = mmap_sample_quins(path.to_str().unwrap(), 8).unwrap();
372        assert_eq!(sampled.len(), 3);
373        let hits = mmap_query_subject(path.to_str().unwrap(), 7).unwrap();
374        assert_eq!(hits.len(), 2);
375        assert!(hits.iter().all(|q| q.subject == 7));
376        let telemetry = lazy_superblock_query(path.to_str().unwrap(), 100).unwrap();
377        assert_eq!(telemetry.blocks_loaded, 1);
378        assert!(telemetry.bytes_decompressed >= crate::q42_volume::SUPERBLOCK_SIZE);
379    }
380}
381
382#[cfg(test)]
383mod context_tests {
384    use super::*;
385
386    #[test]
387    fn test_filter_by_context() {
388        let quins = vec![
389            NQuin {
390                subject: 1,
391                predicate: 2,
392                object: 3,
393                context: 100,
394                metadata: 0,
395                parity: 0,
396            },
397            NQuin {
398                subject: 4,
399                predicate: 5,
400                object: 6,
401                context: 200,
402                metadata: 0,
403                parity: 0,
404            },
405            NQuin {
406                subject: 7,
407                predicate: 8,
408                object: 9,
409                context: 100,
410                metadata: 0,
411                parity: 0,
412            },
413        ];
414
415        let filtered = filter_by_context(&quins, 100);
416        assert_eq!(filtered.len(), 2);
417        assert_eq!(filtered[0].context, 100);
418        assert_eq!(filtered[1].context, 100);
419    }
420
421    #[test]
422    fn test_filter_by_context_wildcard() {
423        let quins = vec![
424            NQuin {
425                subject: 1,
426                predicate: 2,
427                object: 3,
428                context: 100,
429                metadata: 0,
430                parity: 0,
431            },
432            NQuin {
433                subject: 4,
434                predicate: 5,
435                object: 6,
436                context: 200,
437                metadata: 0,
438                parity: 0,
439            },
440        ];
441
442        let filtered = filter_by_context(&quins, 0);
443        assert_eq!(filtered.len(), 2);
444    }
445
446    #[test]
447    fn test_count_by_context() {
448        let quins = vec![
449            NQuin {
450                subject: 1,
451                predicate: 2,
452                object: 3,
453                context: 100,
454                metadata: 0,
455                parity: 0,
456            },
457            NQuin {
458                subject: 4,
459                predicate: 5,
460                object: 6,
461                context: 200,
462                metadata: 0,
463                parity: 0,
464            },
465            NQuin {
466                subject: 7,
467                predicate: 8,
468                object: 9,
469                context: 100,
470                metadata: 0,
471                parity: 0,
472            },
473        ];
474
475        let counts = count_by_context(&quins);
476        assert_eq!(counts.get(&100), Some(&2));
477        assert_eq!(counts.get(&200), Some(&1));
478    }
479
480    #[test]
481    fn test_filter_by_context_and_subject() {
482        let quins = vec![
483            NQuin {
484                subject: 1,
485                predicate: 2,
486                object: 3,
487                context: 100,
488                metadata: 0,
489                parity: 0,
490            },
491            NQuin {
492                subject: 1,
493                predicate: 5,
494                object: 6,
495                context: 200,
496                metadata: 0,
497                parity: 0,
498            },
499            NQuin {
500                subject: 7,
501                predicate: 8,
502                object: 9,
503                context: 100,
504                metadata: 0,
505                parity: 0,
506            },
507        ];
508
509        let filtered = filter_by_context_and_subject(&quins, 100, 1);
510        assert_eq!(filtered.len(), 1);
511        assert_eq!(filtered[0].subject, 1);
512        assert_eq!(filtered[0].context, 100);
513    }
514}
qualia_core_db/query/query_engine.rs

qualia_core_db/query/
query_engine.rs
