linkrs_query/executor/streaming/operators/
intersect_build.rs

1//! Sorted build-side tables for WCO intersect execution.
2//!
3//! Each build side maps a bound-node value to the adjacency rows sharing
4//! it. [`IntersectBuild::finish`] sorts every adjacency list by the
5//! intersect column once, so each probe turns into linear merge-style
6//! intersections instead of repeated hashing.
7
8use std::collections::HashMap;
9
10use linkrs_core::Value;
11
12/// Totally ordered sort key extracted from a [`Value`].
13///
14/// Integer node ids compare numerically; strings compare lexicographically;
15/// every other value falls back to its debug rendering. The fallback is
16/// deterministic within a run, which is all the merge intersection needs.
17#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
18pub enum IntersectKey {
19    Int(i64),
20    Str(String),
21    Other(String),
22}
23
24impl From<&Value> for IntersectKey {
25    fn from(value: &Value) -> Self {
26        match value {
27            Value::SmallInt(v) => Self::Int(*v as i64),
28            Value::Int(v) => Self::Int(*v as i64),
29            Value::BigInt(v) => Self::Int(*v),
30            Value::String(s) => Self::Str(s.to_string()),
31            Value::FixedString(s) => Self::Str(s.clone()),
32            Value::Bool(v) => Self::Other(format!("bool:{v}")),
33            other => Self::Other(format!("{other:?}")),
34        }
35    }
36}
37
38/// One build side of a WCO intersect: bound value -> sorted adjacency rows.
39#[derive(Debug, Default)]
40pub struct IntersectBuild {
41    bound_col: usize,
42    intersect_col: usize,
43    table: HashMap<IntersectKey, Vec<Vec<Value>>>,
44    sealed: bool,
45}
46
47impl IntersectBuild {
48    /// Create a build side reading the bound value at `bound_col` and the
49    /// intersect value at `intersect_col` from each build row.
50    pub fn new(bound_col: usize, intersect_col: usize) -> Self {
51        Self {
52            bound_col,
53            intersect_col,
54            table: HashMap::new(),
55            sealed: false,
56        }
57    }
58
59    pub fn bound_col(&self) -> usize {
60        self.bound_col
61    }
62
63    pub fn intersect_col(&self) -> usize {
64        self.intersect_col
65    }
66
67    /// Append one build row. Rows too narrow to hold both columns are
68    /// skipped so malformed chunks cannot poison the table.
69    pub fn append_row(&mut self, row: Vec<Value>) {
70        let width = self.bound_col.max(self.intersect_col) + 1;
71        if row.len() < width {
72            return;
73        }
74        debug_assert!(!self.sealed, "appended after finish; resort needed");
75        let key = IntersectKey::from(&row[self.bound_col]);
76        self.table.entry(key).or_default().push(row);
77        self.sealed = false;
78    }
79
80    /// Append many build rows.
81    pub fn append_rows(&mut self, rows: &[Vec<Value>]) {
82        for row in rows {
83            self.append_row(row.clone());
84        }
85    }
86
87    /// Sort every adjacency list by the intersect column. Must be called
88    /// after the last append and before the first probe.
89    pub fn finish(&mut self) {
90        for rows in self.table.values_mut() {
91            rows.sort_by(|a, b| {
92                IntersectKey::from(&a[self.intersect_col])
93                    .cmp(&IntersectKey::from(&b[self.intersect_col]))
94            });
95        }
96        self.sealed = true;
97    }
98
99    /// Adjacency rows for one bound value, sorted by the intersect column
100    /// after [`IntersectBuild::finish`].
101    pub fn lookup(&self, bound: &Value) -> &[Vec<Value>] {
102        let key = IntersectKey::from(bound);
103        self.table.get(&key).map(Vec::as_slice).unwrap_or(&[])
104    }
105
106    /// Number of distinct bound values.
107    pub fn num_keys(&self) -> usize {
108        self.table.len()
109    }
110
111    /// Total buffered rows across all bound values.
112    pub fn num_rows(&self) -> usize {
113        self.table.values().map(Vec::len).sum()
114    }
115
116    pub fn is_sealed(&self) -> bool {
117        self.sealed
118    }
119
120    pub fn clear(&mut self) {
121        self.table.clear();
122        self.sealed = false;
123    }
124}
125
126#[cfg(test)]
127mod tests {
128    use super::*;
129
130    fn build() -> IntersectBuild {
131        let mut b = IntersectBuild::new(0, 1);
132        b.append_rows(&[
133            vec![Value::Int(1), Value::Int(20)],
134            vec![Value::Int(1), Value::Int(10)],
135            vec![Value::Int(2), Value::Int(30)],
136        ]);
137        b.finish();
138        b
139    }
140
141    #[test]
142    fn finish_sorts_adjacency_by_intersect_column() {
143        let b = build();
144        let rows = b.lookup(&Value::Int(1));
145        assert_eq!(rows.len(), 2);
146        assert_eq!(rows[0][1], Value::Int(10));
147        assert_eq!(rows[1][1], Value::Int(20));
148    }
149
150    #[test]
151    fn missing_bound_yields_empty_slice() {
152        let b = build();
153        assert!(b.lookup(&Value::Int(99)).is_empty());
154    }
155
156    #[test]
157    fn narrow_rows_are_skipped() {
158        let mut b = IntersectBuild::new(0, 1);
159        b.append_row(vec![Value::Int(1)]);
160        b.append_row(vec![Value::Int(1), Value::Int(5)]);
161        b.finish();
162        assert_eq!(b.num_rows(), 1);
163    }
164
165    #[test]
166    fn key_ordering_groups_integers_before_strings() {
167        assert!(IntersectKey::Int(1) < IntersectKey::Int(2));
168        assert!(IntersectKey::Str("a".to_string()) < IntersectKey::Str("b".to_string()));
169    }
170}
linkrs_query/executor/streaming/operators/intersect_build.rs

linkrs_query/executor/streaming/operators/
intersect_build.rs
