motedb 0.8.0

//! Multi-modal data types for MoteDB

mod spatial;
mod table;
mod tensor;
mod text;
mod timestamp;

pub use spatial::{BoundingBox, BoundingBox3D, Geometry, Point, Point3D};
pub use table::{ColumnDef, ColumnType, IndexDef, IndexType, TTLDuration, TableSchema, TableType};
pub use tensor::Tensor;
pub use text::{Text, TextDoc};
pub use timestamp::Timestamp;

use serde::{Deserialize, Deserializer, Serialize, Serializer};
use std::sync::Arc;

/// Wrapper for Arc<Vec<f32>> with custom serde implementation
#[derive(Debug, Clone, PartialEq)]
pub struct ArcVec(pub Arc<Vec<f32>>);

impl ArcVec {
    pub fn new(vec: Vec<f32>) -> Self {
        ArcVec(Arc::new(vec))
    }

    pub fn len(&self) -> usize {
        self.0.len()
    }

    pub fn is_empty(&self) -> bool {
        self.0.is_empty()
    }

    pub fn iter(&self) -> std::slice::Iter<'_, f32> {
        self.0.iter()
    }

    pub fn to_vec(&self) -> Vec<f32> {
        (*self.0).clone()
    }

    pub fn as_slice(&self) -> &[f32] {
        self.0.as_ref()
    }
}

impl Serialize for ArcVec {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        self.0.as_ref().serialize(serializer)
    }
}

impl<'de> Deserialize<'de> for ArcVec {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let vec = Vec::<f32>::deserialize(deserializer)?;
        Ok(ArcVec(Arc::new(vec)))
    }
}

/// Arc-wrapped str for cheap cloning in Value::Text.
/// Uses `Arc<str>` (1 allocation) instead of `Arc<String>` (2 allocations).
#[derive(Debug, Clone, Eq, PartialEq, Hash)]
pub struct ArcString(pub Arc<str>);

impl ArcString {
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl std::ops::Deref for ArcString {
    type Target = str;
    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl Serialize for ArcString {
    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        self.0.serialize(serializer)
    }
}

impl<'de> Deserialize<'de> for ArcString {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        let s = String::deserialize(deserializer)?;
        Ok(ArcString(Arc::from(s)))
    }
}

impl std::fmt::Display for ArcString {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        self.0.fmt(f)
    }
}

impl PartialOrd for ArcString {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        self.0.partial_cmp(&other.0)
    }
}

impl From<String> for ArcString {
    fn from(s: String) -> Self {
        ArcString(Arc::from(s))
    }
}

impl From<&str> for ArcString {
    fn from(s: &str) -> Self {
        ArcString(Arc::from(s))
    }
}

impl PartialEq<String> for ArcString {
    fn eq(&self, other: &String) -> bool {
        &*self.0 == other.as_str()
    }
}

impl PartialEq<&str> for ArcString {
    fn eq(&self, other: &&str) -> bool {
        &*self.0 == *other
    }
}

/// Unified value type supporting all data modalities
///
/// Size optimization: large variants (Text, Tensor, Spatial, TextDoc) are
/// boxed to keep the enum at 16 bytes instead of 40 bytes. This reduces
/// memory amplification for scalar-heavy rows from 8.6x to ~3.5x.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum Value {
    /// Integer value
    Integer(i64),

    /// Floating point value
    Float(f64),

    /// Boolean value
    Bool(bool),

    /// Text string (Arc for cheap cloning — just atomic increment)
    Text(ArcString),

    /// Vector data (for embeddings)
    Vector(ArcVec),

    /// Tensor data (boxed to reduce enum size)
    Tensor(Box<Tensor>),

    /// Spatial geometry data (boxed to reduce enum size)
    Spatial(Box<Geometry>),

    /// Text document for full-text search (boxed to reduce enum size)
    TextDoc(Box<Text>),

    /// Timestamp data
    Timestamp(Timestamp),

    /// Null value
    Null,
}

/// Precise integer-vs-float comparison that avoids precision loss for |i| > 2^53.
///
/// Strategy:
/// - If the float has a fractional part, compare via f64 (safe: integer is exact in f64 range).
/// - If the float is an exact integer, compare via i64 arithmetic (avoids f64 precision loss).
fn int_float_cmp(i: i64, f: f64) -> Option<std::cmp::Ordering> {
    // If f has a fractional part, the integer can be compared as f64
    // because the fractional part differentiates them regardless.
    let f_trunc = f.trunc();
    if f != f_trunc {
        return (i as f64).partial_cmp(&f);
    }
    // f is an exact integer. Convert to i64 and compare precisely.
    // f.trunc() is guaranteed to be in i64 range because it came from a valid f64
    // that equals an integer value (no overflow since f64 can represent up to 2^1023).
    if f_trunc >= i64::MIN as f64 && f_trunc <= i64::MAX as f64 {
        let f_as_i = f_trunc as i64;
        return Some(i.cmp(&f_as_i));
    }
    // f is astronomically large/small — fall back to f64 comparison
    (i as f64).partial_cmp(&f)
}

impl PartialOrd for Value {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        match (self, other) {
            (Value::Null, Value::Null) => Some(std::cmp::Ordering::Equal),
            (Value::Null, _) => Some(std::cmp::Ordering::Less),
            (_, Value::Null) => Some(std::cmp::Ordering::Greater),
            (Value::Integer(a), Value::Integer(b)) => a.partial_cmp(b),
            (Value::Float(a), Value::Float(b)) => a.partial_cmp(b),
            (Value::Text(a), Value::Text(b)) => a.partial_cmp(b),
            (Value::Bool(a), Value::Bool(b)) => a.partial_cmp(b),
            (Value::Timestamp(a), Value::Timestamp(b)) => a.partial_cmp(b),
            (Value::Integer(a), Value::Float(b)) => int_float_cmp(*a, *b),
            (Value::Float(a), Value::Integer(b)) => int_float_cmp(*b, *a).map(|o| o.reverse()),
            // Timestamp vs Integer: compare timestamp micros to integer value
            (Value::Timestamp(a), Value::Integer(b)) => a.as_micros().partial_cmp(b),
            (Value::Integer(a), Value::Timestamp(b)) => a.partial_cmp(&b.as_micros()),
            // Timestamp vs Float: compare timestamp micros to float value
            (Value::Timestamp(a), Value::Float(b)) => int_float_cmp(a.as_micros(), *b),
            (Value::Float(a), Value::Timestamp(b)) => {
                int_float_cmp(b.as_micros(), *a).map(|o| o.reverse())
            }
            // 🚨 Timestamp vs Text: parse the text as an ISO date/datetime and
            // compare. Without this, `WHERE ts > '2024-01-01'` always returned
            // empty (Text falls through to None → comparison fails). This makes
            // date-range queries on TIMESTAMP columns work correctly.
            (Value::Timestamp(a), Value::Text(s)) => crate::types::Timestamp::parse_iso(s.as_str())
                .and_then(|t| a.as_micros().partial_cmp(&t.as_micros())),
            (Value::Text(s), Value::Timestamp(b)) => crate::types::Timestamp::parse_iso(s.as_str())
                .and_then(|t| t.as_micros().partial_cmp(&b.as_micros())),
            _ => None,
        }
    }
}

/// Total equality for f64: NaN == NaN, -0.0 == 0.0, otherwise bit-equality.
fn float_eq(a: f64, b: f64) -> bool {
    if a.is_nan() && b.is_nan() {
        return true;
    }
    if a == 0.0 && b == 0.0 {
        return true;
    }
    a.to_bits() == b.to_bits()
}

/// Canonical f64 bits for hashing: normalizes NaN to a single representation, -0.0 → 0.0.
fn canonical_float_bits(f: f64) -> u64 {
    if f.is_nan() {
        return u64::MAX;
    }
    if f == 0.0 {
        return 0.0f64.to_bits();
    }
    f.to_bits()
}

/// Precise integer-vs-float equality that avoids precision loss for |i| > 2^53.
fn int_float_eq(i: i64, f: f64) -> bool {
    let f_trunc = f.trunc();
    if f != f_trunc {
        return float_eq(i as f64, f);
    }
    if f_trunc >= i64::MIN as f64 && f_trunc <= i64::MAX as f64 {
        return i == f_trunc as i64;
    }
    float_eq(i as f64, f)
}

impl PartialEq for Value {
    fn eq(&self, other: &Self) -> bool {
        match (self, other) {
            (Value::Integer(a), Value::Integer(b)) => a == b,
            (Value::Float(a), Value::Float(b)) => float_eq(*a, *b),
            (Value::Text(a), Value::Text(b)) => a == b,
            (Value::Bool(a), Value::Bool(b)) => a == b,
            (Value::Null, Value::Null) => true,
            (Value::Timestamp(a), Value::Timestamp(b)) => a == b,
            (Value::Integer(a), Value::Float(b)) => int_float_eq(*a, *b),
            (Value::Float(a), Value::Integer(b)) => int_float_eq(*b, *a),
            (Value::Timestamp(a), Value::Integer(b)) => a.as_micros() == *b,
            (Value::Integer(a), Value::Timestamp(b)) => *a == b.as_micros(),
            (Value::Timestamp(a), Value::Float(b)) => float_eq(a.as_micros() as f64, *b),
            (Value::Float(a), Value::Timestamp(b)) => float_eq(*a, b.as_micros() as f64),
            // 🚨 Timestamp vs Text: parse text as ISO date and compare (so
            // `WHERE ts = '2024-01-15'` works on TIMESTAMP columns).
            (Value::Timestamp(a), Value::Text(s)) => crate::types::Timestamp::parse_iso(s.as_str())
                .is_some_and(|t| a.as_micros() == t.as_micros()),
            (Value::Text(s), Value::Timestamp(b)) => crate::types::Timestamp::parse_iso(s.as_str())
                .is_some_and(|t| t.as_micros() == b.as_micros()),
            _ => false,
        }
    }
}
impl Eq for Value {}

impl std::hash::Hash for Value {
    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
        match self {
            Value::Integer(i) => {
                let f = *i as f64;
                state.write_u8(0); // numeric discriminant
                canonical_float_bits(f).hash(state);
            }
            Value::Float(f) => {
                state.write_u8(0); // numeric discriminant
                canonical_float_bits(*f).hash(state);
            }
            Value::Text(s) => {
                state.write_u8(1);
                s.hash(state);
            }
            Value::Bool(b) => {
                state.write_u8(2);
                b.hash(state);
            }
            Value::Null => {
                state.write_u8(3);
            }
            Value::Timestamp(t) => {
                let micros = t.as_micros();
                let f = micros as f64;
                state.write_u8(0); // numeric discriminant (same as Integer/Float)
                canonical_float_bits(f).hash(state);
            }
            other => {
                state.write_u8(5);
                format!("{:?}", other).hash(state);
            }
        }
    }
}

impl Value {
    /// Create a Text value from a String (1 allocation via Arc<str>).
    pub fn text(s: String) -> Self {
        Value::Text(ArcString(Arc::from(s)))
    }

    /// Create a Text value from a &str (1 allocation via Arc<str>).
    pub fn text_from(s: &str) -> Self {
        Value::Text(ArcString(Arc::from(s)))
    }

    /// Create a Tensor value
    pub fn tensor(t: Tensor) -> Self {
        Value::Tensor(Box::new(t))
    }

    /// Create a Spatial value
    pub fn spatial(g: Geometry) -> Self {
        Value::Spatial(Box::new(g))
    }

    /// Create a TextDoc value
    pub fn textdoc(t: Text) -> Self {
        Value::TextDoc(Box::new(t))
    }

    /// Convert to a hashable string key for use in HashMap/DashMap lookups.
    /// Handles f64 by converting to bits (lossless).
    pub fn to_hash_key(&self) -> String {
        match self {
            Value::Integer(i) => format!("i:{}", i),
            Value::Float(f) => format!("f:{}", f.to_bits()),
            Value::Text(s) => format!("t:{}", s),
            Value::Bool(b) => format!("b:{}", b),
            Value::Timestamp(t) => format!("ts:{}", t.as_micros()),
            _ => format!("{:?}", self),
        }
    }
}

/// A row contains multiple values (for storage engine)
pub type Row = Vec<Value>;

/// A SQL row contains named values (for SQL engine)
pub type SqlRow = std::collections::HashMap<String, Value>;

/// Row identifier (unique across the database)
pub type RowId = u64;

/// Partition identifier for parallel writes
pub type PartitionId = u8;