types.rs

//! Type mapping from DuckLake types to Arrow types

use std::collections::HashMap;

#[cfg(test)]
use std::sync::Arc;

use crate::metadata_provider::DuckLakeTableColumn;
use crate::{DuckLakeError, Result};
use arrow::datatypes::{DataType, Field, IntervalUnit, Schema, TimeUnit};
use parquet::file::metadata::ParquetMetaData;

/// Convert a DuckLake type string to an Arrow DataType
pub fn ducklake_to_arrow_type(ducklake_type: &str) -> Result<DataType> {
    // Normalize type string (lowercase, remove whitespace)
    let normalized = ducklake_type.trim().to_lowercase();

    // Handle parameterized types first
    if let Some(decimal_params) = parse_decimal(&normalized)? {
        return Ok(decimal_params);
    }

    // Handle basic types
    match normalized.as_str() {
        // Boolean
        "boolean" | "bool" => Ok(DataType::Boolean),

        // Integers
        "int8" | "tinyint" => Ok(DataType::Int8),
        "int16" | "smallint" => Ok(DataType::Int16),
        "int32" | "int" | "integer" => Ok(DataType::Int32),
        "int64" | "bigint" | "long" => Ok(DataType::Int64),
        "uint8" | "utinyint" => Ok(DataType::UInt8),
        "uint16" | "usmallint" => Ok(DataType::UInt16),
        "uint32" | "uint" | "uinteger" => Ok(DataType::UInt32),
        "uint64" | "ubigint" => Ok(DataType::UInt64),

        // Floating point
        "float32" | "float" | "real" => Ok(DataType::Float32),
        "float64" | "double" => Ok(DataType::Float64),

        // Temporal types
        "time" => Ok(DataType::Time64(TimeUnit::Microsecond)),
        "date" => Ok(DataType::Date32),
        "timestamp" => Ok(DataType::Timestamp(TimeUnit::Microsecond, None)),
        "timestamptz" | "timestamp with time zone" => Ok(DataType::Timestamp(
            TimeUnit::Microsecond,
            Some("UTC".into()),
        )),
        "timestamp_s" => Ok(DataType::Timestamp(TimeUnit::Second, None)),
        "timestamp_ms" => Ok(DataType::Timestamp(TimeUnit::Millisecond, None)),
        "timestamp_ns" => Ok(DataType::Timestamp(TimeUnit::Nanosecond, None)),
        "interval" => Ok(DataType::Interval(IntervalUnit::MonthDayNano)),

        // String types
        "varchar" | "text" | "string" => Ok(DataType::Utf8),
        "json" => Ok(DataType::Utf8), // JSON stored as UTF8 string

        // Binary types
        "blob" | "binary" | "bytea" => Ok(DataType::Binary),
        "uuid" => Ok(DataType::FixedSizeBinary(16)),

        // Geometry types (stored as binary WKB format)
        "point" | "linestring" | "polygon" | "multipoint" | "multilinestring" | "multipolygon"
        | "geometrycollection" | "linestring z" | "geometry" => Ok(DataType::Binary),

        // Time with timezone - not directly supported, use string
        "timetz" | "time with time zone" => Ok(DataType::Utf8),

        _ => {
            // Check for complex types (list, struct, map)
            if normalized.starts_with("list") || normalized.starts_with("array") {
                Err(DuckLakeError::UnsupportedType(format!(
                    "Complex type '{}' not yet supported. Please open an issue at https://github.com/hotdata-dev/datafusion-ducklake if you need this feature.",
                    ducklake_type
                )))
            } else if normalized.starts_with("struct") {
                Err(DuckLakeError::UnsupportedType(format!(
                    "Struct type '{}' not yet supported. Please open an issue at https://github.com/hotdata-dev/datafusion-ducklake if you need this feature.",
                    ducklake_type
                )))
            } else if normalized.starts_with("map") {
                Err(DuckLakeError::UnsupportedType(format!(
                    "Map type '{}' not yet supported. Please open an issue at https://github.com/hotdata-dev/datafusion-ducklake if you need this feature.",
                    ducklake_type
                )))
            } else {
                Err(DuckLakeError::UnsupportedType(ducklake_type.to_string()))
            }
        },
    }
}

/// Convert an Arrow DataType to a DuckLake type string
///
/// This is the reverse of `ducklake_to_arrow_type()`.
pub fn arrow_to_ducklake_type(arrow_type: &DataType) -> Result<String> {
    match arrow_type {
        // Boolean
        DataType::Boolean => Ok("boolean".to_string()),

        // Integers
        DataType::Int8 => Ok("int8".to_string()),
        DataType::Int16 => Ok("int16".to_string()),
        DataType::Int32 => Ok("int32".to_string()),
        DataType::Int64 => Ok("int64".to_string()),
        DataType::UInt8 => Ok("uint8".to_string()),
        DataType::UInt16 => Ok("uint16".to_string()),
        DataType::UInt32 => Ok("uint32".to_string()),
        DataType::UInt64 => Ok("uint64".to_string()),

        // Floating point
        DataType::Float32 => Ok("float32".to_string()),
        DataType::Float64 => Ok("float64".to_string()),

        // Temporal types
        DataType::Date32 | DataType::Date64 => Ok("date".to_string()),
        DataType::Time32(_) | DataType::Time64(_) => Ok("time".to_string()),
        DataType::Timestamp(TimeUnit::Second, None) => Ok("timestamp_s".to_string()),
        DataType::Timestamp(TimeUnit::Millisecond, None) => Ok("timestamp_ms".to_string()),
        DataType::Timestamp(TimeUnit::Microsecond, None) => Ok("timestamp".to_string()),
        DataType::Timestamp(TimeUnit::Nanosecond, None) => Ok("timestamp_ns".to_string()),
        DataType::Timestamp(_, Some(_)) => Ok("timestamptz".to_string()),
        DataType::Interval(_) => Ok("interval".to_string()),

        // String types
        DataType::Utf8 | DataType::LargeUtf8 => Ok("varchar".to_string()),

        // Binary types
        DataType::Binary | DataType::LargeBinary => Ok("blob".to_string()),
        DataType::FixedSizeBinary(16) => Ok("uuid".to_string()),
        DataType::FixedSizeBinary(_) => Ok("blob".to_string()),

        // Decimal types
        DataType::Decimal128(precision, scale) | DataType::Decimal256(precision, scale) => {
            Ok(format!("decimal({}, {})", precision, scale))
        },

        // Null type - map to varchar as there's no direct equivalent
        DataType::Null => Ok("varchar".to_string()),

        // Complex types - not yet supported for writing
        DataType::List(_) | DataType::LargeList(_) | DataType::FixedSizeList(_, _) => {
            Err(DuckLakeError::UnsupportedType(format!(
                "List type '{}' not yet supported for writing",
                arrow_type
            )))
        },
        DataType::Struct(_) => Err(DuckLakeError::UnsupportedType(format!(
            "Struct type '{}' not yet supported for writing",
            arrow_type
        ))),
        DataType::Map(_, _) => Err(DuckLakeError::UnsupportedType(format!(
            "Map type '{}' not yet supported for writing",
            arrow_type
        ))),

        // Other unsupported types
        other => Err(DuckLakeError::UnsupportedType(format!(
            "Arrow type '{}' has no DuckLake equivalent",
            other
        ))),
    }
}

/// Maximum precision for Arrow Decimal256
const DECIMAL_MAX_PRECISION: u8 = 76;

/// Validate decimal precision and scale bounds
fn validate_decimal_precision_scale(precision: u8, scale: i8, type_str: &str) -> Result<()> {
    if precision == 0 {
        return Err(DuckLakeError::UnsupportedType(format!(
            "Decimal precision must be >= 1, got 0 in type '{}'",
            type_str
        )));
    }
    if precision > DECIMAL_MAX_PRECISION {
        return Err(DuckLakeError::UnsupportedType(format!(
            "Decimal precision must be <= {}, got {} in type '{}'",
            DECIMAL_MAX_PRECISION, precision, type_str
        )));
    }
    if scale >= 0 && scale as u8 > precision {
        return Err(DuckLakeError::UnsupportedType(format!(
            "Decimal scale ({}) must not exceed precision ({}) in type '{}'",
            scale, precision, type_str
        )));
    }
    Ok(())
}

/// Parse decimal type with precision and scale
/// Format: "decimal(precision, scale)" or "decimal(precision)"
///
/// Returns `Ok(None)` if the type string is not a decimal type.
/// Returns `Err` if it is a decimal type but has invalid precision/scale.
fn parse_decimal(type_str: &str) -> Result<Option<DataType>> {
    if !type_str.starts_with("decimal") && !type_str.starts_with("numeric") {
        return Ok(None);
    }

    // Extract parameters from parentheses
    let start = match type_str.find('(') {
        Some(s) => s,
        None => return Ok(None),
    };
    let end = match type_str.find(')') {
        Some(e) => e,
        None => return Ok(None),
    };
    let params = &type_str[start + 1..end];

    let parts: Vec<&str> = params.split(',').map(|s| s.trim()).collect();

    match parts.len() {
        1 => {
            let precision: u8 = parts[0].parse().map_err(|_| {
                DuckLakeError::UnsupportedType(format!(
                    "Invalid decimal precision '{}' in type '{}'",
                    parts[0], type_str
                ))
            })?;
            validate_decimal_precision_scale(precision, 0, type_str)?;
            Ok(Some(DataType::Decimal128(precision, 0)))
        },
        2 => {
            let precision: u8 = parts[0].parse().map_err(|_| {
                DuckLakeError::UnsupportedType(format!(
                    "Invalid decimal precision '{}' in type '{}'",
                    parts[0], type_str
                ))
            })?;
            let scale: i8 = parts[1].parse().map_err(|_| {
                DuckLakeError::UnsupportedType(format!(
                    "Invalid decimal scale '{}' in type '{}'",
                    parts[1], type_str
                ))
            })?;
            validate_decimal_precision_scale(precision, scale, type_str)?;
            if precision > 38 {
                Ok(Some(DataType::Decimal256(precision, scale)))
            } else {
                Ok(Some(DataType::Decimal128(precision, scale)))
            }
        },
        n => Err(DuckLakeError::UnsupportedType(format!(
            "Invalid decimal type: expected at most 2 parameters (precision, scale), got {} in type '{}'",
            n, type_str
        ))),
    }
}

/// Normalize a DuckLake type string to its canonical form.
///
/// Converts aliases and case variants to the canonical DuckLake type string.
/// For example: "int" -> "int32", "INTEGER" -> "int32", "text" -> "varchar".
///
/// Returns the canonical type string, or an error if the type is unrecognized.
pub fn normalize_ducklake_type(ducklake_type: &str) -> Result<String> {
    let arrow_type = ducklake_to_arrow_type(ducklake_type)?;
    arrow_to_ducklake_type(&arrow_type)
}

/// Check if a type can be safely promoted (widened) to another type.
///
/// Type promotion allows safe widening of numeric types during schema evolution.
/// Both type strings are normalized before comparison.
///
/// Supported promotions:
/// - Signed integer widening: int8 -> int16 -> int32 -> int64
/// - Unsigned integer widening: uint8 -> uint16 -> uint32 -> uint64
/// - Float widening: float32 -> float64
/// - Integer to float: any int -> float64
/// - Timestamp: timestamp -> timestamptz
/// - Decimal: smaller precision/scale -> larger precision/scale
pub fn is_promotable(from: &str, to: &str) -> bool {
    let from_arrow = match ducklake_to_arrow_type(from) {
        Ok(t) => t,
        Err(_) => return false,
    };
    let to_arrow = match ducklake_to_arrow_type(to) {
        Ok(t) => t,
        Err(_) => return false,
    };

    is_arrow_promotable(&from_arrow, &to_arrow)
}

/// Check if one Arrow DataType can be safely promoted to another.
fn is_arrow_promotable(from: &DataType, to: &DataType) -> bool {
    use DataType::*;

    // Same type is trivially promotable
    if from == to {
        return true;
    }

    fn signed_int_rank(dt: &DataType) -> Option<u8> {
        match dt {
            Int8 => Some(0),
            Int16 => Some(1),
            Int32 => Some(2),
            Int64 => Some(3),
            _ => None,
        }
    }

    fn unsigned_int_rank(dt: &DataType) -> Option<u8> {
        match dt {
            UInt8 => Some(0),
            UInt16 => Some(1),
            UInt32 => Some(2),
            UInt64 => Some(3),
            _ => None,
        }
    }

    // Signed integer widening
    if let (Some(from_rank), Some(to_rank)) = (signed_int_rank(from), signed_int_rank(to)) {
        return from_rank < to_rank;
    }

    // Unsigned integer widening
    if let (Some(from_rank), Some(to_rank)) = (unsigned_int_rank(from), unsigned_int_rank(to)) {
        return from_rank < to_rank;
    }

    // Float widening
    if matches!(from, Float32) && matches!(to, Float64) {
        return true;
    }

    // Integer to float64 (safe for reasonable values)
    if signed_int_rank(from).is_some() && matches!(to, Float64) {
        return true;
    }

    // Timestamp -> TimestampTZ
    if matches!(from, Timestamp(_, None)) && matches!(to, Timestamp(_, Some(_))) {
        return true;
    }

    // Decimal widening: integer digits don't shrink AND fractional digits don't shrink.
    // We check (tp - ts >= fp - fs) to ensure the integer part has enough room,
    // in addition to (ts >= fs) for the fractional part.
    match (from, to) {
        (Decimal128(fp, fs) | Decimal256(fp, fs), Decimal128(tp, ts) | Decimal256(tp, ts)) => {
            let from_int_digits = *fp as i16 - *fs as i16;
            let to_int_digits = *tp as i16 - *ts as i16;
            ts >= fs && to_int_digits >= from_int_digits
        },
        _ => false,
    }
}

/// Check if two DuckLake type strings are compatible for schema evolution.
///
/// Types are compatible if they normalize to the same canonical type,
/// or if the existing type can be safely promoted to the new type.
pub fn types_compatible(existing_type: &str, new_type: &str) -> bool {
    // First try normalization: if both normalize to the same canonical form, they match
    let existing_normalized = match normalize_ducklake_type(existing_type) {
        Ok(t) => t,
        Err(_) => return false,
    };
    let new_normalized = match normalize_ducklake_type(new_type) {
        Ok(t) => t,
        Err(_) => return false,
    };

    if existing_normalized == new_normalized {
        return true;
    }

    // Then check if promotion is allowed
    is_promotable(existing_type, new_type)
}

/// Build an Arrow schema from a list of DuckLake table columns
pub fn build_arrow_schema(columns: &[DuckLakeTableColumn]) -> Result<Schema> {
    let fields: Result<Vec<Field>> = columns
        .iter()
        .map(|col| {
            let data_type = ducklake_to_arrow_type(&col.column_type)?;
            Ok(Field::new(&col.column_name, data_type, col.is_nullable))
        })
        .collect();

    Ok(Schema::new(fields?))
}

/// Extract field_id to column_name mapping from Parquet metadata.
/// DuckLake column_id == Parquet field_id, enabling column matching after renames.
pub fn extract_parquet_field_ids(metadata: &ParquetMetaData) -> HashMap<i32, String> {
    let schema_descr = metadata.file_metadata().schema_descr();
    let mut field_id_map = HashMap::new();

    for i in 0..schema_descr.num_columns() {
        let column = schema_descr.column(i);
        let basic_info = column.self_type().get_basic_info();

        if basic_info.has_id() {
            let field_id = basic_info.id();
            let column_name = column.name().to_string();
            field_id_map.insert(field_id, column_name);
        }
    }

    field_id_map
}

/// Build a schema for reading Parquet files with renamed columns.
/// Returns (read_schema, name_mapping) where read_schema uses original Parquet names
/// and name_mapping maps old->new for columns that were renamed.
pub fn build_read_schema_with_field_id_mapping(
    current_columns: &[DuckLakeTableColumn],
    parquet_field_ids: &HashMap<i32, String>,
) -> Result<(Schema, HashMap<String, String>)> {
    let mut name_mapping: HashMap<String, String> = HashMap::new();

    let fields: Result<Vec<Field>> = current_columns
        .iter()
        .map(|col| {
            let data_type = ducklake_to_arrow_type(&col.column_type)?;
            let field_id = i32::try_from(col.column_id).map_err(|_| {
                DuckLakeError::Internal(format!(
                    "column_id {} for column '{}' exceeds i32 range for Parquet field_id",
                    col.column_id, col.column_name
                ))
            })?;

            let (read_name, needs_rename) =
                if let Some(parquet_name) = parquet_field_ids.get(&field_id) {
                    if parquet_name != &col.column_name {
                        (parquet_name.clone(), true) // Column was renamed
                    } else {
                        (col.column_name.clone(), false)
                    }
                } else {
                    (col.column_name.clone(), false) // No field_id, use current name
                };

            if needs_rename {
                name_mapping.insert(read_name.clone(), col.column_name.clone());
            }

            Ok(Field::new(read_name, data_type, col.is_nullable))
        })
        .collect();

    Ok((Schema::new(fields?), name_mapping))
}

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

    #[test]
    fn test_build_read_schema_with_renamed_columns() {
        // Simulate: column was originally named "user_id", now renamed to "userId"
        let current_columns = vec![
            DuckLakeTableColumn {
                column_id: 1,
                column_name: "userId".to_string(), // Current name (renamed)
                column_type: "int32".to_string(),
                is_nullable: true,
            },
            DuckLakeTableColumn {
                column_id: 2,
                column_name: "name".to_string(), // Not renamed
                column_type: "varchar".to_string(),
                is_nullable: true,
            },
        ];

        // Parquet file has original names
        let mut parquet_field_ids = HashMap::new();
        parquet_field_ids.insert(1, "user_id".to_string()); // Original name
        parquet_field_ids.insert(2, "name".to_string()); // Same name

        let (read_schema, name_mapping) =
            build_read_schema_with_field_id_mapping(&current_columns, &parquet_field_ids).unwrap();

        // Read schema should have original Parquet names
        assert_eq!(read_schema.field(0).name(), "user_id");
        assert_eq!(read_schema.field(1).name(), "name");

        // Name mapping should map old name to new name
        assert_eq!(name_mapping.len(), 1);
        assert_eq!(name_mapping.get("user_id"), Some(&"userId".to_string()));
    }

    #[test]
    fn test_build_read_schema_no_rename_needed() {
        let current_columns = vec![DuckLakeTableColumn {
            column_id: 1,
            column_name: "id".to_string(),
            column_type: "int32".to_string(),
            is_nullable: true,
        }];

        let mut parquet_field_ids = HashMap::new();
        parquet_field_ids.insert(1, "id".to_string()); // Same name

        let (read_schema, name_mapping) =
            build_read_schema_with_field_id_mapping(&current_columns, &parquet_field_ids).unwrap();

        assert_eq!(read_schema.field(0).name(), "id");
        assert!(name_mapping.is_empty()); // No rename needed
    }

    #[test]
    fn test_build_read_schema_no_field_ids() {
        // External file without field_ids
        let current_columns = vec![DuckLakeTableColumn {
            column_id: 1,
            column_name: "id".to_string(),
            column_type: "int32".to_string(),
            is_nullable: true,
        }];

        let parquet_field_ids = HashMap::new(); // No field_ids in Parquet

        let (read_schema, name_mapping) =
            build_read_schema_with_field_id_mapping(&current_columns, &parquet_field_ids).unwrap();

        // Falls back to current column name
        assert_eq!(read_schema.field(0).name(), "id");
        assert!(name_mapping.is_empty());
    }

    #[test]
    fn test_basic_types() {
        assert_eq!(
            ducklake_to_arrow_type("boolean").unwrap(),
            DataType::Boolean
        );
        assert_eq!(ducklake_to_arrow_type("int32").unwrap(), DataType::Int32);
        assert_eq!(ducklake_to_arrow_type("int64").unwrap(), DataType::Int64);
        assert_eq!(
            ducklake_to_arrow_type("float64").unwrap(),
            DataType::Float64
        );
        assert_eq!(ducklake_to_arrow_type("varchar").unwrap(), DataType::Utf8);
        assert_eq!(ducklake_to_arrow_type("blob").unwrap(), DataType::Binary);
    }

    #[test]
    fn test_decimal_types() {
        assert_eq!(
            ducklake_to_arrow_type("decimal(10, 2)").unwrap(),
            DataType::Decimal128(10, 2)
        );
        assert_eq!(
            ducklake_to_arrow_type("decimal(38, 10)").unwrap(),
            DataType::Decimal128(38, 10)
        );
    }

    #[test]
    fn test_temporal_types() {
        assert_eq!(ducklake_to_arrow_type("date").unwrap(), DataType::Date32);
        assert_eq!(
            ducklake_to_arrow_type("timestamp").unwrap(),
            DataType::Timestamp(TimeUnit::Microsecond, None)
        );
    }

    #[test]
    fn test_unsupported_list_type_errors() {
        // Test list type returns error
        let result = ducklake_to_arrow_type("list<int32>");
        assert!(result.is_err());
        match result {
            Err(DuckLakeError::UnsupportedType(msg)) => {
                assert!(msg.contains("list<int32>"));
                assert!(msg.contains("not yet supported"));
                assert!(msg.contains("open an issue"));
            },
            _ => panic!("Expected UnsupportedType error for list type"),
        }
    }

    #[test]
    fn test_unsupported_array_type_errors() {
        // Test array type returns error
        let result = ducklake_to_arrow_type("array<varchar>");
        assert!(result.is_err());
        match result {
            Err(DuckLakeError::UnsupportedType(msg)) => {
                assert!(msg.contains("array<varchar>"));
                assert!(msg.contains("not yet supported"));
            },
            _ => panic!("Expected UnsupportedType error for array type"),
        }
    }

    #[test]
    fn test_unsupported_struct_type_errors() {
        // Test struct type returns error
        let result = ducklake_to_arrow_type("struct<a:int32,b:varchar>");
        assert!(result.is_err());
        match result {
            Err(DuckLakeError::UnsupportedType(msg)) => {
                assert!(msg.contains("struct<a:int32,b:varchar>"));
                assert!(msg.contains("not yet supported"));
                assert!(msg.contains("open an issue"));
            },
            _ => panic!("Expected UnsupportedType error for struct type"),
        }
    }

    #[test]
    fn test_unsupported_map_type_errors() {
        // Test map type returns error
        let result = ducklake_to_arrow_type("map<varchar,int32>");
        assert!(result.is_err());
        match result {
            Err(DuckLakeError::UnsupportedType(msg)) => {
                assert!(msg.contains("map<varchar,int32>"));
                assert!(msg.contains("not yet supported"));
                assert!(msg.contains("open an issue"));
            },
            _ => panic!("Expected UnsupportedType error for map type"),
        }
    }

    #[test]
    fn test_nested_complex_types_error() {
        // Test nested complex types return error
        let result = ducklake_to_arrow_type("list<struct<a:int32,b:varchar>>");
        assert!(result.is_err());
        match result {
            Err(DuckLakeError::UnsupportedType(msg)) => {
                assert!(msg.contains("list<struct<a:int32,b:varchar>>"));
                assert!(msg.contains("not yet supported"));
            },
            _ => panic!("Expected UnsupportedType error for nested complex type"),
        }
    }

    #[test]
    fn test_unknown_type_error() {
        // Test completely unknown types also return error
        let result = ducklake_to_arrow_type("completely_unknown_type");
        assert!(result.is_err());
        match result {
            Err(DuckLakeError::UnsupportedType(msg)) => {
                assert_eq!(msg, "completely_unknown_type");
            },
            _ => panic!("Expected UnsupportedType error for unknown type"),
        }
    }

    #[test]
    fn test_arrow_to_ducklake_basic_types() {
        assert_eq!(
            arrow_to_ducklake_type(&DataType::Boolean).unwrap(),
            "boolean"
        );
        assert_eq!(arrow_to_ducklake_type(&DataType::Int8).unwrap(), "int8");
        assert_eq!(arrow_to_ducklake_type(&DataType::Int16).unwrap(), "int16");
        assert_eq!(arrow_to_ducklake_type(&DataType::Int32).unwrap(), "int32");
        assert_eq!(arrow_to_ducklake_type(&DataType::Int64).unwrap(), "int64");
        assert_eq!(arrow_to_ducklake_type(&DataType::UInt8).unwrap(), "uint8");
        assert_eq!(arrow_to_ducklake_type(&DataType::UInt16).unwrap(), "uint16");
        assert_eq!(arrow_to_ducklake_type(&DataType::UInt32).unwrap(), "uint32");
        assert_eq!(arrow_to_ducklake_type(&DataType::UInt64).unwrap(), "uint64");
        assert_eq!(
            arrow_to_ducklake_type(&DataType::Float32).unwrap(),
            "float32"
        );
        assert_eq!(
            arrow_to_ducklake_type(&DataType::Float64).unwrap(),
            "float64"
        );
        assert_eq!(arrow_to_ducklake_type(&DataType::Utf8).unwrap(), "varchar");
        assert_eq!(arrow_to_ducklake_type(&DataType::Binary).unwrap(), "blob");
    }

    #[test]
    fn test_arrow_to_ducklake_temporal_types() {
        assert_eq!(arrow_to_ducklake_type(&DataType::Date32).unwrap(), "date");
        assert_eq!(arrow_to_ducklake_type(&DataType::Date64).unwrap(), "date");
        assert_eq!(
            arrow_to_ducklake_type(&DataType::Time64(TimeUnit::Microsecond)).unwrap(),
            "time"
        );
        assert_eq!(
            arrow_to_ducklake_type(&DataType::Timestamp(TimeUnit::Microsecond, None)).unwrap(),
            "timestamp"
        );
        assert_eq!(
            arrow_to_ducklake_type(&DataType::Timestamp(
                TimeUnit::Microsecond,
                Some("UTC".into())
            ))
            .unwrap(),
            "timestamptz"
        );
    }

    #[test]
    fn test_arrow_to_ducklake_decimal() {
        assert_eq!(
            arrow_to_ducklake_type(&DataType::Decimal128(10, 2)).unwrap(),
            "decimal(10, 2)"
        );
        assert_eq!(
            arrow_to_ducklake_type(&DataType::Decimal256(40, 5)).unwrap(),
            "decimal(40, 5)"
        );
    }

    #[test]
    fn test_arrow_to_ducklake_uuid() {
        assert_eq!(
            arrow_to_ducklake_type(&DataType::FixedSizeBinary(16)).unwrap(),
            "uuid"
        );
        // Non-16 byte fixed size binary becomes blob
        assert_eq!(
            arrow_to_ducklake_type(&DataType::FixedSizeBinary(32)).unwrap(),
            "blob"
        );
    }

    #[test]
    fn test_arrow_to_ducklake_roundtrip() {
        // Verify roundtrip: arrow -> ducklake -> arrow for common types
        let test_types = vec![
            DataType::Boolean,
            DataType::Int32,
            DataType::Int64,
            DataType::Float64,
            DataType::Utf8,
            DataType::Binary,
            DataType::Date32,
            DataType::Timestamp(TimeUnit::Microsecond, None),
            DataType::Decimal128(10, 2),
        ];

        for original in test_types {
            let ducklake = arrow_to_ducklake_type(&original).unwrap();
            let back = ducklake_to_arrow_type(&ducklake).unwrap();
            assert_eq!(original, back, "Roundtrip failed for {:?}", original);
        }
    }

    #[test]
    fn test_arrow_to_ducklake_unsupported_list() {
        let list_type = DataType::List(Arc::new(Field::new("item", DataType::Int32, true)));
        let result = arrow_to_ducklake_type(&list_type);
        assert!(result.is_err());
        match result {
            Err(DuckLakeError::UnsupportedType(msg)) => {
                assert!(msg.contains("List type"));
                assert!(msg.contains("not yet supported"));
            },
            _ => panic!("Expected UnsupportedType error"),
        }
    }

    #[test]
    fn test_arrow_to_ducklake_unsupported_struct() {
        let struct_type = DataType::Struct(vec![Field::new("a", DataType::Int32, true)].into());
        let result = arrow_to_ducklake_type(&struct_type);
        assert!(result.is_err());
        match result {
            Err(DuckLakeError::UnsupportedType(msg)) => {
                assert!(msg.contains("Struct type"));
                assert!(msg.contains("not yet supported"));
            },
            _ => panic!("Expected UnsupportedType error"),
        }
    }

    #[test]
    fn test_decimal_precision_zero_rejected() {
        let result = ducklake_to_arrow_type("decimal(0, 0)");
        assert!(result.is_err());
        match result {
            Err(DuckLakeError::UnsupportedType(msg)) => {
                assert!(msg.contains("precision must be >= 1"));
            },
            _ => panic!("Expected UnsupportedType error for precision=0"),
        }
    }

    #[test]
    fn test_decimal_precision_too_large_rejected() {
        let result = ducklake_to_arrow_type("decimal(77, 0)");
        assert!(result.is_err());
        match result {
            Err(DuckLakeError::UnsupportedType(msg)) => {
                assert!(msg.contains("precision must be <= 76"));
            },
            _ => panic!("Expected UnsupportedType error for precision=77"),
        }
    }

    #[test]
    fn test_decimal_precision_255_rejected() {
        let result = ducklake_to_arrow_type("decimal(255, 0)");
        assert!(result.is_err());
        match result {
            Err(DuckLakeError::UnsupportedType(msg)) => {
                assert!(msg.contains("precision must be <= 76"));
            },
            _ => panic!("Expected UnsupportedType error for precision=255"),
        }
    }

    #[test]
    fn test_decimal_scale_exceeds_precision_rejected() {
        let result = ducklake_to_arrow_type("decimal(10, 11)");
        assert!(result.is_err());
        match result {
            Err(DuckLakeError::UnsupportedType(msg)) => {
                assert!(msg.contains("scale (11) must not exceed precision (10)"));
            },
            _ => panic!("Expected UnsupportedType error for scale > precision"),
        }
    }

    #[test]
    fn test_decimal_valid_edge_cases() {
        assert_eq!(
            ducklake_to_arrow_type("decimal(1, 0)").unwrap(),
            DataType::Decimal128(1, 0)
        );
        assert_eq!(
            ducklake_to_arrow_type("decimal(38, 0)").unwrap(),
            DataType::Decimal128(38, 0)
        );
        assert_eq!(
            ducklake_to_arrow_type("decimal(39, 0)").unwrap(),
            DataType::Decimal256(39, 0)
        );
        assert_eq!(
            ducklake_to_arrow_type("decimal(76, 0)").unwrap(),
            DataType::Decimal256(76, 0)
        );
        assert_eq!(
            ducklake_to_arrow_type("decimal(10, 10)").unwrap(),
            DataType::Decimal128(10, 10)
        );
    }

    #[test]
    fn test_decimal_negative_precision_rejected() {
        let result = ducklake_to_arrow_type("decimal(-1, 0)");
        assert!(result.is_err());
    }

    #[test]
    fn test_decimal_too_many_parameters_rejected() {
        let result = ducklake_to_arrow_type("decimal(1,2,3)");
        assert!(result.is_err());
        match result {
            Err(DuckLakeError::UnsupportedType(msg)) => {
                assert!(msg.contains("expected at most 2 parameters"));
                assert!(msg.contains("got 3"));
            },
            _ => panic!("Expected UnsupportedType error for 3 parameters"),
        }

        let result = ducklake_to_arrow_type("decimal(10,2,5,3)");
        assert!(result.is_err());
        match result {
            Err(DuckLakeError::UnsupportedType(msg)) => {
                assert!(msg.contains("expected at most 2 parameters"));
                assert!(msg.contains("got 4"));
            },
            _ => panic!("Expected UnsupportedType error for 4 parameters"),
        }
    }

    #[test]
    fn test_decimal_negative_scale_valid() {
        assert_eq!(
            ducklake_to_arrow_type("decimal(10, -2)").unwrap(),
            DataType::Decimal128(10, -2)
        );
    }

    #[test]
    fn test_build_schema_with_unsupported_type() {
        // Test that build_arrow_schema propagates complex type errors
        let columns = vec![
            DuckLakeTableColumn {
                column_id: 1,
                column_name: "id".to_string(),
                column_type: "int32".to_string(),
                is_nullable: true,
            },
            DuckLakeTableColumn {
                column_id: 2,
                column_name: "data".to_string(),
                column_type: "list<int32>".to_string(),
                is_nullable: true,
            },
        ];

        let result = build_arrow_schema(&columns);
        assert!(result.is_err());
        match result {
            Err(DuckLakeError::UnsupportedType(msg)) => {
                assert!(msg.contains("list<int32>"));
            },
            _ => panic!("Expected UnsupportedType error when building schema with complex type"),
        }
    }

    #[test]
    fn test_column_id_i32_max_succeeds() {
        let columns = vec![DuckLakeTableColumn {
            column_id: i32::MAX as i64,
            column_name: "id".to_string(),
            column_type: "int32".to_string(),
            is_nullable: true,
        }];

        let mut parquet_field_ids = HashMap::new();
        parquet_field_ids.insert(i32::MAX, "id".to_string());

        let result = build_read_schema_with_field_id_mapping(&columns, &parquet_field_ids);
        assert!(result.is_ok(), "column_id = i32::MAX should succeed");
    }

    #[test]
    fn test_column_id_overflow_returns_error() {
        let columns = vec![DuckLakeTableColumn {
            column_id: i32::MAX as i64 + 1, // 2147483648, exceeds i32 range
            column_name: "id".to_string(),
            column_type: "int32".to_string(),
            is_nullable: true,
        }];

        let parquet_field_ids = HashMap::new();

        let result = build_read_schema_with_field_id_mapping(&columns, &parquet_field_ids);
        assert!(result.is_err(), "column_id > i32::MAX should fail");
        match result {
            Err(DuckLakeError::Internal(msg)) => {
                assert!(
                    msg.contains("2147483648"),
                    "Error should contain the overflowing value: {}",
                    msg
                );
                assert!(
                    msg.contains("exceeds i32 range"),
                    "Error should explain the issue: {}",
                    msg
                );
            },
            _ => panic!("Expected Internal error for column_id overflow"),
        }
    }

    #[test]
    fn test_column_id_negative_within_i32_range_succeeds() {
        let columns = vec![DuckLakeTableColumn {
            column_id: -1,
            column_name: "id".to_string(),
            column_type: "int32".to_string(),
            is_nullable: true,
        }];

        let mut parquet_field_ids = HashMap::new();
        parquet_field_ids.insert(-1_i32, "id".to_string());

        let result = build_read_schema_with_field_id_mapping(&columns, &parquet_field_ids);
        assert!(
            result.is_ok(),
            "Negative column_id within i32 range should succeed"
        );
    }

    // ── normalize_ducklake_type tests ──

    #[test]
    fn test_normalize_int_aliases() {
        assert_eq!(normalize_ducklake_type("int").unwrap(), "int32");
        assert_eq!(normalize_ducklake_type("integer").unwrap(), "int32");
        assert_eq!(normalize_ducklake_type("INT").unwrap(), "int32");
        assert_eq!(normalize_ducklake_type("Integer").unwrap(), "int32");
        assert_eq!(normalize_ducklake_type("int32").unwrap(), "int32");
    }

    #[test]
    fn test_normalize_bigint_aliases() {
        assert_eq!(normalize_ducklake_type("bigint").unwrap(), "int64");
        assert_eq!(normalize_ducklake_type("long").unwrap(), "int64");
        assert_eq!(normalize_ducklake_type("BIGINT").unwrap(), "int64");
        assert_eq!(normalize_ducklake_type("int64").unwrap(), "int64");
    }

    #[test]
    fn test_normalize_string_aliases() {
        assert_eq!(normalize_ducklake_type("text").unwrap(), "varchar");
        assert_eq!(normalize_ducklake_type("string").unwrap(), "varchar");
        assert_eq!(normalize_ducklake_type("varchar").unwrap(), "varchar");
        assert_eq!(normalize_ducklake_type("TEXT").unwrap(), "varchar");
        assert_eq!(normalize_ducklake_type("STRING").unwrap(), "varchar");
    }

    #[test]
    fn test_normalize_float_aliases() {
        assert_eq!(normalize_ducklake_type("float").unwrap(), "float32");
        assert_eq!(normalize_ducklake_type("real").unwrap(), "float32");
        assert_eq!(normalize_ducklake_type("FLOAT").unwrap(), "float32");
        assert_eq!(normalize_ducklake_type("float32").unwrap(), "float32");
    }

    #[test]
    fn test_normalize_double_aliases() {
        assert_eq!(normalize_ducklake_type("double").unwrap(), "float64");
        assert_eq!(normalize_ducklake_type("DOUBLE").unwrap(), "float64");
        assert_eq!(normalize_ducklake_type("float64").unwrap(), "float64");
    }

    #[test]
    fn test_normalize_bool_aliases() {
        assert_eq!(normalize_ducklake_type("bool").unwrap(), "boolean");
        assert_eq!(normalize_ducklake_type("boolean").unwrap(), "boolean");
        assert_eq!(normalize_ducklake_type("BOOLEAN").unwrap(), "boolean");
    }

    #[test]
    fn test_normalize_smallint_aliases() {
        assert_eq!(normalize_ducklake_type("smallint").unwrap(), "int16");
        assert_eq!(normalize_ducklake_type("SMALLINT").unwrap(), "int16");
        assert_eq!(normalize_ducklake_type("int16").unwrap(), "int16");
    }

    #[test]
    fn test_normalize_tinyint_aliases() {
        assert_eq!(normalize_ducklake_type("tinyint").unwrap(), "int8");
        assert_eq!(normalize_ducklake_type("TINYINT").unwrap(), "int8");
        assert_eq!(normalize_ducklake_type("int8").unwrap(), "int8");
    }

    #[test]
    fn test_normalize_unknown_type_errors() {
        assert!(normalize_ducklake_type("foobar").is_err());
    }

    // ── is_promotable tests ──

    #[test]
    fn test_promotable_same_type() {
        assert!(is_promotable("int32", "int32"));
        assert!(is_promotable("varchar", "varchar"));
        assert!(is_promotable("float64", "float64"));
    }

    #[test]
    fn test_promotable_signed_int_widening() {
        assert!(is_promotable("int8", "int16"));
        assert!(is_promotable("int8", "int32"));
        assert!(is_promotable("int8", "int64"));
        assert!(is_promotable("int16", "int32"));
        assert!(is_promotable("int16", "int64"));
        assert!(is_promotable("int32", "int64"));
    }

    #[test]
    fn test_promotable_signed_int_narrowing_rejected() {
        assert!(!is_promotable("int64", "int32"));
        assert!(!is_promotable("int32", "int16"));
        assert!(!is_promotable("int16", "int8"));
    }

    #[test]
    fn test_promotable_unsigned_int_widening() {
        assert!(is_promotable("uint8", "uint16"));
        assert!(is_promotable("uint8", "uint32"));
        assert!(is_promotable("uint8", "uint64"));
        assert!(is_promotable("uint16", "uint32"));
        assert!(is_promotable("uint32", "uint64"));
    }

    #[test]
    fn test_promotable_unsigned_narrowing_rejected() {
        assert!(!is_promotable("uint64", "uint32"));
        assert!(!is_promotable("uint32", "uint16"));
    }

    #[test]
    fn test_promotable_float_widening() {
        assert!(is_promotable("float32", "float64"));
    }

    #[test]
    fn test_promotable_float_narrowing_rejected() {
        assert!(!is_promotable("float64", "float32"));
    }

    #[test]
    fn test_promotable_int_to_float64() {
        assert!(is_promotable("int8", "float64"));
        assert!(is_promotable("int16", "float64"));
        assert!(is_promotable("int32", "float64"));
        assert!(is_promotable("int64", "float64"));
    }

    #[test]
    fn test_promotable_int_to_float32_rejected() {
        // We only allow int -> float64, not int -> float32
        assert!(!is_promotable("int32", "float32"));
    }

    #[test]
    fn test_promotable_timestamp_to_timestamptz() {
        assert!(is_promotable("timestamp", "timestamptz"));
    }

    #[test]
    fn test_promotable_timestamptz_to_timestamp_rejected() {
        assert!(!is_promotable("timestamptz", "timestamp"));
    }

    #[test]
    fn test_promotable_decimal_widening() {
        assert!(is_promotable("decimal(10, 2)", "decimal(18, 4)"));
        assert!(is_promotable("decimal(10, 2)", "decimal(10, 2)")); // same
        assert!(is_promotable("decimal(10, 2)", "decimal(20, 2)")); // wider precision
        assert!(is_promotable("decimal(10, 2)", "decimal(12, 4)")); // wider scale with enough integer digits
    }

    #[test]
    fn test_promotable_decimal_narrowing_rejected() {
        assert!(!is_promotable("decimal(18, 4)", "decimal(10, 2)"));
        assert!(!is_promotable("decimal(20, 2)", "decimal(10, 2)")); // narrower precision
        // Scale widening that shrinks integer digits: 10-2=8 integer digits vs 12-8=4
        assert!(!is_promotable("decimal(10, 2)", "decimal(12, 8)"));
    }

    #[test]
    fn test_promotable_incompatible_types() {
        assert!(!is_promotable("int32", "varchar"));
        assert!(!is_promotable("varchar", "int32"));
        assert!(!is_promotable("boolean", "int32"));
        assert!(!is_promotable("date", "timestamp"));
    }

    #[test]
    fn test_promotable_unknown_types() {
        assert!(!is_promotable("foobar", "int32"));
        assert!(!is_promotable("int32", "foobar"));
    }

    #[test]
    fn test_promotable_with_aliases() {
        // Uses normalized forms internally
        assert!(is_promotable("int", "bigint")); // int32 -> int64
        assert!(is_promotable("tinyint", "integer")); // int8 -> int32
        assert!(is_promotable("float", "double")); // float32 -> float64
    }

    // ── types_compatible tests ──

    #[test]
    fn test_types_compatible_same_canonical() {
        assert!(types_compatible("int", "int32"));
        assert!(types_compatible("int32", "int"));
        assert!(types_compatible("integer", "int"));
        assert!(types_compatible("text", "varchar"));
        assert!(types_compatible("string", "text"));
        assert!(types_compatible("bigint", "int64"));
        assert!(types_compatible("float", "real"));
        assert!(types_compatible("double", "float64"));
        assert!(types_compatible("bool", "boolean"));
    }

    #[test]
    fn test_types_compatible_case_insensitive() {
        assert!(types_compatible("INT", "int32"));
        assert!(types_compatible("VARCHAR", "text"));
        assert!(types_compatible("BIGINT", "int64"));
    }

    #[test]
    fn test_types_compatible_with_promotion() {
        assert!(types_compatible("int32", "int64"));
        assert!(types_compatible("float32", "float64"));
        assert!(types_compatible("timestamp", "timestamptz"));
    }

    #[test]
    fn test_types_compatible_narrowing_rejected() {
        assert!(!types_compatible("int64", "int32"));
        assert!(!types_compatible("float64", "float32"));
    }

    #[test]
    fn test_types_compatible_incompatible() {
        assert!(!types_compatible("int32", "varchar"));
        assert!(!types_compatible("varchar", "int32"));
        assert!(!types_compatible("boolean", "float64"));
    }

    #[test]
    fn test_types_compatible_unknown() {
        assert!(!types_compatible("foobar", "int32"));
        assert!(!types_compatible("int32", "foobar"));
        assert!(!types_compatible("foobar", "bazqux"));
    }
}