akar_processor/physical/write_ops/
fts_sync.rs

1//! Commit-time FTS propagation (P107.1).
2//!
3//! [`sync_indexes_on_commit`] is invoked by the connection layer's commit
4//! pipe-line (Step 4.5 of `Connection::commit_write_txn`) with the txn's
5//! written `(table_id, row_id)` pairs. For every registered FTS index over a
6//! touched table it snapshots the current text values of those rows and applies
7//! them to the on-disk Tantivy index via [`akar_fts::build::apply_doc_writes`]:
8//! an update replaces the row's document, a delete (NULL / missing column)
9//! removes it.
10//!
11//! This is the *single* incremental writer AND the *single* reloader: the hook
12//! reloads the shared [`akar_fts::index::FtsIndexHandle`] reader after applying
13//! writes (P107.2). The `PhysicalFtsScan` read path only reuses that cached
14//! reader — never opening, reloading, or writing. Failures here are non-fatal
15//! by contract — they are surfaced as warnings by the caller so a stale index
16//! can never roll back an already-durable commit.
17//!
18//! **Crash recovery (P107.3):** the Tantivy segment commit is the crash
19//! boundary. A crash between the durable akar commit and this hook's
20//! `apply_doc_writes` commit loses only the in-flight increment (the index ends
21//! *stale*, never over-visible, never corrupt — `FtsIndexHandle::open_on_disk`
22//! reopens the last committed generation); the last committed state always
23//! survives. Pinned by `test_fts_crash_recovery_last_committed_survives`
24//! (akar-fts) and `test_fts_crash_recovery_across_db_reopen` (akar-main).
25
26use std::sync::Arc;
27
28use akar_common::types::Value;
29use akar_storage::table::TableCatalog;
30
31/// Apply the written rows of a committed transaction to every FTS index over
32/// a touched table.
33///
34/// `fts_indexes` is the `(name, table_name, column_name)` snapshot from the
35/// catalog; `written_rows` is the txn's deduplicated `(table_id, row_id)`
36/// write set. Returns the number of `(row, text)` writes applied, or an error
37/// message. Indexes with no on-disk Tantivy data yet (index never built /
38/// legacy schema) are skipped with a warning.
39pub fn sync_indexes_on_commit(
40    table_catalog: &Arc<TableCatalog>,
41    fts_indexes: &[(String, String, String)],
42    written_rows: &[(u64, u64)],
43) -> Result<usize, String> {
44    if fts_indexes.is_empty() || written_rows.is_empty() {
45        return Ok(0);
46    }
47
48    // On-disk indexes only: an in-memory database has no `<db_path>/fts/`
49    // directory, so there is nothing to propagate to.
50    let Some(base) = table_catalog.db_path() else {
51        return Ok(0);
52    };
53    if base.to_string_lossy() == ":memory:" {
54        return Ok(0);
55    }
56
57    let written_tables: std::collections::HashSet<u64> = written_rows.iter().map(|(t, _)| *t).collect();
58
59    let mut synced = 0usize;
60    for (name, table_name, column_name) in fts_indexes {
61        // Only indexes whose source table actually wrote rows need syncing.
62        let Some(source_table) = table_catalog.get_node_table_by_name(table_name) else {
63            continue;
64        };
65        let table_id = source_table.table_id;
66        if !written_tables.contains(&table_id) {
67            continue;
68        }
69        let Some(col_idx) = source_table.columns.iter().position(|c| c.name == *column_name) else {
70            continue;
71        };
72
73        let index_dir = base.join("fts").join(name);
74        if !index_dir.join("meta.json").exists() {
75            tracing::warn!(
76                "FTS: index '{name}' has no on-disk Tantivy data; skipping commit-time sync \
77                 (rebuild with DROP + CREATE FTS INDEX if the schema changed)"
78            );
79            continue;
80        }
81
82        // Snapshot the current value of every written row. A single row may
83        // have several undo records within one txn; `get_value` returns the
84        // last committed value. The DashMap `Ref` is scoped out before the
85        // Tantivy writer is opened (DashMap is not re-entrant — the FTS test
86        // flake, P53.x).
87        let writes: Vec<(i64, Option<String>)> = {
88            let mut seen = std::collections::HashSet::with_capacity(written_rows.len());
89            let mut out = Vec::with_capacity(written_rows.len());
90            for (tid, row_id) in written_rows {
91                if *tid != table_id || !seen.insert(*row_id) {
92                    continue;
93                }
94                let Ok(row) = usize::try_from(*row_id) else {
95                    continue;
96                };
97                let text = match source_table.get_value(row, col_idx) {
98                    Some(Value::String(s)) => Some(s.clone()),
99                    _ => None, // NULL (soft-deleted / never set) → delete-term
100                };
101                out.push((*row_id as i64, text));
102            }
103            out
104        };
105
106        if writes.is_empty() {
107            continue;
108        }
109
110        let handle = akar_fts::index::runtime_handle(table_catalog, name, &index_dir)
111            .map_err(|e| format!("FTS: open index '{name}': {e}"))?;
112        akar_fts::build::apply_doc_writes(handle.inner(), column_name, &writes)
113            .map_err(|e| format!("FTS: sync index '{name}': {e}"))?;
114        // P107.2: reload the shared reader HERE, at the akar commit point.
115        // Scans reuse the same cached reader, so this single reload makes every
116        // subsequent scan see the rows just committed — read-after-write
117        // consistency with reload() called only on commit.
118        handle
119            .reload()
120            .map_err(|e| format!("FTS: reload index '{name}': {e}"))?;
121        synced += writes.len();
122    }
123
124    Ok(synced)
125}
akar_processor/physical/write_ops/fts_sync.rs

akar_processor/physical/write_ops/
fts_sync.rs
