nodedb_lite/query/query_ops/joins/
nested_loop.rs

1// SPDX-License-Identifier: Apache-2.0
2//! NestedLoopJoin: fallback for non-equi joins.
3//!
4//! Evaluates an arbitrary `condition` (msgpack-encoded `Vec<ScanFilter>`)
5//! against the merged left+right row per join_type semantics.
6
7use std::collections::HashMap;
8
9use nodedb_query::scan_filter::ScanFilter;
10use nodedb_types::result::QueryResult;
11use nodedb_types::value::Value;
12
13use crate::error::LiteError;
14use crate::query::engine::LiteQueryEngine;
15use crate::storage::engine::StorageEngine;
16
17use super::common::{maps_to_result, merge_rows, scan_collection};
18
19pub async fn execute_nested_loop_join<S: StorageEngine>(
20    engine: &LiteQueryEngine<S>,
21    left_collection: &str,
22    right_collection: &str,
23    condition: &[u8],
24    join_type: &str,
25    limit: usize,
26) -> Result<QueryResult, LiteError> {
27    let left_rows = scan_collection(engine, left_collection).await?;
28    let right_rows = scan_collection(engine, right_collection).await?;
29
30    let filters: Vec<ScanFilter> = if condition.is_empty() {
31        Vec::new()
32    } else {
33        zerompk::from_msgpack(condition).map_err(|e| LiteError::Serialization {
34            detail: format!("decode nested loop condition: {e}"),
35        })?
36    };
37
38    let effective_limit = if limit == 0 { usize::MAX } else { limit };
39    let mut output: Vec<HashMap<String, Value>> = Vec::new();
40
41    'outer: for left_row in &left_rows {
42        let mut matched = false;
43        for right_row in &right_rows {
44            let merged = merge_rows(left_row, right_row, None);
45            let doc = Value::Object(merged.clone());
46            let passes = ScanFilter::all_match_value(&filters, &doc)?;
47            if passes {
48                matched = true;
49                if join_type != "semi" && join_type != "anti" {
50                    output.push(merged);
51                    if output.len() >= effective_limit {
52                        break 'outer;
53                    }
54                } else if join_type == "semi" {
55                    output.push(left_row.clone());
56                    if output.len() >= effective_limit {
57                        break 'outer;
58                    }
59                    break; // one match per left row for semi
60                }
61            }
62        }
63        if !matched {
64            match join_type {
65                "left" | "full" => {
66                    output.push(left_row.clone());
67                    if output.len() >= effective_limit {
68                        break;
69                    }
70                }
71                "anti" => {
72                    output.push(left_row.clone());
73                    if output.len() >= effective_limit {
74                        break;
75                    }
76                }
77                _ => {}
78            }
79        }
80    }
81
82    // Right outer: unmatched right rows.
83    if join_type == "right" || join_type == "full" {
84        'right: for right_row in &right_rows {
85            let mut any_match = false;
86            for left_row in &left_rows {
87                let merged = merge_rows(left_row, right_row, None);
88                let doc = Value::Object(merged);
89                if ScanFilter::all_match_value(&filters, &doc)? {
90                    any_match = true;
91                    break;
92                }
93            }
94            if !any_match {
95                output.push(right_row.clone());
96                if output.len() >= effective_limit {
97                    break 'right;
98                }
99            }
100        }
101    }
102
103    Ok(maps_to_result(output))
104}
105
106#[cfg(test)]
107mod tests {
108    use super::super::common::merge_rows;
109    use nodedb_query::scan_filter::{FilterOp, ScanFilter};
110    use nodedb_types::value::Value;
111    use std::collections::HashMap;
112
113    fn row(fields: &[(&str, Value)]) -> HashMap<String, Value> {
114        fields
115            .iter()
116            .map(|(k, v)| (k.to_string(), v.clone()))
117            .collect()
118    }
119
120    #[test]
121    fn non_equi_condition_evaluates() {
122        // Filter: merged row's "price" > "min_price" (from right side).
123        // This tests that the condition evaluates on the merged row.
124        let left = row(&[
125            ("item", Value::String("apple".into())),
126            ("price", Value::Integer(5)),
127        ]);
128        let right = row(&[("min_price", Value::Integer(3))]);
129        let merged = merge_rows(&left, &right, None);
130
131        let filter = ScanFilter {
132            field: "price".into(),
133            op: FilterOp::Gt,
134            value: Value::Integer(3),
135            clauses: Vec::new(),
136            expr: None,
137        };
138
139        let doc = Value::Object(merged);
140        assert!(
141            filter
142                .matches_value(&doc)
143                .expect("filter evaluation failed")
144        );
145    }
146}
nodedb_lite/query/query_ops/joins/nested_loop.rs

nodedb_lite/query/query_ops/joins/
nested_loop.rs
