kglite 0.14.2

Pure-Rust embedded Cypher knowledge graph engine with in-memory, mmap, and disk storage, and agent-facing schema introspection
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

//! Streaming execution operators for the Cypher generic path.
//!
//! Each operator consumes a [`RowStream`] and produces another. Operators
//! compose into pipelines that mirror what fused operators do today, but
//! generalize to any clause sequence the planner can recognize. The driver
//! in [`super::CypherExecutor::execute`] tries to absorb a contiguous run
//! of clauses into a single pipeline; on success the streaming path runs
//! without materializing intermediate `Vec<ResultRow>` between clauses.
//!
//! # Current scope
//! - The source wraps an already-materialized [`ResultSet`]; pattern matching
//!   itself is not streamed yet.
//! - Streaming aggregate: hash aggregate that builds per-group state while
//!   iterating, replacing the materialize-then-bucket path in
//!   `return_clause::execute_return_with_aggregation`.
//! - Heap top-K: `BinaryHeap<Reverse<…>>` of capacity K, replacing the
//!   full sort + truncate path in `return_clause::execute_order_by` +
//!   `execute_limit`.

use super::super::result::ResultRow;
use super::super::result::ResultSet;

pub mod aggregate;
pub mod heap_top_k;
pub mod pipeline;

/// A typed iterator of result rows with column-name metadata. Used as the
/// composition seam between streaming operators. The underlying iterator is
/// boxed at the seam; concrete operators name their own iterator types and
/// only erase to `dyn` when handing off across module boundaries.
pub struct RowStream<'q> {
    iter: Box<dyn Iterator<Item = Result<ResultRow, String>> + 'q>,
    columns: Vec<String>,
}

impl<'q> RowStream<'q> {
    /// Wrap any iterator yielding `Result<ResultRow, String>`.
    pub fn new<I>(iter: I, columns: Vec<String>) -> Self
    where
        I: Iterator<Item = Result<ResultRow, String>> + 'q,
    {
        RowStream {
            iter: Box::new(iter),
            columns,
        }
    }

    /// Wrap an already-materialized `Vec<ResultRow>`.
    pub fn from_vec(rows: Vec<ResultRow>, columns: Vec<String>) -> Self {
        RowStream::new(rows.into_iter().map(Ok), columns)
    }

    /// Wrap an existing `ResultSet` so a streaming pipeline can splice
    /// onto a materialized prefix.
    pub fn from_result_set(rs: ResultSet) -> Self {
        let columns = rs.columns;
        RowStream::from_vec(rs.rows, columns)
    }

    pub fn columns_owned(&self) -> Vec<String> {
        self.columns.clone()
    }

    /// Drain the stream into a `ResultSet`. Used at the seam where the
    /// streaming pipeline hands its output back to the materialized
    /// executor (e.g. the final `RETURN` projection or a clause the
    /// streaming path doesn't yet absorb).
    pub fn drain(self) -> Result<ResultSet, String> {
        let columns = self.columns;
        let mut rows = Vec::new();
        for row in self.iter {
            rows.push(row?);
        }
        Ok(ResultSet {
            rows,
            columns,
            lazy_return_items: None,
        })
    }
}

impl<'q> Iterator for RowStream<'q> {
    type Item = Result<ResultRow, String>;

    fn next(&mut self) -> Option<Self::Item> {
        self.iter.next()
    }
}