akar_processor/processor/mapper/
map_scan.rs

1use super::ExecutionContext;
2use crate::expression_evaluator::ExpressionEvaluator;
3use crate::physical_operator::*;
4use akar_common::error::ProcessorError;
5use akar_common::types::Value;
6use akar_common::vector::DataChunk;
7use akar_parser::ast::Expression;
8use akar_planner::logical_operator::{LogicalOperator, LogicalScanNode};
9use std::sync::{Arc, Mutex};
10
11fn extract_zone_map_predicate(expr: &Expression, columns: &[String]) -> Option<(usize, String, Value)> {
12    if let Expression::BinaryOp(op, left, right) = expr {
13        let op_str = match op {
14            akar_parser::ast::BinaryOp::Equal => "=",
15            akar_parser::ast::BinaryOp::GreaterThan => ">",
16            akar_parser::ast::BinaryOp::LessThan => "<",
17            akar_parser::ast::BinaryOp::GreaterThanOrEqual => ">=",
18            akar_parser::ast::BinaryOp::LessThanOrEqual => "<=",
19            akar_parser::ast::BinaryOp::NotEqual => "!=",
20            _ => return None,
21        };
22        if let Expression::Variable(var_name) = &**left {
23            if let Expression::Constant(c) = &**right {
24                let col_name = var_name.split('.').next_back().unwrap_or(var_name);
25                if let Some(col_idx) = columns.iter().position(|c| c == col_name) {
26                    let val = match c {
27                        akar_parser::ast::Constant::Integer(i) => Value::Int64(*i),
28                        akar_parser::ast::Constant::Float(f) => Value::Double(*f),
29                        akar_parser::ast::Constant::String(s) => Value::String(s.clone()),
30                        akar_parser::ast::Constant::Bool(b) => Value::Bool(*b),
31                        akar_parser::ast::Constant::Null => Value::Null,
32                    };
33                    return Some((col_idx, op_str.to_string(), val));
34                }
35            }
36        }
37    }
38    None
39}
40
41pub fn map_and_execute_scan_node(
42    s: &LogicalScanNode,
43    next_op: Option<&LogicalOperator>,
44    current_input: Vec<DataChunk>,
45    ctx: &mut ExecutionContext,
46) -> Result<Vec<DataChunk>, ProcessorError> {
47    let mut pred_owned = None;
48    if let Some(LogicalOperator::Filter(f)) = next_op {
49        pred_owned = extract_zone_map_predicate(&f.expression, &s.columns);
50    }
51
52    let pred_ref = pred_owned
53        .as_ref()
54        .map(|(idx, op_str, val)| (*idx, op_str.as_str(), val));
55
56    // Try Arrow fast path first (avoids Vec<Vec<Value>> intermediate)
57    let (arrow_data, columns, arrow_num_rows) = ctx.resolve_scan_arrow_data(&s.table_name);
58    let mut scan = if let Some(arrays) = arrow_data {
59        PhysicalScan::new(s.table_name.clone(), s.table_id, arrow_num_rows.max(1)).with_arrow_data(arrays, columns)
60    } else {
61        // Fallback to legacy Vec<Vec<Value>> data path (MVCC-aware)
62        let (data, fallback_columns, fallback_num_rows) = ctx.resolve_scan_data(&s.table_name, pred_ref);
63        let mut s = PhysicalScan::new(s.table_name.clone(), s.table_id, fallback_num_rows.max(1));
64        if let Some(d) = data {
65            s = s.with_data(d, fallback_columns);
66        }
67        s
68    };
69    if let Some(ref fq) = s.fts_query {
70        scan = scan.with_fts_query(PhysicalFtsScan {
71            index_name: fq.index_name.clone(),
72            query_string: fq.query_string.clone(),
73            table_name: fq.table_name.clone(),
74            column_name: fq.column_name.clone(),
75            table_catalog: ctx
76                .table_catalog
77                .clone()
78                .ok_or_else(|| "Table catalog required for FTS scan".to_string())?,
79        });
80    }
81    if let Some(ref pred) = s.predicate {
82        scan = scan.with_predicate(pred.clone());
83        let registry = ctx
84            .function_registry
85            .clone()
86            .ok_or_else(|| "No function registry available for predicate evaluation".to_string())?;
87        scan = scan.with_evaluator(Arc::new(Mutex::new(ExpressionEvaluator::new(registry))));
88    }
89    let mut result = scan.execute(current_input)?;
90    let prefix = s.alias.as_ref().unwrap_or(&s.table_name);
91
92    for chunk in &mut result {
93        chunk.field_names = chunk.field_names.iter().map(|n| format!("{}.{}", prefix, n)).collect();
94    }
95    Ok(result)
96}
97
98pub fn map_and_execute_scan(
99    op: &LogicalOperator,
100    current_input: Vec<DataChunk>,
101    ctx: &mut ExecutionContext,
102) -> Result<Vec<DataChunk>, ProcessorError> {
103    match op {
104        LogicalOperator::ScanRel(s) => {
105            let (data, columns, _num_rows) = ctx.resolve_scan_data(&s.table_name, None);
106            let scan = PhysicalScanRel {
107                table_name: s.table_name.clone(),
108                table_id: s.table_id,
109                direction: s.direction.clone(),
110                table_data: data,
111                table_columns: columns,
112            };
113            let mut result = scan.execute(current_input)?;
114            let prefix = &s.table_name;
115            for chunk in &mut result {
116                chunk.field_names = chunk.field_names.iter().map(|n| format!("{}.{}", prefix, n)).collect();
117            }
118            Ok(result)
119        }
120        LogicalOperator::VectorSimilarityScan(vs) => {
121            let tc = ctx
122                .table_catalog
123                .clone()
124                .ok_or_else(|| "No table catalog available for VECTOR SIMILARITY SCAN".to_string())?;
125            // Resolve the HNSW index by column (first index on the table whose
126            // indexed column matches), falling back to the first index on the
127            // table — same convention as the explicit CALL handler.
128            let index_name = {
129                let mut by_column = None;
130                let mut first_on_table = None;
131                for entry in tc.all_vector_indexes() {
132                    if entry.table_name == vs.table_name {
133                        if by_column.is_none() && entry.column_name == vs.column_name {
134                            by_column = Some(entry.name.clone());
135                        }
136                        if first_on_table.is_none() {
137                            first_on_table = Some(entry.name.clone());
138                        }
139                    }
140                }
141                by_column.or(first_on_table).ok_or_else(|| {
142                    format!(
143                        "No vector index found on table '{}' for column '{}'",
144                        vs.table_name, vs.column_name
145                    )
146                })?
147            };
148            let scan = PhysicalVectorSimilarityScan {
149                index_name,
150                index_id: 0,
151                query_vector: vs.query_vector.clone(),
152                top_k: vs.top_k,
153                table_name: vs.table_name.clone(),
154                table_catalog: Some(tc),
155            };
156            let mut result = scan.execute(current_input)?;
157            // Prefix the output field names with the node alias (like the
158            // ScanNode branch) so downstream Filter/OrderBy/Projection, which
159            // reference `alias.col`, can resolve their columns. The bare CALL
160            // path has `alias: None` and keeps unprefixed names (distance, _id).
161            if let Some(alias) = &vs.alias {
162                for chunk in &mut result {
163                    chunk.field_names = chunk.field_names.iter().map(|n| format!("{alias}.{n}")).collect();
164                }
165            }
166            Ok(result)
167        }
168        LogicalOperator::ArtIndexRangeScan(ars) => {
169            let scan = PhysicalArtIndexRangeScan {
170                table_name: ars.table_name.clone(),
171                table_id: ars.table_id,
172                lower_bound: ars.lower_bound.clone(),
173                upper_bound: ars.upper_bound.clone(),
174                lower_inclusive: ars.lower_inclusive,
175                upper_inclusive: ars.upper_inclusive,
176                table_catalog: ctx.table_catalog.clone(),
177            };
178            let mut result = scan.execute(current_input)?;
179            let prefix = ars.alias.as_ref().unwrap_or(&ars.table_name);
180            for chunk in &mut result {
181                chunk.field_names = chunk.field_names.iter().map(|n| format!("{}.{}", prefix, n)).collect();
182            }
183            Ok(result)
184        }
185        LogicalOperator::IndexLookup(il) => {
186            let table_catalog = ctx
187                .table_catalog
188                .clone()
189                .ok_or_else(|| "No table catalog available for INDEX LOOKUP".to_string())?;
190            let lookup_op = PhysicalIndexLookup {
191                table_name: il.table_name.clone(),
192                table_id: il.table_id,
193                key_value: il.key_value.clone(),
194                table_catalog,
195            };
196            let result = lookup_op.execute(current_input)?;
197            Ok(result)
198        }
199        LogicalOperator::ExpressionsScan(_es) => Ok(vec![DataChunk::new(vec![], vec![])]),
200        LogicalOperator::PathPropertyProbe(p) => {
201            let properties = p
202                .properties
203                .iter()
204                .map(|(t, is_node, props)| crate::physical::scan_filter::PathPropertySpec {
205                    table_name: t.clone(),
206                    is_node: *is_node,
207                    property_names: props.clone(),
208                })
209                .collect();
210
211            let probe = crate::physical::scan_filter::PhysicalPathPropertyProbe {
212                node_ids_col_idx: p.node_ids_col_idx,
213                edge_ids_col_idx: p.edge_ids_col_idx,
214                properties,
215                table_catalog: ctx
216                    .table_catalog
217                    .clone()
218                    .ok_or_else(|| "table catalog required for PathPropertyProbe".to_string())?,
219            };
220
221            let input = if !p.children.is_empty() {
222                ctx.execute_children(&p.children)?
223            } else {
224                current_input
225            };
226
227            let result = probe.execute(input)?;
228            Ok(result)
229        }
230        _ => Err(format!("Not a scan operator: {:?}", op).into()),
231    }
232}
akar_processor/processor/mapper/map_scan.rs

akar_processor/processor/mapper/
map_scan.rs
