oxirs_arq/streaming/
streamingpatternscan_traits.rs

1//! # StreamingPatternScan - Trait Implementations
2//!
3//! This module contains trait implementations for `StreamingPatternScan`.
4//!
5//! ## Implemented Traits
6//!
7//! - `DataStream`
8//!
9//! 🤖 Generated with [SplitRS](https://github.com/cool-japan/splitrs)
10
11use crate::algebra::Solution;
12use anyhow::Result;
13use std::time::Duration;
14
15use super::functions::DataStream;
16use super::types::{StreamStats, StreamingPatternScan};
17
18impl DataStream for StreamingPatternScan {
19    fn next_batch(&mut self) -> Result<Option<Vec<Solution>>> {
20        if !self.spilled_batches.is_empty() {
21            let spill_id = self.spilled_batches.remove(0);
22            let spilled_solutions: Vec<Solution> = self
23                .spill_manager
24                .lock()
25                .expect("lock poisoned")
26                .read_spill(&spill_id)?;
27            return Ok(Some(spilled_solutions));
28        }
29        if self.batch_index < 10 {
30            let solutions = self.generate_pattern_solutions()?;
31            if solutions.is_empty() {
32                return Ok(None);
33            }
34            if self.should_spill() {
35                self.current_batch = solutions;
36                self.spill_current_batch()?;
37                self.batch_index += 1;
38                return self.next_batch();
39            }
40            self.batch_index += 1;
41            self.total_results += solutions.len();
42            Ok(Some(solutions))
43        } else {
44            Ok(None)
45        }
46    }
47    fn has_more(&self) -> bool {
48        !self.spilled_batches.is_empty() || self.batch_index < 10
49    }
50    fn estimated_size(&self) -> Option<usize> {
51        Some(self.total_results + self.current_batch.len())
52    }
53    fn reset(&mut self) -> Result<()> {
54        self.batch_index = 0;
55        self.total_results = 0;
56        self.current_batch.clear();
57        self.spilled_batches.clear();
58        Ok(())
59    }
60    fn get_stats(&self) -> StreamStats {
61        StreamStats {
62            rows_processed: self.total_results,
63            bytes_processed: 0,
64            processing_time: Duration::from_secs(0),
65            spill_operations: self.spilled_batches.len(),
66            cache_hits: 0,
67            cache_misses: 0,
68        }
69    }
70}
oxirs_arq/streaming/streamingpatternscan_traits.rs

oxirs_arq/streaming/
streamingpatternscan_traits.rs
