Skip to main content

akar_storage/
table.rs

1//! Table storage — columnar node/rel tables with NodeGroup-based storage.
2
3use crate::art_index::ArtPrimaryKeyIndex;
4use crate::art_key::ArtKey;
5use crate::column::Column;
6use crate::column_chunk::ColumnChunk;
7use crate::csr::CsrIndex;
8use crate::index::HashIndex;
9use crate::node_group::NodeGroup;
10use crate::vector_index::VectorIndexTable;
11use akar_common::error::StorageError;
12use akar_common::types::{LogicalTypeID, Value, pk_value_to_string};
13use akar_vector::hnsw::DistanceMetric;
14use dashmap::DashMap;
15use std::collections::HashMap;
16
17/// A column definition within a table.
18#[derive(Debug, Clone)]
19pub struct ColumnDefinition {
20    pub name: String,
21    pub logical_type: LogicalTypeID,
22    pub is_primary_key: bool,
23    pub compression: akar_common::enums::CompressionType,
24}
25
26/// A node table stores properties for a node label using NodeGroup-based
27/// columnar storage. Data is held in-memory as `NodeGroup`s; when a group
28/// reaches `NODE_GROUP_SIZE` rows a new group is automatically created.
29///
30/// Primary key uniqueness is enforced via an in-memory `HashIndex` that maps
31/// PK values to row offsets. For persistent indexes, use `OnDiskHashIndex`
32/// alongside the L1 cache (see `index.rs`).
33///
34/// Optionally, an `ArtPrimaryKeyIndex` can be attached for range-scan
35/// support on the primary key column (see `create_art_index`).
36#[derive(Debug, Clone)]
37pub struct NodeTable {
38    pub table_id: u64,
39    pub name: String,
40    pub columns: Vec<ColumnDefinition>,
41    pub primary_key_column: usize,
42    pub num_rows: u64,
43    /// NodeGroup-based columnar storage. Each group holds up to
44    /// `NODE_GROUP_SIZE` rows across all columns.
45    pub node_groups: Vec<NodeGroup>,
46    /// In-memory hash index for primary key → row lookup and dedup.
47    /// Stores the PK value (as a string representation) mapped to row offset.
48    pub hash_index: HashIndex<String>,
49    /// Optional ART (Adaptive Radix Tree) index for PK range scans.
50    /// When present, `insert_row()` also updates this index automatically.
51    pub art_index: Option<ArtPrimaryKeyIndex>,
52    /// Set when an UPDATE/DELETE touches the table. The durable column mirror
53    /// (see `persistence.rs`) performs a full rewrite when this flag is set.
54    pub persistence_dirty: bool,
55}
56
57/// Sentinel for `NodeTable::primary_key_column` when a node table has no PK
58/// column. SQL always requires a PK (binder rejects `CREATE NODE TABLE` without
59/// one), so via the SQL path this is unreachable; the sentinel only covers
60/// internally/test-constructed tables and keeps "no PK" explicit instead of
61/// silently defaulting to column 0 (P49.2).
62pub const NO_PRIMARY_KEY: usize = usize::MAX;
63
64impl NodeTable {
65    pub fn new(table_id: u64, name: String, columns: Vec<ColumnDefinition>) -> Self {
66        // SQL guarantees a PK exists (binder/mod.rs:861 rejects tables without
67        // one), so the fallback is unreachable in production. Use an explicit
68        // sentinel instead of a silent 0 so a table with no PK column never
69        // accidentally dedups on column 0 (P49.2).
70        let primary_key_column = columns.iter().position(|c| c.is_primary_key).unwrap_or(NO_PRIMARY_KEY);
71        Self {
72            table_id,
73            name,
74            columns,
75            primary_key_column,
76            num_rows: 0,
77            node_groups: Vec::new(),
78            hash_index: HashIndex::new(),
79            art_index: None,
80            persistence_dirty: false,
81        }
82    }
83
84    /// Widen the table schema with a new column (ALTER TABLE ADD). Every
85    /// existing row gets a NULL in the new column and all node groups are
86    /// widened so scans emit it (P53.37). The main DDL catalog is updated by
87    /// the caller; without this storage-side mirror the column silently
88    /// vanishes from scans and the export/projection paths drift.
89    pub fn add_column(&mut self, column: ColumnDefinition) {
90        if self.columns.iter().any(|c| c.name.eq_ignore_ascii_case(&column.name)) {
91            return;
92        }
93        self.columns.push(column);
94        for group in &mut self.node_groups {
95            let mut chunk = ColumnChunk::new();
96            for _ in 0..group.num_nodes {
97                chunk.append(Value::Null);
98            }
99            group.columns.push(chunk);
100        }
101        self.persistence_dirty = true;
102    }
103
104    /// Insert a row of values into the table.
105    ///
106    /// Appends to the current `NodeGroup`; auto-creates a new group when the
107    /// current one is full (reaches `NODE_GROUP_SIZE` rows).
108    ///
109    /// If the table has a primary key column, checks for duplicates and rejects
110    /// rows with already-existing PK values. The hash index is updated after
111    /// a successful insert.
112    ///
113    /// When `txn_id` is `Some(...)`, the insert is recorded in VersionInfo
114    /// for MVCC snapshot isolation.
115    ///
116    /// Returns an error if the number of values doesn't match the number of columns,
117    /// or if a duplicate primary key value is detected.
118    pub fn insert_row(&mut self, values: Vec<Value>) -> Result<u64, StorageError> {
119        self.insert_row_with_txn(values, None)
120    }
121
122    /// Insert a row with an optional transaction ID for MVCC tracking.
123    pub fn insert_row_with_txn(&mut self, mut values: Vec<Value>, txn_id: Option<u64>) -> Result<u64, StorageError> {
124        if values.len() != self.columns.len() {
125            return Err(StorageError::Page(format!(
126                "Column count mismatch: expected {} values, got {}",
127                self.columns.len(),
128                values.len()
129            )));
130        }
131
132        // Reject NULL primary key values
133        if self.primary_key_column < self.columns.len() {
134            let pk_value = &values[self.primary_key_column];
135            if matches!(pk_value, Value::Null) {
136                return Err(StorageError::Page(format!(
137                    "NULL value not allowed for primary key column '{}' in table '{}'",
138                    self.columns[self.primary_key_column].name, self.name
139                )));
140            }
141        }
142
143        // Coerce literal values to the declared column types (P48.12): constant
144        // evaluation (e.g. CREATE `{id: 41}`) produces Int64 literals regardless
145        // of the target column, and a UINT64 column must store Value::UInt64 so
146        // the scan builds the correct Arrow type instead of dropping the value.
147        coerce_values_to_columns(&mut values, &self.columns)?;
148
149        // Check primary key uniqueness
150        if self.primary_key_column < self.columns.len() {
151            let pk_value = &values[self.primary_key_column];
152            let pk_key = pk_value_to_string(pk_value);
153            if self.hash_index.lookup(&pk_key).is_some() {
154                return Err(StorageError::Index(format!(
155                    "Duplicate primary key value: '{pk_key}' in table '{}'",
156                    self.name
157                )));
158            }
159        }
160
161        // Get or create the current node group.
162        let num_cols = self.columns.len();
163        if self.node_groups.is_empty() || self.node_groups.last().unwrap().is_full() {
164            let start_offset = self.num_rows;
165            let mut new_group = NodeGroup::new(num_cols, start_offset);
166            // Enable version info if MVCC tracking is requested
167            if txn_id.is_some() {
168                new_group.enable_version_info();
169            }
170            self.node_groups.push(new_group);
171        }
172
173        let current = self.node_groups.last_mut().unwrap();
174        // Enable version info on existing group if needed
175        if txn_id.is_some() {
176            current.enable_version_info();
177        }
178        current.append_row_with_txn(values.clone(), txn_id)?;
179        self.num_rows += 1;
180
181        // Update hash index with the PK value for this row
182        if self.primary_key_column < self.columns.len() {
183            let pk_value = &values[self.primary_key_column];
184            let pk_key = pk_value_to_string(pk_value);
185            self.hash_index.insert(pk_key, self.num_rows - 1);
186
187            // Also update ART index if present
188            if let Some(ref mut art_idx) = self.art_index
189                && let Some(art_key) = ArtKey::from_value(pk_value)
190            {
191                art_idx.insert(&art_key, self.num_rows - 1);
192            }
193        }
194
195        Ok(self.num_rows - 1)
196    }
197
198    /// Batch insert multiple rows efficiently.
199    /// Validates PK uniqueness, pre-allocates node groups, and bulk-appends.
200    /// When `txn_id` is `Some(...)`, inserts are recorded in VersionInfo for MVCC.
201    pub fn insert_rows_batch(&mut self, rows: &[Vec<Value>]) -> Result<u64, StorageError> {
202        self.insert_rows_batch_with_txn(rows, None)
203    }
204
205    /// Batch insert with optional MVCC tracking.
206    pub fn insert_rows_batch_with_txn(
207        &mut self,
208        rows: &[Vec<Value>],
209        txn_id: Option<u64>,
210    ) -> Result<u64, StorageError> {
211        if rows.is_empty() {
212            return Ok(0);
213        }
214        let num_cols = self.columns.len();
215
216        // Validate all rows first
217        for (i, row) in rows.iter().enumerate() {
218            if row.len() != num_cols {
219                return Err(StorageError::Page(format!(
220                    "Row {} column count mismatch: expected {} values, got {}",
221                    i,
222                    num_cols,
223                    row.len()
224                )));
225            }
226            // Reject NULL primary key values
227            if self.primary_key_column < num_cols {
228                let pk_value = &row[self.primary_key_column];
229                if matches!(pk_value, Value::Null) {
230                    return Err(StorageError::Page(format!(
231                        "NULL value not allowed for primary key column '{}' in table '{}'",
232                        self.columns[self.primary_key_column].name, self.name
233                    )));
234                }
235            }
236            // Check PK uniqueness
237            if self.primary_key_column < num_cols {
238                let pk_key = pk_value_to_string(&row[self.primary_key_column]);
239                if self.hash_index.lookup(&pk_key).is_some() {
240                    return Err(StorageError::Index(format!(
241                        "Duplicate primary key value: '{pk_key}' in table '{}'",
242                        self.name
243                    )));
244                }
245            }
246        }
247
248        let start_offset = self.num_rows;
249        let total_new = rows.len();
250
251        // Coerce literal values to the declared column types (P48.12), mirroring
252        // `insert_row_with_txn`. This must happen before appending and index
253        // updates so PK hash/ART keys use the coerced (UInt64) encoding.
254        let mut coerced_rows = rows.to_vec();
255        for row in &mut coerced_rows {
256            coerce_values_to_columns(row, &self.columns)?;
257        }
258        let rows: &[Vec<Value>] = &coerced_rows;
259
260        // Ensure we have a node group with enough capacity
261        if self.node_groups.is_empty() || self.node_groups.last().unwrap().is_full() {
262            let off = if self.node_groups.is_empty() {
263                start_offset
264            } else {
265                self.num_rows
266            };
267            let mut new_group = NodeGroup::new(num_cols, off);
268            if txn_id.is_some() {
269                new_group.enable_version_info();
270            }
271            self.node_groups.push(new_group);
272        }
273
274        // Append to the last group (spilling into new groups if needed)
275        let mut inserted = 0usize;
276        while inserted < total_new {
277            let current = self.node_groups.last_mut().unwrap();
278            if txn_id.is_some() {
279                current.enable_version_info();
280            }
281            let rem = current.remaining();
282            let take = (total_new - inserted).min(rem);
283            for row in &rows[inserted..inserted + take] {
284                current.append_row_with_txn(row.clone(), txn_id)?;
285            }
286            self.num_rows += take as u64;
287            inserted += take;
288            if inserted < total_new {
289                let off = self.num_rows;
290                let mut new_group = NodeGroup::new(num_cols, off);
291                if txn_id.is_some() {
292                    new_group.enable_version_info();
293                }
294                self.node_groups.push(new_group);
295            }
296        }
297
298        // Batch update indexes
299        for (i, row) in rows.iter().enumerate() {
300            if self.primary_key_column < num_cols {
301                let pk_key = pk_value_to_string(&row[self.primary_key_column]);
302                self.hash_index.insert(pk_key, start_offset + i as u64);
303                if let Some(ref mut art_idx) = self.art_index
304                    && let Some(art_key) = ArtKey::from_value(&row[self.primary_key_column])
305                {
306                    art_idx.insert(&art_key, start_offset + i as u64);
307                }
308            }
309        }
310
311        Ok(rows.len() as u64)
312    }
313
314    /// Look up a row offset by its primary key value.
315    ///
316    /// Returns `Some(row_offset)` if the PK exists, or `None` if not found.
317    /// Uses the in-memory hash index for O(1) lookup.
318    pub fn lookup_by_pk(&self, pk_value: &Value) -> Option<u64> {
319        let pk_key = pk_value_to_string(pk_value);
320        self.hash_index.lookup(&pk_key)
321    }
322
323    /// Batch look up row offsets for multiple primary key values.
324    ///
325    /// Returns a `Vec<Option<u64>>` parallel to the input, where each element
326    /// is `Some(row_offset)` if the PK exists, or `None` if not found.
327    /// Uses the in-memory hash index for O(1) per-key lookup, avoiding
328    /// per-row method-call overhead by inlining the lookup logic.
329    pub fn lookup_by_pk_batch(&self, pk_values: &[Value]) -> Vec<Option<u64>> {
330        pk_values
331            .iter()
332            .map(|pk_value| {
333                let pk_key = pk_value_to_string(pk_value);
334                self.hash_index.lookup(&pk_key)
335            })
336            .collect()
337    }
338
339    /// Perform a range scan on the primary key column using the ART index.
340    ///
341    /// Returns up to `max_results` row offsets for keys within `[lower, upper]`
342    /// (respecting inclusivity flags). Returns an empty vec if no ART index
343    /// exists or no keys match.
344    ///
345    /// This is the bridge function called by `PhysicalArtIndexRangeScan`.
346    pub fn lookup_by_pk_range(
347        &self,
348        lower: Option<&Value>,
349        lower_inclusive: bool,
350        upper: Option<&Value>,
351        upper_inclusive: bool,
352        max_results: u64,
353    ) -> Vec<u64> {
354        match &self.art_index {
355            Some(idx) => {
356                let lower_key = lower.and_then(ArtKey::from_value);
357                let upper_key = upper.and_then(ArtKey::from_value);
358                idx.range_scan(
359                    lower_key.as_ref(),
360                    lower_inclusive,
361                    upper_key.as_ref(),
362                    upper_inclusive,
363                    max_results,
364                )
365            }
366            None => Vec::new(),
367        }
368    }
369
370    /// Scan all values for a given column across all node groups.
371    ///
372    /// Returns a flat `Vec<Value>` containing values from `start` to
373    /// `start + count` (or fewer if the end of the table is reached).
374    ///
375    /// If `snapshot_ts` is `Some(...)`, performs MVCC snapshot isolation:
376    /// rows inserted/deleted by transactions committed after `snapshot_ts`
377    /// are excluded, and versioned updates are resolved.
378    pub fn scan_column(
379        &self,
380        col_idx: usize,
381        start: u64,
382        count: u64,
383        snapshot_ts: Option<u64>,
384        commit_history: &[(u64, u64)],
385    ) -> Vec<Value> {
386        if col_idx >= self.columns.len() || start >= self.num_rows {
387            return Vec::new();
388        }
389        let end = (start + count).min(self.num_rows);
390        let mut result = Vec::with_capacity((end - start) as usize);
391
392        // Find the first node group containing `start`.
393        let group_start = self.find_group(start);
394        let mut remaining = end - start;
395
396        for g_idx in group_start..self.node_groups.len() {
397            if remaining == 0 {
398                break;
399            }
400            let group = &self.node_groups[g_idx];
401            let local_start = if g_idx == group_start {
402                (start - group.start_offset) as usize
403            } else {
404                0
405            };
406            let available = (group.num_nodes as usize).saturating_sub(local_start);
407            let take = available.min(remaining as usize);
408
409            for row in local_start..local_start + take {
410                let val = group.get_value_with_snapshot(row, col_idx, snapshot_ts, commit_history);
411                match val {
412                    Some(v) => result.push(v.clone()),
413                    None => result.push(Value::Null),
414                }
415            }
416            remaining -= take as u64;
417        }
418
419        result
420    }
421
422    /// Rebuild the table's in-memory state from rows loaded off the durable
423    /// column mirror (see `persistence.rs`).
424    ///
425    /// Populates `node_groups`, `num_rows`, the PK hash index, and the ART
426    /// index directly, bypassing PK uniqueness checks so that soft-deleted
427    /// rows (whose PK is `Null`) can be restored at their original row
428    /// offsets.
429    pub fn load_persisted_rows(&mut self, rows: Vec<Vec<Value>>) -> Result<(), StorageError> {
430        let num_cols = self.columns.len();
431        self.node_groups.clear();
432        self.hash_index.clear();
433
434        let mut offset = 0u64;
435        let mut group = NodeGroup::new(num_cols, offset);
436        for row in &rows {
437            if row.len() != num_cols {
438                return Err(StorageError::Page(format!(
439                    "load_persisted_rows: expected {num_cols} values, got {}",
440                    row.len()
441                )));
442            }
443            group.append_row_with_txn(row.clone(), None)?;
444            offset += 1;
445            if group.is_full() {
446                self.node_groups.push(group);
447                group = NodeGroup::new(num_cols, offset);
448            }
449        }
450        if group.num_nodes > 0 {
451            self.node_groups.push(group);
452        }
453        self.num_rows = offset;
454
455        // Rebuild the PK hash index (skip soft-deleted rows with a Null PK).
456        if self.primary_key_column < num_cols {
457            for (row_idx, values) in rows.iter().enumerate() {
458                let pk = &values[self.primary_key_column];
459                if matches!(pk, Value::Null) {
460                    continue;
461                }
462                let key = pk_value_to_string(pk);
463                self.hash_index.insert(key, row_idx as u64);
464            }
465        }
466
467        // Rebuild the ART index if present.
468        if self.art_index.is_some() && self.primary_key_column < num_cols {
469            if let Some(art) = &mut self.art_index {
470                art.clear();
471                for (row_idx, values) in rows.iter().enumerate() {
472                    let pk = &values[self.primary_key_column];
473                    if matches!(pk, Value::Null) {
474                        continue;
475                    }
476                    if let Some(key) = ArtKey::from_value(pk) {
477                        art.insert(&key, row_idx as u64);
478                    }
479                }
480            }
481        }
482        Ok(())
483    }
484
485    /// Update a single cell (row, column) with a new value.
486    pub fn update_cell(&mut self, row_idx: u64, col_idx: usize, value: Value) -> Result<(), StorageError> {
487        if col_idx >= self.columns.len() {
488            return Err(StorageError::Page(format!("Column index {col_idx} out of range")));
489        }
490        if row_idx >= self.num_rows {
491            return Err(StorageError::Page(format!(
492                "Row index {row_idx} out of range (num_rows={})",
493                self.num_rows
494            )));
495        }
496        self.persistence_dirty = true;
497
498        let mut offset = 0u64;
499        for group in &mut self.node_groups {
500            if row_idx < offset + group.num_nodes {
501                let local_row = (row_idx - offset) as usize;
502                if let Some(col_chunk) = group.columns.get_mut(col_idx) {
503                    col_chunk.set_value(local_row, value)?;
504                }
505                return Ok(());
506            }
507            offset += group.num_nodes;
508        }
509        Err(StorageError::Page(format!(
510            "Row index {row_idx} not found in any node group"
511        )))
512    }
513
514    /// Delete a row by its index. Marks the row as null by setting all its column
515    /// values to `Value::Null`. This is a soft delete — the row slot remains.
516    pub fn delete_row(&mut self, row_idx: u64) -> Result<(), StorageError> {
517        self.delete_row_with_txn(row_idx, None)
518    }
519
520    /// Delete a row with optional MVCC tracking.
521    ///
522    /// When `txn_id` is `Some(...)`, the delete is recorded in VersionInfo
523    /// for MVCC snapshot isolation.
524    pub fn delete_row_with_txn(&mut self, row_idx: u64, txn_id: Option<u64>) -> Result<(), StorageError> {
525        if row_idx >= self.num_rows {
526            return Err(StorageError::Page(format!(
527                "Row index {row_idx} out of range (num_rows={})",
528                self.num_rows
529            )));
530        }
531        self.persistence_dirty = true;
532
533        // Locate the node group containing this row
534        let mut offset = 0u64;
535        for group in &mut self.node_groups {
536            if row_idx < offset + group.num_nodes {
537                let local_row = (row_idx - offset) as usize;
538                // Record delete in VersionInfo if MVCC tracking is active
539                if let Some(txn) = txn_id {
540                    if let Some(ref vi) = group.version_info {
541                        vi.delete(txn, local_row as u32);
542                    }
543                }
544                // Capture the PK value before it is nulled so the in-memory
545                // PK indexes can be kept in sync (P52.16).
546                let pk_value = (self.primary_key_column < self.columns.len())
547                    .then(|| group.columns.get(self.primary_key_column))
548                    .flatten()
549                    .and_then(|chunk| chunk.get(local_row))
550                    .cloned();
551                // Set all columns to Null for this row
552                for col_chunk in &mut group.columns {
553                    let _ = col_chunk.set_value(local_row, Value::Null);
554                }
555                // Soft-deleted rows must no longer resolve via PK lookup, and
556                // the same PK must be re-insertable. Drop it from the hash and
557                // ART indexes (P52.16).
558                if let Some(pk) = pk_value
559                    && !matches!(pk, Value::Null)
560                {
561                    let pk_key = pk_value_to_string(&pk);
562                    self.hash_index.delete(&pk_key);
563                    if let Some(ref mut art_idx) = self.art_index
564                        && let Some(art_key) = ArtKey::from_value(&pk)
565                    {
566                        art_idx.delete(&art_key, row_idx);
567                    }
568                }
569                return Ok(());
570            }
571            offset += group.num_nodes;
572        }
573        Err(StorageError::Page(format!(
574            "Row index {row_idx} not found in any node group"
575        )))
576    }
577
578    /// Capture the full row (all columns) as serialized undo bytes.
579    /// Used by the write path to record `UndoType::Delete` records so a
580    /// rollback can restore a soft-deleted row (P52.18).
581    pub fn row_undo_bytes(&self, row_idx: u64) -> Vec<u8> {
582        let mut out = Vec::new();
583        for col in 0..self.columns.len() {
584            let val = self.get_value(row_idx as usize, col).cloned().unwrap_or(Value::Null);
585            out.extend_from_slice(&Column::serialize_value(&val));
586        }
587        out
588    }
589
590    /// Capture a single cell as serialized undo bytes.
591    /// Used to record `UndoType::Update` records for `SET` rollback (P52.18).
592    pub fn cell_undo_bytes(&self, row_idx: u64, col_idx: usize) -> Vec<u8> {
593        let val = self
594            .get_value(row_idx as usize, col_idx)
595            .cloned()
596            .unwrap_or(Value::Null);
597        Column::serialize_value(&val)
598    }
599
600    /// Get a single value at (row, col) by locating the correct `NodeGroup`
601    /// and `ColumnChunk`.
602    pub fn get_value(&self, row: usize, col: usize) -> Option<&Value> {
603        self.get_value_with_snapshot(row, col, None, &[])
604    }
605
606    /// Get a single value with MVCC snapshot isolation.
607    ///
608    /// Checks `VersionInfo` for insert/delete visibility and `UpdateInfo`
609    /// version chains when `snapshot_ts` is provided.
610    pub fn get_value_with_snapshot(
611        &self,
612        row: usize,
613        col: usize,
614        snapshot_ts: Option<u64>,
615        commit_history: &[(u64, u64)],
616    ) -> Option<&Value> {
617        if col >= self.columns.len() || row as u64 >= self.num_rows {
618            return None;
619        }
620        let group_idx = self.find_group(row as u64);
621        let group = self.node_groups.get(group_idx)?;
622        let local_row = row as u64 - group.start_offset;
623        group.get_value_with_snapshot(local_row as usize, col, snapshot_ts, commit_history)
624    }
625
626    /// Reconstruct column-major data (`Vec<Vec<Value>>`) from all node groups.
627    ///
628    /// Used by the processor (`resolve_scan_data`) for backward compatibility.
629    pub fn to_column_major_data(&self) -> Vec<Vec<Value>> {
630        self.to_column_major_data_with_predicate(None)
631    }
632
633    /// Like `to_column_major_data`, but applies an optional zone map predicate
634    /// `(col_idx, op_string, val)` to skip entire node groups.
635    pub fn to_column_major_data_with_predicate(&self, predicate: Option<(usize, &str, &Value)>) -> Vec<Vec<Value>> {
636        let num_cols = self.columns.len();
637        let mut result = vec![Vec::new(); num_cols]; // Avoid allocating self.num_rows if we skip chunks
638
639        for group in &self.node_groups {
640            if let Some((col_idx, op, val)) = predicate
641                && let Some(col_chunk) = group.columns.get(col_idx)
642            {
643                use crate::predicate::{ZoneMapCheckResult, check_zone_map};
644                if check_zone_map(&col_chunk.stats, op, val) == ZoneMapCheckResult::SkipScan {
645                    continue; // Skip this entire node group
646                }
647            }
648
649            for row in 0..group.num_nodes as usize {
650                for (col, res_col) in result.iter_mut().enumerate().take(num_cols) {
651                    match group.get_value(row, col) {
652                        Some(v) => res_col.push(v.clone()),
653                        None => res_col.push(Value::Null),
654                    }
655                }
656            }
657        }
658
659        result
660    }
661
662    /// Like `to_column_major_data`, but with MVCC snapshot isolation.
663    ///
664    /// When `snapshot_ts` is `Some(...)`, rows inserted/deleted by transactions
665    /// committed after `snapshot_ts` are excluded, and versioned updates are
666    /// resolved to the value visible at that snapshot.
667    pub fn to_column_major_data_with_snapshot(
668        &self,
669        snapshot_ts: Option<u64>,
670        commit_history: &[(u64, u64)],
671    ) -> Vec<Vec<Value>> {
672        let num_cols = self.columns.len();
673        let mut result = vec![Vec::new(); num_cols];
674
675        for group in &self.node_groups {
676            for row in 0..group.num_nodes as usize {
677                for (col, res_col) in result.iter_mut().enumerate().take(num_cols) {
678                    match group.get_value_owned_with_snapshot(row, col, snapshot_ts, commit_history) {
679                        Some(v) => res_col.push(v),
680                        None => res_col.push(Value::Null),
681                    }
682                }
683            }
684        }
685
686        result
687    }
688
689    /// Like `to_column_major_data_with_snapshot`, but applies an optional zone
690    /// map predicate to skip entire node groups.
691    pub fn to_column_major_data_with_snapshot_and_predicate(
692        &self,
693        predicate: Option<(usize, &str, &Value)>,
694        snapshot_ts: Option<u64>,
695        commit_history: &[(u64, u64)],
696    ) -> Vec<Vec<Value>> {
697        let num_cols = self.columns.len();
698        let mut result = vec![Vec::new(); num_cols];
699
700        for group in &self.node_groups {
701            if let Some((col_idx, op, val)) = predicate
702                && let Some(col_chunk) = group.columns.get(col_idx)
703            {
704                use crate::predicate::{ZoneMapCheckResult, check_zone_map};
705                if check_zone_map(&col_chunk.stats, op, val) == ZoneMapCheckResult::SkipScan {
706                    continue;
707                }
708            }
709
710            for row in 0..group.num_nodes as usize {
711                for (col, res_col) in result.iter_mut().enumerate().take(num_cols) {
712                    match group.get_value_owned_with_snapshot(row, col, snapshot_ts, commit_history) {
713                        Some(v) => res_col.push(v),
714                        None => res_col.push(Value::Null),
715                    }
716                }
717            }
718        }
719
720        result
721    }
722
723    /// Like `to_column_major_data_with_predicate`, but additionally returns the
724    /// internal node id (global row offset) for every emitted row, in the same
725    /// order as the returned column data. This is the node id space used by the
726    /// processor's extend/insert/join operators (`<var>._id`).
727    pub fn to_column_major_data_with_predicate_and_ids(
728        &self,
729        predicate: Option<(usize, &str, &Value)>,
730    ) -> (Vec<Vec<Value>>, Vec<u64>) {
731        let num_cols = self.columns.len();
732        let mut result = vec![Vec::new(); num_cols];
733        let mut ids = Vec::new();
734
735        for group in &self.node_groups {
736            if let Some((col_idx, op, val)) = predicate
737                && let Some(col_chunk) = group.columns.get(col_idx)
738            {
739                use crate::predicate::{ZoneMapCheckResult, check_zone_map};
740                if check_zone_map(&col_chunk.stats, op, val) == ZoneMapCheckResult::SkipScan {
741                    continue;
742                }
743            }
744
745            for row in 0..group.num_nodes as usize {
746                ids.push(group.start_offset + row as u64);
747                for (col, res_col) in result.iter_mut().enumerate().take(num_cols) {
748                    match group.get_value(row, col) {
749                        Some(v) => res_col.push(v.clone()),
750                        None => res_col.push(Value::Null),
751                    }
752                }
753            }
754        }
755
756        (result, ids)
757    }
758
759    /// Like `to_column_major_data_with_snapshot_and_predicate`, but additionally
760    /// returns the internal node id (global row offset) for every emitted row,
761    /// in the same order as the returned column data.
762    pub fn to_column_major_data_with_snapshot_and_predicate_and_ids(
763        &self,
764        predicate: Option<(usize, &str, &Value)>,
765        snapshot_ts: Option<u64>,
766        commit_history: &[(u64, u64)],
767    ) -> (Vec<Vec<Value>>, Vec<u64>) {
768        let num_cols = self.columns.len();
769        let mut result = vec![Vec::new(); num_cols];
770        let mut ids = Vec::new();
771
772        for group in &self.node_groups {
773            if let Some((col_idx, op, val)) = predicate
774                && let Some(col_chunk) = group.columns.get(col_idx)
775            {
776                use crate::predicate::{ZoneMapCheckResult, check_zone_map};
777                if check_zone_map(&col_chunk.stats, op, val) == ZoneMapCheckResult::SkipScan {
778                    continue;
779                }
780            }
781
782            for row in 0..group.num_nodes as usize {
783                ids.push(group.start_offset + row as u64);
784                for (col, res_col) in result.iter_mut().enumerate().take(num_cols) {
785                    match group.get_value_owned_with_snapshot(row, col, snapshot_ts, commit_history) {
786                        Some(v) => res_col.push(v),
787                        None => res_col.push(Value::Null),
788                    }
789                }
790            }
791        }
792
793        (result, ids)
794    }
795
796    /// Binary-search for the node group that contains `row`.
797    fn find_group(&self, row: u64) -> usize {
798        self.node_groups
799            .binary_search_by_key(&row, |g| g.start_offset)
800            .unwrap_or_else(|i| if i == 0 { 0 } else { i - 1 })
801    }
802}
803
804/// Coerce a row's values to the table's declared column logical types.
805///
806/// Constant evaluation (e.g. CREATE `{id: 41}`) produces `Value::Int64`
807/// literals regardless of the target column type. A UINT64 column must store
808/// `Value::UInt64`: the scan builds its Arrow type from the physical type of
809/// the column, and the ART primary-key index encodes signed and unsigned keys
810/// differently. Mixed signed/unsigned keys would corrupt range scans.
811fn coerce_values_to_columns(values: &mut [Value], columns: &[ColumnDefinition]) -> Result<(), StorageError> {
812    for (i, col) in columns.iter().enumerate() {
813        let coerced = match (col.logical_type, &values[i]) {
814            (LogicalTypeID::UInt64, Value::Int64(x)) if *x >= 0 => Some(Value::UInt64(*x as u64)),
815            (LogicalTypeID::UInt64, Value::Int32(x)) if *x >= 0 => Some(Value::UInt64(*x as u64)),
816            (LogicalTypeID::UInt64, Value::Int16(x)) if *x >= 0 => Some(Value::UInt64(*x as u64)),
817            (LogicalTypeID::UInt64, Value::Int8(x)) if *x >= 0 => Some(Value::UInt64(*x as u64)),
818            (LogicalTypeID::UInt64, Value::UInt32(x)) => Some(Value::UInt64(*x as u64)),
819            (LogicalTypeID::UInt64, Value::UInt16(x)) => Some(Value::UInt64(*x as u64)),
820            (LogicalTypeID::UInt64, Value::UInt8(x)) => Some(Value::UInt64(*x as u64)),
821            (LogicalTypeID::UInt64, Value::Int64(_))
822            | (LogicalTypeID::UInt64, Value::Int32(_))
823            | (LogicalTypeID::UInt64, Value::Int16(_))
824            | (LogicalTypeID::UInt64, Value::Int8(_)) => {
825                return Err(StorageError::Page(format!(
826                    "Cannot store negative value in UINT64 column '{}'",
827                    col.name
828                )));
829            }
830            _ => None,
831        };
832        if let Some(c) = coerced {
833            values[i] = c;
834        }
835    }
836    Ok(())
837}
838
839/// A relationship (edge) table with CSR (Compressed Sparse Row) adjacency storage.
840///
841/// Each edge connects a source node to a destination node and may carry
842/// a set of property values (one per column in `columns`).
843///
844/// # Storage layout
845///
846/// - `edges` — flat edge list: `edge_idx → (src_offset, dst_offset)`
847/// - `fwd_adj` — forward index: `src_offset → Vec<(dst_offset, edge_idx)>`
848/// - `rev_adj` — reverse index: `dst_offset → Vec<(src_offset, edge_idx)>`
849/// - `properties` — column-major property storage: `properties[col_idx][edge_idx]`
850#[derive(Debug, Clone)]
851pub struct RelTable {
852    pub table_id: u64,
853    pub name: String,
854    pub src_table_id: u64,
855    pub dst_table_id: u64,
856    pub columns: Vec<ColumnDefinition>,
857    pub num_rows: u64,
858    /// Flat edge list: edge_idx → (src_offset, dst_offset).
859    pub edges: Vec<(u64, u64)>,
860    /// Forward CSR adjacency: src_offset → [(dst_offset, edge_idx), ...].
861    pub fwd_adj: HashMap<u64, Vec<(u64, usize)>>,
862    /// Reverse CSR adjacency: dst_offset → [(src_offset, edge_idx), ...].
863    pub rev_adj: HashMap<u64, Vec<(u64, usize)>>,
864    /// Specialized CSR Index for fast graph traversals.
865    pub csr_index: Option<CsrIndex>,
866    /// Column-major property storage: properties[col_idx][edge_idx].
867    pub properties: Vec<Vec<Value>>,
868    /// Set when an UPDATE/DELETE touches the table. The durable column mirror
869    /// (see `persistence.rs`) performs a full rewrite when this flag is set.
870    pub persistence_dirty: bool,
871}
872
873impl RelTable {
874    pub fn new(
875        table_id: u64,
876        name: String,
877        src_table_id: u64,
878        dst_table_id: u64,
879        columns: Vec<ColumnDefinition>,
880    ) -> Self {
881        let num_cols = columns.len();
882        Self {
883            table_id,
884            name,
885            src_table_id,
886            dst_table_id,
887            columns,
888            num_rows: 0,
889            edges: Vec::new(),
890            fwd_adj: HashMap::new(),
891            rev_adj: HashMap::new(),
892            csr_index: None,
893            properties: vec![Vec::new(); num_cols],
894            persistence_dirty: false,
895        }
896    }
897
898    /// Widen the rel table schema with a new property column (ALTER TABLE ADD).
899    /// Existing edges get a NULL in the new property column (P53.37).
900    pub fn add_column(&mut self, column: ColumnDefinition) {
901        if self.columns.iter().any(|c| c.name.eq_ignore_ascii_case(&column.name)) {
902            return;
903        }
904        self.columns.push(column);
905        self.properties.push(vec![Value::Null; self.edges.len()]);
906        self.persistence_dirty = true;
907    }
908
909    /// Insert a relationship (edge) between two nodes with property values.
910    ///
911    /// `from` and `to` are the node offsets of the source and destination
912    /// nodes within their respective tables.
913    ///
914    /// Returns an error if the number of values doesn't match the number
915    /// of property columns.
916    pub fn insert_rel(&mut self, from: u64, to: u64, values: Vec<Value>) -> Result<(), StorageError> {
917        if values.len() != self.columns.len() {
918            return Err(StorageError::Page(format!(
919                "Column count mismatch: expected {} values, got {}",
920                self.columns.len(),
921                values.len()
922            )));
923        }
924
925        let edge_idx = self.edges.len();
926        self.edges.push((from, to));
927
928        // Update forward adjacency.
929        self.fwd_adj.entry(from).or_default().push((to, edge_idx));
930
931        // Update reverse adjacency.
932        self.rev_adj.entry(to).or_default().push((from, edge_idx));
933        // Store property values.
934        for (col_idx, val) in values.into_iter().enumerate() {
935            self.properties[col_idx].push(val);
936        }
937        self.num_rows += 1;
938        Ok(())
939    }
940
941    /// Batch insert multiple relations efficiently.
942    /// Each tuple is (from_offset, to_offset, property_values).
943    pub fn insert_rels_batch(&mut self, rels: &[(u64, u64, Vec<Value>)]) -> Result<u64, StorageError> {
944        if rels.is_empty() {
945            return Ok(0);
946        }
947        let num_cols = self.columns.len();
948        let total = rels.len();
949
950        // Validate all rows
951        for (i, (_, _, vals)) in rels.iter().enumerate() {
952            if vals.len() != num_cols {
953                return Err(StorageError::Page(format!(
954                    "Rel {} column count mismatch: expected {} values, got {}",
955                    i,
956                    num_cols,
957                    vals.len()
958                )));
959            }
960        }
961
962        // Pre-allocate
963        self.edges.reserve(total);
964        for col in &mut self.properties {
965            col.reserve(total);
966        }
967
968        let _start_edge_idx = self.edges.len();
969
970        // Batch append
971        for (from, to, vals) in rels {
972            let edge_idx = self.edges.len();
973            self.edges.push((*from, *to));
974            self.fwd_adj.entry(*from).or_default().push((*to, edge_idx));
975            self.rev_adj.entry(*to).or_default().push((*from, edge_idx));
976            for (col_idx, val) in vals.iter().enumerate() {
977                self.properties[col_idx].push(val.clone());
978            }
979        }
980
981        self.num_rows += total as u64;
982        Ok(total as u64)
983    }
984
985    /// Delete an edge by its index. Marks the edge as deleted by removing it from adjacency lists
986    /// and setting its properties to Null.
987    pub fn delete_edge(&mut self, edge_idx: usize) -> Result<(), StorageError> {
988        if edge_idx >= self.edges.len() {
989            return Err(StorageError::Page(format!("Edge index {edge_idx} out of range")));
990        }
991
992        let (src, dst) = self.edges[edge_idx];
993        if src == u64::MAX {
994            // Already deleted
995            return Ok(());
996        }
997
998        // Remove from fwd_adj
999        if let Some(adj) = self.fwd_adj.get_mut(&src) {
1000            adj.retain(|&(_, idx)| idx != edge_idx);
1001        }
1002
1003        // Remove from rev_adj
1004        if let Some(adj) = self.rev_adj.get_mut(&dst) {
1005            adj.retain(|&(_, idx)| idx != edge_idx);
1006        }
1007
1008        // Tombstone the edge
1009        self.edges[edge_idx] = (u64::MAX, u64::MAX);
1010        self.persistence_dirty = true;
1011
1012        // Nullify properties
1013        for col in &mut self.properties {
1014            if edge_idx < col.len() {
1015                col[edge_idx] = Value::Null;
1016            }
1017        }
1018
1019        Ok(())
1020    }
1021
1022    /// Update a single cell (edge property) with a new value.
1023    pub fn update_cell(&mut self, edge_idx: usize, col_idx: usize, value: Value) -> Result<(), StorageError> {
1024        if col_idx >= self.columns.len() {
1025            return Err(StorageError::Page(format!("Column index {col_idx} out of range")));
1026        }
1027        if edge_idx >= self.properties[col_idx].len() {
1028            return Err(StorageError::Page(format!("Edge index {edge_idx} out of range")));
1029        }
1030
1031        self.properties[col_idx][edge_idx] = value;
1032        self.persistence_dirty = true;
1033        Ok(())
1034    }
1035
1036    /// Capture an edge (src, dst) plus all property values as serialized undo
1037    /// bytes: `[src, dst, prop0..propN]`. Used to record `UndoType::Delete`
1038    /// records so a rollback can restore a deleted edge (P52.18).
1039    pub fn edge_undo_bytes(&self, edge_idx: usize) -> Vec<u8> {
1040        let (src, dst) = self.edges.get(edge_idx).copied().unwrap_or((u64::MAX, u64::MAX));
1041        let mut out = Vec::new();
1042        out.extend_from_slice(&Column::serialize_value(&Value::UInt64(src)));
1043        out.extend_from_slice(&Column::serialize_value(&Value::UInt64(dst)));
1044        for p in self.get_edge_properties(edge_idx) {
1045            out.extend_from_slice(&Column::serialize_value(&p));
1046        }
1047        out
1048    }
1049
1050    /// Capture a single edge property as serialized undo bytes.
1051    /// Used to record `UndoType::Update` records for `SET` rollback (P52.18).
1052    pub fn edge_cell_undo_bytes(&self, edge_idx: usize, col_idx: usize) -> Vec<u8> {
1053        let v = self
1054            .get_edge_properties(edge_idx)
1055            .get(col_idx)
1056            .cloned()
1057            .unwrap_or(Value::Null);
1058        Column::serialize_value(&v)
1059    }
1060
1061    /// Restore a tombstoned edge (rollback of a `DELETE` edge). Re-adds the
1062    /// edge to the forward/reverse adjacency lists and restores its properties
1063    /// (P52.18).
1064    pub fn restore_deleted_edge(
1065        &mut self,
1066        edge_idx: usize,
1067        src: u64,
1068        dst: u64,
1069        props: Vec<Value>,
1070    ) -> Result<(), StorageError> {
1071        if edge_idx >= self.edges.len() {
1072            return Err(StorageError::Page(format!("Edge index {edge_idx} out of range")));
1073        }
1074        self.edges[edge_idx] = (src, dst);
1075        self.fwd_adj.entry(src).or_default().push((dst, edge_idx));
1076        self.rev_adj.entry(dst).or_default().push((src, edge_idx));
1077        for (col_idx, val) in props.into_iter().enumerate() {
1078            if col_idx < self.properties.len() {
1079                if edge_idx < self.properties[col_idx].len() {
1080                    self.properties[col_idx][edge_idx] = val;
1081                } else {
1082                    self.properties[col_idx].push(val);
1083                }
1084            }
1085        }
1086        self.persistence_dirty = true;
1087        Ok(())
1088    }
1089
1090    /// Insert a row of values (legacy alias that treats all columns as properties).
1091    /// Only the first two values are treated as (from, to) if the table has
1092    /// at least 2 columns; otherwise they are stored as pure properties.
1093    pub fn insert_row(&mut self, values: Vec<Value>) -> Result<u64, StorageError> {
1094        // If there are at least 2 "structural" columns (src_id, dst_id) plus
1095        // property columns, we assume the first two values are the node offsets.
1096        // This preserves backward compatibility with the old flat API.
1097        let num_prop_cols = self.columns.len();
1098        if values.len() != num_prop_cols {
1099            return Err(StorageError::Page(format!(
1100                "Column count mismatch: expected {} values, got {}",
1101                num_prop_cols,
1102                values.len()
1103            )));
1104        }
1105
1106        // We treat the values as plain properties and use sequential edge IDs
1107        // as (from, to) placeholders. Real callers should use `insert_rel`.
1108        let from = self.num_rows;
1109        let to = self.num_rows;
1110        self.insert_rel(from, to, values)?;
1111        Ok(0) // insert_row on RelTable doesn't have a meaningful row offset right now
1112    }
1113
1114    /// Scan the forward adjacency list for a given source node.
1115    ///
1116    /// Returns a list of `(dst_offset, edge_idx)` pairs, or an empty vec
1117    /// if the node has no outgoing edges.
1118    pub fn scan_adj_list(&self, src_offset: u64) -> &[(u64, usize)] {
1119        self.fwd_adj.get(&src_offset).map(|v| v.as_slice()).unwrap_or(&[])
1120    }
1121
1122    /// Scan the reverse adjacency list for a given destination node.
1123    ///
1124    /// Returns a list of `(src_offset, edge_idx)` pairs, or an empty vec
1125    /// if the node has no incoming edges.
1126    pub fn scan_rev_adj_list(&self, dst_offset: u64) -> &[(u64, usize)] {
1127        self.rev_adj.get(&dst_offset).map(|v| v.as_slice()).unwrap_or(&[])
1128    }
1129
1130    /// Get all outgoing edges from a source node as `(dst_offset, property_values)`.
1131    pub fn get_outgoing_edges(&self, src_offset: u64) -> Vec<(u64, Vec<Value>)> {
1132        self.scan_adj_list(src_offset)
1133            .iter()
1134            .map(|&(dst, edge_idx)| {
1135                let props = self.get_edge_properties(edge_idx);
1136                (dst, props)
1137            })
1138            .collect()
1139    }
1140
1141    /// Get all incoming edges to a destination node as `(src_offset, property_values)`.
1142    pub fn get_incoming_edges(&self, dst_offset: u64) -> Vec<(u64, Vec<Value>)> {
1143        self.scan_rev_adj_list(dst_offset)
1144            .iter()
1145            .map(|&(src, edge_idx)| {
1146                let props = self.get_edge_properties(edge_idx);
1147                (src, props)
1148            })
1149            .collect()
1150    }
1151
1152    /// Get the property values for a specific edge by index.
1153    pub fn get_edge_properties(&self, edge_idx: usize) -> Vec<Value> {
1154        let mut props = Vec::with_capacity(self.columns.len());
1155        for col in &self.properties {
1156            match col.get(edge_idx) {
1157                Some(v) => props.push(v.clone()),
1158                None => props.push(Value::Null),
1159            }
1160        }
1161        props
1162    }
1163
1164    /// Get all values for a given property column (by index) as a slice.
1165    pub fn get_column(&self, col_idx: usize) -> Option<&[Value]> {
1166        self.properties.get(col_idx).map(|v| v.as_slice())
1167    }
1168
1169    /// Reconstruct column-major data from properties for backward compatibility.
1170    pub fn to_column_major_data(&self) -> Vec<Vec<Value>> {
1171        self.properties.clone()
1172    }
1173}
1174
1175/// A collection of tables managed by the storage engine.
1176///
1177/// Uses `DashMap` internally for lock-free concurrent reads.
1178/// Write operations synchronize on individual entries rather than
1179/// the entire catalog, allowing concurrent writers to different
1180/// tables to proceed in parallel.
1181#[derive(Debug, Default)]
1182pub struct TableCatalog {
1183    node_tables: DashMap<u64, NodeTable>,
1184    rel_tables: DashMap<u64, RelTable>,
1185    vector_indexes: DashMap<u64, VectorIndexTable>,
1186    /// Map from table name to table ID for node tables.
1187    node_name_to_id: DashMap<String, u64>,
1188    /// Map from table name to table ID for rel tables.
1189    rel_name_to_id: DashMap<String, u64>,
1190    /// Map from index name to index ID for vector indexes.
1191    vector_index_name_to_id: DashMap<String, u64>,
1192    next_table_id: std::sync::atomic::AtomicU64,
1193}
1194
1195impl TableCatalog {
1196    pub fn new() -> Self {
1197        Self::default()
1198    }
1199
1200    pub fn create_node_table(&self, name: String, columns: Vec<ColumnDefinition>) -> NodeTable {
1201        let table_id = self.next_table_id.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
1202        let table = NodeTable::new(table_id, name.clone(), columns);
1203        self.node_name_to_id.insert(name, table_id);
1204        self.node_tables.insert(table_id, table.clone());
1205        table
1206    }
1207
1208    /// Recreate a node table at a specific table ID (used when restoring a
1209    /// persisted catalog during recovery). Advances `next_table_id` so that
1210    /// subsequent auto-assigned IDs never collide with restored ones.
1211    pub fn create_node_table_with_id(&self, table_id: u64, name: String, columns: Vec<ColumnDefinition>) -> NodeTable {
1212        self.bump_next_table_id(table_id);
1213        let table = NodeTable::new(table_id, name.clone(), columns);
1214        self.node_name_to_id.insert(name, table_id);
1215        self.node_tables.insert(table_id, table.clone());
1216        table
1217    }
1218
1219    pub fn create_rel_table(
1220        &self,
1221        name: String,
1222        src_table_id: u64,
1223        dst_table_id: u64,
1224        columns: Vec<ColumnDefinition>,
1225    ) -> RelTable {
1226        let table_id = self.next_table_id.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
1227        let table = RelTable::new(table_id, name.clone(), src_table_id, dst_table_id, columns);
1228        self.rel_name_to_id.insert(name, table_id);
1229        self.rel_tables.insert(table_id, table.clone());
1230        table
1231    }
1232
1233    /// Recreate a rel table at a specific table ID (used when restoring a
1234    /// persisted catalog during recovery). Advances `next_table_id` so that
1235    /// subsequent auto-assigned IDs never collide with restored ones.
1236    pub fn create_rel_table_with_id(
1237        &self,
1238        table_id: u64,
1239        name: String,
1240        src_table_id: u64,
1241        dst_table_id: u64,
1242        columns: Vec<ColumnDefinition>,
1243    ) -> RelTable {
1244        self.bump_next_table_id(table_id);
1245        let table = RelTable::new(table_id, name.clone(), src_table_id, dst_table_id, columns);
1246        self.rel_name_to_id.insert(name, table_id);
1247        self.rel_tables.insert(table_id, table.clone());
1248        table
1249    }
1250
1251    /// Advance `next_table_id` to be strictly greater than `table_id` so
1252    /// restored IDs are never re-issued by `create_node_table`/`create_rel_table`.
1253    fn bump_next_table_id(&self, table_id: u64) {
1254        let mut next = self.next_table_id.load(std::sync::atomic::Ordering::SeqCst);
1255        while next <= table_id {
1256            match self.next_table_id.compare_exchange(
1257                next,
1258                table_id + 1,
1259                std::sync::atomic::Ordering::SeqCst,
1260                std::sync::atomic::Ordering::SeqCst,
1261            ) {
1262                Ok(_) => break,
1263                Err(current) => next = current,
1264            }
1265        }
1266    }
1267
1268    pub fn get_node_table(&self, table_id: u64) -> Option<dashmap::mapref::one::Ref<'_, u64, NodeTable>> {
1269        self.node_tables.get(&table_id)
1270    }
1271
1272    pub fn get_node_table_mut(&self, table_id: u64) -> Option<dashmap::mapref::one::RefMut<'_, u64, NodeTable>> {
1273        self.node_tables.get_mut(&table_id)
1274    }
1275
1276    pub fn get_node_table_by_name(&self, name: &str) -> Option<dashmap::mapref::one::Ref<'_, u64, NodeTable>> {
1277        let id = self.node_name_to_id.get(name)?;
1278        self.node_tables.get(&*id)
1279    }
1280
1281    pub fn get_node_table_by_name_mut(&self, name: &str) -> Option<dashmap::mapref::one::RefMut<'_, u64, NodeTable>> {
1282        let id = self.node_name_to_id.get(name)?;
1283        self.node_tables.get_mut(&*id)
1284    }
1285
1286    pub fn get_rel_table(&self, table_id: u64) -> Option<dashmap::mapref::one::Ref<'_, u64, RelTable>> {
1287        self.rel_tables.get(&table_id)
1288    }
1289
1290    pub fn get_rel_table_mut(&self, table_id: u64) -> Option<dashmap::mapref::one::RefMut<'_, u64, RelTable>> {
1291        self.rel_tables.get_mut(&table_id)
1292    }
1293
1294    pub fn get_rel_table_by_name(&self, name: &str) -> Option<dashmap::mapref::one::Ref<'_, u64, RelTable>> {
1295        let id = self.rel_name_to_id.get(name)?;
1296        self.rel_tables.get(&*id)
1297    }
1298
1299    pub fn get_rel_table_by_name_mut(&self, name: &str) -> Option<dashmap::mapref::one::RefMut<'_, u64, RelTable>> {
1300        let id = self.rel_name_to_id.get(name)?;
1301        self.rel_tables.get_mut(&*id)
1302    }
1303
1304    /// Check if a node has any incident edges.
1305    pub fn has_incident_edges(&self, table_id: u64, node_idx: u64) -> bool {
1306        for rel_table in self.rel_tables.iter() {
1307            if rel_table.src_table_id == table_id {
1308                if let Some(edges) = rel_table.fwd_adj.get(&node_idx) {
1309                    if !edges.is_empty() {
1310                        return true;
1311                    }
1312                }
1313            }
1314            if rel_table.dst_table_id == table_id {
1315                if let Some(edges) = rel_table.rev_adj.get(&node_idx) {
1316                    if !edges.is_empty() {
1317                        return true;
1318                    }
1319                }
1320            }
1321        }
1322        false
1323    }
1324
1325    /// Delete all incident edges for a given node.
1326    pub fn detach_node(&self, table_id: u64, node_idx: u64) {
1327        for mut rel_table in self.rel_tables.iter_mut() {
1328            let mut edges_to_delete = Vec::new();
1329
1330            if rel_table.src_table_id == table_id {
1331                if let Some(edges) = rel_table.fwd_adj.get(&node_idx) {
1332                    for &(_, edge_idx) in edges {
1333                        edges_to_delete.push(edge_idx);
1334                    }
1335                }
1336            }
1337
1338            if rel_table.dst_table_id == table_id {
1339                if let Some(edges) = rel_table.rev_adj.get(&node_idx) {
1340                    for &(_, edge_idx) in edges {
1341                        edges_to_delete.push(edge_idx);
1342                    }
1343                }
1344            }
1345
1346            for edge_idx in edges_to_delete {
1347                let _ = rel_table.delete_edge(edge_idx);
1348            }
1349        }
1350    }
1351
1352    pub fn all_node_tables(&self) -> Vec<dashmap::mapref::multiple::RefMulti<'_, u64, NodeTable>> {
1353        self.node_tables.iter().collect()
1354    }
1355
1356    pub fn all_rel_tables(&self) -> Vec<dashmap::mapref::multiple::RefMulti<'_, u64, RelTable>> {
1357        self.rel_tables.iter().collect()
1358    }
1359
1360    /// Get the number of rows in a node table by name.
1361    pub fn node_table_num_rows(&self, name: &str) -> u64 {
1362        self.get_node_table_by_name(name).map(|t| t.num_rows).unwrap_or(0)
1363    }
1364
1365    /// Remove a node table by name. Returns true if the table existed.
1366    pub fn drop_node_table(&self, name: &str) -> bool {
1367        if let Some(id) = self.node_name_to_id.get(name) {
1368            let table_id = *id;
1369            drop(id);
1370            self.node_name_to_id.remove(name);
1371            self.node_tables.remove(&table_id).is_some()
1372        } else {
1373            false
1374        }
1375    }
1376
1377    /// Create a new vector index in the catalog.
1378    pub fn create_vector_index(
1379        &self,
1380        name: String,
1381        table_name: String,
1382        column_name: String,
1383        metric: DistanceMetric,
1384        dimensions: u32,
1385    ) -> VectorIndexTable {
1386        let index_id = self.next_table_id.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
1387        let table = VectorIndexTable::new(index_id, name.clone(), table_name, column_name, metric, dimensions);
1388        self.vector_index_name_to_id.insert(name, index_id);
1389        self.vector_indexes.insert(index_id, table.clone());
1390        table
1391    }
1392
1393    /// Get a vector index by its ID.
1394    pub fn get_vector_index(&self, index_id: u64) -> Option<dashmap::mapref::one::Ref<'_, u64, VectorIndexTable>> {
1395        self.vector_indexes.get(&index_id)
1396    }
1397
1398    /// Get a vector index by name.
1399    pub fn get_vector_index_by_name(&self, name: &str) -> Option<dashmap::mapref::one::Ref<'_, u64, VectorIndexTable>> {
1400        let id = self.vector_index_name_to_id.get(name)?;
1401        self.vector_indexes.get(&*id)
1402    }
1403
1404    /// Get a mutable vector index by name.
1405    pub fn get_vector_index_by_name_mut(
1406        &self,
1407        name: &str,
1408    ) -> Option<dashmap::mapref::one::RefMut<'_, u64, VectorIndexTable>> {
1409        let id = self.vector_index_name_to_id.get(name)?;
1410        self.vector_indexes.get_mut(&*id)
1411    }
1412
1413    /// Get a mutable vector index by ID.
1414    pub fn get_vector_index_mut(
1415        &self,
1416        index_id: u64,
1417    ) -> Option<dashmap::mapref::one::RefMut<'_, u64, VectorIndexTable>> {
1418        self.vector_indexes.get_mut(&index_id)
1419    }
1420
1421    /// Remove a vector index by name. Returns true if the index existed.
1422    pub fn drop_vector_index(&self, name: &str) -> bool {
1423        if let Some(id) = self.vector_index_name_to_id.get(name) {
1424            let index_id = *id;
1425            drop(id);
1426            self.vector_index_name_to_id.remove(name);
1427            self.vector_indexes.remove(&index_id).is_some()
1428        } else {
1429            false
1430        }
1431    }
1432
1433    /// Get all vector indexes.
1434    pub fn all_vector_indexes(&self) -> Vec<dashmap::mapref::multiple::RefMulti<'_, u64, VectorIndexTable>> {
1435        self.vector_indexes.iter().collect()
1436    }
1437
1438    /// Rebuild a vector index from the current contents of its node table.
1439    ///
1440    /// The HNSW graph is re-populated with the live row ids and vectors, so
1441    /// INSERT/DELETE after `CREATE VECTOR INDEX` are always reflected (P52.38).
1442    pub fn refresh_vector_index(&self, index_id: u64) {
1443        let (table_name, column_name) = match self.vector_indexes.get(&index_id) {
1444            Some(vi) => (vi.table_name.clone(), vi.column_name.clone()),
1445            None => return,
1446        };
1447
1448        let col_idx = match self.get_node_table_by_name(&table_name) {
1449            Some(t) => match t.columns.iter().position(|c| c.name == column_name) {
1450                Some(idx) => idx,
1451                None => return,
1452            },
1453            None => return,
1454        };
1455
1456        // Collect (row_id, vector) pairs while holding the read reference,
1457        // then re-insert under a single mutable borrow of the index.
1458        let mut data: Vec<(usize, Vec<f64>)> = Vec::new();
1459        if let Some(table) = self.get_node_table_by_name(&table_name) {
1460            for row_id in 0..table.num_rows as usize {
1461                if let Some(val) = table.get_value(row_id, col_idx) {
1462                    if let Ok(vec) = crate::extract_f64_list_from_value(val) {
1463                        data.push((row_id, vec));
1464                    }
1465                }
1466            }
1467        }
1468        if data.is_empty() {
1469            if let Some(mut vi) = self.vector_indexes.get_mut(&index_id) {
1470                vi.hnsw_mut().clear();
1471            }
1472            return;
1473        }
1474
1475        let mut vi = self.vector_indexes.get_mut(&index_id);
1476        if let Some(vi) = vi.as_mut() {
1477            vi.hnsw_mut().clear();
1478            for (row_id, vec) in data {
1479                vi.hnsw_mut().insert(vec, row_id);
1480            }
1481        }
1482    }
1483
1484    /// Rebuild the vector indexes of all node tables written by a statement.
1485    ///
1486    /// Called after successful writes so the on-disk/in-memory HNSW graph never
1487    /// serves stale or wrongly-positioned rows (P52.38).
1488    pub fn refresh_vector_indexes_for_tables(&self, table_ids: &[u64]) {
1489        for table_id in table_ids {
1490            let table_name = match self.get_node_table(*table_id) {
1491                Some(t) => t.name.clone(),
1492                None => continue,
1493            };
1494            let index_ids: Vec<u64> = self
1495                .vector_indexes
1496                .iter()
1497                .filter(|vi| vi.table_name == table_name)
1498                .map(|vi| *vi.key())
1499                .collect();
1500            for index_id in index_ids {
1501                self.refresh_vector_index(index_id);
1502            }
1503        }
1504    }
1505
1506    /// Create an ART (Adaptive Radix Tree) index on a node table's PK column.
1507    ///
1508    /// Creates a new `ArtPrimaryKeyIndex`, backfills it with all existing rows,
1509    /// and attaches it to the `NodeTable`.
1510    ///
1511    /// The `index_name` is used as the BufferManager file name for persistence.
1512    pub fn create_art_index(&self, table_name: &str, index_name: &str) -> Result<(), StorageError> {
1513        let mut table = self
1514            .get_node_table_by_name_mut(table_name)
1515            .ok_or_else(|| StorageError::TableNotFound(format!("Node table '{table_name}' not found")))?;
1516
1517        if table.art_index.is_some() {
1518            return Err(StorageError::Index(format!(
1519                "Table '{table_name}' already has an ART index"
1520            )));
1521        }
1522
1523        let mut art_idx = ArtPrimaryKeyIndex::new(index_name);
1524
1525        // Backfill existing rows
1526        let pk_col = table.primary_key_column;
1527        // Scan all rows via to_column_major_data for backfill
1528        let col_major = table.to_column_major_data();
1529        if pk_col < col_major.len() {
1530            for (row_offset, pk_val) in col_major[pk_col].iter().enumerate() {
1531                if !matches!(pk_val, Value::Null)
1532                    && let Some(art_key) = ArtKey::from_value(pk_val)
1533                {
1534                    art_idx.insert(&art_key, row_offset as u64);
1535                }
1536            }
1537        }
1538
1539        table.art_index = Some(art_idx);
1540        Ok(())
1541    }
1542
1543    /// Drop the ART index from a node table.
1544    pub fn drop_art_index(&self, table_name: &str) -> Result<(), StorageError> {
1545        let mut table = self
1546            .get_node_table_by_name_mut(table_name)
1547            .ok_or_else(|| StorageError::TableNotFound(format!("Node table '{table_name}' not found")))?;
1548
1549        table.art_index = None;
1550        Ok(())
1551    }
1552
1553    /// Get a reference to the ART index for a node table (via table name).
1554    /// Returns `None` if the table has no ART index or doesn't exist.
1555    pub fn get_art_index(&self, table_name: &str) -> Option<ArtPrimaryKeyIndex> {
1556        let table = self.get_node_table_by_name(table_name)?;
1557        table.art_index.clone()
1558    }
1559
1560    /// Check if a node table has an ART index.
1561    pub fn has_art_index(&self, table_name: &str) -> bool {
1562        self.get_node_table_by_name(table_name)
1563            .map(|t| t.art_index.is_some())
1564            .unwrap_or(false)
1565    }
1566
1567    /// Remove a rel table by name. Returns true if the table existed.
1568    pub fn drop_rel_table(&self, name: &str) -> bool {
1569        if let Some(id) = self.rel_name_to_id.get(name) {
1570            let table_id = *id;
1571            drop(id);
1572            self.rel_name_to_id.remove(name);
1573            self.rel_tables.remove(&table_id).is_some()
1574        } else {
1575            false
1576        }
1577    }
1578}
1579
1580// ---------------------------------------------------------------------------
1581// Tests
1582// ---------------------------------------------------------------------------
1583
1584#[cfg(test)]
1585mod tests {
1586    use super::*;
1587    use crate::column_chunk::NODE_GROUP_SIZE;
1588
1589    // ==================== NodeTable tests ====================
1590
1591    #[test]
1592    fn test_node_table_empty() {
1593        let table = NodeTable::new(
1594            1,
1595            "Person".into(),
1596            vec![
1597                ColumnDefinition {
1598                    compression: akar_common::enums::CompressionType::Uncompressed,
1599                    name: "name".into(),
1600                    logical_type: LogicalTypeID::String,
1601                    is_primary_key: true,
1602                },
1603                ColumnDefinition {
1604                    compression: akar_common::enums::CompressionType::Uncompressed,
1605                    name: "age".into(),
1606                    logical_type: LogicalTypeID::Int64,
1607                    is_primary_key: false,
1608                },
1609            ],
1610        );
1611        assert_eq!(table.num_rows, 0);
1612        assert!(table.node_groups.is_empty());
1613    }
1614
1615    #[test]
1616    fn test_node_table_insert_and_get() {
1617        let mut table = NodeTable::new(
1618            1,
1619            "Person".into(),
1620            vec![
1621                ColumnDefinition {
1622                    compression: akar_common::enums::CompressionType::Uncompressed,
1623                    name: "name".into(),
1624                    logical_type: LogicalTypeID::String,
1625                    is_primary_key: true,
1626                },
1627                ColumnDefinition {
1628                    compression: akar_common::enums::CompressionType::Uncompressed,
1629                    name: "age".into(),
1630                    logical_type: LogicalTypeID::Int64,
1631                    is_primary_key: false,
1632                },
1633            ],
1634        );
1635        table
1636            .insert_row(vec![Value::String("Alice".into()), Value::Int64(30)])
1637            .unwrap();
1638        table
1639            .insert_row(vec![Value::String("Bob".into()), Value::Int64(25)])
1640            .unwrap();
1641
1642        assert_eq!(table.num_rows, 2);
1643        assert_eq!(table.get_value(0, 0), Some(&Value::String("Alice".into())));
1644        assert_eq!(table.get_value(1, 1), Some(&Value::Int64(25)));
1645    }
1646
1647    #[test]
1648    fn test_delete_row_removes_pk_from_hash_and_art_index() {
1649        let mut table = NodeTable::new(
1650            1,
1651            "Person".into(),
1652            vec![
1653                ColumnDefinition {
1654                    compression: akar_common::enums::CompressionType::Uncompressed,
1655                    name: "name".into(),
1656                    logical_type: LogicalTypeID::String,
1657                    is_primary_key: true,
1658                },
1659                ColumnDefinition {
1660                    compression: akar_common::enums::CompressionType::Uncompressed,
1661                    name: "age".into(),
1662                    logical_type: LogicalTypeID::Int64,
1663                    is_primary_key: false,
1664                },
1665            ],
1666        );
1667        table.art_index = Some(ArtPrimaryKeyIndex::new("test_art"));
1668        table
1669            .insert_row(vec![Value::String("Alice".into()), Value::Int64(30)])
1670            .unwrap();
1671        table
1672            .insert_row(vec![Value::String("Bob".into()), Value::Int64(25)])
1673            .unwrap();
1674        let art = table.art_index.as_ref().unwrap();
1675        assert_eq!(
1676            art.lookup(&ArtKey::from_value(&Value::String("Alice".into())).unwrap()),
1677            Some(0)
1678        );
1679
1680        table.delete_row(0).unwrap();
1681
1682        // Soft-deleted row must no longer resolve via PK lookup (P52.16).
1683        assert!(table.lookup_by_pk(&Value::String("Alice".into())).is_none());
1684        let art = table.art_index.as_ref().unwrap();
1685        assert_eq!(art.len(), 1, "ART must drop the deleted entry");
1686        assert!(
1687            art.lookup(&ArtKey::from_value(&Value::String("Alice".into())).unwrap())
1688                .is_none()
1689        );
1690        assert!(
1691            art.lookup(&ArtKey::from_value(&Value::String("Bob".into())).unwrap())
1692                .is_some()
1693        );
1694        // Range scan over the ART must not surface the deleted PK.
1695        let hits = table.lookup_by_pk_range(
1696            Some(&Value::String("A".into())),
1697            true,
1698            Some(&Value::String("C".into())),
1699            true,
1700            100,
1701        );
1702        assert_eq!(hits, vec![1], "only 'Bob' (row 1) should be in range");
1703
1704        // Re-inserting the same PK must now succeed (was: duplicate PK error).
1705        table
1706            .insert_row(vec![Value::String("Alice".into()), Value::Int64(31)])
1707            .unwrap();
1708        assert_eq!(table.lookup_by_pk(&Value::String("Alice".into())), Some(2));
1709    }
1710
1711    #[test]
1712    fn test_node_table_scan_column() {
1713        let mut table = NodeTable::new(
1714            1,
1715            "T".into(),
1716            vec![ColumnDefinition {
1717                compression: akar_common::enums::CompressionType::Uncompressed,
1718                name: "val".into(),
1719                logical_type: LogicalTypeID::Int64,
1720                is_primary_key: false,
1721            }],
1722        );
1723        for i in 0..100 {
1724            table.insert_row(vec![Value::Int64(i)]).unwrap();
1725        }
1726        let scanned = table.scan_column(0, 10, 5, None, &[]);
1727        assert_eq!(scanned.len(), 5);
1728        assert_eq!(scanned[0], Value::Int64(10));
1729        assert_eq!(scanned[4], Value::Int64(14));
1730    }
1731
1732    #[test]
1733    fn test_node_table_to_column_major() {
1734        let mut table = NodeTable::new(
1735            1,
1736            "T".into(),
1737            vec![
1738                ColumnDefinition {
1739                    compression: akar_common::enums::CompressionType::Uncompressed,
1740                    name: "x".into(),
1741                    logical_type: LogicalTypeID::Int64,
1742                    is_primary_key: false,
1743                },
1744                ColumnDefinition {
1745                    compression: akar_common::enums::CompressionType::Uncompressed,
1746                    name: "y".into(),
1747                    logical_type: LogicalTypeID::Int64,
1748                    is_primary_key: false,
1749                },
1750            ],
1751        );
1752        table.insert_row(vec![Value::Int64(1), Value::Int64(10)]).unwrap();
1753        table.insert_row(vec![Value::Int64(2), Value::Int64(20)]).unwrap();
1754
1755        let data = table.to_column_major_data();
1756        assert_eq!(data.len(), 2);
1757        assert_eq!(data[0], vec![Value::Int64(1), Value::Int64(2)]);
1758        assert_eq!(data[1], vec![Value::Int64(10), Value::Int64(20)]);
1759    }
1760
1761    #[test]
1762    fn test_node_table_auto_node_group() {
1763        let mut table = NodeTable::new(
1764            1,
1765            "T".into(),
1766            vec![ColumnDefinition {
1767                compression: akar_common::enums::CompressionType::Uncompressed,
1768                name: "v".into(),
1769                logical_type: LogicalTypeID::Int64,
1770                is_primary_key: false,
1771            }],
1772        );
1773        // Insert NODE_GROUP_SIZE + 1 rows to force a second node group
1774        for i in 0..NODE_GROUP_SIZE as u64 + 1 {
1775            table.insert_row(vec![Value::Int64(i as i64)]).unwrap();
1776        }
1777        assert_eq!(table.num_rows, NODE_GROUP_SIZE as u64 + 1);
1778        assert_eq!(table.node_groups.len(), 2);
1779        assert_eq!(table.node_groups[0].num_nodes, NODE_GROUP_SIZE as u64);
1780        assert_eq!(table.node_groups[1].num_nodes, 1);
1781        // Scan should still return all values
1782        assert_eq!(table.get_value(0, 0), Some(&Value::Int64(0)));
1783        assert_eq!(
1784            table.get_value(NODE_GROUP_SIZE, 0),
1785            Some(&Value::Int64(NODE_GROUP_SIZE as i64))
1786        );
1787    }
1788
1789    // ==================== RelTable (CSR) tests ====================
1790
1791    fn make_rel_table() -> RelTable {
1792        RelTable::new(
1793            1,
1794            "Knows".into(),
1795            0,
1796            1,
1797            vec![
1798                ColumnDefinition {
1799                    compression: akar_common::enums::CompressionType::Uncompressed,
1800                    name: "since".into(),
1801                    logical_type: LogicalTypeID::Int64,
1802                    is_primary_key: false,
1803                },
1804                ColumnDefinition {
1805                    compression: akar_common::enums::CompressionType::Uncompressed,
1806                    name: "weight".into(),
1807                    logical_type: LogicalTypeID::Double,
1808                    is_primary_key: false,
1809                },
1810            ],
1811        )
1812    }
1813
1814    #[test]
1815    fn test_rel_table_empty() {
1816        let rel = make_rel_table();
1817        assert_eq!(rel.num_rows, 0);
1818        assert!(rel.edges.is_empty());
1819        assert!(rel.fwd_adj.is_empty());
1820        assert!(rel.rev_adj.is_empty());
1821    }
1822
1823    #[test]
1824    fn test_rel_insert_basic() {
1825        let mut rel = make_rel_table();
1826        rel.insert_rel(0, 1, vec![Value::Int64(2020), Value::Double(0.5)])
1827            .unwrap();
1828        rel.insert_rel(0, 2, vec![Value::Int64(2021), Value::Double(0.8)])
1829            .unwrap();
1830        rel.insert_rel(1, 0, vec![Value::Int64(2020), Value::Double(0.3)])
1831            .unwrap();
1832
1833        assert_eq!(rel.num_rows, 3);
1834        assert_eq!(rel.edges.len(), 3);
1835
1836        // Forward adjacency from node 0
1837        let fwd = rel.scan_adj_list(0);
1838        assert_eq!(fwd.len(), 2);
1839        assert_eq!(fwd[0], (1, 0)); // (dst=1, edge_idx=0)
1840        assert_eq!(fwd[1], (2, 1)); // (dst=2, edge_idx=1)
1841
1842        // Forward from node 1
1843        let fwd1 = rel.scan_adj_list(1);
1844        assert_eq!(fwd1.len(), 1);
1845        assert_eq!(fwd1[0], (0, 2));
1846    }
1847
1848    #[test]
1849    fn test_rel_reverse_adjacency() {
1850        let mut rel = make_rel_table();
1851        rel.insert_rel(0, 5, vec![Value::Int64(2022), Value::Double(1.0)])
1852            .unwrap();
1853        rel.insert_rel(3, 5, vec![Value::Int64(2023), Value::Double(1.5)])
1854            .unwrap();
1855
1856        // Node 5 has two incoming edges
1857        let rev = rel.scan_rev_adj_list(5);
1858        assert_eq!(rev.len(), 2);
1859        assert_eq!(rev[0], (0, 0));
1860        assert_eq!(rev[1], (3, 1));
1861    }
1862
1863    #[test]
1864    fn test_rel_get_edge_properties() {
1865        let mut rel = make_rel_table();
1866        rel.insert_rel(0, 1, vec![Value::Int64(2020), Value::Double(0.5)])
1867            .unwrap();
1868        rel.insert_rel(2, 3, vec![Value::Int64(2021), Value::Double(0.9)])
1869            .unwrap();
1870
1871        let props0 = rel.get_edge_properties(0);
1872        assert_eq!(props0, vec![Value::Int64(2020), Value::Double(0.5)]);
1873
1874        let props1 = rel.get_edge_properties(1);
1875        assert_eq!(props1, vec![Value::Int64(2021), Value::Double(0.9)]);
1876    }
1877
1878    #[test]
1879    fn test_rel_get_outgoing_edges() {
1880        let mut rel = make_rel_table();
1881        rel.insert_rel(0, 10, vec![Value::Int64(2020), Value::Double(1.0)])
1882            .unwrap();
1883        rel.insert_rel(0, 20, vec![Value::Int64(2021), Value::Double(2.0)])
1884            .unwrap();
1885
1886        let outgoing = rel.get_outgoing_edges(0);
1887        assert_eq!(outgoing.len(), 2);
1888        assert_eq!(outgoing[0].0, 10);
1889        assert_eq!(outgoing[0].1, vec![Value::Int64(2020), Value::Double(1.0)]);
1890        assert_eq!(outgoing[1].0, 20);
1891    }
1892
1893    #[test]
1894    fn test_rel_get_incoming_edges() {
1895        let mut rel = make_rel_table();
1896        rel.insert_rel(10, 5, vec![Value::Int64(2020), Value::Double(1.0)])
1897            .unwrap();
1898        rel.insert_rel(20, 5, vec![Value::Int64(2021), Value::Double(2.0)])
1899            .unwrap();
1900
1901        let incoming = rel.get_incoming_edges(5);
1902        assert_eq!(incoming.len(), 2);
1903        assert_eq!(incoming[0].0, 10);
1904        assert_eq!(incoming[1].0, 20);
1905    }
1906
1907    #[test]
1908    fn test_rel_no_edges() {
1909        let rel = make_rel_table();
1910        assert!(rel.scan_adj_list(0).is_empty());
1911        assert!(rel.scan_rev_adj_list(0).is_empty());
1912        assert!(rel.get_outgoing_edges(0).is_empty());
1913        assert!(rel.get_incoming_edges(0).is_empty());
1914    }
1915
1916    #[test]
1917    fn test_rel_insert_row_legacy() {
1918        let mut rel = make_rel_table();
1919        // insert_row treats values as properties with sequential edge IDs
1920        rel.insert_row(vec![Value::Int64(2022), Value::Double(3.0)]).unwrap();
1921        assert_eq!(rel.num_rows, 1);
1922        assert_eq!(rel.edges[0], (0, 0)); // sequential from=0, to=0
1923        assert_eq!(rel.get_edge_properties(0), vec![Value::Int64(2022), Value::Double(3.0)]);
1924    }
1925
1926    #[test]
1927    fn test_rel_wrong_column_count() {
1928        let mut rel = make_rel_table();
1929        let result = rel.insert_rel(0, 1, vec![Value::Int64(42)]); // 1 value, expected 2
1930        assert!(result.is_err());
1931    }
1932
1933    #[test]
1934    fn test_rel_get_column() {
1935        let mut rel = make_rel_table();
1936        rel.insert_rel(0, 1, vec![Value::Int64(2020), Value::Double(1.5)])
1937            .unwrap();
1938        rel.insert_rel(1, 2, vec![Value::Int64(2021), Value::Double(2.5)])
1939            .unwrap();
1940
1941        let since_col = rel.get_column(0).unwrap();
1942        assert_eq!(since_col, &[Value::Int64(2020), Value::Int64(2021)]);
1943
1944        let weight_col = rel.get_column(1).unwrap();
1945        assert_eq!(weight_col, &[Value::Double(1.5), Value::Double(2.5)]);
1946    }
1947
1948    #[test]
1949    fn test_rel_to_column_major() {
1950        let mut rel = make_rel_table();
1951        rel.insert_rel(0, 1, vec![Value::Int64(2020), Value::Double(0.5)])
1952            .unwrap();
1953        rel.insert_rel(2, 3, vec![Value::Int64(2021), Value::Double(0.9)])
1954            .unwrap();
1955
1956        let data = rel.to_column_major_data();
1957        assert_eq!(data.len(), 2);
1958        assert_eq!(data[0], vec![Value::Int64(2020), Value::Int64(2021)]);
1959        assert_eq!(data[1], vec![Value::Double(0.5), Value::Double(0.9)]);
1960    }
1961
1962    // ==================== TableCatalog tests ====================
1963
1964    #[test]
1965    fn test_catalog_create_and_lookup() {
1966        let cat = TableCatalog::new();
1967        let node_table = cat.create_node_table(
1968            "Person".into(),
1969            vec![ColumnDefinition {
1970                compression: akar_common::enums::CompressionType::Uncompressed,
1971                name: "id".into(),
1972                logical_type: LogicalTypeID::Int64,
1973                is_primary_key: true,
1974            }],
1975        );
1976        assert_eq!(node_table.table_id, 0);
1977
1978        let rel_table = cat.create_rel_table(
1979            "Knows".into(),
1980            0,
1981            1,
1982            vec![ColumnDefinition {
1983                compression: akar_common::enums::CompressionType::Uncompressed,
1984                name: "since".into(),
1985                logical_type: LogicalTypeID::Int64,
1986                is_primary_key: false,
1987            }],
1988        );
1989        assert_eq!(rel_table.table_id, 1);
1990
1991        assert!(cat.get_node_table(0).is_some());
1992        assert!(cat.get_rel_table(1).is_some());
1993        assert_eq!(cat.node_table_num_rows("Person"), 0);
1994    }
1995
1996    #[test]
1997    fn test_refresh_vector_index_after_dml() {
1998        // P52.38: the HNSW graph used to be populated only during CREATE VECTOR
1999        // INDEX; rows inserted later went un-indexed. refresh_vector_indexes_for_tables
2000        // must rebuild the graph from the live table so post-index DML is visible.
2001        let cat = TableCatalog::new();
2002        cat.create_node_table(
2003            "Item".into(),
2004            vec![
2005                ColumnDefinition {
2006                    compression: akar_common::enums::CompressionType::Uncompressed,
2007                    name: "id".into(),
2008                    logical_type: LogicalTypeID::Int64,
2009                    is_primary_key: true,
2010                },
2011                ColumnDefinition {
2012                    compression: akar_common::enums::CompressionType::Uncompressed,
2013                    name: "embedding".into(),
2014                    logical_type: LogicalTypeID::List,
2015                    is_primary_key: false,
2016                },
2017            ],
2018        );
2019
2020        let vec3 = |x: f64, y: f64, z: f64| Value::List(vec![Value::Double(x), Value::Double(y), Value::Double(z)]);
2021        {
2022            let mut t = cat.get_node_table_by_name_mut("Item").unwrap();
2023            t.insert_row(vec![Value::Int64(1), vec3(1.0, 0.0, 0.0)]).unwrap();
2024            t.insert_row(vec![Value::Int64(2), vec3(0.0, 1.0, 0.0)]).unwrap();
2025            t.insert_row(vec![Value::Int64(3), vec3(0.0, 0.0, 1.0)]).unwrap();
2026        }
2027
2028        cat.create_vector_index(
2029            "item_vec".into(),
2030            "Item".into(),
2031            "embedding".into(),
2032            DistanceMetric::Cosine,
2033            3,
2034        );
2035
2036        // Seed the index from the 3 pre-existing rows; the vector for row 2 must exist.
2037        cat.refresh_vector_indexes_for_tables(&[0]);
2038        {
2039            let vi = cat.get_vector_index(1).unwrap();
2040            assert_eq!(vi.hnsw().len(), 3);
2041            assert!(vi.hnsw().get_vector(2).is_some());
2042        }
2043
2044        // Rows inserted AFTER the index was created must become searchable.
2045        {
2046            let mut t = cat.get_node_table_by_name_mut("Item").unwrap();
2047            t.insert_row(vec![Value::Int64(4), vec3(1.0, 1.0, 0.0)]).unwrap();
2048            t.insert_row(vec![Value::Int64(5), vec3(0.0, 1.0, 1.0)]).unwrap();
2049        }
2050
2051        cat.refresh_vector_indexes_for_tables(&[0]);
2052        let vi = cat.get_vector_index(1).unwrap();
2053        assert_eq!(vi.hnsw().len(), 5);
2054        assert!(vi.hnsw().get_vector(4).is_some());
2055    }
2056}