use crate::core::{
    entities::{
        properties::prop::{prop_array::*, prop_ref_enum::PropRef, ArrowRow, PropNum, PropType},
        GidRef,
    },
    storage::arc_str::ArcStr,
};
use arrow_array::{
    cast::AsArray,
    types::{
        Date32Type, Date64Type, Decimal128Type, DecimalType, Float32Type, Float64Type, Int32Type,
        Int64Type, TimestampMicrosecondType, TimestampMillisecondType, TimestampNanosecondType,
        TimestampSecondType, UInt16Type, UInt32Type, UInt64Type, UInt8Type,
    },
    Array, ArrayRef, LargeListArray, StructArray,
};
use arrow_schema::{DataType, Field, FieldRef, Fields, TimeUnit, DECIMAL128_MAX_PRECISION};
use bigdecimal::{num_bigint::BigInt, BigDecimal};
use chrono::{DateTime, NaiveDateTime, Utc};
use itertools::Itertools;
use num_traits::{Bounded, FromPrimitive, ToPrimitive};
use rustc_hash::{FxBuildHasher, FxHashMap};
use serde::{
    ser::{Error, SerializeMap, SerializeSeq},
    Deserialize, Serialize, Serializer,
};
use serde_arrow::ArrayBuilder;
use std::{
    cmp::Ordering,
    collections::HashMap,
    fmt,
    fmt::{Display, Formatter},
    hash::{DefaultHasher, Hash, Hasher},
    num::Wrapping,
    sync::Arc,
};
use thiserror::Error;

// Equivalent to parquet decimal(38, 0).
pub const DECIMAL_MAX: i128 = 99999999999999999999999999999999999999i128;

#[derive(Error, Debug)]
#[error("Decimal {0} too large.")]
pub struct InvalidBigDecimal(BigDecimal);

#[derive(Debug, PartialEq, Clone, Serialize)]
#[serde(transparent)]
pub struct PropUntagged(#[serde(with = "PropUntaggedDef")] pub Prop);

#[derive(Serialize, Deserialize)]
#[serde(untagged)]
#[serde(remote = "Prop")]
enum PropUntaggedDef {
    Str(ArcStr),
    U8(u8),
    U16(u16),
    I32(i32),
    I64(i64),
    U32(u32),
    U64(u64),
    F64(f64),
    F32(f32),
    Bool(bool),
    List(PropArray),
    Map(Arc<FxHashMap<ArcStr, Prop>>),
    NDTime(NaiveDateTime),
    DTime(DateTime<Utc>),
    Decimal(BigDecimal),
}

impl From<Prop> for PropUntagged {
    fn from(p: Prop) -> Self {
        PropUntagged(p)
    }
}

impl From<PropUntagged> for Prop {
    fn from(p: PropUntagged) -> Self {
        p.0
    }
}

impl<'de> Deserialize<'de> for PropUntagged {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        #[derive(Deserialize)]
        #[serde(untagged)]
        enum PropUntaggedHelper {
            Bool(bool),
            U8(u8),
            U16(u16),
            U32(u32),
            U64(u64),
            I32(i32),
            I64(i64),
            F64(f64),
            F32(f32),
            Str(ArcStr),
            List(Vec<PropUntagged>), // recursively uses PropUntagged
            Map(FxHashMap<ArcStr, PropUntagged>), // recursively uses PropUntagged
            NDTime(NaiveDateTime),
            DTime(DateTime<Utc>),
            Decimal(BigDecimal),
        }

        let helper = PropUntaggedHelper::deserialize(deserializer)?;
        let prop = match helper {
            PropUntaggedHelper::Bool(v) => Prop::Bool(v),
            PropUntaggedHelper::U8(v) => Prop::U8(v),
            PropUntaggedHelper::U16(v) => Prop::U16(v),
            PropUntaggedHelper::U32(v) => Prop::U32(v),
            PropUntaggedHelper::U64(v) => Prop::U64(v),
            PropUntaggedHelper::I32(v) => Prop::I32(v),
            PropUntaggedHelper::I64(v) => Prop::I64(v),
            PropUntaggedHelper::F64(v) => Prop::F64(v),
            PropUntaggedHelper::F32(v) => Prop::F32(v),
            PropUntaggedHelper::Str(v) => Prop::Str(v),
            PropUntaggedHelper::List(v) => Prop::list(v),
            PropUntaggedHelper::Map(v) => {
                Prop::Map(Arc::new(v.into_iter().map(|(k, p)| (k, p.0)).collect()))
            }
            PropUntaggedHelper::NDTime(v) => Prop::NDTime(v),
            PropUntaggedHelper::DTime(v) => Prop::DTime(v),
            PropUntaggedHelper::Decimal(v) => Prop::Decimal(v),
        };
        Ok(PropUntagged(prop))
    }
}

impl PartialEq<Prop> for PropUntagged {
    fn eq(&self, other: &Prop) -> bool {
        self.0
            .clone()
            .try_cast(other.dtype())
            .map_or(false, |p| p == *other)
    }
}

/// Denotes the types of properties allowed to be stored in the graph.
#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, derive_more::From)]
pub enum Prop {
    Str(ArcStr),
    U8(u8),
    U16(u16),
    I32(i32),
    I64(i64),
    U32(u32),
    U64(u64),
    F64(f64),
    F32(f32),
    Bool(bool),
    List(PropArray),
    Map(Arc<FxHashMap<ArcStr, Prop>>),
    NDTime(NaiveDateTime),
    DTime(DateTime<Utc>),
    Decimal(BigDecimal),
}

impl From<GidRef<'_>> for Prop {
    fn from(value: GidRef<'_>) -> Self {
        match value {
            GidRef::U64(n) => Prop::U64(n),
            GidRef::Str(s) => Prop::str(s),
        }
    }
}

impl<'a> From<PropRef<'a>> for Prop {
    fn from(value: PropRef<'a>) -> Self {
        match value {
            PropRef::Str(s) => Prop::Str(s.into()),
            PropRef::Num(n) => match n {
                PropNum::U8(u) => Prop::U8(u),
                PropNum::U16(u) => Prop::U16(u),
                PropNum::I32(i) => Prop::I32(i),
                PropNum::I64(i) => Prop::I64(i),
                PropNum::U32(u) => Prop::U32(u),
                PropNum::U64(u) => Prop::U64(u),
                PropNum::F32(f) => Prop::F32(f),
                PropNum::F64(f) => Prop::F64(f),
            },
            PropRef::Bool(b) => Prop::Bool(b),
            PropRef::List(v) => Prop::List(v.as_ref().clone()),
            PropRef::Map(m) => m
                .into_prop()
                .unwrap_or_else(|| Prop::Map(Arc::new(Default::default()))),
            PropRef::NDTime(dt) => Prop::NDTime(dt),
            PropRef::DTime(dt) => Prop::DTime(dt),
            PropRef::Decimal { num, scale } => {
                Prop::Decimal(BigDecimal::from_bigint(num.into(), scale as i64))
            }
        }
    }
}

impl Hash for Prop {
    fn hash<H: Hasher>(&self, state: &mut H) {
        match self {
            Prop::Str(s) => s.hash(state),
            Prop::U8(u) => u.hash(state),
            Prop::U16(u) => u.hash(state),
            Prop::I32(i) => i.hash(state),
            Prop::I64(i) => i.hash(state),
            Prop::U32(u) => u.hash(state),
            Prop::U64(u) => u.hash(state),
            Prop::F32(f) => {
                let bits = f.to_bits();
                bits.hash(state);
            }
            Prop::F64(f) => {
                let bits = f.to_bits();
                bits.hash(state);
            }
            Prop::Bool(b) => b.hash(state),
            Prop::NDTime(dt) => dt.hash(state),
            Prop::DTime(dt) => dt.hash(state),
            Prop::List(v) => {
                for prop in v.iter() {
                    prop.hash(state);
                }
            }
            Prop::Map(m) => {
                // Based on python set hash
                let mut hash = Wrapping(1927868237u64);
                hash *= (m.len() as u64).wrapping_add(1);
                for v in m.iter() {
                    let mut inner_hasher = DefaultHasher::new();
                    v.hash(&mut inner_hasher);
                    let inner_hash = Wrapping(inner_hasher.finish());
                    hash ^= (inner_hash ^ (inner_hash << 16) ^ Wrapping(89869747u64))
                        * Wrapping(3644798167u64);
                }
                hash ^= (hash >> 11) ^ (hash >> 25);
                hash *= 69069;
                hash += 907133923;
                state.write_u64(hash.0);
            }
            Prop::Decimal(d) => d.hash(state),
        }
    }
}

impl Eq for Prop {}

impl PartialOrd for Prop {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        match (self, other) {
            (Prop::Str(a), Prop::Str(b)) => a.partial_cmp(b),
            (Prop::U8(a), Prop::U8(b)) => a.partial_cmp(b),
            (Prop::U16(a), Prop::U16(b)) => a.partial_cmp(b),
            (Prop::I32(a), Prop::I32(b)) => a.partial_cmp(b),
            (Prop::I64(a), Prop::I64(b)) => a.partial_cmp(b),
            (Prop::U32(a), Prop::U32(b)) => a.partial_cmp(b),
            (Prop::U64(a), Prop::U64(b)) => a.partial_cmp(b),
            (Prop::F32(a), Prop::F32(b)) => a.partial_cmp(b),
            (Prop::F64(a), Prop::F64(b)) => a.partial_cmp(b),
            (Prop::Bool(a), Prop::Bool(b)) => a.partial_cmp(b),
            (Prop::NDTime(a), Prop::NDTime(b)) => a.partial_cmp(b),
            (Prop::DTime(a), Prop::DTime(b)) => a.partial_cmp(b),
            (Prop::List(a), Prop::List(b)) => a.partial_cmp(b),
            (Prop::Decimal(a), Prop::Decimal(b)) => a.partial_cmp(b),
            _ => None,
        }
    }
}

pub struct SerdeArrowProp<'a>(pub &'a Prop);

#[derive(Clone, Copy, Debug)]
pub struct SerdeArrowList<'a>(pub &'a PropArray);

#[derive(Clone, Copy, Debug)]
pub struct SerdeArrowArray<'a>(pub &'a ArrayRef);

#[derive(Clone, Copy)]
pub struct SerdeArrowMap<'a>(pub &'a HashMap<ArcStr, Prop, FxBuildHasher>);

#[derive(Clone, Copy, Serialize)]
pub struct SerdeRow<P: Serialize> {
    value: Option<P>,
}

impl<'a> Serialize for SerdeArrowList<'a> {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        match &self.0 {
            PropArray::Vec(list) => {
                let mut state = serializer.serialize_seq(Some(self.0.len()))?;

                for prop in list.iter() {
                    state.serialize_element(&SerdeArrowProp(prop))?;
                }

                state.end()
            }
            PropArray::Array(array) => SerdeArrowArray(array).serialize(serializer),
        }
    }
}

impl<'a> Serialize for SerdeArrowMap<'a> {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        let mut state = serializer.serialize_map(Some(self.0.len()))?;
        for (k, v) in self.0.iter() {
            state.serialize_entry(k, &SerdeArrowProp(v))?;
        }
        state.end()
    }
}

impl<'a> Serialize for SerdeArrowProp<'a> {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        match self.0 {
            Prop::I32(i) => serializer.serialize_i32(*i),
            Prop::I64(i) => serializer.serialize_i64(*i),
            Prop::F32(f) => serializer.serialize_f32(*f),
            Prop::F64(f) => serializer.serialize_f64(*f),
            Prop::U8(u) => serializer.serialize_u8(*u),
            Prop::U16(u) => serializer.serialize_u16(*u),
            Prop::U32(u) => serializer.serialize_u32(*u),
            Prop::U64(u) => serializer.serialize_u64(*u),
            Prop::Str(s) => serializer.serialize_str(s),
            Prop::Bool(b) => serializer.serialize_bool(*b),
            Prop::DTime(dt) => serializer.serialize_i64(dt.timestamp_millis()),
            Prop::NDTime(dt) => serializer.serialize_i64(dt.and_utc().timestamp_millis()),
            Prop::List(l) => SerdeArrowList(l).serialize(serializer),
            Prop::Map(m) => SerdeArrowMap(m).serialize(serializer),
            Prop::Decimal(dec) => {
                // Serialize BigDecimal as string manually to match
                // the Arrow Decimal128 format.
                let (num, scale) = dec.as_bigint_and_scale();

                let num_i128 = num.to_i128().ok_or_else(|| {
                    serde::ser::Error::custom(format!(
                        "decimal value {dec} is out of range for i128 representation"
                    ))
                })?;

                let num_formatted =
                    Decimal128Type::format_decimal(num_i128, DECIMAL128_MAX_PRECISION, scale as i8);

                serializer.serialize_str(&num_formatted)
            }
        }
    }
}

impl<'a> Serialize for SerdeArrowArray<'a> {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        let dtype = self.0.data_type();
        let len = self.0.len();
        let mut state = serializer.serialize_seq(Some(len))?;

        match dtype {
            DataType::Boolean => {
                for v in self.0.as_boolean().iter() {
                    state.serialize_element(&v)?;
                }
            }
            DataType::Int32 => {
                for v in self.0.as_primitive::<Int32Type>().iter() {
                    state.serialize_element(&v)?;
                }
            }
            DataType::Int64 => {
                for v in self.0.as_primitive::<Int64Type>().iter() {
                    state.serialize_element(&v)?;
                }
            }
            DataType::UInt8 => {
                for v in self.0.as_primitive::<UInt8Type>().iter() {
                    state.serialize_element(&v)?;
                }
            }
            DataType::UInt16 => {
                for v in self.0.as_primitive::<UInt16Type>().iter() {
                    state.serialize_element(&v)?;
                }
            }
            DataType::UInt32 => {
                for v in self.0.as_primitive::<UInt32Type>().iter() {
                    state.serialize_element(&v)?;
                }
            }
            DataType::UInt64 => {
                for v in self.0.as_primitive::<UInt64Type>().iter() {
                    state.serialize_element(&v)?;
                }
            }
            DataType::Float32 => {
                for v in self.0.as_primitive::<Float32Type>().iter() {
                    state.serialize_element(&v)?;
                }
            }
            DataType::Float64 => {
                for v in self.0.as_primitive::<Float64Type>().iter() {
                    state.serialize_element(&v)?;
                }
            }
            DataType::Timestamp(unit, _) => match unit {
                TimeUnit::Second => {
                    for v in self.0.as_primitive::<TimestampSecondType>().iter() {
                        state.serialize_element(&v)?;
                    }
                }
                TimeUnit::Millisecond => {
                    for v in self.0.as_primitive::<TimestampMillisecondType>().iter() {
                        state.serialize_element(&v)?;
                    }
                }
                TimeUnit::Microsecond => {
                    for v in self.0.as_primitive::<TimestampMicrosecondType>().iter() {
                        state.serialize_element(&v)?;
                    }
                }
                TimeUnit::Nanosecond => {
                    for v in self.0.as_primitive::<TimestampNanosecondType>().iter() {
                        state.serialize_element(&v)?;
                    }
                }
            },
            DataType::Date32 => {
                for v in self.0.as_primitive::<Date32Type>().iter() {
                    state.serialize_element(&v)?;
                }
            }
            DataType::Date64 => {
                for v in self.0.as_primitive::<Date64Type>().iter() {
                    state.serialize_element(&v)?;
                }
            }
            DataType::Utf8 => {
                for v in self.0.as_string::<i32>().iter() {
                    state.serialize_element(&v)?;
                }
            }
            DataType::LargeUtf8 => {
                for v in self.0.as_string::<i64>().iter() {
                    state.serialize_element(&v)?;
                }
            }
            DataType::Utf8View => {
                for v in self.0.as_string_view().iter() {
                    state.serialize_element(&v)?;
                }
            }
            DataType::Decimal128(precision, scale) => {
                for v in self.0.as_primitive::<Decimal128Type>().iter() {
                    // i128 is not supported directly by serde_arrow,
                    // so we format as string manually.
                    let element = v.map(|v| Decimal128Type::format_decimal(v, *precision, *scale));

                    state.serialize_element(&element)?
                }
            }
            DataType::Struct(_) => {
                let struct_array = self.0.as_struct();
                match struct_array.nulls() {
                    None => {
                        for i in 0..struct_array.len() {
                            state.serialize_element(&ArrowRow::new(struct_array, i))?;
                        }
                    }
                    Some(nulls) => {
                        for (i, is_valid) in nulls.iter().enumerate() {
                            state.serialize_element(
                                &is_valid.then_some(ArrowRow::new(struct_array, i)),
                            )?;
                        }
                    }
                }
            }
            DataType::List(_) => {
                let list = self.0.as_list::<i32>();
                for array in list.iter() {
                    state.serialize_element(&array.as_ref().map(SerdeArrowArray))?;
                }
            }
            DataType::LargeList(_) => {
                let list = self.0.as_list::<i64>();
                for array in list.iter() {
                    state.serialize_element(&array.as_ref().map(SerdeArrowArray))?;
                }
            }
            DataType::Null => {
                for _ in 0..self.0.len() {
                    state.serialize_element(&None::<()>)?;
                }
            }
            dtype => Err(Error::custom(format!("unsuported data type {dtype:?}")))?,
        }
        state.end()
    }
}

pub fn validate_bd(bd: &BigDecimal) -> Result<(), InvalidBigDecimal> {
    let (bint, scale) = bd.as_bigint_and_exponent();
    if bint <= BigInt::from(DECIMAL_MAX) && scale <= 38 {
        Ok(())
    } else {
        Err(InvalidBigDecimal(bd.clone()))
    }
}

impl Prop {
    // auxiliary function to help with numerical conversion
    pub fn cast_num<T>(self) -> Option<T>
    where
        T: FromPrimitive + Bounded,
    {
        match self {
            Prop::U8(v) => T::from_u8(v),
            Prop::U16(v) => T::from_u16(v),
            Prop::I32(v) => T::from_i32(v),
            Prop::I64(v) => T::from_i64(v),
            Prop::U32(v) => T::from_u32(v),
            Prop::U64(v) => T::from_u64(v),
            Prop::F32(v) => T::from_f32(v),
            Prop::F64(v) => T::from_f64(v),
            _ => None,
        }
    }

    /// convert prop into another prop type (primarily for numerical conversions)
    pub fn try_cast(self, prop_type: PropType) -> Option<Prop> {
        // Early return if casting to the same type
        if self.dtype() == prop_type {
            return Some(self);
        }

        match self {
            Prop::Str(v) => match prop_type {
                PropType::Str => Some(Prop::Str(v)),
                PropType::U8 => v.parse::<u8>().map(Prop::U8).ok(),
                PropType::U16 => v.parse::<u16>().map(Prop::U16).ok(),
                PropType::I32 => v.parse::<i32>().map(Prop::I32).ok(),
                PropType::I64 => v.parse::<i64>().map(Prop::I64).ok(),
                PropType::U32 => v.parse::<u32>().map(Prop::U32).ok(),
                PropType::U64 => v.parse::<u64>().map(Prop::U64).ok(),
                PropType::F32 => v.parse::<f32>().map(Prop::F32).ok(),
                PropType::F64 => v.parse::<f64>().map(Prop::F64).ok(),
                PropType::Bool => v.parse::<bool>().map(Prop::Bool).ok(),
                PropType::NDTime => v.parse::<NaiveDateTime>().map(Prop::NDTime).ok(),
                PropType::DTime => v.parse::<DateTime<Utc>>().map(Prop::DTime).ok(),
                PropType::Decimal { scale } => v
                    .parse::<BigDecimal>()
                    .map(|v| Prop::Decimal(v.with_scale(scale)))
                    .ok(),
                _ => None,
            },
            Prop::Bool(v) => match prop_type {
                PropType::Str => Some(Prop::Str(v.to_string().into())),
                PropType::U8 => Some(Prop::U8(v as _)),
                PropType::U16 => Some(Prop::U16(v as _)),
                PropType::I32 => Some(Prop::I32(v as _)),
                PropType::I64 => Some(Prop::I64(v as _)),
                PropType::U32 => Some(Prop::U32(v as _)),
                PropType::U64 => Some(Prop::U64(v as _)),
                PropType::F32 => Some(Prop::F32(if v { 1.0 } else { 0.0 })),
                PropType::F64 => Some(Prop::F64(if v { 1.0 } else { 0.0 })),
                PropType::Bool => unreachable!("Same type case handled above"),
                PropType::Decimal { scale } => {
                    let val = if v {
                        BigDecimal::from(1)
                    } else {
                        BigDecimal::from(0)
                    };
                    Some(Prop::Decimal(val.with_scale(scale)))
                }
                _ => None,
            },
            Prop::List(_v) => None,
            Prop::Map(_v) => None,
            Prop::NDTime(v) => match prop_type {
                PropType::Str => Some(Prop::Str(v.to_string().into())),
                PropType::I64 => Some(Prop::I64(v.and_utc().timestamp())),
                PropType::U64 => {
                    let ts = v.and_utc().timestamp();
                    if ts >= 0 {
                        Some(Prop::U64(ts as u64))
                    } else {
                        None
                    }
                }
                PropType::DTime => Some(Prop::DTime(v.and_utc())),
                PropType::NDTime => unreachable!("Same type case handled above"),
                _ => None,
            },
            Prop::DTime(v) => match prop_type {
                PropType::Str => Some(Prop::Str(v.to_rfc3339().into())),
                PropType::I64 => Some(Prop::I64(v.timestamp())),
                PropType::U64 => {
                    let ts = v.timestamp();
                    if ts >= 0 {
                        Some(Prop::U64(ts as u64))
                    } else {
                        None
                    }
                }
                PropType::NDTime => Some(Prop::NDTime(v.naive_utc())),
                PropType::DTime => unreachable!("Same type case handled above"),
                _ => None,
            },
            Prop::Decimal(v) => match prop_type {
                PropType::Str => Some(Prop::Str(v.to_string().into())),
                PropType::U8 => {
                    let as_i64 = v.to_i64()?;
                    u8::from_i64(as_i64).map(Prop::U8)
                }
                PropType::U16 => {
                    let as_i64 = v.to_i64()?;
                    u16::from_i64(as_i64).map(Prop::U16)
                }
                PropType::I32 => {
                    let as_i64 = v.to_i64()?;
                    i32::from_i64(as_i64).map(Prop::I32)
                }
                PropType::I64 => v.to_i64().map(Prop::I64),
                PropType::U32 => {
                    let as_i64 = v.to_i64()?;
                    u32::from_i64(as_i64).map(Prop::U32)
                }
                PropType::U64 => {
                    let as_i64 = v.to_i64()?;
                    u64::from_i64(as_i64).map(Prop::U64)
                }
                PropType::F32 => v.to_f32().map(Prop::F32),
                PropType::F64 => v.to_f64().map(Prop::F64),
                PropType::Bool => {
                    let as_i64 = v.to_i64()?;
                    Some(Prop::Bool(as_i64 != 0))
                }
                PropType::Decimal { scale } => Some(Prop::Decimal(v.with_scale(scale))),
                _ => None,
            },
            _ => match prop_type {
                // Numeric conversions using num_traits
                PropType::U8 => self.cast_num::<u8>().map(Prop::U8),
                PropType::U16 => self.cast_num::<u16>().map(Prop::U16),
                PropType::I32 => self.cast_num::<i32>().map(Prop::I32),
                PropType::I64 => self.cast_num::<i64>().map(Prop::I64),
                PropType::U32 => self.cast_num::<u32>().map(Prop::U32),
                PropType::U64 => self.cast_num::<u64>().map(Prop::U64),
                PropType::F32 => self.cast_num::<f32>().map(Prop::F32),
                PropType::F64 => self.cast_num::<f64>().map(Prop::F64),
                _ => None,
            },
        }
    }

    /// Losslessly widen unsigned integer variants to u64.
    #[inline]
    pub fn as_u64_lossless(&self) -> Option<u64> {
        match self {
            Prop::U8(v) => Some(*v as u64),
            Prop::U16(v) => Some(*v as u64),
            Prop::U32(v) => Some(*v as u64),
            Prop::U64(v) => Some(*v),
            _ => None,
        }
    }

    /// Losslessly widen signed integer variants to i64.
    #[inline]
    pub fn as_i64_lossless(&self) -> Option<i64> {
        match self {
            Prop::I32(v) => Some(*v as i64),
            Prop::I64(v) => Some(*v),
            _ => None,
        }
    }

    /// Losslessly widen float variants to f64.
    #[inline]
    pub fn as_f64_lossless(&self) -> Option<f64> {
        match self {
            Prop::F32(v) => Some(*v as f64),
            Prop::F64(v) => Some(*v),
            _ => None,
        }
    }

    pub fn try_from_bd(bd: BigDecimal) -> Result<Prop, InvalidBigDecimal> {
        validate_bd(&bd)?;
        Ok(Prop::Decimal(bd))
    }

    pub fn map(vals: impl IntoIterator<Item = (impl Into<ArcStr>, impl Into<Prop>)>) -> Self {
        let h_map: FxHashMap<_, _> = vals
            .into_iter()
            .map(|(k, v)| (k.into(), v.into()))
            .collect();
        Prop::Map(h_map.into())
    }

    pub fn as_map(&self) -> Option<SerdeArrowMap<'_>> {
        match self {
            Prop::Map(map) => Some(SerdeArrowMap(map)),
            _ => None,
        }
    }

    pub fn dtype(&self) -> PropType {
        match self {
            Prop::Str(_) => PropType::Str,
            Prop::U8(_) => PropType::U8,
            Prop::U16(_) => PropType::U16,
            Prop::I32(_) => PropType::I32,
            Prop::I64(_) => PropType::I64,
            Prop::U32(_) => PropType::U32,
            Prop::U64(_) => PropType::U64,
            Prop::F32(_) => PropType::F32,
            Prop::F64(_) => PropType::F64,
            Prop::Bool(_) => PropType::Bool,
            Prop::List(list) => PropType::List(Box::new(list.dtype())),
            Prop::Map(map) => PropType::map(map.iter().map(|(k, v)| (k, v.dtype()))),
            Prop::NDTime(_) => PropType::NDTime,
            Prop::DTime(_) => PropType::DTime,
            Prop::Decimal(d) => PropType::Decimal {
                scale: d.as_bigint_and_scale().1,
            },
        }
    }

    pub fn str<S: Into<ArcStr>>(s: S) -> Prop {
        Prop::Str(s.into())
    }

    pub fn list<P: Into<Prop>, I: IntoIterator<Item = P>>(vals: I) -> Prop {
        Prop::List(PropArray::Vec(
            vals.into_iter().map_into().collect::<Vec<_>>().into(),
        ))
    }

    pub fn add(self, other: Prop) -> Option<Prop> {
        match (self, other) {
            (Prop::U8(a), Prop::U8(b)) => Some(Prop::U8(a + b)),
            (Prop::U16(a), Prop::U16(b)) => Some(Prop::U16(a + b)),
            (Prop::I32(a), Prop::I32(b)) => Some(Prop::I32(a + b)),
            (Prop::I64(a), Prop::I64(b)) => Some(Prop::I64(a + b)),
            (Prop::U32(a), Prop::U32(b)) => Some(Prop::U32(a + b)),
            (Prop::U64(a), Prop::U64(b)) => Some(Prop::U64(a + b)),
            (Prop::F32(a), Prop::F32(b)) => Some(Prop::F32(a + b)),
            (Prop::F64(a), Prop::F64(b)) => Some(Prop::F64(a + b)),
            (Prop::Str(a), Prop::Str(b)) => Some(Prop::Str((a.to_string() + b.as_ref()).into())),
            (Prop::Decimal(a), Prop::Decimal(b)) => Some(Prop::Decimal(a + b)),
            _ => None,
        }
    }

    pub fn min(self, other: Prop) -> Option<Prop> {
        self.partial_cmp(&other).map(|ord| match ord {
            Ordering::Less => self,
            Ordering::Equal => self,
            Ordering::Greater => other,
        })
    }

    pub fn max(self, other: Prop) -> Option<Prop> {
        self.partial_cmp(&other).map(|ord| match ord {
            Ordering::Less => other,
            Ordering::Equal => self,
            Ordering::Greater => self,
        })
    }

    pub fn divide(self, other: Prop) -> Option<Prop> {
        match (self, other) {
            (Prop::U8(a), Prop::U8(b)) if b != 0 => Some(Prop::U8(a / b)),
            (Prop::U16(a), Prop::U16(b)) if b != 0 => Some(Prop::U16(a / b)),
            (Prop::I32(a), Prop::I32(b)) if b != 0 => Some(Prop::I32(a / b)),
            (Prop::I64(a), Prop::I64(b)) if b != 0 => Some(Prop::I64(a / b)),
            (Prop::U32(a), Prop::U32(b)) if b != 0 => Some(Prop::U32(a / b)),
            (Prop::U64(a), Prop::U64(b)) if b != 0 => Some(Prop::U64(a / b)),
            (Prop::F32(a), Prop::F32(b)) => Some(Prop::F32(a / b)),
            (Prop::F64(a), Prop::F64(b)) => Some(Prop::F64(a / b)),
            (Prop::Decimal(a), Prop::Decimal(b)) if b != BigDecimal::from(0) => {
                Some(Prop::Decimal(a / b))
            }
            _ => None,
        }
    }
}

pub fn list_array_from_props<P: Serialize + fmt::Debug + Clone>(
    dt: &DataType,
    props: impl IntoIterator<Item = Option<P>>,
) -> Result<LargeListArray, serde_arrow::Error> {
    let fields: Fields = vec![Field::new("value", dt.clone(), true)].into();
    let mut builder = ArrayBuilder::from_arrow(&fields)?;

    for value in props {
        builder.push(SerdeRow { value })?;
    }

    let arrays = builder.to_arrow()?;

    Ok(arrays.first().unwrap().as_list::<i64>().clone())
}

pub fn struct_array_from_props<P: Serialize>(
    dt: &DataType,
    props: impl IntoIterator<Item = Option<P>>,
) -> Result<StructArray, serde_arrow::Error> {
    let fields = [FieldRef::new(Field::new("value", dt.clone(), true))];
    let mut builder = ArrayBuilder::from_arrow(&fields)?;

    for p in props {
        builder.push(SerdeRow { value: p })?
    }

    let arrays = builder.to_arrow()?;

    Ok(arrays.first().unwrap().as_struct().clone())
}

impl Display for Prop {
    fn fmt(&self, f: &mut Formatter) -> fmt::Result {
        match self {
            Prop::Str(value) => write!(f, "{}", value),
            Prop::U8(value) => write!(f, "{}", value),
            Prop::U16(value) => write!(f, "{}", value),
            Prop::I32(value) => write!(f, "{}", value),
            Prop::I64(value) => write!(f, "{}", value),
            Prop::U32(value) => write!(f, "{}", value),
            Prop::U64(value) => write!(f, "{}", value),
            Prop::F32(value) => write!(f, "{}", value),
            Prop::F64(value) => write!(f, "{}", value),
            Prop::Bool(value) => write!(f, "{}", value),
            Prop::DTime(value) => write!(f, "{}", value),
            Prop::NDTime(value) => write!(f, "{}", value),
            Prop::List(value) => {
                write!(
                    f,
                    "[{}]",
                    value
                        .iter()
                        .map(|item| {
                            match item {
                                Prop::Str(_) => {
                                    format!("\"{}\"", item)
                                }
                                _ => {
                                    format!("{}", item)
                                }
                            }
                        })
                        .join(", ")
                )
            }
            Prop::Map(value) => {
                write!(
                    f,
                    "{{{}}}",
                    value
                        .iter()
                        .map(|(key, val)| {
                            match val {
                                Prop::Str(_) => {
                                    format!("\"{}\": \"{}\"", key, val)
                                }
                                _ => {
                                    format!("\"{}\": {}", key, val)
                                }
                            }
                        })
                        .join(", ")
                )
            }
            Prop::Decimal(d) => write!(f, "Decimal({})", d.as_bigint_and_scale().1),
        }
    }
}

impl From<&str> for Prop {
    fn from(s: &str) -> Self {
        Prop::Str(s.into())
    }
}

impl From<String> for Prop {
    fn from(s: String) -> Self {
        Prop::Str(s.into())
    }
}

impl From<HashMap<ArcStr, Prop>> for Prop {
    fn from(value: HashMap<ArcStr, Prop>) -> Self {
        Prop::Map(Arc::new(value.into_iter().collect()))
    }
}

impl From<FxHashMap<ArcStr, Prop>> for Prop {
    fn from(value: FxHashMap<ArcStr, Prop>) -> Self {
        Prop::Map(Arc::new(value))
    }
}

impl From<Vec<Prop>> for Prop {
    fn from(value: Vec<Prop>) -> Self {
        Prop::List(value.into())
    }
}

impl From<&Prop> for Prop {
    fn from(value: &Prop) -> Self {
        value.clone()
    }
}

impl From<ArrayRef> for Prop {
    fn from(value: ArrayRef) -> Self {
        Prop::List(PropArray::from(value))
    }
}

pub trait IntoPropMap {
    fn into_prop_map(self) -> Prop;
}

impl<I: IntoIterator<Item = (K, V)>, K: Into<ArcStr>, V: Into<Prop>> IntoPropMap for I {
    fn into_prop_map(self) -> Prop {
        Prop::Map(Arc::new(
            self.into_iter()
                .map(|(k, v)| (k.into(), v.into()))
                .collect(),
        ))
    }
}

pub trait IntoPropList {
    fn into_prop_list(self) -> Prop;
}

impl<I: IntoIterator<Item = K>, K: Into<Prop>> IntoPropList for I {
    fn into_prop_list(self) -> Prop {
        let vec = self.into_iter().map(|v| v.into()).collect::<Vec<_>>();
        Prop::List(vec.into())
    }
}

pub trait IntoProp {
    fn into_prop(self) -> Prop;
}

impl<T: Into<Prop>> IntoProp for T {
    fn into_prop(self) -> Prop {
        self.into()
    }
}

pub fn sort_comparable_props(props: Vec<&Prop>) -> Vec<&Prop> {
    // Filter out non-comparable props
    let mut comparable_props: Vec<_> = props
        .into_iter()
        .filter(|p| {
            matches!(
                p,
                Prop::Str(_)
                    | Prop::U8(_)
                    | Prop::U16(_)
                    | Prop::I32(_)
                    | Prop::I64(_)
                    | Prop::U32(_)
                    | Prop::U64(_)
                    | Prop::F32(_)
                    | Prop::F64(_)
                    | Prop::Bool(_)
                    | Prop::NDTime(_)
                    | Prop::DTime(_)
            )
        })
        .collect();

    // Sort the comparable props
    comparable_props.sort_by(|a, b| a.partial_cmp(b).unwrap_or(Ordering::Equal));

    comparable_props
}
