scylla_cql_core/

value.rs

//! Defines CQL values of various types and their representations,
//! as well as conversion between them and other types.

use std::net::IpAddr;
use std::time::{SystemTime, UNIX_EPOCH};

use thiserror::Error;
use uuid::Uuid;

use crate::utils::safe_format::IteratorSafeFormatExt;

/// Error type indicating that the value is too large to fit in the destination type.
///
/// Intended to be used when converting between CQL types and other types
/// in case the source type is larger than the destination type.
#[derive(Debug, Error, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
#[error(
    "Conversion between CQL type and another type is not possible because\
    value of one of them is too large to fit in the other"
)]
pub struct ValueOverflow;

/// Represents an unset value
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub struct Unset;

/// Represents an counter value
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub struct Counter(pub i64);

/// Enum providing a way to represent a value that might be unset
#[derive(Debug, Clone, Copy, Default)]
pub enum MaybeUnset<V> {
    /// The value is unset, so the server's state about this value will not be changed.
    #[default]
    Unset,
    /// The value is set, so the server's state will be changed to this value.
    Set(V),
}

impl<V> MaybeUnset<V> {
    /// Converts an `Option<V>` into a `MaybeUnset<V>`.
    #[inline]
    pub fn from_option(opt: Option<V>) -> Self {
        match opt {
            Some(v) => Self::Set(v),
            None => Self::Unset,
        }
    }
}

/// Values that may be empty or not.
///
/// In CQL, some types can have a special value of "empty", represented as
/// a serialized value of length 0. An example of this are integral types:
/// the "int" type can actually hold 2^32 + 1 possible values because of this
/// quirk. Note that this is distinct from being NULL.
///
/// Rust types that cannot represent an empty value (e.g. i32) should implement
/// this trait in order to be deserialized as [`MaybeEmpty`] or serialized
/// from it.
pub trait Emptiable {}

// Implementations of Emptiable for types that support empty CQL values.

impl Emptiable for bool {}
impl Emptiable for i8 {}
impl Emptiable for i16 {}
impl Emptiable for i32 {}
impl Emptiable for i64 {}
impl Emptiable for f32 {}
impl Emptiable for f64 {}

impl Emptiable for CqlVarint {}
impl<'b> Emptiable for CqlVarintBorrowed<'b> {}
impl Emptiable for CqlDecimal {}
impl<'b> Emptiable for CqlDecimalBorrowed<'b> {}
impl Emptiable for CqlDate {}
impl Emptiable for CqlTime {}
impl Emptiable for CqlTimestamp {}
impl Emptiable for CqlTimeuuid {}

impl Emptiable for std::net::IpAddr {}
impl Emptiable for uuid::Uuid {}

#[cfg(feature = "num-bigint-03")]
impl Emptiable for num_bigint_03::BigInt {}
#[cfg(feature = "num-bigint-04")]
impl Emptiable for num_bigint_04::BigInt {}
#[cfg(feature = "bigdecimal-04")]
impl Emptiable for bigdecimal_04::BigDecimal {}

#[cfg(feature = "chrono-04")]
impl Emptiable for chrono_04::NaiveDate {}
#[cfg(feature = "chrono-04")]
impl Emptiable for chrono_04::NaiveTime {}
#[cfg(feature = "chrono-04")]
impl Emptiable for chrono_04::DateTime<chrono_04::Utc> {}

#[cfg(feature = "time-03")]
impl Emptiable for time_03::Date {}
#[cfg(feature = "time-03")]
impl Emptiable for time_03::Time {}
#[cfg(feature = "time-03")]
impl Emptiable for time_03::OffsetDateTime {}

/// A value that may be empty or not.
///
/// `MaybeEmpty` was introduced to help support the quirk described in [`Emptiable`]
/// for Rust types which can't represent the empty, additional value.
///
/// This type can be both serialized and deserialized. When serializing,
/// [`MaybeEmpty::Empty`] will produce an empty value (0 bytes) for emptiable types.
/// When deserializing, an empty value will be represented as [`MaybeEmpty::Empty`].
#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Copy)]
pub enum MaybeEmpty<T: Emptiable> {
    /// Represents an empty value (0 bytes in the serialized form).
    Empty,
    /// Represents a non-empty value.
    Value(T),
}

/// Represents timeuuid (uuid V1) value
///
/// This type has custom comparison logic which follows ScyllaDB/Cassandra semantics.
/// For details, see [`Ord` implementation](#impl-Ord-for-CqlTimeuuid).
#[derive(Debug, Clone, Copy, Eq)]
pub struct CqlTimeuuid(Uuid);

/// [`Uuid`] delegate methods
impl CqlTimeuuid {
    /// Creates a new nil `CqlTimeuuid`.
    /// See [`Uuid::nil`] for details.
    pub fn nil() -> Self {
        Self(Uuid::nil())
    }

    /// Returns byte representation of the `CqlTimeuuid`.
    /// See [`Uuid::as_bytes`] for details.
    pub fn as_bytes(&self) -> &[u8; 16] {
        self.0.as_bytes()
    }

    /// Returns a `u128` representation of the `CqlTimeuuid`.
    /// See [`Uuid::as_u128`] for details.
    pub fn as_u128(&self) -> u128 {
        self.0.as_u128()
    }

    /// Returns a representation of the `CqlTimeuuid` as a list of its logical fields.
    /// See [`Uuid::as_fields`] for details.
    pub fn as_fields(&self) -> (u32, u16, u16, &[u8; 8]) {
        self.0.as_fields()
    }

    /// Returns a representation of the `CqlTimeuuid` as a pair of `u64` values.
    /// See [`Uuid::as_u64_pair`] for details.
    pub fn as_u64_pair(&self) -> (u64, u64) {
        self.0.as_u64_pair()
    }

    /// Creates a new `CqlTimeuuid` from a big-endian byte representation.
    /// See [`Uuid::from_slice`] for details.
    pub fn from_slice(b: &[u8]) -> Result<Self, uuid::Error> {
        Ok(Self(Uuid::from_slice(b)?))
    }

    /// Creates a new `CqlTimeuuid` from a little-endian byte representation.
    /// See [`Uuid::from_slice_le`] for details.
    pub fn from_slice_le(b: &[u8]) -> Result<Self, uuid::Error> {
        Ok(Self(Uuid::from_slice_le(b)?))
    }

    /// Creates a new `CqlTimeuuid` from a big-endian byte representation.
    /// See [`Uuid::from_bytes`] for details.
    pub fn from_bytes(bytes: [u8; 16]) -> Self {
        Self(Uuid::from_bytes(bytes))
    }

    /// Creates a new `CqlTimeuuid` from a little-endian byte representation.
    /// See [`Uuid::from_bytes_le`] for details.
    pub fn from_bytes_le(bytes: [u8; 16]) -> Self {
        Self(Uuid::from_bytes_le(bytes))
    }

    /// Creates a new `CqlTimeuuid` from a big-endian byte representation of its fields.
    /// See [`Uuid::from_fields`] for details.
    pub fn from_fields(d1: u32, d2: u16, d3: u16, d4: &[u8; 8]) -> Self {
        Self(Uuid::from_fields(d1, d2, d3, d4))
    }

    /// Creates a new `CqlTimeuuid` from a little-endian byte representation of its fields.
    /// See [`Uuid::from_fields_le`] for details.
    pub fn from_fields_le(d1: u32, d2: u16, d3: u16, d4: &[u8; 8]) -> Self {
        Self(Uuid::from_fields_le(d1, d2, d3, d4))
    }

    /// Creates a new `CqlTimeuuid` from a big-endian `u128` value.
    /// See [`Uuid::from_u128`] for details.
    pub fn from_u128(v: u128) -> Self {
        Self(Uuid::from_u128(v))
    }

    /// Creates a new `CqlTimeuuid` from a little-endian `u128` value.
    /// See [`Uuid::from_u128_le`] for details.
    pub fn from_u128_le(v: u128) -> Self {
        Self(Uuid::from_u128_le(v))
    }

    /// Creates a new `CqlTimeuuid` from a pair of `u64` values.
    /// See [`Uuid::from_u64_pair`] for details.
    pub fn from_u64_pair(high_bits: u64, low_bits: u64) -> Self {
        Self(Uuid::from_u64_pair(high_bits, low_bits))
    }
}

impl CqlTimeuuid {
    /// Read 8 most significant bytes of timeuuid from serialized bytes
    fn msb(&self) -> u64 {
        // Scylla and Cassandra use a standard UUID memory layout for MSB:
        // 4 bytes    2 bytes    2 bytes
        // time_low - time_mid - time_hi_and_version
        let bytes = self.0.as_bytes();
        u64::from_be_bytes([
            bytes[6] & 0x0f,
            bytes[7],
            bytes[4],
            bytes[5],
            bytes[0],
            bytes[1],
            bytes[2],
            bytes[3],
        ])
    }

    fn lsb(&self) -> u64 {
        let bytes = self.0.as_bytes();
        u64::from_be_bytes([
            bytes[8], bytes[9], bytes[10], bytes[11], bytes[12], bytes[13], bytes[14], bytes[15],
        ])
    }

    /// Returns the least significant bytes transformed so that an unsigned
    /// comparison of the result reproduces Scylla/Cassandra's ordering of those
    /// bytes for two timeuuids with equal `msb` (used by the [`Ord`] impl).
    ///
    /// Cassandra legacy compares the 8 low bytes as *signed bytes* — each byte
    /// independently, most-significant byte first — not as one signed 64-bit
    /// integer. Scylla implements this in `timeuuid_tri_compare` (utils/UUID.hh)
    /// as `lsb ^ 0x8080808080808080` followed by an unsigned compare: flipping
    /// the top bit of every byte maps each byte's signed order onto its unsigned
    /// order, so a big-endian `u64` comparison of the XORed value matches the
    /// byte-wise signed ordering.
    fn lsb_signed(&self) -> u64 {
        self.lsb() ^ 0x8080808080808080
    }
}

impl std::str::FromStr for CqlTimeuuid {
    type Err = uuid::Error;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        Ok(Self(Uuid::from_str(s)?))
    }
}

impl std::fmt::Display for CqlTimeuuid {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.0)
    }
}

impl AsRef<Uuid> for CqlTimeuuid {
    fn as_ref(&self) -> &Uuid {
        &self.0
    }
}

impl From<CqlTimeuuid> for Uuid {
    fn from(value: CqlTimeuuid) -> Self {
        value.0
    }
}

impl From<Uuid> for CqlTimeuuid {
    fn from(value: Uuid) -> Self {
        Self(value)
    }
}

/// Compare two values of timeuuid type.
///
/// Cassandra legacy requires:
/// - converting 8 most significant bytes to date, which is then compared.
/// - masking off UUID version from the 8 ms-bytes during compare, to
///   treat possible non-version-1 UUID the same way as UUID.
/// - using signed compare for least significant bits.
impl Ord for CqlTimeuuid {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        let mut res = self.msb().cmp(&other.msb());
        if let std::cmp::Ordering::Equal = res {
            res = self.lsb_signed().cmp(&other.lsb_signed());
        }
        res
    }
}

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

impl PartialEq for CqlTimeuuid {
    fn eq(&self, other: &Self) -> bool {
        self.cmp(other) == std::cmp::Ordering::Equal
    }
}

impl std::hash::Hash for CqlTimeuuid {
    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
        self.lsb_signed().hash(state);
        self.msb().hash(state);
    }
}

/// Native CQL `varint` representation.
///
/// Represented as two's-complement binary in big-endian order.
///
/// This type is a raw representation in bytes. It's the default
/// implementation of `varint` type - independent of any
/// external crates and crate features.
///
/// The type is not very useful in most use cases.
/// However, users can make use of more complex types
/// such as `num_bigint::BigInt` (v0.3/v0.4).
/// The library support (e.g. conversion from [`CqlValue`]) for these types is
/// enabled via `num-bigint-03` and `num-bigint-04` crate features.
///
/// This struct holds owned bytes. If you wish to borrow the bytes instead,
/// see [`CqlVarintBorrowed`] documentation.
///
/// # DB data format
/// Notice that [constructors](CqlVarint#impl-CqlVarint)
/// don't perform any normalization on the provided data.
/// This means that underlying bytes may contain leading zeros.
///
/// Currently, Scylla and Cassandra support non-normalized `varint` values.
/// Bytes provided by the user via constructor are passed to DB as is.
///
/// The implementation of [`PartialEq`], however, normalizes the underlying bytes
/// before comparison. For details, check [examples](#impl-PartialEq-for-CqlVarint).
#[derive(Clone, Eq, Debug)]
pub struct CqlVarint(Vec<u8>);

/// A borrowed version of native CQL `varint` representation.
///
/// Refer to the documentation of [`CqlVarint`].
/// Especially, see the disclaimer about [non-normalized values](CqlVarint#db-data-format).
#[derive(Clone, Eq, Debug)]
pub struct CqlVarintBorrowed<'b>(&'b [u8]);

/// Constructors from bytes
impl CqlVarint {
    /// Creates a [`CqlVarint`] from an array of bytes in
    /// two's complement big-endian binary representation.
    ///
    /// See: disclaimer about [non-normalized values](CqlVarint#db-data-format).
    pub fn from_signed_bytes_be(digits: Vec<u8>) -> Self {
        Self(digits)
    }

    /// Creates a [`CqlVarint`] from a slice of bytes in
    /// two's complement binary big-endian representation.
    ///
    /// See: disclaimer about [non-normalized values](CqlVarint#db-data-format).
    pub fn from_signed_bytes_be_slice(digits: &[u8]) -> Self {
        Self::from_signed_bytes_be(digits.to_vec())
    }
}

/// Constructors from bytes
impl<'b> CqlVarintBorrowed<'b> {
    /// Creates a [`CqlVarintBorrowed`] from a slice of bytes in
    /// two's complement binary big-endian representation.
    ///
    /// See: disclaimer about [non-normalized values](CqlVarint#db-data-format).
    pub fn from_signed_bytes_be_slice(digits: &'b [u8]) -> Self {
        Self(digits)
    }
}

/// Conversion to bytes
impl CqlVarint {
    /// Converts [`CqlVarint`] to an array of bytes in two's
    /// complement binary big-endian representation.
    pub fn into_signed_bytes_be(self) -> Vec<u8> {
        self.0
    }

    /// Returns a slice of bytes in two's complement
    /// binary big-endian representation.
    pub fn as_signed_bytes_be_slice(&self) -> &[u8] {
        &self.0
    }
}

/// Conversion to bytes
impl CqlVarintBorrowed<'_> {
    /// Returns a slice of bytes in two's complement
    /// binary big-endian representation.
    pub fn as_signed_bytes_be_slice(&self) -> &[u8] {
        self.0
    }
}

/// An internal utility trait used to implement [`AsNormalizedVarintSlice`]
/// for both [`CqlVarint`] and [`CqlVarintBorrowed`].
trait AsVarintSlice {
    fn as_slice(&self) -> &[u8];
}
impl AsVarintSlice for CqlVarint {
    fn as_slice(&self) -> &[u8] {
        self.as_signed_bytes_be_slice()
    }
}
impl AsVarintSlice for CqlVarintBorrowed<'_> {
    fn as_slice(&self) -> &[u8] {
        self.as_signed_bytes_be_slice()
    }
}

/// An internal utility trait used to implement [`PartialEq`] and [`std::hash::Hash`]
/// for [`CqlVarint`] and [`CqlVarintBorrowed`].
trait AsNormalizedVarintSlice {
    fn as_normalized_slice(&self) -> &[u8];
}
impl<V: AsVarintSlice> AsNormalizedVarintSlice for V {
    fn as_normalized_slice(&self) -> &[u8] {
        let digits = self.as_slice();
        if digits.is_empty() {
            // num-bigint crate normalizes empty vector to 0.
            // We will follow the same approach.
            return &[0];
        }

        let non_zero_position = match digits.iter().position(|b| *b != 0) {
            Some(pos) => pos,
            None => {
                // Vector is filled with zeros. Represent it as 0.
                return &[0];
            }
        };

        if non_zero_position > 0 {
            // There were some leading zeros.
            // Now, there are two cases:
            let zeros_to_remove = if digits[non_zero_position] > 0x7f {
                // Most significant bit is 1, so we need to include one of the leading
                // zeros as originally it represented a positive number.
                non_zero_position - 1
            } else {
                // Most significant bit is 0 - positive number with no leading zeros.
                non_zero_position
            };
            return &digits[zeros_to_remove..];
        }

        // There were no leading zeros at all - leave as is.
        digits
    }
}

/// Compares two [`CqlVarint`] values after normalization.
///
/// # Example
///
/// ```rust
/// # use scylla_cql_core::value::CqlVarint;
/// let non_normalized_bytes = vec![0x00, 0x01];
/// let normalized_bytes = vec![0x01];
/// assert_eq!(
///     CqlVarint::from_signed_bytes_be(non_normalized_bytes),
///     CqlVarint::from_signed_bytes_be(normalized_bytes)
/// );
/// ```
impl PartialEq for CqlVarint {
    fn eq(&self, other: &Self) -> bool {
        self.as_normalized_slice() == other.as_normalized_slice()
    }
}

/// Computes the hash of normalized [`CqlVarint`].
impl std::hash::Hash for CqlVarint {
    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
        self.as_normalized_slice().hash(state)
    }
}

/// Compares two [`CqlVarintBorrowed`] values after normalization.
///
/// # Example
///
/// ```rust
/// # use scylla_cql_core::value::CqlVarintBorrowed;
/// let non_normalized_bytes = &[0x00, 0x01];
/// let normalized_bytes = &[0x01];
/// assert_eq!(
///     CqlVarintBorrowed::from_signed_bytes_be_slice(non_normalized_bytes),
///     CqlVarintBorrowed::from_signed_bytes_be_slice(normalized_bytes)
/// );
/// ```
impl PartialEq for CqlVarintBorrowed<'_> {
    fn eq(&self, other: &Self) -> bool {
        self.as_normalized_slice() == other.as_normalized_slice()
    }
}

/// Computes the hash of normalized [`CqlVarintBorrowed`].
impl std::hash::Hash for CqlVarintBorrowed<'_> {
    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
        self.as_normalized_slice().hash(state)
    }
}

#[cfg(feature = "num-bigint-03")]
impl From<num_bigint_03::BigInt> for CqlVarint {
    fn from(value: num_bigint_03::BigInt) -> Self {
        Self(value.to_signed_bytes_be())
    }
}

#[cfg(feature = "num-bigint-03")]
impl From<CqlVarint> for num_bigint_03::BigInt {
    fn from(val: CqlVarint) -> Self {
        num_bigint_03::BigInt::from_signed_bytes_be(&val.0)
    }
}

#[cfg(feature = "num-bigint-03")]
impl From<CqlVarintBorrowed<'_>> for num_bigint_03::BigInt {
    fn from(val: CqlVarintBorrowed<'_>) -> Self {
        num_bigint_03::BigInt::from_signed_bytes_be(val.0)
    }
}

#[cfg(feature = "num-bigint-04")]
impl From<num_bigint_04::BigInt> for CqlVarint {
    fn from(value: num_bigint_04::BigInt) -> Self {
        Self(value.to_signed_bytes_be())
    }
}

#[cfg(feature = "num-bigint-04")]
impl From<CqlVarint> for num_bigint_04::BigInt {
    fn from(val: CqlVarint) -> Self {
        num_bigint_04::BigInt::from_signed_bytes_be(&val.0)
    }
}

#[cfg(feature = "num-bigint-04")]
impl From<CqlVarintBorrowed<'_>> for num_bigint_04::BigInt {
    fn from(val: CqlVarintBorrowed<'_>) -> Self {
        num_bigint_04::BigInt::from_signed_bytes_be(val.0)
    }
}

/// Native CQL `decimal` representation.
///
/// Represented as a pair:
/// - a [`CqlVarint`] value
/// - 32-bit integer which determines the position of the decimal point
///
/// This struct holds owned bytes. If you wish to borrow the bytes instead,
/// see [`CqlDecimalBorrowed`] documentation.
///
/// The type is not very useful in most use cases.
/// However, users can make use of more complex types
/// such as `bigdecimal::BigDecimal` (v0.4).
/// The library support (e.g. conversion from [`CqlValue`]) for the type is
/// enabled via `bigdecimal-04` crate feature.
///
/// # DB data format
/// Notice that [constructors](CqlDecimal#impl-CqlDecimal)
/// don't perform any normalization on the provided data.
/// For more details, see [`CqlVarint`] documentation.
#[derive(Clone, PartialEq, Eq, Debug)]
pub struct CqlDecimal {
    int_val: CqlVarint,
    scale: i32,
}

/// Borrowed version of native CQL `decimal` representation.
///
/// Represented as a pair:
/// - a [`CqlVarintBorrowed`] value
/// - 32-bit integer which determines the position of the decimal point
///
/// Refer to the documentation of [`CqlDecimal`].
/// Especially, see the disclaimer about [non-normalized values](CqlDecimal#db-data-format).
#[derive(Clone, PartialEq, Eq, Debug)]
pub struct CqlDecimalBorrowed<'b> {
    int_val: CqlVarintBorrowed<'b>,
    scale: i32,
}

/// Constructors
impl CqlDecimal {
    /// Creates a [`CqlDecimal`] from an array of bytes
    /// representing [`CqlVarint`] and a 32-bit scale.
    ///
    /// See: disclaimer about [non-normalized values](CqlVarint#db-data-format).
    pub fn from_signed_be_bytes_and_exponent(bytes: Vec<u8>, scale: i32) -> Self {
        Self {
            int_val: CqlVarint::from_signed_bytes_be(bytes),
            scale,
        }
    }

    /// Creates a [`CqlDecimal`] from a slice of bytes
    /// representing [`CqlVarint`] and a 32-bit scale.
    ///
    /// See: disclaimer about [non-normalized values](CqlVarint#db-data-format).
    pub fn from_signed_be_bytes_slice_and_exponent(bytes: &[u8], scale: i32) -> Self {
        Self::from_signed_be_bytes_and_exponent(bytes.to_vec(), scale)
    }
}

/// Constructors
impl<'b> CqlDecimalBorrowed<'b> {
    /// Creates a [`CqlDecimalBorrowed`] from a slice of bytes
    /// representing [`CqlVarintBorrowed`] and a 32-bit scale.
    ///
    /// See: disclaimer about [non-normalized values](CqlVarint#db-data-format).
    pub fn from_signed_be_bytes_slice_and_exponent(bytes: &'b [u8], scale: i32) -> Self {
        Self {
            int_val: CqlVarintBorrowed::from_signed_bytes_be_slice(bytes),
            scale,
        }
    }
}

/// Conversion to raw bytes
impl CqlDecimal {
    /// Returns a slice of bytes in two's complement
    /// binary big-endian representation and a scale.
    pub fn as_signed_be_bytes_slice_and_exponent(&self) -> (&[u8], i32) {
        (self.int_val.as_signed_bytes_be_slice(), self.scale)
    }

    /// Converts [`CqlDecimal`] to an array of bytes in two's
    /// complement binary big-endian representation and a scale.
    pub fn into_signed_be_bytes_and_exponent(self) -> (Vec<u8>, i32) {
        (self.int_val.into_signed_bytes_be(), self.scale)
    }
}

/// Conversion to raw bytes
impl CqlDecimalBorrowed<'_> {
    /// Returns a slice of bytes in two's complement
    /// binary big-endian representation and a scale.
    pub fn as_signed_be_bytes_slice_and_exponent(&self) -> (&[u8], i32) {
        (self.int_val.as_signed_bytes_be_slice(), self.scale)
    }
}

#[cfg(feature = "bigdecimal-04")]
impl From<CqlDecimal> for bigdecimal_04::BigDecimal {
    fn from(value: CqlDecimal) -> Self {
        Self::from((
            bigdecimal_04::num_bigint::BigInt::from_signed_bytes_be(
                value.int_val.as_signed_bytes_be_slice(),
            ),
            value.scale as i64,
        ))
    }
}

#[cfg(feature = "bigdecimal-04")]
impl From<CqlDecimalBorrowed<'_>> for bigdecimal_04::BigDecimal {
    fn from(value: CqlDecimalBorrowed) -> Self {
        Self::from((
            bigdecimal_04::num_bigint::BigInt::from_signed_bytes_be(
                value.int_val.as_signed_bytes_be_slice(),
            ),
            value.scale as i64,
        ))
    }
}

#[cfg(feature = "bigdecimal-04")]
impl TryFrom<bigdecimal_04::BigDecimal> for CqlDecimal {
    type Error = <i64 as TryInto<i32>>::Error;

    fn try_from(value: bigdecimal_04::BigDecimal) -> Result<Self, Self::Error> {
        let (bigint, scale) = value.into_bigint_and_exponent();
        let bytes = bigint.to_signed_bytes_be();
        Ok(Self::from_signed_be_bytes_and_exponent(
            bytes,
            scale.try_into()?,
        ))
    }
}

/// Native CQL date representation that allows for a bigger range of dates (-262145-1-1 to 262143-12-31).
///
/// Represented as number of days since -5877641-06-23 i.e. 2^31 days before unix epoch.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub struct CqlDate(pub u32);

/// Native CQL timestamp representation that allows full supported timestamp range.
///
/// Represented as signed milliseconds since unix epoch.
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Debug)]
pub struct CqlTimestamp(pub i64);

/// Native CQL time representation.
///
/// Represented as nanoseconds since midnight.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub struct CqlTime(pub i64);

impl CqlDate {
    fn try_to_chrono_04_naive_date(&self) -> Result<chrono_04::NaiveDate, ValueOverflow> {
        let days_since_unix_epoch = self.0 as i64 - (1 << 31);

        // date_days is u32 then converted to i64 then we subtract 2^31;
        // Max value is 2^31, min value is -2^31. Both values can safely fit in chrono::Duration, this call won't panic
        let duration_since_unix_epoch =
            chrono_04::Duration::try_days(days_since_unix_epoch).unwrap();

        chrono_04::NaiveDate::from_yo_opt(1970, 1)
            .unwrap()
            .checked_add_signed(duration_since_unix_epoch)
            .ok_or(ValueOverflow)
    }
}

#[cfg(feature = "chrono-04")]
impl From<chrono_04::NaiveDate> for CqlDate {
    fn from(value: chrono_04::NaiveDate) -> Self {
        let unix_epoch = chrono_04::NaiveDate::from_yo_opt(1970, 1).unwrap();

        // `NaiveDate` range is -262145-01-01 to 262143-12-31
        // Both values are well within supported range
        let days = ((1 << 31) + value.signed_duration_since(unix_epoch).num_days()) as u32;

        Self(days)
    }
}

#[cfg(feature = "chrono-04")]
impl TryInto<chrono_04::NaiveDate> for CqlDate {
    type Error = ValueOverflow;

    fn try_into(self) -> Result<chrono_04::NaiveDate, Self::Error> {
        self.try_to_chrono_04_naive_date()
    }
}

impl CqlTimestamp {
    /// The earliest representable `CqlTimestamp` (milliseconds = `i64::MIN`).
    pub const MIN: CqlTimestamp = CqlTimestamp(i64::MIN);

    /// The latest representable `CqlTimestamp` (milliseconds = `i64::MAX`).
    pub const MAX: CqlTimestamp = CqlTimestamp(i64::MAX);

    /// Returns the current wall-clock time as a `CqlTimestamp`.
    ///
    /// The returned value holds the number of milliseconds since the Unix epoch
    /// (1970-01-01 00:00:00 UTC), matching the CQL `timestamp` type semantics.
    ///
    /// This is implemented in terms of [`std::time::SystemTime::now`]
    /// and is therefore **not monotonic**.
    /// Because of OS/NTP clock adjustments, a later call to `now()` can return
    /// a smaller value than an earlier call.
    /// Do not use `now()` to measure elapsed time;
    /// use [`std::time::Instant`] for that purpose instead.
    ///
    /// # Panics
    ///
    /// Panics if the system clock is so far removed from the Unix epoch
    /// (in either direction) that it cannot be represented as `i64` milliseconds.
    pub fn now() -> Self {
        match SystemTime::now().duration_since(UNIX_EPOCH) {
            Ok(d) => Self(
                i64::try_from(d.as_millis())
                    .expect("system clock is too far past the Unix epoch to fit in a CqlTimestamp"),
            ),
            Err(e) => {
                let ms = i64::try_from(e.duration().as_millis()).expect(
                    "system clock is too far before the Unix epoch to fit in a CqlTimestamp",
                );
                // `ms` is the result of converting a `u128` (always non-negative) into `i64`,
                // so it is in the range [0, i64::MAX]. Negating it therefore cannot overflow.
                Self(-ms)
            }
        }
    }

    /// Returns the amount of time elapsed from `earlier` to `self`.
    ///
    /// Mirrors [`std::time::Instant::checked_duration_since`].
    /// Returns `None` if `earlier` is actually later than `self`,
    /// instead of silently saturating at zero, since it is easy to swap
    /// the operands by mistake.
    pub fn checked_duration_since(self, earlier: CqlTimestamp) -> Option<std::time::Duration> {
        if self.0 < earlier.0 {
            return None;
        }
        // The ordering check above guarantees `self.0 >= earlier.0`,
        // so `abs_diff` gives the correct non-negative `u64` difference even
        // when it wouldn't fit in `i64` (e.g. MAX - MIN).
        Some(std::time::Duration::from_millis(self.0.abs_diff(earlier.0)))
    }

    fn try_to_chrono_04_datetime_utc(
        &self,
    ) -> Result<chrono_04::DateTime<chrono_04::Utc>, ValueOverflow> {
        use chrono_04::TimeZone;
        match chrono_04::Utc.timestamp_millis_opt(self.0) {
            chrono_04::LocalResult::Single(datetime) => Ok(datetime),
            _ => Err(ValueOverflow),
        }
    }
}

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

    /// # Panics
    ///
    /// Panics if the resulting timestamp would overflow `i64` milliseconds.
    fn add(self, rhs: std::time::Duration) -> CqlTimestamp {
        let rhs_ms = i128::try_from(rhs.as_millis()).unwrap_or(i128::MAX);
        let result = i128::from(self.0) + rhs_ms;
        CqlTimestamp(i64::try_from(result).expect("overflow when adding Duration to CqlTimestamp"))
    }
}

impl std::ops::AddAssign<std::time::Duration> for CqlTimestamp {
    /// # Panics
    ///
    /// Panics if the resulting timestamp would overflow `i64` milliseconds.
    fn add_assign(&mut self, rhs: std::time::Duration) {
        *self = *self + rhs;
    }
}

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

    /// # Panics
    ///
    /// Panics if the resulting timestamp would overflow `i64` milliseconds.
    fn sub(self, rhs: std::time::Duration) -> CqlTimestamp {
        let rhs_ms = i128::try_from(rhs.as_millis()).unwrap_or(i128::MAX);
        let result = i128::from(self.0) - rhs_ms;
        CqlTimestamp(
            i64::try_from(result).expect("overflow when subtracting Duration from CqlTimestamp"),
        )
    }
}

impl std::ops::SubAssign<std::time::Duration> for CqlTimestamp {
    /// # Panics
    ///
    /// Panics if the resulting timestamp would overflow `i64` milliseconds.
    fn sub_assign(&mut self, rhs: std::time::Duration) {
        *self = *self - rhs;
    }
}

#[cfg(feature = "chrono-04")]
impl From<chrono_04::DateTime<chrono_04::Utc>> for CqlTimestamp {
    fn from(value: chrono_04::DateTime<chrono_04::Utc>) -> Self {
        Self(value.timestamp_millis())
    }
}

#[cfg(feature = "chrono-04")]
impl TryInto<chrono_04::DateTime<chrono_04::Utc>> for CqlTimestamp {
    type Error = ValueOverflow;

    fn try_into(self) -> Result<chrono_04::DateTime<chrono_04::Utc>, Self::Error> {
        self.try_to_chrono_04_datetime_utc()
    }
}

#[cfg(feature = "chrono-04")]
impl TryFrom<chrono_04::NaiveTime> for CqlTime {
    type Error = ValueOverflow;

    fn try_from(value: chrono_04::NaiveTime) -> Result<Self, Self::Error> {
        let nanos = value
            .signed_duration_since(chrono_04::NaiveTime::MIN)
            .num_nanoseconds()
            .unwrap();

        // Value can exceed max CQL time in case of leap second
        if nanos <= 86399999999999 {
            Ok(Self(nanos))
        } else {
            Err(ValueOverflow)
        }
    }
}

#[cfg(feature = "chrono-04")]
impl TryInto<chrono_04::NaiveTime> for CqlTime {
    type Error = ValueOverflow;

    fn try_into(self) -> Result<chrono_04::NaiveTime, Self::Error> {
        let secs = (self.0 / 1_000_000_000)
            .try_into()
            .map_err(|_| ValueOverflow)?;
        let nanos = (self.0 % 1_000_000_000)
            .try_into()
            .map_err(|_| ValueOverflow)?;
        chrono_04::NaiveTime::from_num_seconds_from_midnight_opt(secs, nanos).ok_or(ValueOverflow)
    }
}

#[cfg(feature = "time-03")]
impl From<time_03::Date> for CqlDate {
    fn from(value: time_03::Date) -> Self {
        const JULIAN_DAY_OFFSET: i64 =
            (1 << 31) - time_03::OffsetDateTime::UNIX_EPOCH.date().to_julian_day() as i64;

        // Statically assert that no possible value will ever overflow
        const _: () = assert!(
            time_03::Date::MAX.to_julian_day() as i64 + JULIAN_DAY_OFFSET < u32::MAX as i64
        );
        const _: () = assert!(
            time_03::Date::MIN.to_julian_day() as i64 + JULIAN_DAY_OFFSET > u32::MIN as i64
        );

        let days = value.to_julian_day() as i64 + JULIAN_DAY_OFFSET;

        Self(days as u32)
    }
}

#[cfg(feature = "time-03")]
impl TryInto<time_03::Date> for CqlDate {
    type Error = ValueOverflow;

    fn try_into(self) -> Result<time_03::Date, Self::Error> {
        const JULIAN_DAY_OFFSET: i64 =
            (1 << 31) - time_03::OffsetDateTime::UNIX_EPOCH.date().to_julian_day() as i64;

        let julian_days = (self.0 as i64 - JULIAN_DAY_OFFSET)
            .try_into()
            .map_err(|_| ValueOverflow)?;

        time_03::Date::from_julian_day(julian_days).map_err(|_| ValueOverflow)
    }
}

#[cfg(feature = "time-03")]
impl From<time_03::OffsetDateTime> for CqlTimestamp {
    fn from(value: time_03::OffsetDateTime) -> Self {
        // Statically assert that no possible value will ever overflow. OffsetDateTime doesn't allow offset to overflow
        // the UTC PrimitiveDateTime value value
        const _: () = assert!(
            time_03::PrimitiveDateTime::MAX
                .assume_utc()
                .unix_timestamp_nanos()
                // Nanos to millis
                / 1_000_000
                < i64::MAX as i128
        );
        const _: () = assert!(
            time_03::PrimitiveDateTime::MIN
                .assume_utc()
                .unix_timestamp_nanos()
                / 1_000_000
                > i64::MIN as i128
        );

        // Edge cases were statically asserted above, checked math is not required
        Self(value.unix_timestamp() * 1000 + value.millisecond() as i64)
    }
}

#[cfg(feature = "time-03")]
impl TryInto<time_03::OffsetDateTime> for CqlTimestamp {
    type Error = ValueOverflow;

    fn try_into(self) -> Result<time_03::OffsetDateTime, Self::Error> {
        time_03::OffsetDateTime::from_unix_timestamp_nanos(self.0 as i128 * 1_000_000)
            .map_err(|_| ValueOverflow)
    }
}

#[cfg(feature = "time-03")]
impl From<time_03::Time> for CqlTime {
    fn from(value: time_03::Time) -> Self {
        let (h, m, s, n) = value.as_hms_nano();

        // no need for checked arithmetic as all these types are guaranteed to fit in i64 without overflow
        let nanos = (h as i64 * 3600 + m as i64 * 60 + s as i64) * 1_000_000_000 + n as i64;

        Self(nanos)
    }
}

#[cfg(feature = "time-03")]
impl TryInto<time_03::Time> for CqlTime {
    type Error = ValueOverflow;

    fn try_into(self) -> Result<time_03::Time, Self::Error> {
        let h = self.0 / 3_600_000_000_000;
        let m = self.0 / 60_000_000_000 % 60;
        let s = self.0 / 1_000_000_000 % 60;
        let n = self.0 % 1_000_000_000;

        time_03::Time::from_hms_nano(
            h.try_into().map_err(|_| ValueOverflow)?,
            m as u8,
            s as u8,
            n as u32,
        )
        .map_err(|_| ValueOverflow)
    }
}

/// Represents a CQL Duration value
#[derive(Clone, Debug, Copy, PartialEq, Eq)]
pub struct CqlDuration {
    /// Number of months.
    pub months: i32,
    /// Number of days.
    pub days: i32,
    /// Number of nanoseconds.
    pub nanoseconds: i64,
}

/// Represents all possible CQL values that can be returned by the database.
///
/// This type can represent a CQL value of any type. Therefore, it should be used in places
/// where dynamic capabilities are needed, while, for efficiency purposes, avoided in places
/// where the type of the value is known in the compile time.
#[derive(Clone, Debug, PartialEq)]
#[non_exhaustive]
pub enum CqlValue {
    /// ASCII-only string.
    Ascii(String),
    /// Boolean value.
    Boolean(bool),
    /// Binary data of any length.
    Blob(Vec<u8>),
    /// Counter value, represented as a 64-bit integer.
    Counter(Counter),
    /// Variable-precision decimal.
    Decimal(CqlDecimal),
    /// Days since -5877641-06-23 i.e. 2^31 days before unix epoch
    /// Can be converted to chrono::NaiveDate (-262145-1-1 to 262143-12-31) using [TryInto].
    Date(CqlDate),
    /// 64-bit IEEE-754 floating point number.
    Double(f64),
    /// A duration with nanosecond precision.
    Duration(CqlDuration),
    /// An empty value, which is distinct from null and is some DB legacy.
    Empty,
    /// 32-bit IEEE-754 floating point number.
    Float(f32),
    /// 32-bit signed integer.
    Int(i32),
    /// 64-bit signed integer.
    BigInt(i64),
    /// UTF-8 encoded string.
    Text(String),
    /// Milliseconds since unix epoch.
    Timestamp(CqlTimestamp),
    /// IPv4 or IPv6 address.
    Inet(IpAddr),
    /// A list of CQL values of the same types.
    List(Vec<CqlValue>),
    /// A map of CQL values, whose all keys have the same type
    /// and all values have the same type.
    Map(Vec<(CqlValue, CqlValue)>),
    /// A set of CQL values of the same types.
    Set(Vec<CqlValue>),
    /// A user-defined type (UDT) value.
    /// UDT is composed of fields, each with a name
    /// and an optional value of its own type.
    UserDefinedType {
        /// Keyspace the type belongs to.
        keyspace: String,
        /// Name of the user-defined type.
        name: String,
        /// Fields of the user-defined type - (name, value) pairs.
        fields: Vec<(String, Option<CqlValue>)>,
    },
    /// 16-bit signed integer.
    SmallInt(i16),
    /// 8-bit signed integer.
    TinyInt(i8),
    /// Nanoseconds since midnight.
    Time(CqlTime),
    /// Version 1 UUID, generally used as a "conflict-free" timestamp.
    Timeuuid(CqlTimeuuid),
    /// A tuple of CQL values of independent types each, where each element can be `None`
    /// if the value is null. The length of the tuple is part of its CQL type.
    Tuple(Vec<Option<CqlValue>>),
    /// Universally unique identifier (UUID) of any version.
    Uuid(Uuid),
    /// Arbitrary-precision integer.
    Varint(CqlVarint),
    /// A vector of CQL values of the same type.
    /// The length of the vector is part of its CQL type.
    Vector(Vec<CqlValue>),
}

impl CqlValue {
    /// Casts the value to ASCII string if it is of that type.
    pub fn as_ascii(&self) -> Option<&String> {
        match self {
            Self::Ascii(s) => Some(s),
            _ => None,
        }
    }

    /// Casts the value to CQL Date if it is of that type.
    pub fn as_cql_date(&self) -> Option<CqlDate> {
        match self {
            Self::Date(d) => Some(*d),
            _ => None,
        }
    }

    /// Casts the value to CQL Timestamp if it is of that type.
    pub fn as_cql_timestamp(&self) -> Option<CqlTimestamp> {
        match self {
            Self::Timestamp(i) => Some(*i),
            _ => None,
        }
    }

    /// Casts the value to CQL Time if it is of that type.
    pub fn as_cql_time(&self) -> Option<CqlTime> {
        match self {
            Self::Time(i) => Some(*i),
            _ => None,
        }
    }

    /// Casts the value to CQL Duration if it is of that type.
    pub fn as_cql_duration(&self) -> Option<CqlDuration> {
        match self {
            Self::Duration(i) => Some(*i),
            _ => None,
        }
    }

    /// Casts the value to CQL Counter if it is of that type.
    pub fn as_counter(&self) -> Option<Counter> {
        match self {
            Self::Counter(i) => Some(*i),
            _ => None,
        }
    }

    /// Casts the value to bool if it is of that type.
    pub fn as_boolean(&self) -> Option<bool> {
        match self {
            Self::Boolean(i) => Some(*i),
            _ => None,
        }
    }

    /// Casts the value to double-precision float if it is of that type.
    pub fn as_double(&self) -> Option<f64> {
        match self {
            Self::Double(d) => Some(*d),
            _ => None,
        }
    }

    /// Casts the value to UUID if it is of that type.
    pub fn as_uuid(&self) -> Option<Uuid> {
        match self {
            Self::Uuid(u) => Some(*u),
            _ => None,
        }
    }

    /// Casts the value to single-precision float if it is of that type.
    pub fn as_float(&self) -> Option<f32> {
        match self {
            Self::Float(f) => Some(*f),
            _ => None,
        }
    }

    /// Casts the value to 32-bit signed integer if it is of that type.
    pub fn as_int(&self) -> Option<i32> {
        match self {
            Self::Int(i) => Some(*i),
            _ => None,
        }
    }

    /// Casts the value to 64-bit signed integer if it is of that type.
    pub fn as_bigint(&self) -> Option<i64> {
        match self {
            Self::BigInt(i) => Some(*i),
            _ => None,
        }
    }

    /// Casts the value to 8-bit signed integer if it is of that type.
    pub fn as_tinyint(&self) -> Option<i8> {
        match self {
            Self::TinyInt(i) => Some(*i),
            _ => None,
        }
    }

    /// Casts the value to 16-bit signed integer if it is of that type.
    pub fn as_smallint(&self) -> Option<i16> {
        match self {
            Self::SmallInt(i) => Some(*i),
            _ => None,
        }
    }

    /// Casts the value to a byte sequence if it is of `blob` type.
    pub fn as_blob(&self) -> Option<&Vec<u8>> {
        match self {
            Self::Blob(v) => Some(v),
            _ => None,
        }
    }

    /// Casts the value to UTF-8 encoded string if it is of `text` type.
    pub fn as_text(&self) -> Option<&String> {
        match self {
            Self::Text(s) => Some(s),
            _ => None,
        }
    }

    /// Casts the value to CQL Timeuuid if it is of that type.
    pub fn as_timeuuid(&self) -> Option<CqlTimeuuid> {
        match self {
            Self::Timeuuid(u) => Some(*u),
            _ => None,
        }
    }

    /// Converts the value to string if it is of `ascii` or `text` type.
    pub fn into_string(self) -> Option<String> {
        match self {
            Self::Ascii(s) => Some(s),
            Self::Text(s) => Some(s),
            _ => None,
        }
    }

    /// Converts the value to a byte sequence if it is of `blob` type.
    pub fn into_blob(self) -> Option<Vec<u8>> {
        match self {
            Self::Blob(b) => Some(b),
            _ => None,
        }
    }

    /// Casts the value to an IP address if it is of `inet` type.
    pub fn as_inet(&self) -> Option<IpAddr> {
        match self {
            Self::Inet(a) => Some(*a),
            _ => None,
        }
    }

    /// Casts the value to a vec of CQL values if it is of `list` type.
    pub fn as_list(&self) -> Option<&Vec<CqlValue>> {
        match self {
            Self::List(s) => Some(s),
            _ => None,
        }
    }

    /// Casts the value to a vec of CQL values if it is of `set` type.
    pub fn as_set(&self) -> Option<&Vec<CqlValue>> {
        match self {
            Self::Set(s) => Some(s),
            _ => None,
        }
    }

    /// Casts the value to a vec of CQL values if it is of `vector` type.
    pub fn as_vector(&self) -> Option<&Vec<CqlValue>> {
        match self {
            Self::Vector(s) => Some(s),
            _ => None,
        }
    }

    /// Casts the value to a vec of pairs of CQL values if it is of `map` type,
    /// where each pair is a key-value pair.
    pub fn as_map(&self) -> Option<&Vec<(CqlValue, CqlValue)>> {
        match self {
            Self::Map(s) => Some(s),
            _ => None,
        }
    }

    /// Casts the value to a user-defined type (UDT) if it is of that type.
    /// The UDT is represented as a vector of pairs,
    /// where each pair consists of a field name and an optional (=nullable) value.
    pub fn as_udt(&self) -> Option<&Vec<(String, Option<CqlValue>)>> {
        match self {
            Self::UserDefinedType { fields, .. } => Some(fields),
            _ => None,
        }
    }

    /// Converts the value to a vector of CQL values if it is of `list`, `set`, or `vector` type.
    pub fn into_vec(self) -> Option<Vec<CqlValue>> {
        match self {
            Self::List(s) => Some(s),
            Self::Set(s) => Some(s),
            Self::Vector(s) => Some(s),
            _ => None,
        }
    }

    /// Converts the value to a vec of pairs of CQL values if it is of `map` type,
    /// where each pair is a key-value pair.
    pub fn into_pair_vec(self) -> Option<Vec<(CqlValue, CqlValue)>> {
        match self {
            Self::Map(s) => Some(s),
            _ => None,
        }
    }

    /// Converts the value to a vec of pairs if it is a user-defined type (UDT).
    /// Each pair consists of a field name and an optional (=nullable) value.
    pub fn into_udt_pair_vec(self) -> Option<Vec<(String, Option<CqlValue>)>> {
        match self {
            Self::UserDefinedType { fields, .. } => Some(fields),
            _ => None,
        }
    }

    /// Converts the value to CQL Varint if it is of that type.
    pub fn into_cql_varint(self) -> Option<CqlVarint> {
        match self {
            Self::Varint(i) => Some(i),
            _ => None,
        }
    }

    /// Converts the value to CQL Decimal if it is of that type.
    pub fn into_cql_decimal(self) -> Option<CqlDecimal> {
        match self {
            Self::Decimal(i) => Some(i),
            _ => None,
        }
    }
    // TODO
}

/// Displays a CqlValue. The syntax should resemble the CQL literal syntax
/// (but no guarantee is given that it's always the same).
impl std::fmt::Display for CqlValue {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        use crate::pretty::{
            CqlStringLiteralDisplayer, HexBytes, MaybeNullDisplayer, PairDisplayer,
        };

        match self {
            // Scalar types
            CqlValue::Ascii(a) => write!(f, "{}", CqlStringLiteralDisplayer(a))?,
            CqlValue::Text(t) => write!(f, "{}", CqlStringLiteralDisplayer(t))?,
            CqlValue::Blob(b) => write!(f, "0x{:x}", HexBytes(b))?,
            CqlValue::Empty => write!(f, "0x")?,
            CqlValue::Decimal(d) => {
                let (bytes, scale) = d.as_signed_be_bytes_slice_and_exponent();
                write!(
                    f,
                    "blobAsDecimal(0x{:x}{:x})",
                    HexBytes(&scale.to_be_bytes()),
                    HexBytes(bytes)
                )?
            }
            CqlValue::Float(fl) => write!(f, "{fl}")?,
            CqlValue::Double(d) => write!(f, "{d}")?,
            CqlValue::Boolean(b) => write!(f, "{b}")?,
            CqlValue::Int(i) => write!(f, "{i}")?,
            CqlValue::BigInt(bi) => write!(f, "{bi}")?,
            CqlValue::Inet(i) => write!(f, "'{i}'")?,
            CqlValue::SmallInt(si) => write!(f, "{si}")?,
            CqlValue::TinyInt(ti) => write!(f, "{ti}")?,
            CqlValue::Varint(vi) => write!(
                f,
                "blobAsVarint(0x{:x})",
                HexBytes(vi.as_signed_bytes_be_slice())
            )?,
            CqlValue::Counter(c) => write!(f, "{}", c.0)?,
            CqlValue::Date(d) => {
                // TODO: chrono::NaiveDate does not handle the whole range
                // supported by the `date` datatype
                match d.try_to_chrono_04_naive_date() {
                    Ok(d) => write!(f, "'{d}'")?,
                    Err(_) => f.write_str("<date out of representable range>")?,
                }
            }
            CqlValue::Duration(d) => write!(f, "{}mo{}d{}ns", d.months, d.days, d.nanoseconds)?,
            CqlValue::Time(CqlTime(t)) => {
                write!(
                    f,
                    "'{:02}:{:02}:{:02}.{:09}'",
                    t / 3_600_000_000_000,
                    t / 60_000_000_000 % 60,
                    t / 1_000_000_000 % 60,
                    t % 1_000_000_000,
                )?;
            }
            CqlValue::Timestamp(ts) => match ts.try_to_chrono_04_datetime_utc() {
                Ok(d) => write!(f, "{}", d.format("'%Y-%m-%d %H:%M:%S%.3f%z'"))?,
                Err(_) => f.write_str("<timestamp out of representable range>")?,
            },
            CqlValue::Timeuuid(t) => write!(f, "{t}")?,
            CqlValue::Uuid(u) => write!(f, "{u}")?,

            // Compound types
            CqlValue::Tuple(t) => {
                f.write_str("(")?;
                t.iter()
                    .map(|x| MaybeNullDisplayer(x.as_ref()))
                    .safe_format(",")
                    .fmt(f)?;
                f.write_str(")")?;
            }
            CqlValue::List(v) | CqlValue::Vector(v) => {
                f.write_str("[")?;
                v.iter().safe_format(",").fmt(f)?;
                f.write_str("]")?;
            }
            CqlValue::Set(v) => {
                f.write_str("{")?;
                v.iter().safe_format(",").fmt(f)?;
                f.write_str("}")?;
            }
            CqlValue::Map(m) => {
                f.write_str("{")?;
                m.iter()
                    .map(|(k, v)| PairDisplayer(k, v))
                    .safe_format(",")
                    .fmt(f)?;
                f.write_str("}")?;
            }
            CqlValue::UserDefinedType {
                keyspace: _,
                name: _,
                fields,
            } => {
                f.write_str("{")?;
                fields
                    .iter()
                    .map(|(k, v)| PairDisplayer(k, MaybeNullDisplayer(v.as_ref())))
                    .safe_format(",")
                    .fmt(f)?;
                f.write_str("}")?;
            }
        }
        Ok(())
    }
}

/// A row in a CQL result set, containing a vector of columns.
/// Each column can be either a `CqlValue` or `None` if the column
/// is null.
///
/// This type can represent any row:
/// - with any number of columns,
/// - with any column types.
///
/// Therefore, this type should be used in places where dynamic capabilities are needed,
/// while, for efficiency purposes, avoided in places where the row structure is known at compile time.
#[derive(Debug, Default, PartialEq)]
pub struct Row {
    /// A vector of columns in the row.
    ///
    /// Each column is represented as an `Option<CqlValue>`, where `None` indicates a null value.
    pub columns: Vec<Option<CqlValue>>,
}

#[cfg(test)]
mod tests {
    use std::str::FromStr as _;
    use std::time::Duration;

    use super::*;

    #[test]
    fn timeuuid_msb_byte_order() {
        let uuid = CqlTimeuuid::from_str("00010203-0405-0607-0809-0a0b0c0d0e0f").unwrap();

        assert_eq!(0x0607040500010203, uuid.msb());
    }

    #[test]
    fn timeuuid_msb_clears_version_bits() {
        // UUID version nibble should be cleared
        let uuid = CqlTimeuuid::from_str("ffffffff-ffff-ffff-ffff-ffffffffffff").unwrap();

        assert_eq!(0x0fffffffffffffff, uuid.msb());
    }

    #[test]
    fn timeuuid_lsb_byte_order() {
        let uuid = CqlTimeuuid::from_str("00010203-0405-0607-0809-0a0b0c0d0e0f").unwrap();

        assert_eq!(0x08090a0b0c0d0e0f, uuid.lsb());
    }

    #[test]
    fn timeuuid_lsb_modifies_no_bits() {
        let uuid = CqlTimeuuid::from_str("ffffffff-ffff-ffff-ffff-ffffffffffff").unwrap();

        assert_eq!(0xffffffffffffffff, uuid.lsb());
    }

    #[test]
    fn timeuuid_nil() {
        let uuid = CqlTimeuuid::nil();

        assert_eq!(0x0000000000000000, uuid.msb());
        assert_eq!(0x0000000000000000, uuid.lsb());
    }

    #[test]
    fn test_cql_value_displayer() {
        assert_eq!(format!("{}", CqlValue::Boolean(true)), "true");
        assert_eq!(format!("{}", CqlValue::Int(123)), "123");
        assert_eq!(
            format!(
                "{}",
                // 123.456
                CqlValue::Decimal(CqlDecimal::from_signed_be_bytes_and_exponent(
                    vec![0x01, 0xE2, 0x40],
                    3
                ))
            ),
            "blobAsDecimal(0x0000000301e240)"
        );
        assert_eq!(format!("{}", CqlValue::Float(12.75)), "12.75");
        assert_eq!(
            format!("{}", CqlValue::Text("Ala ma kota".to_owned())),
            "'Ala ma kota'"
        );
        assert_eq!(
            format!("{}", CqlValue::Text("Foo's".to_owned())),
            "'Foo''s'"
        );

        // Time types are the most tricky
        assert_eq!(
            format!("{}", CqlValue::Date(CqlDate(40 + (1 << 31)))),
            "'1970-02-10'"
        );
        assert_eq!(
            format!(
                "{}",
                CqlValue::Duration(CqlDuration {
                    months: 1,
                    days: 2,
                    nanoseconds: 3,
                })
            ),
            "1mo2d3ns"
        );
        let t = chrono_04::NaiveTime::from_hms_nano_opt(6, 5, 4, 123)
            .unwrap()
            .signed_duration_since(chrono_04::NaiveTime::MIN);
        let t = t.num_nanoseconds().unwrap();
        assert_eq!(
            format!("{}", CqlValue::Time(CqlTime(t))),
            "'06:05:04.000000123'"
        );

        let t = chrono_04::NaiveDate::from_ymd_opt(2005, 4, 2)
            .unwrap()
            .and_time(chrono_04::NaiveTime::from_hms_opt(19, 37, 42).unwrap());
        assert_eq!(
            format!(
                "{}",
                CqlValue::Timestamp(CqlTimestamp(
                    t.signed_duration_since(chrono_04::NaiveDateTime::default())
                        .num_milliseconds()
                ))
            ),
            "'2005-04-02 19:37:42.000+0000'"
        );

        // Compound types
        let list_or_set = vec![CqlValue::Int(1), CqlValue::Int(3), CqlValue::Int(2)];
        assert_eq!(
            format!("{}", CqlValue::List(list_or_set.clone())),
            "[1,3,2]"
        );
        assert_eq!(format!("{}", CqlValue::Set(list_or_set.clone())), "{1,3,2}");

        let tuple: Vec<_> = list_or_set
            .into_iter()
            .map(Some)
            .chain(std::iter::once(None))
            .collect();
        assert_eq!(format!("{}", CqlValue::Tuple(tuple)), "(1,3,2,null)");

        let map = vec![
            (CqlValue::Text("foo".to_owned()), CqlValue::Int(123)),
            (CqlValue::Text("bar".to_owned()), CqlValue::Int(321)),
        ];
        assert_eq!(format!("{}", CqlValue::Map(map)), "{'foo':123,'bar':321}");

        let fields = vec![
            ("foo".to_owned(), Some(CqlValue::Int(123))),
            ("bar".to_owned(), Some(CqlValue::Int(321))),
        ];
        assert_eq!(
            format!(
                "{}",
                CqlValue::UserDefinedType {
                    keyspace: "ks".to_owned(),
                    name: "typ".to_owned(),
                    fields,
                }
            ),
            "{foo:123,bar:321}"
        );
    }

    #[test]
    fn cql_timestamp_sentinels() {
        assert_eq!(CqlTimestamp::MIN.0, i64::MIN);
        assert_eq!(CqlTimestamp::MAX.0, i64::MAX);
    }

    #[test]
    fn cql_timestamp_add_duration() {
        let epoch = CqlTimestamp(0);
        assert_eq!(epoch + Duration::from_millis(1_000), CqlTimestamp(1_000));
        assert_eq!(epoch + Duration::from_secs(1), CqlTimestamp(1_000));

        let t = CqlTimestamp(1_000);
        assert_eq!(t + Duration::from_millis(500), CqlTimestamp(1_500));
    }

    #[test]
    #[should_panic]
    fn cql_timestamp_add_duration_panics_on_overflow() {
        let _ = CqlTimestamp::MAX + Duration::from_millis(1);
    }

    #[test]
    fn cql_timestamp_add_duration_does_not_panic_when_result_fits() {
        assert_eq!(
            CqlTimestamp::MIN + Duration::from_millis(u64::MAX),
            CqlTimestamp::MAX
        );
    }

    #[test]
    fn cql_timestamp_sub_duration() {
        let t = CqlTimestamp(2_000);
        assert_eq!(t - Duration::from_millis(500), CqlTimestamp(1_500));
        assert_eq!(t - Duration::from_secs(1), CqlTimestamp(1_000));
        assert_eq!(t - Duration::from_millis(2_000), CqlTimestamp(0));
    }

    #[test]
    #[should_panic]
    fn cql_timestamp_sub_duration_panics_on_overflow() {
        let _ = CqlTimestamp::MIN - Duration::from_millis(1);
    }

    #[test]
    fn cql_timestamp_sub_duration_does_not_panic_when_result_fits() {
        assert_eq!(
            CqlTimestamp::MAX - Duration::from_millis(u64::MAX),
            CqlTimestamp::MIN
        );
    }

    #[test]
    fn cql_timestamp_add_assign_and_sub_assign() {
        let mut t = CqlTimestamp(1_000);
        t += Duration::from_millis(500);
        assert_eq!(t, CqlTimestamp(1_500));
        t -= Duration::from_millis(1_500);
        assert_eq!(t, CqlTimestamp(0));
    }

    #[test]
    fn cql_timestamp_checked_duration_since() {
        let later = CqlTimestamp(3_000);
        let earlier = CqlTimestamp(1_000);
        assert_eq!(
            later.checked_duration_since(earlier),
            Some(Duration::from_millis(2_000))
        );
        assert_eq!(
            later.checked_duration_since(CqlTimestamp(0)),
            Some(Duration::from_millis(3_000))
        );
    }

    #[test]
    fn cql_timestamp_checked_duration_since_none_when_earlier_is_later() {
        let earlier = CqlTimestamp(1_000);
        let later = CqlTimestamp(3_000);
        assert_eq!(earlier.checked_duration_since(later), None);
    }

    #[test]
    fn cql_timestamp_checked_duration_since_no_overflow_on_extreme_range() {
        let diff = CqlTimestamp::MAX
            .checked_duration_since(CqlTimestamp::MIN)
            .unwrap();
        // MAX - MIN = i64::MAX - i64::MIN = u64::MAX, which is the full
        // non-negative range and must not be truncated to i64::MAX.
        assert_eq!(diff, Duration::from_millis(u64::MAX));
    }
}