akar_processor/physical/write_ops/
ddl_fts.rs

1//! Auto-extracted from physical_operator.rs
2use crate::physical::types::{OperatorResult, PhysicalOperatorExec};
3use akar_common::types::{PhysicalTypeID, Value};
4use akar_common::vector::{DataChunk, ValueVector};
5use akar_storage::table::TableCatalog;
6use std::sync::Arc;
7
8// ==================== DDL & FTS ====================
9
10/// Physical COUNT on rel table — optimized via CSR metadata (Ladybug).
11/// Instead of scanning all edges, directly reads the edge count from the RelTable.
12pub struct PhysicalCountRelTable {
13    pub table_name: String,
14    pub table_id: u64,
15    pub table_catalog: Option<Arc<TableCatalog>>,
16}
17
18impl PhysicalOperatorExec for PhysicalCountRelTable {
19    fn operator_type(&self) -> &str {
20        "count_rel_table"
21    }
22
23    fn execute(&self, _input: Vec<DataChunk>) -> OperatorResult {
24        let tc = self
25            .table_catalog
26            .as_ref()
27            .ok_or_else(|| "No table catalog for CountRelTable".to_string())?;
28
29        let count = if let Some(table) = tc.get_rel_table(self.table_id) {
30            table.num_rows as i64
31        } else {
32            0
33        };
34
35        let mut v = ValueVector::new(PhysicalTypeID::Int64, 1);
36        v.resize(1);
37        v.set_i64(0, count);
38        let arr = akar_common::arrow_vector::ArrowVector::from_legacy(&v).array;
39        Ok(vec![DataChunk::new(vec![arr], vec![PhysicalTypeID::Int64])])
40    }
41}
42
43/// Physical operator for `CREATE FTS INDEX` — builds the **Tantivy** FTS index
44/// over the source column (P104.1/P104.4).
45///
46/// The index persists as a Tantivy directory under `<db_path>/fts/<index_name>`
47/// for disk-backed catalogs. The legacy `fts_{idx}_docs` / `fts_{idx}_terms` /
48/// `fts_{idx}_appears_in` macro tables are **gone** (P104.2 clean break) — the
49/// Tantivy directory is the *only* FTS representation.
50pub struct PhysicalCreateFtsIndex {
51    pub index_name: String,
52    pub table_name: String,
53    pub column_name: String,
54    /// Tokenizer name from `WITH TOKENIZER('...')` (P109.1); `None` resolves to
55    /// the `en_stem` default via [`akar_fts::tokenizer::resolve`].
56    pub tokenizer: Option<String>,
57    pub if_not_exists: bool,
58    pub table_catalog: Arc<TableCatalog>,
59}
60
61impl PhysicalOperatorExec for PhysicalCreateFtsIndex {
62    fn operator_type(&self) -> &str {
63        "create_fts_index"
64    }
65
66    fn execute(&self, _input: Vec<DataChunk>) -> OperatorResult {
67        // Locate the source table and snapshot its schema + data. The DashMap
68        // `Ref` MUST be dropped before any write lock below — DashMap is not
69        // re-entrant, and holding a read `Ref` while acquiring a write lock on
70        // the same shard self-deadlocks (the FTS test flake, P53.x).
71        let (columns, col_idx, num_rows, source_data) = {
72            let source_table = match self.table_catalog.get_node_table_by_name(&self.table_name) {
73                Some(t) => t,
74                None => return Err(format!("Table '{}' not found", self.table_name).into()),
75            };
76            let col_idx = source_table
77                .columns
78                .iter()
79                .position(|c| c.name == self.column_name)
80                .ok_or_else(|| format!("Column '{}' not found in '{}'", self.column_name, self.table_name))?;
81            (
82                source_table.columns.clone(),
83                col_idx,
84                source_table.num_rows as usize,
85                source_table.to_column_major_data(),
86            )
87        };
88
89        // Materialize the (doc_id, text) rows from the source snapshot.
90        let mut rows: Vec<(i64, String)> = Vec::with_capacity(num_rows);
91        for row_idx in 0..num_rows {
92            let text = source_data
93                .get(col_idx)
94                .and_then(|col| col.get(row_idx))
95                .and_then(|v| match v {
96                    Value::String(s) => Some(s.clone()),
97                    _ => None,
98                })
99                .unwrap_or_default();
100            rows.push((row_idx as i64, text));
101        }
102
103        // Resolve the tokenizer (P109.1): `WITH TOKENIZER('...')` or the
104        // `en_stem` default; an unsupported name fails the statement.
105        let tokenizer = akar_fts::tokenizer::resolve(self.tokenizer.as_deref())?;
106
107        // Build the Tantivy index (P104.1) — the only FTS representation
108        // (P104.2 clean break). Disk-backed catalogs persist under
109        // `<db_path>/fts/<index_name>`; in-memory ones build an ephemeral index
110        // (searching it is not supported and errors clearly).
111        let index_dir = self
112            .table_catalog
113            .db_path()
114            .filter(|p| p.to_string_lossy() != ":memory:")
115            .map(|p| p.join("fts").join(&self.index_name));
116        akar_fts::build::build_index(&columns, &self.column_name, &tokenizer, index_dir.as_deref(), &rows)?;
117
118        let mut result_vec = akar_common::vector::ValueVector::new(akar_common::types::PhysicalTypeID::String, 1);
119        result_vec.resize(1);
120        result_vec
121            .set_value(
122                0,
123                &Value::String(format!("FTS index '{}' built successfully.", self.index_name)),
124            )
125            .unwrap();
126        let arr = akar_common::arrow_vector::ArrowVector::from_legacy(&result_vec).array;
127        let mut result = DataChunk::new(vec![arr], vec![akar_common::types::PhysicalTypeID::String]);
128        result.size = 1;
129        result.field_names = vec!["result".to_string()];
130        Ok(vec![result])
131    }
132}
133
134/// Physical operator for `USING FTS INDEX` scan — queries the **Tantivy**
135/// index and returns ranked (doc_id, score) pairs (P104.2 clean break).
136#[derive(Debug, Clone)]
137pub struct PhysicalFtsScan {
138    pub index_name: String,
139    pub query_string: String,
140    /// Source node table/column the index was created on (P52.39) — used to
141    /// filter deleted rows at query time (P107.1 keeps the index in sync at
142    /// commit, so no scan-side catch-up is needed).
143    pub table_name: String,
144    pub column_name: String,
145    pub table_catalog: Arc<TableCatalog>,
146}
147
148impl PhysicalOperatorExec for PhysicalFtsScan {
149    fn operator_type(&self) -> &str {
150        "fts_scan"
151    }
152
153    fn execute(&self, _input: Vec<DataChunk>) -> OperatorResult {
154        let index_dir = self.index_dir().ok_or_else(|| {
155            "FTS scan requires a disk-backed database (the FTS index lives on disk; in-memory DBs are not supported — P104.2)"
156                .to_string()
157        })?;
158
159        // P107.1/P107.2: the index is kept in sync at commit time (the commit
160        // hook is the single incremental writer AND the single reloader); the
161        // scan only reuses the shared cached reader.
162        let reader = self.open_reader(&index_dir)?;
163
164        // Parse and run the query against the Tantivy searcher (P105.1).
165        let searcher = reader.searcher();
166        let search_schema = searcher.schema();
167        let text_field = search_schema.get_field(&self.column_name).map_err(|_| {
168            format!(
169                "FTS: column '{}' not found in index '{}'",
170                self.column_name, self.index_name
171            )
172        })?;
173        let doc_id_field = search_schema
174            .get_field(akar_fts::schema::DOC_ID_FIELD)
175            .map_err(|_| "FTS: internal doc_id field missing".to_string())?;
176
177        let limit = searcher.num_docs() as usize;
178        let hits = akar_fts::index::TantivyIndex::search_doc_ids(
179            &reader,
180            &self.query_string,
181            vec![text_field],
182            doc_id_field,
183            limit,
184        )
185        .map_err(|e| format!("FTS: search '{}': {e}", self.query_string))?;
186
187        // Doc validity: a doc is searchable only while its source row still
188        // exists and its text column is non-NULL (soft-deleted rows are
189        // filtered out, P52.39).
190        let source_table = self.table_catalog.get_node_table_by_name(&self.table_name);
191        let source_col = source_table
192            .as_ref()
193            .and_then(|t| t.columns.iter().position(|c| c.name == self.column_name));
194        let doc_valid = |doc_id: i64| -> bool {
195            let Ok(r) = usize::try_from(doc_id) else {
196                return false;
197            };
198            match (&source_table, source_col) {
199                (Some(t), Some(ci)) => r < t.num_rows as usize && matches!(t.get_value(r, ci), Some(Value::String(_))),
200                _ => true, // no source info → keep everything
201            }
202        };
203
204        // Tantivy BM25 (k1=1.2, b=0.75) ranks by descending relevance (P106.1
205        // verifies parity); keep the doc_id (source row index) contract intact.
206        let mut ranked: Vec<(i64, f64)> = Vec::with_capacity(hits.len());
207        for (doc_id, score) in hits {
208            if doc_valid(doc_id) {
209                ranked.push((doc_id, score as f64));
210            }
211        }
212
213        // Return (doc_id, score) data chunks
214        let n = ranked.len();
215        let mut id_vec = akar_common::vector::ValueVector::new(akar_common::types::PhysicalTypeID::Int64, n);
216        let mut score_vec = akar_common::vector::ValueVector::new(akar_common::types::PhysicalTypeID::Double, n);
217        id_vec.resize(n);
218        score_vec.resize(n);
219        for (i, (doc_id, score)) in ranked.into_iter().enumerate() {
220            id_vec.set_i64(i, doc_id);
221            score_vec.set_double(i, score);
222        }
223        let arr1 = akar_common::arrow_vector::ArrowVector::from_legacy(&id_vec).array;
224        let arr2 = akar_common::arrow_vector::ArrowVector::from_legacy(&score_vec).array;
225        let mut chunk = DataChunk::new(
226            vec![arr1, arr2],
227            vec![
228                akar_common::types::PhysicalTypeID::Int64,
229                akar_common::types::PhysicalTypeID::Double,
230            ],
231        );
232        chunk.size = n;
233        chunk.field_names = vec!["doc_id".to_string(), "score".to_string()];
234        Ok(vec![chunk])
235    }
236}
237
238impl PhysicalFtsScan {
239    /// On-disk location of this index's Tantivy directory
240    /// (`<db_path>/fts/<index_name>`).
241    fn index_dir(&self) -> Option<std::path::PathBuf> {
242        self.table_catalog
243            .db_path()
244            .filter(|p| p.to_string_lossy() != ":memory:")
245            .map(|p| p.join("fts").join(&self.index_name))
246    }
247
248    /// Resolve the shared read handle for this index (opening + registering it
249    /// in the catalog on first use) and return its cached reader.
250    ///
251    /// P107.2: **no reload here.** The reader is refreshed by only one place —
252    /// the commit-time sync hook, via
253    /// [`crate::physical::write_ops::fts_sync::sync_indexes_on_commit`]. The
254    /// scan must not write and must not reload; this guarantees every scan
255    /// observes exactly what the last akar commit wrote.
256    fn open_reader(&self, index_dir: &std::path::Path) -> Result<akar_fts::index::IndexReader, String> {
257        if !index_dir.join("meta.json").exists() {
258            return Err(format!(
259                "FTS index '{}' not found on disk at '{}' (run CREATE FTS INDEX first)",
260                self.index_name,
261                index_dir.display()
262            ));
263        }
264
265        let handle = akar_fts::index::runtime_handle(&self.table_catalog, &self.index_name, index_dir)?;
266        handle.reader().map_err(|e| format!("FTS: reader: {e}"))
267    }
268}
akar_processor/physical/write_ops/ddl_fts.rs

akar_processor/physical/write_ops/
ddl_fts.rs
