oxirs_arq/streaming/
streamingprojection_traits.rs

1//! # StreamingProjection - Trait Implementations
2//!
3//! This module contains trait implementations for `StreamingProjection`.
4//!
5//! ## Implemented Traits
6//!
7//! - `DataStream`
8//!
9//! 🤖 Generated with [SplitRS](https://github.com/cool-japan/splitrs)
10
11use crate::algebra::{Binding, Solution};
12use anyhow::Result;
13
14use super::functions::DataStream;
15use super::types::{StreamStats, StreamingHashJoin, StreamingProjection};
16
17impl DataStream for StreamingProjection {
18    fn next_batch(&mut self) -> Result<Option<Vec<Solution>>> {
19        if let Some(batch) = self.input.next_batch()? {
20            let projected: Vec<Solution> = batch
21                .into_iter()
22                .map(|solution| {
23                    let mut projected_binding = Binding::new();
24                    for var in &self.variables {
25                        if let Some(term) = StreamingHashJoin::get_solution_value(&solution, var) {
26                            projected_binding.insert(var.clone(), term.clone());
27                        }
28                    }
29                    vec![projected_binding]
30                })
31                .collect();
32            Ok(Some(projected))
33        } else {
34            Ok(None)
35        }
36    }
37    fn has_more(&self) -> bool {
38        self.input.has_more()
39    }
40    fn estimated_size(&self) -> Option<usize> {
41        self.input.estimated_size()
42    }
43    fn reset(&mut self) -> Result<()> {
44        self.input.reset()
45    }
46    fn get_stats(&self) -> StreamStats {
47        self.input.get_stats()
48    }
49}
oxirs_arq/streaming/streamingprojection_traits.rs

oxirs_arq/streaming/
streamingprojection_traits.rs
