exarrow-rs 0.29.0

//! Shared type conversion utilities.
//!
//! This module provides common parsing functions for date, timestamp, and decimal
//! values used across the codebase. The functions return simple `Result<T, String>`
//! to allow callers to wrap errors into their specific error types.

use arrow::datatypes::{DataType, IntervalMonthDayNano, TimeUnit};
use chrono::Datelike;

use crate::types::ExasolType;

const MICROSECOND_DIGITS: usize = 6;
const NANOSECOND_DIGITS: usize = 9;
const MONTHS_PER_YEAR: i64 = 12;
const MILLIS_PER_DAY: i64 = 86_400_000;
const MILLIS_PER_SECOND: u64 = 1_000;
const MILLIS_PER_MINUTE: u64 = 60 * MILLIS_PER_SECOND;
const MILLIS_PER_HOUR: u64 = 60 * MILLIS_PER_MINUTE;
const SECONDS_PER_MINUTE: u64 = 60;
const MINUTES_PER_HOUR: u64 = 60;
/// The most days that an INTERVAL DAY TO SECOND column of leading precision 9 holds.
const MAX_INTERVAL_DAYS: u64 = 999_999_999;
const NANOS_PER_MILLISECOND: i64 = 1_000_000;
const NANOS_PER_SECOND: i64 = 1_000_000_000;
const NANOS_PER_MINUTE: i64 = 60 * NANOS_PER_SECOND;
const NANOS_PER_HOUR: i64 = 60 * NANOS_PER_MINUTE;

/// Parses a date string (YYYY-MM-DD) to days since Unix epoch (1970-01-01).
///
/// # Arguments
/// * `date_str` - A date string in YYYY-MM-DD format
///
/// # Returns
/// * `Ok(i32)` - Days since Unix epoch
/// * `Err(String)` - Description of the parsing error
///
pub fn parse_date_to_days(date_str: &str) -> Result<i32, String> {
    let parts: Vec<&str> = date_str.split('-').collect();
    if parts.len() != 3 {
        return Err(format!(
            "Invalid date format: {} (expected YYYY-MM-DD)",
            date_str
        ));
    }

    let year: i32 = parts[0]
        .parse()
        .map_err(|_| format!("Invalid year: {}", parts[0]))?;
    let month: u32 = parts[1]
        .parse()
        .map_err(|_| format!("Invalid month: {}", parts[1]))?;
    let day: u32 = parts[2]
        .parse()
        .map_err(|_| format!("Invalid day: {}", parts[2]))?;

    if !(1..=12).contains(&month) {
        return Err(format!("Month out of range: {}", month));
    }
    if !(1..=31).contains(&day) {
        return Err(format!("Day out of range: {}", day));
    }

    Ok(ymd_to_days(year, month, day))
}

/// Converts a year/month/day triple to days since 1970-01-01 in the proleptic Gregorian calendar, for any year.
/// Floor division keeps the leap-day count correct before 1970; the function does no validation.
pub fn ymd_to_days(year: i32, month: u32, day: u32) -> i32 {
    let days_from_year = (year - 1970) * 365 + (year - 1969).div_euclid(4)
        - (year - 1901).div_euclid(100)
        + (year - 1601).div_euclid(400);

    let days_from_month = match month {
        1 => 0,
        2 => 31,
        3 => 59,
        4 => 90,
        5 => 120,
        6 => 151,
        7 => 181,
        8 => 212,
        9 => 243,
        10 => 273,
        11 => 304,
        12 => 334,
        _ => 0,
    };

    let is_leap_year = (year % 4 == 0 && year % 100 != 0) || (year % 400 == 0);
    let leap_adjustment = if month > 2 && is_leap_year { 1 } else { 0 };

    days_from_year + days_from_month + day as i32 - 1 + leap_adjustment
}

/// Converts a pre-decoded timestamp (y, m, d, h, min, s, nanos) to microseconds
/// since Unix epoch. Nanoseconds are truncated to microsecond precision.
pub fn ymd_hms_nanos_to_micros(
    year: i32,
    month: u32,
    day: u32,
    hour: u64,
    minute: u64,
    second: u64,
    nanos: i32,
) -> i64 {
    let days = ymd_to_days(year, month, day);
    let mut micros = days as i64 * 86_400 * 1_000_000;
    micros += hour as i64 * 3_600 * 1_000_000;
    micros += minute as i64 * 60 * 1_000_000;
    micros += second as i64 * 1_000_000;
    micros += nanos as i64 / 1_000;
    micros
}

const EXASOL_YEARS: std::ops::RangeInclusive<i32> = 1..=9999;
const DAYS_FROM_CE_TO_EPOCH: i32 = 719_163;

/// Formats Date32 days since 1970-01-01 as `YYYY-MM-DD` in the proleptic Gregorian calendar.
///
/// Fails for a date outside the years 1 to 9999.
pub fn format_date32(days: i32) -> Result<String, String> {
    let date = days
        .checked_add(DAYS_FROM_CE_TO_EPOCH)
        .and_then(chrono::NaiveDate::from_num_days_from_ce_opt)
        .filter(|date| EXASOL_YEARS.contains(&date.year()))
        .ok_or_else(|| {
            format!(
                "Date32 value {days} is outside 0001-01-01 to 9999-12-31, the range Exasol accepts"
            )
        })?;
    Ok(date.format("%Y-%m-%d").to_string())
}

/// Formats the day that holds the Date64 instant, in milliseconds since 1970-01-01, as `YYYY-MM-DD`.
///
/// Fails for a date outside the years 1 to 9999.
pub fn format_date64(milliseconds: i64) -> Result<String, String> {
    let outside_range = || {
        format!(
            "Date64 value {milliseconds} is outside 0001-01-01 to 9999-12-31, the range Exasol accepts"
        )
    };
    // Flooring keeps a pre-epoch value on the day that holds it.
    let days =
        i32::try_from(milliseconds.div_euclid(MILLIS_PER_DAY)).map_err(|_| outside_range())?;
    format_date32(days).map_err(|_| outside_range())
}

/// Formats microseconds since the epoch as `YYYY-MM-DD HH:MM:SS.ffffff`.
///
/// Fails for an instant outside the years 1 to 9999.
pub fn format_timestamp_micros(micros: i64) -> Result<String, String> {
    let instant = chrono::DateTime::from_timestamp_micros(micros)
        .filter(|instant| EXASOL_YEARS.contains(&instant.year()))
        .ok_or_else(|| {
            format!(
                "Timestamp value {micros} microseconds is outside 0001-01-01 00:00:00 to 9999-12-31 23:59:59.999999, the range Exasol accepts"
            )
        })?;
    Ok(instant.format("%Y-%m-%d %H:%M:%S%.6f").to_string())
}

/// Converts the value to whole microseconds, flooring a nanosecond value toward the earlier instant.
/// Fails when the conversion overflows `i64` or the instant is outside the years 1 to 9999.
pub fn format_timestamp(unit: &TimeUnit, value: i64) -> Result<String, String> {
    format_timestamp_micros(timestamp_unit_to_micros(unit, value)?).map_err(|_| {
        format!(
            "Timestamp value {value} in unit {unit:?} is outside 0001-01-01 00:00:00 to 9999-12-31 23:59:59.999999, the range Exasol accepts"
        )
    })
}

fn timestamp_unit_to_micros(unit: &TimeUnit, value: i64) -> Result<i64, String> {
    let micros = match unit {
        TimeUnit::Second => value.checked_mul(1_000_000),
        TimeUnit::Millisecond => value.checked_mul(1_000),
        TimeUnit::Microsecond => Some(value),
        // Rounds a pre-epoch value down to the microsecond, as a positive one already is.
        TimeUnit::Nanosecond => Some(value.div_euclid(1_000)),
    };
    micros.ok_or_else(|| {
        format!("Timestamp value {value} in unit {unit:?} overflows when converted to microseconds")
    })
}

/// Accepted form of DATE parameter text, quoted in rejection messages.
pub const DATE_TEXT_FORM: &str = "YYYY-MM-DD";
/// Accepted form of TIMESTAMP parameter text, quoted in rejection messages.
pub const TIMESTAMP_TEXT_FORM: &str = "YYYY-MM-DD HH:MI:SS.FFFFFFFFF";

/// Date and time fields exactly as written in parameter text, with no calendar check.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TemporalFields {
    pub year: u16,
    pub month: u8,
    pub day: u8,
    pub hour: u8,
    pub minute: u8,
    pub second: u8,
    pub nanos: u32,
}

fn digits(part: &str, max_len: usize) -> Option<u32> {
    if part.is_empty() || part.len() > max_len || !part.bytes().all(|b| b.is_ascii_digit()) {
        return None;
    }
    part.parse().ok()
}

fn parse_temporal_fields(text: &str) -> Option<TemporalFields> {
    let text = text.trim_matches(' ');
    let (date, time) = match text.split_once(' ') {
        Some((date, time)) => (date, Some(time.trim_start_matches(' '))),
        None => (text, None),
    };
    let mut date_parts = date.split('-');
    let year = digits(date_parts.next()?, 4)?;
    let month = digits(date_parts.next()?, 2)?;
    let day = digits(date_parts.next()?, 2)?;
    if date_parts.next().is_some() {
        return None;
    }
    let mut fields = TemporalFields {
        year: year as u16,
        month: month as u8,
        day: day as u8,
        hour: 0,
        minute: 0,
        second: 0,
        nanos: 0,
    };
    if let Some(time) = time {
        let mut parts = time.splitn(3, ':');
        fields.hour = digits(parts.next()?, 2)? as u8;
        if let Some(minute) = parts.next() {
            fields.minute = digits(minute, 2)? as u8;
        }
        if let Some(seconds) = parts.next() {
            let (sec, frac) = seconds.split_once('.').unwrap_or((seconds, ""));
            fields.second = digits(sec, 2)? as u8;
            if frac.len() > 9 || !frac.bytes().all(|b| b.is_ascii_digit()) {
                return None;
            }
            fields.nanos = format!("{frac:0<9}").parse().ok()?;
        }
    }
    Some(fields)
}

/// Parses TIMESTAMP parameter text: `Y-M-D`, optionally followed by spaces and `h`, `h:m`, `h:m:s`
/// or `h:m:s.f` (up to 9 fraction digits). Leading and trailing spaces are ignored.
pub fn parse_timestamp_text(text: &str) -> Result<TemporalFields, String> {
    parse_temporal_fields(text).ok_or_else(|| {
        format!("Cannot read '{text}' as a timestamp, expected {TIMESTAMP_TEXT_FORM}")
    })
}

/// Parses DATE parameter text in the forms of [`parse_timestamp_text`]. A DATE keeps only the
/// date fields; the time fields must stay within the bounds Exasol's text cast accepts.
pub fn parse_date_text(text: &str) -> Result<TemporalFields, String> {
    parse_temporal_fields(text)
        .filter(|f| f.hour <= 23 && f.minute <= 60 && f.second <= 60)
        .ok_or_else(|| format!("Cannot read '{text}' as a date, expected {DATE_TEXT_FORM}"))
}

/// Parses a timestamp string to microseconds since Unix epoch.
///
/// Supports formats:
/// - "YYYY-MM-DD" (date only, time defaults to 00:00:00)
/// - "YYYY-MM-DD HH:MM:SS" (date and time)
/// - "YYYY-MM-DD HH:MM:SS.ffffff" (date, time, and fractional seconds)
///
/// # Arguments
/// * `timestamp_str` - A timestamp string
///
/// # Returns
/// * `Ok(i64)` - Microseconds since Unix epoch
/// * `Err(String)` - Description of the parsing error
pub fn parse_timestamp_to_micros(timestamp_str: &str) -> Result<i64, String> {
    let segments: Vec<&str> = timestamp_str.split(' ').collect();

    let days = parse_date_to_days(segments[0])?;
    let date_micros = days as i64 * 86400 * 1_000_000;

    match segments.get(1) {
        Some(time_str) => Ok(date_micros + parse_time_of_day_to_micros(time_str)?),
        None => Ok(date_micros),
    }
}

/// Converts a "HH:MM[:SS[.ffffff]]" time-of-day field to microseconds.
///
/// A field with fewer than two colon-separated components carries no
/// recognizable time and contributes zero, matching Exasol's tolerance for
/// date-only timestamp literals.
fn parse_time_of_day_to_micros(time_str: &str) -> Result<i64, String> {
    let time_parts: Vec<&str> = time_str.split(':').collect();
    if time_parts.len() < 2 {
        return Ok(0);
    }

    let hours: i64 = time_parts[0]
        .parse()
        .map_err(|_| format!("Invalid hour: {}", time_parts[0]))?;
    let minutes: i64 = time_parts[1]
        .parse()
        .map_err(|_| format!("Invalid minute: {}", time_parts[1]))?;

    let mut micros = hours * 3600 * 1_000_000 + minutes * 60 * 1_000_000;
    if let Some(seconds_field) = time_parts.get(2) {
        micros += parse_seconds_field_to_micros(seconds_field)?;
    }
    Ok(micros)
}

/// Converts a "SS[.ffffff]" seconds field to microseconds.
fn parse_seconds_field_to_micros(seconds_field: &str) -> Result<i64, String> {
    let sec_parts: Vec<&str> = seconds_field.split('.').collect();
    let seconds: i64 = sec_parts[0]
        .parse()
        .map_err(|_| format!("Invalid second: {}", sec_parts[0]))?;

    let mut micros = seconds * 1_000_000;
    if let Some(fraction) = sec_parts.get(1) {
        micros += fractional_seconds_to_micros(fraction);
    }
    Ok(micros)
}

/// Converts the digits following a seconds field's decimal point to whole
/// microseconds, right-padding shorter fractions and truncating longer ones.
///
/// Exasol emits fractional seconds with a precision that varies by column, so
/// every caller has to normalize to a fixed width; this keeps that rule in one
/// place. Digits that do not parse count as zero rather than failing, because a
/// malformed fraction never invalidates the whole-second part of a value.
pub(crate) fn fractional_seconds_to_micros(fraction: &str) -> i64 {
    fractional_seconds_to_units(fraction, MICROSECOND_DIGITS)
}

/// Nanosecond-precision counterpart of `fractional_seconds_to_micros`, used for
/// Exasol INTERVAL DAY TO SECOND values.
pub(crate) fn fractional_seconds_to_nanos(fraction: &str) -> i64 {
    fractional_seconds_to_units(fraction, NANOSECOND_DIGITS)
}

/// Parses the leading digits of `fraction` and scales the result to
/// `unit_digits` places, without allocating: a shorter fraction is scaled up
/// by the missing trailing zeros instead of being padded and re-parsed as a
/// new string.
fn fractional_seconds_to_units(fraction: &str, unit_digits: usize) -> i64 {
    if fraction.len() > unit_digits {
        return fraction[..unit_digits].parse::<i64>().unwrap_or(0);
    }
    let Ok(value) = fraction.parse::<i64>() else {
        return 0;
    };
    let missing_digits = unit_digits - fraction.len();
    value.saturating_mul(10i64.pow(missing_digits as u32))
}

/// Parses the text of an Exasol INTERVAL value into an Arrow `IntervalMonthDayNano`.
///
/// Exasol sends YEAR TO MONTH as `+YY-MM` and DAY TO SECOND as
/// `+DD HH:MM:SS.fff`, so the `-` between years and months, which only the
/// first form holds after its sign, tells them apart. The sign applies to every
/// component. A month count outside the `i32` range is an error rather than a
/// wrapped value, because INTERVAL YEAR(9) reaches 999,999,999 years.
pub fn parse_interval(text: &str) -> Result<IntervalMonthDayNano, String> {
    let (negative, body) = match text.as_bytes().first() {
        Some(b'-') => (true, &text[1..]),
        Some(b'+') => (false, &text[1..]),
        _ => (false, text),
    };
    if body.contains('-') {
        let months = parse_interval_months(body, text)?;
        // The sign applies before the i32 check, because i32::MIN months has no positive i32.
        let signed_months = if negative { -months } else { months };
        let months = i32::try_from(signed_months)
            .map_err(|_| format!("Interval month count out of range: {text}"))?;
        return Ok(IntervalMonthDayNano::new(months, 0, 0));
    }
    let (days, nanoseconds) = parse_interval_days_and_nanos(body)?;
    Ok(if negative {
        IntervalMonthDayNano::new(0, -days, -nanoseconds)
    } else {
        IntervalMonthDayNano::new(0, days, nanoseconds)
    })
}

fn parse_interval_months(body: &str, text: &str) -> Result<i64, String> {
    let parts: Vec<&str> = body.split('-').collect();
    if parts.len() != 2 {
        return Err(format!("Invalid interval format: {text}"));
    }
    let years: i64 = parts[0]
        .parse()
        .map_err(|_| format!("Invalid years: {}", parts[0]))?;
    let months: i64 = parts[1]
        .parse()
        .map_err(|_| format!("Invalid months: {}", parts[1]))?;
    years
        .checked_mul(MONTHS_PER_YEAR)
        .and_then(|total| total.checked_add(months))
        .ok_or_else(|| format!("Interval month count out of range: {text}"))
}

/// Writes an Arrow Interval(YearMonth) month count as Exasol INTERVAL YEAR TO MONTH text,
/// such as `5-03` or `-5-03`.
///
/// The text is the form that [`parse_interval`] reads, with no `+` sign and no zero-padded
/// years. The count is split in `i64`, so `i32::MIN` months becomes `-178956970-08`.
#[must_use]
pub fn format_interval_year_to_month(months: i32) -> String {
    let months = i64::from(months);
    let sign = if months < 0 { "-" } else { "" };
    let magnitude = months.abs();
    format!(
        "{sign}{}-{:02}",
        magnitude / MONTHS_PER_YEAR,
        magnitude % MONTHS_PER_YEAR
    )
}

/// Writes a day count and a time part in milliseconds as Exasol INTERVAL DAY TO SECOND
/// text, such as `2 12:50:10.123` or `-2 12:50:10.123`.
///
/// The days and the time part add up as days of 86,400,000 ms, so one sign applies to the
/// whole value, as in the form that [`parse_interval`] reads.
///
/// # Errors
/// Fails for a value of more than 999,999,999 days, the most that a column of leading
/// precision 9 holds.
pub fn format_interval_day_to_second(days: i32, milliseconds: i64) -> Result<String, String> {
    let out_of_range = || {
        format!(
            "Interval value of {days} days and {milliseconds} milliseconds is out of range, because INTERVAL DAY(9) TO SECOND holds at most 999,999,999 days"
        )
    };
    let total = i64::from(days)
        .checked_mul(MILLIS_PER_DAY)
        .and_then(|day_millis| day_millis.checked_add(milliseconds))
        .ok_or_else(out_of_range)?;
    let magnitude = total.unsigned_abs();
    let whole_days = magnitude / MILLIS_PER_DAY as u64;
    if whole_days > MAX_INTERVAL_DAYS {
        return Err(out_of_range());
    }
    let time = magnitude % MILLIS_PER_DAY as u64;
    let sign = if total < 0 { "-" } else { "" };
    Ok(format!(
        "{sign}{whole_days} {:02}:{:02}:{:02}.{:03}",
        time / MILLIS_PER_HOUR,
        time / MILLIS_PER_MINUTE % MINUTES_PER_HOUR,
        time / MILLIS_PER_SECOND % SECONDS_PER_MINUTE,
        time % MILLIS_PER_SECOND
    ))
}

/// Writes an Arrow Interval(MonthDayNano) value as Exasol INTERVAL DAY TO SECOND text, in
/// the form of [`format_interval_day_to_second`].
///
/// # Errors
/// Fails for a month component, which INTERVAL DAY TO SECOND does not have, for nanoseconds
/// that are not whole milliseconds, which Exasol cannot store without loss, and for more
/// than 999,999,999 days.
pub fn format_interval_month_day_nano(value: IntervalMonthDayNano) -> Result<String, String> {
    let IntervalMonthDayNano {
        months,
        days,
        nanoseconds,
    } = value;
    let described =
        format!("Interval value of {months} months, {days} days, and {nanoseconds} nanoseconds");
    if months != 0 {
        return Err(format!(
            "{described} has a month component, but Exasol INTERVAL DAY TO SECOND has no month component"
        ));
    }
    if nanoseconds % NANOS_PER_MILLISECOND != 0 {
        return Err(format!(
            "{described} has a fraction of a millisecond, but Exasol stores INTERVAL DAY TO SECOND values in whole milliseconds"
        ));
    }
    format_interval_day_to_second(days, nanoseconds / NANOS_PER_MILLISECOND)
}

fn parse_interval_days_and_nanos(body: &str) -> Result<(i32, i64), String> {
    let mut segments = body.split(' ');
    let days_text = segments.next().unwrap_or_default();
    let days: i32 = days_text
        .parse()
        .map_err(|_| format!("Invalid days: {days_text}"))?;
    let nanos = match segments.next() {
        Some(time_text) => parse_interval_time_to_nanos(time_text)?,
        None => 0,
    };
    Ok((days, nanos))
}

/// A time part with fewer than two colon-separated fields carries no
/// recognizable time and contributes zero, matching Exasol's day-only literals.
fn parse_interval_time_to_nanos(time_text: &str) -> Result<i64, String> {
    let fields: Vec<&str> = time_text.split(':').collect();
    if fields.len() < 2 {
        return Ok(0);
    }
    let hours: i64 = fields[0]
        .parse()
        .map_err(|_| format!("Invalid hours: {}", fields[0]))?;
    let minutes: i64 = fields[1]
        .parse()
        .map_err(|_| format!("Invalid minutes: {}", fields[1]))?;
    let mut nanos = hours
        .checked_mul(NANOS_PER_HOUR)
        .zip(minutes.checked_mul(NANOS_PER_MINUTE))
        .and_then(|(h, m)| h.checked_add(m))
        .ok_or_else(|| format!("Interval time out of range: {time_text}"))?;
    if let Some(seconds_text) = fields.get(2) {
        nanos = nanos
            .checked_add(parse_interval_seconds_to_nanos(seconds_text)?)
            .ok_or_else(|| format!("Interval time out of range: {time_text}"))?;
    }
    Ok(nanos)
}

fn parse_interval_seconds_to_nanos(seconds_text: &str) -> Result<i64, String> {
    let mut parts = seconds_text.split('.');
    let whole_text = parts.next().unwrap_or_default();
    let seconds: i64 = whole_text
        .parse()
        .map_err(|_| format!("Invalid seconds: {whole_text}"))?;
    let whole_nanos = seconds
        .checked_mul(NANOS_PER_SECOND)
        .ok_or_else(|| format!("Interval seconds out of range: {seconds_text}"))?;
    whole_nanos
        .checked_add(parts.next().map_or(0, fractional_seconds_to_nanos))
        .ok_or_else(|| format!("Interval seconds out of range: {seconds_text}"))
}

/// Parses a decimal string to i128 with the given scale.
///
/// Supports formats:
/// - "123" (integer)
/// - "123.45" (decimal)
/// - "-123.45" (negative decimal)
///
/// # Arguments
/// * `value_str` - A decimal string
/// * `scale` - The scale (number of decimal places) to use
///
/// # Returns
/// * `Ok(i128)` - The decimal value scaled by 10^scale
/// * `Err(String)` - Description of the parsing error
pub fn parse_decimal_to_i128(value_str: &str, scale: i8) -> Result<i128, String> {
    // Parse string representation of decimal
    // Format: "123.45" or "123" or "-123.45"
    let parts: Vec<&str> = value_str.split('.').collect();

    let (integer_part, decimal_part) = match parts.len() {
        1 => (parts[0], ""),
        2 => (parts[0], parts[1]),
        _ => return Err(format!("Invalid decimal format: {}", value_str)),
    };

    // Parse the integer part
    let mut result: i128 = integer_part
        .parse()
        .map_err(|_| format!("Invalid integer part: {}", integer_part))?;

    // Scale up by 10^scale
    result = result
        .checked_mul(10_i128.pow(scale as u32))
        .ok_or_else(|| format!("Numeric overflow for value: {}", value_str))?;

    // Add the decimal part
    if !decimal_part.is_empty() {
        let decimal_digits = decimal_part.len().min(scale as usize);
        let decimal_value: i128 = decimal_part[..decimal_digits]
            .parse()
            .map_err(|_| format!("Invalid decimal part: {}", decimal_part))?;

        // Scale the decimal part appropriately
        let scale_diff = scale as usize - decimal_digits;
        let scaled_decimal = decimal_value * 10_i128.pow(scale_diff as u32);

        result = if integer_part.starts_with('-') {
            result
                .checked_sub(scaled_decimal)
                .ok_or_else(|| format!("Numeric overflow for value: {}", value_str))?
        } else {
            result
                .checked_add(scaled_decimal)
                .ok_or_else(|| format!("Numeric overflow for value: {}", value_str))?
        };
    }

    Ok(result)
}

/// Maps an Exasol type to an Arrow DataType.
///
/// # Arguments
/// * `exasol_type` - The Exasol type to convert
///
/// # Returns
/// * `Ok(DataType)` - The corresponding Arrow DataType
/// * `Err(String)` - Description if the type is unsupported
pub fn exasol_type_to_arrow(exasol_type: &ExasolType) -> Result<DataType, String> {
    match exasol_type {
        ExasolType::Boolean => Ok(DataType::Boolean),
        ExasolType::Char { .. } | ExasolType::Varchar { .. } => Ok(DataType::Utf8),
        ExasolType::Decimal { precision, scale } => Ok(DataType::Decimal128(*precision, *scale)),
        ExasolType::Double => Ok(DataType::Float64),
        ExasolType::Date => Ok(DataType::Date32),
        ExasolType::Timestamp {
            with_local_time_zone,
        } => {
            if *with_local_time_zone {
                Ok(DataType::Timestamp(
                    TimeUnit::Microsecond,
                    Some("UTC".into()),
                ))
            } else {
                Ok(DataType::Timestamp(TimeUnit::Microsecond, None))
            }
        }
        ExasolType::IntervalYearToMonth => Ok(DataType::Int64), // Months as i64
        ExasolType::IntervalDayToSecond { .. } => Ok(DataType::Int64), // Nanoseconds as i64
        ExasolType::Geometry { .. } | ExasolType::Hashtype { .. } => Ok(DataType::Binary),
    }
}

#[cfg(test)]
mod tests {
    /// Scenario: Bound Arrow Timestamp and Date32 values are stored on both transports
    #[test]
    fn formatter_output_parses_back_to_its_fields() {
        for (days, y, m, d) in [
            (-719_162, 1, 1, 1),
            (2_932_896, 9999, 12, 31),
            (-1, 1969, 12, 31),
        ] {
            let f = parse_date_text(&format_date32(days).unwrap()).unwrap();
            assert_eq!((f.year, f.month, f.day), (y, m, d));
        }
        for (micros, y, mo, d, h, mi, s, nanos) in [
            (-62_135_596_800_000_000, 1, 1, 1, 0, 0, 0, 0),
            (
                253_402_300_799_999_999,
                9999,
                12,
                31,
                23,
                59,
                59,
                999_999_000,
            ),
            (-999_999, 1969, 12, 31, 23, 59, 59, 1000),
        ] {
            let text = format_timestamp(&TimeUnit::Microsecond, micros).unwrap();
            let f = parse_timestamp_text(&text).unwrap();
            assert_eq!(
                (f.year, f.month, f.day, f.hour, f.minute, f.second, f.nanos),
                (y, mo, d, h, mi, s, nanos)
            );
        }
    }

    use super::*;

    // Tests for pre-1970 day counts

    /// Scenario: DATE and TIMESTAMP values count days in the proleptic Gregorian calendar
    #[test]
    fn test_ymd_to_days_matches_chrono_for_every_exasol_date() {
        use chrono::{Datelike, NaiveDate};

        let epoch = NaiveDate::from_ymd_opt(1970, 1, 1).unwrap();
        let last = NaiveDate::from_ymd_opt(9999, 12, 31).unwrap();
        let mut date = NaiveDate::from_ymd_opt(1, 1, 1).unwrap();
        loop {
            let expected = (date - epoch).num_days() as i32;
            assert_eq!(
                ymd_to_days(date.year(), date.month(), date.day()),
                expected,
                "day count of {date}"
            );
            if date == last {
                break;
            }
            date = date.succ_opt().unwrap();
        }
    }

    /// Scenario: DATE and TIMESTAMP values count days in the proleptic Gregorian calendar
    #[test]
    fn test_parse_timestamp_to_micros_matches_chrono_for_every_exasol_year() {
        use chrono::NaiveDate;

        for year in 1..=9999 {
            let text = format!("{year:04}-03-01 12:34:56.789012");
            let expected = NaiveDate::from_ymd_opt(year, 3, 1)
                .unwrap()
                .and_hms_micro_opt(12, 34, 56, 789_012)
                .unwrap()
                .and_utc()
                .timestamp_micros();
            assert_eq!(
                parse_timestamp_to_micros(&text).unwrap(),
                expected,
                "microseconds of {text}"
            );
        }
    }

    /// Scenario: DATE and TIMESTAMP values count days in the proleptic Gregorian calendar
    #[test]
    fn test_parse_date_to_days_before_1970() {
        assert_eq!(parse_date_to_days("1968-01-01").unwrap(), -731);
        assert_eq!(parse_date_to_days("1900-03-01").unwrap(), -25508);
        assert_eq!(parse_date_to_days("1600-03-01").unwrap(), -135080);
        assert_eq!(parse_date_to_days("0001-01-01").unwrap(), -719162);
        assert_eq!(parse_date_to_days("9999-12-31").unwrap(), 2932896);
    }

    /// Scenario: DATE and TIMESTAMP values count days in the proleptic Gregorian calendar
    #[test]
    fn test_parse_date_to_days_century_leap_years() {
        assert_eq!(parse_date_to_days("1600-02-29").unwrap(), -135081);
        assert_eq!(parse_date_to_days("1900-02-28").unwrap(), -25509);
        assert_eq!(parse_date_to_days("1900-03-01").unwrap(), -25508);
        assert_eq!(parse_date_to_days("2000-02-29").unwrap(), 11016);
        assert_eq!(parse_date_to_days("2000-03-01").unwrap(), 11017);
    }

    /// Scenario: DATE and TIMESTAMP values count days in the proleptic Gregorian calendar
    #[test]
    fn test_parse_timestamp_to_micros_before_1970() {
        assert_eq!(
            parse_timestamp_to_micros("1950-06-15 00:00:00").unwrap(),
            -616_896_000_000_000
        );
        assert_eq!(
            parse_timestamp_to_micros("1969-12-31 23:59:59.999999").unwrap(),
            -1
        );
        assert_eq!(
            parse_timestamp_to_micros("1969-12-31 12:00:00").unwrap(),
            -43_200_000_000
        );
        assert_eq!(
            parse_timestamp_to_micros("0001-01-01 00:00:00").unwrap(),
            -62_135_596_800_000_000
        );
        assert_eq!(
            parse_timestamp_to_micros("9999-12-31 23:59:59.999999").unwrap(),
            253_402_300_799_999_999
        );
    }

    /// Scenario: DATE and TIMESTAMP values count days in the proleptic Gregorian calendar
    #[test]
    fn test_ymd_hms_nanos_to_micros_before_epoch() {
        assert_eq!(
            ymd_hms_nanos_to_micros(1969, 12, 31, 23, 59, 59, 999_999_000),
            -1
        );
        assert_eq!(
            ymd_hms_nanos_to_micros(1950, 6, 15, 0, 0, 0, 0),
            -616_896_000_000_000
        );
    }

    /// Scenario: DATE and TIMESTAMP values count days in the proleptic Gregorian calendar
    /// Scenario: Lossless conversion validation
    #[test]
    fn test_ymd_to_days_julian_only_leap_day_reads_as_march_first() {
        assert_eq!(ymd_to_days(1500, 2, 29), -171_605);
        assert_eq!(ymd_to_days(1500, 3, 1), -171_605);
        assert_eq!(parse_date_to_days("1500-02-29"), Ok(-171_605));
    }

    // Tests for parse_date_to_days

    #[test]
    fn test_parse_date_to_days_unix_epoch() {
        assert_eq!(parse_date_to_days("1970-01-01").unwrap(), 0);
    }

    #[test]
    fn test_parse_date_to_days_positive_date() {
        // 2024-01-01 should be a positive number of days
        let days = parse_date_to_days("2024-01-01").unwrap();
        assert!(days > 0);
    }

    #[test]
    fn test_parse_date_to_days_negative_date() {
        // 1969-01-01 should be negative (before Unix epoch)
        let days = parse_date_to_days("1969-01-01").unwrap();
        assert!(days < 0);
    }

    #[test]
    fn test_parse_date_to_days_leap_year() {
        // Feb 29, 2024 (leap year) should be valid
        let days = parse_date_to_days("2024-02-29").unwrap();
        assert!(days > 0);
    }

    #[test]
    fn test_parse_date_to_days_invalid_format() {
        assert!(parse_date_to_days("2024/01/01").is_err());
        // "01-01-2024" parses as year=1, month=1, day=2024, which fails day validation
        assert!(parse_date_to_days("01-01-2024").is_err());
        assert!(parse_date_to_days("2024-1-1").is_ok()); // Single digits should work
    }

    #[test]
    fn test_parse_date_to_days_invalid_month() {
        assert!(parse_date_to_days("2024-13-01").is_err());
        assert!(parse_date_to_days("2024-00-01").is_err());
    }

    #[test]
    fn test_parse_date_to_days_invalid_day() {
        assert!(parse_date_to_days("2024-01-32").is_err());
        assert!(parse_date_to_days("2024-01-00").is_err());
    }

    // Tests for parse_timestamp_to_micros

    #[test]
    fn test_parse_timestamp_to_micros_date_only() {
        let micros = parse_timestamp_to_micros("1970-01-01").unwrap();
        assert_eq!(micros, 0);
    }

    #[test]
    fn test_parse_timestamp_to_micros_with_time() {
        let micros = parse_timestamp_to_micros("1970-01-01 01:00:00").unwrap();
        assert_eq!(micros, 3600 * 1_000_000);
    }

    #[test]
    fn test_parse_timestamp_to_micros_with_fractional_seconds() {
        let micros = parse_timestamp_to_micros("1970-01-01 00:00:00.123456").unwrap();
        assert_eq!(micros, 123456);
    }

    #[test]
    fn test_parse_timestamp_to_micros_with_short_fraction() {
        // .123 should be interpreted as 123000 microseconds
        let micros = parse_timestamp_to_micros("1970-01-01 00:00:00.123").unwrap();
        assert_eq!(micros, 123000);
    }

    #[test]
    fn test_parse_timestamp_to_micros_hours_and_minutes_only() {
        let result = parse_timestamp_to_micros("1970-01-01 01:30").unwrap();
        assert_eq!(result, 5400 * 1_000_000);
    }

    #[test]
    fn test_parse_timestamp_to_micros_empty_string_reports_date_format() {
        let error = parse_timestamp_to_micros("").unwrap_err();
        assert_eq!(error, "Invalid date format:  (expected YYYY-MM-DD)");
    }

    #[test]
    fn test_parse_timestamp_to_micros_time_without_colon_is_ignored() {
        // A time segment with fewer than two colon-separated fields contributes nothing
        let micros = parse_timestamp_to_micros("1970-01-01 12").unwrap();
        assert_eq!(micros, 0);
    }

    #[test]
    fn test_parse_timestamp_to_micros_ignores_segments_after_the_time() {
        // Only the first two space-separated segments are interpreted
        let micros = parse_timestamp_to_micros("1970-01-01 01:00:00 CET").unwrap();
        assert_eq!(micros, 3600 * 1_000_000);
    }

    #[test]
    fn test_parse_timestamp_to_micros_ignores_fields_after_the_seconds() {
        let micros = parse_timestamp_to_micros("1970-01-01 00:00:01:99").unwrap();
        assert_eq!(micros, 1_000_000);
    }

    #[test]
    fn test_parse_timestamp_to_micros_before_unix_epoch() {
        let micros = parse_timestamp_to_micros("1969-12-31 23:00:00").unwrap();
        assert_eq!(micros, -86_400 * 1_000_000 + 23 * 3_600 * 1_000_000);
    }

    #[test]
    fn test_parse_timestamp_to_micros_invalid_hour() {
        let error = parse_timestamp_to_micros("1970-01-01 XX:00:00").unwrap_err();
        assert_eq!(error, "Invalid hour: XX");
    }

    #[test]
    fn test_parse_timestamp_to_micros_invalid_minute() {
        let error = parse_timestamp_to_micros("1970-01-01 00:XX:00").unwrap_err();
        assert_eq!(error, "Invalid minute: XX");
    }

    #[test]
    fn test_parse_timestamp_to_micros_invalid_second() {
        let error = parse_timestamp_to_micros("1970-01-01 00:00:XX").unwrap_err();
        assert_eq!(error, "Invalid second: XX");
    }

    #[test]
    fn test_parse_timestamp_to_micros_non_numeric_fraction_counts_as_zero() {
        let micros = parse_timestamp_to_micros("1970-01-01 00:00:01.abc").unwrap();
        assert_eq!(micros, 1_000_000);
    }

    #[test]
    fn test_parse_date_to_days_each_month_of_a_non_leap_year() {
        let cumulative_days = [0, 31, 59, 90, 120, 151, 181, 212, 243, 273, 304, 334];
        for (index, expected) in cumulative_days.iter().enumerate() {
            let month = index + 1;
            let date = format!("1970-{:02}-01", month);
            assert_eq!(
                parse_date_to_days(&date).unwrap(),
                *expected,
                "mismatch for {}",
                date
            );
        }
    }

    #[test]
    fn test_parse_date_to_days_reports_invalid_year() {
        assert_eq!(
            parse_date_to_days("XXXX-01-01").unwrap_err(),
            "Invalid year: XXXX"
        );
    }

    #[test]
    fn test_parse_date_to_days_reports_unparsable_month() {
        assert_eq!(
            parse_date_to_days("2024-XX-01").unwrap_err(),
            "Invalid month: XX"
        );
    }

    #[test]
    fn test_parse_date_to_days_reports_unparsable_day() {
        assert_eq!(
            parse_date_to_days("2024-01-XX").unwrap_err(),
            "Invalid day: XX"
        );
    }

    #[test]
    fn test_parse_date_to_days_reports_month_out_of_range() {
        assert_eq!(
            parse_date_to_days("2024-13-01").unwrap_err(),
            "Month out of range: 13"
        );
    }

    #[test]
    fn test_parse_date_to_days_reports_day_out_of_range() {
        assert_eq!(
            parse_date_to_days("2024-01-32").unwrap_err(),
            "Day out of range: 32"
        );
    }

    // Tests for ymd_to_days

    #[test]
    fn test_ymd_to_days_unix_epoch() {
        assert_eq!(ymd_to_days(1970, 1, 1), 0);
    }

    #[test]
    fn test_ymd_to_days_matches_string_parser() {
        for date in &[
            "1970-01-01",
            "1969-12-31",
            "2024-02-29",
            "2000-02-29",
            "1900-03-01",
            "2024-01-01",
            "2100-03-01",
            "2023-12-31",
            "1601-01-01",
        ] {
            let parts: Vec<&str> = date.split('-').collect();
            let y: i32 = parts[0].parse().unwrap();
            let m: u32 = parts[1].parse().unwrap();
            let d: u32 = parts[2].parse().unwrap();
            assert_eq!(
                ymd_to_days(y, m, d),
                parse_date_to_days(date).unwrap(),
                "mismatch for {}",
                date
            );
        }
    }

    #[test]
    fn test_ymd_to_days_month_out_of_range_contributes_no_month_offset() {
        // ymd_to_days does no validation: an out-of-range month adds zero days
        assert_eq!(ymd_to_days(1970, 13, 1), 0);
        assert_eq!(ymd_to_days(1970, 0, 1), 0);
    }

    #[test]
    fn test_ymd_to_days_leap_year_after_feb() {
        // 2024 is a leap year; March 1 2024 is one day later than a non-leap year would be
        let leap_mar1 = ymd_to_days(2024, 3, 1);
        let non_leap_mar1 = ymd_to_days(2023, 3, 1);
        assert_eq!(leap_mar1 - non_leap_mar1, 366);
    }

    // Tests for ymd_hms_nanos_to_micros

    #[test]
    fn test_ymd_hms_nanos_to_micros_epoch() {
        assert_eq!(ymd_hms_nanos_to_micros(1970, 1, 1, 0, 0, 0, 0), 0);
    }

    #[test]
    fn test_ymd_hms_nanos_to_micros_with_time() {
        let micros = ymd_hms_nanos_to_micros(1970, 1, 1, 1, 0, 0, 0);
        assert_eq!(micros, 3600 * 1_000_000);
    }

    #[test]
    fn test_ymd_hms_nanos_to_micros_sub_second_precision() {
        // 123456789 nanos → 123456 micros (truncated)
        let micros = ymd_hms_nanos_to_micros(1970, 1, 1, 0, 0, 0, 123_456_789);
        assert_eq!(micros, 123_456);
    }

    #[test]
    fn test_ymd_hms_nanos_to_micros_full_timestamp() {
        // 2024-01-02 03:04:05.678 → known value
        let micros = ymd_hms_nanos_to_micros(2024, 1, 2, 3, 4, 5, 678_000_000);
        let days = ymd_to_days(2024, 1, 2) as i64;
        let expected = days * 86_400 * 1_000_000
            + 3 * 3_600 * 1_000_000
            + 4 * 60 * 1_000_000
            + 5 * 1_000_000
            + 678_000;
        assert_eq!(micros, expected);
    }

    // Tests for fractional_seconds_to_micros / fractional_seconds_to_nanos

    #[test]
    fn test_fractional_seconds_to_micros_pads_short_fractions() {
        assert_eq!(fractional_seconds_to_micros("1"), 100_000);
        assert_eq!(fractional_seconds_to_micros("12"), 120_000);
        assert_eq!(fractional_seconds_to_micros("123456"), 123_456);
    }

    #[test]
    fn test_fractional_seconds_to_micros_truncates_long_fractions() {
        assert_eq!(fractional_seconds_to_micros("123456789"), 123_456);
    }

    #[test]
    fn test_fractional_seconds_to_micros_empty_is_zero() {
        assert_eq!(fractional_seconds_to_micros(""), 0);
    }

    #[test]
    fn test_fractional_seconds_to_micros_unparsable_is_zero() {
        assert_eq!(fractional_seconds_to_micros("abc"), 0);
        assert_eq!(fractional_seconds_to_micros("abcdefghi"), 0);
    }

    #[test]
    fn test_fractional_seconds_to_nanos_pads_short_fractions() {
        assert_eq!(fractional_seconds_to_nanos("1"), 100_000_000);
        assert_eq!(fractional_seconds_to_nanos("123456"), 123_456_000);
        assert_eq!(fractional_seconds_to_nanos("123456789"), 123_456_789);
    }

    #[test]
    fn test_fractional_seconds_to_nanos_truncates_long_fractions() {
        assert_eq!(fractional_seconds_to_nanos("123456789012"), 123_456_789);
    }

    #[test]
    fn test_fractional_seconds_to_nanos_empty_is_zero() {
        assert_eq!(fractional_seconds_to_nanos(""), 0);
    }

    #[test]
    fn test_fractional_seconds_to_nanos_unparsable_is_zero() {
        assert_eq!(fractional_seconds_to_nanos("abc"), 0);
    }

    // Tests for the Arrow-to-Exasol date and timestamp formatters

    #[test]
    fn test_format_date32_round_trips_every_month_boundary() {
        let epoch = chrono::NaiveDate::from_ymd_opt(1970, 1, 1).unwrap();
        for year in 1..=9999 {
            for month in 1..=12 {
                let first = (chrono::NaiveDate::from_ymd_opt(year, month, 1).unwrap() - epoch)
                    .num_days() as i32;
                for day in [first - 1, first] {
                    if day == -719_163 {
                        continue;
                    }
                    assert_eq!(
                        parse_date_to_days(&format_date32(day).unwrap()),
                        Ok(day),
                        "day {day}"
                    );
                }
            }
        }
    }

    #[test]
    fn test_format_date32_literals() {
        let cases = [
            (0, "1970-01-01"),
            (1, "1970-01-02"),
            (365, "1971-01-01"),
            (-1, "1969-12-31"),
            (-731, "1968-01-01"),
            (-25567, "1900-01-01"),
            (10956, "1999-12-31"),
            (11016, "2000-02-29"),
            (11017, "2000-03-01"),
            (19737, "2024-01-15"),
            (-719_162, "0001-01-01"),
            (2_932_896, "9999-12-31"),
        ];
        for (days, expected) in cases {
            assert_eq!(format_date32(days).as_deref(), Ok(expected), "day {days}");
        }
    }

    /// Scenario: RecordBatch import rejects DATE and TIMESTAMP values outside Exasol's range
    #[test]
    fn test_format_date64_floors_to_the_day_that_holds_it_and_rejects_dates_outside_exasol_range() {
        for (milliseconds, expected) in [
            (-1, "1969-12-31"),
            (-86_400_001, "1969-12-30"),
            (1_577_836_800_000, "2020-01-01"),
        ] {
            assert_eq!(
                format_date64(milliseconds).as_deref(),
                Ok(expected),
                "{milliseconds} ms"
            );
        }
        for milliseconds in [253_402_300_800_000, i64::MAX] {
            let message = format_date64(milliseconds).unwrap_err();
            assert!(message.contains(&milliseconds.to_string()), "{message}");
        }
    }

    /// Scenario: RecordBatch import rejects DATE and TIMESTAMP values outside Exasol's range
    #[test]
    fn test_format_date32_rejects_days_outside_exasol_range() {
        for days in [-719_163, 2_932_897, i32::MIN, i32::MAX] {
            assert!(format_date32(days).is_err(), "day {days}");
        }
    }

    /// Scenario: RecordBatch import formats pre-epoch timestamps of every time unit as times before the epoch
    #[test]
    fn test_format_timestamp_micros_literals() {
        let cases = [
            (0, "1970-01-01 00:00:00.000000"),
            (1_000_000, "1970-01-01 00:00:01.000000"),
            (86_400_000_000, "1970-01-02 00:00:00.000000"),
            (123_456, "1970-01-01 00:00:00.123456"),
            (-1, "1969-12-31 23:59:59.999999"),
            (-1_000_000, "1969-12-31 23:59:59.000000"),
            (-86_400_000_000, "1969-12-31 00:00:00.000000"),
            (-43_200_000_000, "1969-12-31 12:00:00.000000"),
            (-62_135_596_800_000_000, "0001-01-01 00:00:00.000000"),
            (253_402_300_799_999_999, "9999-12-31 23:59:59.999999"),
        ];
        for (micros, expected) in cases {
            assert_eq!(
                format_timestamp_micros(micros).as_deref(),
                Ok(expected),
                "micros {micros}"
            );
        }
    }

    /// Scenario: RecordBatch import rejects DATE and TIMESTAMP values outside Exasol's range
    #[test]
    fn test_format_timestamp_micros_rejects_micros_outside_exasol_range() {
        for micros in [
            253_402_300_800_000_000,
            -62_135_596_800_000_001,
            i64::MIN,
            i64::MAX,
        ] {
            assert!(format_timestamp_micros(micros).is_err(), "micros {micros}");
        }
    }

    /// Scenario: RecordBatch import formats pre-epoch timestamps of every time unit as times before the epoch
    #[test]
    fn test_format_timestamp_floors_every_unit() {
        let cases = [
            (TimeUnit::Second, -1, "1969-12-31 23:59:59.000000"),
            (TimeUnit::Millisecond, -1, "1969-12-31 23:59:59.999000"),
            (TimeUnit::Microsecond, -1, "1969-12-31 23:59:59.999999"),
            (TimeUnit::Nanosecond, -1, "1969-12-31 23:59:59.999999"),
            (TimeUnit::Nanosecond, -1000, "1969-12-31 23:59:59.999999"),
            (TimeUnit::Nanosecond, -1001, "1969-12-31 23:59:59.999998"),
            (TimeUnit::Nanosecond, 1999, "1970-01-01 00:00:00.000001"),
        ];
        for (unit, value, expected) in cases {
            assert_eq!(
                format_timestamp(&unit, value).as_deref(),
                Ok(expected),
                "{unit:?} {value}"
            );
        }
    }

    /// Scenario: RecordBatch import rejects DATE and TIMESTAMP values outside Exasol's range
    #[test]
    fn test_format_timestamp_rejects_overflow_and_out_of_range() {
        let cases = [
            (TimeUnit::Second, i64::MAX),
            (TimeUnit::Millisecond, i64::MIN),
            (TimeUnit::Microsecond, 253_402_300_800_000_000),
        ];
        for (unit, value) in cases {
            assert!(format_timestamp(&unit, value).is_err(), "{unit:?} {value}");
        }
    }

    #[test]
    fn test_format_timestamp_names_the_value_in_its_own_unit() {
        let error = format_timestamp(&TimeUnit::Second, 253_402_300_800).unwrap_err();
        assert!(error.contains("253402300800 in unit Second"), "{error}");
        assert!(!error.contains("microseconds"), "{error}");
    }

    // Tests for parse_decimal_to_i128

    #[test]
    fn test_parse_decimal_to_i128_integer() {
        let result = parse_decimal_to_i128("123", 2).unwrap();
        assert_eq!(result, 12300); // 123 * 10^2
    }

    #[test]
    fn test_parse_decimal_to_i128_with_decimal() {
        let result = parse_decimal_to_i128("123.45", 2).unwrap();
        assert_eq!(result, 12345);
    }

    #[test]
    fn test_parse_decimal_to_i128_negative() {
        let result = parse_decimal_to_i128("-123.45", 2).unwrap();
        assert_eq!(result, -12345);
    }

    #[test]
    fn test_parse_decimal_to_i128_scale_larger_than_digits() {
        // "12.3" with scale 4 should give 123000
        let result = parse_decimal_to_i128("12.3", 4).unwrap();
        assert_eq!(result, 123000);
    }

    #[test]
    fn test_parse_decimal_to_i128_invalid_format() {
        assert!(parse_decimal_to_i128("12.34.56", 2).is_err());
        assert!(parse_decimal_to_i128("abc", 2).is_err());
    }

    // Tests for exasol_type_to_arrow

    #[test]
    fn test_exasol_type_to_arrow_boolean() {
        assert_eq!(
            exasol_type_to_arrow(&ExasolType::Boolean).unwrap(),
            DataType::Boolean
        );
    }

    #[test]
    fn test_exasol_type_to_arrow_varchar() {
        assert_eq!(
            exasol_type_to_arrow(&ExasolType::Varchar { size: 100 }).unwrap(),
            DataType::Utf8
        );
    }

    #[test]
    fn test_exasol_type_to_arrow_decimal() {
        assert_eq!(
            exasol_type_to_arrow(&ExasolType::Decimal {
                precision: 18,
                scale: 2
            })
            .unwrap(),
            DataType::Decimal128(18, 2)
        );
    }

    #[test]
    fn test_exasol_type_to_arrow_timestamp() {
        assert_eq!(
            exasol_type_to_arrow(&ExasolType::Timestamp {
                with_local_time_zone: false
            })
            .unwrap(),
            DataType::Timestamp(TimeUnit::Microsecond, None)
        );
    }

    #[test]
    fn test_exasol_type_to_arrow_timestamp_with_tz() {
        assert_eq!(
            exasol_type_to_arrow(&ExasolType::Timestamp {
                with_local_time_zone: true
            })
            .unwrap(),
            DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into()))
        );
    }

    #[test]
    fn test_exasol_type_to_arrow_date() {
        assert_eq!(
            exasol_type_to_arrow(&ExasolType::Date).unwrap(),
            DataType::Date32
        );
    }

    #[test]
    fn test_exasol_type_to_arrow_double() {
        assert_eq!(
            exasol_type_to_arrow(&ExasolType::Double).unwrap(),
            DataType::Float64
        );
    }

    #[test]
    fn test_parse_interval_reads_every_day_to_second_component() {
        let (days, nanos) = days_and_nanos_of("+01 12:30:45.123456789");
        assert_eq!(days, 1);
        assert_eq!(
            nanos,
            12 * 3600 * 1_000_000_000 + 30 * 60 * 1_000_000_000 + 45 * 1_000_000_000 + 123_456_789
        );
    }

    #[test]
    fn test_parse_interval_ym_zero_years() {
        let months = months_of("+00-06");
        assert_eq!(months, 6);
    }

    #[test]
    fn test_parse_interval_ym_zero_months() {
        let months = months_of("+05-00");
        assert_eq!(months, 60);
    }

    #[test]
    fn test_parse_interval_ym_large_value() {
        let months = months_of("+99-11");
        assert_eq!(months, 99 * 12 + 11);
    }

    #[test]
    fn test_parse_interval_ym_negative_large_value() {
        let months = months_of("-99-11");
        assert_eq!(months, -(99 * 12 + 11));
    }

    #[test]
    fn test_parse_interval_reads_unsigned_text_as_positive() {
        assert_eq!(months_of("01-06"), 18);
    }

    #[test]
    fn test_parse_interval_ym_invalid_years() {
        let result = parse_interval("+XX-06");
        assert!(result.is_err());
    }

    #[test]
    fn test_parse_interval_ym_invalid_months() {
        let result = parse_interval("+01-XX");
        assert!(result.is_err());
    }

    #[test]
    fn test_parse_interval_ds_days_only() {
        let (days, nanos) = days_and_nanos_of("+05");
        assert_eq!(days, 5);
        assert_eq!(nanos, 0);
    }

    #[test]
    fn test_parse_interval_ds_hours_minutes_only() {
        let (days, nanos) = days_and_nanos_of("+01 12:30");
        assert_eq!(days, 1);
        assert_eq!(nanos, 12 * 3600 * 1_000_000_000 + 30 * 60 * 1_000_000_000);
    }

    #[test]
    fn test_parse_interval_ds_negative_values() {
        let (days, nanos) = days_and_nanos_of("-03 06:15:30");
        assert_eq!(days, -3);
        assert_eq!(
            nanos,
            -(6 * 3600 * 1_000_000_000 + 15 * 60 * 1_000_000_000 + 30 * 1_000_000_000)
        );
    }

    #[test]
    fn test_parse_interval_ds_with_fractional_3_digits() {
        let (days, nanos) = days_and_nanos_of("+00 00:00:00.123");
        assert_eq!(days, 0);
        assert_eq!(nanos, 123_000_000);
    }

    #[test]
    fn test_parse_interval_ds_with_fractional_6_digits() {
        let (days, nanos) = days_and_nanos_of("+00 00:00:00.123456");
        assert_eq!(days, 0);
        assert_eq!(nanos, 123_456_000);
    }

    #[test]
    fn test_parse_interval_ds_with_fractional_9_digits() {
        let (days, nanos) = days_and_nanos_of("+00 00:00:00.123456789");
        assert_eq!(days, 0);
        assert_eq!(nanos, 123_456_789);
    }

    #[test]
    fn test_parse_interval_ds_with_fractional_12_digits() {
        let (days, nanos) = days_and_nanos_of("+00 00:00:00.123456789012");
        assert_eq!(days, 0);
        assert_eq!(nanos, 123_456_789);
    }

    #[test]
    fn test_parse_interval_ds_invalid_days() {
        let result = parse_interval("+XX 12:30:45");
        assert!(result.is_err());
    }

    #[test]
    fn test_parse_interval_ds_time_without_colon_is_ignored() {
        let (days, nanos) = days_and_nanos_of("+01 12");
        assert_eq!(days, 1);
        assert_eq!(nanos, 0);
    }

    #[test]
    fn test_parse_interval_ds_ignores_segments_after_the_time() {
        let (days, nanos) = days_and_nanos_of("+01 01:00:00 extra");
        assert_eq!(days, 1);
        assert_eq!(nanos, 3600 * 1_000_000_000);
    }

    #[test]
    fn test_parse_interval_ds_ignores_fields_after_the_seconds() {
        let (days, nanos) = days_and_nanos_of("+00 00:00:01:99");
        assert_eq!(days, 0);
        assert_eq!(nanos, 1_000_000_000);
    }

    #[test]
    fn test_parse_interval_ds_empty_string_reports_invalid_days() {
        assert_eq!(parse_interval("").unwrap_err(), "Invalid days: ");
    }

    #[test]
    fn test_parse_interval_ds_reports_invalid_hours() {
        let message = parse_interval("+01 XX:30").unwrap_err();
        assert_eq!(message, "Invalid hours: XX");
    }

    #[test]
    fn test_parse_interval_ds_reports_invalid_minutes() {
        let message = parse_interval("+01 12:XX").unwrap_err();
        assert_eq!(message, "Invalid minutes: XX");
    }

    #[test]
    fn test_parse_interval_ds_reports_invalid_seconds() {
        let message = parse_interval("+01 12:30:XX").unwrap_err();
        assert_eq!(message, "Invalid seconds: XX");
    }

    #[test]
    fn test_parse_interval_ds_non_numeric_fraction_counts_as_zero() {
        let (days, nanos) = days_and_nanos_of("+00 00:00:01.abc");
        assert_eq!(days, 0);
        assert_eq!(nanos, 1_000_000_000);
    }

    #[test]
    fn test_parse_interval_ds_negative_nanos_only() {
        let (days, nanos) = days_and_nanos_of("-00 01:00:00");
        assert_eq!(days, 0);
        assert_eq!(nanos, -3600 * 1_000_000_000);
    }

    /// Scenario: INTERVAL type mapping
    #[test]
    fn test_parse_interval_reads_exasol_wire_text_of_both_forms() {
        assert_eq!(
            parse_interval("+01-02").unwrap(),
            IntervalMonthDayNano::new(14, 0, 0)
        );
        assert_eq!(
            parse_interval("-01-02").unwrap(),
            IntervalMonthDayNano::new(-14, 0, 0)
        );
        assert_eq!(
            parse_interval("-03 04:05:06.789").unwrap(),
            IntervalMonthDayNano::new(0, -3, -14_706_789_000_000)
        );
        assert_eq!(
            parse_interval("+03 04:05:06").unwrap(),
            IntervalMonthDayNano::new(0, 3, 14_706_000_000_000)
        );
    }

    #[test]
    fn test_parse_interval_reads_the_largest_day_to_second_value() {
        assert_eq!(
            parse_interval("-999999999 23:59:59.999000000").unwrap(),
            IntervalMonthDayNano::new(0, -999_999_999, -86_399_999_000_000)
        );
    }

    /// Scenario: INTERVAL YEAR TO MONTH value outside the Arrow month range
    #[test]
    fn test_parse_interval_rejects_month_count_outside_i32_range() {
        let message = parse_interval("+999999999-11").unwrap_err();
        assert!(message.contains("out of range"), "{message}");
        assert!(parse_interval("-999999999-11").is_err());
    }

    #[test]
    fn test_parse_interval_accepts_the_largest_year_to_month_value_of_i32() {
        assert_eq!(
            parse_interval("+178956970-07").unwrap(),
            IntervalMonthDayNano::new(i32::MAX, 0, 0)
        );
        assert!(parse_interval("+178956970-08").is_err());
    }

    #[test]
    fn test_parse_interval_rejects_extra_dash_parts() {
        assert!(parse_interval("+1-2-3")
            .unwrap_err()
            .contains("Invalid interval format"));
    }

    #[test]
    fn test_parse_interval_rejects_time_that_overflows_nanoseconds() {
        assert!(parse_interval("+00 9223372036854775807:00").is_err());
    }

    #[test]
    fn test_parse_interval_reads_the_smallest_month_count_of_i32() {
        assert_eq!(
            parse_interval("-178956970-08").unwrap(),
            IntervalMonthDayNano::new(i32::MIN, 0, 0)
        );
        assert!(parse_interval("-178956970-09").is_err());
    }

    fn duration_millis(interval: IntervalMonthDayNano) -> i64 {
        i64::from(interval.days) * MILLIS_PER_DAY + interval.nanoseconds / NANOS_PER_MILLISECOND
    }

    /// Scenario: RecordBatch import writes Arrow interval values as Exasol INTERVAL text
    #[test]
    fn test_interval_text_writer_writes_exasol_interval_text() {
        for (months, expected) in [
            (63, "5-03"),
            (-63, "-5-03"),
            (i32::MAX, "178956970-07"),
            (i32::MIN, "-178956970-08"),
            (0, "0-00"),
        ] {
            let text = format_interval_year_to_month(months);
            assert_eq!(text, expected, "{months} months");
            assert_eq!(
                parse_interval(&text).unwrap(),
                IntervalMonthDayNano::new(months, 0, 0),
                "{text}"
            );
        }

        let day_time = [
            ((2, 46_210_123), "2 12:50:10.123"),
            ((-2, -46_210_123), "-2 12:50:10.123"),
            ((0, 86_400_000), "1 00:00:00.000"),
            ((0, -1), "-0 00:00:00.001"),
            ((-999_999_999, -86_399_999), "-999999999 23:59:59.999"),
        ];
        for ((days, milliseconds), expected) in day_time {
            let text = format_interval_day_to_second(days, milliseconds).unwrap();
            assert_eq!(text, expected, "{days} days {milliseconds} ms");
            let parsed = parse_interval(&text).unwrap();
            assert_eq!(parsed.months, 0, "{text}");
            assert_eq!(
                duration_millis(parsed),
                i64::from(days) * MILLIS_PER_DAY + milliseconds,
                "{text}"
            );
        }

        let month_day_nano = [
            (
                IntervalMonthDayNano::new(0, 2, 46_210_123_000_000),
                "2 12:50:10.123",
            ),
            (
                IntervalMonthDayNano::new(0, -2, -46_210_123_000_000),
                "-2 12:50:10.123",
            ),
            (
                IntervalMonthDayNano::new(0, 1, -1_000_000),
                "0 23:59:59.999",
            ),
        ];
        for (value, expected) in month_day_nano {
            let text = format_interval_month_day_nano(value).unwrap();
            assert_eq!(text, expected, "{value:?}");
            let parsed = parse_interval(&text).unwrap();
            assert_eq!(parsed.months, 0, "{text}");
            assert_eq!(duration_millis(parsed), duration_millis(value), "{text}");
        }
    }

    /// Scenario: The import CSV conversion rejects interval values that INTERVAL DAY TO SECOND cannot hold
    #[test]
    fn test_interval_text_writer_rejects_values_that_day_to_second_cannot_hold() {
        let cases = [
            (
                format_interval_month_day_nano(IntervalMonthDayNano::new(1, 0, 0)),
                "1 months",
                "Exasol INTERVAL DAY TO SECOND has no month component",
            ),
            (
                format_interval_month_day_nano(IntervalMonthDayNano::new(0, 0, 1_500_000)),
                "1500000 nanoseconds",
                "Exasol stores INTERVAL DAY TO SECOND values in whole milliseconds",
            ),
            (
                format_interval_month_day_nano(IntervalMonthDayNano::new(0, 1_000_000_000, 0)),
                "1000000000 days",
                "INTERVAL DAY(9) TO SECOND holds at most 999,999,999 days",
            ),
            (
                format_interval_day_to_second(-1_000_000_000, 0),
                "-1000000000 days",
                "INTERVAL DAY(9) TO SECOND holds at most 999,999,999 days",
            ),
            (
                format_interval_day_to_second(999_999_999, 86_400_000),
                "86400000 milliseconds",
                "INTERVAL DAY(9) TO SECOND holds at most 999,999,999 days",
            ),
            (
                format_interval_day_to_second(0, i64::MAX),
                "9223372036854775807 milliseconds",
                "INTERVAL DAY(9) TO SECOND holds at most 999,999,999 days",
            ),
        ];

        for (result, value, reason) in cases {
            let message = result.unwrap_err();
            assert!(
                message.contains(value) && message.contains(reason),
                "{message}"
            );
        }
    }

    fn months_of(text: &str) -> i32 {
        let interval = parse_interval(text).unwrap();
        assert_eq!((interval.days, interval.nanoseconds), (0, 0));
        interval.months
    }

    fn days_and_nanos_of(text: &str) -> (i32, i64) {
        let interval = parse_interval(text).unwrap();
        assert_eq!(interval.months, 0);
        (interval.days, interval.nanoseconds)
    }
}