delta_kernel 0.30.0

//! Partition key and type validation.
//!
//! Validates that the connector provided the correct partition column names and value types
//! before serialization.
//!
//! ```text
//! Step 1 (THIS MODULE):  {"YEAR": Integer(2024)} --> {"Year": Integer(2024)}
//!                         (case-normalize keys, check types against schema)
//! Step 2 (serialization): Integer(2024) --> "2024"
//! Step 3 (path encoding): "Year=2024/"
//! ```
//!
//! See the encoding tables in the [`super`] module for the full set of partition-eligible
//! types and their expected serializations.
//!
//! The entry points are [`validate_partition_values`] for logical keys and
//! [`validate_physical_partition_values`] for physical keys. Both combine:
//!
//! - [`validate_keys`]: checks key completeness using the selected matching rule and re-keys values
//!   to logical schema names.
//! - [`validate_types`]: checks that each `Scalar`'s type matches the partition column's schema
//!   type, and that non-null partition columns are never assigned a value that would serialize to a
//!   null partition value.
use std::collections::HashMap;

use crate::expressions::Scalar;
use crate::partition::serialization::would_serialize_to_null;
use crate::schema::{DataType, StructType};
use crate::table_features::ColumnMappingMode;
use crate::{KernelError, KernelResult};

/// Validates and normalizes partition keys and value types against the table schema.
/// Returns the map re-keyed to schema case.
///
/// This is the primary entry point for partition validation. It combines [`validate_keys`]
/// (key completeness and case normalization) with [`validate_types`] (scalar type checking).
///
/// # Parameters
/// - `logical_partition_columns`: logical partition column names from kernel's table metadata.
/// - `logical_schema`: the table's logical schema.
/// - `logical_partition_values`: connector-provided map from logical column names (any case) to
///   typed values.
pub(crate) fn validate_partition_values(
    logical_partition_columns: &[String],
    logical_schema: &StructType,
    logical_partition_values: HashMap<String, Scalar>,
) -> KernelResult<HashMap<String, Scalar>> {
    let expected_keys: Vec<_> = logical_partition_columns
        .iter()
        .map(|name| ExpectedPartitionKey {
            input_name: name,
            logical_name: name,
        })
        .collect();
    validate_and_rekey_partition_values(
        &expected_keys,
        logical_schema,
        logical_partition_values,
        PartitionKeyMatching::LogicalCaseInsensitive,
    )
}

/// Validates physical partition keys exactly and returns values keyed by logical schema names.
/// Physical names are opaque and may differ only by case, so they must not be case-normalized.
pub(crate) fn validate_physical_partition_values(
    logical_partition_columns: &[String],
    logical_schema: &StructType,
    column_mapping_mode: ColumnMappingMode,
    physical_partition_values: HashMap<String, Scalar>,
) -> KernelResult<HashMap<String, Scalar>> {
    let mut expected_keys = Vec::with_capacity(logical_partition_columns.len());
    for logical_name in logical_partition_columns {
        let field = logical_schema.field(logical_name).ok_or_else(|| {
            KernelError::invalid_partition_values(format!(
                "partition column '{logical_name}' not found in table schema"
            ))
        })?;
        let physical_name = field.physical_name(column_mapping_mode);
        expected_keys.push(ExpectedPartitionKey {
            input_name: physical_name,
            logical_name,
        });
    }
    validate_and_rekey_partition_values(
        &expected_keys,
        logical_schema,
        physical_partition_values,
        PartitionKeyMatching::PhysicalExact,
    )
}

#[derive(Clone, Copy)]
enum PartitionKeyMatching {
    LogicalCaseInsensitive,
    PhysicalExact,
}

impl PartitionKeyMatching {
    fn lookup_key(self, name: &str) -> String {
        match self {
            Self::LogicalCaseInsensitive => name.to_lowercase(),
            Self::PhysicalExact => name.to_string(),
        }
    }

    fn namespace(self) -> &'static str {
        match self {
            Self::LogicalCaseInsensitive => "logical",
            Self::PhysicalExact => "physical",
        }
    }
}

struct ExpectedPartitionKey<'a> {
    input_name: &'a str,
    logical_name: &'a str,
}

fn validate_and_rekey_partition_values(
    expected_keys: &[ExpectedPartitionKey<'_>],
    logical_schema: &StructType,
    partition_values: HashMap<String, Scalar>,
    matching: PartitionKeyMatching,
) -> KernelResult<HashMap<String, Scalar>> {
    let normalized = validate_keys(expected_keys, partition_values, matching)?;
    validate_types(logical_schema, &normalized)?;
    Ok(normalized)
}

/// Validates that a connector-provided partition value map contains exactly the expected
/// partition columns and returns the map re-keyed to logical schema names.
///
/// # Parameters
/// - `expected_keys`: accepted input names and their corresponding logical schema names.
/// - `partition_values`: connector-provided map from partition column names to typed values.
/// - `matching`: matching rule for connector-provided names.
///
/// # Errors
/// - A partition column is missing from the map
/// - An extra key is present that is not a partition column
/// - Two provided keys resolve to the same logical partition column
/// - Two expected input names collide under the selected matching rule
fn validate_keys(
    expected_keys: &[ExpectedPartitionKey<'_>],
    partition_values: HashMap<String, Scalar>,
    matching: PartitionKeyMatching,
) -> KernelResult<HashMap<String, Scalar>> {
    let mut schema_lookup = HashMap::with_capacity(expected_keys.len());
    for expected in expected_keys {
        if schema_lookup
            .insert(matching.lookup_key(expected.input_name), expected)
            .is_some()
        {
            return Err(KernelError::invalid_partition_values(format!(
                "duplicate {} partition column '{}' in table schema",
                matching.namespace(),
                expected.input_name
            )));
        }
    }
    let expected_names = expected_keys
        .iter()
        .map(|expected| expected.input_name)
        .collect::<Vec<_>>()
        .join(", ");
    let provided_names = partition_values.keys().cloned().collect::<Vec<_>>();

    let mut normalized = HashMap::with_capacity(partition_values.len());
    for (key, value) in partition_values {
        let lookup_key = matching.lookup_key(&key);
        let expected = schema_lookup.get(&lookup_key).ok_or_else(|| {
            KernelError::invalid_partition_values(format!(
                "unknown partition column '{key}'. Expected one of: [{}]",
                expected_names
            ))
        })?;
        if normalized.contains_key(expected.logical_name) {
            return Err(KernelError::invalid_partition_values(format!(
                "duplicate partition column '{key}' (normalized to same key as a previously provided entry)"
            )));
        }
        normalized.insert(expected.logical_name.to_string(), value);
    }

    for expected in expected_keys {
        if !normalized.contains_key(expected.logical_name) {
            return Err(KernelError::invalid_partition_values(format!(
                "missing partition column '{}'. Provided: [{}]",
                expected.input_name,
                provided_names.join(", ")
            )));
        }
    }

    Ok(normalized)
}

/// Validates that each [`Scalar`] value's type is compatible with the corresponding
/// partition column's type in the table schema. Values that would serialize to a null
/// partition value skip the value type check, but a
/// null partition value is only legal when the schema field is nullable.
///
/// Nullability is enforced before any non-null type check, so the rejection treats null
/// scalars and their serialization-equivalent forms (empty strings, empty binary)
/// uniformly -- otherwise an empty `Scalar::String` for a `nullable: false` column would
/// pass validation, then collapse to JSON null at serialization, landing a null value in
/// a NOT NULL column.
///
/// # Parameters
/// - `logical_schema`: the table's logical schema.
/// - `logical_partition_values`: map from logical column names (schema case) to typed values. Keys
///   must use the exact logical names from the schema, as returned by [`validate_keys`]. This
///   function uses case-sensitive schema lookup.
///
/// # Errors
/// - A partition column is not found in the table schema
/// - A partition column's schema type is not a primitive (struct, array, map are rejected)
/// - A null-equivalent value (null scalar, empty string, or empty binary) is provided for a
///   non-nullable partition column
/// - A non-null value's type does not match the partition column's schema type
fn validate_types(
    logical_schema: &StructType,
    logical_partition_values: &HashMap<String, Scalar>,
) -> KernelResult<()> {
    for (col_name, value) in logical_partition_values {
        let field = logical_schema.field(col_name).ok_or_else(|| {
            KernelError::invalid_partition_values(format!(
                "partition column '{col_name}' not found in table schema"
            ))
        })?;
        let expected_type = field.data_type();
        if matches!(
            expected_type,
            DataType::Struct(_) | DataType::Array(_) | DataType::Map(_)
        ) {
            return Err(KernelError::invalid_partition_values(format!(
                "partition column '{col_name}' has non-primitive type {expected_type:?}. \
                 Partition columns must be primitive types."
            )));
        }
        if would_serialize_to_null(value) {
            if !field.nullable {
                return Err(KernelError::invalid_partition_values(format!(
                    "partition column '{col_name}' is not nullable but received a value that \
                     serializes to null (null scalar, empty string, or empty binary)"
                )));
            }
            continue;
        }
        let actual_type = value.data_type();
        if *expected_type != actual_type {
            return Err(KernelError::invalid_partition_values(format!(
                "partition column '{col_name}' has type {expected_type:?} but got \
                 value of type {actual_type:?}"
            )));
        }
    }
    Ok(())
}

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

    use super::*;
    use crate::expressions::Scalar;
    use crate::schema::{schema, ArrayType, DataType, MapType};

    fn assert_type_ok(data_type: DataType, value: Scalar) {
        let schema = schema! { not_null "p": (data_type) };
        let values = HashMap::from([("p".to_string(), value)]);
        validate_types(&schema, &values).unwrap();
    }

    fn assert_type_err(data_type: DataType, value: Scalar) -> String {
        let schema = schema! { not_null "p": (data_type) };
        let values = HashMap::from([("p".to_string(), value)]);
        validate_types(&schema, &values).unwrap_err().to_string()
    }

    /// Like [`assert_type_ok`] but uses a `nullable` schema field. Used by null-value cases,
    /// since null scalars are only valid for nullable partition columns.
    fn assert_type_ok_nullable(data_type: DataType, value: Scalar) {
        let schema = schema! { nullable "p": (data_type) };
        let values = HashMap::from([("p".to_string(), value)]);
        validate_types(&schema, &values).unwrap();
    }

    // ============================================================================
    // validate_keys
    // ============================================================================

    fn validate_logical_keys(
        logical_partition_columns: &[String],
        logical_partition_values: HashMap<String, Scalar>,
    ) -> KernelResult<HashMap<String, Scalar>> {
        let expected_keys = logical_partition_columns
            .iter()
            .map(|name| ExpectedPartitionKey {
                input_name: name,
                logical_name: name,
            })
            .collect::<Vec<_>>();
        validate_keys(
            &expected_keys,
            logical_partition_values,
            PartitionKeyMatching::LogicalCaseInsensitive,
        )
    }

    #[test]
    fn test_validate_partition_keys_matching_keys_returns_ok() {
        let cols = vec!["year".to_string(), "region".to_string()];
        let values = HashMap::from([
            ("year".to_string(), Scalar::Integer(2024)),
            ("region".to_string(), Scalar::String("US".into())),
        ]);
        let result = validate_logical_keys(&cols, values).unwrap();
        assert_eq!(result.len(), 2);
        assert_eq!(result.get("year"), Some(&Scalar::Integer(2024)));
        assert_eq!(result.get("region"), Some(&Scalar::String("US".into())));
    }

    #[test]
    fn test_validate_partition_keys_empty_columns_and_values_returns_ok() {
        let cols: Vec<String> = vec![];
        let values = HashMap::new();
        let result = validate_logical_keys(&cols, values).unwrap();
        assert!(result.is_empty());
    }

    #[test]
    fn test_validate_partition_keys_missing_key_returns_error() {
        let cols = vec!["year".to_string(), "region".to_string()];
        let values = HashMap::from([("year".to_string(), Scalar::Integer(2024))]);
        let result = validate_logical_keys(&cols, values);
        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(err.contains("missing partition column 'region'"), "{err}");
    }

    #[test]
    fn test_validate_partition_keys_extra_key_returns_error() {
        let cols = vec!["year".to_string()];
        let values = HashMap::from([
            ("year".to_string(), Scalar::Integer(2024)),
            ("region".to_string(), Scalar::String("US".into())),
        ]);
        let result = validate_logical_keys(&cols, values);
        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(err.contains("unknown partition column 'region'"), "{err}");
    }

    #[test]
    fn test_validate_partition_keys_case_normalizes_to_schema_case() {
        let cols = vec!["Year".to_string()];
        let values = HashMap::from([("YEAR".to_string(), Scalar::Integer(2024))]);
        let result = validate_logical_keys(&cols, values).unwrap();
        assert!(result.contains_key("Year"));
        assert!(!result.contains_key("YEAR"));
    }

    #[test]
    fn test_validate_partition_keys_duplicate_after_normalization_returns_error() {
        let cols = vec!["col".to_string()];
        let values = HashMap::from([
            ("COL".to_string(), Scalar::Integer(1)),
            ("col".to_string(), Scalar::Integer(2)),
        ]);
        let result = validate_logical_keys(&cols, values);
        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(err.contains("duplicate"), "{err}");
    }

    // ============================================================================
    // validate_types
    // ============================================================================
    //
    // Covers every partition-eligible type from the encoding tables in `partition/mod.rs`.
    // Each row tests that the correct Scalar variant passes type validation against the
    // corresponding schema DataType.

    /// Every non-null row from the "Encoding by type" table in `partition/mod.rs`.
    /// Row numbers reference that table. Validation only checks types (not values), but
    /// covering every row mirrors the reference table 1:1 for completeness.
    #[rstest]
    // INT (rows 1-4)
    #[case(DataType::INTEGER, Scalar::Integer(0))] // row 1
    #[case(DataType::INTEGER, Scalar::Integer(-1))] // row 2
    #[case(DataType::INTEGER, Scalar::Integer(i32::MAX))] // row 3
    #[case(DataType::INTEGER, Scalar::Integer(i32::MIN))] // row 4
    // BIGINT (rows 6-7)
    #[case(DataType::LONG, Scalar::Long(i64::MAX))] // row 6
    #[case(DataType::LONG, Scalar::Long(i64::MIN))] // row 7
    // TINYINT (rows 9-10)
    #[case(DataType::BYTE, Scalar::Byte(127))] // row 9
    #[case(DataType::BYTE, Scalar::Byte(-128))] // row 10
    // SMALLINT (rows 12-13)
    #[case(DataType::SHORT, Scalar::Short(32767))] // row 12
    #[case(DataType::SHORT, Scalar::Short(-32768))] // row 13
    // DOUBLE (rows 15-21)
    #[case(DataType::DOUBLE, Scalar::Double(0.0))] // row 15
    #[case(DataType::DOUBLE, Scalar::Double(-0.0))] // row 16
    #[case(DataType::DOUBLE, Scalar::Double(f64::MAX))] // row 17
    #[case(DataType::DOUBLE, Scalar::Double(f64::MIN_POSITIVE))] // row 18
    #[case(DataType::DOUBLE, Scalar::Double(f64::NAN))] // row 19
    #[case(DataType::DOUBLE, Scalar::Double(f64::INFINITY))] // row 20
    #[case(DataType::DOUBLE, Scalar::Double(f64::NEG_INFINITY))] // row 21
    // FLOAT (rows 23-26)
    #[case(DataType::FLOAT, Scalar::Float(0.0))] // row 23
    #[case(DataType::FLOAT, Scalar::Float(f32::NAN))] // row 24
    #[case(DataType::FLOAT, Scalar::Float(f32::INFINITY))] // row 25
    #[case(DataType::FLOAT, Scalar::Float(f32::NEG_INFINITY))] // row 26
    // BOOLEAN (rows 28-29)
    #[case(DataType::BOOLEAN, Scalar::Boolean(true))] // row 28
    #[case(DataType::BOOLEAN, Scalar::Boolean(false))] // row 29
    // DECIMAL(38,18) (rows 31-33)
    #[case(DataType::decimal(38, 18).unwrap(), Scalar::decimal(0, 38, 18).unwrap())] // row 31
    #[case(DataType::decimal(38, 18).unwrap(), Scalar::decimal(1_230_000_000_000_000_000i128, 38, 18).unwrap())] // row 32
    #[case(DataType::decimal(38, 18).unwrap(), Scalar::decimal(-1_230_000_000_000_000_000i128, 38, 18).unwrap())] // row 33
    // DATE (rows 35-38)
    #[case(DataType::DATE, Scalar::Date(19723))] // row 35: 2024-01-01
    #[case(DataType::DATE, Scalar::Date(0))] // row 36: 1970-01-01 (epoch)
    #[case(DataType::DATE, Scalar::Date(-719_162))] // row 37: 0001-01-01
    #[case(DataType::DATE, Scalar::Date(2_932_896))] // row 38: 9999-12-31
    // TIMESTAMP (rows 40-42)
    #[case(DataType::TIMESTAMP, Scalar::Timestamp(1_718_451_045_000_000))] // row 40
    #[case(DataType::TIMESTAMP, Scalar::Timestamp(0))] // row 41
    #[case(DataType::TIMESTAMP, Scalar::Timestamp(1_718_499_599_999_999))] // row 42
    // TIMESTAMP_NTZ (rows 44-45)
    #[case(DataType::TIMESTAMP_NTZ, Scalar::TimestampNtz(1_718_451_045_000_000))] // row 44
    #[case(DataType::TIMESTAMP_NTZ, Scalar::TimestampNtz(0))] // row 45
    fn test_validate_types_encoding_table_rows_return_ok(
        #[case] data_type: DataType,
        #[case] value: Scalar,
    ) {
        assert_type_ok(data_type, value);
    }

    /// Every non-null row from the "String and binary encoding" table in
    /// `partition/mod.rs`. Row numbers reference that table.
    #[rstest]
    // STRING rows. Rows 49 (empty binary) and 65 (empty string) are null-equivalent under
    // the protocol and are covered by the null-into-nullable / null-into-non-nullable tests
    // further down -- they are intentionally absent from this "type-match passes" group.
    #[case(DataType::STRING, Scalar::String("\x00".into()))] // row 47
    #[case(DataType::STRING, Scalar::String("before\x00after".into()))] // row 48
    #[case(DataType::STRING, Scalar::String("a{b".into()))] // row 54
    #[case(DataType::STRING, Scalar::String("a}b".into()))] // row 55
    #[case(DataType::STRING, Scalar::String("hello world".into()))] // row 56
    #[case(DataType::STRING, Scalar::String("M\u{00FC}nchen".into()))] // row 57
    #[case(DataType::STRING, Scalar::String("\u{65E5}\u{672C}\u{8A9E}".into()))] // row 58
    #[case(DataType::STRING, Scalar::String("\u{1F3B5}\u{1F3B6}".into()))] // row 59
    #[case(DataType::STRING, Scalar::String("a<b>c|d".into()))] // row 60
    #[case(DataType::STRING, Scalar::String("a@b!c(d)".into()))] // row 61
    #[case(DataType::STRING, Scalar::String("a&b+c$d;e,f".into()))] // row 62
    #[case(DataType::STRING, Scalar::String("Serbia/srb%".into()))] // row 63
    #[case(DataType::STRING, Scalar::String("100%25".into()))] // row 64
    #[case(DataType::STRING, Scalar::String(" ".into()))] // row 67
    #[case(DataType::STRING, Scalar::String("  ".into()))] // row 68
    // BINARY rows
    #[case(DataType::BINARY, Scalar::Binary(vec![0xDE, 0xAD, 0xBE, 0xEF]))] // row 50
    #[case(DataType::BINARY, Scalar::Binary(vec![0x48, 0x45, 0x4C, 0x4C, 0x4F]))] // row 51
    #[case(DataType::BINARY, Scalar::Binary(vec![0x00, 0xFF]))] // row 52
    #[case(DataType::BINARY, Scalar::Binary(vec![0x2F, 0x3D, 0x25]))] // row 53
    fn test_validate_types_string_binary_table_rows_return_ok(
        #[case] data_type: DataType,
        #[case] value: Scalar,
    ) {
        assert_type_ok(data_type, value);
    }

    /// Null-equivalent rows from the encoding table. Under the Delta protocol, three Scalar
    /// shapes serialize to JSON null in `partitionValues`: `Scalar::Null`, empty `Scalar::String`,
    /// and empty `Scalar::Binary`. All three are valid for every partition column type when the
    /// schema field is nullable, regardless of the `Scalar::Null` inner type.
    #[rstest]
    #[case(DataType::INTEGER, Scalar::Null(DataType::INTEGER))] // row 5
    #[case(DataType::LONG, Scalar::Null(DataType::LONG))] // row 8
    #[case(DataType::BYTE, Scalar::Null(DataType::BYTE))] // row 11
    #[case(DataType::SHORT, Scalar::Null(DataType::SHORT))] // row 14
    #[case(DataType::DOUBLE, Scalar::Null(DataType::DOUBLE))] // row 22
    #[case(DataType::FLOAT, Scalar::Null(DataType::FLOAT))] // row 27
    #[case(DataType::BOOLEAN, Scalar::Null(DataType::BOOLEAN))] // row 30
    #[case(DataType::decimal(38, 18).unwrap(), Scalar::Null(DataType::decimal(38, 18).unwrap()))] // row 34
    #[case(DataType::DATE, Scalar::Null(DataType::DATE))] // row 39
    #[case(DataType::TIMESTAMP, Scalar::Null(DataType::TIMESTAMP))] // row 43
    #[case(DataType::TIMESTAMP_NTZ, Scalar::Null(DataType::TIMESTAMP_NTZ))] // row 46
    #[case(DataType::STRING, Scalar::Null(DataType::STRING))] // row 66
    #[case(DataType::BINARY, Scalar::Null(DataType::BINARY))] // (binary NULL)
    #[case(DataType::STRING, Scalar::String(String::new()))] // row 65: empty string
    #[case(DataType::BINARY, Scalar::Binary(Vec::new()))] // row 49: empty binary
    fn test_validate_types_null_into_nullable_returns_ok(
        #[case] data_type: DataType,
        #[case] value: Scalar,
    ) {
        assert_type_ok_nullable(data_type, value);
    }

    /// Null skips the value type check even when the `Scalar::Null` inner type does not match
    /// the schema column type. This is intentional: null carries no concrete type to compare
    /// against.
    #[test]
    fn test_validate_types_null_with_mismatched_inner_type_returns_ok() {
        assert_type_ok_nullable(DataType::INTEGER, Scalar::Null(DataType::STRING));
    }

    /// A null-equivalent partition value (null scalar, empty string, or empty binary) for
    /// a non-nullable partition column is rejected
    #[rstest]
    #[case(DataType::INTEGER, Scalar::Null(DataType::INTEGER))]
    #[case(DataType::LONG, Scalar::Null(DataType::LONG))]
    #[case(DataType::STRING, Scalar::Null(DataType::STRING))]
    #[case(DataType::BINARY, Scalar::Null(DataType::BINARY))]
    #[case(DataType::BOOLEAN, Scalar::Null(DataType::BOOLEAN))]
    #[case(DataType::DATE, Scalar::Null(DataType::DATE))]
    #[case(DataType::TIMESTAMP, Scalar::Null(DataType::TIMESTAMP))]
    #[case(DataType::TIMESTAMP_NTZ, Scalar::Null(DataType::TIMESTAMP_NTZ))]
    #[case(DataType::decimal(10, 2).unwrap(), Scalar::Null(DataType::decimal(10, 2).unwrap()))]
    // Null with mismatched inner type is also rejected the schema's nullability is the
    // only thing checked for null scalars.
    #[case(DataType::INTEGER, Scalar::Null(DataType::STRING))]
    // Empty string and empty binary serialize to JSON null per the Delta protocol; they
    // must be rejected for NOT NULL partition columns alongside `Scalar::Null` so a
    // serialization-equivalent value cannot bypass the nullability check.
    #[case(DataType::STRING, Scalar::String(String::new()))]
    #[case(DataType::BINARY, Scalar::Binary(Vec::new()))]
    fn test_validate_types_null_into_non_nullable_returns_error(
        #[case] data_type: DataType,
        #[case] value: Scalar,
    ) {
        let err = assert_type_err(data_type, value);
        assert!(err.contains("not nullable"), "{err}");
        assert!(err.contains("'p'"), "{err}");
    }

    /// Type mismatch: every non-null Scalar variant against the wrong schema type.
    #[rstest]
    #[case(DataType::STRING, Scalar::Integer(1))]
    #[case(DataType::INTEGER, Scalar::String("x".into()))]
    #[case(DataType::INTEGER, Scalar::Long(1))]
    #[case(DataType::LONG, Scalar::Integer(1))]
    #[case(DataType::DOUBLE, Scalar::Float(1.0))]
    #[case(DataType::FLOAT, Scalar::Double(1.0))]
    #[case(DataType::DATE, Scalar::Timestamp(0))]
    #[case(DataType::TIMESTAMP, Scalar::TimestampNtz(0))]
    #[case(DataType::STRING, Scalar::Binary(vec![0x41]))]
    #[case(DataType::BINARY, Scalar::String("A".into()))]
    #[case(DataType::BOOLEAN, Scalar::Integer(1))]
    fn test_validate_types_mismatch_returns_error(
        #[case] data_type: DataType,
        #[case] value: Scalar,
    ) {
        let err = assert_type_err(data_type, value);
        assert!(err.contains("p"), "{err}");
    }

    /// Complex types (struct, array, map) are rejected as partition column types.
    #[rstest]
    #[case(
        DataType::from(schema! { not_null "x": INTEGER }),
        Scalar::Null(DataType::STRING)
    )]
    #[case(
        DataType::from(ArrayType::new(DataType::INTEGER, false)),
        Scalar::Null(DataType::STRING)
    )]
    #[case(
        DataType::from(MapType::new(DataType::STRING, DataType::INTEGER, false)),
        Scalar::Null(DataType::STRING)
    )]
    fn test_validate_types_complex_type_returns_error(
        #[case] data_type: DataType,
        #[case] value: Scalar,
    ) {
        let err = assert_type_err(data_type, value);
        assert!(err.contains("non-primitive type"), "{err}");
    }

    #[test]
    fn test_validate_types_column_not_in_schema_returns_error() {
        let schema = schema! { not_null "year": INTEGER };
        let values = HashMap::from([("nonexistent".to_string(), Scalar::Integer(42))]);
        let result = validate_types(&schema, &values);
        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(err.contains("not found in table schema"), "{err}");
    }

    /// validate_partition_values normalizes keys to schema case and type-checks in one call.
    #[test]
    fn test_validate_partition_values_with_case_mismatch_succeeds() {
        let partition_cols = vec!["Year".to_string(), "Region".to_string()];
        let schema = schema! {
            not_null "id": INTEGER,
            not_null "Year": INTEGER,
            nullable "Region": STRING,
        };
        let values = HashMap::from([
            ("YEAR".to_string(), Scalar::Integer(2024)),
            ("region".to_string(), Scalar::String("US".into())),
        ]);
        let result = validate_partition_values(&partition_cols, &schema, values).unwrap();
        assert_eq!(result.get("Year"), Some(&Scalar::Integer(2024)));
        assert_eq!(result.get("Region"), Some(&Scalar::String("US".into())));
    }
}