exarrow-rs 0.28.0

//! Array builders for converting JSON values to Arrow arrays.
//!
//! This module provides type-specific builders that convert columnar JSON data
//! into Arrow arrays with proper NULL handling.

use crate::error::{ColumnConversionError, ConversionError};
use crate::types::{
    conversion::{
        parse_date_to_days as parse_date_to_days_impl,
        parse_decimal_to_i128 as parse_decimal_to_i128_impl, parse_interval,
        parse_timestamp_to_micros as parse_timestamp_to_micros_impl,
    },
    ExasolType, TypeMapper,
};
use arrow::array::{
    ArrayRef, BinaryBuilder, BooleanBuilder, Date32Builder, Decimal128Builder, Float64Builder,
    IntervalMonthDayNanoBuilder, StringBuilder, TimestampMicrosecondBuilder,
};
use serde_json::Value;
use std::sync::Arc;

/// Trait for building Arrow arrays from JSON values.
pub trait ArrayBuilder {
    /// Append a JSON value to the array builder.
    ///
    /// # Arguments
    /// * `value` - The JSON value to append (or null)
    /// * `row` - The row index for error reporting
    /// * `column` - The column index for error reporting
    fn append_value(
        &mut self,
        value: &Value,
        row: usize,
        column: usize,
    ) -> Result<(), ConversionError>;

    /// Finish building and return the Arrow array.
    fn finish(&mut self) -> Result<ArrayRef, ConversionError>;
}

/// Build the Arrow array for one result column from its JSON values.
///
/// A value that fails to convert returns an error that names the column, the
/// row, the JSON value, and the Arrow data type that drove parsing.
pub fn build_array(
    exasol_type: &ExasolType,
    values: &[&Value],
    column: usize,
    column_name: &str,
) -> Result<ArrayRef, ConversionError> {
    // Same mapping SchemaBuilder uses, so the error names the type that drove parsing.
    let target_type = TypeMapper::exasol_to_arrow(exasol_type, true)?;
    build_typed_array(exasol_type, values)
        .map_err(|error| error.into_conversion_error(column, column_name, values, &target_type))
}

fn build_typed_array(
    exasol_type: &ExasolType,
    values: &[&Value],
) -> Result<ArrayRef, ColumnConversionError> {
    match exasol_type {
        ExasolType::Boolean => build_boolean_array(values),
        ExasolType::Char { .. } | ExasolType::Varchar { .. } => build_string_array(values),
        // Exasol DECIMAL precision is 1-36; always fits in Decimal128
        ExasolType::Decimal { precision, scale } => {
            build_decimal128_array(values, *precision, *scale)
        }
        ExasolType::Double => build_double_array(values),
        ExasolType::Date => build_date_array(values),
        ExasolType::Timestamp {
            with_local_time_zone,
        } => build_timestamp_array(values, *with_local_time_zone),
        ExasolType::IntervalYearToMonth | ExasolType::IntervalDayToSecond { .. } => {
            build_interval_array(values)
        }
        ExasolType::Geometry { .. } | ExasolType::Hashtype { .. } => build_binary_array(values),
    }
}

fn build_boolean_array(values: &[&Value]) -> Result<ArrayRef, ColumnConversionError> {
    let mut builder = BooleanBuilder::with_capacity(values.len());

    for (row, value) in values.iter().enumerate() {
        if value.is_null() {
            builder.append_null();
        } else if let Some(b) = value.as_bool() {
            builder.append_value(b);
        } else {
            return Err(ColumnConversionError::invalid_value(
                row,
                format!("Expected boolean, got: {:?}", value),
            ));
        }
    }

    Ok(Arc::new(builder.finish()))
}

fn estimate_string_capacity(values: &[&Value]) -> usize {
    const SAMPLE_SIZE: usize = 10;
    const DEFAULT_AVG_LEN: usize = 32;

    let mut total_len = 0;
    let mut count = 0;

    for value in values.iter().take(SAMPLE_SIZE) {
        if let Some(s) = value.as_str() {
            total_len += s.len();
            count += 1;
        }
    }

    total_len
        .checked_div(count)
        .map_or(DEFAULT_AVG_LEN, |avg| avg + 8)
        * values.len()
}

fn build_string_array(values: &[&Value]) -> Result<ArrayRef, ColumnConversionError> {
    // Use better capacity estimation based on actual string lengths
    let estimated_bytes = estimate_string_capacity(values);
    let mut builder = StringBuilder::with_capacity(values.len(), estimated_bytes);

    for (row, value) in values.iter().enumerate() {
        if value.is_null() {
            builder.append_null();
        } else if let Some(s) = value.as_str() {
            // UTF-8 validation is automatic in Rust strings
            builder.append_value(s);
        } else {
            return Err(ColumnConversionError::invalid_value(
                row,
                format!("Expected string, got: {:?}", value),
            ));
        }
    }

    Ok(Arc::new(builder.finish()))
}

/// Validate that a decimal string fits within the target precision.
///
/// This function checks that the total number of significant digits in the
/// decimal value does not exceed the specified precision.
///
/// # Arguments
/// * `value_str` - The decimal value as a string
/// * `precision` - The maximum number of significant digits allowed
/// * `_scale` - The scale (reserved for future use)
/// * `row` - The row index for error reporting
///
/// # Returns
/// `Ok(())` if the value fits within precision, or `ColumnConversionError` flagged as overflow otherwise.
fn validate_decimal_precision(
    value_str: &str,
    precision: u8,
    _scale: i8,
    row: usize,
) -> Result<(), ColumnConversionError> {
    // Remove sign and decimal point for digit counting
    let digits: String = value_str.chars().filter(|c| c.is_ascii_digit()).collect();

    let total_digits = digits.len();
    let max_digits = precision as usize;

    // Also strip leading zeros for accurate precision check
    let significant_digits = digits.trim_start_matches('0');
    let significant_count = if significant_digits.is_empty() {
        1 // "0" has 1 significant digit
    } else {
        significant_digits.len()
    };

    if significant_count > max_digits {
        return Err(ColumnConversionError::overflow(
            row,
            format!("{significant_count} significant digits exceed precision {max_digits}"),
        ));
    }

    // Also check total raw digit count as a sanity check
    // (precision includes both integer and fractional parts)
    if total_digits > max_digits + 1 {
        // +1 for potential leading zero
        return Err(ColumnConversionError::overflow(
            row,
            format!("{total_digits} digits exceed precision {max_digits}"),
        ));
    }

    Ok(())
}

fn build_decimal128_array(
    values: &[&Value],
    precision: u8,
    scale: i8,
) -> Result<ArrayRef, ColumnConversionError> {
    let mut builder = Decimal128Builder::with_capacity(values.len())
        .with_precision_and_scale(precision, scale)
        .map_err(|e| ConversionError::ArrowError(e.to_string()))?;

    for (row, value) in values.iter().enumerate() {
        if value.is_null() {
            builder.append_null();
        } else {
            let decimal_value = parse_decimal_to_i128(value, precision, scale, row)?;
            builder.append_value(decimal_value);
        }
    }

    Ok(Arc::new(builder.finish()))
}

/// Parse a JSON value to i128 for Decimal128.
fn parse_decimal_to_i128(
    value: &Value,
    precision: u8,
    scale: i8,
    row: usize,
) -> Result<i128, ColumnConversionError> {
    // Handle non-string JSON values directly
    if let Some(n) = value.as_i64() {
        return Ok((n as i128) * 10_i128.pow(scale as u32));
    } else if let Some(n) = value.as_f64() {
        // Convert float to decimal by multiplying by 10^scale
        let scaled = (n * 10_f64.powi(scale as i32)).round();
        return scaled
            .to_string()
            .parse::<i128>()
            .map_err(|_| ColumnConversionError::overflow(row, "Value does not fit in 128 bits"));
    }

    // Handle string values using shared implementation
    let value_str = value.as_str().ok_or_else(|| {
        ColumnConversionError::invalid_value(
            row,
            format!("Expected numeric value, got: {:?}", value),
        )
    })?;

    // Validate precision before conversion
    validate_decimal_precision(value_str, precision, scale, row)?;

    // Use shared decimal parsing implementation
    parse_decimal_to_i128_impl(value_str, scale)
        .map_err(|e| ColumnConversionError::invalid_value(row, e))
}

fn build_double_array(values: &[&Value]) -> Result<ArrayRef, ColumnConversionError> {
    let mut builder = Float64Builder::with_capacity(values.len());

    for (row, value) in values.iter().enumerate() {
        if value.is_null() {
            builder.append_null();
        } else if let Some(f) = value.as_f64() {
            builder.append_value(f);
        } else if let Some(i) = value.as_i64() {
            builder.append_value(i as f64);
        } else if let Some(s) = value.as_str() {
            // Handle special values like "Infinity", "-Infinity", "NaN"
            let f = match s {
                "Infinity" => f64::INFINITY,
                "-Infinity" => f64::NEG_INFINITY,
                "NaN" => f64::NAN,
                _ => s.parse::<f64>().map_err(|_| {
                    ColumnConversionError::invalid_value(row, format!("Invalid float value: {}", s))
                })?,
            };
            builder.append_value(f);
        } else {
            return Err(ColumnConversionError::invalid_value(
                row,
                format!("Expected number, got: {:?}", value),
            ));
        }
    }

    Ok(Arc::new(builder.finish()))
}

fn build_date_array(values: &[&Value]) -> Result<ArrayRef, ColumnConversionError> {
    let mut builder = Date32Builder::with_capacity(values.len());

    for (row, value) in values.iter().enumerate() {
        if value.is_null() {
            builder.append_null();
        } else if let Some(s) = value.as_str() {
            // Parse date string in format "YYYY-MM-DD"
            let days = parse_date_to_days(s, row)?;
            builder.append_value(days);
        } else {
            return Err(ColumnConversionError::invalid_value(
                row,
                format!("Expected date string, got: {:?}", value),
            ));
        }
    }

    Ok(Arc::new(builder.finish()))
}

/// Parse a date string to days since Unix epoch (1970-01-01).
fn parse_date_to_days(date_str: &str, row: usize) -> Result<i32, ColumnConversionError> {
    parse_date_to_days_impl(date_str).map_err(|e| ColumnConversionError::invalid_value(row, e))
}

fn build_timestamp_array(
    values: &[&Value],
    with_local_time_zone: bool,
) -> Result<ArrayRef, ColumnConversionError> {
    let mut builder = TimestampMicrosecondBuilder::with_capacity(values.len());

    for (row, value) in values.iter().enumerate() {
        if value.is_null() {
            builder.append_null();
        } else if let Some(s) = value.as_str() {
            // Parse timestamp string in format "YYYY-MM-DD HH:MM:SS.ffffff"
            let micros = parse_timestamp_to_micros(s, row)?;
            builder.append_value(micros);
        } else {
            return Err(ColumnConversionError::invalid_value(
                row,
                format!("Expected timestamp string, got: {:?}", value),
            ));
        }
    }

    let array = builder.finish();
    if with_local_time_zone {
        Ok(Arc::new(array.with_timezone("UTC")))
    } else {
        Ok(Arc::new(array))
    }
}

/// Parse a timestamp string to microseconds since Unix epoch.
fn parse_timestamp_to_micros(
    timestamp_str: &str,
    row: usize,
) -> Result<i64, ColumnConversionError> {
    parse_timestamp_to_micros_impl(timestamp_str)
        .map_err(|e| ColumnConversionError::invalid_value(row, e))
}

fn build_interval_array(values: &[&Value]) -> Result<ArrayRef, ColumnConversionError> {
    let mut builder = IntervalMonthDayNanoBuilder::with_capacity(values.len());

    for (row, value) in values.iter().enumerate() {
        if value.is_null() {
            builder.append_null();
        } else if let Some(s) = value.as_str() {
            let interval =
                parse_interval(s).map_err(|e| ColumnConversionError::invalid_value(row, e))?;
            builder.append_value(interval);
        } else {
            return Err(ColumnConversionError::invalid_value(
                row,
                format!("Expected interval string, got: {:?}", value),
            ));
        }
    }

    Ok(Arc::new(builder.finish()))
}

fn build_binary_array(values: &[&Value]) -> Result<ArrayRef, ColumnConversionError> {
    // Estimate binary capacity: hex strings are 2 chars per byte
    // Sample up to 10 values to estimate average size
    const SAMPLE_SIZE: usize = 10;
    let sample_count = values.len().min(SAMPLE_SIZE);
    let sample_total: usize = values
        .iter()
        .take(SAMPLE_SIZE)
        .filter_map(|v| v.as_str())
        .map(|s| s.len() / 2)
        .sum();

    let estimated_bytes = sample_total
        .checked_div(sample_count)
        .map_or(32, |avg| avg.max(1))
        * values.len();

    let mut builder = BinaryBuilder::with_capacity(values.len(), estimated_bytes.max(1024));

    for (row, value) in values.iter().enumerate() {
        if value.is_null() {
            builder.append_null();
        } else if let Some(s) = value.as_str() {
            // Assume hex-encoded string, decode to bytes
            let bytes = decode_hex(s).map_err(|_| {
                ColumnConversionError::invalid_value(row, format!("Invalid hex string: {}", s))
            })?;
            builder.append_value(&bytes);
        } else {
            return Err(ColumnConversionError::invalid_value(
                row,
                format!("Expected hex string, got: {:?}", value),
            ));
        }
    }

    Ok(Arc::new(builder.finish()))
}

/// Decode a hex string to bytes.
fn decode_hex(s: &str) -> Result<Vec<u8>, ()> {
    if !s.len().is_multiple_of(2) {
        return Err(());
    }

    (0..s.len())
        .step_by(2)
        .map(|i| u8::from_str_radix(&s[i..i + 2], 16).map_err(|_| ()))
        .collect()
}

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

    /// Helper to convert owned values to references for testing
    fn to_refs(values: &[Value]) -> Vec<&Value> {
        values.iter().collect()
    }

    /// Destructure a value failure into (row, reason, overflow).
    fn value_failure(error: ColumnConversionError) -> (usize, String, bool) {
        match error {
            ColumnConversionError::Value {
                row,
                reason,
                overflow,
            } => (row, reason, overflow),
            other => panic!("Expected a value failure, got {:?}", other),
        }
    }

    /// Destructure a value-conversion failure into (row, reason).
    fn value_conversion_failure(error: ColumnConversionError) -> (usize, String) {
        match value_failure(error) {
            (row, reason, false) => (row, reason),
            (_, reason, true) => panic!("Expected a non-overflow failure, got overflow: {reason}"),
        }
    }

    /// The reason of a value-conversion failure, ignoring its position.
    fn conversion_failure_message(error: ColumnConversionError) -> String {
        value_conversion_failure(error).1
    }

    #[test]
    fn test_build_boolean_array() {
        let values = vec![json!(true), json!(false), json!(null), json!(true)];
        let refs = to_refs(&values);
        let array = build_boolean_array(&refs).unwrap();
        assert_eq!(array.len(), 4);
        assert_eq!(array.null_count(), 1);
    }

    #[test]
    fn test_build_string_array() {
        let values = vec![json!("hello"), json!("world"), json!(null), json!("test")];
        let refs = to_refs(&values);
        let array = build_string_array(&refs).unwrap();
        assert_eq!(array.len(), 4);
        assert_eq!(array.null_count(), 1);
    }

    #[test]
    fn test_build_decimal128_array() {
        let values = vec![json!("123.45"), json!("678.90"), json!(null)];
        let refs = to_refs(&values);
        let array = build_decimal128_array(&refs, 10, 2).unwrap();
        assert_eq!(array.len(), 3);
        assert_eq!(array.null_count(), 1);
    }

    #[test]
    fn test_build_double_array() {
        let values = vec![json!(1.5), json!(2.7), json!(null), json!("Infinity")];
        let refs = to_refs(&values);
        let array = build_double_array(&refs).unwrap();
        assert_eq!(array.len(), 4);
        assert_eq!(array.null_count(), 1);
    }

    #[test]
    fn test_parse_date_to_days() {
        // 1970-01-01 should be day 0
        let days = parse_date_to_days("1970-01-01", 0).unwrap();
        assert_eq!(days, 0);

        // 1970-01-02 should be day 1
        let days = parse_date_to_days("1970-01-02", 0).unwrap();
        assert_eq!(days, 1);
    }

    #[test]
    fn test_parse_timestamp_to_micros() {
        // 1970-01-01 00:00:00 should be 0 microseconds
        let micros = parse_timestamp_to_micros("1970-01-01 00:00:00", 0).unwrap();
        assert_eq!(micros, 0);

        // 1970-01-01 00:00:01 should be 1,000,000 microseconds
        let micros = parse_timestamp_to_micros("1970-01-01 00:00:01", 0).unwrap();
        assert_eq!(micros, 1_000_000);
    }

    #[test]
    fn test_decode_hex() {
        let bytes = decode_hex("48656c6c6f").unwrap();
        assert_eq!(bytes, b"Hello");

        let bytes = decode_hex("").unwrap();
        assert_eq!(bytes, b"");

        assert!(decode_hex("xyz").is_err());
        assert!(decode_hex("1").is_err()); // Odd length
    }

    #[test]
    fn test_parse_decimal_to_i128() {
        // Test integer
        let result = parse_decimal_to_i128(&json!("123"), 10, 2, 0).unwrap();
        assert_eq!(result, 12300); // 123 * 10^2

        // Test decimal
        let result = parse_decimal_to_i128(&json!("123.45"), 10, 2, 0).unwrap();
        assert_eq!(result, 12345); // 123.45 * 10^2

        // Test negative
        let result = parse_decimal_to_i128(&json!("-123.45"), 10, 2, 0).unwrap();
        assert_eq!(result, -12345);

        // Test from number
        let result = parse_decimal_to_i128(&json!(123), 10, 2, 0).unwrap();
        assert_eq!(result, 12300);
    }

    #[test]
    fn test_invalid_conversions() {
        // Invalid boolean
        let values = vec![json!("not a bool")];
        let refs = to_refs(&values);
        assert!(build_boolean_array(&refs).is_err());

        // Invalid date
        let values = vec![json!("2024-13-01")]; // Invalid month
        let refs = to_refs(&values);
        assert!(build_date_array(&refs).is_err());

        // Invalid hex
        let values = vec![json!("zzz")];
        let refs = to_refs(&values);
        assert!(build_binary_array(&refs).is_err());
    }

    #[test]
    fn test_validate_decimal_precision() {
        // Valid: 5 digits fits in precision 10
        assert!(validate_decimal_precision("12345", 10, 2, 0).is_ok());

        // Valid: decimal point doesn't count
        assert!(validate_decimal_precision("123.45", 10, 2, 0).is_ok());

        // Valid: sign doesn't count
        assert!(validate_decimal_precision("-12345", 10, 2, 0).is_ok());

        // Invalid: 6 significant digits in precision 5
        assert!(validate_decimal_precision("123456", 5, 2, 0).is_err());

        // Valid: leading zeros don't count as significant
        assert!(validate_decimal_precision("00123", 5, 2, 0).is_ok());

        // Valid: zero
        assert!(validate_decimal_precision("0", 1, 0, 0).is_ok());

        // Valid: zero with leading zeros
        assert!(validate_decimal_precision("000", 5, 0, 0).is_ok());
    }

    #[test]
    fn test_decimal_precision_overflow() {
        // Test that precision overflow is detected
        let values = vec![json!("12345678901234567890")]; // 20 digits
        let refs = to_refs(&values);
        let result = build_decimal128_array(&refs, 10, 2);
        assert!(result.is_err());

        // Should succeed with sufficient precision
        let values = vec![json!("12345")]; // 5 digits
        let refs = to_refs(&values);
        let result = build_decimal128_array(&refs, 10, 2);
        assert!(result.is_ok());
    }

    #[test]
    fn test_estimate_string_capacity() {
        // Test with normal strings
        let values = vec![json!("hello"), json!("world"), json!("test")];
        let refs = to_refs(&values);
        let capacity = estimate_string_capacity(&refs);
        // Average length is about 5, plus buffer, times 3 values
        assert!(capacity > 0);
        assert!(capacity >= 15); // At least the actual total length

        // Test with empty values - should return 0
        let empty_values: Vec<Value> = vec![];
        let empty_refs = to_refs(&empty_values);
        let empty_capacity = estimate_string_capacity(&empty_refs);
        assert_eq!(empty_capacity, 0);

        // Test with all nulls - should use default average
        let null_values = vec![json!(null), json!(null), json!(null)];
        let null_refs = to_refs(&null_values);
        let null_capacity = estimate_string_capacity(&null_refs);
        assert!(null_capacity > 0);

        // Test with mixed nulls and strings
        let mixed_values = vec![json!("hello"), json!(null), json!("world")];
        let mixed_refs = to_refs(&mixed_values);
        let mixed_capacity = estimate_string_capacity(&mixed_refs);
        assert!(mixed_capacity > 0);
    }

    // ==========================================================================
    // Tests for build_array dispatch
    // ==========================================================================

    #[test]
    fn test_build_array_dispatches_to_boolean() {
        let values = vec![json!(true), json!(false)];
        let refs = to_refs(&values);
        let result = build_array(&ExasolType::Boolean, &refs, 0, "COL").unwrap();
        assert_eq!(result.len(), 2);
    }

    #[test]
    fn test_build_array_dispatches_to_char() {
        let values = vec![json!("abc"), json!("def")];
        let refs = to_refs(&values);
        let result = build_array(&ExasolType::Char { size: 3 }, &refs, 0, "COL").unwrap();
        assert_eq!(result.len(), 2);
    }

    #[test]
    fn test_build_array_dispatches_to_varchar() {
        let values = vec![json!("hello"), json!("world")];
        let refs = to_refs(&values);
        let result = build_array(&ExasolType::Varchar { size: 100 }, &refs, 0, "COL").unwrap();
        assert_eq!(result.len(), 2);
    }

    #[test]
    fn test_build_array_dispatches_to_decimal() {
        let values = vec![json!("123.45"), json!("678.90")];
        let refs = to_refs(&values);
        let result = build_array(
            &ExasolType::Decimal {
                precision: 10,
                scale: 2,
            },
            &refs,
            0,
            "COL",
        )
        .unwrap();
        assert_eq!(result.len(), 2);
    }

    #[test]
    fn test_build_array_dispatches_to_double() {
        let values = vec![json!(1.5), json!(2.5)];
        let refs = to_refs(&values);
        let result = build_array(&ExasolType::Double, &refs, 0, "COL").unwrap();
        assert_eq!(result.len(), 2);
    }

    #[test]
    fn test_build_array_dispatches_to_date() {
        let values = vec![json!("2024-01-15"), json!("2024-06-20")];
        let refs = to_refs(&values);
        let result = build_array(&ExasolType::Date, &refs, 0, "COL").unwrap();
        assert_eq!(result.len(), 2);
    }

    #[test]
    fn test_build_array_dispatches_to_timestamp_without_tz() {
        let values = vec![json!("2024-01-15 10:30:00"), json!("2024-06-20 14:45:30")];
        let refs = to_refs(&values);
        let result = build_array(
            &ExasolType::Timestamp {
                with_local_time_zone: false,
            },
            &refs,
            0,
            "COL",
        )
        .unwrap();
        assert_eq!(result.len(), 2);
    }

    #[test]
    fn test_build_array_dispatches_to_timestamp_with_tz() {
        let values = vec![json!("2024-01-15 10:30:00"), json!("2024-06-20 14:45:30")];
        let refs = to_refs(&values);
        let result = build_array(
            &ExasolType::Timestamp {
                with_local_time_zone: true,
            },
            &refs,
            0,
            "COL",
        )
        .unwrap();
        assert_eq!(result.len(), 2);
    }

    #[test]
    fn test_build_array_dispatches_to_interval_year_to_month() {
        let values = vec![json!("+01-06"), json!("-02-03")];
        let refs = to_refs(&values);
        let result = build_array(&ExasolType::IntervalYearToMonth, &refs, 0, "COL").unwrap();
        assert_eq!(result.len(), 2);
    }

    #[test]
    fn test_build_array_dispatches_to_interval_day_to_second() {
        let values = vec![json!("+01 12:30:45"), json!("-02 08:15:30")];
        let refs = to_refs(&values);
        let result = build_array(
            &ExasolType::IntervalDayToSecond { precision: 3 },
            &refs,
            0,
            "COL",
        )
        .unwrap();
        assert_eq!(result.len(), 2);
    }

    #[test]
    fn test_build_array_dispatches_to_geometry() {
        let values = vec![json!("48656c6c6f"), json!("576f726c64")];
        let refs = to_refs(&values);
        let result =
            build_array(&ExasolType::Geometry { srid: Some(4326) }, &refs, 0, "COL").unwrap();
        assert_eq!(result.len(), 2);
    }

    #[test]
    fn test_build_array_dispatches_to_hashtype() {
        let values = vec![json!("deadbeef"), json!("cafebabe")];
        let refs = to_refs(&values);
        let result = build_array(&ExasolType::Hashtype { byte_size: 16 }, &refs, 0, "COL").unwrap();
        assert_eq!(result.len(), 2);
    }

    // ==========================================================================
    // Tests for build_date_array
    // ==========================================================================

    #[test]
    fn test_build_date_array_with_valid_dates() {
        let values = vec![
            json!("2024-01-15"),
            json!("2023-06-20"),
            json!("1970-01-01"),
        ];
        let refs = to_refs(&values);
        let result = build_date_array(&refs).unwrap();
        assert_eq!(result.len(), 3);
        assert_eq!(result.null_count(), 0);
    }

    #[test]
    fn test_build_date_array_with_null_values() {
        let values = vec![json!("2024-01-15"), json!(null), json!("2023-06-20")];
        let refs = to_refs(&values);
        let result = build_date_array(&refs).unwrap();
        assert_eq!(result.len(), 3);
        assert_eq!(result.null_count(), 1);
    }

    #[test]
    fn test_build_date_array_invalid_format_not_string() {
        let values = vec![json!(12345)];
        let refs = to_refs(&values);
        let message = conversion_failure_message(build_date_array(&refs).unwrap_err());
        assert!(message.contains("Expected date string"), "{}", message);
    }

    #[test]
    fn test_build_date_array_invalid_date_format() {
        let values = vec![json!("2024/01/15")]; // Wrong separator
        let refs = to_refs(&values);
        let result = build_date_array(&refs);
        assert!(result.is_err());
    }

    #[test]
    fn test_build_date_array_invalid_month_out_of_range() {
        let values = vec![json!("2024-13-01")]; // Month 13
        let refs = to_refs(&values);
        let result = build_date_array(&refs);
        assert!(result.is_err());
    }

    #[test]
    fn test_build_date_array_invalid_day_out_of_range() {
        let values = vec![json!("2024-01-32")]; // Day 32
        let refs = to_refs(&values);
        let result = build_date_array(&refs);
        assert!(result.is_err());
    }

    #[test]
    fn test_build_date_array_invalid_month_zero() {
        let values = vec![json!("2024-00-15")]; // Month 0
        let refs = to_refs(&values);
        let result = build_date_array(&refs);
        assert!(result.is_err());
    }

    #[test]
    fn test_build_date_array_invalid_day_zero() {
        let values = vec![json!("2024-01-00")]; // Day 0
        let refs = to_refs(&values);
        let result = build_date_array(&refs);
        assert!(result.is_err());
    }

    // ==========================================================================
    // Tests for parse_date_to_days - all months
    // ==========================================================================

    #[test]
    fn test_parse_date_to_days_january() {
        let days = parse_date_to_days("1970-01-15", 0).unwrap();
        assert_eq!(days, 14); // 15th day, 0-indexed
    }

    #[test]
    fn test_parse_date_to_days_february() {
        let days = parse_date_to_days("1970-02-01", 0).unwrap();
        assert_eq!(days, 31); // Jan has 31 days
    }

    #[test]
    fn test_parse_date_to_days_march() {
        let days = parse_date_to_days("1970-03-01", 0).unwrap();
        assert_eq!(days, 59); // Jan(31) + Feb(28)
    }

    #[test]
    fn test_parse_date_to_days_april() {
        let days = parse_date_to_days("1970-04-01", 0).unwrap();
        assert_eq!(days, 90); // Jan(31) + Feb(28) + Mar(31)
    }

    #[test]
    fn test_parse_date_to_days_may() {
        let days = parse_date_to_days("1970-05-01", 0).unwrap();
        assert_eq!(days, 120); // Jan(31) + Feb(28) + Mar(31) + Apr(30)
    }

    #[test]
    fn test_parse_date_to_days_june() {
        let days = parse_date_to_days("1970-06-01", 0).unwrap();
        assert_eq!(days, 151);
    }

    #[test]
    fn test_parse_date_to_days_july() {
        let days = parse_date_to_days("1970-07-01", 0).unwrap();
        assert_eq!(days, 181);
    }

    #[test]
    fn test_parse_date_to_days_august() {
        let days = parse_date_to_days("1970-08-01", 0).unwrap();
        assert_eq!(days, 212);
    }

    #[test]
    fn test_parse_date_to_days_september() {
        let days = parse_date_to_days("1970-09-01", 0).unwrap();
        assert_eq!(days, 243);
    }

    #[test]
    fn test_parse_date_to_days_october() {
        let days = parse_date_to_days("1970-10-01", 0).unwrap();
        assert_eq!(days, 273);
    }

    #[test]
    fn test_parse_date_to_days_november() {
        let days = parse_date_to_days("1970-11-01", 0).unwrap();
        assert_eq!(days, 304);
    }

    #[test]
    fn test_parse_date_to_days_december() {
        let days = parse_date_to_days("1970-12-01", 0).unwrap();
        assert_eq!(days, 334);
    }

    // ==========================================================================
    // Tests for leap year handling
    // ==========================================================================

    #[test]
    fn test_parse_date_to_days_leap_year_2000() {
        // 2000 is a leap year (divisible by 400)
        // March 1, 2000 should include the leap day
        let march_1_2000 = parse_date_to_days("2000-03-01", 0).unwrap();
        let feb_28_2000 = parse_date_to_days("2000-02-28", 0).unwrap();
        assert_eq!(march_1_2000 - feb_28_2000, 2); // Feb 29 exists
    }

    #[test]
    fn test_parse_date_to_days_leap_year_2004() {
        // 2004 is a leap year (divisible by 4, not by 100)
        let march_1_2004 = parse_date_to_days("2004-03-01", 0).unwrap();
        let feb_28_2004 = parse_date_to_days("2004-02-28", 0).unwrap();
        assert_eq!(march_1_2004 - feb_28_2004, 2); // Feb 29 exists
    }

    #[test]
    fn test_parse_date_to_days_non_leap_year_1900() {
        // 1900 is NOT a leap year (divisible by 100, not by 400)
        let march_1_1900 = parse_date_to_days("1900-03-01", 0).unwrap();
        let feb_28_1900 = parse_date_to_days("1900-02-28", 0).unwrap();
        assert_eq!(march_1_1900 - feb_28_1900, 1); // No Feb 29
    }

    #[test]
    fn test_parse_date_to_days_non_leap_year_2023() {
        // 2023 is NOT a leap year
        let march_1_2023 = parse_date_to_days("2023-03-01", 0).unwrap();
        let feb_28_2023 = parse_date_to_days("2023-02-28", 0).unwrap();
        assert_eq!(march_1_2023 - feb_28_2023, 1); // No Feb 29
    }

    #[test]
    fn test_parse_date_to_days_leap_adjustment_only_after_february() {
        // In a leap year, January should not have the leap adjustment
        let jan_31_2000 = parse_date_to_days("2000-01-31", 0).unwrap();
        let jan_30_2000 = parse_date_to_days("2000-01-30", 0).unwrap();
        assert_eq!(jan_31_2000 - jan_30_2000, 1); // Normal day difference
    }

    #[test]
    fn test_parse_date_to_days_before_unix_epoch() {
        // 1969-12-31 should be -1
        let days = parse_date_to_days("1969-12-31", 0).unwrap();
        assert_eq!(days, -1);
    }

    #[test]
    fn test_parse_date_to_days_far_future() {
        // Test a date far in the future
        let days = parse_date_to_days("2100-01-01", 0).unwrap();
        assert!(days > 0);
    }

    // ==========================================================================
    // Tests for parse_date_to_days error cases
    // ==========================================================================

    #[test]
    fn test_parse_date_to_days_invalid_format_missing_parts() {
        let result = parse_date_to_days("2024-01", 0);
        assert!(result.is_err());
    }

    #[test]
    fn test_parse_date_to_days_invalid_format_too_many_parts() {
        let result = parse_date_to_days("2024-01-15-extra", 0);
        assert!(result.is_err());
    }

    #[test]
    fn test_parse_date_to_days_invalid_year() {
        let result = parse_date_to_days("XXXX-01-15", 0);
        assert!(result.is_err());
    }

    #[test]
    fn test_parse_date_to_days_invalid_month() {
        let result = parse_date_to_days("2024-XX-15", 0);
        assert!(result.is_err());
    }

    #[test]
    fn test_parse_date_to_days_invalid_day() {
        let result = parse_date_to_days("2024-01-XX", 0);
        assert!(result.is_err());
    }

    // ==========================================================================
    // Tests for build_timestamp_array
    // ==========================================================================

    #[test]
    fn test_build_timestamp_array_without_timezone() {
        let values = vec![
            json!("2024-01-15 10:30:00"),
            json!("2024-06-20 14:45:30.123456"),
        ];
        let refs = to_refs(&values);
        let result = build_timestamp_array(&refs, false).unwrap();
        assert_eq!(result.len(), 2);
        assert_eq!(result.null_count(), 0);
    }

    #[test]
    fn test_build_timestamp_array_with_timezone() {
        let values = vec![
            json!("2024-01-15 10:30:00"),
            json!("2024-06-20 14:45:30.123456"),
        ];
        let refs = to_refs(&values);
        let result = build_timestamp_array(&refs, true).unwrap();
        assert_eq!(result.len(), 2);
        assert_eq!(result.null_count(), 0);
    }

    #[test]
    fn test_build_timestamp_array_with_null_values() {
        let values = vec![
            json!("2024-01-15 10:30:00"),
            json!(null),
            json!("2024-06-20 14:45:30"),
        ];
        let refs = to_refs(&values);
        let result = build_timestamp_array(&refs, false).unwrap();
        assert_eq!(result.len(), 3);
        assert_eq!(result.null_count(), 1);
    }

    #[test]
    fn test_build_timestamp_array_date_only() {
        // Test timestamp with only date part (no time)
        let values = vec![json!("2024-01-15")];
        let refs = to_refs(&values);
        let result = build_timestamp_array(&refs, false).unwrap();
        assert_eq!(result.len(), 1);
    }

    #[test]
    fn test_build_timestamp_array_invalid_not_string() {
        let values = vec![json!(12345)];
        let refs = to_refs(&values);
        let message = conversion_failure_message(build_timestamp_array(&refs, false).unwrap_err());
        assert!(message.contains("Expected timestamp string"), "{}", message);
    }

    // ==========================================================================
    // Tests for parse_timestamp_to_micros
    // ==========================================================================

    #[test]
    fn test_parse_timestamp_to_micros_with_fractional_seconds_3_digits() {
        let micros = parse_timestamp_to_micros("1970-01-01 00:00:00.123", 0).unwrap();
        assert_eq!(micros, 123_000);
    }

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

    #[test]
    fn test_parse_timestamp_to_micros_with_fractional_seconds_9_digits() {
        // Only first 6 digits are used for microseconds
        let micros = parse_timestamp_to_micros("1970-01-01 00:00:00.123456789", 0).unwrap();
        assert_eq!(micros, 123_456);
    }

    #[test]
    fn test_parse_timestamp_to_micros_with_fractional_seconds_1_digit() {
        let micros = parse_timestamp_to_micros("1970-01-01 00:00:00.1", 0).unwrap();
        assert_eq!(micros, 100_000);
    }

    #[test]
    fn test_parse_timestamp_to_micros_with_fractional_seconds_2_digits() {
        let micros = parse_timestamp_to_micros("1970-01-01 00:00:00.12", 0).unwrap();
        assert_eq!(micros, 120_000);
    }

    #[test]
    fn test_parse_timestamp_to_micros_hours_minutes_only() {
        let micros = parse_timestamp_to_micros("1970-01-01 12:30", 0).unwrap();
        assert_eq!(micros, 12 * 3600 * 1_000_000 + 30 * 60 * 1_000_000);
    }

    #[test]
    fn test_parse_timestamp_to_micros_full_time() {
        let micros = parse_timestamp_to_micros("1970-01-01 12:30:45", 0).unwrap();
        assert_eq!(
            micros,
            12 * 3600 * 1_000_000 + 30 * 60 * 1_000_000 + 45 * 1_000_000
        );
    }

    #[test]
    fn test_parse_timestamp_to_micros_empty_string() {
        let result = parse_timestamp_to_micros("", 0);
        assert!(result.is_err());
    }

    #[test]
    fn test_parse_timestamp_to_micros_invalid_hour() {
        let result = parse_timestamp_to_micros("1970-01-01 XX:30:00", 0);
        assert!(result.is_err());
    }

    #[test]
    fn test_parse_timestamp_to_micros_invalid_minute() {
        let result = parse_timestamp_to_micros("1970-01-01 12:XX:00", 0);
        assert!(result.is_err());
    }

    #[test]
    fn test_parse_timestamp_to_micros_invalid_second() {
        let result = parse_timestamp_to_micros("1970-01-01 12:30:XX", 0);
        assert!(result.is_err());
    }

    #[test]
    fn test_build_interval_array_rejects_a_value_that_is_not_a_string() {
        let values = vec![json!(12345)];
        let refs = to_refs(&values);
        let message = conversion_failure_message(build_interval_array(&refs).unwrap_err());
        assert!(message.contains("Expected interval string"), "{}", message);
    }

    #[test]
    fn test_build_interval_array_stores_month_day_nano_values() {
        let values = vec![json!("+01-02"), json!("-03 04:05:06.789"), json!(null)];
        let refs = to_refs(&values);
        let result = build_interval_array(&refs).unwrap();
        let intervals = result
            .as_any()
            .downcast_ref::<arrow::array::IntervalMonthDayNanoArray>()
            .unwrap();
        assert_eq!(
            intervals.value(0),
            arrow::datatypes::IntervalMonthDayNano::new(14, 0, 0)
        );
        assert_eq!(
            intervals.value(1),
            arrow::datatypes::IntervalMonthDayNano::new(0, -3, -14_706_789_000_000)
        );
        assert!(arrow::array::Array::is_null(intervals, 2));
    }

    #[test]
    fn test_build_interval_array_rejects_month_count_outside_i32_range() {
        let values = vec![json!("+01-02"), json!("+999999999-11")];
        let refs = to_refs(&values);
        let (row, message) = value_conversion_failure(build_interval_array(&refs).unwrap_err());
        assert_eq!(row, 1);
        assert!(message.contains("out of range"), "{}", message);
    }

    // ==========================================================================
    // Tests for build_binary_array
    // ==========================================================================

    #[test]
    fn test_build_binary_array_valid_hex() {
        let values = vec![json!("48656c6c6f"), json!("576f726c64")];
        let refs = to_refs(&values);
        let result = build_binary_array(&refs).unwrap();
        assert_eq!(result.len(), 2);
        assert_eq!(result.null_count(), 0);
    }

    #[test]
    fn test_build_binary_array_with_null_values() {
        let values = vec![json!("48656c6c6f"), json!(null), json!("576f726c64")];
        let refs = to_refs(&values);
        let result = build_binary_array(&refs).unwrap();
        assert_eq!(result.len(), 3);
        assert_eq!(result.null_count(), 1);
    }

    #[test]
    fn test_build_binary_array_empty_hex() {
        let values = vec![json!("")];
        let refs = to_refs(&values);
        let result = build_binary_array(&refs).unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result.null_count(), 0);
    }

    #[test]
    fn test_build_binary_array_invalid_not_string() {
        let values = vec![json!(12345)];
        let refs = to_refs(&values);
        let message = conversion_failure_message(build_binary_array(&refs).unwrap_err());
        assert!(message.contains("Expected hex string"), "{}", message);
    }

    #[test]
    fn test_build_binary_array_invalid_hex_odd_length() {
        let values = vec![json!("abc")]; // Odd length
        let refs = to_refs(&values);
        let result = build_binary_array(&refs);
        assert!(result.is_err());
    }

    #[test]
    fn test_build_binary_array_invalid_hex_chars() {
        let values = vec![json!("ghij")]; // Invalid hex chars
        let refs = to_refs(&values);
        let result = build_binary_array(&refs);
        assert!(result.is_err());
    }

    #[test]
    fn test_build_binary_array_capacity_estimation_with_samples() {
        // Create more than 10 values to test sampling
        let values: Vec<Value> = (0..15).map(|_| json!("deadbeef")).collect();
        let refs = to_refs(&values);
        let result = build_binary_array(&refs).unwrap();
        assert_eq!(result.len(), 15);
    }

    #[test]
    fn test_build_binary_array_capacity_estimation_all_nulls() {
        // All null values for capacity estimation
        let values: Vec<Value> = (0..5).map(|_| json!(null)).collect();
        let refs = to_refs(&values);
        let result = build_binary_array(&refs).unwrap();
        assert_eq!(result.len(), 5);
        assert_eq!(result.null_count(), 5);
    }

    #[test]
    fn test_build_double_array_from_integer() {
        let values = vec![json!(42), json!(100)];
        let refs = to_refs(&values);
        let result = build_double_array(&refs).unwrap();
        assert_eq!(result.len(), 2);
    }

    #[test]
    fn test_build_double_array_from_string_numeric() {
        let values = vec![json!("3.14159")];
        let refs = to_refs(&values);
        let result = build_double_array(&refs).unwrap();
        assert_eq!(result.len(), 1);
    }

    #[test]
    fn test_build_double_array_neg_infinity() {
        let values = vec![json!("-Infinity")];
        let refs = to_refs(&values);
        let result = build_double_array(&refs).unwrap();
        assert_eq!(result.len(), 1);
    }

    #[test]
    fn test_build_double_array_nan() {
        let values = vec![json!("NaN")];
        let refs = to_refs(&values);
        let result = build_double_array(&refs).unwrap();
        assert_eq!(result.len(), 1);
    }

    #[test]
    fn test_build_double_array_invalid_string() {
        let values = vec![json!("not a number")];
        let refs = to_refs(&values);
        let result = build_double_array(&refs);
        assert!(result.is_err());
    }

    #[test]
    fn test_parse_decimal_to_i128_from_float() {
        let result = parse_decimal_to_i128(&json!(123.45), 10, 2, 0).unwrap();
        assert_eq!(result, 12345);
    }

    #[test]
    fn test_parse_decimal_to_i128_invalid_decimal_format() {
        let result = parse_decimal_to_i128(&json!("1.2.3"), 10, 2, 0);
        assert!(result.is_err());
    }

    #[test]
    fn test_parse_decimal_to_i128_invalid_integer_part() {
        let result = parse_decimal_to_i128(&json!("abc.45"), 10, 2, 0);
        assert!(result.is_err());
    }

    #[test]
    fn test_parse_decimal_to_i128_invalid_decimal_part() {
        let result = parse_decimal_to_i128(&json!("123.xyz"), 10, 2, 0);
        assert!(result.is_err());
    }

    #[test]
    fn test_parse_decimal_to_i128_non_numeric_value() {
        let result = parse_decimal_to_i128(&json!({"key": "value"}), 10, 2, 0);
        assert!(result.is_err());
    }

    #[test]
    fn test_build_string_array_rejects_non_string() {
        let values = vec![json!("ok"), json!(42)];
        let refs = to_refs(&values);
        let (row, message) = value_conversion_failure(build_string_array(&refs).unwrap_err());
        assert_eq!(row, 1);
        assert!(message.starts_with("Expected string, got:"), "{}", message);
    }

    #[test]
    fn test_build_double_array_rejects_boolean() {
        let values = vec![json!(true)];
        let refs = to_refs(&values);
        let (row, message) = value_conversion_failure(build_double_array(&refs).unwrap_err());
        assert_eq!(row, 0);
        assert!(message.starts_with("Expected number, got:"), "{}", message);
    }

    #[test]
    fn test_validate_decimal_precision_rejects_excess_raw_digits() {
        // Leading zeros keep the significant-digit count within precision, but the
        // raw digit count still exceeds what the column can hold
        let result = validate_decimal_precision("0000123", 3, 0, 4);
        let (row, _, overflow) = value_failure(result.unwrap_err());
        assert_eq!(row, 4);
        assert!(overflow);
    }

    #[test]
    fn test_validate_decimal_precision_allows_one_leading_zero() {
        // "0.12" carries 3 raw digits for precision 2, which is tolerated
        assert!(validate_decimal_precision("0.12", 2, 2, 0).is_ok());
    }

    #[test]
    fn test_decode_hex_uppercase() {
        let bytes = decode_hex("DEADBEEF").unwrap();
        assert_eq!(bytes, vec![0xDE, 0xAD, 0xBE, 0xEF]);
    }

    #[test]
    fn test_decode_hex_mixed_case() {
        let bytes = decode_hex("DeAdBeEf").unwrap();
        assert_eq!(bytes, vec![0xDE, 0xAD, 0xBE, 0xEF]);
    }
}