akar_processor/processor/mapper/
map_join.rs

1use super::ExecutionContext;
2use crate::physical_operator::*;
3use akar_common::error::ProcessorError;
4use akar_common::vector::DataChunk;
5use akar_parser::ast::Expression;
6use akar_planner::logical_operator::LogicalOperator;
7
8use crate::processor::join_helpers::derive_join_column_indices;
9use crate::processor::union_helpers::{flatten_union_child, merge_optional_chunks};
10
11pub fn map_and_execute_join(
12    op: &LogicalOperator,
13    current_input: Vec<DataChunk>,
14    ctx: &mut ExecutionContext,
15) -> Result<Vec<DataChunk>, ProcessorError> {
16    match op {
17        LogicalOperator::HashJoin(h) => {
18            let left_ops = flatten_union_child(&h.build_side);
19            let right_ops = flatten_union_child(&h.probe_side);
20
21            let build_chunks = ctx.execute_children(&left_ops)?;
22            let probe_chunks = ctx.execute_children(&right_ops)?;
23
24            let (build_cols, probe_cols) = derive_join_column_indices(&h.join_keys, &build_chunks, &probe_chunks);
25            let join = PhysicalHashJoin::new(build_cols, probe_cols);
26            let result = join.execute_binary(&build_chunks, &probe_chunks)?;
27            Ok(result)
28        }
29        LogicalOperator::SemiJoin(s) => {
30            let left_ops = flatten_union_child(&s.left);
31            let right_ops = flatten_union_child(&s.right);
32
33            let build_chunks = ctx.execute_children(&left_ops)?;
34            let probe_chunks = ctx.execute_children(&right_ops)?;
35
36            let semi = PhysicalSemiJoin {
37                build_columns: vec![0],
38                probe_columns: vec![0],
39            };
40            let result = semi.execute_binary(&build_chunks, &probe_chunks)?;
41            Ok(result)
42        }
43        LogicalOperator::AntiJoin(a) => {
44            let left_ops = flatten_union_child(&a.left);
45            let right_ops = flatten_union_child(&a.right);
46
47            let build_chunks = ctx.execute_children(&left_ops)?;
48            let probe_chunks = ctx.execute_children(&right_ops)?;
49
50            let anti = PhysicalAntiJoin {
51                build_columns: vec![0],
52                probe_columns: vec![0],
53            };
54            let result = anti.execute_binary(&build_chunks, &probe_chunks)?;
55            Ok(result)
56        }
57        LogicalOperator::Intersect(ic) => {
58            // Build side is a (possibly nested) Union of per-pattern pipelines,
59            // each executed independently so the intersect has one hash table per
60            // pattern. Probe side is the shared-node scan.
61            let build_sides = collect_union_sides(&ic.left);
62            let probe_ops = flatten_union_child(&ic.right);
63
64            let build_chunk_sides: Vec<Vec<DataChunk>> = build_sides
65                .iter()
66                .map(|ops| ctx.execute_children(ops))
67                .collect::<Result<_, _>>()?;
68            let probe_chunks = ctx.execute_children(&probe_ops)?;
69
70            let (probe_key_col, build_key_col) =
71                resolve_intersect_key_cols(&ic.build_key_exprs, &probe_chunks, &build_chunk_sides);
72
73            let intersect = PhysicalIntersect {
74                num_build_sides: build_chunk_sides.len() as u32,
75                probe_key_col,
76                build_key_col,
77            };
78            let result = intersect.execute_sides(&build_chunk_sides, &probe_chunks)?;
79            Ok(result)
80        }
81        LogicalOperator::CrossProduct(cp) => {
82            let left_ops = flatten_union_child(&cp.left);
83            let right_ops = flatten_union_child(&cp.right);
84            let build_chunks = ctx.execute_children(&left_ops)?;
85            let probe_chunks = ctx.execute_children(&right_ops)?;
86            let cross = PhysicalCrossProduct;
87            let result = cross.execute_binary(&build_chunks, &probe_chunks)?;
88            Ok(result)
89        }
90        LogicalOperator::OptionalMatch(om) => {
91            // Execute left (required) subtree
92            let left_ops = flatten_union_child(&om.left);
93            let left_result = ctx.execute_children(&left_ops)?;
94
95            // Execute right (optional) subtree
96            let right_ops = flatten_union_child(&om.right);
97            let right_result = ctx.execute_children(&right_ops)?;
98
99            // Combine: use flattened row-level merge
100            let merged = merge_optional_chunks(left_result, right_result)?;
101            Ok(merged)
102        }
103        LogicalOperator::RecursiveExtend(re) => {
104            let scan = PhysicalRecursiveExtend {
105                source_table_id: re.source_table_id,
106                rel_table_ids: re.rel_table_ids.clone(),
107                lower_bound: re.lower_bound,
108                upper_bound: re.upper_bound,
109                direction: re.direction,
110                semantic: re.semantic,
111                table_catalog: ctx.table_catalog.clone(),
112                weight_property: re.weight_property.clone(),
113                cost_output_name: re.cost_output_name.clone(),
114            };
115            let result = scan.execute(current_input)?;
116            Ok(result)
117        }
118        _ => Err(format!("Not a join operator: {:?}", op).into()),
119    }
120}
121
122/// Flatten a (possibly nested) `Union` subtree into a list of independent
123/// operator pipelines — one per WCOJ build side.
124fn collect_union_sides(op: &LogicalOperator) -> Vec<Vec<LogicalOperator>> {
125    match op {
126        LogicalOperator::Union(u) => {
127            let mut sides = collect_union_sides(&u.left);
128            sides.extend(collect_union_sides(&u.right));
129            sides
130        }
131        other => vec![flatten_union_child(other)],
132    }
133}
134
135/// Resolve the shared-node key column index on the probe and build sides.
136///
137/// The key is derived from the first build key expression (a reference to the
138/// shared variable, e.g. `a`), resolved against `field_names` as `a.id`.
139fn resolve_intersect_key_cols(
140    build_key_exprs: &[Expression],
141    probe_chunks: &[DataChunk],
142    build_sides: &[Vec<DataChunk>],
143) -> (u32, u32) {
144    let var = build_key_exprs.first().and_then(|e| match e {
145        Expression::Variable(v) => Some(v.clone()),
146        Expression::PropertyAccess(obj, _) => {
147            if let Expression::Variable(v) = &**obj {
148                Some(v.clone())
149            } else {
150                None
151            }
152        }
153        _ => None,
154    });
155
156    let probe_names: Vec<&str> = probe_chunks
157        .first()
158        .map(|c| c.field_names.iter().map(|s| s.as_str()).collect())
159        .unwrap_or_default();
160    let build_names: Vec<&str> = build_sides
161        .first()
162        .and_then(|s| s.first())
163        .map(|c| c.field_names.iter().map(|s| s.as_str()).collect())
164        .unwrap_or_default();
165
166    let mut probe_col = 0u32;
167    let mut build_col = 0u32;
168    if let Some(var) = var {
169        let candidates = [format!("{var}._id"), format!("{var}.id"), var];
170        for c in &candidates {
171            if let Some(idx) = probe_names.iter().position(|n| n == c) {
172                probe_col = idx as u32;
173                break;
174            }
175        }
176        for c in &candidates {
177            if let Some(idx) = build_names.iter().position(|n| n == c) {
178                build_col = idx as u32;
179                break;
180            }
181        }
182    }
183    (probe_col, build_col)
184}
akar_processor/processor/mapper/map_join.rs

akar_processor/processor/mapper/
map_join.rs
