velesdb-core 1.9.3

//! Column-oriented storage for high-performance metadata filtering.
//!
//! This module provides a columnar storage format for frequently filtered fields,
//! avoiding the overhead of JSON parsing during filter operations.
//!
//! # Performance Goals
//!
//! - Maintain 50M+ items/sec throughput at 100k items (vs 19M/s with JSON)
//! - Cache-friendly sequential memory access
//! - Support for common filter operations: Eq, Gt, Lt, In, Range
//!
//! # Architecture
//!
//! ```text
//! ColumnStore
//! ├── columns: HashMap<field_name, TypedColumn>
//! │   ├── "category" -> StringColumn(Vec<Option<StringId>>)
//! │   ├── "price"    -> IntColumn(Vec<Option<i64>>)
//! │   └── "rating"   -> FloatColumn(Vec<Option<f64>>)
//! ```

// SAFETY: Numeric casts in column store are intentional:
// - All casts are for columnar data processing and statistics
// - u64/usize conversions for row indices and bitmap operations
// - Values bounded by column cardinality and row count
// - Precision loss acceptable for column statistics
#![allow(clippy::cast_precision_loss)]
#![allow(clippy::cast_possible_truncation)]
#![allow(clippy::doc_markdown)] // Column-store docs include many storage type identifiers.

mod batch;
#[cfg(test)]
mod batch_tests;
mod filter;
#[cfg(test)]
mod filter_tests;
mod string_table;
mod types;
mod vacuum;
#[cfg(test)]
mod vacuum_tests;

use roaring::RoaringBitmap;
use rustc_hash::FxHashMap;
use std::collections::HashMap;

pub use string_table::StringTable;
pub use types::{
    AutoVacuumConfig, BatchUpdate, BatchUpdateResult, BatchUpsertResult, ColumnStoreError,
    ColumnType, ColumnValue, ExpireResult, StringId, TypedColumn, UpsertResult, VacuumConfig,
    VacuumStats,
};

/// Column store for high-performance filtering.
#[derive(Debug, Default)]
pub struct ColumnStore {
    /// Columns indexed by field name
    pub(crate) columns: HashMap<String, TypedColumn>,
    /// String interning table
    pub(crate) string_table: StringTable,
    /// Number of rows
    pub(crate) row_count: usize,
    /// Primary key column name (if any)
    pub(crate) primary_key_column: Option<String>,
    /// Primary key index: pk_value → row_idx (O(1) lookup)
    pub(crate) primary_index: HashMap<i64, usize>,
    /// Reverse index: row_idx → pk_value (O(1) reverse lookup for expire_rows)
    pub(crate) row_idx_to_pk: HashMap<usize, i64>,
    /// Deleted row indices (tombstones) - FxHashSet for backward compatibility
    pub(crate) deleted_rows: rustc_hash::FxHashSet<usize>,
    /// Deleted row bitmap (EPIC-043 US-002) - RoaringBitmap for O(1) contains
    pub(crate) deletion_bitmap: RoaringBitmap,
    /// Row expiry timestamps: row_idx → expiry_timestamp (US-004 TTL)
    pub(crate) row_expiry: HashMap<usize, u64>,
}

impl ColumnStore {
    /// Creates a new empty column store.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Creates a column store with pre-defined indexed fields.
    #[must_use]
    pub fn with_schema(fields: &[(&str, ColumnType)]) -> Self {
        let mut store = Self::new();
        for (name, col_type) in fields {
            store.add_column(name, *col_type);
        }
        store
    }

    /// Creates a column store with a primary key for O(1) lookups.
    ///
    /// # Errors
    ///
    /// Returns `Error::ColumnStoreError` if `pk_column` is not found in `fields`
    /// or is not of type `Int`.
    pub fn with_primary_key(
        fields: &[(&str, ColumnType)],
        pk_column: &str,
    ) -> crate::error::Result<Self> {
        let pk_field = fields
            .iter()
            .find(|(name, _)| *name == pk_column)
            .ok_or_else(|| {
                crate::error::Error::ColumnStoreError(format!(
                    "Primary key column '{}' not found in fields: {:?}",
                    pk_column,
                    fields.iter().map(|(n, _)| *n).collect::<Vec<_>>()
                ))
            })?;
        if !matches!(pk_field.1, ColumnType::Int) {
            return Err(crate::error::Error::ColumnStoreError(format!(
                "Primary key column '{}' must be Int type, got {:?}",
                pk_column, pk_field.1
            )));
        }

        let mut store = Self::with_schema(fields);
        store.primary_key_column = Some(pk_column.to_string());
        store.primary_index = HashMap::new();
        Ok(store)
    }

    /// Returns the primary key column name if set.
    #[must_use]
    pub fn primary_key_column(&self) -> Option<&str> {
        self.primary_key_column.as_deref()
    }

    /// Adds a new column to the store.
    pub fn add_column(&mut self, name: &str, col_type: ColumnType) {
        let column = match col_type {
            ColumnType::Int => TypedColumn::new_int(0),
            ColumnType::Float => TypedColumn::new_float(0),
            ColumnType::String => TypedColumn::new_string(0),
            ColumnType::Bool => TypedColumn::new_bool(0),
        };
        self.columns.insert(name.to_string(), column);
    }

    /// Returns the total number of rows in the store (including deleted/tombstoned rows).
    #[must_use]
    pub fn row_count(&self) -> usize {
        self.row_count
    }

    /// Returns the number of active (non-deleted) rows in the store.
    #[must_use]
    pub fn active_row_count(&self) -> usize {
        self.row_count.saturating_sub(self.deleted_rows.len())
    }

    /// Returns the number of deleted (tombstoned) rows.
    #[must_use]
    pub fn deleted_row_count(&self) -> usize {
        self.deleted_rows.len()
    }

    /// Returns the string table for string interning.
    #[must_use]
    pub fn string_table(&self) -> &StringTable {
        &self.string_table
    }

    /// Returns a mutable reference to the string table.
    pub fn string_table_mut(&mut self) -> &mut StringTable {
        &mut self.string_table
    }

    /// Pushes values for a new row (low-level, no validation).
    pub fn push_row_unchecked(&mut self, values: &[(&str, ColumnValue)]) {
        let value_map: FxHashMap<&str, &ColumnValue> =
            values.iter().map(|(k, v)| (*k, v)).collect();

        for (name, column) in &mut self.columns {
            if let Some(value) = value_map.get(name.as_str()) {
                column.push_typed(value);
            } else {
                column.push_null();
            }
        }
        self.row_count += 1;
    }

    /// Convenience alias for [`push_row_unchecked()`](Self::push_row_unchecked).
    #[inline]
    pub fn push_row(&mut self, values: &[(&str, ColumnValue)]) {
        self.push_row_unchecked(values);
    }

    /// Inserts a row with primary key validation and index update.
    ///
    /// # Errors
    ///
    /// Returns an error if the primary key is missing, duplicated, or any
    /// provided value does not match the target column type.
    pub fn insert_row(
        &mut self,
        values: &[(&str, ColumnValue)],
    ) -> Result<usize, ColumnStoreError> {
        let Some(ref pk_col) = self.primary_key_column else {
            self.push_row(values);
            return Ok(self.row_count - 1);
        };

        let pk_value = Self::extract_pk_value(values, pk_col)?;

        if let Some(&existing_idx) = self.primary_index.get(&pk_value) {
            return self.reinsert_or_reject(values, existing_idx, pk_value);
        }

        let row_idx = self.row_count;
        self.push_row(values);
        self.primary_index.insert(pk_value, row_idx);
        self.row_idx_to_pk.insert(row_idx, pk_value);
        Ok(row_idx)
    }

    /// Extracts the integer primary key value from a row's values.
    fn extract_pk_value(
        values: &[(&str, ColumnValue)],
        pk_col: &str,
    ) -> Result<i64, ColumnStoreError> {
        values
            .iter()
            .find(|(name, _)| *name == pk_col)
            .and_then(|(_, value)| {
                if let ColumnValue::Int(v) = value {
                    Some(*v)
                } else {
                    None
                }
            })
            .ok_or(ColumnStoreError::MissingPrimaryKey)
    }

    /// Handles insert into a previously-deleted row slot, or rejects as duplicate.
    fn reinsert_or_reject(
        &mut self,
        values: &[(&str, ColumnValue)],
        existing_idx: usize,
        pk_value: i64,
    ) -> Result<usize, ColumnStoreError> {
        if !self.deleted_rows.contains(&existing_idx) {
            return Err(ColumnStoreError::DuplicateKey(pk_value));
        }

        Self::validate_value_types(&self.columns, values, None)?;
        self.undelete_row(existing_idx);
        self.set_row_values(values, existing_idx, None)?;
        Ok(existing_idx)
    }

    /// Validates that all non-null values match their target column types.
    ///
    /// Optionally skips a column (e.g. primary key). Shared by both `insert_row`
    /// and upsert paths in `batch.rs`.
    pub(super) fn validate_value_types(
        columns: &HashMap<String, TypedColumn>,
        values: &[(&str, ColumnValue)],
        skip_col: Option<&str>,
    ) -> Result<(), ColumnStoreError> {
        for (col_name, value) in values {
            if skip_col.is_some_and(|s| s == *col_name) {
                continue;
            }
            if let Some(col) = columns.get(*col_name) {
                if !matches!(value, ColumnValue::Null) {
                    Self::validate_type_match(col, value)?;
                }
            }
        }
        Ok(())
    }

    /// Marks a tombstoned row as live again.
    fn undelete_row(&mut self, row_idx: usize) {
        self.deleted_rows.remove(&row_idx);
        if let Ok(idx) = u32::try_from(row_idx) {
            self.deletion_bitmap.remove(idx);
        }
        self.row_expiry.remove(&row_idx);
    }

    /// Writes column values for a row, optionally skipping a column (e.g. primary key).
    ///
    /// Missing columns are set to null. Shared by `reinsert_or_reject` and
    /// `write_row_values` (batch module) to eliminate duplication.
    pub(super) fn set_row_values(
        &mut self,
        values: &[(&str, ColumnValue)],
        row_idx: usize,
        skip_col: Option<&str>,
    ) -> Result<(), ColumnStoreError> {
        let value_map: std::collections::HashMap<&str, &ColumnValue> =
            values.iter().map(|(k, v)| (*k, v)).collect();
        let col_names: Vec<String> = self.columns.keys().cloned().collect();
        for col_name in col_names {
            if skip_col.is_some_and(|s| s == col_name) {
                continue;
            }
            if let Some(col) = self.columns.get_mut(&col_name) {
                let val = value_map
                    .get(col_name.as_str())
                    .map_or(ColumnValue::Null, |v| (*v).clone());
                Self::set_column_value(col, row_idx, val)?;
            }
        }
        Ok(())
    }

    /// Gets the row index by primary key value - O(1) lookup.
    #[must_use]
    pub fn get_row_idx_by_pk(&self, pk: i64) -> Option<usize> {
        let row_idx = self.primary_index.get(&pk).copied()?;
        if self.deleted_rows.contains(&row_idx) {
            return None;
        }
        Some(row_idx)
    }

    /// Deletes a row by primary key value.
    ///
    /// Also clears any TTL metadata to prevent false-positive expirations.
    /// Updates both FxHashSet and RoaringBitmap (EPIC-043 US-002).
    pub fn delete_by_pk(&mut self, pk: i64) -> bool {
        let Some(&row_idx) = self.primary_index.get(&pk) else {
            return false;
        };
        if self.deleted_rows.contains(&row_idx) {
            return false;
        }
        self.deleted_rows.insert(row_idx);
        // EPIC-043 US-002: Also update RoaringBitmap for O(1) contains
        if let Ok(idx) = u32::try_from(row_idx) {
            self.deletion_bitmap.insert(idx);
        }
        self.row_expiry.remove(&row_idx);
        true
    }

    /// Updates a single column value for a row identified by primary key - O(1).
    ///
    /// # Errors
    ///
    /// Returns an error if the row does not exist, the column does not exist,
    /// the update targets the primary-key column, or the value type mismatches
    /// the column type.
    pub fn update_by_pk(
        &mut self,
        pk: i64,
        column: &str,
        value: ColumnValue,
    ) -> Result<(), ColumnStoreError> {
        if self
            .primary_key_column
            .as_ref()
            .is_some_and(|pk_col| pk_col == column)
        {
            return Err(ColumnStoreError::PrimaryKeyUpdate);
        }

        let row_idx = self.resolve_live_row(pk)?;

        let col = self
            .columns
            .get_mut(column)
            .ok_or_else(|| ColumnStoreError::ColumnNotFound(column.to_string()))?;

        Self::set_column_value(col, row_idx, value)
    }

    /// Updates multiple columns atomically for a row identified by primary key.
    ///
    /// # Panics
    ///
    /// This function will not panic under normal operation. The internal expect
    /// is guarded by prior validation that all columns exist.
    ///
    /// # Errors
    ///
    /// Returns an error if the row does not exist, one of the columns does not
    /// exist, one update attempts to modify the primary key, or a value type
    /// mismatches its target column type.
    pub fn update_multi_by_pk(
        &mut self,
        pk: i64,
        updates: &[(&str, ColumnValue)],
    ) -> Result<(), ColumnStoreError> {
        let row_idx = self.resolve_live_row(pk)?;
        self.validate_multi_update(updates)?;

        for (col_name, value) in updates {
            let col = self
                .columns
                .get_mut(*col_name)
                .ok_or_else(|| ColumnStoreError::ColumnNotFound((*col_name).to_string()))?;
            Self::set_column_value(col, row_idx, value.clone())?;
        }

        Ok(())
    }

    /// Resolves a primary key to a live (non-deleted) row index.
    fn resolve_live_row(&self, pk: i64) -> Result<usize, ColumnStoreError> {
        let row_idx = *self
            .primary_index
            .get(&pk)
            .ok_or(ColumnStoreError::RowNotFound(pk))?;
        if self.deleted_rows.contains(&row_idx) {
            return Err(ColumnStoreError::RowNotFound(pk));
        }
        Ok(row_idx)
    }

    /// Validates that no update targets the primary key and all types match.
    fn validate_multi_update(
        &self,
        updates: &[(&str, ColumnValue)],
    ) -> Result<(), ColumnStoreError> {
        for (col_name, value) in updates {
            if self
                .primary_key_column
                .as_ref()
                .is_some_and(|pk_col| pk_col == *col_name)
            {
                return Err(ColumnStoreError::PrimaryKeyUpdate);
            }

            let col = self
                .columns
                .get(*col_name)
                .ok_or_else(|| ColumnStoreError::ColumnNotFound((*col_name).to_string()))?;

            if !matches!(value, ColumnValue::Null) {
                Self::validate_type_match(col, value)?;
            }
        }
        Ok(())
    }

    /// Gets a column by name.
    #[must_use]
    pub fn get_column(&self, name: &str) -> Option<&TypedColumn> {
        self.columns.get(name)
    }

    /// Returns an iterator over column names.
    pub fn column_names(&self) -> impl Iterator<Item = &str> {
        self.columns.keys().map(String::as_str)
    }

    /// Gets a value from a column at a specific row index as JSON.
    #[must_use]
    pub fn get_value_as_json(&self, column: &str, row_idx: usize) -> Option<serde_json::Value> {
        if self.deleted_rows.contains(&row_idx) {
            return None;
        }

        let col = self.columns.get(column)?;
        // String columns need special handling for intern-table resolution.
        if let TypedColumn::String(v) = col {
            return v.get(row_idx).and_then(|opt| {
                opt.and_then(|id| self.string_table.get(id).map(|s| serde_json::json!(s)))
            });
        }
        col.get_as_json_non_string(row_idx)
    }
}