velesdb_core/column_store/
mod.rs

1//! Column-oriented storage for high-performance metadata filtering.
2//!
3//! This module provides a columnar storage format for frequently filtered fields,
4//! avoiding the overhead of JSON parsing during filter operations.
5//!
6//! # Performance Goals
7//!
8//! - Maintain 50M+ items/sec filter throughput at 100k items (vs 19M/s with
9//!   JSON) — measured by the `column_filter_benchmark` micro-benchmark of
10//!   this module's filtering API. The `SELECT ... WHERE` query path invokes
11//!   these typed filters through the per-collection payload mirror
12//!   (`collection::payload_mirror`), built adaptively for scan-heavy
13//!   workloads; the `ColumnStore` also backs JOIN execution.
14//! - Cache-friendly sequential memory access
15//! - Support for common filter operations: Eq, Gt, Lt, In, Range
16//!
17//! # Architecture
18//!
19//! ```text
20//! ColumnStore
21//! ├── columns: HashMap<field_name, TypedColumn>
22//! │   ├── "category" -> StringColumn(Vec<Option<StringId>>)
23//! │   ├── "price"    -> IntColumn(Vec<Option<i64>>)
24//! │   └── "rating"   -> FloatColumn(Vec<Option<f64>>)
25//! ```
26
27// Reason: Numeric casts in column store are intentional:
28// - All casts are for columnar data processing and statistics
29// - u64/usize conversions for row indices and bitmap operations
30// - Values bounded by column cardinality and row count
31// - Precision loss acceptable for column statistics
32#![allow(clippy::cast_precision_loss)]
33#![allow(clippy::cast_possible_truncation)]
34#![allow(clippy::doc_markdown)] // Column-store docs include many storage type identifiers.
35
36mod batch;
37#[cfg(test)]
38mod batch_tests;
39mod filter;
40mod filter_array;
41mod filter_geo;
42#[cfg(test)]
43mod filter_tests;
44pub(crate) mod haversine;
45#[cfg(test)]
46mod haversine_tests;
47mod primary_key_ops;
48mod string_table;
49mod types;
50mod vacuum;
51#[cfg(test)]
52mod vacuum_tests;
53
54use roaring::RoaringBitmap;
55use rustc_hash::FxHashMap;
56use std::collections::HashMap;
57
58pub use filter_geo::{CompareOp, GeoBboxParams, GeoDistanceParams};
59pub use string_table::StringTable;
60pub use types::{
61    AutoVacuumConfig, BatchUpdate, BatchUpdateResult, BatchUpsertResult, ColumnStoreError,
62    ColumnType, ColumnValue, ExpireResult, StringId, TypedColumn, UpsertResult, VacuumConfig,
63    VacuumStats,
64};
65
66/// Column store for high-performance filtering.
67#[derive(Debug, Default)]
68pub struct ColumnStore {
69    /// Columns indexed by field name
70    pub(crate) columns: HashMap<String, TypedColumn>,
71    /// String interning table
72    pub(crate) string_table: StringTable,
73    /// Number of rows
74    pub(crate) row_count: usize,
75    /// Primary key column name (if any)
76    pub(crate) primary_key_column: Option<String>,
77    /// Primary key index: pk_value → row_idx (O(1) lookup)
78    pub(crate) primary_index: HashMap<i64, usize>,
79    /// Reverse index: row_idx → pk_value (O(1) reverse lookup for expire_rows)
80    pub(crate) row_idx_to_pk: HashMap<usize, i64>,
81    /// Deleted row indices (tombstones) - FxHashSet for backward compatibility
82    pub(crate) deleted_rows: rustc_hash::FxHashSet<usize>,
83    /// Deleted row bitmap (EPIC-043 US-002) - RoaringBitmap for O(1) contains
84    pub(crate) deletion_bitmap: RoaringBitmap,
85    /// Row expiry timestamps: row_idx → expiry_timestamp (US-004 TTL)
86    pub(crate) row_expiry: HashMap<usize, u64>,
87}
88
89impl ColumnStore {
90    /// Creates a new empty column store.
91    #[must_use]
92    pub fn new() -> Self {
93        Self::default()
94    }
95
96    /// Creates a column store with pre-defined indexed fields.
97    #[must_use]
98    pub fn with_schema(fields: &[(&str, ColumnType)]) -> Self {
99        let mut store = Self::new();
100        for (name, col_type) in fields {
101            store.add_column(name, col_type);
102        }
103        store
104    }
105
106    /// Creates a column store with validated schema (rejects nested arrays).
107    ///
108    /// # Errors
109    ///
110    /// Returns `ColumnStoreError::TypeMismatch` if any column uses nested arrays.
111    pub fn with_schema_validated(fields: &[(&str, ColumnType)]) -> Result<Self, ColumnStoreError> {
112        for (name, col_type) in fields {
113            Self::reject_nested_array(name, col_type)?;
114        }
115        Ok(Self::with_schema(fields))
116    }
117
118    /// Creates a column store with a primary key for O(1) lookups.
119    ///
120    /// # Errors
121    ///
122    /// Returns `Error::ColumnStoreError` if `pk_column` is not found in `fields`
123    /// or is not of type `Int`.
124    pub fn with_primary_key(
125        fields: &[(&str, ColumnType)],
126        pk_column: &str,
127    ) -> crate::error::Result<Self> {
128        let pk_field = fields
129            .iter()
130            .find(|(name, _)| *name == pk_column)
131            .ok_or_else(|| {
132                crate::error::Error::ColumnStoreError(format!(
133                    "Primary key column '{}' not found in fields: {:?}",
134                    pk_column,
135                    fields.iter().map(|(n, _)| *n).collect::<Vec<_>>()
136                ))
137            })?;
138        if !matches!(pk_field.1, ColumnType::Int) {
139            return Err(crate::error::Error::ColumnStoreError(format!(
140                "Primary key column '{}' must be Int type, got {:?}",
141                pk_column, pk_field.1
142            )));
143        }
144
145        let mut store = Self::with_schema(fields);
146        store.primary_key_column = Some(pk_column.to_string());
147        store.primary_index = HashMap::new();
148        Ok(store)
149    }
150
151    /// Returns the primary key column name if set.
152    #[must_use]
153    pub fn primary_key_column(&self) -> Option<&str> {
154        self.primary_key_column.as_deref()
155    }
156
157    /// Adds a new column to the store.
158    ///
159    /// # Panics
160    ///
161    /// Does not panic. Nested arrays are silently treated as scalar arrays.
162    /// Use `add_column_validated` for strict schema validation.
163    pub fn add_column(&mut self, name: &str, col_type: &ColumnType) {
164        let column = match col_type {
165            ColumnType::Int => TypedColumn::new_int(0),
166            ColumnType::Float => TypedColumn::new_float(0),
167            ColumnType::String => TypedColumn::new_string(0),
168            ColumnType::Bool => TypedColumn::new_bool(0),
169            ColumnType::Array(inner) => TypedColumn::new_array((**inner).clone(), 0),
170            ColumnType::GeoPoint => TypedColumn::new_geopoint(0),
171        };
172        self.columns.insert(name.to_string(), column);
173    }
174
175    /// Adds a column after rows already exist, backfilling nulls so the new
176    /// column stays aligned with `row_count` (required by the per-collection
177    /// payload mirror, whose schema evolves as new fields appear).
178    pub(crate) fn add_column_backfilled(&mut self, name: &str, col_type: &ColumnType) {
179        self.add_column(name, col_type);
180        if let Some(column) = self.columns.get_mut(name) {
181            for _ in 0..self.row_count {
182                column.push_null();
183            }
184        }
185    }
186
187    /// Tombstones a row by index without requiring the primary-key machinery.
188    ///
189    /// Used by the payload mirror, which maps point ids to row indices
190    /// externally (`u64` ids do not fit the `i64` primary-key API).
191    pub(crate) fn tombstone_row(&mut self, row_idx: usize) {
192        if row_idx >= self.row_count {
193            return;
194        }
195        self.deleted_rows.insert(row_idx);
196        if let Ok(idx) = u32::try_from(row_idx) {
197            self.deletion_bitmap.insert(idx);
198        }
199    }
200
201    /// Returns the total number of rows in the store (including deleted/tombstoned rows).
202    #[must_use]
203    pub fn row_count(&self) -> usize {
204        self.row_count
205    }
206
207    /// Returns the number of active (non-deleted) rows in the store.
208    #[must_use]
209    pub fn active_row_count(&self) -> usize {
210        self.row_count.saturating_sub(self.deleted_rows.len())
211    }
212
213    /// Returns the number of deleted (tombstoned) rows.
214    #[must_use]
215    pub fn deleted_row_count(&self) -> usize {
216        self.deleted_rows.len()
217    }
218
219    /// Returns the string table for string interning.
220    #[must_use]
221    pub fn string_table(&self) -> &StringTable {
222        &self.string_table
223    }
224
225    /// Returns a mutable reference to the string table.
226    pub fn string_table_mut(&mut self) -> &mut StringTable {
227        &mut self.string_table
228    }
229
230    /// Pushes values for a new row (low-level, no validation).
231    pub fn push_row_unchecked(&mut self, values: &[(&str, ColumnValue)]) {
232        let value_map: FxHashMap<&str, &ColumnValue> =
233            values.iter().map(|(k, v)| (*k, v)).collect();
234
235        for (name, column) in &mut self.columns {
236            if let Some(value) = value_map.get(name.as_str()) {
237                column.push_typed(value);
238            } else {
239                column.push_null();
240            }
241        }
242        self.row_count += 1;
243    }
244
245    /// Convenience alias for [`push_row_unchecked()`](Self::push_row_unchecked).
246    #[inline]
247    pub fn push_row(&mut self, values: &[(&str, ColumnValue)]) {
248        self.push_row_unchecked(values);
249    }
250
251    /// Gets a column by name.
252    #[must_use]
253    pub fn get_column(&self, name: &str) -> Option<&TypedColumn> {
254        self.columns.get(name)
255    }
256
257    /// Rejects nested array types (`Array(Array(...))`) at schema creation time.
258    fn reject_nested_array(name: &str, col_type: &ColumnType) -> Result<(), ColumnStoreError> {
259        if let ColumnType::Array(inner) = col_type {
260            if matches!(inner.as_ref(), ColumnType::Array(_)) {
261                return Err(ColumnStoreError::TypeMismatch {
262                    expected: "scalar element type (Int, Float, String, Bool)".to_string(),
263                    actual: format!("nested Array in column '{name}'"),
264                });
265            }
266        }
267        Ok(())
268    }
269
270    /// Returns an iterator over column names.
271    pub fn column_names(&self) -> impl Iterator<Item = &str> {
272        self.columns.keys().map(String::as_str)
273    }
274
275    /// Gets a value from a column at a specific row index as JSON.
276    #[must_use]
277    pub fn get_value_as_json(&self, column: &str, row_idx: usize) -> Option<serde_json::Value> {
278        if self.deleted_rows.contains(&row_idx) {
279            return None;
280        }
281
282        let col = self.columns.get(column)?;
283        // String columns need special handling for intern-table resolution.
284        if let TypedColumn::String(v) = col {
285            return v.get(row_idx).and_then(|opt| {
286                opt.and_then(|id| self.string_table.get(id).map(|s| serde_json::json!(s)))
287            });
288        }
289        // Array columns need special handling for string element resolution.
290        if let TypedColumn::Array { data, .. } = col {
291            return self.get_array_as_json(data, row_idx);
292        }
293        // GeoPoint columns return {"lat": f64, "lng": f64}.
294        if let TypedColumn::GeoPoint(v) = col {
295            return v
296                .get(row_idx)
297                .and_then(|opt| opt.map(|(lat, lng)| serde_json::json!({"lat": lat, "lng": lng})));
298        }
299        col.get_as_json_non_string(row_idx)
300    }
301
302    /// Converts an array column cell to a JSON array, resolving string IDs.
303    fn get_array_as_json(
304        &self,
305        data: &[Option<smallvec::SmallVec<[ColumnValue; 8]>>],
306        row_idx: usize,
307    ) -> Option<serde_json::Value> {
308        let arr = data.get(row_idx)?.as_ref()?;
309        let json_arr: Vec<serde_json::Value> =
310            arr.iter().map(|v| self.column_value_to_json(v)).collect();
311        Some(serde_json::Value::Array(json_arr))
312    }
313
314    /// Converts a single `ColumnValue` to its JSON representation.
315    fn column_value_to_json(&self, value: &ColumnValue) -> serde_json::Value {
316        match value {
317            ColumnValue::Int(v) => serde_json::json!(v),
318            ColumnValue::Float(v) => serde_json::json!(v),
319            ColumnValue::Bool(v) => serde_json::json!(v),
320            ColumnValue::String(id) => self
321                .string_table
322                .get(*id)
323                .map_or(serde_json::Value::Null, |s| serde_json::json!(s)),
324            ColumnValue::Null => serde_json::Value::Null,
325            ColumnValue::Array(inner) => {
326                let arr: Vec<serde_json::Value> =
327                    inner.iter().map(|v| self.column_value_to_json(v)).collect();
328                serde_json::Value::Array(arr)
329            }
330            ColumnValue::GeoPoint(lat, lng) => {
331                serde_json::json!({"lat": lat, "lng": lng})
332            }
333        }
334    }
335}
velesdb_core/column_store/mod.rs

velesdb_core/column_store/
mod.rs
