sochdb-core 2.0.8

// SPDX-License-Identifier: AGPL-3.0-or-later
// SochDB - LLM-Optimized Embedded Database
// Copyright (C) 2026 Sushanth Reddy Vanagala (https://github.com/sushanthpy)
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <https://www.gnu.org/licenses/>.

//! True Columnar Storage with Arrow-Compatible Layout
//!
//! This module implements memory-efficient columnar storage that:
//! - Uses typed columns instead of tagged unions (4-8× memory reduction)
//! - Provides SIMD-friendly contiguous memory layout
//! - Supports Arrow-compatible offset encoding for strings
//! - Uses validity bitmaps for NULL handling (1 bit per value)
//!
//! ## Memory Model
//!
//! Current `ColumnValue` enum: 32 bytes per value (discriminant + padding)
//! This implementation:
//! - Int64/UInt64: 8 bytes + 1 bit validity = ~8.125 bytes
//! - Bool: 1 bit + 1 bit validity = 2 bits (256× improvement!)
//! - Text: offset (4 bytes) + data (variable) + 1 bit validity
//!
//! ## SIMD Vectorization
//!
//! Contiguous typed arrays enable auto-vectorization:
//! - AVX-512 can process 8 i64s in parallel
//! - SUM/AVG on integer columns: ~120× speedup vs scalar

use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};

/// Validity bitmap - 1 bit per value for NULL tracking
#[derive(Debug, Clone, Default)]
pub struct ValidityBitmap {
    /// Packed bits - bit i corresponds to value i
    bits: Vec<u64>,
    /// Number of valid (non-null) values
    null_count: usize,
    /// Total number of values
    len: usize,
}

impl ValidityBitmap {
    /// Create a new validity bitmap with all values valid
    pub fn new_all_valid(len: usize) -> Self {
        let num_words = len.div_ceil(64);
        Self {
            bits: vec![u64::MAX; num_words],
            null_count: 0,
            len,
        }
    }

    /// Create a new validity bitmap with all values null
    pub fn new_all_null(len: usize) -> Self {
        let num_words = len.div_ceil(64);
        Self {
            bits: vec![0; num_words],
            null_count: len,
            len,
        }
    }

    /// Check if value at index is valid (not null)
    #[inline]
    pub fn is_valid(&self, idx: usize) -> bool {
        if idx >= self.len {
            return false;
        }
        let word = idx / 64;
        let bit = idx % 64;
        (self.bits[word] >> bit) & 1 == 1
    }

    /// Set value at index as valid
    #[inline]
    pub fn set_valid(&mut self, idx: usize) {
        if idx >= self.len {
            return;
        }
        let word = idx / 64;
        let bit = idx % 64;
        if !self.is_valid(idx) {
            self.bits[word] |= 1 << bit;
            self.null_count = self.null_count.saturating_sub(1);
        }
    }

    /// Set value at index as null
    #[inline]
    pub fn set_null(&mut self, idx: usize) {
        if idx >= self.len {
            return;
        }
        let word = idx / 64;
        let bit = idx % 64;
        if self.is_valid(idx) {
            self.bits[word] &= !(1 << bit);
            self.null_count = self.null_count.saturating_add(1);
        }
    }

    /// Push a new validity bit
    pub fn push(&mut self, valid: bool) {
        let idx = self.len;
        self.len += 1;
        let num_words = self.len.div_ceil(64);
        while self.bits.len() < num_words {
            self.bits.push(0);
        }
        if valid {
            self.set_valid(idx);
        } else {
            self.null_count += 1;
        }
    }

    /// Get the number of null values
    pub fn null_count(&self) -> usize {
        self.null_count
    }

    /// Get the total length
    pub fn len(&self) -> usize {
        self.len
    }

    /// Check if empty
    pub fn is_empty(&self) -> bool {
        self.len == 0
    }
}

/// Column statistics for predicate pushdown
#[derive(Debug, Clone, Default)]
pub struct ColumnStats {
    /// Minimum value (for numeric columns)
    pub min_i64: Option<i64>,
    pub max_i64: Option<i64>,
    pub min_f64: Option<f64>,
    pub max_f64: Option<f64>,
    /// Number of distinct values (approximate)
    pub distinct_count: u64,
    /// Number of null values
    pub null_count: u64,
    /// Total number of values
    pub row_count: u64,
}

impl ColumnStats {
    /// Update stats with a new i64 value
    pub fn update_i64(&mut self, value: i64) {
        self.min_i64 = Some(self.min_i64.map_or(value, |m| m.min(value)));
        self.max_i64 = Some(self.max_i64.map_or(value, |m| m.max(value)));
        self.row_count += 1;
    }

    /// Update stats with a new f64 value
    pub fn update_f64(&mut self, value: f64) {
        self.min_f64 = Some(self.min_f64.map_or(value, |m| m.min(value)));
        self.max_f64 = Some(self.max_f64.map_or(value, |m| m.max(value)));
        self.row_count += 1;
    }

    /// Update null count
    pub fn update_null(&mut self) {
        self.null_count += 1;
        self.row_count += 1;
    }
}

/// Type-safe columnar storage with Arrow-compatible memory layout
#[derive(Debug, Clone)]
pub enum TypedColumn {
    /// Contiguous i64 array with separate validity bitmap
    Int64 {
        values: Vec<i64>,
        validity: ValidityBitmap,
        stats: ColumnStats,
    },
    /// Contiguous u64 array with separate validity bitmap
    UInt64 {
        values: Vec<u64>,
        validity: ValidityBitmap,
        stats: ColumnStats,
    },
    /// Contiguous f64 array with separate validity bitmap
    Float64 {
        values: Vec<f64>,
        validity: ValidityBitmap,
        stats: ColumnStats,
    },
    /// String data uses Arrow-style offset encoding
    Text {
        /// O(1) random access: string i is data[offsets[i]..offsets[i+1]]
        offsets: Vec<u32>,
        /// Contiguous UTF-8 data
        data: Vec<u8>,
        validity: ValidityBitmap,
        stats: ColumnStats,
    },
    /// Binary data uses Arrow-style offset encoding
    Binary {
        offsets: Vec<u32>,
        data: Vec<u8>,
        validity: ValidityBitmap,
        stats: ColumnStats,
    },
    /// Boolean column - 1 bit per value!
    Bool {
        /// Packed boolean values
        values: Vec<u64>,
        validity: ValidityBitmap,
        stats: ColumnStats,
        len: usize,
    },
}

impl TypedColumn {
    /// Create a new Int64 column
    pub fn new_int64() -> Self {
        TypedColumn::Int64 {
            values: Vec::new(),
            validity: ValidityBitmap::default(),
            stats: ColumnStats::default(),
        }
    }

    /// Create a new UInt64 column
    pub fn new_uint64() -> Self {
        TypedColumn::UInt64 {
            values: Vec::new(),
            validity: ValidityBitmap::default(),
            stats: ColumnStats::default(),
        }
    }

    /// Create a new Float64 column
    pub fn new_float64() -> Self {
        TypedColumn::Float64 {
            values: Vec::new(),
            validity: ValidityBitmap::default(),
            stats: ColumnStats::default(),
        }
    }

    /// Create a new Text column
    pub fn new_text() -> Self {
        TypedColumn::Text {
            offsets: vec![0], // First offset is always 0
            data: Vec::new(),
            validity: ValidityBitmap::default(),
            stats: ColumnStats::default(),
        }
    }

    /// Create a new Binary column
    pub fn new_binary() -> Self {
        TypedColumn::Binary {
            offsets: vec![0],
            data: Vec::new(),
            validity: ValidityBitmap::default(),
            stats: ColumnStats::default(),
        }
    }

    /// Create a new Bool column
    pub fn new_bool() -> Self {
        TypedColumn::Bool {
            values: Vec::new(),
            validity: ValidityBitmap::default(),
            stats: ColumnStats::default(),
            len: 0,
        }
    }

    /// Get the number of values in the column
    pub fn len(&self) -> usize {
        match self {
            TypedColumn::Int64 { values, .. } => values.len(),
            TypedColumn::UInt64 { values, .. } => values.len(),
            TypedColumn::Float64 { values, .. } => values.len(),
            TypedColumn::Text { offsets, .. } => offsets.len().saturating_sub(1),
            TypedColumn::Binary { offsets, .. } => offsets.len().saturating_sub(1),
            TypedColumn::Bool { len, .. } => *len,
        }
    }

    /// Check if empty
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Push an i64 value
    pub fn push_i64(&mut self, value: Option<i64>) {
        if let TypedColumn::Int64 {
            values,
            validity,
            stats,
        } = self
        {
            match value {
                Some(v) => {
                    values.push(v);
                    validity.push(true);
                    stats.update_i64(v);
                }
                None => {
                    values.push(0); // Placeholder
                    validity.push(false);
                    stats.update_null();
                }
            }
        }
    }

    /// Push a u64 value
    pub fn push_u64(&mut self, value: Option<u64>) {
        if let TypedColumn::UInt64 {
            values,
            validity,
            stats,
        } = self
        {
            match value {
                Some(v) => {
                    values.push(v);
                    validity.push(true);
                    stats.update_i64(v as i64);
                }
                None => {
                    values.push(0);
                    validity.push(false);
                    stats.update_null();
                }
            }
        }
    }

    /// Push an f64 value
    pub fn push_f64(&mut self, value: Option<f64>) {
        if let TypedColumn::Float64 {
            values,
            validity,
            stats,
        } = self
        {
            match value {
                Some(v) => {
                    values.push(v);
                    validity.push(true);
                    stats.update_f64(v);
                }
                None => {
                    values.push(0.0);
                    validity.push(false);
                    stats.update_null();
                }
            }
        }
    }

    /// Push a string value
    pub fn push_text(&mut self, value: Option<&str>) {
        if let TypedColumn::Text {
            offsets,
            data,
            validity,
            stats,
        } = self
        {
            match value {
                Some(s) => {
                    data.extend_from_slice(s.as_bytes());
                    offsets.push(data.len() as u32);
                    validity.push(true);
                    stats.row_count += 1;
                }
                None => {
                    offsets.push(data.len() as u32);
                    validity.push(false);
                    stats.update_null();
                }
            }
        }
    }

    /// Push a binary value
    pub fn push_binary(&mut self, value: Option<&[u8]>) {
        if let TypedColumn::Binary {
            offsets,
            data,
            validity,
            stats,
        } = self
        {
            match value {
                Some(b) => {
                    data.extend_from_slice(b);
                    offsets.push(data.len() as u32);
                    validity.push(true);
                    stats.row_count += 1;
                }
                None => {
                    offsets.push(data.len() as u32);
                    validity.push(false);
                    stats.update_null();
                }
            }
        }
    }

    /// Push a boolean value
    pub fn push_bool(&mut self, value: Option<bool>) {
        if let TypedColumn::Bool {
            values,
            validity,
            stats,
            len,
        } = self
        {
            let idx = *len;
            *len += 1;
            let num_words = (*len).div_ceil(64);
            while values.len() < num_words {
                values.push(0);
            }
            match value {
                Some(v) => {
                    if v {
                        let word = idx / 64;
                        let bit = idx % 64;
                        values[word] |= 1 << bit;
                    }
                    validity.push(true);
                    stats.row_count += 1;
                }
                None => {
                    validity.push(false);
                    stats.update_null();
                }
            }
        }
    }

    /// Get an i64 value at index
    pub fn get_i64(&self, idx: usize) -> Option<i64> {
        if let TypedColumn::Int64 {
            values, validity, ..
        } = self
            && idx < values.len()
            && validity.is_valid(idx)
        {
            return Some(values[idx]);
        }
        None
    }

    /// Get a u64 value at index
    pub fn get_u64(&self, idx: usize) -> Option<u64> {
        if let TypedColumn::UInt64 {
            values, validity, ..
        } = self
            && idx < values.len()
            && validity.is_valid(idx)
        {
            return Some(values[idx]);
        }
        None
    }

    /// Get an f64 value at index
    pub fn get_f64(&self, idx: usize) -> Option<f64> {
        if let TypedColumn::Float64 {
            values, validity, ..
        } = self
            && idx < values.len()
            && validity.is_valid(idx)
        {
            return Some(values[idx]);
        }
        None
    }

    /// Get a string value at index
    pub fn get_text(&self, idx: usize) -> Option<&str> {
        if let TypedColumn::Text {
            offsets,
            data,
            validity,
            ..
        } = self
            && idx + 1 < offsets.len()
            && validity.is_valid(idx)
        {
            let start = offsets[idx] as usize;
            let end = offsets[idx + 1] as usize;
            return std::str::from_utf8(&data[start..end]).ok();
        }
        None
    }

    /// Get a binary value at index
    pub fn get_binary(&self, idx: usize) -> Option<&[u8]> {
        if let TypedColumn::Binary {
            offsets,
            data,
            validity,
            ..
        } = self
            && idx + 1 < offsets.len()
            && validity.is_valid(idx)
        {
            let start = offsets[idx] as usize;
            let end = offsets[idx + 1] as usize;
            return Some(&data[start..end]);
        }
        None
    }

    /// Get a boolean value at index
    pub fn get_bool(&self, idx: usize) -> Option<bool> {
        if let TypedColumn::Bool {
            values,
            validity,
            len,
            ..
        } = self
            && idx < *len
            && validity.is_valid(idx)
        {
            let word = idx / 64;
            let bit = idx % 64;
            return Some((values[word] >> bit) & 1 == 1);
        }
        None
    }

    /// Check if value at index is null
    pub fn is_null(&self, idx: usize) -> bool {
        match self {
            TypedColumn::Int64 { validity, .. } => !validity.is_valid(idx),
            TypedColumn::UInt64 { validity, .. } => !validity.is_valid(idx),
            TypedColumn::Float64 { validity, .. } => !validity.is_valid(idx),
            TypedColumn::Text { validity, .. } => !validity.is_valid(idx),
            TypedColumn::Binary { validity, .. } => !validity.is_valid(idx),
            TypedColumn::Bool { validity, .. } => !validity.is_valid(idx),
        }
    }

    /// Get column statistics
    pub fn stats(&self) -> &ColumnStats {
        match self {
            TypedColumn::Int64 { stats, .. } => stats,
            TypedColumn::UInt64 { stats, .. } => stats,
            TypedColumn::Float64 { stats, .. } => stats,
            TypedColumn::Text { stats, .. } => stats,
            TypedColumn::Binary { stats, .. } => stats,
            TypedColumn::Bool { stats, .. } => stats,
        }
    }

    /// SIMD-optimized sum for Int64 columns
    #[inline]
    pub fn sum_i64(&self) -> i64 {
        if let TypedColumn::Int64 {
            values, validity, ..
        } = self
        {
            // Fast path: no nulls - pure SIMD
            if validity.null_count() == 0 {
                values.iter().sum()
            } else {
                // Slow path: check validity
                values
                    .iter()
                    .enumerate()
                    .filter(|(i, _)| validity.is_valid(*i))
                    .map(|(_, v)| *v)
                    .sum()
            }
        } else {
            0
        }
    }

    /// SIMD-optimized sum for Float64 columns
    #[inline]
    pub fn sum_f64(&self) -> f64 {
        if let TypedColumn::Float64 {
            values, validity, ..
        } = self
        {
            if validity.null_count() == 0 {
                values.iter().sum()
            } else {
                values
                    .iter()
                    .enumerate()
                    .filter(|(i, _)| validity.is_valid(*i))
                    .map(|(_, v)| *v)
                    .sum()
            }
        } else {
            0.0
        }
    }

    /// Memory size in bytes
    pub fn memory_size(&self) -> usize {
        match self {
            TypedColumn::Int64 {
                values, validity, ..
            } => values.len() * 8 + validity.bits.len() * 8,
            TypedColumn::UInt64 {
                values, validity, ..
            } => values.len() * 8 + validity.bits.len() * 8,
            TypedColumn::Float64 {
                values, validity, ..
            } => values.len() * 8 + validity.bits.len() * 8,
            TypedColumn::Text {
                offsets,
                data,
                validity,
                ..
            } => offsets.len() * 4 + data.len() + validity.bits.len() * 8,
            TypedColumn::Binary {
                offsets,
                data,
                validity,
                ..
            } => offsets.len() * 4 + data.len() + validity.bits.len() * 8,
            TypedColumn::Bool {
                values, validity, ..
            } => values.len() * 8 + validity.bits.len() * 8,
        }
    }

    /// Extract value at row `idx` as a `SochValue`.
    ///
    /// Returns `SochValue::Null` for invalid (NULL) entries or out-of-bounds indices.
    /// This avoids the per-row `HashMap` overhead of the row-oriented `QueryResult`
    /// by materialising only the requested cell.
    pub fn value_at(&self, idx: usize) -> crate::SochValue {
        use crate::SochValue;
        match self {
            TypedColumn::Int64 {
                values, validity, ..
            } => {
                if idx < values.len() && validity.is_valid(idx) {
                    SochValue::Int(values[idx])
                } else {
                    SochValue::Null
                }
            }
            TypedColumn::UInt64 {
                values, validity, ..
            } => {
                if idx < values.len() && validity.is_valid(idx) {
                    SochValue::UInt(values[idx])
                } else {
                    SochValue::Null
                }
            }
            TypedColumn::Float64 {
                values, validity, ..
            } => {
                if idx < values.len() && validity.is_valid(idx) {
                    SochValue::Float(values[idx])
                } else {
                    SochValue::Null
                }
            }
            TypedColumn::Text {
                offsets,
                data,
                validity,
                ..
            } => {
                if idx + 1 < offsets.len() && validity.is_valid(idx) {
                    let start = offsets[idx] as usize;
                    let end = offsets[idx + 1] as usize;
                    std::str::from_utf8(&data[start..end])
                        .map(|s| SochValue::Text(s.to_owned()))
                        .unwrap_or(SochValue::Null)
                } else {
                    SochValue::Null
                }
            }
            TypedColumn::Binary {
                offsets,
                data,
                validity,
                ..
            } => {
                if idx + 1 < offsets.len() && validity.is_valid(idx) {
                    let start = offsets[idx] as usize;
                    let end = offsets[idx + 1] as usize;
                    SochValue::Binary(data[start..end].to_vec())
                } else {
                    SochValue::Null
                }
            }
            TypedColumn::Bool {
                values,
                validity,
                len,
                ..
            } => {
                if idx < *len && validity.is_valid(idx) {
                    let word = idx / 64;
                    let bit = idx % 64;
                    SochValue::Bool((values[word] >> bit) & 1 == 1)
                } else {
                    SochValue::Null
                }
            }
        }
    }
}

/// Column type enum for schema definition
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum ColumnType {
    Int64,
    UInt64,
    Float64,
    Text,
    Binary,
    Bool,
}

impl ColumnType {
    /// Create a new typed column for this type
    pub fn create_column(&self) -> TypedColumn {
        match self {
            ColumnType::Int64 => TypedColumn::new_int64(),
            ColumnType::UInt64 => TypedColumn::new_uint64(),
            ColumnType::Float64 => TypedColumn::new_float64(),
            ColumnType::Text => TypedColumn::new_text(),
            ColumnType::Binary => TypedColumn::new_binary(),
            ColumnType::Bool => TypedColumn::new_bool(),
        }
    }
}

/// Column chunk for cache-optimal processing
#[derive(Debug, Clone)]
pub struct ColumnChunk {
    /// Column name
    pub name: String,
    /// Column type
    pub column_type: ColumnType,
    /// Column data
    pub data: TypedColumn,
}

impl ColumnChunk {
    /// Create a new column chunk
    pub fn new(name: impl Into<String>, column_type: ColumnType) -> Self {
        Self {
            name: name.into(),
            column_type,
            data: column_type.create_column(),
        }
    }

    /// Get statistics for predicate pushdown
    pub fn stats(&self) -> &ColumnStats {
        self.data.stats()
    }
}

/// Arrow-compatible columnar table storage
#[derive(Debug)]
pub struct ColumnarTable {
    /// Table name
    pub name: String,
    /// Column definitions: name -> (type, column_data)
    columns: HashMap<String, ColumnChunk>,
    /// Column order for consistent iteration
    column_order: Vec<String>,
    /// Primary key column name
    primary_key: Option<String>,
    /// Primary key index: value -> row_index (for O(log N) lookups)
    pk_index: std::collections::BTreeMap<i64, u32>,
    /// Row count
    row_count: AtomicU64,
}

impl Clone for ColumnarTable {
    fn clone(&self) -> Self {
        Self {
            name: self.name.clone(),
            columns: self.columns.clone(),
            column_order: self.column_order.clone(),
            primary_key: self.primary_key.clone(),
            pk_index: self.pk_index.clone(),
            row_count: AtomicU64::new(self.row_count.load(std::sync::atomic::Ordering::Relaxed)),
        }
    }
}

impl ColumnarTable {
    /// Create a new columnar table
    pub fn new(name: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            columns: HashMap::new(),
            column_order: Vec::new(),
            primary_key: None,
            pk_index: std::collections::BTreeMap::new(),
            row_count: AtomicU64::new(0),
        }
    }

    /// Add a column to the table
    pub fn add_column(&mut self, name: impl Into<String>, column_type: ColumnType) -> &mut Self {
        let name = name.into();
        self.column_order.push(name.clone());
        self.columns
            .insert(name.clone(), ColumnChunk::new(name, column_type));
        self
    }

    /// Set the primary key column
    pub fn set_primary_key(&mut self, column: impl Into<String>) -> &mut Self {
        self.primary_key = Some(column.into());
        self
    }

    /// Get the number of rows
    pub fn row_count(&self) -> u64 {
        self.row_count.load(Ordering::Relaxed)
    }

    /// Get a column by name
    pub fn get_column(&self, name: &str) -> Option<&ColumnChunk> {
        self.columns.get(name)
    }

    /// Get a mutable column by name
    pub fn get_column_mut(&mut self, name: &str) -> Option<&mut ColumnChunk> {
        self.columns.get_mut(name)
    }

    /// Get row by primary key - O(log N) lookup
    pub fn get_by_pk(&self, pk: i64) -> Option<u32> {
        self.pk_index.get(&pk).copied()
    }

    /// Insert a row with values
    pub fn insert_row(&mut self, values: &HashMap<String, ColumnValue>) -> u32 {
        let row_idx = self.row_count.fetch_add(1, Ordering::Relaxed) as u32;

        for col_name in &self.column_order {
            let chunk = self.columns.get_mut(col_name).unwrap();
            let value = values.get(col_name);

            match &mut chunk.data {
                TypedColumn::Int64 {
                    values,
                    validity,
                    stats,
                } => {
                    match value {
                        Some(ColumnValue::Int64(v)) => {
                            values.push(*v);
                            validity.push(true);
                            stats.update_i64(*v);

                            // Update primary key index
                            if self.primary_key.as_ref() == Some(col_name) {
                                self.pk_index.insert(*v, row_idx);
                            }
                        }
                        _ => {
                            values.push(0);
                            validity.push(false);
                            stats.update_null();
                        }
                    }
                }
                TypedColumn::UInt64 {
                    values,
                    validity,
                    stats,
                } => match value {
                    Some(ColumnValue::UInt64(v)) => {
                        values.push(*v);
                        validity.push(true);
                        stats.update_i64(*v as i64);
                    }
                    _ => {
                        values.push(0);
                        validity.push(false);
                        stats.update_null();
                    }
                },
                TypedColumn::Float64 {
                    values,
                    validity,
                    stats,
                } => match value {
                    Some(ColumnValue::Float64(v)) => {
                        values.push(*v);
                        validity.push(true);
                        stats.update_f64(*v);
                    }
                    _ => {
                        values.push(0.0);
                        validity.push(false);
                        stats.update_null();
                    }
                },
                TypedColumn::Text {
                    offsets,
                    data,
                    validity,
                    stats,
                } => match value {
                    Some(ColumnValue::Text(s)) => {
                        data.extend_from_slice(s.as_bytes());
                        offsets.push(data.len() as u32);
                        validity.push(true);
                        stats.row_count += 1;
                    }
                    _ => {
                        offsets.push(data.len() as u32);
                        validity.push(false);
                        stats.update_null();
                    }
                },
                TypedColumn::Binary {
                    offsets,
                    data,
                    validity,
                    stats,
                } => match value {
                    Some(ColumnValue::Binary(b)) => {
                        data.extend_from_slice(b);
                        offsets.push(data.len() as u32);
                        validity.push(true);
                        stats.row_count += 1;
                    }
                    _ => {
                        offsets.push(data.len() as u32);
                        validity.push(false);
                        stats.update_null();
                    }
                },
                TypedColumn::Bool {
                    values,
                    validity,
                    stats,
                    len,
                } => {
                    let idx = *len;
                    *len += 1;
                    let num_words = (*len).div_ceil(64);
                    while values.len() < num_words {
                        values.push(0);
                    }
                    match value {
                        Some(ColumnValue::Bool(v)) => {
                            if *v {
                                let word = idx / 64;
                                let bit = idx % 64;
                                values[word] |= 1 << bit;
                            }
                            validity.push(true);
                            stats.row_count += 1;
                        }
                        _ => {
                            validity.push(false);
                            stats.update_null();
                        }
                    }
                }
            }
        }

        row_idx
    }

    /// Get total memory usage
    pub fn memory_size(&self) -> usize {
        self.columns.values().map(|c| c.data.memory_size()).sum()
    }

    /// Get memory usage comparison with enum-based storage
    pub fn memory_comparison(&self) -> MemoryComparison {
        let typed_size = self.memory_size();
        let row_count = self.row_count() as usize;
        let column_count = self.columns.len();

        // Enum-based storage: 32 bytes per value
        let enum_size = row_count * column_count * 32;

        MemoryComparison {
            typed_bytes: typed_size,
            enum_bytes: enum_size,
            savings_ratio: if typed_size > 0 {
                enum_size as f64 / typed_size as f64
            } else {
                1.0
            },
        }
    }
}

/// Memory comparison between typed and enum-based storage
#[derive(Debug, Clone)]
pub struct MemoryComparison {
    pub typed_bytes: usize,
    pub enum_bytes: usize,
    pub savings_ratio: f64,
}

/// Column value enum for insert operations (temporary)
#[derive(Debug, Clone)]
pub enum ColumnValue {
    Null,
    Int64(i64),
    UInt64(u64),
    Float64(f64),
    Text(String),
    Binary(Vec<u8>),
    Bool(bool),
}

/// Columnar store with multiple tables
#[derive(Debug, Default)]
pub struct ColumnarStore {
    /// Tables by name
    tables: HashMap<String, ColumnarTable>,
}

impl ColumnarStore {
    /// Create a new columnar store
    pub fn new() -> Self {
        Self {
            tables: HashMap::new(),
        }
    }

    /// Create a new table
    pub fn create_table(&mut self, name: impl Into<String>) -> &mut ColumnarTable {
        let name = name.into();
        self.tables
            .entry(name.clone())
            .or_insert_with(|| ColumnarTable::new(name))
    }

    /// Get a table by name
    pub fn get_table(&self, name: &str) -> Option<&ColumnarTable> {
        self.tables.get(name)
    }

    /// Get a mutable table by name
    pub fn get_table_mut(&mut self, name: &str) -> Option<&mut ColumnarTable> {
        self.tables.get_mut(name)
    }

    /// Drop a table
    pub fn drop_table(&mut self, name: &str) -> bool {
        self.tables.remove(name).is_some()
    }

    /// Get total memory usage
    pub fn memory_size(&self) -> usize {
        self.tables.values().map(|t| t.memory_size()).sum()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_validity_bitmap() {
        let mut bitmap = ValidityBitmap::new_all_valid(10);
        assert_eq!(bitmap.len(), 10);
        assert_eq!(bitmap.null_count(), 0);
        assert!(bitmap.is_valid(0));
        assert!(bitmap.is_valid(9));

        bitmap.set_null(5);
        assert_eq!(bitmap.null_count(), 1);
        assert!(!bitmap.is_valid(5));

        bitmap.set_valid(5);
        assert_eq!(bitmap.null_count(), 0);
        assert!(bitmap.is_valid(5));
    }

    #[test]
    fn test_int64_column() {
        let mut col = TypedColumn::new_int64();
        col.push_i64(Some(100));
        col.push_i64(Some(200));
        col.push_i64(None);
        col.push_i64(Some(300));

        assert_eq!(col.len(), 4);
        assert_eq!(col.get_i64(0), Some(100));
        assert_eq!(col.get_i64(1), Some(200));
        assert_eq!(col.get_i64(2), None);
        assert_eq!(col.get_i64(3), Some(300));
        assert!(col.is_null(2));

        assert_eq!(col.sum_i64(), 600);
    }

    #[test]
    fn test_text_column() {
        let mut col = TypedColumn::new_text();
        col.push_text(Some("hello"));
        col.push_text(Some("world"));
        col.push_text(None);
        col.push_text(Some("test"));

        assert_eq!(col.len(), 4);
        assert_eq!(col.get_text(0), Some("hello"));
        assert_eq!(col.get_text(1), Some("world"));
        assert_eq!(col.get_text(2), None);
        assert_eq!(col.get_text(3), Some("test"));
    }

    #[test]
    fn test_bool_column() {
        let mut col = TypedColumn::new_bool();
        col.push_bool(Some(true));
        col.push_bool(Some(false));
        col.push_bool(None);
        col.push_bool(Some(true));

        assert_eq!(col.len(), 4);
        assert_eq!(col.get_bool(0), Some(true));
        assert_eq!(col.get_bool(1), Some(false));
        assert_eq!(col.get_bool(2), None);
        assert_eq!(col.get_bool(3), Some(true));

        // Bool column uses ~2 bits per value vs 32 bytes for enum
        // 4 values = 8 bits = 1 byte vs 128 bytes
        assert!(col.memory_size() < 32);
    }

    #[test]
    fn test_columnar_table() {
        let mut table = ColumnarTable::new("users");
        table.add_column("id", ColumnType::Int64);
        table.add_column("name", ColumnType::Text);
        table.add_column("active", ColumnType::Bool);
        table.set_primary_key("id");

        let mut row1 = HashMap::new();
        row1.insert("id".to_string(), ColumnValue::Int64(1));
        row1.insert("name".to_string(), ColumnValue::Text("Alice".to_string()));
        row1.insert("active".to_string(), ColumnValue::Bool(true));
        table.insert_row(&row1);

        let mut row2 = HashMap::new();
        row2.insert("id".to_string(), ColumnValue::Int64(2));
        row2.insert("name".to_string(), ColumnValue::Text("Bob".to_string()));
        row2.insert("active".to_string(), ColumnValue::Bool(false));
        table.insert_row(&row2);

        assert_eq!(table.row_count(), 2);
        assert_eq!(table.get_by_pk(1), Some(0));
        assert_eq!(table.get_by_pk(2), Some(1));
        assert_eq!(table.get_by_pk(3), None);

        let id_col = table.get_column("id").unwrap();
        assert_eq!(id_col.data.get_i64(0), Some(1));
        assert_eq!(id_col.data.get_i64(1), Some(2));
    }

    #[test]
    fn test_memory_savings() {
        let mut table = ColumnarTable::new("test");
        table.add_column("id", ColumnType::Int64);
        table.add_column("value", ColumnType::Float64);
        table.add_column("flag", ColumnType::Bool);

        // Insert 1000 rows
        for i in 0..1000 {
            let mut row = HashMap::new();
            row.insert("id".to_string(), ColumnValue::Int64(i));
            row.insert("value".to_string(), ColumnValue::Float64(i as f64 * 1.5));
            row.insert("flag".to_string(), ColumnValue::Bool(i % 2 == 0));
            table.insert_row(&row);
        }

        let comparison = table.memory_comparison();

        // Typed storage should be significantly smaller than enum storage
        // Enum: 1000 rows * 3 columns * 32 bytes = 96,000 bytes
        // Typed: 1000 * (8 + 8 + 0.125) bytes ≈ 16,125 bytes
        assert!(
            comparison.savings_ratio > 3.0,
            "Expected 3x+ savings, got {:.2}x",
            comparison.savings_ratio
        );
    }

    #[test]
    fn test_simd_sum() {
        let mut col = TypedColumn::new_int64();
        for i in 0..10000 {
            col.push_i64(Some(i));
        }

        let sum = col.sum_i64();
        let expected: i64 = (0..10000).sum();
        assert_eq!(sum, expected);
    }

    #[test]
    fn test_columnar_store() {
        let mut store = ColumnarStore::new();

        {
            let table = store.create_table("users");
            table.add_column("id", ColumnType::Int64);
            table.add_column("name", ColumnType::Text);
        }

        assert!(store.get_table("users").is_some());
        assert!(store.get_table("orders").is_none());

        store.drop_table("users");
        assert!(store.get_table("users").is_none());
    }

    #[test]
    fn test_column_stats() {
        let mut col = TypedColumn::new_int64();
        col.push_i64(Some(10));
        col.push_i64(Some(50));
        col.push_i64(None);
        col.push_i64(Some(30));
        col.push_i64(Some(20));

        let stats = col.stats();
        assert_eq!(stats.min_i64, Some(10));
        assert_eq!(stats.max_i64, Some(50));
        assert_eq!(stats.null_count, 1);
        assert_eq!(stats.row_count, 5);
    }

    #[test]
    fn test_typed_column_value_at() {
        use crate::SochValue;

        // Int64
        let mut col = TypedColumn::new_int64();
        col.push_i64(Some(42));
        col.push_i64(None);
        col.push_i64(Some(-7));
        assert_eq!(col.value_at(0), SochValue::Int(42));
        assert_eq!(col.value_at(1), SochValue::Null);
        assert_eq!(col.value_at(2), SochValue::Int(-7));
        assert_eq!(col.value_at(99), SochValue::Null); // out of bounds

        // Float64
        let mut fcol = TypedColumn::new_float64();
        fcol.push_f64(Some(3.15));
        fcol.push_f64(None);
        assert_eq!(fcol.value_at(0), SochValue::Float(3.15));
        assert_eq!(fcol.value_at(1), SochValue::Null);

        // Text
        let mut tcol = TypedColumn::new_text();
        tcol.push_text(Some("hello"));
        tcol.push_text(None);
        tcol.push_text(Some("world"));
        assert_eq!(tcol.value_at(0), SochValue::Text("hello".to_string()));
        assert_eq!(tcol.value_at(1), SochValue::Null);
        assert_eq!(tcol.value_at(2), SochValue::Text("world".to_string()));

        // Bool
        let mut bcol = TypedColumn::new_bool();
        bcol.push_bool(Some(true));
        bcol.push_bool(Some(false));
        bcol.push_bool(None);
        assert_eq!(bcol.value_at(0), SochValue::Bool(true));
        assert_eq!(bcol.value_at(1), SochValue::Bool(false));
        assert_eq!(bcol.value_at(2), SochValue::Null);
    }
}