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

akar_processor/processor/mapper/
map_scan.rs
