Skip to main content

scirs2_io/database/
table.rs

1//! Pure Rust in-memory columnar table with join, group-by, filter, sort, project.
2//!
3//! Provides a DataFrame-like in-memory table that operates entirely in pure Rust
4//! without any external SQL dependencies.
5//!
6//! # Example
7//!
8//! ```rust
9//! use scirs2_io::database::table::{InMemoryTable, ColumnValue, TableFilter, Predicate};
10//!
11//! let mut table = InMemoryTable::new(vec![
12//!     ("id".to_string(),    scirs2_io::database::table::ColumnType::Int64),
13//!     ("name".to_string(),  scirs2_io::database::table::ColumnType::Utf8),
14//!     ("score".to_string(), scirs2_io::database::table::ColumnType::Float64),
15//! ]);
16//!
17//! table.push_row(&[
18//!     ColumnValue::Int(1),
19//!     ColumnValue::Utf8("Alice".to_string()),
20//!     ColumnValue::Float(95.0),
21//! ]).unwrap();
22//!
23//! table.push_row(&[
24//!     ColumnValue::Int(2),
25//!     ColumnValue::Utf8("Bob".to_string()),
26//!     ColumnValue::Float(82.5),
27//! ]).unwrap();
28//!
29//! let filtered = TableFilter::new(&table)
30//!     .predicate(Predicate::Greater("score".to_string(), ColumnValue::Float(90.0)))
31//!     .apply()
32//!     .unwrap();
33//! assert_eq!(filtered.row_count(), 1);
34//! ```
35
36#![allow(missing_docs)]
37
38use crate::error::{IoError, Result};
39use serde::{Deserialize, Serialize};
40use std::cmp::Ordering;
41use std::collections::HashMap;
42
43// ─── Column types and values ─────────────────────────────────────────────────
44
45/// The type of a column in an [`InMemoryTable`].
46#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
47pub enum ColumnType {
48    /// 64-bit signed integer
49    Int64,
50    /// 64-bit floating point
51    Float64,
52    /// UTF-8 string
53    Utf8,
54    /// Boolean
55    Boolean,
56    /// Nullable wrapper
57    Nullable(Box<ColumnType>),
58}
59
60impl ColumnType {
61    /// Return a string identifier.
62    pub fn as_str(&self) -> &str {
63        match self {
64            ColumnType::Int64 => "int64",
65            ColumnType::Float64 => "float64",
66            ColumnType::Utf8 => "utf8",
67            ColumnType::Boolean => "boolean",
68            ColumnType::Nullable(_) => "nullable",
69        }
70    }
71}
72
73/// A single cell value in an [`InMemoryTable`].
74#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
75pub enum ColumnValue {
76    /// NULL / missing
77    Null,
78    /// 64-bit signed integer
79    Int(i64),
80    /// 64-bit floating point
81    Float(f64),
82    /// UTF-8 string
83    Utf8(String),
84    /// Boolean
85    Boolean(bool),
86}
87
88impl ColumnValue {
89    /// Return the column type for this value.
90    pub fn column_type(&self) -> ColumnType {
91        match self {
92            ColumnValue::Null => ColumnType::Nullable(Box::new(ColumnType::Utf8)),
93            ColumnValue::Int(_) => ColumnType::Int64,
94            ColumnValue::Float(_) => ColumnType::Float64,
95            ColumnValue::Utf8(_) => ColumnType::Utf8,
96            ColumnValue::Boolean(_) => ColumnType::Boolean,
97        }
98    }
99
100    /// Try to extract as f64 (for aggregation).
101    pub fn as_f64(&self) -> Option<f64> {
102        match self {
103            ColumnValue::Float(f) => Some(*f),
104            ColumnValue::Int(i) => Some(*i as f64),
105            _ => None,
106        }
107    }
108
109    /// Try to extract as i64.
110    pub fn as_i64(&self) -> Option<i64> {
111        match self {
112            ColumnValue::Int(i) => Some(*i),
113            ColumnValue::Float(f) if f.fract() == 0.0 => Some(*f as i64),
114            _ => None,
115        }
116    }
117
118    /// Partial comparison for ordering.
119    pub fn partial_cmp_value(&self, other: &Self) -> Option<Ordering> {
120        match (self, other) {
121            (ColumnValue::Int(a), ColumnValue::Int(b)) => a.partial_cmp(b),
122            (ColumnValue::Float(a), ColumnValue::Float(b)) => a.partial_cmp(b),
123            (ColumnValue::Int(a), ColumnValue::Float(b)) => (*a as f64).partial_cmp(b),
124            (ColumnValue::Float(a), ColumnValue::Int(b)) => a.partial_cmp(&(*b as f64)),
125            (ColumnValue::Utf8(a), ColumnValue::Utf8(b)) => a.partial_cmp(b),
126            (ColumnValue::Boolean(a), ColumnValue::Boolean(b)) => a.partial_cmp(b),
127            (ColumnValue::Null, ColumnValue::Null) => Some(Ordering::Equal),
128            (ColumnValue::Null, _) => Some(Ordering::Less),
129            (_, ColumnValue::Null) => Some(Ordering::Greater),
130            _ => None,
131        }
132    }
133
134    /// Convert to JSON value.
135    pub fn to_json(&self) -> serde_json::Value {
136        match self {
137            ColumnValue::Null => serde_json::Value::Null,
138            ColumnValue::Int(i) => serde_json::json!(i),
139            ColumnValue::Float(f) => serde_json::json!(f),
140            ColumnValue::Utf8(s) => serde_json::json!(s),
141            ColumnValue::Boolean(b) => serde_json::json!(b),
142        }
143    }
144
145    /// Try to convert from a JSON value.
146    pub fn from_json(v: &serde_json::Value, col_type: &ColumnType) -> Self {
147        match (col_type, v) {
148            (_, serde_json::Value::Null) => ColumnValue::Null,
149            (ColumnType::Int64, serde_json::Value::Number(n)) => {
150                ColumnValue::Int(n.as_i64().unwrap_or_default())
151            }
152            (ColumnType::Float64, serde_json::Value::Number(n)) => {
153                ColumnValue::Float(n.as_f64().unwrap_or_default())
154            }
155            (ColumnType::Utf8, serde_json::Value::String(s)) => ColumnValue::Utf8(s.clone()),
156            (ColumnType::Boolean, serde_json::Value::Bool(b)) => ColumnValue::Boolean(*b),
157            (ColumnType::Nullable(inner), val) => Self::from_json(val, inner),
158            _ => ColumnValue::Null,
159        }
160    }
161}
162
163impl std::fmt::Display for ColumnValue {
164    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
165        match self {
166            ColumnValue::Null => write!(f, "NULL"),
167            ColumnValue::Int(i) => write!(f, "{i}"),
168            ColumnValue::Float(v) => write!(f, "{v}"),
169            ColumnValue::Utf8(s) => write!(f, "{s}"),
170            ColumnValue::Boolean(b) => write!(f, "{b}"),
171        }
172    }
173}
174
175// ─── Column schema ───────────────────────────────────────────────────────────
176
177/// Describes a column in an [`InMemoryTable`].
178#[derive(Debug, Clone, Serialize, Deserialize)]
179pub struct ColumnSchema {
180    /// Column name
181    pub name: String,
182    /// Column type
183    pub col_type: ColumnType,
184}
185
186// ─── InMemoryTable ───────────────────────────────────────────────────────────
187
188/// A columnar in-memory table.
189///
190/// Data is stored in row-major form (`Vec<Vec<ColumnValue>>`) to simplify row
191/// operations like filtering and joining, while still supporting column projection.
192#[derive(Debug, Clone, Default, Serialize, Deserialize)]
193pub struct InMemoryTable {
194    /// Column schemas (ordered)
195    pub columns: Vec<ColumnSchema>,
196    /// Row data: each row is `Vec<ColumnValue>` with one entry per column
197    pub rows: Vec<Vec<ColumnValue>>,
198    /// Optional table name
199    pub name: Option<String>,
200}
201
202impl InMemoryTable {
203    /// Create a new empty table with the given column name/type pairs.
204    pub fn new(columns: Vec<(String, ColumnType)>) -> Self {
205        Self {
206            columns: columns
207                .into_iter()
208                .map(|(n, t)| ColumnSchema {
209                    name: n,
210                    col_type: t,
211                })
212                .collect(),
213            rows: Vec::new(),
214            name: None,
215        }
216    }
217
218    /// Set the table name.
219    pub fn with_name(mut self, name: impl Into<String>) -> Self {
220        self.name = Some(name.into());
221        self
222    }
223
224    /// Return the number of rows.
225    pub fn row_count(&self) -> usize {
226        self.rows.len()
227    }
228
229    /// Return the number of columns.
230    pub fn column_count(&self) -> usize {
231        self.columns.len()
232    }
233
234    /// Return column index by name.
235    pub fn column_index(&self, name: &str) -> Option<usize> {
236        self.columns.iter().position(|c| c.name == name)
237    }
238
239    /// Add a row. Returns an error if the row length doesn't match the schema.
240    pub fn push_row(&mut self, row: &[ColumnValue]) -> Result<()> {
241        if row.len() != self.columns.len() {
242            return Err(IoError::ValidationError(format!(
243                "Row has {} values but table has {} columns",
244                row.len(),
245                self.columns.len()
246            )));
247        }
248        self.rows.push(row.to_vec());
249        Ok(())
250    }
251
252    /// Add a row from a map. Missing columns get NULL.
253    pub fn push_row_map(&mut self, map: HashMap<String, ColumnValue>) -> Result<()> {
254        let row: Vec<ColumnValue> = self
255            .columns
256            .iter()
257            .map(|col| map.get(&col.name).cloned().unwrap_or(ColumnValue::Null))
258            .collect();
259        self.rows.push(row);
260        Ok(())
261    }
262
263    /// Get a cell value.
264    pub fn get(&self, row: usize, col: usize) -> Option<&ColumnValue> {
265        self.rows.get(row)?.get(col)
266    }
267
268    /// Get a column as a slice of values.
269    pub fn get_column(&self, name: &str) -> Option<Vec<&ColumnValue>> {
270        let idx = self.column_index(name)?;
271        Some(self.rows.iter().map(|r| &r[idx]).collect())
272    }
273
274    /// Return a row as a HashMap.
275    pub fn row_as_map(&self, row_idx: usize) -> Option<HashMap<String, ColumnValue>> {
276        let row = self.rows.get(row_idx)?;
277        Some(
278            self.columns
279                .iter()
280                .zip(row.iter())
281                .map(|(col, val)| (col.name.clone(), val.clone()))
282                .collect(),
283        )
284    }
285
286    /// Return rows as a Vec of JSON objects.
287    pub fn to_json_rows(&self) -> Vec<serde_json::Value> {
288        self.rows
289            .iter()
290            .map(|row| {
291                let obj: serde_json::Map<String, serde_json::Value> = self
292                    .columns
293                    .iter()
294                    .zip(row.iter())
295                    .map(|(col, val)| (col.name.clone(), val.to_json()))
296                    .collect();
297                serde_json::Value::Object(obj)
298            })
299            .collect()
300    }
301
302    /// Append all rows from another table (must have compatible schemas).
303    pub fn append(&mut self, other: &InMemoryTable) -> Result<()> {
304        if self.columns.len() != other.columns.len() {
305            return Err(IoError::ValidationError(
306                "Column count mismatch in append".to_string(),
307            ));
308        }
309        for (a, b) in self.columns.iter().zip(other.columns.iter()) {
310            if a.name != b.name {
311                return Err(IoError::ValidationError(format!(
312                    "Column name mismatch: '{}' vs '{}'",
313                    a.name, b.name
314                )));
315            }
316        }
317        self.rows.extend(other.rows.clone());
318        Ok(())
319    }
320}
321
322// ─── TableFilter ─────────────────────────────────────────────────────────────
323
324/// A filter predicate for rows.
325#[derive(Debug, Clone)]
326pub enum Predicate {
327    /// column == value
328    Eq(String, ColumnValue),
329    /// column != value
330    Ne(String, ColumnValue),
331    /// column > value
332    Greater(String, ColumnValue),
333    /// column >= value
334    GreaterEq(String, ColumnValue),
335    /// column < value
336    Less(String, ColumnValue),
337    /// column <= value
338    LessEq(String, ColumnValue),
339    /// column IS NULL
340    IsNull(String),
341    /// column IS NOT NULL
342    IsNotNull(String),
343    /// String column LIKE pattern (% = any chars, _ = single char)
344    Like(String, String),
345    /// column value is one of
346    In(String, Vec<ColumnValue>),
347    /// Both predicates must hold
348    And(Box<Predicate>, Box<Predicate>),
349    /// Either predicate must hold
350    Or(Box<Predicate>, Box<Predicate>),
351    /// Predicate must not hold
352    Not(Box<Predicate>),
353}
354
355impl Predicate {
356    fn eval(&self, row: &[ColumnValue], columns: &[ColumnSchema]) -> bool {
357        match self {
358            Predicate::Eq(col, val) => get_col_val(row, columns, col)
359                .map(|v| v == val)
360                .unwrap_or(false),
361            Predicate::Ne(col, val) => get_col_val(row, columns, col)
362                .map(|v| v != val)
363                .unwrap_or(false),
364            Predicate::Greater(col, val) => get_col_val(row, columns, col)
365                .and_then(|v| v.partial_cmp_value(val))
366                .map(|o| o == Ordering::Greater)
367                .unwrap_or(false),
368            Predicate::GreaterEq(col, val) => get_col_val(row, columns, col)
369                .and_then(|v| v.partial_cmp_value(val))
370                .map(|o| o != Ordering::Less)
371                .unwrap_or(false),
372            Predicate::Less(col, val) => get_col_val(row, columns, col)
373                .and_then(|v| v.partial_cmp_value(val))
374                .map(|o| o == Ordering::Less)
375                .unwrap_or(false),
376            Predicate::LessEq(col, val) => get_col_val(row, columns, col)
377                .and_then(|v| v.partial_cmp_value(val))
378                .map(|o| o != Ordering::Greater)
379                .unwrap_or(false),
380            Predicate::IsNull(col) => get_col_val(row, columns, col)
381                .map(|v| matches!(v, ColumnValue::Null))
382                .unwrap_or(true),
383            Predicate::IsNotNull(col) => get_col_val(row, columns, col)
384                .map(|v| !matches!(v, ColumnValue::Null))
385                .unwrap_or(false),
386            Predicate::Like(col, pattern) => get_col_val(row, columns, col)
387                .and_then(|v| {
388                    if let ColumnValue::Utf8(s) = v {
389                        Some(like_match(s, pattern))
390                    } else {
391                        None
392                    }
393                })
394                .unwrap_or(false),
395            Predicate::In(col, values) => get_col_val(row, columns, col)
396                .map(|v| values.contains(v))
397                .unwrap_or(false),
398            Predicate::And(a, b) => a.eval(row, columns) && b.eval(row, columns),
399            Predicate::Or(a, b) => a.eval(row, columns) || b.eval(row, columns),
400            Predicate::Not(p) => !p.eval(row, columns),
401        }
402    }
403}
404
405fn get_col_val<'a>(
406    row: &'a [ColumnValue],
407    columns: &[ColumnSchema],
408    name: &str,
409) -> Option<&'a ColumnValue> {
410    let idx = columns.iter().position(|c| c.name == name)?;
411    row.get(idx)
412}
413
414/// Simple LIKE pattern matching (% = any, _ = one char).
415fn like_match(s: &str, pattern: &str) -> bool {
416    like_match_recursive(s.as_bytes(), pattern.as_bytes())
417}
418
419fn like_match_recursive(s: &[u8], p: &[u8]) -> bool {
420    match (s, p) {
421        (_, []) => s.is_empty(),
422        (_, [b'%', rest @ ..]) => {
423            // % matches zero or more characters
424            for i in 0..=s.len() {
425                if like_match_recursive(&s[i..], rest) {
426                    return true;
427                }
428            }
429            false
430        }
431        ([], _) => false,
432        ([sc, s_rest @ ..], [b'_', p_rest @ ..]) => {
433            like_match_recursive(s_rest, p_rest)
434                || (sc.is_ascii() && like_match_recursive(s_rest, p_rest))
435        }
436        ([sc, s_rest @ ..], [pc, p_rest @ ..]) => {
437            sc.eq_ignore_ascii_case(pc) && like_match_recursive(s_rest, p_rest)
438        }
439    }
440}
441
442/// Predicate-based row filtering.
443pub struct TableFilter<'a> {
444    table: &'a InMemoryTable,
445    predicates: Vec<Predicate>,
446}
447
448impl<'a> TableFilter<'a> {
449    /// Create a new filter builder.
450    pub fn new(table: &'a InMemoryTable) -> Self {
451        Self {
452            table,
453            predicates: Vec::new(),
454        }
455    }
456
457    /// Add a predicate (combined with AND).
458    pub fn predicate(mut self, p: Predicate) -> Self {
459        self.predicates.push(p);
460        self
461    }
462
463    /// Apply all predicates and return a new table with matching rows.
464    pub fn apply(&self) -> Result<InMemoryTable> {
465        let mut result = InMemoryTable {
466            columns: self.table.columns.clone(),
467            rows: Vec::new(),
468            name: self.table.name.clone(),
469        };
470        for row in &self.table.rows {
471            let matches = self
472                .predicates
473                .iter()
474                .all(|p| p.eval(row, &self.table.columns));
475            if matches {
476                result.rows.push(row.clone());
477            }
478        }
479        Ok(result)
480    }
481}
482
483// ─── TableSort ───────────────────────────────────────────────────────────────
484
485/// Sort direction.
486#[derive(Debug, Clone, Copy, PartialEq, Eq)]
487pub enum SortDirection {
488    /// Ascending order
489    Asc,
490    /// Descending order
491    Desc,
492}
493
494/// A column + direction sort key.
495#[derive(Debug, Clone)]
496pub struct SortKey {
497    /// Column name
498    pub column: String,
499    /// Sort direction
500    pub direction: SortDirection,
501    /// Whether NULLs sort first
502    pub nulls_first: bool,
503}
504
505impl SortKey {
506    /// Ascending sort on a column.
507    pub fn asc(column: impl Into<String>) -> Self {
508        Self {
509            column: column.into(),
510            direction: SortDirection::Asc,
511            nulls_first: false,
512        }
513    }
514
515    /// Descending sort on a column.
516    pub fn desc(column: impl Into<String>) -> Self {
517        Self {
518            column: column.into(),
519            direction: SortDirection::Desc,
520            nulls_first: false,
521        }
522    }
523}
524
525/// Multi-column sort.
526pub struct TableSort;
527
528impl TableSort {
529    /// Sort `table` by the given keys and return a new sorted table.
530    pub fn sort(table: &InMemoryTable, keys: &[SortKey]) -> Result<InMemoryTable> {
531        // Validate that all sort columns exist
532        for key in keys {
533            if table.column_index(&key.column).is_none() {
534                return Err(IoError::ValidationError(format!(
535                    "Sort column '{}' not found in table",
536                    key.column
537                )));
538            }
539        }
540
541        let mut rows = table.rows.clone();
542        rows.sort_by(|a, b| {
543            for key in keys {
544                let idx = table
545                    .columns
546                    .iter()
547                    .position(|c| c.name == key.column)
548                    .unwrap_or(0);
549                let va = &a[idx];
550                let vb = &b[idx];
551
552                let ord = match (va, vb) {
553                    (ColumnValue::Null, ColumnValue::Null) => Ordering::Equal,
554                    (ColumnValue::Null, _) => {
555                        if key.nulls_first {
556                            Ordering::Less
557                        } else {
558                            Ordering::Greater
559                        }
560                    }
561                    (_, ColumnValue::Null) => {
562                        if key.nulls_first {
563                            Ordering::Greater
564                        } else {
565                            Ordering::Less
566                        }
567                    }
568                    _ => va.partial_cmp_value(vb).unwrap_or(Ordering::Equal),
569                };
570
571                let ord = if key.direction == SortDirection::Desc {
572                    ord.reverse()
573                } else {
574                    ord
575                };
576
577                if ord != Ordering::Equal {
578                    return ord;
579                }
580            }
581            Ordering::Equal
582        });
583
584        Ok(InMemoryTable {
585            columns: table.columns.clone(),
586            rows,
587            name: table.name.clone(),
588        })
589    }
590}
591
592// ─── TableProjection ────────────────────────────────────────────────────────
593
594/// Column selection and renaming.
595pub struct TableProjection<'a> {
596    table: &'a InMemoryTable,
597    selections: Vec<(String, Option<String>)>, // (original_name, alias)
598}
599
600impl<'a> TableProjection<'a> {
601    /// Create a projection builder.
602    pub fn new(table: &'a InMemoryTable) -> Self {
603        Self {
604            table,
605            selections: Vec::new(),
606        }
607    }
608
609    /// Select a column (keep original name).
610    pub fn column(mut self, name: impl Into<String>) -> Self {
611        self.selections.push((name.into(), None));
612        self
613    }
614
615    /// Select a column with an alias.
616    pub fn column_as(mut self, name: impl Into<String>, alias: impl Into<String>) -> Self {
617        self.selections.push((name.into(), Some(alias.into())));
618        self
619    }
620
621    /// Apply the projection and return a new table.
622    pub fn apply(&self) -> Result<InMemoryTable> {
623        // Resolve column indices
624        let mut indices: Vec<(usize, String)> = Vec::new();
625        for (orig, alias) in &self.selections {
626            let idx = self.table.column_index(orig).ok_or_else(|| {
627                IoError::ValidationError(format!("Projection column '{}' not found in table", orig))
628            })?;
629            let out_name = alias.as_deref().unwrap_or(orig.as_str()).to_string();
630            indices.push((idx, out_name));
631        }
632
633        let new_columns: Vec<ColumnSchema> = indices
634            .iter()
635            .map(|(idx, name)| ColumnSchema {
636                name: name.clone(),
637                col_type: self.table.columns[*idx].col_type.clone(),
638            })
639            .collect();
640
641        let new_rows: Vec<Vec<ColumnValue>> = self
642            .table
643            .rows
644            .iter()
645            .map(|row| indices.iter().map(|(idx, _)| row[*idx].clone()).collect())
646            .collect();
647
648        Ok(InMemoryTable {
649            columns: new_columns,
650            rows: new_rows,
651            name: self.table.name.clone(),
652        })
653    }
654}
655
656// ─── GroupBy ─────────────────────────────────────────────────────────────────
657
658/// Aggregation function for group-by.
659#[derive(Debug, Clone)]
660pub enum AggFunc {
661    /// Count rows in group
662    Count,
663    /// Sum of column values
664    Sum(String),
665    /// Arithmetic mean
666    Mean(String),
667    /// Minimum value
668    Min(String),
669    /// Maximum value
670    Max(String),
671    /// Sample standard deviation
672    Std(String),
673    /// Collect distinct count
674    CountDistinct(String),
675}
676
677impl AggFunc {
678    /// Return the output column name for this aggregation.
679    pub fn output_name(&self) -> String {
680        match self {
681            AggFunc::Count => "count".to_string(),
682            AggFunc::Sum(col) => format!("sum_{col}"),
683            AggFunc::Mean(col) => format!("mean_{col}"),
684            AggFunc::Min(col) => format!("min_{col}"),
685            AggFunc::Max(col) => format!("max_{col}"),
686            AggFunc::Std(col) => format!("std_{col}"),
687            AggFunc::CountDistinct(col) => format!("count_distinct_{col}"),
688        }
689    }
690
691    fn compute(&self, rows: &[&Vec<ColumnValue>], columns: &[ColumnSchema]) -> ColumnValue {
692        match self {
693            AggFunc::Count => ColumnValue::Int(rows.len() as i64),
694            AggFunc::Sum(col) => {
695                let sum: f64 = rows
696                    .iter()
697                    .filter_map(|r| get_col_val(r, columns, col)?.as_f64())
698                    .sum();
699                ColumnValue::Float(sum)
700            }
701            AggFunc::Mean(col) => {
702                let vals: Vec<f64> = rows
703                    .iter()
704                    .filter_map(|r| get_col_val(r, columns, col)?.as_f64())
705                    .collect();
706                if vals.is_empty() {
707                    ColumnValue::Null
708                } else {
709                    ColumnValue::Float(vals.iter().sum::<f64>() / vals.len() as f64)
710                }
711            }
712            AggFunc::Min(col) => rows
713                .iter()
714                .filter_map(|r| get_col_val(r, columns, col))
715                .filter(|v| !matches!(v, ColumnValue::Null))
716                .min_by(|a, b| a.partial_cmp_value(b).unwrap_or(Ordering::Equal))
717                .cloned()
718                .unwrap_or(ColumnValue::Null),
719            AggFunc::Max(col) => rows
720                .iter()
721                .filter_map(|r| get_col_val(r, columns, col))
722                .filter(|v| !matches!(v, ColumnValue::Null))
723                .max_by(|a, b| a.partial_cmp_value(b).unwrap_or(Ordering::Equal))
724                .cloned()
725                .unwrap_or(ColumnValue::Null),
726            AggFunc::Std(col) => {
727                let vals: Vec<f64> = rows
728                    .iter()
729                    .filter_map(|r| get_col_val(r, columns, col)?.as_f64())
730                    .collect();
731                if vals.len() < 2 {
732                    ColumnValue::Float(0.0)
733                } else {
734                    let n = vals.len() as f64;
735                    let mean = vals.iter().sum::<f64>() / n;
736                    let variance = vals.iter().map(|v| (v - mean).powi(2)).sum::<f64>() / (n - 1.0);
737                    ColumnValue::Float(variance.sqrt())
738                }
739            }
740            AggFunc::CountDistinct(col) => {
741                let mut seen: Vec<String> = Vec::new();
742                for r in rows {
743                    if let Some(v) = get_col_val(r, columns, col) {
744                        let s = v.to_string();
745                        if !seen.contains(&s) {
746                            seen.push(s);
747                        }
748                    }
749                }
750                ColumnValue::Int(seen.len() as i64)
751            }
752        }
753    }
754}
755
756/// Group-by aggregation operator.
757pub struct GroupBy<'a> {
758    table: &'a InMemoryTable,
759    group_cols: Vec<String>,
760    agg_funcs: Vec<AggFunc>,
761}
762
763impl<'a> GroupBy<'a> {
764    /// Create a group-by builder.
765    pub fn new(table: &'a InMemoryTable, group_cols: Vec<String>) -> Self {
766        Self {
767            table,
768            group_cols,
769            agg_funcs: Vec::new(),
770        }
771    }
772
773    /// Add an aggregation.
774    pub fn agg(mut self, func: AggFunc) -> Self {
775        self.agg_funcs.push(func);
776        self
777    }
778
779    /// Execute the group-by and return an aggregated table.
780    pub fn apply(&self) -> Result<InMemoryTable> {
781        // Validate group columns exist
782        let group_indices: Vec<usize> = self
783            .group_cols
784            .iter()
785            .map(|col| {
786                self.table.column_index(col).ok_or_else(|| {
787                    IoError::ValidationError(format!("Group-by column '{col}' not found"))
788                })
789            })
790            .collect::<Result<Vec<_>>>()?;
791
792        // Group rows by key
793        let mut groups: HashMap<Vec<String>, Vec<&Vec<ColumnValue>>> = HashMap::new();
794        for row in &self.table.rows {
795            let key: Vec<String> = group_indices.iter().map(|&i| row[i].to_string()).collect();
796            groups.entry(key).or_default().push(row);
797        }
798
799        // Build output schema
800        let mut out_columns: Vec<ColumnSchema> = self
801            .group_cols
802            .iter()
803            .map(|name| {
804                let idx = self.table.column_index(name).unwrap_or(0);
805                ColumnSchema {
806                    name: name.clone(),
807                    col_type: self.table.columns[idx].col_type.clone(),
808                }
809            })
810            .collect();
811        for agg in &self.agg_funcs {
812            out_columns.push(ColumnSchema {
813                name: agg.output_name(),
814                col_type: ColumnType::Float64,
815            });
816        }
817
818        // Compute results
819        let mut out_rows: Vec<Vec<ColumnValue>> = Vec::new();
820        // Sort groups for determinism
821        let mut keys: Vec<Vec<String>> = groups.keys().cloned().collect();
822        keys.sort();
823
824        for key in &keys {
825            let group_rows = &groups[key];
826            // Get representative row for group key values
827            let first_row = group_rows[0];
828            let mut out_row: Vec<ColumnValue> = group_indices
829                .iter()
830                .map(|&i| first_row[i].clone())
831                .collect();
832            for agg in &self.agg_funcs {
833                out_row.push(agg.compute(group_rows, &self.table.columns));
834            }
835            out_rows.push(out_row);
836        }
837
838        Ok(InMemoryTable {
839            columns: out_columns,
840            rows: out_rows,
841            name: None,
842        })
843    }
844}
845
846// ─── TableJoin ───────────────────────────────────────────────────────────────
847
848/// Join type.
849#[derive(Debug, Clone, Copy, PartialEq, Eq)]
850pub enum JoinType {
851    /// Inner join (only matching rows)
852    Inner,
853    /// Left outer join (all left rows, NULLs for unmatched right)
854    Left,
855    /// Right outer join (all right rows, NULLs for unmatched left)
856    Right,
857    /// Cross join (cartesian product)
858    Cross,
859}
860
861/// Table join operations.
862pub struct TableJoin;
863
864impl TableJoin {
865    /// Hash join two tables on the given key columns.
866    pub fn hash_join(
867        left: &InMemoryTable,
868        right: &InMemoryTable,
869        left_key: &str,
870        right_key: &str,
871        join_type: JoinType,
872    ) -> Result<InMemoryTable> {
873        let left_key_idx = left.column_index(left_key).ok_or_else(|| {
874            IoError::ValidationError(format!("Left join key '{left_key}' not found"))
875        })?;
876        let right_key_idx = right.column_index(right_key).ok_or_else(|| {
877            IoError::ValidationError(format!("Right join key '{right_key}' not found"))
878        })?;
879
880        // Build output schema: left columns + right columns (excluding join key)
881        let mut out_columns: Vec<ColumnSchema> = left.columns.clone();
882        for (i, col) in right.columns.iter().enumerate() {
883            if i != right_key_idx {
884                // Avoid duplicate name
885                let name = if left.column_index(&col.name).is_some() {
886                    format!("right_{}", col.name)
887                } else {
888                    col.name.clone()
889                };
890                out_columns.push(ColumnSchema {
891                    name,
892                    col_type: col.col_type.clone(),
893                });
894            }
895        }
896
897        let null_left: Vec<ColumnValue> = left.columns.iter().map(|_| ColumnValue::Null).collect();
898        let null_right_no_key: Vec<ColumnValue> = right
899            .columns
900            .iter()
901            .enumerate()
902            .filter(|(i, _)| *i != right_key_idx)
903            .map(|_| ColumnValue::Null)
904            .collect();
905
906        match join_type {
907            JoinType::Cross => {
908                let mut rows = Vec::new();
909                for l_row in &left.rows {
910                    for r_row in &right.rows {
911                        let mut out_row = l_row.clone();
912                        for (i, v) in r_row.iter().enumerate() {
913                            if i != right_key_idx {
914                                out_row.push(v.clone());
915                            }
916                        }
917                        rows.push(out_row);
918                    }
919                }
920                return Ok(InMemoryTable {
921                    columns: out_columns,
922                    rows,
923                    name: None,
924                });
925            }
926            JoinType::Inner | JoinType::Left | JoinType::Right => {}
927        }
928
929        // Build hash map from right key → rows
930        let mut right_map: HashMap<String, Vec<usize>> = HashMap::new();
931        for (i, r_row) in right.rows.iter().enumerate() {
932            let key = r_row[right_key_idx].to_string();
933            right_map.entry(key).or_default().push(i);
934        }
935
936        let mut rows: Vec<Vec<ColumnValue>> = Vec::new();
937        let mut right_matched: Vec<bool> = vec![false; right.rows.len()];
938
939        for l_row in &left.rows {
940            let key = l_row[left_key_idx].to_string();
941            match right_map.get(&key) {
942                Some(r_indices) => {
943                    for &ri in r_indices {
944                        right_matched[ri] = true;
945                        let mut out_row = l_row.clone();
946                        for (i, v) in right.rows[ri].iter().enumerate() {
947                            if i != right_key_idx {
948                                out_row.push(v.clone());
949                            }
950                        }
951                        rows.push(out_row);
952                    }
953                }
954                None => {
955                    if join_type == JoinType::Left {
956                        let mut out_row = l_row.clone();
957                        out_row.extend(null_right_no_key.iter().cloned());
958                        rows.push(out_row);
959                    }
960                }
961            }
962        }
963
964        // Right outer join: include unmatched right rows
965        if join_type == JoinType::Right {
966            for (i, r_row) in right.rows.iter().enumerate() {
967                if !right_matched[i] {
968                    let mut out_row = null_left.clone();
969                    for (j, v) in r_row.iter().enumerate() {
970                        if j != right_key_idx {
971                            out_row.push(v.clone());
972                        }
973                    }
974                    rows.push(out_row);
975                }
976            }
977        }
978
979        Ok(InMemoryTable {
980            columns: out_columns,
981            rows,
982            name: None,
983        })
984    }
985
986    /// Merge join two pre-sorted tables on the given key columns.
987    /// Both tables must already be sorted ascending by their key columns.
988    pub fn merge_join(
989        left: &InMemoryTable,
990        right: &InMemoryTable,
991        left_key: &str,
992        right_key: &str,
993    ) -> Result<InMemoryTable> {
994        let left_key_idx = left.column_index(left_key).ok_or_else(|| {
995            IoError::ValidationError(format!("Left merge key '{left_key}' not found"))
996        })?;
997        let right_key_idx = right.column_index(right_key).ok_or_else(|| {
998            IoError::ValidationError(format!("Right merge key '{right_key}' not found"))
999        })?;
1000
1001        let mut out_columns: Vec<ColumnSchema> = left.columns.clone();
1002        for (i, col) in right.columns.iter().enumerate() {
1003            if i != right_key_idx {
1004                let name = if left.column_index(&col.name).is_some() {
1005                    format!("right_{}", col.name)
1006                } else {
1007                    col.name.clone()
1008                };
1009                out_columns.push(ColumnSchema {
1010                    name,
1011                    col_type: col.col_type.clone(),
1012                });
1013            }
1014        }
1015
1016        let mut rows = Vec::new();
1017        let mut li = 0usize;
1018        let mut ri = 0usize;
1019
1020        while li < left.rows.len() && ri < right.rows.len() {
1021            let lk = &left.rows[li][left_key_idx];
1022            let rk = &right.rows[ri][right_key_idx];
1023
1024            match lk.partial_cmp_value(rk).unwrap_or(Ordering::Equal) {
1025                Ordering::Equal => {
1026                    // Collect all matching right rows
1027                    let mut rj = ri;
1028                    while rj < right.rows.len() {
1029                        let rk2 = &right.rows[rj][right_key_idx];
1030                        if lk.partial_cmp_value(rk2) != Some(Ordering::Equal) {
1031                            break;
1032                        }
1033                        let mut out_row = left.rows[li].clone();
1034                        for (k, v) in right.rows[rj].iter().enumerate() {
1035                            if k != right_key_idx {
1036                                out_row.push(v.clone());
1037                            }
1038                        }
1039                        rows.push(out_row);
1040                        rj += 1;
1041                    }
1042                    li += 1;
1043                }
1044                Ordering::Less => li += 1,
1045                Ordering::Greater => ri += 1,
1046            }
1047        }
1048
1049        Ok(InMemoryTable {
1050            columns: out_columns,
1051            rows,
1052            name: None,
1053        })
1054    }
1055}
1056
1057// ─── Tests ───────────────────────────────────────────────────────────────────
1058
1059#[cfg(test)]
1060mod tests {
1061    use super::*;
1062
1063    fn make_table() -> InMemoryTable {
1064        let mut t = InMemoryTable::new(vec![
1065            ("id".to_string(), ColumnType::Int64),
1066            ("name".to_string(), ColumnType::Utf8),
1067            ("score".to_string(), ColumnType::Float64),
1068            ("dept".to_string(), ColumnType::Utf8),
1069        ]);
1070        t.push_row(&[
1071            ColumnValue::Int(1),
1072            ColumnValue::Utf8("Alice".to_string()),
1073            ColumnValue::Float(95.0),
1074            ColumnValue::Utf8("eng".to_string()),
1075        ])
1076        .unwrap();
1077        t.push_row(&[
1078            ColumnValue::Int(2),
1079            ColumnValue::Utf8("Bob".to_string()),
1080            ColumnValue::Float(82.5),
1081            ColumnValue::Utf8("eng".to_string()),
1082        ])
1083        .unwrap();
1084        t.push_row(&[
1085            ColumnValue::Int(3),
1086            ColumnValue::Utf8("Carol".to_string()),
1087            ColumnValue::Float(91.0),
1088            ColumnValue::Utf8("hr".to_string()),
1089        ])
1090        .unwrap();
1091        t.push_row(&[
1092            ColumnValue::Int(4),
1093            ColumnValue::Utf8("Dave".to_string()),
1094            ColumnValue::Float(78.0),
1095            ColumnValue::Utf8("hr".to_string()),
1096        ])
1097        .unwrap();
1098        t
1099    }
1100
1101    #[test]
1102    fn test_filter_greater() {
1103        let t = make_table();
1104        let filtered = TableFilter::new(&t)
1105            .predicate(Predicate::Greater(
1106                "score".to_string(),
1107                ColumnValue::Float(90.0),
1108            ))
1109            .apply()
1110            .unwrap();
1111        assert_eq!(filtered.row_count(), 2); // Alice(95) and Carol(91)
1112    }
1113
1114    #[test]
1115    fn test_filter_eq() {
1116        let t = make_table();
1117        let filtered = TableFilter::new(&t)
1118            .predicate(Predicate::Eq(
1119                "dept".to_string(),
1120                ColumnValue::Utf8("eng".to_string()),
1121            ))
1122            .apply()
1123            .unwrap();
1124        assert_eq!(filtered.row_count(), 2);
1125    }
1126
1127    #[test]
1128    fn test_sort_asc() {
1129        let t = make_table();
1130        let sorted = TableSort::sort(&t, &[SortKey::asc("score")]).unwrap();
1131        let scores: Vec<f64> = sorted
1132            .get_column("score")
1133            .unwrap()
1134            .into_iter()
1135            .filter_map(|v| v.as_f64())
1136            .collect();
1137        assert_eq!(scores, vec![78.0, 82.5, 91.0, 95.0]);
1138    }
1139
1140    #[test]
1141    fn test_sort_desc() {
1142        let t = make_table();
1143        let sorted = TableSort::sort(&t, &[SortKey::desc("score")]).unwrap();
1144        let scores: Vec<f64> = sorted
1145            .get_column("score")
1146            .unwrap()
1147            .into_iter()
1148            .filter_map(|v| v.as_f64())
1149            .collect();
1150        assert_eq!(scores, vec![95.0, 91.0, 82.5, 78.0]);
1151    }
1152
1153    #[test]
1154    fn test_projection() {
1155        let t = make_table();
1156        let projected = TableProjection::new(&t)
1157            .column("id")
1158            .column_as("name", "full_name")
1159            .apply()
1160            .unwrap();
1161        assert_eq!(projected.column_count(), 2);
1162        assert_eq!(projected.columns[1].name, "full_name");
1163        assert_eq!(projected.row_count(), 4);
1164    }
1165
1166    #[test]
1167    fn test_group_by_sum_mean() {
1168        let t = make_table();
1169        let grouped = GroupBy::new(&t, vec!["dept".to_string()])
1170            .agg(AggFunc::Count)
1171            .agg(AggFunc::Sum("score".to_string()))
1172            .agg(AggFunc::Mean("score".to_string()))
1173            .apply()
1174            .unwrap();
1175
1176        assert_eq!(grouped.row_count(), 2); // eng, hr
1177                                            // eng has Alice(95) + Bob(82.5) = 177.5
1178        let eng_row = grouped
1179            .rows
1180            .iter()
1181            .find(|r| r[0] == ColumnValue::Utf8("eng".to_string()))
1182            .expect("eng group missing");
1183        assert_eq!(eng_row[1], ColumnValue::Int(2)); // count
1184        if let ColumnValue::Float(sum) = eng_row[2] {
1185            assert!((sum - 177.5).abs() < 1e-9);
1186        } else {
1187            panic!("Expected float sum");
1188        }
1189    }
1190
1191    #[test]
1192    fn test_inner_join() {
1193        let mut left = InMemoryTable::new(vec![
1194            ("id".to_string(), ColumnType::Int64),
1195            ("val".to_string(), ColumnType::Float64),
1196        ]);
1197        left.push_row(&[ColumnValue::Int(1), ColumnValue::Float(1.0)])
1198            .unwrap();
1199        left.push_row(&[ColumnValue::Int(2), ColumnValue::Float(2.0)])
1200            .unwrap();
1201        left.push_row(&[ColumnValue::Int(3), ColumnValue::Float(3.0)])
1202            .unwrap();
1203
1204        let mut right = InMemoryTable::new(vec![
1205            ("id".to_string(), ColumnType::Int64),
1206            ("label".to_string(), ColumnType::Utf8),
1207        ]);
1208        right
1209            .push_row(&[ColumnValue::Int(1), ColumnValue::Utf8("one".to_string())])
1210            .unwrap();
1211        right
1212            .push_row(&[ColumnValue::Int(2), ColumnValue::Utf8("two".to_string())])
1213            .unwrap();
1214
1215        let joined = TableJoin::hash_join(&left, &right, "id", "id", JoinType::Inner).unwrap();
1216        assert_eq!(joined.row_count(), 2);
1217    }
1218
1219    #[test]
1220    fn test_left_join() {
1221        let mut left = InMemoryTable::new(vec![("id".to_string(), ColumnType::Int64)]);
1222        left.push_row(&[ColumnValue::Int(1)]).unwrap();
1223        left.push_row(&[ColumnValue::Int(2)]).unwrap();
1224        left.push_row(&[ColumnValue::Int(3)]).unwrap(); // no match in right
1225
1226        let mut right = InMemoryTable::new(vec![
1227            ("id".to_string(), ColumnType::Int64),
1228            ("x".to_string(), ColumnType::Float64),
1229        ]);
1230        right
1231            .push_row(&[ColumnValue::Int(1), ColumnValue::Float(10.0)])
1232            .unwrap();
1233        right
1234            .push_row(&[ColumnValue::Int(2), ColumnValue::Float(20.0)])
1235            .unwrap();
1236
1237        let joined = TableJoin::hash_join(&left, &right, "id", "id", JoinType::Left).unwrap();
1238        assert_eq!(joined.row_count(), 3);
1239        // Row for id=3 should have NULL for 'x'
1240        let row3 = joined
1241            .rows
1242            .iter()
1243            .find(|r| r[0] == ColumnValue::Int(3))
1244            .expect("row 3 missing");
1245        assert_eq!(row3[1], ColumnValue::Null);
1246    }
1247
1248    #[test]
1249    fn test_cross_join() {
1250        let mut a = InMemoryTable::new(vec![("a".to_string(), ColumnType::Int64)]);
1251        a.push_row(&[ColumnValue::Int(1)]).unwrap();
1252        a.push_row(&[ColumnValue::Int(2)]).unwrap();
1253
1254        let mut b = InMemoryTable::new(vec![("b".to_string(), ColumnType::Int64)]);
1255        b.push_row(&[ColumnValue::Int(10)]).unwrap();
1256        b.push_row(&[ColumnValue::Int(20)]).unwrap();
1257        b.push_row(&[ColumnValue::Int(30)]).unwrap();
1258
1259        let crossed = TableJoin::hash_join(&a, &b, "a", "b", JoinType::Cross).unwrap();
1260        assert_eq!(crossed.row_count(), 6); // 2 × 3
1261    }
1262
1263    #[test]
1264    fn test_like_match() {
1265        assert!(like_match("hello world", "%world"));
1266        assert!(like_match("hello world", "hello%"));
1267        assert!(like_match("hello world", "%lo w%"));
1268        assert!(!like_match("hello world", "xyz%"));
1269        assert!(like_match("abc", "a_c"));
1270        assert!(!like_match("axyz", "a_c"));
1271    }
1272}