oxirs_arq/streaming/
types.rs

1//! Auto-generated module
2//!
3//! 🤖 Generated with [SplitRS](https://github.com/cool-japan/splitrs)
4
5use crate::algebra::{Binding, Expression, Solution, Term, TriplePattern, Variable};
6use anyhow::{anyhow, Result};
7use oxirs_core::model::NamedNode;
8use serde::{Deserialize, Serialize};
9use std::collections::{HashMap, VecDeque};
10use std::io::BufRead;
11use std::path::PathBuf;
12use std::sync::{Arc, Mutex};
13use std::time::{Duration, Instant};
14use tracing::{debug, info, warn};
15
16use super::functions::{evaluate_literal_as_boolean, DataStream};
17
18/// Statistics for data streams
19#[derive(Debug, Clone, Default)]
20pub struct StreamStats {
21    pub rows_processed: usize,
22    pub bytes_processed: usize,
23    pub processing_time: Duration,
24    pub spill_operations: usize,
25    pub cache_hits: usize,
26    pub cache_misses: usize,
27}
28/// Streaming pattern scan for large result sets with spilling support
29pub struct StreamingPatternScan {
30    pattern: TriplePattern,
31    memory_monitor: Arc<MemoryMonitor>,
32    pub(super) spill_manager: Arc<Mutex<SpillManager>>,
33    config: StreamingConfig,
34    pub(super) current_batch: Vec<Solution>,
35    pub(super) batch_index: usize,
36    pub(super) total_results: usize,
37    pub(super) spilled_batches: Vec<String>,
38}
39impl StreamingPatternScan {
40    pub fn new(
41        pattern: TriplePattern,
42        memory_monitor: Arc<MemoryMonitor>,
43        spill_manager: Arc<Mutex<SpillManager>>,
44        config: StreamingConfig,
45    ) -> Result<Self> {
46        Ok(Self {
47            pattern,
48            memory_monitor,
49            spill_manager,
50            config,
51            current_batch: Vec::new(),
52            batch_index: 0,
53            total_results: 0,
54            spilled_batches: Vec::new(),
55        })
56    }
57    /// Generate solutions for the pattern (simplified simulation)
58    pub(super) fn generate_pattern_solutions(&mut self) -> Result<Vec<Solution>> {
59        let mut solutions = Vec::new();
60        let solution_count = match (
61            matches!(self.pattern.subject, Term::Variable(_)),
62            matches!(self.pattern.predicate, Term::Variable(_)),
63            matches!(self.pattern.object, Term::Variable(_)),
64        ) {
65            (true, true, true) => self.config.batch_size * 10,
66            (false, true, true) => self.config.batch_size * 5,
67            (true, false, true) => self.config.batch_size * 3,
68            (true, true, false) => self.config.batch_size * 5,
69            (false, false, true) => self.config.batch_size * 2,
70            (false, true, false) => self.config.batch_size,
71            (true, false, false) => self.config.batch_size * 2,
72            (false, false, false) => 1,
73        };
74        for i in 0..solution_count.min(self.config.batch_size) {
75            let mut binding = Binding::new();
76            for var in self.pattern.variables() {
77                let value = Term::Iri(
78                    NamedNode::new(format!("http://example.org/resource_{i}"))
79                        .expect("generated URL should be valid"),
80                );
81                binding.insert(var, value);
82            }
83            if !binding.is_empty() {
84                solutions.push(vec![binding]);
85            }
86        }
87        Ok(solutions)
88    }
89    /// Check if spilling is needed based on memory pressure
90    pub(super) fn should_spill(&self) -> bool {
91        let current_usage = self.memory_monitor.get_current_usage();
92        let max_usage = self
93            .memory_monitor
94            .inner
95            .lock()
96            .expect("lock poisoned")
97            .max_allowed;
98        (current_usage as f64 / max_usage as f64) > self.config.spill_threshold
99    }
100    /// Spill current batch to disk
101    pub(super) fn spill_current_batch(&mut self) -> Result<()> {
102        if !self.current_batch.is_empty() {
103            let spill_id = self
104                .spill_manager
105                .lock()
106                .expect("lock poisoned")
107                .spill_data(&self.current_batch, SpillDataType::Solutions)?;
108            self.spilled_batches.push(spill_id);
109            self.current_batch.clear();
110            debug!("Spilled batch {} for pattern scan", self.batch_index);
111        }
112        Ok(())
113    }
114}
115/// Buffered pattern scan for smaller result sets
116pub struct BufferedPatternScan {
117    pattern: TriplePattern,
118    pub(super) batch_size: usize,
119    pub(super) solutions: Vec<Solution>,
120    pub(super) current_index: usize,
121    pub(super) exhausted: bool,
122}
123impl BufferedPatternScan {
124    pub fn new(pattern: TriplePattern, batch_size: usize) -> Result<Self> {
125        let mut scan = Self {
126            pattern,
127            batch_size,
128            solutions: Vec::new(),
129            current_index: 0,
130            exhausted: false,
131        };
132        scan.generate_all_solutions()?;
133        Ok(scan)
134    }
135    /// Generate all solutions for the pattern upfront
136    pub(super) fn generate_all_solutions(&mut self) -> Result<()> {
137        let solution_count = match (
138            matches!(self.pattern.subject, Term::Variable(_)),
139            matches!(self.pattern.predicate, Term::Variable(_)),
140            matches!(self.pattern.object, Term::Variable(_)),
141        ) {
142            (true, true, true) => 1000,
143            (false, true, true) => 100,
144            (true, false, true) => 50,
145            (true, true, false) => 100,
146            (false, false, true) => 20,
147            (false, true, false) => 10,
148            (true, false, false) => 20,
149            (false, false, false) => 1,
150        };
151        for i in 0..solution_count {
152            let mut binding = Binding::new();
153            for var in self.pattern.variables() {
154                let value = Term::Iri(
155                    NamedNode::new(format!("http://example.org/item_{i}"))
156                        .expect("generated URL should be valid"),
157                );
158                binding.insert(var, value);
159            }
160            if !binding.is_empty() {
161                self.solutions.push(vec![binding]);
162            }
163        }
164        Ok(())
165    }
166}
167/// Spill manager for handling memory overflow
168pub struct SpillManager {
169    pub(super) spill_directory: PathBuf,
170    pub(super) active_spills: HashMap<String, SpillInfo>,
171    pub(super) spill_counter: usize,
172    pub(super) compression_enabled: bool,
173    pub(super) compression_level: u32,
174}
175impl SpillManager {
176    pub(super) fn new(spill_directory: PathBuf, compression_level: u32) -> Result<Self> {
177        std::fs::create_dir_all(&spill_directory)?;
178        Ok(Self {
179            spill_directory,
180            active_spills: HashMap::new(),
181            spill_counter: 0,
182            compression_enabled: compression_level > 0,
183            compression_level,
184        })
185    }
186    pub(super) fn spill_data<T: Serialize>(
187        &mut self,
188        data: &T,
189        data_type: SpillDataType,
190    ) -> Result<String> {
191        self.spill_counter += 1;
192        let spill_id = format!("spill_{c}", c = self.spill_counter);
193        let file_path = self.spill_directory.join(format!("{spill_id}.bin"));
194        let start_time = Instant::now();
195        let serialized = oxicode::serde::encode_to_vec(&data, oxicode::config::standard())?;
196        let original_size = serialized.len();
197        let final_data = if self.compression_enabled {
198            self.compress_data(&serialized)?
199        } else {
200            serialized
201        };
202        std::fs::write(&file_path, &final_data)?;
203        let spill_info = SpillInfo {
204            file_path: file_path.clone(),
205            original_size,
206            compressed_size: final_data.len(),
207            data_type,
208            creation_time: start_time,
209            access_count: 0,
210        };
211        self.active_spills.insert(spill_id.clone(), spill_info);
212        info!("Spilled {} bytes to {}", original_size, file_path.display());
213        Ok(spill_id)
214    }
215    pub(super) fn read_spill<T: for<'de> Deserialize<'de>>(&mut self, spill_id: &str) -> Result<T> {
216        let spill_info = self
217            .active_spills
218            .get_mut(spill_id)
219            .ok_or_else(|| anyhow!("Spill not found: {}", spill_id))?;
220        spill_info.access_count += 1;
221        let data = std::fs::read(&spill_info.file_path)?;
222        let decompressed = if self.compression_enabled {
223            self.decompress_data(&data)?
224        } else {
225            data
226        };
227        let deserialized =
228            oxicode::serde::decode_from_slice(&decompressed, oxicode::config::standard())
229                .map(|(v, _)| v)?;
230        Ok(deserialized)
231    }
232    pub(super) fn delete_spill(&mut self, spill_id: &str) -> Result<()> {
233        if let Some(spill_info) = self.active_spills.remove(spill_id) {
234            std::fs::remove_file(&spill_info.file_path)?;
235            debug!("Deleted spill file: {}", spill_info.file_path.display());
236        }
237        Ok(())
238    }
239    pub(super) fn cleanup_all(&mut self) -> Result<()> {
240        for spill_id in self.active_spills.keys().cloned().collect::<Vec<_>>() {
241            self.delete_spill(&spill_id)?;
242        }
243        Ok(())
244    }
245    pub(super) fn compress_data(&self, data: &[u8]) -> Result<Vec<u8>> {
246        // Gzip (RFC 1952) compression via Pure-Rust oxiarc-deflate.
247        // oxiarc-deflate takes a u8 level (0-9); clamp the configured u32 level.
248        let level = self.compression_level.min(9) as u8;
249        Ok(oxiarc_deflate::gzip_compress(data, level)?)
250    }
251    pub(super) fn decompress_data(&self, data: &[u8]) -> Result<Vec<u8>> {
252        Ok(oxiarc_deflate::gzip_decompress(data)?)
253    }
254}
255/// Streaming aggregation operator
256pub struct StreamingAggregation {
257    #[allow(dead_code)]
258    input_stream: Box<dyn DataStream>,
259    #[allow(dead_code)]
260    group_variables: Vec<Variable>,
261    #[allow(dead_code)]
262    aggregation_functions: Vec<AggregationFunction>,
263    #[allow(dead_code)]
264    partial_results: HashMap<String, AggregationState>,
265    #[allow(dead_code)]
266    memory_monitor: Arc<MemoryMonitor>,
267    #[allow(dead_code)]
268    spill_manager: Arc<Mutex<SpillManager>>,
269    #[allow(dead_code)]
270    config: StreamingConfig,
271}
272/// Types of data that can be spilled
273#[derive(Debug, Clone)]
274pub enum SpillDataType {
275    Solutions,
276    HashTable,
277    SortBuffer,
278    IntermediateResults,
279    Index,
280}
281/// Memory monitoring and management
282pub struct MemoryMonitor {
283    pub(super) inner: Arc<Mutex<MemoryMonitorInner>>,
284}
285impl MemoryMonitor {
286    pub(super) fn new(max_allowed: usize) -> Self {
287        Self {
288            inner: Arc::new(Mutex::new(MemoryMonitorInner {
289                current_usage: 0,
290                peak_usage: 0,
291                max_allowed,
292                allocation_history: VecDeque::new(),
293            })),
294        }
295    }
296    pub(super) fn allocate(&self, size: usize, operation: &str) -> bool {
297        let mut inner = self.inner.lock().expect("lock poisoned");
298        if inner.current_usage + size > inner.max_allowed {
299            return false;
300        }
301        inner.current_usage += size;
302        inner.peak_usage = inner.peak_usage.max(inner.current_usage);
303        inner.allocation_history.push_back(MemoryAllocation {
304            timestamp: Instant::now(),
305            size,
306            operation: operation.to_string(),
307            freed: false,
308        });
309        if inner.allocation_history.len() > 10000 {
310            inner.allocation_history.pop_front();
311        }
312        true
313    }
314    pub(super) fn deallocate(&self, size: usize) {
315        let mut inner = self.inner.lock().expect("lock poisoned");
316        inner.current_usage = inner.current_usage.saturating_sub(size);
317    }
318    #[allow(dead_code)]
319    pub(super) fn should_spill(&self, threshold: f64) -> bool {
320        let inner = self.inner.lock().expect("lock poisoned");
321        inner.current_usage as f64 > inner.max_allowed as f64 * threshold
322    }
323    #[allow(dead_code)]
324    pub(super) fn get_usage_percentage(&self) -> f64 {
325        let inner = self.inner.lock().expect("lock poisoned");
326        inner.current_usage as f64 / inner.max_allowed as f64
327    }
328    pub(super) fn get_current_usage(&self) -> usize {
329        let inner = self.inner.lock().expect("lock poisoned");
330        inner.current_usage
331    }
332}
333/// Memory allocation record
334#[derive(Debug, Clone)]
335pub struct MemoryAllocation {
336    pub timestamp: Instant,
337    pub size: usize,
338    pub operation: String,
339    pub freed: bool,
340}
341/// Information about a spilled data structure
342#[derive(Debug, Clone)]
343pub struct SpillInfo {
344    pub file_path: PathBuf,
345    pub original_size: usize,
346    pub compressed_size: usize,
347    pub data_type: SpillDataType,
348    pub creation_time: Instant,
349    pub access_count: usize,
350}
351pub struct StreamingUnion {
352    pub(super) left: Box<dyn DataStream>,
353    pub(super) right: Box<dyn DataStream>,
354    pub(super) left_exhausted: bool,
355}
356impl StreamingUnion {
357    pub(super) fn new(left: Box<dyn DataStream>, right: Box<dyn DataStream>) -> Self {
358        Self {
359            left,
360            right,
361            left_exhausted: false,
362        }
363    }
364}
365pub struct StreamingMinus {
366    pub(super) left: Box<dyn DataStream>,
367    pub(super) right: Box<dyn DataStream>,
368    #[allow(dead_code)]
369    memory_monitor: Arc<MemoryMonitor>,
370    #[allow(dead_code)]
371    spill_manager: Arc<Mutex<SpillManager>>,
372}
373impl StreamingMinus {
374    #[allow(dead_code)]
375    pub(super) fn new(
376        left: Box<dyn DataStream>,
377        right: Box<dyn DataStream>,
378        memory_monitor: Arc<MemoryMonitor>,
379        spill_manager: Arc<Mutex<SpillManager>>,
380    ) -> Self {
381        Self {
382            left,
383            right,
384            memory_monitor,
385            spill_manager,
386        }
387    }
388}
389impl StreamingMinus {
390    /// Check if two solutions are compatible according to SPARQL MINUS semantics
391    /// Two solutions are compatible if they don't disagree on any shared variables
392    pub(super) fn solutions_compatible(&self, left: &Solution, right: &Solution) -> bool {
393        let left_binding = match left.first() {
394            Some(binding) => binding,
395            None => return false,
396        };
397        let right_binding = match right.first() {
398            Some(binding) => binding,
399            None => return false,
400        };
401        for (var, left_term) in left_binding.iter() {
402            if let Some(right_term) = right_binding.get(var) {
403                if left_term != right_term {
404                    return false;
405                }
406            }
407        }
408        true
409    }
410}
411pub struct StreamingSelection {
412    pub(super) input: Box<dyn DataStream>,
413    pub(super) condition: crate::algebra::Expression,
414}
415impl StreamingSelection {
416    #[allow(dead_code)]
417    pub(super) fn new(input: Box<dyn DataStream>, condition: crate::algebra::Expression) -> Self {
418        Self { input, condition }
419    }
420}
421impl StreamingSelection {
422    /// Evaluate the filter condition against a solution
423    pub(super) fn evaluate_condition(&self, solution: &Solution) -> Result<bool> {
424        use crate::algebra::{BinaryOperator, UnaryOperator};
425        let binding = match solution.first() {
426            Some(binding) => binding,
427            None => return Ok(false),
428        };
429        match &self.condition {
430            Expression::Variable(var) => Ok(binding.contains_key(var)),
431            Expression::Literal(literal) => evaluate_literal_as_boolean(literal),
432            Expression::Binary { op, left, right } => match op {
433                BinaryOperator::Equal => {
434                    let left_val = self.evaluate_expression(left, binding)?;
435                    let right_val = self.evaluate_expression(right, binding)?;
436                    Ok(left_val == right_val)
437                }
438                BinaryOperator::NotEqual => {
439                    let left_val = self.evaluate_expression(left, binding)?;
440                    let right_val = self.evaluate_expression(right, binding)?;
441                    Ok(left_val != right_val)
442                }
443                BinaryOperator::And => {
444                    let left_result = self.evaluate_condition_expr(left, binding)?;
445                    let right_result = self.evaluate_condition_expr(right, binding)?;
446                    Ok(left_result && right_result)
447                }
448                BinaryOperator::Or => {
449                    let left_result = self.evaluate_condition_expr(left, binding)?;
450                    let right_result = self.evaluate_condition_expr(right, binding)?;
451                    Ok(left_result || right_result)
452                }
453                _ => {
454                    warn!("Unsupported binary operator in filter: {:?}", op);
455                    Ok(true)
456                }
457            },
458            Expression::Unary { op, operand } => match op {
459                UnaryOperator::Not => {
460                    let result = self.evaluate_condition_expr(operand, binding)?;
461                    Ok(!result)
462                }
463                _ => {
464                    warn!("Unsupported unary operator in filter: {:?}", op);
465                    Ok(true)
466                }
467            },
468            Expression::Bound(var) => Ok(binding.contains_key(var)),
469            _ => {
470                warn!("Unsupported expression type in filter, defaulting to true");
471                Ok(true)
472            }
473        }
474    }
475    /// Helper to evaluate sub-expressions that return boolean values
476    pub(super) fn evaluate_condition_expr(
477        &self,
478        expr: &Expression,
479        binding: &Binding,
480    ) -> Result<bool> {
481        let temp_solution = vec![binding.clone()];
482        let temp_filter = StreamingSelection {
483            input: Box::new(EmptyStream::new()),
484            condition: expr.clone(),
485        };
486        temp_filter.evaluate_condition(&temp_solution)
487    }
488    /// Helper to evaluate expressions that return Term values
489    pub(super) fn evaluate_expression(
490        &self,
491        expr: &Expression,
492        binding: &Binding,
493    ) -> Result<Option<Term>> {
494        match expr {
495            Expression::Variable(var) => Ok(binding.get(var).cloned()),
496            Expression::Literal(literal) => Ok(Some(Term::Literal(literal.clone()))),
497            _ => Ok(None),
498        }
499    }
500}
501/// Streaming statistics
502#[derive(Debug, Clone, Default)]
503pub struct StreamingStats {
504    pub total_memory_used: usize,
505    pub peak_memory_used: usize,
506    pub spill_operations: usize,
507    pub total_spill_size: usize,
508    pub total_execution_time: Duration,
509    pub rows_processed: usize,
510    pub cache_hit_rate: f64,
511}
512/// Streaming hash join implementation
513pub struct StreamingHashJoin {
514    pub(super) left_stream: Box<dyn DataStream>,
515    pub(super) right_stream: Box<dyn DataStream>,
516    join_variables: Vec<Variable>,
517    pub(super) hash_table: HashMap<String, Vec<Solution>>,
518    pub(super) memory_monitor: Arc<MemoryMonitor>,
519    spill_manager: Arc<Mutex<SpillManager>>,
520    #[allow(dead_code)]
521    config: StreamingConfig,
522    pub(super) left_exhausted: bool,
523    pub(super) current_batch: Option<Vec<Solution>>,
524    spilled_partitions: Vec<String>,
525    #[allow(dead_code)]
526    current_spill_index: usize,
527}
528impl StreamingHashJoin {
529    pub(super) fn new(
530        left: Box<dyn DataStream>,
531        right: Box<dyn DataStream>,
532        join_variables: Vec<Variable>,
533        memory_monitor: Arc<MemoryMonitor>,
534        spill_manager: Arc<Mutex<SpillManager>>,
535        config: StreamingConfig,
536    ) -> Result<Self> {
537        Ok(Self {
538            left_stream: left,
539            right_stream: right,
540            join_variables,
541            hash_table: HashMap::new(),
542            memory_monitor,
543            spill_manager,
544            config,
545            left_exhausted: false,
546            current_batch: None,
547            spilled_partitions: Vec::new(),
548            current_spill_index: 0,
549        })
550    }
551}
552impl StreamingHashJoin {
553    pub(super) fn extract_join_key(&self, solution: &Solution) -> String {
554        self.join_variables
555            .iter()
556            .map(|var| {
557                Self::get_solution_value(solution, var)
558                    .map(|term| format!("{term:?}"))
559                    .unwrap_or_else(|| "NULL".to_string())
560            })
561            .collect::<Vec<_>>()
562            .join("|")
563    }
564    /// Helper function to get a value from a solution
565    pub(super) fn get_solution_value<'a>(
566        solution: &'a Solution,
567        var: &Variable,
568    ) -> Option<&'a Term> {
569        solution.first().and_then(|binding| binding.get(var))
570    }
571    pub(super) fn join_solutions(&self, left: &Solution, right: &Solution) -> Option<Solution> {
572        for var in &self.join_variables {
573            let left_val = Self::get_solution_value(left, var);
574            let right_val = Self::get_solution_value(right, var);
575            match (left_val, right_val) {
576                (Some(l), Some(r)) if l != r => return None,
577                _ => {}
578            }
579        }
580        let mut result_binding = Binding::new();
581        if let Some(left_binding) = left.first() {
582            for (var, term) in left_binding.iter() {
583                result_binding.insert(var.clone(), term.clone());
584            }
585        }
586        if let Some(right_binding) = right.first() {
587            for (var, term) in right_binding.iter() {
588                result_binding.insert(var.clone(), term.clone());
589            }
590        }
591        Some(vec![result_binding])
592    }
593    /// Spill current hash table to disk to free memory
594    pub(super) fn spill_hash_table(&mut self) -> Result<()> {
595        if self.hash_table.is_empty() {
596            return Ok(());
597        }
598        let spill_id = self
599            .spill_manager
600            .lock()
601            .expect("lock poisoned")
602            .spill_data(&self.hash_table, SpillDataType::HashTable)?;
603        self.spilled_partitions.push(spill_id);
604        let total_size: usize = self
605            .hash_table
606            .iter()
607            .map(|(key, solutions)| key.len() + solutions.len() * std::mem::size_of::<Solution>())
608            .sum();
609        self.hash_table.clear();
610        self.memory_monitor.deallocate(total_size);
611        debug!("Spilled hash table partition with {} entries", total_size);
612        Ok(())
613    }
614    /// Load spilled hash table partition back into memory
615    #[allow(dead_code)]
616    pub(super) fn load_spilled_partition(
617        &mut self,
618        spill_id: &str,
619    ) -> Result<HashMap<String, Vec<Solution>>> {
620        let partition: HashMap<String, Vec<Solution>> = self
621            .spill_manager
622            .lock()
623            .expect("lock poisoned")
624            .read_spill(spill_id)?;
625        Ok(partition)
626    }
627}
628pub struct StreamingProjection {
629    pub(super) input: Box<dyn DataStream>,
630    pub(super) variables: Vec<Variable>,
631}
632impl StreamingProjection {
633    #[allow(dead_code)]
634    pub(super) fn new(input: Box<dyn DataStream>, variables: Vec<Variable>) -> Self {
635        Self { input, variables }
636    }
637}
638/// Memory-mapped file stream for large datasets
639pub struct MemoryMappedStream {
640    #[allow(dead_code)]
641    file_path: PathBuf,
642    #[allow(dead_code)]
643    current_position: usize,
644    #[allow(dead_code)]
645    total_size: usize,
646    #[allow(dead_code)]
647    batch_size: usize,
648    #[allow(dead_code)]
649    stats: StreamStats,
650}
651/// Compressed spill file stream
652pub struct CompressedSpillStream {
653    #[allow(dead_code)]
654    file_path: PathBuf,
655    #[allow(dead_code)]
656    reader: Option<Box<dyn BufRead>>,
657    #[allow(dead_code)]
658    batch_size: usize,
659    #[allow(dead_code)]
660    stats: StreamStats,
661}
662/// Internal state for memory monitor
663pub(super) struct MemoryMonitorInner {
664    pub(super) current_usage: usize,
665    pub(super) peak_usage: usize,
666    pub(super) max_allowed: usize,
667    pub(super) allocation_history: VecDeque<MemoryAllocation>,
668}
669/// Configuration for streaming execution
670#[derive(Debug, Clone)]
671pub struct StreamingConfig {
672    /// Maximum memory usage before spilling (bytes)
673    pub max_memory_usage: usize,
674    /// Memory threshold for spilling (0.0 to 1.0)
675    pub spill_threshold: f64,
676    /// Batch size for processing
677    pub batch_size: usize,
678    /// Number of parallel workers
679    pub parallel_workers: usize,
680    /// Compression level for spilled data (0-9)
681    pub compression_level: u32,
682    /// Enable memory mapping for large files
683    pub enable_memory_mapping: bool,
684    /// Buffer size for I/O operations
685    pub io_buffer_size: usize,
686    /// Enable adaptive batch sizing
687    pub adaptive_batching: bool,
688}
689/// Streaming sort-merge join implementation
690pub struct StreamingSortMergeJoin {
691    pub(super) left_stream: Box<dyn DataStream>,
692    pub(super) right_stream: Box<dyn DataStream>,
693    join_variables: Vec<Variable>,
694    pub(super) left_buffer: VecDeque<Solution>,
695    pub(super) right_buffer: VecDeque<Solution>,
696    #[allow(dead_code)]
697    memory_monitor: Arc<MemoryMonitor>,
698    #[allow(dead_code)]
699    spill_manager: Arc<Mutex<SpillManager>>,
700    #[allow(dead_code)]
701    pub(super) config: StreamingConfig,
702}
703impl StreamingSortMergeJoin {
704    pub(super) fn new(
705        left: Box<dyn DataStream>,
706        right: Box<dyn DataStream>,
707        join_variables: Vec<Variable>,
708        memory_monitor: Arc<MemoryMonitor>,
709        spill_manager: Arc<Mutex<SpillManager>>,
710        config: StreamingConfig,
711    ) -> Result<Self> {
712        Ok(Self {
713            left_stream: left,
714            right_stream: right,
715            join_variables,
716            left_buffer: VecDeque::new(),
717            right_buffer: VecDeque::new(),
718            memory_monitor,
719            spill_manager,
720            config,
721        })
722    }
723}
724impl StreamingSortMergeJoin {
725    /// Refill buffers with sorted data from input streams
726    pub(super) fn refill_sorted_buffers(&mut self) -> Result<()> {
727        if self.left_buffer.is_empty() {
728            if let Some(mut batch) = self.left_stream.next_batch()? {
729                batch.sort_by(|a, b| self.compare_solution_keys(a, b));
730                self.left_buffer.extend(batch);
731            }
732        }
733        if self.right_buffer.is_empty() {
734            if let Some(mut batch) = self.right_stream.next_batch()? {
735                batch.sort_by(|a, b| self.compare_solution_keys(a, b));
736                self.right_buffer.extend(batch);
737            }
738        }
739        Ok(())
740    }
741    /// Compare join keys of two solutions
742    pub(super) fn compare_join_keys(
743        &self,
744        left: &Solution,
745        right: &Solution,
746    ) -> std::cmp::Ordering {
747        self.compare_solution_keys(left, right)
748    }
749    /// Compare solutions by their join variable values
750    pub(super) fn compare_solution_keys(
751        &self,
752        left: &Solution,
753        right: &Solution,
754    ) -> std::cmp::Ordering {
755        for var in &self.join_variables {
756            let left_val = Self::get_solution_value(left, var);
757            let right_val = Self::get_solution_value(right, var);
758            match (left_val, right_val) {
759                (Some(left_term), Some(right_term)) => {
760                    let cmp = format!("{left_term:?}").cmp(&format!("{right_term:?}"));
761                    if cmp != std::cmp::Ordering::Equal {
762                        return cmp;
763                    }
764                }
765                (Some(_), None) => return std::cmp::Ordering::Greater,
766                (None, Some(_)) => return std::cmp::Ordering::Less,
767                (None, None) => continue,
768            }
769        }
770        std::cmp::Ordering::Equal
771    }
772    /// Extract join key from solution
773    pub(super) fn extract_join_key(&self, solution: &Solution) -> String {
774        self.join_variables
775            .iter()
776            .map(|var| {
777                Self::get_solution_value(solution, var)
778                    .map(|term| format!("{term:?}"))
779                    .unwrap_or_else(|| "NULL".to_string())
780            })
781            .collect::<Vec<_>>()
782            .join("|")
783    }
784    /// Helper function to get a value from a solution
785    pub(super) fn get_solution_value<'a>(
786        solution: &'a Solution,
787        var: &Variable,
788    ) -> Option<&'a Term> {
789        solution.first().and_then(|binding| binding.get(var))
790    }
791    /// Join two compatible solutions
792    pub(super) fn join_solutions(&self, left: &Solution, right: &Solution) -> Option<Solution> {
793        for var in &self.join_variables {
794            let left_val = Self::get_solution_value(left, var);
795            let right_val = Self::get_solution_value(right, var);
796            match (left_val, right_val) {
797                (Some(l), Some(r)) if l != r => return None,
798                _ => {}
799            }
800        }
801        let mut result_binding = Binding::new();
802        if let Some(left_binding) = left.first() {
803            for (var, term) in left_binding.iter() {
804                result_binding.insert(var.clone(), term.clone());
805            }
806        }
807        if let Some(right_binding) = right.first() {
808            for (var, term) in right_binding.iter() {
809                result_binding.insert(var.clone(), term.clone());
810            }
811        }
812        Some(vec![result_binding])
813    }
814}
815pub struct StreamingSort {
816    pub(super) input: Box<dyn DataStream>,
817    sort_variables: Vec<Variable>,
818    #[allow(dead_code)]
819    memory_monitor: Arc<MemoryMonitor>,
820    pub(super) spill_manager: Arc<Mutex<SpillManager>>,
821    pub(super) config: StreamingConfig,
822    pub(super) sorted_batches: Vec<String>,
823    pub(super) current_batch_index: usize,
824    pub(super) fully_sorted: bool,
825}
826impl StreamingSort {
827    #[allow(dead_code)]
828    pub(super) fn new(
829        input: Box<dyn DataStream>,
830        sort_variables: Vec<Variable>,
831        memory_monitor: Arc<MemoryMonitor>,
832        spill_manager: Arc<Mutex<SpillManager>>,
833        config: StreamingConfig,
834    ) -> Result<Self> {
835        Ok(Self {
836            input,
837            sort_variables,
838            memory_monitor,
839            spill_manager,
840            config,
841            sorted_batches: Vec::new(),
842            current_batch_index: 0,
843            fully_sorted: false,
844        })
845    }
846}
847impl StreamingSort {
848    pub(super) fn compare_solutions(&self, a: &Solution, b: &Solution) -> std::cmp::Ordering {
849        for var in &self.sort_variables {
850            let a_val = StreamingHashJoin::get_solution_value(a, var);
851            let b_val = StreamingHashJoin::get_solution_value(b, var);
852            match (a_val, b_val) {
853                (Some(a_term), Some(b_term)) => {
854                    let cmp = format!("{a_term:?}").cmp(&format!("{b_term:?}"));
855                    if cmp != std::cmp::Ordering::Equal {
856                        return cmp;
857                    }
858                }
859                (Some(_), None) => return std::cmp::Ordering::Greater,
860                (None, Some(_)) => return std::cmp::Ordering::Less,
861                (None, None) => continue,
862            }
863        }
864        std::cmp::Ordering::Equal
865    }
866}
867pub struct EmptyStream {
868    pub(super) exhausted: bool,
869}
870impl EmptyStream {
871    pub(super) fn new() -> Self {
872        Self { exhausted: false }
873    }
874}
875/// Aggregation function types
876#[derive(Debug, Clone)]
877pub enum AggregationFunction {
878    Count,
879    Sum(Variable),
880    Avg(Variable),
881    Min(Variable),
882    Max(Variable),
883    GroupConcat(Variable, Option<String>),
884}
885/// State for aggregation computation
886#[derive(Debug, Clone)]
887pub struct AggregationState {
888    pub count: usize,
889    pub sum: f64,
890    pub min: Option<Term>,
891    pub max: Option<Term>,
892    pub values: Vec<Term>,
893}
oxirs_arq/streaming/types.rs

oxirs_arq/streaming/
types.rs
