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            docs_table: fq.docs_table.clone(),
74            terms_table: fq.terms_table.clone(),
75            posting_table: fq.posting_table.clone(),
76            table_name: fq.table_name.clone(),
77            column_name: fq.column_name.clone(),
78            table_catalog: ctx
79                .table_catalog
80                .clone()
81                .ok_or_else(|| "Table catalog required for FTS scan".to_string())?,
82        });
83    }
84    if let Some(ref pred) = s.predicate {
85        scan = scan.with_predicate(pred.clone());
86        let registry = ctx
87            .function_registry
88            .clone()
89            .ok_or_else(|| "No function registry available for predicate evaluation".to_string())?;
90        scan = scan.with_evaluator(Arc::new(Mutex::new(ExpressionEvaluator::new(registry))));
91    }
92    let mut result = scan.execute(current_input)?;
93    let prefix = s.alias.as_ref().unwrap_or(&s.table_name);
94
95    for chunk in &mut result {
96        chunk.field_names = chunk.field_names.iter().map(|n| format!("{}.{}", prefix, n)).collect();
97    }
98    Ok(result)
99}
100
101pub fn map_and_execute_scan(
102    op: &LogicalOperator,
103    current_input: Vec<DataChunk>,
104    ctx: &mut ExecutionContext,
105) -> Result<Vec<DataChunk>, ProcessorError> {
106    match op {
107        LogicalOperator::ScanRel(s) => {
108            let (data, columns, _num_rows) = ctx.resolve_scan_data(&s.table_name, None);
109            let scan = PhysicalScanRel {
110                table_name: s.table_name.clone(),
111                table_id: s.table_id,
112                direction: s.direction.clone(),
113                table_data: data,
114                table_columns: columns,
115            };
116            let mut result = scan.execute(current_input)?;
117            let prefix = &s.table_name;
118            for chunk in &mut result {
119                chunk.field_names = chunk.field_names.iter().map(|n| format!("{}.{}", prefix, n)).collect();
120            }
121            Ok(result)
122        }
123        LogicalOperator::VectorSimilarityScan(vs) => {
124            let scan = PhysicalVectorSimilarityScan {
125                index_name: vs.index_name.clone(),
126                index_id: vs.index_id,
127                query_vector: vs.query_vector.clone(),
128                top_k: vs.top_k,
129                table_name: vs.table_name.clone(),
130                table_catalog: ctx.table_catalog.clone(),
131            };
132            let result = scan.execute(current_input)?;
133            Ok(result)
134        }
135        LogicalOperator::ArtIndexRangeScan(ars) => {
136            let scan = PhysicalArtIndexRangeScan {
137                table_name: ars.table_name.clone(),
138                table_id: ars.table_id,
139                lower_bound: ars.lower_bound.clone(),
140                upper_bound: ars.upper_bound.clone(),
141                lower_inclusive: ars.lower_inclusive,
142                upper_inclusive: ars.upper_inclusive,
143                table_catalog: ctx.table_catalog.clone(),
144            };
145            let mut result = scan.execute(current_input)?;
146            let prefix = ars.alias.as_ref().unwrap_or(&ars.table_name);
147            for chunk in &mut result {
148                chunk.field_names = chunk.field_names.iter().map(|n| format!("{}.{}", prefix, n)).collect();
149            }
150            Ok(result)
151        }
152        LogicalOperator::IndexLookup(il) => {
153            let table_catalog = ctx
154                .table_catalog
155                .clone()
156                .ok_or_else(|| "No table catalog available for INDEX LOOKUP".to_string())?;
157            let lookup_op = PhysicalIndexLookup {
158                table_name: il.table_name.clone(),
159                table_id: il.table_id,
160                key_value: il.key_value.clone(),
161                table_catalog,
162            };
163            let result = lookup_op.execute(current_input)?;
164            Ok(result)
165        }
166        LogicalOperator::ExpressionsScan(_es) => Ok(vec![DataChunk::new(vec![], vec![])]),
167        LogicalOperator::PathPropertyProbe(p) => {
168            let properties = p
169                .properties
170                .iter()
171                .map(|(t, is_node, props)| crate::physical::scan_filter::PathPropertySpec {
172                    table_name: t.clone(),
173                    is_node: *is_node,
174                    property_names: props.clone(),
175                })
176                .collect();
177
178            let probe = crate::physical::scan_filter::PhysicalPathPropertyProbe {
179                node_ids_col_idx: p.node_ids_col_idx,
180                edge_ids_col_idx: p.edge_ids_col_idx,
181                properties,
182                table_catalog: ctx
183                    .table_catalog
184                    .clone()
185                    .ok_or_else(|| "table catalog required for PathPropertyProbe".to_string())?,
186            };
187
188            let input = if !p.children.is_empty() {
189                ctx.execute_children(&p.children)?
190            } else {
191                current_input
192            };
193
194            let result = probe.execute(input)?;
195            Ok(result)
196        }
197        _ => Err(format!("Not a scan operator: {:?}", op).into()),
198    }
199}
akar_processor/processor/mapper/map_scan.rs

akar_processor/processor/mapper/
map_scan.rs
