akar_processor/physical/write_ops/
optionalextend.rs

1use crate::physical::common::store_value_in_vector;
2use crate::physical::scan_filter::PhysicalScan;
3use akar_common::error::ProcessorError;
4use akar_common::types::{PhysicalTypeID, Value};
5use akar_common::vector::{DataChunk, ValueVector};
6use akar_parser::ast::EdgeDirection;
7use akar_storage::table::TableCatalog;
8use std::collections::HashMap;
9use std::sync::Arc;
10
11// ==================== OptionalExtend ====================
12
13/// Physical operator for `OPTIONAL MATCH` over an already-bound pair of node
14/// variables, e.g. `OPTIONAL MATCH (a)-[existing:Connected]-(b)` where both
15/// `a` and `b` come from the mandatory side (P53.25).
16///
17/// For each input row the relationship-table adjacency is probed for an edge
18/// between the source and destination node ids. When an edge exists, its
19/// property columns (plus an internal `{rel_var}._id` holding the edge index)
20/// are emitted; when no edge exists, those columns are NULL-padded and exactly
21/// one row is still produced (outer-join semantics).
22///
23/// Fan-out mode (P53.40): when `dst_node_var` is empty the destination is
24/// anonymous — `OPTIONAL MATCH (m)-[r:R]-(:T)` — so EVERY edge incident on the
25/// source becomes one output row (zero edges still yield one NULL-padded row).
26/// This preserves multi-edge cardinality for downstream aggregates such as
27/// `WITH m, COUNT(r) AS cnt`; the previous fallback routed such patterns
28/// through a cross-product merge that duplicated every left row.
29///
30/// Output layout: `[input_fields | rel_properties | {rel_var}._id]`.
31pub struct PhysicalOptionalExtend {
32    /// Name of the relationship table to probe.
33    pub rel_table_name: String,
34    /// ID of the relationship table.
35    pub rel_table_id: u64,
36    /// Variable name of the relationship (e.g., "existing"); prefix for the
37    /// emitted edge property columns.
38    pub rel_var: String,
39    /// Variable name of the bound source node (e.g., "a").
40    pub src_node_var: String,
41    /// Variable name of the bound destination node (e.g., "b"). Empty string
42    /// selects fan-out mode: all edges incident on the source are emitted.
43    pub dst_node_var: String,
44    /// Direction of the probe (forward, backward, or both).
45    pub direction: EdgeDirection,
46    /// Table catalog for data access.
47    pub table_catalog: Arc<TableCatalog>,
48}
49
50/// Resolve the internal node-id column (`{var}._id`, falling back to the bare
51/// variable or the primary key column) in the input chunk.
52fn find_node_id_col(chunk: &DataChunk, var: &str) -> Result<usize, ProcessorError> {
53    let name_id = format!("{}.{}", var, "_id");
54    let name_pk = format!("{}.{}", var, "id");
55    let idx = chunk
56        .field_names
57        .iter()
58        .position(|n| n == &name_id)
59        .or_else(|| chunk.field_names.iter().position(|n| n == var))
60        .or_else(|| chunk.field_names.iter().position(|n| n == &name_pk));
61    idx.ok_or_else(|| {
62        format!(
63            "Node variable {} not found in OptionalExtend input. Available fields: {:?}",
64            var, chunk.field_names
65        )
66        .into()
67    })
68}
69
70/// Find an edge index between `src` and `dst` in the forward/reverse adjacency
71/// maps, honoring the probe direction. Returns the first matching edge index.
72fn probe_edge(
73    fwd_adj: &HashMap<u64, Vec<(u64, usize)>>,
74    rev_adj: &HashMap<u64, Vec<(u64, usize)>>,
75    src: u64,
76    dst: u64,
77    direction: &EdgeDirection,
78) -> Option<usize> {
79    match direction {
80        EdgeDirection::LeftToRight => fwd_adj
81            .get(&src)
82            .and_then(|e| e.iter().find(|(o, _)| *o == dst).map(|(_, i)| *i)),
83        EdgeDirection::RightToLeft => rev_adj
84            .get(&src)
85            .and_then(|e| e.iter().find(|(o, _)| *o == dst).map(|(_, i)| *i)),
86        EdgeDirection::Both => fwd_adj
87            .get(&src)
88            .and_then(|e| e.iter().find(|(o, _)| *o == dst).map(|(_, i)| *i))
89            .or_else(|| {
90                rev_adj
91                    .get(&src)
92                    .and_then(|e| e.iter().find(|(o, _)| *o == dst).map(|(_, i)| *i))
93            }),
94    }
95}
96
97/// Fan-out probe (P53.40): every edge index incident on `src`, honoring the
98/// probe direction. Under `Both` a self-loop appears in both adjacency lists,
99/// so indexes are deduplicated (first-seen adjacency order preserved).
100fn probe_incident_edges(
101    fwd_adj: &HashMap<u64, Vec<(u64, usize)>>,
102    rev_adj: &HashMap<u64, Vec<(u64, usize)>>,
103    src: u64,
104    direction: &EdgeDirection,
105) -> Vec<usize> {
106    let mut idxs: Vec<usize> = Vec::new();
107    match direction {
108        EdgeDirection::LeftToRight => {
109            if let Some(entries) = fwd_adj.get(&src) {
110                idxs.extend(entries.iter().map(|(_, i)| *i));
111            }
112        }
113        EdgeDirection::RightToLeft => {
114            if let Some(entries) = rev_adj.get(&src) {
115                idxs.extend(entries.iter().map(|(_, i)| *i));
116            }
117        }
118        EdgeDirection::Both => {
119            for list in [fwd_adj.get(&src), rev_adj.get(&src)].into_iter().flatten() {
120                for (_, i) in list {
121                    if !idxs.contains(i) {
122                        idxs.push(*i);
123                    }
124                }
125            }
126        }
127    }
128    idxs
129}
130
131impl PhysicalOptionalExtend {
132    pub fn execute(&self, input: Vec<DataChunk>) -> Result<Vec<DataChunk>, ProcessorError> {
133        if input.is_empty() {
134            return Ok(input);
135        }
136
137        // Collect rel table data upfront (owned)
138        let (fwd_adj, rev_adj, rel_props, rel_cols) = {
139            let rel_table = self
140                .table_catalog
141                .get_rel_table_by_name(&self.rel_table_name)
142                .ok_or_else(|| format!("Rel table {} not found", self.rel_table_name))?;
143            (
144                rel_table.fwd_adj.clone(),
145                rel_table.rev_adj.clone(),
146                rel_table.properties.clone(),
147                rel_table.columns.clone(),
148            )
149        };
150
151        let num_rel_cols = rel_cols.len();
152        let rel_prefix = if self.rel_var.is_empty() {
153            self.rel_table_name.clone()
154        } else {
155            self.rel_var.clone()
156        };
157        let rel_field_names: Vec<String> = rel_cols.iter().map(|c| format!("{}.{}", rel_prefix, c.name)).collect();
158
159        let mut output = Vec::with_capacity(input.len());
160        let fan_out = self.dst_node_var.is_empty();
161
162        for chunk in input {
163            if chunk.size == 0 {
164                output.push(chunk);
165                continue;
166            }
167
168            let src_idx = find_node_id_col(&chunk, &self.src_node_var)?;
169            // Fan-out mode has no destination variable to resolve.
170            let dst_idx = if fan_out {
171                0
172            } else {
173                find_node_id_col(&chunk, &self.dst_node_var)?
174            };
175
176            let num_input_fields = chunk.fields.len();
177            let num_out_cols = num_input_fields + num_rel_cols + 1;
178            let mut out_data: Vec<Vec<Value>> = vec![Vec::new(); num_out_cols];
179
180            for i in 0..chunk.size {
181                let input_vals: Vec<Value> = (0..num_input_fields)
182                    .map(|col| chunk.get_value(col, i).unwrap_or(Value::Null))
183                    .collect();
184
185                // Edge matches per input row: fan-out emits one row per
186                // incident edge; the bound-pair probe emits at most one. An
187                // empty match still yields a NULL-padded row (outer join).
188                let matches: Vec<Option<usize>> = if fan_out {
189                    match input_vals.get(src_idx) {
190                        Some(Value::Int64(s)) => {
191                            let found = probe_incident_edges(&fwd_adj, &rev_adj, *s as u64, &self.direction);
192                            if found.is_empty() {
193                                vec![None]
194                            } else {
195                                found.into_iter().map(Some).collect()
196                            }
197                        }
198                        _ => vec![None],
199                    }
200                } else {
201                    let edge_idx = match (chunk.get_value(src_idx, i), chunk.get_value(dst_idx, i)) {
202                        (Some(Value::Int64(s)), Some(Value::Int64(d))) => {
203                            probe_edge(&fwd_adj, &rev_adj, s as u64, d as u64, &self.direction)
204                        }
205                        _ => None,
206                    };
207                    vec![edge_idx]
208                };
209
210                for edge_idx in matches {
211                    for col in 0..num_input_fields {
212                        out_data[col].push(input_vals[col].clone());
213                    }
214                    for col in 0..num_rel_cols {
215                        let val = match edge_idx {
216                            Some(ei) => rel_props
217                                .get(col)
218                                .and_then(|c| c.get(ei))
219                                .cloned()
220                                .unwrap_or(Value::Null),
221                            None => Value::Null,
222                        };
223                        out_data[num_input_fields + col].push(val);
224                    }
225                    // Internal edge index — non-NULL iff an edge exists, so a
226                    // zero-property rel table still answers `{rel_var} IS NULL`.
227                    out_data[num_input_fields + num_rel_cols].push(match edge_idx {
228                        Some(ei) => Value::Int64(ei as i64),
229                        None => Value::Null,
230                    });
231                }
232            }
233
234            let out_size = out_data.first().map(Vec::len).unwrap_or(chunk.size);
235
236            // Build output columns.
237            let mut fields = Vec::with_capacity(num_out_cols);
238            let mut field_types = Vec::with_capacity(num_out_cols);
239            let mut field_names = Vec::with_capacity(num_out_cols);
240
241            for col in 0..num_input_fields {
242                let phys_type = chunk.field_types[col];
243                let mut v = ValueVector::new(phys_type, out_size);
244                v.resize(out_size);
245                for row in 0..out_size {
246                    store_value_in_vector(&mut v, row, &out_data[col][row])?;
247                }
248                fields.push(v);
249                field_types.push(phys_type);
250                field_names.push(if col < chunk.field_names.len() {
251                    chunk.field_names[col].clone()
252                } else {
253                    format!("field_{}", col)
254                });
255            }
256            for col in 0..num_rel_cols {
257                let phys_type = if col < rel_cols.len() {
258                    PhysicalScan::logical_to_physical(&rel_cols[col].logical_type)
259                } else {
260                    PhysicalTypeID::Int64
261                };
262                let mut v = ValueVector::new(phys_type, out_size);
263                v.resize(out_size);
264                for row in 0..out_size {
265                    store_value_in_vector(&mut v, row, &out_data[num_input_fields + col][row])?;
266                }
267                fields.push(v);
268                field_types.push(phys_type);
269                field_names.push(rel_field_names[col].clone());
270            }
271            // Internal edge index column (`{rel_var}._id`).
272            let mut id_v = ValueVector::new(PhysicalTypeID::Int64, out_size);
273            id_v.resize(out_size);
274            for row in 0..out_size {
275                store_value_in_vector(&mut id_v, row, &out_data[num_input_fields + num_rel_cols][row])?;
276            }
277            fields.push(id_v);
278            field_types.push(PhysicalTypeID::Int64);
279            field_names.push(format!("{}.{}", rel_prefix, "_id"));
280
281            let arrow_fields = fields
282                .iter()
283                .map(|v| akar_common::arrow_vector::ArrowVector::from_legacy(v).array)
284                .collect::<Vec<_>>();
285            let arrow_field_types = fields.iter().map(|v| v.physical_type()).collect::<Vec<_>>();
286            output.push(DataChunk {
287                fields: arrow_fields,
288                field_types: arrow_field_types,
289                size: out_size,
290                field_names,
291                sel_vector: None,
292            });
293        }
294
295        Ok(output)
296    }
297}
akar_processor/physical/write_ops/optionalextend.rs

akar_processor/physical/write_ops/
optionalextend.rs
