hamelin_eval 0.10.13

use crate::eval::environment::Environment;
use crate::eval::error::{EvalError, EvalResult};
use chrono::{DateTime, Datelike, Duration, FixedOffset, Timelike, Utc};
use chrono_tz::Tz;
use chronoutil::RelativeDuration;
use derive_more::derive::TryUnwrap;
use hamelin_lib::tree::ast::identifier::SimpleIdentifier;
use hamelin_lib::tree::typed_ast::expression::TypedLambda;
use linear_map::LinearMap;
use ordermap::OrderMap;
use serde_json;
use std::fmt::{self, Display, Formatter};
use std::rc::Rc;

/// Represents a timezone for timestamp operations
#[derive(Debug, Clone, PartialEq)]
pub enum TimeZone {
    /// Named timezone from the IANA timezone database (includes UTC as Tz::UTC)
    Named(Tz),
    /// Fixed offset from UTC (e.g., +05:30, -08:00) - for parsed timestamps with offset
    FixedOffset(FixedOffset),
    /// Unconstrained timezone - only valid in reverse evaluation constraints.
    /// Represents "any timezone with this instant" - the instant is constrained but
    /// the timezone representation is not.
    Any,
}

impl TimeZone {
    /// Create a UTC timezone
    pub fn utc() -> Self {
        TimeZone::Named(Tz::UTC)
    }

    /// Returns true if this is the Any variant
    pub fn is_any(&self) -> bool {
        matches!(self, TimeZone::Any)
    }

    /// Returns true if this is UTC
    pub fn is_utc(&self) -> bool {
        matches!(self, TimeZone::Named(tz) if *tz == Tz::UTC)
    }
}

/// Represents a timestamp value with timezone information
#[derive(Debug, Clone)]
pub struct TimestampValue {
    /// The actual instant in time (always stored as UTC internally)
    instant: DateTime<Utc>,
    /// The timezone for operations like truncation, extract, comparison
    timezone: TimeZone,
}

impl TimestampValue {
    /// Create a new TimestampValue with the given instant and timezone
    pub fn new(instant: DateTime<Utc>, timezone: TimeZone) -> Self {
        Self { instant, timezone }
    }

    /// Create a new TimestampValue in UTC
    pub fn utc(instant: DateTime<Utc>) -> Self {
        Self {
            instant,
            timezone: TimeZone::utc(),
        }
    }

    /// Create a new TimestampValue with a different timezone
    /// (same instant, different representation)
    pub fn with_timezone(self, timezone: TimeZone) -> Self {
        Self {
            instant: self.instant,
            timezone,
        }
    }

    /// Get the instant in time
    pub fn instant(&self) -> &DateTime<Utc> {
        &self.instant
    }

    /// Get the timezone
    pub fn timezone(&self) -> &TimeZone {
        &self.timezone
    }

    /// Convert to a DateTime<Tz> in a named timezone
    /// Returns an error if timezone is not Named or if timezone is TimeZone::Any
    pub fn to_datetime_tz(&self) -> EvalResult<chrono::DateTime<Tz>> {
        match &self.timezone {
            TimeZone::Named(tz) => Ok(self.instant.with_timezone(tz)),
            TimeZone::FixedOffset(_) => Err(EvalError::execution(
                "Cannot convert timestamp with fixed offset to named timezone DateTime".to_string(),
            )),
            TimeZone::Any => Err(EvalError::execution(
                "Cannot convert timestamp with unconstrained timezone (TimeZone::Any) to datetime"
                    .to_string(),
            )),
        }
    }

    /// Convert to a DateTime<FixedOffset> with a fixed offset
    /// Returns an error if timezone is not FixedOffset or if timezone is TimeZone::Any
    pub fn to_datetime_fixed(&self) -> EvalResult<chrono::DateTime<FixedOffset>> {
        match &self.timezone {
            TimeZone::Named(_) => Err(EvalError::execution(
                "Cannot convert timestamp with named timezone to FixedOffset DateTime".to_string(),
            )),
            TimeZone::FixedOffset(offset) => Ok(self.instant.with_timezone(offset)),
            TimeZone::Any => Err(EvalError::execution(
                "Cannot convert timestamp with unconstrained timezone (TimeZone::Any) to datetime"
                    .to_string(),
            )),
        }
    }

    // Component accessors - these convert to the timezone first, then extract the component
    // Following the pattern: "When using, first convert to the stored zone, then use"

    /// Get the year component in the timestamp's timezone
    pub fn year(&self) -> EvalResult<i32> {
        match &self.timezone {
            TimeZone::Named(tz) => Ok(self.instant.with_timezone(tz).year()),
            TimeZone::FixedOffset(offset) => Ok(self.instant.with_timezone(offset).year()),
            TimeZone::Any => Err(EvalError::execution(
                "Cannot extract year from timestamp with unconstrained timezone (TimeZone::Any)",
            )),
        }
    }

    /// Get the month component in the timestamp's timezone
    pub fn month(&self) -> EvalResult<u32> {
        match &self.timezone {
            TimeZone::Named(tz) => Ok(self.instant.with_timezone(tz).month()),
            TimeZone::FixedOffset(offset) => Ok(self.instant.with_timezone(offset).month()),
            TimeZone::Any => Err(EvalError::execution(
                "Cannot extract month from timestamp with unconstrained timezone (TimeZone::Any)",
            )),
        }
    }

    /// Get the day component in the timestamp's timezone
    pub fn day(&self) -> EvalResult<u32> {
        match &self.timezone {
            TimeZone::Named(tz) => Ok(self.instant.with_timezone(tz).day()),
            TimeZone::FixedOffset(offset) => Ok(self.instant.with_timezone(offset).day()),
            TimeZone::Any => Err(EvalError::execution(
                "Cannot extract day from timestamp with unconstrained timezone (TimeZone::Any)",
            )),
        }
    }

    /// Get the hour component in the timestamp's timezone
    pub fn hour(&self) -> EvalResult<u32> {
        match &self.timezone {
            TimeZone::Named(tz) => Ok(self.instant.with_timezone(tz).hour()),
            TimeZone::FixedOffset(offset) => Ok(self.instant.with_timezone(offset).hour()),
            TimeZone::Any => Err(EvalError::execution(
                "Cannot extract hour from timestamp with unconstrained timezone (TimeZone::Any)",
            )),
        }
    }

    /// Get the minute component in the timestamp's timezone
    pub fn minute(&self) -> EvalResult<u32> {
        match &self.timezone {
            TimeZone::Named(tz) => Ok(self.instant.with_timezone(tz).minute()),
            TimeZone::FixedOffset(offset) => Ok(self.instant.with_timezone(offset).minute()),
            TimeZone::Any => Err(EvalError::execution(
                "Cannot extract minute from timestamp with unconstrained timezone (TimeZone::Any)",
            )),
        }
    }

    /// Get the second component in the timestamp's timezone
    pub fn second(&self) -> EvalResult<u32> {
        match &self.timezone {
            TimeZone::Named(tz) => Ok(self.instant.with_timezone(tz).second()),
            TimeZone::FixedOffset(offset) => Ok(self.instant.with_timezone(offset).second()),
            TimeZone::Any => Err(EvalError::execution(
                "Cannot extract second from timestamp with unconstrained timezone (TimeZone::Any)",
            )),
        }
    }

    /// Get the weekday in the timestamp's timezone
    pub fn weekday(&self) -> EvalResult<chrono::Weekday> {
        match &self.timezone {
            TimeZone::Named(tz) => Ok(self.instant.with_timezone(tz).weekday()),
            TimeZone::FixedOffset(offset) => Ok(self.instant.with_timezone(offset).weekday()),
            TimeZone::Any => Err(EvalError::execution(
                "Cannot extract weekday from timestamp with unconstrained timezone (TimeZone::Any)",
            )),
        }
    }

    // Unix timestamp accessors - these are timezone-independent
    pub fn timestamp(&self) -> i64 {
        self.instant.timestamp()
    }

    pub fn timestamp_millis(&self) -> i64 {
        self.instant.timestamp_millis()
    }

    pub fn timestamp_micros(&self) -> i64 {
        self.instant.timestamp_micros()
    }

    pub fn timestamp_nanos_opt(&self) -> Option<i64> {
        self.instant.timestamp_nanos_opt()
    }

    pub fn timestamp_subsec_micros(&self) -> u32 {
        self.instant.timestamp_subsec_micros()
    }
}

/// Represents runtime values during expression evaluation
#[derive(Debug, Clone, PartialEq, TryUnwrap)]
pub enum Value {
    /// Binary data
    Binary(Vec<u8>),
    /// Boolean value
    Boolean(bool),
    /// Interval value (for time intervals)
    Interval(Duration),
    /// Calendar interval (number of months)
    CalendarInterval(i32),
    /// 64-bit integer
    Int(i64),
    /// 64-bit floating point
    Double(f64),
    /// Number of rows (for aggregate contexts)
    Rows(i64),
    /// UTF-8 string
    String(String),
    /// Timestamp with timezone information
    Timestamp(TimestampValue),
    /// Unknown/undefined value
    Unknown,
    /// Decimal value with precision and scale
    Decimal(DecimalValue),
    /// Array of values
    Array(Vec<Value>),
    /// Key-value map
    Map(LinearMap<Value, Value>),
    /// Tuple of values
    Tuple(Vec<Value>),
    /// Variant type (for union types)
    Variant(serde_json::Value),
    /// Range of values
    Range(Box<RangeValue>),
    /// Structured record with named fields
    Struct(OrderMap<SimpleIdentifier, Value>),
    /// Closure (lambda with captured environment)
    Closure(Closure),
    /// SQL NULL value
    Null,
}

/// A closure captures a lambda expression along with its environment.
#[derive(Debug, Clone)]
pub struct Closure {
    pub lambda: Rc<TypedLambda>,
    pub captured_env: Environment,
}

impl PartialEq for Closure {
    fn eq(&self, other: &Self) -> bool {
        // Closures use reference equality - two closures are equal only if
        // they point to the same lambda (by Rc pointer)
        Rc::ptr_eq(&self.lambda, &other.lambda)
    }
}

/// Represents a decimal value with arbitrary precision
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct DecimalValue {
    /// The unscaled value
    pub unscaled: i128,
    /// The scale (number of decimal places)
    pub scale: i32,
}

impl DecimalValue {
    /// Create a new DecimalValue
    pub fn new(unscaled: i128, scale: i32) -> Self {
        Self { unscaled, scale }
    }

    /// Add two decimal values
    pub fn add(&self, other: &DecimalValue) -> DecimalValue {
        use std::cmp::max;
        let common_scale = max(self.scale, other.scale);
        let left_scaled = self.unscaled * 10_i128.pow((common_scale - self.scale) as u32);
        let right_scaled = other.unscaled * 10_i128.pow((common_scale - other.scale) as u32);

        DecimalValue {
            unscaled: left_scaled + right_scaled,
            scale: common_scale,
        }
    }

    /// Subtract two decimal values
    pub fn sub(&self, other: &DecimalValue) -> DecimalValue {
        use std::cmp::max;
        let common_scale = max(self.scale, other.scale);
        let left_scaled = self.unscaled * 10_i128.pow((common_scale - self.scale) as u32);
        let right_scaled = other.unscaled * 10_i128.pow((common_scale - other.scale) as u32);

        DecimalValue {
            unscaled: left_scaled - right_scaled,
            scale: common_scale,
        }
    }

    /// Multiply two decimal values
    pub fn mul(&self, other: &DecimalValue) -> DecimalValue {
        DecimalValue {
            unscaled: self.unscaled * other.unscaled,
            scale: self.scale + other.scale,
        }
    }

    /// Divide two decimal values
    pub fn div(&self, other: &DecimalValue) -> anyhow::Result<DecimalValue> {
        if other.unscaled == 0 {
            anyhow::bail!("Division by zero");
        }

        // Scale up the dividend to maintain precision
        let scale_adjustment = 6; // Add 6 decimal places for precision
        let dividend = self.unscaled * 10_i128.pow(scale_adjustment);
        let result_scale = self.scale - other.scale + scale_adjustment as i32;

        Ok(DecimalValue {
            unscaled: dividend / other.unscaled,
            scale: result_scale,
        })
    }

    /// Compare two decimal values for equality
    pub fn eq(&self, other: &DecimalValue) -> bool {
        use std::cmp::max;
        let common_scale = max(self.scale, other.scale);
        let left_scaled = self.unscaled * 10_i128.pow((common_scale - self.scale) as u32);
        let right_scaled = other.unscaled * 10_i128.pow((common_scale - other.scale) as u32);
        left_scaled == right_scaled
    }

    /// Compare two decimal values for less than
    pub fn lt(&self, other: &DecimalValue) -> bool {
        use std::cmp::max;
        let common_scale = max(self.scale, other.scale);
        let left_scaled = self.unscaled * 10_i128.pow((common_scale - self.scale) as u32);
        let right_scaled = other.unscaled * 10_i128.pow((common_scale - other.scale) as u32);
        left_scaled < right_scaled
    }

    /// Compare two decimal values for less than or equal
    pub fn le(&self, other: &DecimalValue) -> bool {
        use std::cmp::max;
        let common_scale = max(self.scale, other.scale);
        let left_scaled = self.unscaled * 10_i128.pow((common_scale - self.scale) as u32);
        let right_scaled = other.unscaled * 10_i128.pow((common_scale - other.scale) as u32);
        left_scaled <= right_scaled
    }

    /// Compare two decimal values for greater than
    pub fn gt(&self, other: &DecimalValue) -> bool {
        use std::cmp::max;
        let common_scale = max(self.scale, other.scale);
        let left_scaled = self.unscaled * 10_i128.pow((common_scale - self.scale) as u32);
        let right_scaled = other.unscaled * 10_i128.pow((common_scale - other.scale) as u32);
        left_scaled > right_scaled
    }

    /// Compare two decimal values for greater than or equal
    pub fn ge(&self, other: &DecimalValue) -> bool {
        use std::cmp::max;
        let common_scale = max(self.scale, other.scale);
        let left_scaled = self.unscaled * 10_i128.pow((common_scale - self.scale) as u32);
        let right_scaled = other.unscaled * 10_i128.pow((common_scale - other.scale) as u32);
        left_scaled >= right_scaled
    }
}

impl From<i64> for DecimalValue {
    fn from(i: i64) -> Self {
        Self {
            unscaled: i as i128,
            scale: 0,
        }
    }
}

impl From<f64> for DecimalValue {
    fn from(f: f64) -> Self {
        // Handle special cases
        if !f.is_finite() {
            return Self {
                unscaled: 0,
                scale: 0,
            };
        }

        // Use a reasonable scale for floating point precision
        let scale = 6;
        let scale_factor = 10_f64.powi(scale);
        let unscaled = (f * scale_factor).round() as i128;

        Self { unscaled, scale }
    }
}

impl From<DecimalValue> for f64 {
    fn from(dec: DecimalValue) -> Self {
        let scale_factor = 10_f64.powi(dec.scale);
        dec.unscaled as f64 / scale_factor
    }
}

impl From<&DecimalValue> for f64 {
    fn from(dec: &DecimalValue) -> Self {
        let scale_factor = 10_f64.powi(dec.scale);
        dec.unscaled as f64 / scale_factor
    }
}

impl std::str::FromStr for DecimalValue {
    type Err = DecimalParseError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        let s = s.trim();
        if s.is_empty() {
            return Err(DecimalParseError::Empty);
        }

        // Handle optional sign
        let (is_negative, s) = match s.strip_prefix('-') {
            Some(rest) => (true, rest),
            None => (false, s.strip_prefix('+').unwrap_or(s)),
        };

        // Split on decimal point
        let (integer_part, fractional_part) = match s.split_once('.') {
            Some((int, frac)) => (int, frac),
            None => (s, ""),
        };

        // Require at least one digit in either part
        if integer_part.is_empty() && fractional_part.is_empty() {
            return Err(DecimalParseError::InvalidDigit);
        }

        // Parse integer part
        let integer_value: i128 = if integer_part.is_empty() {
            0
        } else {
            integer_part.parse().map_err(parse_int_error_to_decimal)?
        };

        // Parse fractional part
        let scale = fractional_part.len() as i32;
        let fractional_value: i128 = if fractional_part.is_empty() {
            0
        } else {
            fractional_part
                .parse()
                .map_err(parse_int_error_to_decimal)?
        };

        // Combine: unscaled = integer * 10^scale + fractional
        let scale_factor = 10_i128
            .checked_pow(scale as u32)
            .ok_or(DecimalParseError::Overflow)?;
        let unscaled = integer_value
            .checked_mul(scale_factor)
            .and_then(|v| v.checked_add(fractional_value))
            .ok_or(DecimalParseError::Overflow)?;

        let unscaled = if is_negative {
            unscaled.checked_neg().ok_or(DecimalParseError::Overflow)?
        } else {
            unscaled
        };

        Ok(DecimalValue { unscaled, scale })
    }
}

#[derive(Debug, Clone, thiserror::Error)]
pub enum DecimalParseError {
    #[error("empty string")]
    Empty,
    #[error("invalid digit in decimal")]
    InvalidDigit,
    #[error("decimal value overflow")]
    Overflow,
}

/// Convert a ParseIntError to the appropriate DecimalParseError variant
fn parse_int_error_to_decimal(e: std::num::ParseIntError) -> DecimalParseError {
    match e.kind() {
        std::num::IntErrorKind::PosOverflow | std::num::IntErrorKind::NegOverflow => {
            DecimalParseError::Overflow
        }
        _ => DecimalParseError::InvalidDigit,
    }
}

/// Represents a range of values
#[derive(Debug, Clone, PartialEq)]
pub struct RangeValue {
    /// Lower bound (inclusive if Some)
    pub lower: Option<Value>,
    /// Upper bound (exclusive if Some)
    pub upper: Option<Value>,
}

impl Display for Value {
    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Value::Binary(bytes) => {
                write!(f, "0x")?;
                for byte in bytes {
                    write!(f, "{:02x}", byte)?;
                }
                Ok(())
            }
            Value::Boolean(b) => write!(f, "{}", b),
            Value::Interval(i) => write!(f, "{}", i),
            Value::CalendarInterval(months) => {
                write!(
                    f,
                    "{}",
                    RelativeDuration::months(*months).format_to_iso8601()
                )
            }
            Value::Int(i) => write!(f, "{}", i),
            Value::Double(d) => write!(f, "{}", d),
            Value::Rows(r) => write!(f, "{} rows", r),
            Value::String(s) => write!(f, "'{}'", s),
            Value::Timestamp(t) => {
                // First convert the instant to the stored timezone, then format
                match &t.timezone {
                    TimeZone::Named(tz) => {
                        let dt_in_tz = t.instant.with_timezone(tz);
                        write!(f, "{}", dt_in_tz.to_rfc3339())
                    }
                    TimeZone::FixedOffset(offset) => {
                        let dt_with_offset = t.instant.with_timezone(offset);
                        write!(f, "{}", dt_with_offset.to_rfc3339())
                    }
                    TimeZone::Any => {
                        // Can't convert to a specific timezone, just show UTC with annotation
                        write!(f, "{} <any timezone>", t.instant.to_rfc3339())
                    }
                }
            }
            Value::Unknown => write!(f, "UNKNOWN"),
            Value::Decimal(d) => {
                let scale_factor = 10_i128.pow(d.scale as u32);
                let integer_part = d.unscaled / scale_factor;
                let fractional_part = (d.unscaled % scale_factor).abs();
                write!(
                    f,
                    "{}.{:0width$}",
                    integer_part,
                    fractional_part,
                    width = d.scale as usize
                )
            }
            Value::Array(values) => {
                write!(f, "[")?;
                for (i, v) in values.iter().enumerate() {
                    if i > 0 {
                        write!(f, ", ")?;
                    }
                    write!(f, "{}", v)?;
                }
                write!(f, "]")
            }
            Value::Map(map) => {
                write!(f, "{{")?;
                for (i, (k, v)) in map.iter().enumerate() {
                    if i > 0 {
                        write!(f, ", ")?;
                    }
                    write!(f, "{}: {}", k, v)?;
                }
                write!(f, "}}")
            }
            Value::Tuple(values) => {
                write!(f, "(")?;
                for (i, v) in values.iter().enumerate() {
                    if i > 0 {
                        write!(f, ", ")?;
                    }
                    write!(f, "{}", v)?;
                }
                write!(f, ")")
            }
            Value::Variant(v) => write!(f, "{}", v),
            Value::Range(r) => {
                if let Some(ref lower) = r.lower {
                    write!(f, "{}", lower)?;
                }
                write!(f, "..")?;
                if let Some(ref upper) = r.upper {
                    write!(f, "{}", upper)?;
                }
                Ok(())
            }
            Value::Struct(fields) => {
                write!(f, "{{")?;
                for (i, (k, v)) in fields.iter().enumerate() {
                    if i > 0 {
                        write!(f, ", ")?;
                    }
                    write!(f, "{}: {}", k, v)?;
                }
                write!(f, "}}")
            }
            Value::Closure(closure) => {
                let params: Vec<_> = closure
                    .lambda
                    .parameters
                    .iter()
                    .map(|p| format!("{}: {}", p.name.name(), p.typ))
                    .collect();
                let return_type = &closure.lambda.body.resolved_type;
                write!(f, "<closure({}) -> {}>", params.join(", "), return_type)
            }
            Value::Null => write!(f, "NULL"),
        }
    }
}

impl Eq for Value {}

impl Value {
    /// Check if this value is NULL
    pub fn is_null(&self) -> bool {
        matches!(self, Value::Null)
    }

    /// Extract an integer value, returning an error if not an integer
    pub fn try_int(&self) -> EvalResult<i64> {
        match self {
            Value::Int(i) => Ok(*i),
            _ => Err(EvalError::execution(format!(
                "Expected integer, got {}",
                self.type_name()
            ))),
        }
    }

    /// Extract a double value, returning an error if not a double
    pub fn try_double(&self) -> EvalResult<f64> {
        match self {
            Value::Double(d) => Ok(*d),
            _ => Err(EvalError::execution(format!(
                "Expected double, got {}",
                self.type_name()
            ))),
        }
    }

    /// Extract a string reference, returning an error if not a string
    pub fn try_string(&self) -> EvalResult<&String> {
        match self {
            Value::String(s) => Ok(s),
            _ => Err(EvalError::execution(format!(
                "Expected string, got {}",
                self.type_name()
            ))),
        }
    }

    /// Extract a boolean value, returning an error if not a boolean
    pub fn try_bool(&self) -> EvalResult<bool> {
        match self {
            Value::Boolean(b) => Ok(*b),
            _ => Err(EvalError::execution(format!(
                "Expected boolean, got {}",
                self.type_name()
            ))),
        }
    }

    /// Extract a struct reference, returning an error if not a struct
    pub fn try_struct(&self) -> EvalResult<&OrderMap<SimpleIdentifier, Value>> {
        match self {
            Value::Struct(s) => Ok(s),
            _ => Err(EvalError::execution(format!(
                "Expected struct, got {}",
                self.type_name()
            ))),
        }
    }

    /// Extract an array reference, returning an error if not an array
    pub fn try_array(&self) -> EvalResult<&Vec<Value>> {
        match self {
            Value::Array(a) => Ok(a),
            _ => Err(EvalError::execution(format!(
                "Expected array, got {}",
                self.type_name()
            ))),
        }
    }

    /// Extract a tuple reference, returning an error if not a tuple
    pub fn try_tuple(&self) -> EvalResult<&Vec<Value>> {
        match self {
            Value::Tuple(t) => Ok(t),
            _ => Err(EvalError::execution(format!(
                "Expected tuple, got {}",
                self.type_name()
            ))),
        }
    }

    /// Extract a map reference, returning an error if not a map
    pub fn try_map(&self) -> EvalResult<&LinearMap<Value, Value>> {
        match self {
            Value::Map(m) => Ok(m),
            _ => Err(EvalError::execution(format!(
                "Expected map, got {}",
                self.type_name()
            ))),
        }
    }

    /// Extract a timestamp value, returning an error if not a timestamp
    pub fn try_timestamp(&self) -> EvalResult<&TimestampValue> {
        match self {
            Value::Timestamp(t) => Ok(t),
            _ => Err(EvalError::execution(format!(
                "Expected timestamp, got {}",
                self.type_name()
            ))),
        }
    }

    /// Extract an interval value, returning an error if not an interval
    pub fn try_interval(&self) -> EvalResult<&Duration> {
        match self {
            Value::Interval(i) => Ok(i),
            _ => Err(EvalError::execution(format!(
                "Expected interval, got {}",
                self.type_name()
            ))),
        }
    }

    /// Extract a calendar interval value (number of months), returning an error if not a calendar interval
    pub fn try_calendar_interval(&self) -> EvalResult<i32> {
        match self {
            Value::CalendarInterval(months) => Ok(*months),
            _ => Err(EvalError::execution(format!(
                "Expected calendar interval, got {}",
                self.type_name()
            ))),
        }
    }

    /// Consume and extract a string, returning an error if not a string
    pub fn require_string(self) -> EvalResult<String> {
        match self {
            Value::String(s) => Ok(s),
            _ => Err(EvalError::execution(format!(
                "Expected string, got {}",
                self.type_name()
            ))),
        }
    }

    /// Consume and extract an integer value, returning an error if not an integer
    pub fn require_int(self) -> EvalResult<i64> {
        match self {
            Value::Int(i) => Ok(i),
            _ => Err(EvalError::execution(format!(
                "Expected integer, got {}",
                self.type_name()
            ))),
        }
    }

    /// Consume and extract a double value, returning an error if not a double
    pub fn require_double(self) -> EvalResult<f64> {
        match self {
            Value::Double(d) => Ok(d),
            _ => Err(EvalError::execution(format!(
                "Expected double, got {}",
                self.type_name()
            ))),
        }
    }

    /// Consume and extract a boolean value, returning an error if not a boolean
    pub fn require_bool(self) -> EvalResult<bool> {
        match self {
            Value::Boolean(b) => Ok(b),
            _ => Err(EvalError::execution(format!(
                "Expected boolean, got {}",
                self.type_name()
            ))),
        }
    }

    /// Consume and extract a timestamp value, returning an error if not a timestamp
    pub fn require_timestamp(self) -> EvalResult<TimestampValue> {
        match self {
            Value::Timestamp(t) => Ok(t),
            _ => Err(EvalError::execution(format!(
                "Expected timestamp, got {}",
                self.type_name()
            ))),
        }
    }

    /// Consume and extract an interval value, returning an error if not an interval
    pub fn require_interval(self) -> EvalResult<Duration> {
        match self {
            Value::Interval(i) => Ok(i),
            _ => Err(EvalError::execution(format!(
                "Expected interval, got {}",
                self.type_name()
            ))),
        }
    }

    /// Consume and extract a calendar interval value (number of months), returning an error if not a calendar interval
    pub fn require_calendar_interval(self) -> EvalResult<i32> {
        match self {
            Value::CalendarInterval(months) => Ok(months),
            _ => Err(EvalError::execution(format!(
                "Expected calendar interval, got {}",
                self.type_name()
            ))),
        }
    }

    /// Extract a variant value, returning an error if not a variant
    pub fn try_variant(&self) -> EvalResult<&serde_json::Value> {
        match self {
            Value::Variant(v) => Ok(v),
            _ => Err(EvalError::execution(format!(
                "Expected variant, got {}",
                self.type_name()
            ))),
        }
    }

    /// Consume and extract a variant value, returning an error if not a variant
    pub fn require_variant(self) -> EvalResult<serde_json::Value> {
        match self {
            Value::Variant(v) => Ok(v),
            _ => Err(EvalError::execution(format!(
                "Expected variant, got {}",
                self.type_name()
            ))),
        }
    }

    /// Consume and extract an array value, returning an error if not an array
    pub fn require_array(self) -> EvalResult<Vec<Value>> {
        match self {
            Value::Array(arr) => Ok(arr),
            _ => Err(EvalError::execution(format!(
                "Expected array, got {}",
                self.type_name()
            ))),
        }
    }

    /// Consume and extract a map value, returning an error if not a map
    pub fn require_map(self) -> EvalResult<LinearMap<Value, Value>> {
        match self {
            Value::Map(map) => Ok(map),
            _ => Err(EvalError::execution(format!(
                "Expected map, got {}",
                self.type_name()
            ))),
        }
    }

    /// Coerce this value to f64, supporting both Int and Double variants
    pub fn coerce_require_double(&self) -> EvalResult<f64> {
        match self {
            Value::Int(i) => Ok(*i as f64),
            Value::Double(d) => Ok(*d),
            _ => Err(EvalError::execution(format!(
                "Cannot coerce {} to double",
                self.type_name()
            ))),
        }
    }

    /// Get the type name of this value for error messages
    pub fn type_name(&self) -> &'static str {
        match self {
            Value::Binary(_) => "binary",
            Value::Boolean(_) => "boolean",
            Value::Interval(_) => "interval",
            Value::CalendarInterval(_) => "calendar_interval",
            Value::Int(_) => "int",
            Value::Double(_) => "double",
            Value::Rows(_) => "rows",
            Value::String(_) => "string",
            Value::Timestamp(_) => "timestamp",
            Value::Unknown => "unknown",
            Value::Decimal(_) => "decimal",
            Value::Array(_) => "array",
            Value::Map(_) => "map",
            Value::Tuple(_) => "tuple",
            Value::Variant(_) => "variant",
            Value::Range(_) => "range",
            Value::Struct(_) => "struct",
            Value::Closure(_) => "closure",
            Value::Null => "null",
        }
    }

    /// Extract a closure reference, returning an error if not a closure
    pub fn try_closure(&self) -> EvalResult<&Closure> {
        match self {
            Value::Closure(c) => Ok(c),
            _ => Err(EvalError::execution(format!(
                "Expected closure, got {}",
                self.type_name()
            ))),
        }
    }

    /// Consume and extract a closure, returning an error if not a closure
    pub fn require_closure(self) -> EvalResult<Closure> {
        match self {
            Value::Closure(c) => Ok(c),
            _ => Err(EvalError::execution(format!(
                "Expected closure, got {}",
                self.type_name()
            ))),
        }
    }

    /// Check if this value is a closure
    pub fn is_closure(&self) -> bool {
        matches!(self, Value::Closure(_))
    }

    /// Compare two values for greater than or equal (>=)
    /// Returns true if this value >= other value
    pub fn gte(&self, other: &Value) -> bool {
        match (self, other) {
            (Value::Int(l), Value::Int(r)) => l >= r,
            (Value::Double(l), Value::Double(r)) => l >= r,
            (Value::Int(l), Value::Double(r)) => (*l as f64) >= *r,
            (Value::Double(l), Value::Int(r)) => *l >= (*r as f64),
            (Value::Decimal(l), Value::Decimal(r)) => l.ge(r),
            (Value::Decimal(l), Value::Int(r)) => l.ge(&(*r).into()),
            (Value::Int(l), Value::Decimal(r)) => DecimalValue::from(*l).ge(r),
            (Value::Decimal(l), Value::Double(r)) => f64::from(l) >= *r,
            (Value::Double(l), Value::Decimal(r)) => *l >= f64::from(r),
            (Value::String(l), Value::String(r)) => l >= r,
            _ => false, // Can't compare other types
        }
    }

    /// Compare two values for less than or equal (<=)
    /// Returns true if this value <= other value
    pub fn lte(&self, other: &Value) -> bool {
        match (self, other) {
            (Value::Int(l), Value::Int(r)) => l <= r,
            (Value::Double(l), Value::Double(r)) => l <= r,
            (Value::Int(l), Value::Double(r)) => (*l as f64) <= *r,
            (Value::Double(l), Value::Int(r)) => *l <= (*r as f64),
            (Value::Decimal(l), Value::Decimal(r)) => l.le(r),
            (Value::Decimal(l), Value::Int(r)) => l.le(&(*r).into()),
            (Value::Int(l), Value::Decimal(r)) => DecimalValue::from(*l).le(r),
            (Value::Decimal(l), Value::Double(r)) => f64::from(l) <= *r,
            (Value::Double(l), Value::Decimal(r)) => *l <= f64::from(r),
            (Value::String(l), Value::String(r)) => l <= r,
            _ => false, // Can't compare other types
        }
    }

    /// Compare two values for greater than (>)
    /// Returns true if this value > other value
    pub fn gt(&self, other: &Value) -> bool {
        match (self, other) {
            (Value::Int(l), Value::Int(r)) => l > r,
            (Value::Double(l), Value::Double(r)) => l > r,
            (Value::Int(l), Value::Double(r)) => (*l as f64) > *r,
            (Value::Double(l), Value::Int(r)) => *l > (*r as f64),
            (Value::Decimal(l), Value::Decimal(r)) => l.gt(r),
            (Value::Decimal(l), Value::Int(r)) => l.gt(&(*r).into()),
            (Value::Int(l), Value::Decimal(r)) => DecimalValue::from(*l).gt(r),
            (Value::Decimal(l), Value::Double(r)) => f64::from(l) > *r,
            (Value::Double(l), Value::Decimal(r)) => *l > f64::from(r),
            (Value::String(l), Value::String(r)) => l > r,
            _ => false, // Can't compare other types
        }
    }

    /// Compare two values for less than (<)
    /// Returns true if this value < other value
    pub fn lt(&self, other: &Value) -> bool {
        match (self, other) {
            (Value::Int(l), Value::Int(r)) => l < r,
            (Value::Double(l), Value::Double(r)) => l < r,
            (Value::Int(l), Value::Double(r)) => (*l as f64) < *r,
            (Value::Double(l), Value::Int(r)) => *l < (*r as f64),
            (Value::Decimal(l), Value::Decimal(r)) => l.lt(r),
            (Value::Decimal(l), Value::Int(r)) => l.lt(&(*r).into()),
            (Value::Int(l), Value::Decimal(r)) => DecimalValue::from(*l).lt(r),
            (Value::Decimal(l), Value::Double(r)) => f64::from(l) < *r,
            (Value::Double(l), Value::Decimal(r)) => *l < f64::from(r),
            (Value::String(l), Value::String(r)) => l < r,
            _ => false, // Can't compare other types
        }
    }
}

impl From<TimestampValue> for Value {
    fn from(ts: TimestampValue) -> Self {
        Value::Timestamp(ts)
    }
}

// Implement arithmetic and comparison operations for TimestampValue
// These operate on the instant while preserving the timezone

impl std::ops::Add<Duration> for TimestampValue {
    type Output = TimestampValue;

    fn add(self, duration: Duration) -> TimestampValue {
        TimestampValue::new(self.instant + duration, self.timezone)
    }
}

impl std::ops::Add<Duration> for &TimestampValue {
    type Output = TimestampValue;

    fn add(self, duration: Duration) -> TimestampValue {
        TimestampValue::new(self.instant + duration, self.timezone.clone())
    }
}

impl std::ops::Sub<Duration> for TimestampValue {
    type Output = TimestampValue;

    fn sub(self, duration: Duration) -> TimestampValue {
        TimestampValue::new(self.instant - duration, self.timezone)
    }
}

impl std::ops::Sub<Duration> for &TimestampValue {
    type Output = TimestampValue;

    fn sub(self, duration: Duration) -> TimestampValue {
        TimestampValue::new(self.instant - duration, self.timezone.clone())
    }
}

impl std::ops::Sub<TimestampValue> for TimestampValue {
    type Output = Duration;

    fn sub(self, other: TimestampValue) -> Duration {
        self.instant - other.instant
    }
}

impl std::ops::Sub<&TimestampValue> for &TimestampValue {
    type Output = Duration;

    fn sub(self, other: &TimestampValue) -> Duration {
        self.instant - other.instant
    }
}

impl std::ops::Add<RelativeDuration> for TimestampValue {
    type Output = TimestampValue;

    fn add(self, duration: RelativeDuration) -> TimestampValue {
        TimestampValue::new(self.instant + duration, self.timezone)
    }
}

impl std::ops::Add<RelativeDuration> for &TimestampValue {
    type Output = TimestampValue;

    fn add(self, duration: RelativeDuration) -> TimestampValue {
        TimestampValue::new(self.instant + duration, self.timezone.clone())
    }
}

impl PartialEq for TimestampValue {
    fn eq(&self, other: &Self) -> bool {
        self.instant == other.instant
    }
}

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

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

    mod decimal_value_from_str {
        use super::*;

        #[test]
        fn parses_positive_integer() {
            let dec: DecimalValue = "123".parse().unwrap();
            assert_eq!(dec.unscaled, 123);
            assert_eq!(dec.scale, 0);
        }

        #[test]
        fn parses_negative_integer() {
            let dec: DecimalValue = "-456".parse().unwrap();
            assert_eq!(dec.unscaled, -456);
            assert_eq!(dec.scale, 0);
        }

        #[test]
        fn parses_decimal_with_fractional() {
            let dec: DecimalValue = "3.14".parse().unwrap();
            assert_eq!(dec.unscaled, 314);
            assert_eq!(dec.scale, 2);
        }

        #[test]
        fn parses_negative_decimal() {
            let dec: DecimalValue = "-3.14".parse().unwrap();
            assert_eq!(dec.unscaled, -314);
            assert_eq!(dec.scale, 2);
        }

        #[test]
        fn parses_explicit_positive_sign() {
            let dec: DecimalValue = "+42".parse().unwrap();
            assert_eq!(dec.unscaled, 42);
            assert_eq!(dec.scale, 0);
        }

        #[test]
        fn parses_leading_decimal() {
            let dec: DecimalValue = ".5".parse().unwrap();
            assert_eq!(dec.unscaled, 5);
            assert_eq!(dec.scale, 1);
        }

        #[test]
        fn parses_trailing_decimal() {
            let dec: DecimalValue = "5.".parse().unwrap();
            assert_eq!(dec.unscaled, 5);
            assert_eq!(dec.scale, 0);
        }

        #[test]
        fn rejects_empty_string() {
            let result: Result<DecimalValue, _> = "".parse();
            assert!(matches!(result, Err(DecimalParseError::Empty)));
        }

        #[test]
        fn rejects_whitespace_only() {
            let result: Result<DecimalValue, _> = "   ".parse();
            assert!(matches!(result, Err(DecimalParseError::Empty)));
        }

        #[test]
        fn rejects_sign_only_plus() {
            let result: Result<DecimalValue, _> = "+".parse();
            assert!(matches!(result, Err(DecimalParseError::InvalidDigit)));
        }

        #[test]
        fn rejects_sign_only_minus() {
            let result: Result<DecimalValue, _> = "-".parse();
            assert!(matches!(result, Err(DecimalParseError::InvalidDigit)));
        }

        #[test]
        fn rejects_dot_only() {
            let result: Result<DecimalValue, _> = ".".parse();
            assert!(matches!(result, Err(DecimalParseError::InvalidDigit)));
        }

        #[test]
        fn rejects_sign_and_dot_only() {
            let result: Result<DecimalValue, _> = "+.".parse();
            assert!(matches!(result, Err(DecimalParseError::InvalidDigit)));

            let result: Result<DecimalValue, _> = "-.".parse();
            assert!(matches!(result, Err(DecimalParseError::InvalidDigit)));
        }

        #[test]
        fn rejects_non_numeric() {
            let result: Result<DecimalValue, _> = "abc".parse();
            assert!(matches!(result, Err(DecimalParseError::InvalidDigit)));
        }

        #[test]
        fn rejects_excessive_scale() {
            // 10^39 overflows i128, so 39+ decimal places should return Overflow
            let input = format!("1.{}", "0".repeat(39));
            let result: Result<DecimalValue, _> = input.parse();
            assert!(matches!(result, Err(DecimalParseError::Overflow)));
        }

        #[test]
        fn parses_i128_max() {
            let dec: DecimalValue = i128::MAX.to_string().parse().unwrap();
            assert_eq!(dec.unscaled, i128::MAX);
            assert_eq!(dec.scale, 0);
        }

        #[test]
        fn parses_negative_i128_max() {
            // -i128::MAX is valid (magnitude fits in i128, negation is safe)
            let input = format!("-{}", i128::MAX);
            let dec: DecimalValue = input.parse().unwrap();
            assert_eq!(dec.unscaled, -i128::MAX);
            assert_eq!(dec.scale, 0);
        }

        #[test]
        fn rejects_i128_min_magnitude() {
            // i128::MIN's magnitude (170141183460469231731687303715884105728) exceeds i128::MAX,
            // so it can't be parsed as the absolute value first - returns Overflow, not InvalidDigit.
            let result: Result<DecimalValue, _> = i128::MIN.to_string().parse();
            assert!(matches!(result, Err(DecimalParseError::Overflow)));
        }
    }
}