akar_processor/physical/
index_lookup.rs

1//! PhysicalIndexLookup — point lookup via ART index.
2//!
3//! Uses the ART index on a table's primary key column to efficiently
4//! find a single row. Produces a DataChunk with the matching row's
5//! column values or empty result if the key is not found.
6
7use crate::physical::common::store_value_in_vector;
8use crate::physical::types::{OperatorResult, PhysicalOperatorExec};
9use akar_common::types::Value;
10use akar_common::vector::{DataChunk, ValueVector};
11use akar_storage::table::TableCatalog;
12use std::sync::Arc;
13
14/// Physical operator for index-based point lookup.
15///
16/// Uses the ART index on a node table's primary key to find a single
17/// row matching `key_value`. Returns the full row data or empty if
18/// no match is found.
19pub struct PhysicalIndexLookup {
20    pub table_name: String,
21    pub table_id: u64,
22    /// The key value to look up in the ART index.
23    pub key_value: Value,
24    pub table_catalog: Arc<TableCatalog>,
25}
26
27impl PhysicalOperatorExec for PhysicalIndexLookup {
28    fn operator_type(&self) -> &str {
29        "index_lookup"
30    }
31
32    fn execute(&self, _input: Vec<DataChunk>) -> OperatorResult {
33        let node_table = self
34            .table_catalog
35            .get_node_table_by_name(&self.table_name)
36            .ok_or_else(|| format!("Node table '{}' not found for IndexLookup", self.table_name))?;
37
38        // Use ART index for point lookup
39        let row_ids = node_table.lookup_by_pk_range(
40            Some(&self.key_value),
41            true, // lower_inclusive
42            Some(&self.key_value),
43            true, // upper_inclusive
44            1,    // max results (point lookup = single key)
45        );
46
47        if row_ids.is_empty() {
48            return Ok(vec![DataChunk::new(vec![], vec![])]);
49        }
50
51        let row_id = row_ids[0] as usize;
52        let num_cols = node_table.columns.len();
53
54        let mut fields = Vec::with_capacity(num_cols);
55        let mut field_types = Vec::with_capacity(num_cols);
56        let mut field_names = Vec::with_capacity(num_cols);
57
58        for col_idx in 0..num_cols {
59            let val = node_table.get_value(row_id, col_idx).cloned().unwrap_or(Value::Null);
60
61            let phys_type = val.physical_type();
62            let mut v = ValueVector::new(phys_type, 1);
63            v.resize(1);
64            if matches!(val, Value::Null) {
65                v.set_null(0, true);
66            } else {
67                store_value_in_vector(&mut v, 0, &val)?;
68            }
69            fields.push(akar_common::arrow_vector::ArrowVector::from_legacy(&v).array);
70            field_types.push(phys_type);
71            field_names.push(node_table.columns[col_idx].name.clone());
72        }
73
74        drop(node_table);
75
76        Ok(vec![DataChunk {
77            fields,
78            field_types,
79            size: 1,
80            field_names,
81            sel_vector: None,
82        }])
83    }
84}
akar_processor/physical/index_lookup.rs

akar_processor/physical/
index_lookup.rs
