delta_funnel/table_formats/

delta.rs

//! Delta table-format source loading.

use std::collections::{BTreeMap, HashMap, HashSet};

use crate::{DeltaFunnelError, error::DeltaSourceSchemaSnafu, observability};
use delta_kernel::object_store::aws::AmazonS3ConfigKey;

mod deletion_vector;
mod kernel;
mod protocol;
mod read;
mod snapshot;
#[cfg(test)]
pub(crate) mod test_support;
mod uri;

use super::validate_table_source_names;
#[allow(unused_imports)]
pub(crate) use deletion_vector::{
    KernelDeletionVectorReadRequest, KernelDeletionVectorReader, KernelDeletionVectorReaderConfig,
    ProviderDeletionVectorSelection, ProviderDeletionVectorSelectionContext,
};
#[cfg(test)]
pub(crate) use kernel::KernelStructType;
use kernel::{ArrowSchemaRef, Version, snapshot_arrow_schema};
pub(crate) use kernel::{
    ColumnName as KernelColumnName, DecimalData as KernelDecimalData,
    DeltaKernelPartitionScalarAdapterError, DeltaKernelPredicate, KernelColumnMetadataKey,
    KernelDataType, KernelMetadataColumnSpec, KernelMetadataValue, KernelPrimitiveType,
    KernelSchemaRef, KernelStructField, Scalar as KernelScalar,
    arrow_partition_type_to_kernel_primitive, datafusion_expr_to_kernel_predicate,
    kernel_partition_scalar_to_datafusion_scalar,
};
pub(crate) use protocol::preflight_delta_protocol_with_tracing;
pub use protocol::{ProtocolPreflight, preflight_delta_protocol, preflight_delta_sources};
pub(crate) use read::{
    KernelDataFilePredicateEvalRequest, KernelDataFileReadRequest, KernelDataFileReader,
    KernelDataFileReaderConfig, KernelDataFileTransformRequest, KernelScanReadSchema,
};
use snapshot::{LoadedDeltaTableSnapshot, load_delta_table_snapshot, s3_auth_mode_hint_for_source};

/// Metadata-only expansion of one kernel scan.
///
/// This is the direct output of `delta_kernel` scan metadata expansion. It is
/// not the provider execution task model; provider planning converts
/// these file metadata records into Delta-aware file tasks and grouped scan
/// partitions.
#[allow(dead_code)]
pub(crate) struct KernelScanMetadataExpansion {
    /// File-level metadata rows yielded by Delta Kernel scan metadata expansion.
    pub(crate) files: Vec<KernelScanFileMetadata>,
    /// Whether the kernel scan metadata iterator was consumed to completion.
    ///
    /// This records that the adapter did not stop after a partial metadata
    /// batch. Later planning layers can rely on `files` representing the full
    /// kernel-selected file set for this scan.
    pub(crate) scan_metadata_exhausted: bool,
}

/// Metadata for one file selected by kernel scan planning.
#[allow(dead_code)]
pub(crate) struct KernelScanFileMetadata {
    /// Delta add-action path for the selected data file.
    ///
    /// This is the table-relative path that Delta Kernel exposes in
    /// `ScanFile`, not a resolved object-store URI.
    pub(crate) path: String,
    /// File size in bytes from the Delta add action.
    pub(crate) size: i64,
    /// Last modification timestamp from the Delta add action.
    ///
    /// Delta log metadata stores this value in milliseconds since the Unix
    /// epoch.
    pub(crate) modification_time: i64,
    /// Parsed file statistics from Delta Kernel when stats are available.
    pub(crate) stats: Option<KernelScanFileStats>,
    /// Deletion-vector metadata selected with this file, if one exists.
    ///
    /// The handle intentionally preserves kernel-owned metadata only. It must
    /// not read or materialize the deletion-vector payload during metadata
    /// expansion.
    pub(crate) deletion_vector: KernelScanDeletionVectorMetadata,
    /// Optional physical-to-logical transform required for this file.
    ///
    /// Delta Kernel uses this for features where physical Parquet columns do
    /// not directly match the logical scan schema. The expression is preserved
    /// for provider execution and is not evaluated here.
    pub(crate) physical_to_logical_transform: KernelPhysicalToLogicalTransform,
    /// Partition values from the Delta add action keyed by partition column.
    ///
    /// Values are the raw string metadata values exposed by Delta Kernel.
    pub(crate) partition_values: HashMap<String, String>,
}

/// Parsed statistics for one kernel-selected scan file.
#[allow(dead_code)]
pub(crate) struct KernelScanFileStats {
    /// Number of records parsed from the Delta add-action stats JSON.
    pub(crate) num_records: u64,
}

/// Opaque metadata for a deletion vector selected by kernel scan planning.
///
/// This preserves the kernel-owned deletion-vector descriptor for provider
/// execution without loading the deletion-vector payload.
#[allow(dead_code)]
#[derive(Clone)]
pub(crate) enum KernelScanDeletionVectorMetadata {
    /// The selected file has no deletion vector.
    NotPresent,
    /// The selected file has deletion-vector metadata preserved for execution.
    Present(KernelScanDeletionVectorHandle),
}

/// Opaque handle for kernel-owned deletion-vector metadata.
///
/// The wrapped `DvInfo` is kept private so metadata expansion cannot
/// accidentally load the deletion-vector payload. Execution code must load
/// payloads through the private deletion-vector reader boundary.
#[allow(dead_code)]
#[derive(Clone)]
pub(crate) struct KernelScanDeletionVectorHandle {
    /// Kernel deletion-vector metadata for the selected file.
    dv_info: kernel::DvInfo,
}

impl KernelScanDeletionVectorHandle {
    #[cfg(test)]
    pub(crate) fn from_test_dv_info(dv_info: kernel::DvInfo) -> Self {
        Self { dv_info }
    }
}

impl KernelScanDeletionVectorMetadata {
    #[must_use]
    pub(crate) fn is_present(&self) -> bool {
        matches!(self, Self::Present(_))
    }

    #[cfg(test)]
    #[must_use]
    pub(crate) fn test_present_from_descriptor(
        descriptor: kernel::DeletionVectorDescriptor,
    ) -> Self {
        Self::Present(KernelScanDeletionVectorHandle {
            dv_info: descriptor.into(),
        })
    }
}

/// Opaque physical-to-logical transform metadata selected for one scan file.
///
/// The transform is applied only after physical Parquet data has been read,
/// which belongs to provider execution rather than metadata expansion.
#[allow(dead_code)]
#[derive(Clone)]
pub(crate) enum KernelPhysicalToLogicalTransform {
    /// The physical Parquet file already matches the logical scan shape.
    NotRequired,
    /// A kernel expression must be applied after physical data is read.
    Required(KernelPhysicalToLogicalTransformHandle),
}

/// Opaque handle for a kernel physical-to-logical transform expression.
///
/// The expression is metadata for provider execution. It is not evaluated while
/// expanding scan metadata.
#[allow(dead_code)]
#[derive(Clone)]
pub(crate) struct KernelPhysicalToLogicalTransformHandle {
    /// Kernel expression that maps physical file data into logical scan data.
    transform: kernel::ExpressionRef,
}

/// Storage options for one Delta source.
pub type DeltaStorageOptions = BTreeMap<String, String>;

/// Caller-provided configuration for one named Delta source.
pub struct DeltaSourceConfig {
    /// DataFusion table name that will identify this source.
    pub name: String,
    /// Caller-provided Delta table location.
    pub table_uri: String,
    /// Optional fixed Delta table version.
    pub version: Option<Version>,
    /// Source-local options forwarded to Delta Kernel object-store construction.
    pub storage_options: DeltaStorageOptions,
}

impl DeltaSourceConfig {
    /// Builds a Delta source config with no fixed version or storage options.
    #[must_use]
    pub fn new(name: impl Into<String>, table_uri: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            table_uri: table_uri.into(),
            version: None,
            storage_options: DeltaStorageOptions::default(),
        }
    }

    /// Sets an optional fixed Delta table version.
    #[must_use]
    pub fn with_version(mut self, version: Option<Version>) -> Self {
        self.version = version;
        self
    }

    /// Sets source-local storage options.
    #[must_use]
    pub fn with_storage_options(mut self, storage_options: DeltaStorageOptions) -> Self {
        self.storage_options = storage_options;
        self
    }
}

/// Loaded named Delta source state.
pub struct PlannedDeltaSource {
    name: String,
    requested_table_uri: String,
    storage_options: DeltaStorageOptions,
    snapshot: LoadedDeltaTableSnapshot,
}

/// Kernel-backed scan state for one projected Delta scan.
#[allow(dead_code)]
pub(crate) struct ProjectedDeltaScan {
    scan: kernel::Scan,
    kernel_schema: kernel::KernelSchemaRef,
}

impl ProjectedDeltaScan {
    /// Returns the projected kernel schema selected for this scan.
    #[allow(dead_code)]
    #[must_use]
    pub(crate) fn kernel_schema(&self) -> &kernel::KernelSchemaRef {
        &self.kernel_schema
    }

    /// Returns the kernel scan handle that metadata planning consumes.
    #[allow(dead_code)]
    #[must_use]
    pub(crate) fn kernel_scan(&self) -> &kernel::Scan {
        &self.scan
    }

    /// Returns the kernel scan schema state needed by data-file reads.
    #[allow(dead_code)]
    #[must_use]
    pub(crate) fn read_schema(&self) -> KernelScanReadSchema {
        KernelScanReadSchema::new(
            self.scan.physical_schema().clone(),
            self.scan.logical_schema().clone(),
            self.scan.physical_predicate().clone(),
        )
    }

    /// Expands this kernel scan into metadata-only scan file records.
    ///
    /// This exhausts kernel scan metadata for one provider scan and does not
    /// read Parquet data, load deletion-vector payloads, or apply physical data
    /// transforms.
    #[allow(dead_code)]
    pub(crate) fn expand_kernel_scan_metadata(
        &self,
        table_uri: &str,
        storage_options: &DeltaStorageOptions,
    ) -> Result<KernelScanMetadataExpansion, delta_kernel::Error> {
        fn collect_scan_file(files: &mut Vec<KernelScanFileMetadata>, file: kernel::ScanFile) {
            files.push(KernelScanFileMetadata::from_kernel_scan_file(file));
        }

        let table_url = kernel::try_parse_uri(table_uri)?;
        let store = kernel::store_from_url_opts(
            &table_url,
            storage_options
                .iter()
                .map(|(key, value)| (key.as_str(), value.as_str())),
        )?;
        let engine = kernel::DefaultEngineBuilder::new(store).build();
        let mut files = Vec::new();

        for scan_metadata in self.scan.scan_metadata(&engine)? {
            files = scan_metadata?.visit_scan_files(files, collect_scan_file)?;
        }

        Ok(KernelScanMetadataExpansion {
            files,
            scan_metadata_exhausted: true,
        })
    }

    #[cfg(test)]
    /// Returns scan file paths after kernel scan planning.
    pub(crate) fn scan_file_paths(
        &self,
        table_uri: &str,
        storage_options: &DeltaStorageOptions,
    ) -> Result<Vec<String>, Box<dyn std::error::Error>> {
        let mut paths = self
            .expand_kernel_scan_metadata(table_uri, storage_options)?
            .files
            .into_iter()
            .map(|file| file.path)
            .collect::<Vec<_>>();
        paths.sort();
        Ok(paths)
    }
}

impl KernelScanFileMetadata {
    fn from_kernel_scan_file(file: kernel::ScanFile) -> Self {
        let kernel::ScanFile {
            path,
            size,
            modification_time,
            stats,
            dv_info,
            transform,
            partition_values,
        } = file;

        Self {
            path,
            size,
            modification_time,
            stats: stats.map(|stats| KernelScanFileStats {
                num_records: stats.num_records,
            }),
            deletion_vector: KernelScanDeletionVectorMetadata::from_dv_info(dv_info),
            physical_to_logical_transform: KernelPhysicalToLogicalTransform::from_transform(
                transform,
            ),
            partition_values,
        }
    }
}

impl KernelScanDeletionVectorMetadata {
    fn from_dv_info(dv_info: kernel::DvInfo) -> Self {
        if dv_info.has_vector() {
            Self::Present(KernelScanDeletionVectorHandle { dv_info })
        } else {
            Self::NotPresent
        }
    }
}

impl KernelPhysicalToLogicalTransform {
    #[cfg(test)]
    #[must_use]
    pub(crate) fn is_required(&self) -> bool {
        matches!(self, Self::Required(_))
    }

    #[cfg(test)]
    pub(crate) fn test_required_column_transform(column_name: &str) -> Self {
        let transform = std::sync::Arc::new(kernel::Expression::Column(kernel::ColumnName::new([
            column_name,
        ])));

        Self::Required(KernelPhysicalToLogicalTransformHandle { transform })
    }

    #[cfg(test)]
    pub(crate) fn test_required_orders_customer_name_replacement_transform(
        replacement: &str,
    ) -> Self {
        let transform = std::sync::Arc::new(kernel::Expression::struct_from([
            kernel::Expression::Column(kernel::ColumnName::new(["id"])),
            kernel::Expression::Literal(kernel::Scalar::String(replacement.to_owned())),
        ]));

        Self::Required(KernelPhysicalToLogicalTransformHandle { transform })
    }

    fn from_transform(transform: Option<kernel::ExpressionRef>) -> Self {
        match transform {
            Some(transform) => Self::Required(KernelPhysicalToLogicalTransformHandle { transform }),
            None => Self::NotRequired,
        }
    }
}

impl PlannedDeltaSource {
    /// DataFusion table name for this source.
    #[must_use]
    pub fn name(&self) -> &str {
        &self.name
    }

    /// Caller-provided Delta table location.
    #[must_use]
    pub fn requested_table_uri(&self) -> &str {
        &self.requested_table_uri
    }

    /// Effective source-local options forwarded to object-store construction.
    #[must_use]
    pub fn storage_options(&self) -> &DeltaStorageOptions {
        &self.storage_options
    }

    /// Normalized Delta table URI used for snapshot loading.
    #[must_use]
    pub fn table_uri(&self) -> &str {
        self.loaded_snapshot().table_uri()
    }

    /// Loaded Delta table version.
    #[must_use]
    pub fn version(&self) -> Version {
        self.loaded_snapshot().version()
    }

    pub(crate) fn loaded_snapshot(&self) -> &LoadedDeltaTableSnapshot {
        &self.snapshot
    }
}

pub(crate) fn delta_source_arrow_schema(
    source: &PlannedDeltaSource,
) -> Result<ArrowSchemaRef, DeltaFunnelError> {
    match snapshot_arrow_schema(source.loaded_snapshot().kernel_snapshot()) {
        Ok(schema) => Ok(schema),
        Err(error) => DeltaSourceSchemaSnafu {
            source_name: source.name().to_owned(),
            table_uri: source.table_uri().to_owned(),
            reason: error.to_string(),
        }
        .fail(),
    }
}

/// Builds kernel-backed scan state for a loaded Delta source projection.
#[allow(dead_code)]
pub(crate) fn build_projected_delta_scan(
    source: &PlannedDeltaSource,
    projected_column_names: Option<&[String]>,
) -> Result<ProjectedDeltaScan, delta_kernel::Error> {
    build_projected_predicated_delta_scan(source, projected_column_names, None)
}

/// Builds kernel-backed scan state for a loaded Delta source projection and predicate.
///
/// `projected_column_names` is the schema passed to delta_kernel's scan
/// builder, not necessarily the final DataFusion output schema. Callers may
/// include hidden predicate columns here when the final projection omits them.
#[allow(dead_code)]
pub(crate) fn build_projected_predicated_delta_scan(
    source: &PlannedDeltaSource,
    projected_column_names: Option<&[String]>,
    predicate: Option<DeltaKernelPredicate>,
) -> Result<ProjectedDeltaScan, delta_kernel::Error> {
    let (scan, kernel_schema) = kernel::build_projected_predicated_scan(
        source.loaded_snapshot().kernel_snapshot(),
        projected_column_names,
        predicate,
    )?;

    Ok(ProjectedDeltaScan {
        scan,
        kernel_schema,
    })
}

/// Builds kernel-backed scan state with parsed file stats exposed in scan metadata.
#[allow(dead_code)]
pub(crate) fn build_projected_predicated_stats_delta_scan(
    source: &PlannedDeltaSource,
    projected_column_names: Option<&[String]>,
    predicate: Option<DeltaKernelPredicate>,
) -> Result<ProjectedDeltaScan, delta_kernel::Error> {
    let (scan, kernel_schema) = kernel::build_projected_predicated_stats_scan(
        source.loaded_snapshot().kernel_snapshot(),
        projected_column_names,
        predicate,
    )?;

    Ok(ProjectedDeltaScan {
        scan,
        kernel_schema,
    })
}

/// Loads one named Delta source.
///
/// Name validation runs before URI normalization, engine construction, or
/// snapshot loading.
///
/// # Errors
///
/// Returns [`DeltaFunnelError::InvalidSourceName`] for an invalid source name,
/// [`DeltaFunnelError::InvalidSourceUri`] for an invalid table URI,
/// [`DeltaFunnelError::DeltaSourceEngine`] when engine construction fails, or
/// [`DeltaFunnelError::DeltaSnapshotLoad`] when the requested snapshot cannot
/// be loaded.
pub fn load_delta_source(
    config: DeltaSourceConfig,
) -> Result<PlannedDeltaSource, DeltaFunnelError> {
    validate_table_source_names([config.name.as_str()])?;

    load_delta_source_after_name_validation(config, false)
}

pub(crate) fn load_delta_source_with_tracing(
    config: DeltaSourceConfig,
) -> Result<PlannedDeltaSource, DeltaFunnelError> {
    validate_table_source_names([config.name.as_str()])?;

    load_delta_source_after_name_validation(config, true)
}

/// Loads configured Delta sources after validating all names.
///
/// Name validation and duplicate detection run before any URI normalization,
/// engine construction, or snapshot loading.
///
/// # Errors
///
/// Returns [`DeltaFunnelError::InvalidSourceName`] or
/// [`DeltaFunnelError::DuplicateSourceName`] before loading any snapshot when
/// source names are invalid or ambiguous. Otherwise returns the same URI,
/// engine, and snapshot-loading errors as [`load_delta_source`].
pub fn load_delta_sources<I>(configs: I) -> Result<Vec<PlannedDeltaSource>, DeltaFunnelError>
where
    I: IntoIterator<Item = DeltaSourceConfig>,
{
    let configs: Vec<_> = configs.into_iter().collect();

    validate_table_source_names(configs.iter().map(|config| config.name.as_str()))?;

    configs
        .into_iter()
        .map(|config| load_delta_source_after_name_validation(config, false))
        .collect()
}

fn load_delta_source_after_name_validation(
    mut config: DeltaSourceConfig,
    emit_tracing: bool,
) -> Result<PlannedDeltaSource, DeltaFunnelError> {
    config.storage_options = effective_storage_options_for_source_from_env(
        &config.table_uri,
        config.storage_options,
        std::env::vars(),
    );
    let source_name = config.name.clone();
    let s3_auth_mode_hint =
        s3_auth_mode_hint_for_source(&config.table_uri, &config.storage_options);
    if emit_tracing {
        observability::source_loading_started(
            &source_name,
            s3_auth_mode_hint.map(|hint| hint.as_str()),
        );
    }

    let DeltaSourceConfig {
        name,
        table_uri,
        version,
        storage_options,
    } = config;

    let result = load_delta_table_snapshot(&table_uri, version, &storage_options).map(|snapshot| {
        PlannedDeltaSource {
            name,
            requested_table_uri: table_uri,
            storage_options,
            snapshot,
        }
    });

    if emit_tracing {
        match &result {
            Ok(source) => observability::source_loading_completed(
                source.name(),
                source.version(),
                s3_auth_mode_hint.map(|hint| hint.as_str()),
            ),
            Err(error) => observability::source_loading_failed(
                &source_name,
                error,
                s3_auth_mode_hint.map(|hint| hint.as_str()),
            ),
        }
    }

    result
}

fn effective_storage_options_for_source_from_env<I>(
    table_uri: &str,
    storage_options: DeltaStorageOptions,
    env: I,
) -> DeltaStorageOptions
where
    I: IntoIterator<Item = (String, String)>,
{
    if s3_auth_mode_hint_for_source(table_uri, &storage_options).is_none() {
        return storage_options;
    }

    let caller_s3_keys = storage_options
        .keys()
        .filter_map(|key| s3_option_precedence_key(key))
        .collect::<HashSet<_>>();
    let mut effective = env
        .into_iter()
        .filter(|(key, _)| key.starts_with("AWS_"))
        .filter(|(key, _)| match s3_option_precedence_key(key) {
            Some(key) => !caller_s3_keys.contains(&key),
            None => true,
        })
        .collect::<DeltaStorageOptions>();
    effective.extend(storage_options);
    effective
}

fn s3_option_precedence_key(key: &str) -> Option<String> {
    let key = key.to_ascii_lowercase().parse::<AmazonS3ConfigKey>().ok()?;
    Some(match key {
        AmazonS3ConfigKey::DefaultRegion | AmazonS3ConfigKey::Region => "aws_region".to_owned(),
        _ => key.as_ref().to_owned(),
    })
}

#[cfg(test)]
mod tests {
    use std::collections::{BTreeMap, HashMap};
    use std::fs;
    use std::path::{Path, PathBuf};
    use std::sync::{Arc, Mutex};
    use std::time::{SystemTime, UNIX_EPOCH};

    use datafusion::common::ScalarValue;
    use datafusion::logical_expr::{Expr, col, lit};
    use delta_kernel::arrow::array::{Array, StructArray};
    use delta_kernel::arrow::record_batch::RecordBatch;
    use delta_kernel::object_store::{
        aws::{AmazonS3Builder, AmazonS3ConfigKey},
        memory::InMemory,
        path::Path as ObjectStorePath,
    };

    use super::kernel::{ColumnName, Expression, Predicate, Scalar};
    use super::{
        DeltaKernelPredicate, DeltaSourceConfig, DeltaStorageOptions, KernelDataFileReader,
        KernelDataFileReaderConfig, KernelDeletionVectorReader, KernelDeletionVectorReaderConfig,
        ProjectedDeltaScan, build_projected_delta_scan, build_projected_predicated_delta_scan,
        datafusion_expr_to_kernel_predicate, effective_storage_options_for_source_from_env,
        load_delta_source, load_delta_sources, load_delta_table_snapshot,
    };
    use crate::DeltaFunnelError;

    struct DeltaLogTable {
        path: PathBuf,
    }

    impl Drop for DeltaLogTable {
        fn drop(&mut self) {
            let _ = fs::remove_dir_all(&self.path);
        }
    }

    impl DeltaLogTable {
        fn new(name: &str) -> Result<Self, Box<dyn std::error::Error>> {
            Self::new_with_metadata_and_adds(name, METADATA_JSON, &[add_json("part-00001.parquet")])
        }

        fn new_with_metadata_and_adds(
            name: &str,
            metadata_json: &str,
            add_jsons: &[String],
        ) -> Result<Self, Box<dyn std::error::Error>> {
            Self::new_with_protocol_metadata_and_adds(name, PROTOCOL_JSON, metadata_json, add_jsons)
        }

        fn new_with_protocol_metadata_and_adds(
            name: &str,
            protocol_json: &str,
            metadata_json: &str,
            add_jsons: &[String],
        ) -> Result<Self, Box<dyn std::error::Error>> {
            let path = Path::new("target")
                .join("delta-funnel-named-source-tests")
                .join(unique_name(name)?);
            let log_path = path.join("_delta_log");
            fs::create_dir_all(&log_path)?;
            fs::write(
                log_path.join("00000000000000000000.json"),
                format!("{protocol_json}\n{metadata_json}\n"),
            )?;
            fs::write(log_path.join("00000000000000000001.json"), {
                let mut actions = add_jsons.join("\n");
                actions.push('\n');
                actions
            })?;

            Ok(Self { path })
        }
    }

    const PROTOCOL_JSON: &str = r#"{"protocol":{"minReaderVersion":1,"minWriterVersion":2}}"#;
    const METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1587968585495}}"#;
    const BOOLEAN_DATA_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"is_current\",\"type\":\"boolean\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1587968585495}}"#;
    const BINARY_DATA_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"payload\",\"type\":\"binary\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1587968585495}}"#;
    const DATE_DATA_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"event_date\",\"type\":\"date\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1587968585495}}"#;
    const DECIMAL_DATA_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"amount\",\"type\":\"decimal(10,2)\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1587968585495}}"#;
    const FLOATING_DATA_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"float_score\",\"type\":\"float\",\"nullable\":true,\"metadata\":{}},{\"name\":\"double_score\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1587968585495}}"#;
    const STRING_DATA_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"customer_name\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1587968585495}}"#;
    const TIMESTAMP_DATA_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"event_ts\",\"type\":\"timestamp\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1587968585495}}"#;
    const TIMESTAMP_NTZ_DATA_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"event_ts_ntz\",\"type\":\"timestamp_ntz\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1587968585495}}"#;
    const PARTITIONED_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"region\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["region"],"configuration":{},"createdTime":1587968585495}}"#;
    const INTEGER_PARTITIONED_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"byte_part\",\"type\":\"byte\",\"nullable\":true,\"metadata\":{}},{\"name\":\"short_part\",\"type\":\"short\",\"nullable\":true,\"metadata\":{}},{\"name\":\"int_part\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"long_part\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["byte_part","short_part","int_part","long_part"],"configuration":{},"createdTime":1587968585495}}"#;
    const BOOLEAN_PARTITIONED_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"is_current\",\"type\":\"boolean\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["is_current"],"configuration":{},"createdTime":1587968585495}}"#;
    const DATE_PARTITIONED_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"event_date\",\"type\":\"date\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["event_date"],"configuration":{},"createdTime":1587968585495}}"#;
    const DECIMAL_PARTITIONED_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"amount\",\"type\":\"decimal(10,2)\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["amount"],"configuration":{},"createdTime":1587968585495}}"#;
    const FLOATING_PARTITIONED_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"float_part\",\"type\":\"float\",\"nullable\":true,\"metadata\":{}},{\"name\":\"double_part\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["float_part","double_part"],"configuration":{},"createdTime":1587968585495}}"#;
    const BINARY_PARTITIONED_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"payload\",\"type\":\"binary\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["payload"],"configuration":{},"createdTime":1587968585495}}"#;
    const TIMESTAMP_PARTITIONED_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"event_ts\",\"type\":\"timestamp\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["event_ts"],"configuration":{},"createdTime":1587968585495}}"#;
    const TIMESTAMP_NTZ_PROTOCOL_JSON: &str = r#"{"protocol":{"minReaderVersion":3,"minWriterVersion":7,"readerFeatures":["timestampNtz"],"writerFeatures":["timestampNtz"]}}"#;
    const TIMESTAMP_NTZ_PARTITIONED_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"event_ts_ntz\",\"type\":\"timestamp_ntz\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["event_ts_ntz"],"configuration":{},"createdTime":1587968585495}}"#;
    const DELETION_VECTOR_PROTOCOL_JSON: &str = r#"{"protocol":{"minReaderVersion":3,"minWriterVersion":7,"readerFeatures":["deletionVectors"],"writerFeatures":["deletionVectors"]}}"#;

    fn add_json(path: &str) -> String {
        format!(
            r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true}}}}"#
        )
    }

    fn partitioned_add_json(path: &str, partition_values_json: &str) -> String {
        format!(
            r#"{{"add":{{"path":"{path}","partitionValues":{partition_values_json},"size":0,"modificationTime":1587968586000,"dataChange":true}}}}"#
        )
    }

    fn add_json_with_id_stats(
        path: &str,
        num_records: i64,
        min_value: i32,
        max_value: i32,
        null_count: i64,
    ) -> String {
        format!(
            r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{\"id\":{min_value}}},\"maxValues\":{{\"id\":{max_value}}},\"nullCount\":{{\"id\":{null_count}}}}}"}}}}"#
        )
    }

    fn add_json_with_boolean_stats(
        path: &str,
        num_records: i64,
        min_value: bool,
        max_value: bool,
        null_count: i64,
    ) -> String {
        format!(
            r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{\"is_current\":{min_value}}},\"maxValues\":{{\"is_current\":{max_value}}},\"nullCount\":{{\"is_current\":{null_count}}}}}"}}}}"#
        )
    }

    fn add_json_with_partial_boolean_stats(
        path: &str,
        num_records: i64,
        min_value: Option<bool>,
        max_value: Option<bool>,
        null_count: Option<i64>,
    ) -> String {
        let min_values = min_value
            .map(|value| format!(r#"\"is_current\":{value}"#))
            .unwrap_or_default();
        let max_values = max_value
            .map(|value| format!(r#"\"is_current\":{value}"#))
            .unwrap_or_default();
        let null_count = null_count
            .map(|value| format!(r#"\"is_current\":{value}"#))
            .unwrap_or_default();

        format!(
            r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{{min_values}}},\"maxValues\":{{{max_values}}},\"nullCount\":{{{null_count}}}}}"}}}}"#
        )
    }

    fn add_json_with_date_stats(
        path: &str,
        num_records: i64,
        min_value: &str,
        max_value: &str,
        null_count: i64,
    ) -> String {
        format!(
            r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{\"event_date\":\"{min_value}\"}},\"maxValues\":{{\"event_date\":\"{max_value}\"}},\"nullCount\":{{\"event_date\":{null_count}}}}}"}}}}"#
        )
    }

    fn add_json_with_partial_date_stats(
        path: &str,
        num_records: i64,
        min_value: Option<&str>,
        max_value: Option<&str>,
        null_count: Option<i64>,
    ) -> String {
        let min_values = min_value
            .map(|value| format!(r#"\"event_date\":\"{value}\""#))
            .unwrap_or_default();
        let max_values = max_value
            .map(|value| format!(r#"\"event_date\":\"{value}\""#))
            .unwrap_or_default();
        let null_count = null_count
            .map(|value| format!(r#"\"event_date\":{value}"#))
            .unwrap_or_default();

        format!(
            r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{{min_values}}},\"maxValues\":{{{max_values}}},\"nullCount\":{{{null_count}}}}}"}}}}"#
        )
    }

    fn add_json_with_decimal_stats(
        path: &str,
        num_records: i64,
        min_value: &str,
        max_value: &str,
        null_count: i64,
    ) -> String {
        format!(
            r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{\"amount\":\"{min_value}\"}},\"maxValues\":{{\"amount\":\"{max_value}\"}},\"nullCount\":{{\"amount\":{null_count}}}}}"}}}}"#
        )
    }

    fn add_json_with_partial_decimal_stats(
        path: &str,
        num_records: i64,
        min_value: Option<&str>,
        max_value: Option<&str>,
        null_count: Option<i64>,
    ) -> String {
        let min_values = min_value
            .map(|value| format!(r#"\"amount\":\"{value}\""#))
            .unwrap_or_default();
        let max_values = max_value
            .map(|value| format!(r#"\"amount\":\"{value}\""#))
            .unwrap_or_default();
        let null_count = null_count
            .map(|value| format!(r#"\"amount\":{value}"#))
            .unwrap_or_default();

        format!(
            r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{{min_values}}},\"maxValues\":{{{max_values}}},\"nullCount\":{{{null_count}}}}}"}}}}"#
        )
    }

    fn add_json_with_binary_stats(
        path: &str,
        num_records: i64,
        min_value: &str,
        max_value: &str,
        null_count: i64,
    ) -> String {
        format!(
            r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{\"payload\":\"{min_value}\"}},\"maxValues\":{{\"payload\":\"{max_value}\"}},\"nullCount\":{{\"payload\":{null_count}}}}}"}}}}"#
        )
    }

    fn add_json_with_partial_binary_stats(
        path: &str,
        num_records: i64,
        min_value: Option<&str>,
        max_value: Option<&str>,
        null_count: Option<i64>,
    ) -> String {
        let min_values = min_value
            .map(|value| format!(r#"\"payload\":\"{value}\""#))
            .unwrap_or_default();
        let max_values = max_value
            .map(|value| format!(r#"\"payload\":\"{value}\""#))
            .unwrap_or_default();
        let null_count = null_count
            .map(|value| format!(r#"\"payload\":{value}"#))
            .unwrap_or_default();

        format!(
            r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{{min_values}}},\"maxValues\":{{{max_values}}},\"nullCount\":{{{null_count}}}}}"}}}}"#
        )
    }

    fn add_json_with_floating_stats(
        path: &str,
        num_records: i64,
        float_min_value: &str,
        float_max_value: &str,
        double_min_value: &str,
        double_max_value: &str,
        null_count: i64,
    ) -> String {
        format!(
            r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{\"float_score\":{float_min_value},\"double_score\":{double_min_value}}},\"maxValues\":{{\"float_score\":{float_max_value},\"double_score\":{double_max_value}}},\"nullCount\":{{\"float_score\":{null_count},\"double_score\":{null_count}}}}}"}}}}"#
        )
    }

    fn add_json_with_partial_floating_stats(
        path: &str,
        num_records: i64,
        float_min_value: Option<&str>,
        float_max_value: Option<&str>,
        double_min_value: Option<&str>,
        double_max_value: Option<&str>,
        null_count: Option<i64>,
    ) -> String {
        let mut min_values = Vec::new();
        if let Some(value) = float_min_value {
            min_values.push(format!(r#"\"float_score\":{value}"#));
        }
        if let Some(value) = double_min_value {
            min_values.push(format!(r#"\"double_score\":{value}"#));
        }

        let mut max_values = Vec::new();
        if let Some(value) = float_max_value {
            max_values.push(format!(r#"\"float_score\":{value}"#));
        }
        if let Some(value) = double_max_value {
            max_values.push(format!(r#"\"double_score\":{value}"#));
        }

        let null_count = null_count
            .map(|value| format!(r#"\"float_score\":{value},\"double_score\":{value}"#))
            .unwrap_or_default();

        format!(
            r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{{}}},\"maxValues\":{{{}}},\"nullCount\":{{{null_count}}}}}"}}}}"#,
            min_values.join(","),
            max_values.join(",")
        )
    }

    fn add_json_with_string_stats(
        path: &str,
        num_records: i64,
        min_value: &str,
        max_value: &str,
        null_count: i64,
    ) -> String {
        format!(
            r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{\"customer_name\":\"{min_value}\"}},\"maxValues\":{{\"customer_name\":\"{max_value}\"}},\"nullCount\":{{\"customer_name\":{null_count}}}}}"}}}}"#
        )
    }

    fn add_json_with_partial_string_stats(
        path: &str,
        num_records: i64,
        min_value: Option<&str>,
        max_value: Option<&str>,
        null_count: Option<i64>,
    ) -> String {
        let min_values = min_value
            .map(|value| format!(r#"\"customer_name\":\"{value}\""#))
            .unwrap_or_default();
        let max_values = max_value
            .map(|value| format!(r#"\"customer_name\":\"{value}\""#))
            .unwrap_or_default();
        let null_count = null_count
            .map(|value| format!(r#"\"customer_name\":{value}"#))
            .unwrap_or_default();

        format!(
            r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{{min_values}}},\"maxValues\":{{{max_values}}},\"nullCount\":{{{null_count}}}}}"}}}}"#
        )
    }

    fn add_json_with_timestamp_stats(
        path: &str,
        column_name: &str,
        num_records: i64,
        min_value: &str,
        max_value: &str,
        null_count: i64,
    ) -> String {
        format!(
            r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{\"{column_name}\":\"{min_value}\"}},\"maxValues\":{{\"{column_name}\":\"{max_value}\"}},\"nullCount\":{{\"{column_name}\":{null_count}}}}}"}}}}"#
        )
    }

    fn add_json_with_partial_timestamp_stats(
        path: &str,
        column_name: &str,
        num_records: i64,
        min_value: Option<&str>,
        max_value: Option<&str>,
        null_count: Option<i64>,
    ) -> String {
        let min_values = min_value
            .map(|value| format!(r#"\"{column_name}\":\"{value}\""#))
            .unwrap_or_default();
        let max_values = max_value
            .map(|value| format!(r#"\"{column_name}\":\"{value}\""#))
            .unwrap_or_default();
        let null_count = null_count
            .map(|value| format!(r#"\"{column_name}\":{value}"#))
            .unwrap_or_default();

        format!(
            r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{{min_values}}},\"maxValues\":{{{max_values}}},\"nullCount\":{{{null_count}}}}}"}}}}"#
        )
    }

    fn dv_add_json_with_id_stats(
        path: &str,
        num_records: i64,
        min_value: i32,
        max_value: i32,
        null_count: i64,
    ) -> String {
        format!(
            r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{\"id\":{min_value}}},\"maxValues\":{{\"id\":{max_value}}},\"nullCount\":{{\"id\":{null_count}}}}}","deletionVector":{{"storageType":"u","pathOrInlineDv":"vBn[lx{{q8@P<9BNH/isA","offset":1,"sizeInBytes":36,"cardinality":2}}}}}}"#
        )
    }

    fn partitioned_dv_add_json(path: &str, partition_values_json: &str) -> String {
        format!(
            r#"{{"add":{{"path":"{path}","partitionValues":{partition_values_json},"size":0,"modificationTime":1587968586000,"dataChange":true,"deletionVector":{{"storageType":"u","pathOrInlineDv":"vBn[lx{{q8@P<9BNH/isA","offset":1,"sizeInBytes":36,"cardinality":2}}}}}}"#
        )
    }

    fn partitioned_dv_add_json_with_id_stats_and_size(
        path: &str,
        partition_values_json: &str,
        size: i64,
        num_records: i64,
        min_value: i32,
        max_value: i32,
        null_count: i64,
    ) -> String {
        format!(
            r#"{{"add":{{"path":"{path}","partitionValues":{partition_values_json},"size":{size},"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{\"id\":{min_value}}},\"maxValues\":{{\"id\":{max_value}}},\"nullCount\":{{\"id\":{null_count}}}}}","deletionVector":{{"storageType":"u","pathOrInlineDv":"vBn[lx{{q8@P<9BNH/isA","offset":1,"sizeInBytes":36,"cardinality":2}}}}}}"#
        )
    }

    fn integer_partitioned_add_json(path: &str, value: &str) -> String {
        partitioned_add_json(
            path,
            &format!(
                r#"{{"byte_part":"{value}","short_part":"{value}","int_part":"{value}","long_part":"{value}"}}"#
            ),
        )
    }

    fn unique_name(name: &str) -> Result<String, Box<dyn std::error::Error>> {
        let nanos = SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos();

        Ok(format!("{}-{}-{nanos}", std::process::id(), name))
    }

    type CapturedStorageOptions = Arc<Mutex<Vec<DeltaStorageOptions>>>;

    fn storage_options(entries: &[(&str, &str)]) -> DeltaStorageOptions {
        entries
            .iter()
            .map(|(key, value)| ((*key).to_owned(), (*value).to_owned()))
            .collect()
    }

    fn env_vars(entries: &[(&str, &str)]) -> Vec<(String, String)> {
        entries
            .iter()
            .map(|(key, value)| ((*key).to_owned(), (*value).to_owned()))
            .collect()
    }

    fn s3_builder_from_storage_options(options: &DeltaStorageOptions) -> AmazonS3Builder {
        options.iter().fold(
            AmazonS3Builder::new().with_url("s3://bucket/root"),
            |builder, (key, value)| match key.to_ascii_lowercase().parse::<AmazonS3ConfigKey>() {
                Ok(key) => builder.with_config(key, value.as_str()),
                Err(_) => builder,
            },
        )
    }

    fn unique_storage_scheme(name: &str) -> Result<String, Box<dyn std::error::Error>> {
        let nanos = SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos();
        let sanitized_name = name
            .chars()
            .filter(|character| character.is_ascii_alphanumeric())
            .collect::<String>();

        Ok(format!("df{sanitized_name}{}{}", std::process::id(), nanos))
    }

    fn register_capturing_storage_handler(
        scheme: &str,
        captured: CapturedStorageOptions,
    ) -> Result<(), Box<dyn std::error::Error>> {
        super::kernel::insert_url_handler(
            scheme,
            Arc::new(move |_url, options| {
                let options = options.into_iter().collect::<BTreeMap<_, _>>();
                captured
                    .lock()
                    .map_err(|_| delta_kernel::object_store::Error::Generic {
                        store: "capture",
                        source: std::io::Error::other("captured storage options lock poisoned")
                            .into(),
                    })?
                    .push(options);

                Ok((Box::new(InMemory::new()), ObjectStorePath::from("")))
            }),
        )?;

        Ok(())
    }

    fn register_failing_storage_handler(
        scheme: &str,
        captured: CapturedStorageOptions,
        secret_value: &'static str,
    ) -> Result<(), Box<dyn std::error::Error>> {
        super::kernel::insert_url_handler(
            scheme,
            Arc::new(move |_url, options| {
                let options = options.into_iter().collect::<BTreeMap<_, _>>();
                captured
                    .lock()
                    .map_err(|_| delta_kernel::object_store::Error::Generic {
                        store: "capture",
                        source: std::io::Error::other("captured storage options lock poisoned")
                            .into(),
                    })?
                    .push(options);

                Err(delta_kernel::object_store::Error::Generic {
                    store: "capture",
                    source: std::io::Error::other(secret_value).into(),
                })
            }),
        )?;

        Ok(())
    }

    fn captured_storage_options(captured: &CapturedStorageOptions) -> Vec<DeltaStorageOptions> {
        captured
            .lock()
            .map(|options| options.clone())
            .unwrap_or_default()
    }

    fn kernel_scan_file_paths(
        scan: &ProjectedDeltaScan,
        table_uri: &str,
    ) -> Result<Vec<String>, Box<dyn std::error::Error>> {
        let storage_options = DeltaStorageOptions::default();
        let mut paths = scan
            .expand_kernel_scan_metadata(table_uri, &storage_options)?
            .files
            .into_iter()
            .map(|file| file.path)
            .collect::<Vec<_>>();
        paths.sort();
        Ok(paths)
    }

    #[derive(Debug, PartialEq, Eq)]
    struct KernelScanFileBoundary {
        path: String,
        has_deletion_vector: bool,
    }

    fn kernel_scan_file_boundaries(
        scan: &ProjectedDeltaScan,
        table_uri: &str,
    ) -> Result<Vec<KernelScanFileBoundary>, Box<dyn std::error::Error>> {
        let storage_options = DeltaStorageOptions::default();
        let mut boundaries = scan
            .expand_kernel_scan_metadata(table_uri, &storage_options)?
            .files
            .into_iter()
            .map(|file| KernelScanFileBoundary {
                path: file.path,
                has_deletion_vector: file.deletion_vector.is_present(),
            })
            .collect::<Vec<_>>();
        boundaries.sort_by(|left, right| left.path.cmp(&right.path));
        Ok(boundaries)
    }

    #[test]
    fn kernel_scan_metadata_expansion_preserves_file_metadata_without_data_reads()
    -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_protocol_metadata_and_adds(
            "kernel-scan-metadata-expansion",
            DELETION_VECTOR_PROTOCOL_JSON,
            PARTITIONED_METADATA_JSON,
            &[partitioned_dv_add_json_with_id_stats_and_size(
                "region-west-dv.parquet",
                r#"{"region":"us-west"}"#,
                123,
                7,
                10,
                20,
                0,
            )],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;
        let scan = build_projected_predicated_stats_delta_scan(&source, None)?;

        let expansion =
            scan.expand_kernel_scan_metadata(source.table_uri(), source.storage_options())?;

        assert!(expansion.scan_metadata_exhausted);
        assert_eq!(expansion.files.len(), 1);
        let file = &expansion.files[0];
        assert_eq!(file.path, "region-west-dv.parquet");
        assert_eq!(file.size, 123);
        assert_eq!(file.modification_time, 1587968586000);
        assert_eq!(file.stats.as_ref().map(|stats| stats.num_records), Some(7));
        assert!(file.deletion_vector.is_present());
        assert_eq!(
            file.partition_values.get("region").map(String::as_str),
            Some("us-west")
        );

        Ok(())
    }

    #[test]
    fn kernel_scan_metadata_expansion_reports_empty_scan_as_exhausted()
    -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-scan-metadata-empty",
            METADATA_JSON,
            &[],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;
        let scan = build_projected_predicated_stats_delta_scan(&source, None)?;

        let expansion =
            scan.expand_kernel_scan_metadata(source.table_uri(), source.storage_options())?;

        assert!(expansion.scan_metadata_exhausted);
        assert!(expansion.files.is_empty());

        Ok(())
    }

    #[test]
    fn kernel_scan_file_metadata_preserves_transform_handle_without_execution()
    -> Result<(), Box<dyn std::error::Error>> {
        let transform = Arc::new(Expression::Column(ColumnName::new(["physical_id"])));
        let partition_values = HashMap::from([("region".to_owned(), "us-west".to_owned())]);

        let file = super::kernel::ScanFile {
            path: "part-with-transform.parquet".to_owned(),
            size: 456,
            modification_time: 1587968587000,
            stats: None,
            dv_info: super::kernel::DvInfo::default(),
            transform: Some(Arc::clone(&transform)),
            partition_values,
        };

        let metadata = super::KernelScanFileMetadata::from_kernel_scan_file(file);

        assert_eq!(metadata.path, "part-with-transform.parquet");
        assert_eq!(metadata.size, 456);
        assert_eq!(metadata.modification_time, 1587968587000);
        assert!(metadata.stats.is_none());
        assert!(!metadata.deletion_vector.is_present());
        assert!(metadata.physical_to_logical_transform.is_required());
        assert_eq!(
            metadata.partition_values.get("region").map(String::as_str),
            Some("us-west")
        );
        match metadata.physical_to_logical_transform {
            super::KernelPhysicalToLogicalTransform::Required(handle) => {
                assert!(Arc::ptr_eq(&handle.transform, &transform));
            }
            super::KernelPhysicalToLogicalTransform::NotRequired => {
                return Err("transform handle should be preserved".into());
            }
        }

        Ok(())
    }

    fn kernel_partition_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-partition-characterization",
            PARTITIONED_METADATA_JSON,
            &[
                partitioned_add_json("region-us-west.parquet", r#"{"region":"us-west"}"#),
                partitioned_add_json("region-us-east.parquet", r#"{"region":"us-east"}"#),
                partitioned_add_json("region-null.parquet", r#"{"region":null}"#),
                partitioned_add_json("region-missing.parquet", r#"{}"#),
                partitioned_add_json("region-empty-string.parquet", r#"{"region":""}"#),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_integer_partition_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-integer-partition-characterization",
            INTEGER_PARTITIONED_METADATA_JSON,
            &[
                integer_partitioned_add_json("integer--1.parquet", "-1"),
                integer_partitioned_add_json("integer-2.parquet", "2"),
                integer_partitioned_add_json("integer-10.parquet", "10"),
                partitioned_add_json(
                    "integer-null.parquet",
                    r#"{"byte_part":null,"short_part":null,"int_part":null,"long_part":null}"#,
                ),
                integer_partitioned_add_json("integer-empty.parquet", ""),
                partitioned_add_json("integer-missing.parquet", r#"{}"#),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_boolean_partition_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-boolean-partition-characterization",
            BOOLEAN_PARTITIONED_METADATA_JSON,
            &[
                partitioned_add_json("boolean-true.parquet", r#"{"is_current":"true"}"#),
                partitioned_add_json("boolean-false.parquet", r#"{"is_current":"false"}"#),
                partitioned_add_json("boolean-null.parquet", r#"{"is_current":null}"#),
                partitioned_add_json("boolean-empty.parquet", r#"{"is_current":""}"#),
                partitioned_add_json("boolean-missing.parquet", r#"{}"#),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_date_partition_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-date-partition-characterization",
            DATE_PARTITIONED_METADATA_JSON,
            &[
                partitioned_add_json("date-pre-epoch.parquet", r#"{"event_date":"1969-12-31"}"#),
                partitioned_add_json("date-epoch.parquet", r#"{"event_date":"1970-01-01"}"#),
                partitioned_add_json("date-leap-day.parquet", r#"{"event_date":"2024-02-29"}"#),
                partitioned_add_json("date-new-year.parquet", r#"{"event_date":"2026-01-01"}"#),
                partitioned_add_json("date-null.parquet", r#"{"event_date":null}"#),
                partitioned_add_json("date-empty.parquet", r#"{"event_date":""}"#),
                partitioned_add_json("date-missing.parquet", r#"{}"#),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_decimal_partition_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-decimal-partition-characterization",
            DECIMAL_PARTITIONED_METADATA_JSON,
            &[
                partitioned_add_json("decimal-negative.parquet", r#"{"amount":"-1.23"}"#),
                partitioned_add_json("decimal-zero.parquet", r#"{"amount":"0.00"}"#),
                partitioned_add_json("decimal-two.parquet", r#"{"amount":"2.00"}"#),
                partitioned_add_json("decimal-ten.parquet", r#"{"amount":"10.00"}"#),
                partitioned_add_json("decimal-large.parquet", r#"{"amount":"123.45"}"#),
                partitioned_add_json("decimal-null.parquet", r#"{"amount":null}"#),
                partitioned_add_json("decimal-empty.parquet", r#"{"amount":""}"#),
                partitioned_add_json("decimal-missing.parquet", r#"{}"#),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_floating_partition_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-floating-partition-characterization",
            FLOATING_PARTITIONED_METADATA_JSON,
            &[
                partitioned_add_json(
                    "floating-neg.parquet",
                    r#"{"float_part":"-1.5","double_part":"-2.25"}"#,
                ),
                partitioned_add_json(
                    "floating-neg-zero.parquet",
                    r#"{"float_part":"-0.0","double_part":"-0.0"}"#,
                ),
                partitioned_add_json(
                    "floating-pos-zero.parquet",
                    r#"{"float_part":"0.0","double_part":"0.0"}"#,
                ),
                partitioned_add_json(
                    "floating-one.parquet",
                    r#"{"float_part":"1.5","double_part":"2.25"}"#,
                ),
                partitioned_add_json(
                    "floating-ten.parquet",
                    r#"{"float_part":"10.0","double_part":"10.0"}"#,
                ),
                partitioned_add_json(
                    "floating-null.parquet",
                    r#"{"float_part":null,"double_part":null}"#,
                ),
                partitioned_add_json(
                    "floating-empty.parquet",
                    r#"{"float_part":"","double_part":""}"#,
                ),
                partitioned_add_json("floating-missing.parquet", r#"{}"#),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_binary_partition_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-binary-partition-characterization",
            BINARY_PARTITIONED_METADATA_JSON,
            &[
                partitioned_add_json("binary-HELLO.parquet", r#"{"payload":"HELLO"}"#),
                partitioned_add_json("binary-hello.parquet", r#"{"payload":"hello"}"#),
                partitioned_add_json("binary-special.parquet", r#"{"payload":"/=%"}"#),
                partitioned_add_json("binary-null.parquet", r#"{"payload":null}"#),
                partitioned_add_json("binary-empty.parquet", r#"{"payload":""}"#),
                partitioned_add_json("binary-missing.parquet", r#"{}"#),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_timestamp_partition_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-timestamp-partition-characterization",
            TIMESTAMP_PARTITIONED_METADATA_JSON,
            &[
                partitioned_add_json(
                    "timestamp-pre-epoch.parquet",
                    r#"{"event_ts":"1969-12-31T23:59:59.999999Z"}"#,
                ),
                partitioned_add_json(
                    "timestamp-low.parquet",
                    r#"{"event_ts":"2025-12-31T23:59:59.999999Z"}"#,
                ),
                partitioned_add_json(
                    "timestamp-target.parquet",
                    r#"{"event_ts":"2026-01-01T00:00:00.123456Z"}"#,
                ),
                partitioned_add_json(
                    "timestamp-high.parquet",
                    r#"{"event_ts":"2026-01-01T00:00:00.123457Z"}"#,
                ),
                partitioned_add_json("timestamp-null.parquet", r#"{"event_ts":null}"#),
                partitioned_add_json("timestamp-empty.parquet", r#"{"event_ts":""}"#),
                partitioned_add_json("timestamp-missing.parquet", r#"{}"#),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_timestamp_ntz_partition_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_protocol_metadata_and_adds(
            "kernel-timestamp-ntz-partition-characterization",
            TIMESTAMP_NTZ_PROTOCOL_JSON,
            TIMESTAMP_NTZ_PARTITIONED_METADATA_JSON,
            &[
                partitioned_add_json(
                    "timestamp-ntz-pre-epoch.parquet",
                    r#"{"event_ts_ntz":"1969-12-31 23:59:59.999999"}"#,
                ),
                partitioned_add_json(
                    "timestamp-ntz-low-space.parquet",
                    r#"{"event_ts_ntz":"2025-12-31 23:59:59.999999"}"#,
                ),
                partitioned_add_json(
                    "timestamp-ntz-target-space.parquet",
                    r#"{"event_ts_ntz":"2026-01-01 00:00:00.123456"}"#,
                ),
                partitioned_add_json(
                    "timestamp-ntz-high.parquet",
                    r#"{"event_ts_ntz":"2026-01-01 00:00:00.123457"}"#,
                ),
                partitioned_add_json("timestamp-ntz-null.parquet", r#"{"event_ts_ntz":null}"#),
                partitioned_add_json("timestamp-ntz-empty.parquet", r#"{"event_ts_ntz":""}"#),
                partitioned_add_json("timestamp-ntz-missing.parquet", r#"{}"#),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_data_stats_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-data-stats-characterization",
            METADATA_JSON,
            &[
                add_json_with_id_stats("id-impossible.parquet", 10, 1, 50, 0),
                add_json_with_id_stats("id-possible.parquet", 10, 101, 150, 0),
                add_json("id-missing-stats.parquet"),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_boolean_data_stats_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-boolean-data-stats-characterization",
            BOOLEAN_DATA_METADATA_JSON,
            &[
                add_json_with_boolean_stats("boolean-false-only.parquet", 10, false, false, 0),
                add_json_with_boolean_stats("boolean-true-only.parquet", 10, true, true, 0),
                add_json_with_boolean_stats("boolean-mixed.parquet", 10, false, true, 0),
                add_json_with_boolean_stats("boolean-false-with-null.parquet", 10, false, false, 2),
                add_json_with_boolean_stats("boolean-true-with-null.parquet", 10, true, true, 2),
                add_json_with_partial_boolean_stats(
                    "boolean-all-null.parquet",
                    10,
                    None,
                    None,
                    Some(10),
                ),
                add_json("boolean-missing-stats.parquet"),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_partial_boolean_data_stats_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-partial-boolean-data-stats-characterization",
            BOOLEAN_DATA_METADATA_JSON,
            &[
                add_json_with_partial_boolean_stats(
                    "boolean-min-only-false.parquet",
                    10,
                    Some(false),
                    None,
                    Some(0),
                ),
                add_json_with_partial_boolean_stats(
                    "boolean-max-only-true.parquet",
                    10,
                    None,
                    Some(true),
                    Some(0),
                ),
                add_json_with_partial_boolean_stats(
                    "boolean-counts-only.parquet",
                    10,
                    None,
                    None,
                    Some(0),
                ),
                add_json_with_partial_boolean_stats(
                    "boolean-missing-null-count.parquet",
                    10,
                    Some(false),
                    Some(true),
                    None,
                ),
                add_json("boolean-missing-stats.parquet"),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn assert_boolean_stats_min_max_error(
        result: Result<Vec<String>, Box<dyn std::error::Error>>,
    ) -> Result<(), Box<dyn std::error::Error>> {
        let error = match result {
            Ok(paths) => {
                return Err(format!(
                    "boolean min/max stats predicate should fail kernel scan: {paths:?}"
                )
                .into());
            }
            Err(error) => error,
        };
        let message = error.to_string();
        let debug_message = format!("{error:?}");
        assert!(
            message.contains("minValues") || debug_message.contains("minValues"),
            "{message}\n{debug_message}"
        );

        Ok(())
    }

    fn assert_binary_stats_min_max_error(
        result: Result<Vec<String>, Box<dyn std::error::Error>>,
    ) -> Result<(), Box<dyn std::error::Error>> {
        let error = match result {
            Ok(paths) => {
                return Err(format!(
                    "binary min/max stats predicate should fail kernel scan: {paths:?}"
                )
                .into());
            }
            Err(error) => error,
        };
        let message = error.to_string();
        let debug_message = format!("{error:?}");
        assert!(
            message.contains("minValues")
                || message.contains("maxValues")
                || debug_message.contains("minValues")
                || debug_message.contains("maxValues"),
            "{message}\n{debug_message}"
        );

        Ok(())
    }

    fn assert_unsupported_literal_error(
        result: Result<Vec<String>, Box<dyn std::error::Error>>,
    ) -> Result<(), Box<dyn std::error::Error>> {
        let error = match result {
            Ok(paths) => {
                return Err(format!(
                    "unsupported literal predicate should fail before kernel scan: {paths:?}"
                )
                .into());
            }
            Err(error) => error,
        };
        let message = error.to_string();
        let debug_message = format!("{error:?}");
        assert!(
            message.contains("unsupported DataFusion literal")
                || debug_message.contains("UnsupportedLiteral"),
            "{message}\n{debug_message}"
        );

        Ok(())
    }

    fn kernel_date_data_stats_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-date-data-stats-characterization",
            DATE_DATA_METADATA_JSON,
            &[
                add_json_with_date_stats(
                    "date-pre-epoch-only.parquet",
                    10,
                    "1969-12-31",
                    "1969-12-31",
                    0,
                ),
                add_json_with_date_stats(
                    "date-leap-only.parquet",
                    10,
                    "2024-02-29",
                    "2024-02-29",
                    0,
                ),
                add_json_with_date_stats(
                    "date-new-year-only.parquet",
                    10,
                    "2026-01-01",
                    "2026-01-01",
                    0,
                ),
                add_json_with_date_stats("date-range.parquet", 10, "2024-02-29", "2026-01-01", 0),
                add_json_with_date_stats(
                    "date-new-year-with-null.parquet",
                    10,
                    "2026-01-01",
                    "2026-01-01",
                    2,
                ),
                add_json_with_partial_date_stats("date-all-null.parquet", 10, None, None, Some(10)),
                add_json("date-missing-stats.parquet"),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_partial_date_data_stats_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-partial-date-data-stats-characterization",
            DATE_DATA_METADATA_JSON,
            &[
                add_json_with_partial_date_stats(
                    "date-min-only-high.parquet",
                    10,
                    Some("2026-01-01"),
                    None,
                    Some(0),
                ),
                add_json_with_partial_date_stats(
                    "date-max-only-low.parquet",
                    10,
                    None,
                    Some("2024-02-29"),
                    Some(0),
                ),
                add_json_with_partial_date_stats(
                    "date-counts-only.parquet",
                    10,
                    None,
                    None,
                    Some(0),
                ),
                add_json_with_partial_date_stats(
                    "date-missing-null-count.parquet",
                    10,
                    Some("2024-02-29"),
                    Some("2026-01-01"),
                    None,
                ),
                add_json("date-missing-stats.parquet"),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_decimal_data_stats_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-decimal-data-stats-characterization",
            DECIMAL_DATA_METADATA_JSON,
            &[
                add_json_with_decimal_stats(
                    "decimal-negative-only.parquet",
                    10,
                    "-1.23",
                    "-1.23",
                    0,
                ),
                add_json_with_decimal_stats("decimal-zero-only.parquet", 10, "0.00", "0.00", 0),
                add_json_with_decimal_stats("decimal-two-only.parquet", 10, "2.00", "2.00", 0),
                add_json_with_decimal_stats("decimal-ten-only.parquet", 10, "10.00", "10.00", 0),
                add_json_with_decimal_stats(
                    "decimal-large-only.parquet",
                    10,
                    "123.45",
                    "123.45",
                    0,
                ),
                add_json_with_decimal_stats("decimal-range.parquet", 10, "0.00", "10.00", 0),
                add_json_with_decimal_stats("decimal-two-with-null.parquet", 10, "2.00", "2.00", 2),
                add_json_with_partial_decimal_stats(
                    "decimal-all-null.parquet",
                    10,
                    None,
                    None,
                    Some(10),
                ),
                add_json("decimal-missing-stats.parquet"),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_partial_decimal_data_stats_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-partial-decimal-data-stats-characterization",
            DECIMAL_DATA_METADATA_JSON,
            &[
                add_json_with_partial_decimal_stats(
                    "decimal-min-only-high.parquet",
                    10,
                    Some("10.00"),
                    None,
                    Some(0),
                ),
                add_json_with_partial_decimal_stats(
                    "decimal-max-only-low.parquet",
                    10,
                    None,
                    Some("0.00"),
                    Some(0),
                ),
                add_json_with_partial_decimal_stats(
                    "decimal-counts-only.parquet",
                    10,
                    None,
                    None,
                    Some(0),
                ),
                add_json_with_partial_decimal_stats(
                    "decimal-missing-null-count.parquet",
                    10,
                    Some("0.00"),
                    Some("10.00"),
                    None,
                ),
                add_json("decimal-missing-stats.parquet"),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_binary_data_stats_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-binary-data-stats-characterization",
            BINARY_DATA_METADATA_JSON,
            &[
                add_json_with_binary_stats("binary-HELLO.parquet", 10, "HELLO", "HELLO", 0),
                add_json_with_binary_stats("binary-empty.parquet", 10, "", "", 0),
                add_json_with_binary_stats("binary-hello.parquet", 10, "hello", "hello", 0),
                add_json_with_binary_stats("binary-range.parquet", 10, "a", "z", 0),
                add_json_with_binary_stats("binary-special.parquet", 10, "/=%", "/=%", 0),
                add_json_with_binary_stats("binary-with-null.parquet", 10, "hello", "hello", 2),
                add_json_with_partial_binary_stats(
                    "binary-all-null.parquet",
                    10,
                    None,
                    None,
                    Some(10),
                ),
                add_json("binary-missing-stats.parquet"),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_partial_binary_data_stats_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-partial-binary-data-stats-characterization",
            BINARY_DATA_METADATA_JSON,
            &[
                add_json_with_partial_binary_stats(
                    "binary-min-only-high.parquet",
                    10,
                    Some("m"),
                    None,
                    Some(0),
                ),
                add_json_with_partial_binary_stats(
                    "binary-max-only-low.parquet",
                    10,
                    None,
                    Some("a"),
                    Some(0),
                ),
                add_json_with_partial_binary_stats(
                    "binary-counts-only.parquet",
                    10,
                    None,
                    None,
                    Some(0),
                ),
                add_json_with_partial_binary_stats(
                    "binary-missing-null-count.parquet",
                    10,
                    Some("a"),
                    Some("z"),
                    None,
                ),
                add_json("binary-missing-stats.parquet"),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_floating_data_stats_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-floating-data-stats-characterization",
            FLOATING_DATA_METADATA_JSON,
            &[
                add_json_with_floating_stats(
                    "floating-neg.parquet",
                    10,
                    "-1.5",
                    "-1.5",
                    "-2.25",
                    "-2.25",
                    0,
                ),
                add_json_with_floating_stats(
                    "floating-neg-zero.parquet",
                    10,
                    "-0.0",
                    "-0.0",
                    "-0.0",
                    "-0.0",
                    0,
                ),
                add_json_with_floating_stats(
                    "floating-pos-zero.parquet",
                    10,
                    "0.0",
                    "0.0",
                    "0.0",
                    "0.0",
                    0,
                ),
                add_json_with_floating_stats(
                    "floating-one.parquet",
                    10,
                    "1.5",
                    "1.5",
                    "2.25",
                    "2.25",
                    0,
                ),
                add_json_with_floating_stats(
                    "floating-range.parquet",
                    10,
                    "-1.0",
                    "2.0",
                    "-2.0",
                    "3.0",
                    0,
                ),
                add_json_with_floating_stats(
                    "floating-ten.parquet",
                    10,
                    "10.0",
                    "10.0",
                    "10.0",
                    "10.0",
                    0,
                ),
                add_json_with_floating_stats(
                    "floating-one-with-null.parquet",
                    10,
                    "1.5",
                    "1.5",
                    "2.25",
                    "2.25",
                    2,
                ),
                add_json_with_partial_floating_stats(
                    "floating-all-null.parquet",
                    10,
                    None,
                    None,
                    None,
                    None,
                    Some(10),
                ),
                add_json("floating-missing-stats.parquet"),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_partial_floating_data_stats_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-partial-floating-data-stats-characterization",
            FLOATING_DATA_METADATA_JSON,
            &[
                add_json_with_partial_floating_stats(
                    "floating-min-only-high.parquet",
                    10,
                    Some("2.0"),
                    None,
                    Some("2.0"),
                    None,
                    Some(0),
                ),
                add_json_with_partial_floating_stats(
                    "floating-max-only-low.parquet",
                    10,
                    None,
                    Some("0.0"),
                    None,
                    Some("0.0"),
                    Some(0),
                ),
                add_json_with_partial_floating_stats(
                    "floating-counts-only.parquet",
                    10,
                    None,
                    None,
                    None,
                    None,
                    Some(0),
                ),
                add_json_with_partial_floating_stats(
                    "floating-missing-null-count.parquet",
                    10,
                    Some("-1.0"),
                    Some("2.0"),
                    Some("-1.0"),
                    Some("2.0"),
                    None,
                ),
                add_json("floating-missing-stats.parquet"),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_nonfinite_floating_data_stats_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-nonfinite-floating-data-stats-characterization",
            FLOATING_DATA_METADATA_JSON,
            &[
                add_json_with_floating_stats(
                    "floating-valid.parquet",
                    10,
                    "1.5",
                    "1.5",
                    "2.25",
                    "2.25",
                    0,
                ),
                add_json_with_floating_stats(
                    "floating-nan.parquet",
                    10,
                    "\\\"NaN\\\"",
                    "\\\"NaN\\\"",
                    "\\\"NaN\\\"",
                    "\\\"NaN\\\"",
                    0,
                ),
                add_json_with_floating_stats(
                    "floating-inf.parquet",
                    10,
                    "\\\"Infinity\\\"",
                    "\\\"Infinity\\\"",
                    "\\\"Infinity\\\"",
                    "\\\"Infinity\\\"",
                    0,
                ),
                add_json_with_floating_stats(
                    "floating-neg-inf.parquet",
                    10,
                    "\\\"-Infinity\\\"",
                    "\\\"-Infinity\\\"",
                    "\\\"-Infinity\\\"",
                    "\\\"-Infinity\\\"",
                    0,
                ),
                add_json("floating-missing-stats.parquet"),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_string_data_stats_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-string-data-stats-characterization",
            STRING_DATA_METADATA_JSON,
            &[
                add_json_with_string_stats("string-empty-only.parquet", 10, "", "", 0),
                add_json_with_string_stats(
                    "string-mixed-case-only.parquet",
                    10,
                    "Alice",
                    "Alice",
                    0,
                ),
                add_json_with_string_stats("string-alice-only.parquet", 10, "alice", "alice", 0),
                add_json_with_string_stats("string-bob-only.parquet", 10, "bob", "bob", 0),
                add_json_with_string_stats("string-range.parquet", 10, "alice", "morgan", 0),
                add_json_with_string_stats("string-zed-only.parquet", 10, "zed", "zed", 0),
                add_json_with_string_stats(
                    "string-alice-with-null.parquet",
                    10,
                    "alice",
                    "alice",
                    2,
                ),
                add_json_with_partial_string_stats(
                    "string-all-null.parquet",
                    10,
                    None,
                    None,
                    Some(10),
                ),
                add_json("string-missing-stats.parquet"),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_partial_string_data_stats_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-partial-string-data-stats-characterization",
            STRING_DATA_METADATA_JSON,
            &[
                add_json_with_partial_string_stats(
                    "string-min-only-morgan.parquet",
                    10,
                    Some("morgan"),
                    None,
                    Some(0),
                ),
                add_json_with_partial_string_stats(
                    "string-max-only-alice.parquet",
                    10,
                    None,
                    Some("alice"),
                    Some(0),
                ),
                add_json_with_partial_string_stats(
                    "string-counts-only.parquet",
                    10,
                    None,
                    None,
                    Some(0),
                ),
                add_json_with_partial_string_stats(
                    "string-missing-null-count.parquet",
                    10,
                    Some("alice"),
                    Some("morgan"),
                    None,
                ),
                add_json("string-missing-stats.parquet"),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_non_ascii_string_data_stats_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-non-ascii-string-data-stats-characterization",
            STRING_DATA_METADATA_JSON,
            &[
                add_json_with_string_stats("string-ascii-cafe.parquet", 10, "cafe", "cafe", 0),
                add_json_with_string_stats("string-ascii-zulu.parquet", 10, "zulu", "zulu", 0),
                add_json_with_string_stats(
                    "string-eclair.parquet",
                    10,
                    "\\u00e9clair",
                    "\\u00e9clair",
                    0,
                ),
                add_json_with_string_stats(
                    "string-emile.parquet",
                    10,
                    "\\u00e9mile",
                    "\\u00e9mile",
                    0,
                ),
                add_json("string-missing-stats.parquet"),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_timestamp_data_stats_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-timestamp-data-stats-characterization",
            TIMESTAMP_DATA_METADATA_JSON,
            &[
                add_json_with_timestamp_stats(
                    "timestamp-pre-epoch-only.parquet",
                    "event_ts",
                    10,
                    "1969-12-31T23:59:59.999999Z",
                    "1969-12-31T23:59:59.999999Z",
                    0,
                ),
                add_json_with_timestamp_stats(
                    "timestamp-low-only.parquet",
                    "event_ts",
                    10,
                    "2025-12-31T23:59:59.999999Z",
                    "2025-12-31T23:59:59.999999Z",
                    0,
                ),
                add_json_with_timestamp_stats(
                    "timestamp-target-only.parquet",
                    "event_ts",
                    10,
                    "2026-01-01T00:00:00.123456Z",
                    "2026-01-01T00:00:00.123456Z",
                    0,
                ),
                add_json_with_timestamp_stats(
                    "timestamp-high-only.parquet",
                    "event_ts",
                    10,
                    "2026-01-01T00:00:00.123457Z",
                    "2026-01-01T00:00:00.123457Z",
                    0,
                ),
                add_json_with_timestamp_stats(
                    "timestamp-range.parquet",
                    "event_ts",
                    10,
                    "2025-12-31T23:59:59.999999Z",
                    "2026-01-01T00:00:00.123456Z",
                    0,
                ),
                add_json_with_timestamp_stats(
                    "timestamp-target-with-null.parquet",
                    "event_ts",
                    10,
                    "2026-01-01T00:00:00.123456Z",
                    "2026-01-01T00:00:00.123456Z",
                    2,
                ),
                add_json_with_partial_timestamp_stats(
                    "timestamp-all-null.parquet",
                    "event_ts",
                    10,
                    None,
                    None,
                    Some(10),
                ),
                add_json("timestamp-missing-stats.parquet"),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_timestamp_ntz_data_stats_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_protocol_metadata_and_adds(
            "kernel-timestamp-ntz-data-stats-characterization",
            TIMESTAMP_NTZ_PROTOCOL_JSON,
            TIMESTAMP_NTZ_DATA_METADATA_JSON,
            &[
                add_json_with_timestamp_stats(
                    "timestamp-ntz-pre-epoch-only.parquet",
                    "event_ts_ntz",
                    10,
                    "1969-12-31 23:59:59.999999",
                    "1969-12-31 23:59:59.999999",
                    0,
                ),
                add_json_with_timestamp_stats(
                    "timestamp-ntz-low-only.parquet",
                    "event_ts_ntz",
                    10,
                    "2025-12-31 23:59:59.999999",
                    "2025-12-31 23:59:59.999999",
                    0,
                ),
                add_json_with_timestamp_stats(
                    "timestamp-ntz-target-only.parquet",
                    "event_ts_ntz",
                    10,
                    "2026-01-01 00:00:00.123456",
                    "2026-01-01 00:00:00.123456",
                    0,
                ),
                add_json_with_timestamp_stats(
                    "timestamp-ntz-high-only.parquet",
                    "event_ts_ntz",
                    10,
                    "2026-01-01 00:00:00.123457",
                    "2026-01-01 00:00:00.123457",
                    0,
                ),
                add_json_with_timestamp_stats(
                    "timestamp-ntz-range.parquet",
                    "event_ts_ntz",
                    10,
                    "2025-12-31 23:59:59.999999",
                    "2026-01-01 00:00:00.123456",
                    0,
                ),
                add_json_with_timestamp_stats(
                    "timestamp-ntz-target-with-null.parquet",
                    "event_ts_ntz",
                    10,
                    "2026-01-01 00:00:00.123456",
                    "2026-01-01 00:00:00.123456",
                    2,
                ),
                add_json_with_partial_timestamp_stats(
                    "timestamp-ntz-all-null.parquet",
                    "event_ts_ntz",
                    10,
                    None,
                    None,
                    Some(10),
                ),
                add_json("timestamp-ntz-missing-stats.parquet"),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_partial_timestamp_data_stats_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-partial-timestamp-data-stats-characterization",
            TIMESTAMP_DATA_METADATA_JSON,
            &[
                add_json_with_partial_timestamp_stats(
                    "timestamp-min-only-target.parquet",
                    "event_ts",
                    10,
                    Some("2026-01-01T00:00:00.123456Z"),
                    None,
                    Some(0),
                ),
                add_json_with_partial_timestamp_stats(
                    "timestamp-max-only-low.parquet",
                    "event_ts",
                    10,
                    None,
                    Some("2025-12-31T23:59:59.999999Z"),
                    Some(0),
                ),
                add_json_with_partial_timestamp_stats(
                    "timestamp-counts-only.parquet",
                    "event_ts",
                    10,
                    None,
                    None,
                    Some(0),
                ),
                add_json_with_partial_timestamp_stats(
                    "timestamp-missing-null-count.parquet",
                    "event_ts",
                    10,
                    Some("2025-12-31T23:59:59.999999Z"),
                    Some("2026-01-01T00:00:00.123456Z"),
                    None,
                ),
                add_json("timestamp-missing-stats.parquet"),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_partial_timestamp_ntz_data_stats_characterization_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_protocol_metadata_and_adds(
            "kernel-partial-timestamp-ntz-data-stats-characterization",
            TIMESTAMP_NTZ_PROTOCOL_JSON,
            TIMESTAMP_NTZ_DATA_METADATA_JSON,
            &[
                add_json_with_partial_timestamp_stats(
                    "timestamp-ntz-min-only-target.parquet",
                    "event_ts_ntz",
                    10,
                    Some("2026-01-01 00:00:00.123456"),
                    None,
                    Some(0),
                ),
                add_json_with_partial_timestamp_stats(
                    "timestamp-ntz-max-only-low.parquet",
                    "event_ts_ntz",
                    10,
                    None,
                    Some("2025-12-31 23:59:59.999999"),
                    Some(0),
                ),
                add_json_with_partial_timestamp_stats(
                    "timestamp-ntz-counts-only.parquet",
                    "event_ts_ntz",
                    10,
                    None,
                    None,
                    Some(0),
                ),
                add_json_with_partial_timestamp_stats(
                    "timestamp-ntz-missing-null-count.parquet",
                    "event_ts_ntz",
                    10,
                    Some("2025-12-31 23:59:59.999999"),
                    Some("2026-01-01 00:00:00.123456"),
                    None,
                ),
                add_json("timestamp-ntz-missing-stats.parquet"),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_all_sufficient_data_stats_source()
    -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-all-sufficient-data-stats",
            METADATA_JSON,
            &[
                add_json_with_id_stats("id-low-a.parquet", 10, 1, 50, 0),
                add_json_with_id_stats("id-low-b.parquet", 10, 51, 100, 0),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        Ok((table, source))
    }

    fn kernel_predicated_file_paths(
        source: &super::PlannedDeltaSource,
        filter: &datafusion::logical_expr::Expr,
    ) -> Result<Vec<String>, Box<dyn std::error::Error>> {
        let predicate = datafusion_expr_to_kernel_predicate(filter)?;
        kernel_predicate_file_paths(source, predicate)
    }

    fn kernel_predicate_file_paths(
        source: &super::PlannedDeltaSource,
        predicate: DeltaKernelPredicate,
    ) -> Result<Vec<String>, Box<dyn std::error::Error>> {
        let scan = build_projected_predicated_delta_scan(source, None, Some(predicate))?;

        kernel_scan_file_paths(&scan, source.table_uri())
    }

    fn build_projected_predicated_stats_delta_scan(
        source: &super::PlannedDeltaSource,
        predicate: Option<DeltaKernelPredicate>,
    ) -> Result<ProjectedDeltaScan, delta_kernel::Error> {
        let (scan, kernel_schema) = super::kernel::build_projected_predicated_stats_scan(
            source.loaded_snapshot().kernel_snapshot(),
            None,
            predicate,
        )?;

        Ok(ProjectedDeltaScan {
            scan,
            kernel_schema,
        })
    }

    fn kernel_predicated_stats_file_paths(
        source: &super::PlannedDeltaSource,
        filter: &datafusion::logical_expr::Expr,
    ) -> Result<Vec<String>, Box<dyn std::error::Error>> {
        let predicate = datafusion_expr_to_kernel_predicate(filter)?;
        let scan = build_projected_predicated_stats_delta_scan(source, Some(predicate))?;

        kernel_scan_file_paths(&scan, source.table_uri())
    }

    #[derive(Debug, Default, PartialEq, Eq)]
    struct KernelStatsMetadataBoundary {
        selected_rows: usize,
        selected_stats_rows: usize,
        selected_missing_stats_rows: usize,
        has_id_min_values: bool,
        has_id_max_values: bool,
        has_id_null_count: bool,
    }

    fn kernel_stats_metadata_boundary(
        scan: &ProjectedDeltaScan,
        table_uri: &str,
    ) -> Result<KernelStatsMetadataBoundary, Box<dyn std::error::Error>> {
        let storage_options = DeltaStorageOptions::default();
        let table_url = super::kernel::try_parse_uri(table_uri)?;
        let store = super::kernel::store_from_url_opts(
            &table_url,
            storage_options
                .iter()
                .map(|(key, value)| (key.as_str(), value.as_str())),
        )?;
        let engine = super::kernel::DefaultEngineBuilder::new(store).build();
        let mut boundary = KernelStatsMetadataBoundary::default();

        for scan_metadata in scan.kernel_scan().scan_metadata(&engine)? {
            let (underlying_data, selection_vector) = scan_metadata?.scan_files.into_parts();
            let batch: RecordBatch =
                super::kernel::ArrowEngineData::try_from_engine_data(underlying_data)?.into();
            let stats_parsed = stats_parsed_column(&batch)?;
            let min_values = stats_struct_child(stats_parsed, "minValues")?;
            let max_values = stats_struct_child(stats_parsed, "maxValues")?;
            let null_count = stats_struct_child(stats_parsed, "nullCount")?;
            let min_id = stats_struct_child_column(min_values, "id")?;
            let max_id = stats_struct_child_column(max_values, "id")?;
            let null_count_id = stats_struct_child_column(null_count, "id")?;

            boundary.has_id_min_values |= min_values.column_by_name("id").is_some();
            boundary.has_id_max_values |= max_values.column_by_name("id").is_some();
            boundary.has_id_null_count |= null_count.column_by_name("id").is_some();

            for (row_index, selected) in selection_vector.iter().copied().enumerate() {
                if selected {
                    boundary.selected_rows += 1;
                    if !min_id.is_null(row_index)
                        && !max_id.is_null(row_index)
                        && !null_count_id.is_null(row_index)
                    {
                        boundary.selected_stats_rows += 1;
                    } else {
                        boundary.selected_missing_stats_rows += 1;
                    }
                }
            }
        }

        Ok(boundary)
    }

    fn stats_parsed_column(
        batch: &RecordBatch,
    ) -> Result<&StructArray, Box<dyn std::error::Error>> {
        let Some(column) = batch.column_by_name("stats_parsed") else {
            return Err("scan metadata did not expose stats_parsed".into());
        };
        let Some(stats_parsed) = column.as_any().downcast_ref::<StructArray>() else {
            return Err("scan metadata stats_parsed was not a struct array".into());
        };

        Ok(stats_parsed)
    }

    fn stats_struct_child<'a>(
        stats_parsed: &'a StructArray,
        name: &str,
    ) -> Result<&'a StructArray, Box<dyn std::error::Error>> {
        let Some(column) = stats_parsed.column_by_name(name) else {
            return Err(format!("scan metadata stats_parsed did not expose {name}").into());
        };
        let Some(stats_child) = column.as_any().downcast_ref::<StructArray>() else {
            return Err(format!("scan metadata stats_parsed.{name} was not a struct array").into());
        };

        Ok(stats_child)
    }

    fn stats_struct_child_column<'a>(
        stats_child: &'a StructArray,
        name: &str,
    ) -> Result<&'a dyn Array, Box<dyn std::error::Error>> {
        let Some(column) = stats_child.column_by_name(name) else {
            return Err(format!("scan metadata stats_parsed child did not expose {name}").into());
        };

        Ok(column.as_ref())
    }

    fn int8_lit(value: i8) -> Expr {
        Expr::Literal(ScalarValue::Int8(Some(value)), None)
    }

    fn int16_lit(value: i16) -> Expr {
        Expr::Literal(ScalarValue::Int16(Some(value)), None)
    }

    fn int32_lit(value: i32) -> Expr {
        Expr::Literal(ScalarValue::Int32(Some(value)), None)
    }

    fn int64_lit(value: i64) -> Expr {
        Expr::Literal(ScalarValue::Int64(Some(value)), None)
    }

    fn bool_lit(value: bool) -> Expr {
        Expr::Literal(ScalarValue::Boolean(Some(value)), None)
    }

    fn date_lit(value: i32) -> Expr {
        Expr::Literal(ScalarValue::Date32(Some(value)), None)
    }

    fn timestamp_lit(value: i64, timezone: &str) -> Expr {
        Expr::Literal(
            ScalarValue::TimestampMicrosecond(Some(value), Some(timezone.into())),
            None,
        )
    }

    fn timestamp_ntz_lit(value: i64) -> Expr {
        Expr::Literal(ScalarValue::TimestampMicrosecond(Some(value), None), None)
    }

    fn decimal_lit(value: i128) -> Expr {
        Expr::Literal(ScalarValue::Decimal128(Some(value), 10, 2), None)
    }

    fn decimal_lit_with_type(value: i128, precision: u8, scale: i8) -> Expr {
        Expr::Literal(ScalarValue::Decimal128(Some(value), precision, scale), None)
    }

    fn string_lit(value: &str) -> Expr {
        Expr::Literal(ScalarValue::Utf8(Some(value.to_owned())), None)
    }

    fn binary_lit(value: &[u8]) -> Expr {
        Expr::Literal(ScalarValue::Binary(Some(value.to_vec())), None)
    }

    fn large_string_lit(value: &str) -> Expr {
        Expr::Literal(ScalarValue::LargeUtf8(Some(value.to_owned())), None)
    }

    fn float32_lit(value: f32) -> Expr {
        Expr::Literal(ScalarValue::Float32(Some(value)), None)
    }

    fn float64_lit(value: f64) -> Expr {
        Expr::Literal(ScalarValue::Float64(Some(value)), None)
    }

    fn binary_partition_column() -> Expression {
        Expression::Column(ColumnName::new(["payload"]))
    }

    fn binary_partition_value(value: &[u8]) -> Expression {
        Expression::Literal(Scalar::Binary(value.to_vec()))
    }

    fn binary_partition_eq(value: &[u8]) -> DeltaKernelPredicate {
        DeltaKernelPredicate::new(Predicate::eq(
            binary_partition_column(),
            binary_partition_value(value),
        ))
    }

    fn binary_partition_ne(value: &[u8]) -> DeltaKernelPredicate {
        DeltaKernelPredicate::new(Predicate::ne(
            binary_partition_column(),
            binary_partition_value(value),
        ))
    }

    fn timestamp_partition_column() -> Expression {
        Expression::Column(ColumnName::new(["event_ts"]))
    }

    fn timestamp_partition_value(value: i64) -> Expression {
        Expression::Literal(Scalar::Timestamp(value))
    }

    fn timestamp_partition_eq(value: i64) -> DeltaKernelPredicate {
        DeltaKernelPredicate::new(Predicate::eq(
            timestamp_partition_column(),
            timestamp_partition_value(value),
        ))
    }

    fn timestamp_partition_ne(value: i64) -> DeltaKernelPredicate {
        DeltaKernelPredicate::new(Predicate::ne(
            timestamp_partition_column(),
            timestamp_partition_value(value),
        ))
    }

    fn timestamp_ntz_partition_column() -> Expression {
        Expression::Column(ColumnName::new(["event_ts_ntz"]))
    }

    fn timestamp_ntz_partition_value(value: i64) -> Expression {
        Expression::Literal(Scalar::TimestampNtz(value))
    }

    fn timestamp_ntz_partition_eq(value: i64) -> DeltaKernelPredicate {
        DeltaKernelPredicate::new(Predicate::eq(
            timestamp_ntz_partition_column(),
            timestamp_ntz_partition_value(value),
        ))
    }

    fn timestamp_ntz_partition_ne(value: i64) -> DeltaKernelPredicate {
        DeltaKernelPredicate::new(Predicate::ne(
            timestamp_ntz_partition_column(),
            timestamp_ntz_partition_value(value),
        ))
    }

    fn assert_invalid_integer_partition_error(
        result: Result<Vec<String>, Box<dyn std::error::Error>>,
    ) -> Result<(), Box<dyn std::error::Error>> {
        let error = match result {
            Ok(paths) => {
                return Err(
                    format!("invalid integer metadata should fail kernel scan: {paths:?}").into(),
                );
            }
            Err(error) => error,
        };
        let message = error.to_string();
        let debug_message = format!("{error:?}");
        assert!(message.contains("not-an-integer"));
        assert!(debug_message.contains("Primitive(Long)"));

        Ok(())
    }

    fn assert_invalid_boolean_partition_error(
        result: Result<Vec<String>, Box<dyn std::error::Error>>,
    ) -> Result<(), Box<dyn std::error::Error>> {
        let error = match result {
            Ok(paths) => {
                return Err(
                    format!("invalid boolean metadata should fail kernel scan: {paths:?}").into(),
                );
            }
            Err(error) => error,
        };
        let message = error.to_string();
        let debug_message = format!("{error:?}");
        assert!(message.contains("not-a-boolean"));
        assert!(debug_message.contains("Primitive(Boolean)"));

        Ok(())
    }

    fn assert_invalid_date_partition_error(
        result: Result<Vec<String>, Box<dyn std::error::Error>>,
    ) -> Result<(), Box<dyn std::error::Error>> {
        let error = match result {
            Ok(paths) => {
                return Err(
                    format!("invalid date metadata should fail kernel scan: {paths:?}").into(),
                );
            }
            Err(error) => error,
        };
        let message = error.to_string();
        let debug_message = format!("{error:?}");
        assert!(message.contains("not-a-date"));
        assert!(debug_message.contains("Primitive(Date)"));

        Ok(())
    }

    fn assert_invalid_decimal_partition_error(
        result: Result<Vec<String>, Box<dyn std::error::Error>>,
    ) -> Result<(), Box<dyn std::error::Error>> {
        let error = match result {
            Ok(paths) => {
                return Err(
                    format!("invalid decimal metadata should fail kernel scan: {paths:?}").into(),
                );
            }
            Err(error) => error,
        };
        let message = error.to_string();
        let debug_message = format!("{error:?}");
        assert!(
            message.contains("not-a-decimal")
                || message.contains("123.450")
                || message.contains("Could not parse int")
                || debug_message.contains("ParseIntError")
                || debug_message.contains("InvalidDecimal"),
            "{message}\n{debug_message}"
        );
        assert!(
            debug_message.contains("Decimal") || debug_message.contains("ParseIntError"),
            "{debug_message}"
        );

        Ok(())
    }

    fn assert_invalid_floating_partition_error(
        result: Result<Vec<String>, Box<dyn std::error::Error>>,
    ) -> Result<(), Box<dyn std::error::Error>> {
        let error = match result {
            Ok(paths) => {
                return Err(format!(
                    "invalid floating metadata should fail kernel scan: {paths:?}"
                )
                .into());
            }
            Err(error) => error,
        };
        let message = error.to_string();
        let debug_message = format!("{error:?}");
        assert!(
            message.contains("not-a-float")
                || message.contains("not-a-double")
                || debug_message.contains("InvalidFloat")
                || debug_message.contains("ParseFloatError"),
            "{message}\n{debug_message}"
        );

        Ok(())
    }

    fn assert_invalid_timestamp_partition_error(
        result: Result<Vec<String>, Box<dyn std::error::Error>>,
    ) -> Result<(), Box<dyn std::error::Error>> {
        let error = match result {
            Ok(paths) => {
                return Err(format!(
                    "invalid timestamp metadata should fail kernel scan: {paths:?}"
                )
                .into());
            }
            Err(error) => error,
        };
        let message = error.to_string();
        let debug_message = format!("{error:?}");
        assert!(
            message.contains("not-a-timestamp")
                || debug_message.contains("not-a-timestamp")
                || debug_message.contains("Timestamp"),
            "{message}\n{debug_message}"
        );

        Ok(())
    }

    fn assert_invalid_timestamp_ntz_partition_error(
        result: Result<Vec<String>, Box<dyn std::error::Error>>,
    ) -> Result<(), Box<dyn std::error::Error>> {
        let error = match result {
            Ok(paths) => {
                return Err(format!(
                    "invalid timestamp_ntz metadata should fail kernel scan: {paths:?}"
                )
                .into());
            }
            Err(error) => error,
        };
        let message = error.to_string();
        let debug_message = format!("{error:?}");
        assert!(
            message.contains("not-a-timestamp")
                || debug_message.contains("not-a-timestamp")
                || debug_message.contains("TimestampNtz"),
            "{message}\n{debug_message}"
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_uses_official_delta_kernel_0_23_0() {
        let manifest = include_str!("../../Cargo.toml");
        let lockfile = include_str!("../../../../Cargo.lock");

        assert!(manifest.contains(r#"delta_kernel = { version = "0.23.0""#));
        assert!(lockfile.contains(
            "name = \"delta_kernel\"\nversion = \"0.23.0\"\nsource = \"registry+https://github.com/rust-lang/crates.io-index\""
        ));
        assert!(!manifest.contains("deltalake"));
        assert!(!manifest.contains("buoyant_kernel"));
    }

    #[test]
    fn kernel_predicated_scan_prunes_files() -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-predicated-scan-files",
            PARTITIONED_METADATA_JSON,
            &[
                partitioned_add_json("region-us-west.parquet", r#"{"region":"us-west"}"#),
                partitioned_add_json("region-us-east.parquet", r#"{"region":"us-east"}"#),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;
        let unfiltered_scan = build_projected_delta_scan(&source, None)?;
        let predicate = datafusion_expr_to_kernel_predicate(&col("region").eq(lit("us-west")))?;
        let predicated_scan =
            build_projected_predicated_delta_scan(&source, None, Some(predicate))?;

        assert_eq!(
            kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
            vec!["region-us-east.parquet", "region-us-west.parquet"]
        );
        assert_eq!(
            kernel_scan_file_paths(&predicated_scan, source.table_uri())?,
            vec!["region-us-west.parquet"]
        );

        Ok(())
    }

    #[test]
    fn kernel_stats_scan_metadata_exposes_parsed_file_stats_without_data_file_reads()
    -> Result<(), Box<dyn std::error::Error>> {
        let (table, source) = kernel_data_stats_characterization_source()?;
        let stats_scan = build_projected_predicated_stats_delta_scan(&source, None)?;

        assert!(!table.path.join("id-impossible.parquet").exists());
        assert!(!table.path.join("id-possible.parquet").exists());
        assert!(!table.path.join("id-missing-stats.parquet").exists());
        assert_eq!(
            kernel_scan_file_paths(&stats_scan, source.table_uri())?,
            vec![
                "id-impossible.parquet",
                "id-missing-stats.parquet",
                "id-possible.parquet",
            ]
        );
        assert_eq!(
            kernel_stats_metadata_boundary(&stats_scan, source.table_uri())?,
            KernelStatsMetadataBoundary {
                selected_rows: 3,
                selected_stats_rows: 2,
                selected_missing_stats_rows: 1,
                has_id_min_values: true,
                has_id_max_values: true,
                has_id_null_count: true,
            }
        );

        Ok(())
    }

    #[test]
    fn kernel_data_column_stats_pruning_keeps_possible_and_missing_stats_files()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("id").gt(int32_lit(100)))?,
            vec!["id-missing-stats.parquet", "id-possible.parquet"]
        );

        Ok(())
    }

    #[test]
    fn kernel_data_column_stats_pruning_can_return_empty_when_all_stats_are_impossible()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_all_sufficient_data_stats_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("id").gt(int32_lit(100)))?,
            Vec::<String>::new()
        );

        Ok(())
    }

    #[test]
    fn kernel_boolean_data_column_stats_pruning_documents_min_max_boundary()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_boolean_data_stats_characterization_source()?;

        assert_boolean_stats_min_max_error(kernel_predicated_stats_file_paths(
            &source,
            &col("is_current").eq(bool_lit(true)),
        ))?;
        assert_boolean_stats_min_max_error(kernel_predicated_stats_file_paths(
            &source,
            &col("is_current").eq(bool_lit(false)),
        ))?;
        assert_boolean_stats_min_max_error(kernel_predicated_stats_file_paths(
            &source,
            &col("is_current").not_eq(bool_lit(true)),
        ))?;
        assert_boolean_stats_min_max_error(kernel_predicated_stats_file_paths(
            &source,
            &col("is_current").not_eq(bool_lit(false)),
        ))?;

        Ok(())
    }

    #[test]
    fn kernel_boolean_data_column_stats_pruning_documents_null_checks()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_boolean_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("is_current").is_null())?,
            vec![
                "boolean-all-null.parquet",
                "boolean-false-with-null.parquet",
                "boolean-missing-stats.parquet",
                "boolean-true-with-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("is_current").is_not_null())?,
            vec![
                "boolean-false-only.parquet",
                "boolean-false-with-null.parquet",
                "boolean-missing-stats.parquet",
                "boolean-mixed.parquet",
                "boolean-true-only.parquet",
                "boolean-true-with-null.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_boolean_data_column_stats_pruning_keeps_partial_null_counts_uncertain()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_partial_boolean_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("is_current").is_null())?,
            vec![
                "boolean-missing-null-count.parquet",
                "boolean-missing-stats.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("is_current").is_not_null())?,
            vec![
                "boolean-counts-only.parquet",
                "boolean-max-only-true.parquet",
                "boolean-min-only-false.parquet",
                "boolean-missing-null-count.parquet",
                "boolean-missing-stats.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_date_data_column_stats_pruning_documents_comparisons()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_date_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("event_date").gt(date_lit(19_782)))?,
            vec![
                "date-missing-stats.parquet",
                "date-new-year-only.parquet",
                "date-new-year-with-null.parquet",
                "date-range.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("event_date").gt_eq(date_lit(20_454)),
            )?,
            vec![
                "date-missing-stats.parquet",
                "date-new-year-only.parquet",
                "date-new-year-with-null.parquet",
                "date-range.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("event_date").lt_eq(date_lit(-1)),)?,
            vec!["date-missing-stats.parquet", "date-pre-epoch-only.parquet"]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &date_lit(20_454).gt(col("event_date")))?,
            vec![
                "date-leap-only.parquet",
                "date-missing-stats.parquet",
                "date-pre-epoch-only.parquet",
                "date-range.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_date_data_column_stats_pruning_documents_equality_and_not_equals()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_date_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("event_date").eq(date_lit(20_454)))?,
            vec![
                "date-missing-stats.parquet",
                "date-new-year-only.parquet",
                "date-new-year-with-null.parquet",
                "date-range.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("event_date").not_eq(date_lit(20_454)),
            )?,
            vec![
                "date-leap-only.parquet",
                "date-missing-stats.parquet",
                "date-pre-epoch-only.parquet",
                "date-range.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_date_data_column_stats_pruning_documents_null_checks()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_date_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("event_date").is_null())?,
            vec![
                "date-all-null.parquet",
                "date-missing-stats.parquet",
                "date-new-year-with-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("event_date").is_not_null())?,
            vec![
                "date-leap-only.parquet",
                "date-missing-stats.parquet",
                "date-new-year-only.parquet",
                "date-new-year-with-null.parquet",
                "date-pre-epoch-only.parquet",
                "date-range.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_date_data_column_stats_pruning_documents_partial_stats_boundary()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_partial_date_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("event_date").gt(date_lit(19_782)))?,
            vec![
                "date-counts-only.parquet",
                "date-min-only-high.parquet",
                "date-missing-null-count.parquet",
                "date-missing-stats.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("event_date").is_null())?,
            vec![
                "date-missing-null-count.parquet",
                "date-missing-stats.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("event_date").is_not_null())?,
            vec![
                "date-counts-only.parquet",
                "date-max-only-low.parquet",
                "date-min-only-high.parquet",
                "date-missing-null-count.parquet",
                "date-missing-stats.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_decimal_data_column_stats_pruning_documents_comparisons()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_decimal_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("amount").gt(decimal_lit(200)))?,
            vec![
                "decimal-large-only.parquet",
                "decimal-missing-stats.parquet",
                "decimal-range.parquet",
                "decimal-ten-only.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("amount").gt_eq(decimal_lit(1_000)))?,
            vec![
                "decimal-large-only.parquet",
                "decimal-missing-stats.parquet",
                "decimal-range.parquet",
                "decimal-ten-only.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("amount").lt_eq(decimal_lit(-123)))?,
            vec![
                "decimal-missing-stats.parquet",
                "decimal-negative-only.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &decimal_lit(1_000).gt(col("amount")))?,
            vec![
                "decimal-missing-stats.parquet",
                "decimal-negative-only.parquet",
                "decimal-range.parquet",
                "decimal-two-only.parquet",
                "decimal-two-with-null.parquet",
                "decimal-zero-only.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_decimal_data_column_stats_pruning_documents_equality_and_not_equals()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_decimal_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("amount").eq(decimal_lit(200)))?,
            vec![
                "decimal-missing-stats.parquet",
                "decimal-range.parquet",
                "decimal-two-only.parquet",
                "decimal-two-with-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("amount").not_eq(decimal_lit(200)))?,
            vec![
                "decimal-large-only.parquet",
                "decimal-missing-stats.parquet",
                "decimal-negative-only.parquet",
                "decimal-range.parquet",
                "decimal-ten-only.parquet",
                "decimal-zero-only.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_decimal_data_column_stats_pruning_documents_null_checks()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_decimal_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("amount").is_null())?,
            vec![
                "decimal-all-null.parquet",
                "decimal-missing-stats.parquet",
                "decimal-two-with-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("amount").is_not_null())?,
            vec![
                "decimal-large-only.parquet",
                "decimal-missing-stats.parquet",
                "decimal-negative-only.parquet",
                "decimal-range.parquet",
                "decimal-ten-only.parquet",
                "decimal-two-only.parquet",
                "decimal-two-with-null.parquet",
                "decimal-zero-only.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_decimal_data_column_stats_pruning_documents_partial_stats_boundary()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_partial_decimal_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("amount").gt(decimal_lit(200)))?,
            vec![
                "decimal-counts-only.parquet",
                "decimal-min-only-high.parquet",
                "decimal-missing-null-count.parquet",
                "decimal-missing-stats.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("amount").is_null())?,
            vec![
                "decimal-missing-null-count.parquet",
                "decimal-missing-stats.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("amount").is_not_null())?,
            vec![
                "decimal-counts-only.parquet",
                "decimal-max-only-low.parquet",
                "decimal-min-only-high.parquet",
                "decimal-missing-null-count.parquet",
                "decimal-missing-stats.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_decimal_data_column_stats_pruning_documents_precision_and_scale_boundary()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_decimal_data_stats_characterization_source()?;

        for (description, literal) in [
            ("scale mismatch", decimal_lit_with_type(2_000, 11, 3)),
            ("precision mismatch", decimal_lit_with_type(200, 11, 2)),
        ] {
            let error = kernel_predicated_stats_file_paths(&source, &col("amount").eq(literal))
                .err()
                .ok_or_else(|| {
                    format!("{description} decimal predicate should fail kernel scan")
                })?;
            let message = error.to_string();
            let debug_message = format!("{error:?}");
            assert!(
                message.contains("Invalid comparison operation")
                    || debug_message.contains("Invalid comparison operation"),
                "{message}\n{debug_message}"
            );
        }

        Ok(())
    }

    #[test]
    fn kernel_binary_data_column_stats_pruning_documents_equality_and_empty_boundary()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_binary_data_stats_characterization_source()?;

        assert_binary_stats_min_max_error(kernel_predicated_stats_file_paths(
            &source,
            &col("payload").eq(binary_lit(b"hello")),
        ))?;
        assert_binary_stats_min_max_error(kernel_predicated_stats_file_paths(
            &source,
            &col("payload").not_eq(binary_lit(b"hello")),
        ))?;
        assert_unsupported_literal_error(kernel_predicated_stats_file_paths(
            &source,
            &col("payload").eq(binary_lit(b"")),
        ))?;
        assert_unsupported_literal_error(kernel_predicated_stats_file_paths(
            &source,
            &col("payload").not_eq(binary_lit(b"")),
        ))?;

        Ok(())
    }

    #[test]
    fn kernel_binary_data_column_stats_pruning_documents_ordering()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_binary_data_stats_characterization_source()?;

        assert_binary_stats_min_max_error(kernel_predicated_stats_file_paths(
            &source,
            &col("payload").gt(binary_lit(b"hello")),
        ))?;
        assert_binary_stats_min_max_error(kernel_predicated_stats_file_paths(
            &source,
            &col("payload").lt(binary_lit(b"hello")),
        ))?;
        assert_binary_stats_min_max_error(kernel_predicated_stats_file_paths(
            &source,
            &binary_lit(b"hello").gt(col("payload")),
        ))?;

        Ok(())
    }

    #[test]
    fn kernel_binary_data_column_stats_pruning_documents_null_checks()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_binary_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("payload").is_null())?,
            vec![
                "binary-all-null.parquet",
                "binary-missing-stats.parquet",
                "binary-with-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("payload").is_not_null())?,
            vec![
                "binary-HELLO.parquet",
                "binary-empty.parquet",
                "binary-hello.parquet",
                "binary-missing-stats.parquet",
                "binary-range.parquet",
                "binary-special.parquet",
                "binary-with-null.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_binary_data_column_stats_pruning_documents_partial_stats_boundary()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_partial_binary_data_stats_characterization_source()?;

        assert_binary_stats_min_max_error(kernel_predicated_stats_file_paths(
            &source,
            &col("payload").gt(binary_lit(b"hello")),
        ))?;
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("payload").is_null())?,
            vec![
                "binary-missing-null-count.parquet",
                "binary-missing-stats.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("payload").is_not_null())?,
            vec![
                "binary-counts-only.parquet",
                "binary-max-only-low.parquet",
                "binary-min-only-high.parquet",
                "binary-missing-null-count.parquet",
                "binary-missing-stats.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_floating_data_column_stats_pruning_documents_comparisons()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_floating_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("float_score").gt(float32_lit(1.5)))?,
            vec![
                "floating-missing-stats.parquet",
                "floating-range.parquet",
                "floating-ten.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("double_score").lt(float64_lit(0.0)),)?,
            vec![
                "floating-missing-stats.parquet",
                "floating-neg-zero.parquet",
                "floating-neg.parquet",
                "floating-range.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &float64_lit(0.0).gt(col("double_score")))?,
            vec![
                "floating-missing-stats.parquet",
                "floating-neg-zero.parquet",
                "floating-neg.parquet",
                "floating-range.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("float_score").lt(float32_lit(10.0)))?,
            vec![
                "floating-missing-stats.parquet",
                "floating-neg-zero.parquet",
                "floating-neg.parquet",
                "floating-one-with-null.parquet",
                "floating-one.parquet",
                "floating-pos-zero.parquet",
                "floating-range.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_floating_data_column_stats_pruning_documents_equality_and_signed_zero()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_floating_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("float_score").eq(float32_lit(1.5)))?,
            vec![
                "floating-missing-stats.parquet",
                "floating-one-with-null.parquet",
                "floating-one.parquet",
                "floating-range.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("float_score").not_eq(float32_lit(1.5)),
            )?,
            vec![
                "floating-missing-stats.parquet",
                "floating-neg-zero.parquet",
                "floating-neg.parquet",
                "floating-pos-zero.parquet",
                "floating-range.parquet",
                "floating-ten.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("float_score").eq(float32_lit(-0.0)))?,
            vec![
                "floating-missing-stats.parquet",
                "floating-neg-zero.parquet",
                "floating-range.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("float_score").eq(float32_lit(0.0)))?,
            vec![
                "floating-missing-stats.parquet",
                "floating-pos-zero.parquet",
                "floating-range.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_floating_data_column_stats_pruning_documents_signed_zero_operator_boundary()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_floating_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("float_score").lt(float32_lit(0.0)))?,
            vec![
                "floating-missing-stats.parquet",
                "floating-neg-zero.parquet",
                "floating-neg.parquet",
                "floating-range.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("float_score").lt_eq(float32_lit(0.0)),
            )?,
            vec![
                "floating-missing-stats.parquet",
                "floating-neg-zero.parquet",
                "floating-neg.parquet",
                "floating-pos-zero.parquet",
                "floating-range.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("float_score").gt(float32_lit(0.0)))?,
            vec![
                "floating-missing-stats.parquet",
                "floating-one-with-null.parquet",
                "floating-one.parquet",
                "floating-range.parquet",
                "floating-ten.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("float_score").gt_eq(float32_lit(0.0)),
            )?,
            vec![
                "floating-missing-stats.parquet",
                "floating-one-with-null.parquet",
                "floating-one.parquet",
                "floating-pos-zero.parquet",
                "floating-range.parquet",
                "floating-ten.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("float_score").not_eq(float32_lit(0.0)),
            )?,
            vec![
                "floating-missing-stats.parquet",
                "floating-neg-zero.parquet",
                "floating-neg.parquet",
                "floating-one-with-null.parquet",
                "floating-one.parquet",
                "floating-range.parquet",
                "floating-ten.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_floating_data_column_stats_pruning_documents_null_checks()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_floating_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("float_score").is_null())?,
            vec![
                "floating-all-null.parquet",
                "floating-missing-stats.parquet",
                "floating-one-with-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("double_score").is_not_null())?,
            vec![
                "floating-missing-stats.parquet",
                "floating-neg-zero.parquet",
                "floating-neg.parquet",
                "floating-one-with-null.parquet",
                "floating-one.parquet",
                "floating-pos-zero.parquet",
                "floating-range.parquet",
                "floating-ten.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_floating_data_column_stats_pruning_documents_partial_stats_boundary()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_partial_floating_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("float_score").gt(float32_lit(1.0)))?,
            vec![
                "floating-counts-only.parquet",
                "floating-min-only-high.parquet",
                "floating-missing-null-count.parquet",
                "floating-missing-stats.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("float_score").is_null())?,
            vec![
                "floating-missing-null-count.parquet",
                "floating-missing-stats.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("float_score").is_not_null())?,
            vec![
                "floating-counts-only.parquet",
                "floating-max-only-low.parquet",
                "floating-min-only-high.parquet",
                "floating-missing-null-count.parquet",
                "floating-missing-stats.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_floating_data_column_stats_pruning_documents_nonfinite_stats_boundary()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_nonfinite_floating_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("float_score").gt(float32_lit(1.0)))?,
            vec![
                "floating-inf.parquet",
                "floating-missing-stats.parquet",
                "floating-nan.parquet",
                "floating-valid.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("double_score").lt(float64_lit(0.0)))?,
            vec!["floating-missing-stats.parquet", "floating-neg-inf.parquet"]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("float_score").eq(float32_lit(1.5)),)?,
            vec!["floating-missing-stats.parquet", "floating-valid.parquet"]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("float_score").not_eq(float32_lit(1.5)),
            )?,
            vec![
                "floating-inf.parquet",
                "floating-missing-stats.parquet",
                "floating-nan.parquet",
                "floating-neg-inf.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_string_data_column_stats_pruning_documents_comparisons()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_string_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("customer_name").gt(string_lit("m")),)?,
            vec![
                "string-missing-stats.parquet",
                "string-range.parquet",
                "string-zed-only.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("customer_name").gt_eq(string_lit("morgan")),
            )?,
            vec![
                "string-missing-stats.parquet",
                "string-range.parquet",
                "string-zed-only.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("customer_name").lt_eq(string_lit("Alice")),
            )?,
            vec![
                "string-empty-only.parquet",
                "string-missing-stats.parquet",
                "string-mixed-case-only.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &string_lit("m").gt(col("customer_name")),)?,
            vec![
                "string-alice-only.parquet",
                "string-alice-with-null.parquet",
                "string-bob-only.parquet",
                "string-empty-only.parquet",
                "string-missing-stats.parquet",
                "string-mixed-case-only.parquet",
                "string-range.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_string_data_column_stats_pruning_documents_equality_and_not_equals()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_string_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("customer_name").eq(string_lit("alice")),
            )?,
            vec![
                "string-alice-only.parquet",
                "string-alice-with-null.parquet",
                "string-missing-stats.parquet",
                "string-range.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("customer_name").not_eq(string_lit("alice")),
            )?,
            vec![
                "string-bob-only.parquet",
                "string-empty-only.parquet",
                "string-missing-stats.parquet",
                "string-mixed-case-only.parquet",
                "string-range.parquet",
                "string-zed-only.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_string_data_column_stats_pruning_documents_null_checks()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_string_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("customer_name").is_null())?,
            vec![
                "string-alice-with-null.parquet",
                "string-all-null.parquet",
                "string-missing-stats.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("customer_name").is_not_null())?,
            vec![
                "string-alice-only.parquet",
                "string-alice-with-null.parquet",
                "string-bob-only.parquet",
                "string-empty-only.parquet",
                "string-missing-stats.parquet",
                "string-mixed-case-only.parquet",
                "string-range.parquet",
                "string-zed-only.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_string_data_column_stats_pruning_documents_partial_stats_boundary()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_partial_string_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("customer_name").gt(string_lit("m")),)?,
            vec![
                "string-counts-only.parquet",
                "string-min-only-morgan.parquet",
                "string-missing-null-count.parquet",
                "string-missing-stats.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("customer_name").is_null())?,
            vec![
                "string-missing-null-count.parquet",
                "string-missing-stats.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("customer_name").is_not_null())?,
            vec![
                "string-counts-only.parquet",
                "string-max-only-alice.parquet",
                "string-min-only-morgan.parquet",
                "string-missing-null-count.parquet",
                "string-missing-stats.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_string_data_column_stats_pruning_documents_large_utf8_literal_boundary()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_string_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("customer_name").eq(large_string_lit("alice")),
            )?,
            vec![
                "string-alice-only.parquet",
                "string-alice-with-null.parquet",
                "string-missing-stats.parquet",
                "string-range.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("customer_name").gt(large_string_lit("m")),
            )?,
            vec![
                "string-missing-stats.parquet",
                "string-range.parquet",
                "string-zed-only.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_string_data_column_stats_pruning_documents_non_ascii_ordering()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_non_ascii_string_data_stats_characterization_source()?;
        let eclair = string_lit("\u{00e9}clair");

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("customer_name").eq(eclair.clone()),)?,
            vec!["string-eclair.parquet", "string-missing-stats.parquet"]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("customer_name").eq(large_string_lit("\u{00e9}clair")),
            )?,
            vec!["string-eclair.parquet", "string-missing-stats.parquet"]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("customer_name").gt_eq(eclair.clone()),
            )?,
            vec![
                "string-eclair.parquet",
                "string-emile.parquet",
                "string-missing-stats.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("customer_name").lt(eclair.clone()),)?,
            vec![
                "string-ascii-cafe.parquet",
                "string-ascii-zulu.parquet",
                "string-missing-stats.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("customer_name").gt(eclair),)?,
            vec!["string-emile.parquet", "string-missing-stats.parquet"]
        );

        Ok(())
    }

    #[test]
    fn kernel_timestamp_data_column_stats_pruning_documents_comparisons()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_timestamp_data_stats_characterization_source()?;
        let low = 1_767_225_599_999_999_i64;
        let target = 1_767_225_600_123_456_i64;
        let high = 1_767_225_600_123_457_i64;

        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("event_ts").lt(timestamp_lit(target, "UTC")),
            )?,
            vec![
                "timestamp-low-only.parquet",
                "timestamp-missing-stats.parquet",
                "timestamp-pre-epoch-only.parquet",
                "timestamp-range.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("event_ts").gt_eq(timestamp_lit(target, "UTC")),
            )?,
            vec![
                "timestamp-high-only.parquet",
                "timestamp-missing-stats.parquet",
                "timestamp-range.parquet",
                "timestamp-target-only.parquet",
                "timestamp-target-with-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &timestamp_lit(high, "UTC").gt(col("event_ts")),
            )?,
            vec![
                "timestamp-low-only.parquet",
                "timestamp-missing-stats.parquet",
                "timestamp-pre-epoch-only.parquet",
                "timestamp-range.parquet",
                "timestamp-target-only.parquet",
                "timestamp-target-with-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("event_ts").eq(timestamp_lit(low, "UTC")),
            )?,
            vec![
                "timestamp-low-only.parquet",
                "timestamp-missing-stats.parquet",
                "timestamp-range.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("event_ts").not_eq(timestamp_lit(target, "UTC")),
            )?,
            vec![
                "timestamp-high-only.parquet",
                "timestamp-low-only.parquet",
                "timestamp-missing-stats.parquet",
                "timestamp-pre-epoch-only.parquet",
                "timestamp-range.parquet",
                "timestamp-target-only.parquet",
                "timestamp-target-with-null.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_timestamp_data_column_stats_pruning_documents_null_checks()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_timestamp_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("event_ts").is_null())?,
            vec![
                "timestamp-all-null.parquet",
                "timestamp-missing-stats.parquet",
                "timestamp-target-with-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("event_ts").is_not_null())?,
            vec![
                "timestamp-high-only.parquet",
                "timestamp-low-only.parquet",
                "timestamp-missing-stats.parquet",
                "timestamp-pre-epoch-only.parquet",
                "timestamp-range.parquet",
                "timestamp-target-only.parquet",
                "timestamp-target-with-null.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_timestamp_data_column_stats_pruning_documents_partial_stats_boundary()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_partial_timestamp_data_stats_characterization_source()?;
        let low = 1_767_225_599_999_999_i64;

        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("event_ts").gt(timestamp_lit(low, "UTC")),
            )?,
            vec![
                "timestamp-counts-only.parquet",
                "timestamp-max-only-low.parquet",
                "timestamp-min-only-target.parquet",
                "timestamp-missing-null-count.parquet",
                "timestamp-missing-stats.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("event_ts").is_null())?,
            vec![
                "timestamp-missing-null-count.parquet",
                "timestamp-missing-stats.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("event_ts").is_not_null())?,
            vec![
                "timestamp-counts-only.parquet",
                "timestamp-max-only-low.parquet",
                "timestamp-min-only-target.parquet",
                "timestamp-missing-null-count.parquet",
                "timestamp-missing-stats.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_timestamp_ntz_data_column_stats_pruning_documents_comparisons()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_timestamp_ntz_data_stats_characterization_source()?;
        let low = 1_767_225_599_999_999_i64;
        let target = 1_767_225_600_123_456_i64;
        let high = 1_767_225_600_123_457_i64;

        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("event_ts_ntz").lt(timestamp_ntz_lit(target)),
            )?,
            vec![
                "timestamp-ntz-low-only.parquet",
                "timestamp-ntz-missing-stats.parquet",
                "timestamp-ntz-pre-epoch-only.parquet",
                "timestamp-ntz-range.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("event_ts_ntz").gt_eq(timestamp_ntz_lit(target)),
            )?,
            vec![
                "timestamp-ntz-high-only.parquet",
                "timestamp-ntz-missing-stats.parquet",
                "timestamp-ntz-range.parquet",
                "timestamp-ntz-target-only.parquet",
                "timestamp-ntz-target-with-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &timestamp_ntz_lit(high).gt(col("event_ts_ntz")),
            )?,
            vec![
                "timestamp-ntz-low-only.parquet",
                "timestamp-ntz-missing-stats.parquet",
                "timestamp-ntz-pre-epoch-only.parquet",
                "timestamp-ntz-range.parquet",
                "timestamp-ntz-target-only.parquet",
                "timestamp-ntz-target-with-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("event_ts_ntz").eq(timestamp_ntz_lit(low)),
            )?,
            vec![
                "timestamp-ntz-low-only.parquet",
                "timestamp-ntz-missing-stats.parquet",
                "timestamp-ntz-range.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("event_ts_ntz").not_eq(timestamp_ntz_lit(target)),
            )?,
            vec![
                "timestamp-ntz-high-only.parquet",
                "timestamp-ntz-low-only.parquet",
                "timestamp-ntz-missing-stats.parquet",
                "timestamp-ntz-pre-epoch-only.parquet",
                "timestamp-ntz-range.parquet",
                "timestamp-ntz-target-only.parquet",
                "timestamp-ntz-target-with-null.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_timestamp_ntz_data_column_stats_pruning_documents_null_checks()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_timestamp_ntz_data_stats_characterization_source()?;

        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("event_ts_ntz").is_null())?,
            vec![
                "timestamp-ntz-all-null.parquet",
                "timestamp-ntz-missing-stats.parquet",
                "timestamp-ntz-target-with-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("event_ts_ntz").is_not_null())?,
            vec![
                "timestamp-ntz-high-only.parquet",
                "timestamp-ntz-low-only.parquet",
                "timestamp-ntz-missing-stats.parquet",
                "timestamp-ntz-pre-epoch-only.parquet",
                "timestamp-ntz-range.parquet",
                "timestamp-ntz-target-only.parquet",
                "timestamp-ntz-target-with-null.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_timestamp_ntz_data_column_stats_pruning_documents_partial_stats_boundary()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_partial_timestamp_ntz_data_stats_characterization_source()?;
        let low = 1_767_225_599_999_999_i64;

        assert_eq!(
            kernel_predicated_stats_file_paths(
                &source,
                &col("event_ts_ntz").gt(timestamp_ntz_lit(low)),
            )?,
            vec![
                "timestamp-ntz-counts-only.parquet",
                "timestamp-ntz-max-only-low.parquet",
                "timestamp-ntz-min-only-target.parquet",
                "timestamp-ntz-missing-null-count.parquet",
                "timestamp-ntz-missing-stats.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("event_ts_ntz").is_null())?,
            vec![
                "timestamp-ntz-missing-null-count.parquet",
                "timestamp-ntz-missing-stats.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_stats_file_paths(&source, &col("event_ts_ntz").is_not_null())?,
            vec![
                "timestamp-ntz-counts-only.parquet",
                "timestamp-ntz-max-only-low.parquet",
                "timestamp-ntz-min-only-target.parquet",
                "timestamp-ntz-missing-null-count.parquet",
                "timestamp-ntz-missing-stats.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_predicated_stats_scan_exposes_stats_for_surviving_files()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_data_stats_characterization_source()?;
        let predicate = datafusion_expr_to_kernel_predicate(&col("id").gt(int32_lit(100)))?;
        let stats_scan = build_projected_predicated_stats_delta_scan(&source, Some(predicate))?;

        assert_eq!(
            kernel_scan_file_paths(&stats_scan, source.table_uri())?,
            vec!["id-missing-stats.parquet", "id-possible.parquet"]
        );
        assert_eq!(
            kernel_stats_metadata_boundary(&stats_scan, source.table_uri())?,
            KernelStatsMetadataBoundary {
                selected_rows: 2,
                selected_stats_rows: 1,
                selected_missing_stats_rows: 1,
                has_id_min_values: true,
                has_id_max_values: true,
                has_id_null_count: true,
            }
        );

        Ok(())
    }

    #[test]
    fn kernel_stats_pruning_preserves_surviving_dv_metadata_boundary()
    -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_protocol_metadata_and_adds(
            "kernel-dv-data-stats-boundary",
            DELETION_VECTOR_PROTOCOL_JSON,
            METADATA_JSON,
            &[
                dv_add_json_with_id_stats("id-dv-impossible.parquet", 10, 1, 50, 0),
                dv_add_json_with_id_stats("id-dv-possible.parquet", 10, 101, 150, 0),
                add_json_with_id_stats("id-plain-impossible.parquet", 10, 1, 50, 0),
                add_json("id-plain-missing-stats.parquet"),
                partitioned_dv_add_json("id-dv-missing-stats.parquet", r#"{}"#),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;
        let predicate = datafusion_expr_to_kernel_predicate(&col("id").gt(int32_lit(100)))?;
        let stats_scan = build_projected_predicated_stats_delta_scan(&source, Some(predicate))?;

        assert_eq!(
            kernel_scan_file_boundaries(&stats_scan, source.table_uri())?,
            vec![
                KernelScanFileBoundary {
                    path: "id-dv-missing-stats.parquet".to_owned(),
                    has_deletion_vector: true,
                },
                KernelScanFileBoundary {
                    path: "id-dv-possible.parquet".to_owned(),
                    has_deletion_vector: true,
                },
                KernelScanFileBoundary {
                    path: "id-plain-missing-stats.parquet".to_owned(),
                    has_deletion_vector: false,
                },
            ]
        );

        Ok(())
    }

    #[test]
    fn delta_table_format_production_paths_do_not_load_dv_payloads()
    -> Result<(), Box<dyn std::error::Error>> {
        let source_root = Path::new(env!("CARGO_MANIFEST_DIR"))
            .join("src")
            .join("table_formats");
        let production_source = [
            source_root.join("delta.rs"),
            source_root.join("delta").join("kernel.rs"),
        ]
        .iter()
        .map(fs::read_to_string)
        .collect::<Result<Vec<_>, _>>()?
        .into_iter()
        .map(|source| {
            source
                .split("\n#[cfg(test)]")
                .next()
                .unwrap_or(source.as_str())
                .to_owned()
        })
        .collect::<Vec<_>>()
        .join("\n");

        assert!(!production_source.contains("get_selection_vector"));
        assert!(!production_source.contains("get_row_indexes"));

        Ok(())
    }

    #[test]
    fn kernel_predicated_scan_preserves_dv_metadata_boundary()
    -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_protocol_metadata_and_adds(
            "kernel-dv-partition-boundary",
            DELETION_VECTOR_PROTOCOL_JSON,
            PARTITIONED_METADATA_JSON,
            &[
                partitioned_dv_add_json("region-us-west-dv.parquet", r#"{"region":"us-west"}"#),
                partitioned_add_json("region-us-west-plain.parquet", r#"{"region":"us-west"}"#),
                partitioned_dv_add_json("region-us-east-dv.parquet", r#"{"region":"us-east"}"#),
                partitioned_add_json("region-us-east-plain.parquet", r#"{"region":"us-east"}"#),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;
        let unfiltered_scan = build_projected_delta_scan(&source, None)?;
        let predicate = datafusion_expr_to_kernel_predicate(&col("region").eq(lit("us-west")))?;
        let predicated_scan =
            build_projected_predicated_delta_scan(&source, None, Some(predicate))?;

        assert_eq!(
            kernel_scan_file_boundaries(&unfiltered_scan, source.table_uri())?,
            vec![
                KernelScanFileBoundary {
                    path: "region-us-east-dv.parquet".to_owned(),
                    has_deletion_vector: true,
                },
                KernelScanFileBoundary {
                    path: "region-us-east-plain.parquet".to_owned(),
                    has_deletion_vector: false,
                },
                KernelScanFileBoundary {
                    path: "region-us-west-dv.parquet".to_owned(),
                    has_deletion_vector: true,
                },
                KernelScanFileBoundary {
                    path: "region-us-west-plain.parquet".to_owned(),
                    has_deletion_vector: false,
                },
            ]
        );
        assert_eq!(
            kernel_scan_file_boundaries(&predicated_scan, source.table_uri())?,
            vec![
                KernelScanFileBoundary {
                    path: "region-us-west-dv.parquet".to_owned(),
                    has_deletion_vector: true,
                },
                KernelScanFileBoundary {
                    path: "region-us-west-plain.parquet".to_owned(),
                    has_deletion_vector: false,
                },
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_string_null_and_empty_semantics()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_partition_characterization_source()?;
        let unfiltered_scan = build_projected_delta_scan(&source, None)?;

        assert_eq!(
            kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
            vec![
                "region-empty-string.parquet",
                "region-missing.parquet",
                "region-null.parquet",
                "region-us-east.parquet",
                "region-us-west.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("region").is_null())?,
            vec![
                "region-empty-string.parquet",
                "region-missing.parquet",
                "region-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("region").is_not_null())?,
            vec!["region-us-east.parquet", "region-us-west.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("region").eq(lit("us-west")))?,
            vec!["region-us-west.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("region").not_eq(lit("us-west")))?,
            vec!["region-us-east.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("region").eq(lit("")))?,
            Vec::<String>::new()
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("region").not_eq(lit("")))?,
            vec!["region-us-east.parquet", "region-us-west.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("region").in_list(vec![lit("us-west"), lit("")], false),
            )?,
            vec!["region-us-west.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("region").in_list(vec![lit("us-west"), lit("")], true),
            )?,
            vec!["region-us-east.parquet"]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_boolean_null_and_empty_semantics()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_boolean_partition_characterization_source()?;
        let unfiltered_scan = build_projected_delta_scan(&source, None)?;

        assert_eq!(
            kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
            vec![
                "boolean-empty.parquet",
                "boolean-false.parquet",
                "boolean-missing.parquet",
                "boolean-null.parquet",
                "boolean-true.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("is_current").is_null())?,
            vec![
                "boolean-empty.parquet",
                "boolean-missing.parquet",
                "boolean-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("is_current").is_not_null())?,
            vec!["boolean-false.parquet", "boolean-true.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("is_current").eq(bool_lit(true)))?,
            vec!["boolean-true.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("is_current").eq(bool_lit(false)))?,
            vec!["boolean-false.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("is_current").not_eq(bool_lit(true)))?,
            vec!["boolean-false.parquet"]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_boolean_membership_and_composition()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_boolean_partition_characterization_source()?;

        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("is_current").in_list(vec![bool_lit(true), bool_lit(false)], false),
            )?,
            vec!["boolean-false.parquet", "boolean-true.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("is_current").in_list(vec![bool_lit(true)], true),
            )?,
            vec!["boolean-false.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("is_current")
                    .eq(bool_lit(true))
                    .or(col("is_current").is_null()),
            )?,
            vec![
                "boolean-empty.parquet",
                "boolean-missing.parquet",
                "boolean-null.parquet",
                "boolean-true.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("is_current")
                    .eq(bool_lit(true))
                    .and(col("is_current").is_not_null()),
            )?,
            vec!["boolean-true.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &Expr::Not(Box::new(col("is_current").eq(bool_lit(true)))),
            )?,
            vec!["boolean-false.parquet"]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_date_ordering()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_date_partition_characterization_source()?;
        let unfiltered_scan = build_projected_delta_scan(&source, None)?;

        assert_eq!(
            kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
            vec![
                "date-empty.parquet",
                "date-epoch.parquet",
                "date-leap-day.parquet",
                "date-missing.parquet",
                "date-new-year.parquet",
                "date-null.parquet",
                "date-pre-epoch.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("event_date").gt(date_lit(19_782)))?,
            vec!["date-new-year.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("event_date").lt_eq(date_lit(0)))?,
            vec!["date-epoch.parquet", "date-pre-epoch.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &date_lit(20_454).gt(col("event_date")))?,
            vec![
                "date-epoch.parquet",
                "date-leap-day.parquet",
                "date-pre-epoch.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_date_null_and_empty_semantics()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_date_partition_characterization_source()?;

        assert_eq!(
            kernel_predicated_file_paths(&source, &col("event_date").is_null())?,
            vec![
                "date-empty.parquet",
                "date-missing.parquet",
                "date-null.parquet"
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("event_date").is_not_null())?,
            vec![
                "date-epoch.parquet",
                "date-leap-day.parquet",
                "date-new-year.parquet",
                "date-pre-epoch.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("event_date").eq(date_lit(20_454)))?,
            vec!["date-new-year.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("event_date").not_eq(date_lit(20_454)))?,
            vec![
                "date-epoch.parquet",
                "date-leap-day.parquet",
                "date-pre-epoch.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_date_membership_between_and_composition()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_date_partition_characterization_source()?;

        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("event_date").in_list(vec![date_lit(-1), date_lit(20_454)], false),
            )?,
            vec!["date-new-year.parquet", "date-pre-epoch.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("event_date").in_list(vec![date_lit(19_782)], true),
            )?,
            vec![
                "date-epoch.parquet",
                "date-new-year.parquet",
                "date-pre-epoch.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("event_date").between(date_lit(-1), date_lit(19_782)),
            )?,
            vec![
                "date-epoch.parquet",
                "date-leap-day.parquet",
                "date-pre-epoch.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("event_date").not_between(date_lit(-1), date_lit(19_782)),
            )?,
            vec!["date-new-year.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("event_date")
                    .gt_eq(date_lit(0))
                    .and(col("event_date").lt(date_lit(20_454))),
            )?,
            vec!["date-epoch.parquet", "date-leap-day.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("event_date")
                    .eq(date_lit(-1))
                    .or(col("event_date").eq(date_lit(20_454))),
            )?,
            vec!["date-new-year.parquet", "date-pre-epoch.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &Expr::Not(Box::new(col("event_date").eq(date_lit(19_782)))),
            )?,
            vec![
                "date-epoch.parquet",
                "date-new-year.parquet",
                "date-pre-epoch.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_decimal_numeric_ordering()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_decimal_partition_characterization_source()?;
        let unfiltered_scan = build_projected_delta_scan(&source, None)?;

        assert_eq!(
            kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
            vec![
                "decimal-empty.parquet",
                "decimal-large.parquet",
                "decimal-missing.parquet",
                "decimal-negative.parquet",
                "decimal-null.parquet",
                "decimal-ten.parquet",
                "decimal-two.parquet",
                "decimal-zero.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("amount").gt(decimal_lit(200)))?,
            vec!["decimal-large.parquet", "decimal-ten.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("amount").lt(decimal_lit(1000)))?,
            vec![
                "decimal-negative.parquet",
                "decimal-two.parquet",
                "decimal-zero.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &decimal_lit(1000).gt(col("amount")))?,
            vec![
                "decimal-negative.parquet",
                "decimal-two.parquet",
                "decimal-zero.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_decimal_null_and_empty_semantics()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_decimal_partition_characterization_source()?;

        assert_eq!(
            kernel_predicated_file_paths(&source, &col("amount").is_null())?,
            vec![
                "decimal-empty.parquet",
                "decimal-missing.parquet",
                "decimal-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("amount").is_not_null())?,
            vec![
                "decimal-large.parquet",
                "decimal-negative.parquet",
                "decimal-ten.parquet",
                "decimal-two.parquet",
                "decimal-zero.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("amount").eq(decimal_lit(12_345)))?,
            vec!["decimal-large.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("amount").not_eq(decimal_lit(12_345)))?,
            vec![
                "decimal-negative.parquet",
                "decimal-ten.parquet",
                "decimal-two.parquet",
                "decimal-zero.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_decimal_membership_between_and_composition()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_decimal_partition_characterization_source()?;

        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("amount").in_list(vec![decimal_lit(-123), decimal_lit(12_345)], false),
            )?,
            vec!["decimal-large.parquet", "decimal-negative.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("amount").in_list(vec![decimal_lit(200)], true),
            )?,
            vec![
                "decimal-large.parquet",
                "decimal-negative.parquet",
                "decimal-ten.parquet",
                "decimal-zero.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("amount").between(decimal_lit(-123), decimal_lit(200)),
            )?,
            vec![
                "decimal-negative.parquet",
                "decimal-two.parquet",
                "decimal-zero.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("amount").not_between(decimal_lit(-123), decimal_lit(200)),
            )?,
            vec!["decimal-large.parquet", "decimal-ten.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("amount")
                    .gt_eq(decimal_lit(0))
                    .and(col("amount").lt(decimal_lit(12345))),
            )?,
            vec![
                "decimal-ten.parquet",
                "decimal-two.parquet",
                "decimal-zero.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("amount")
                    .eq(decimal_lit(-123))
                    .or(col("amount").eq(decimal_lit(12_345))),
            )?,
            vec!["decimal-large.parquet", "decimal-negative.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &Expr::Not(Box::new(col("amount").eq(decimal_lit(200)))),
            )?,
            vec![
                "decimal-large.parquet",
                "decimal-negative.parquet",
                "decimal-ten.parquet",
                "decimal-zero.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_invalid_decimal_metadata()
    -> Result<(), Box<dyn std::error::Error>> {
        let invalid_text_table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-invalid-decimal-characterization",
            DECIMAL_PARTITIONED_METADATA_JSON,
            &[
                partitioned_add_json("decimal-valid.parquet", r#"{"amount":"123.45"}"#),
                partitioned_add_json("decimal-invalid.parquet", r#"{"amount":"not-a-decimal"}"#),
            ],
        )?;
        let invalid_text_source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: invalid_text_table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;
        let unfiltered_scan = build_projected_delta_scan(&invalid_text_source, None)?;

        assert_invalid_decimal_partition_error(kernel_scan_file_paths(
            &unfiltered_scan,
            invalid_text_source.table_uri(),
        ))?;
        assert_invalid_decimal_partition_error(kernel_predicated_file_paths(
            &invalid_text_source,
            &col("amount").eq(decimal_lit(12_345)),
        ))?;

        let scale_mismatch_table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-invalid-decimal-scale-characterization",
            DECIMAL_PARTITIONED_METADATA_JSON,
            &[
                partitioned_add_json("decimal-valid.parquet", r#"{"amount":"123.45"}"#),
                partitioned_add_json("decimal-scale.parquet", r#"{"amount":"123.450"}"#),
            ],
        )?;
        let scale_mismatch_source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: scale_mismatch_table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        assert_invalid_decimal_partition_error(kernel_predicated_file_paths(
            &scale_mismatch_source,
            &col("amount").eq(decimal_lit(12_345)),
        ))?;

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_floating_numeric_ordering()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_floating_partition_characterization_source()?;
        let unfiltered_scan = build_projected_delta_scan(&source, None)?;

        assert_eq!(
            kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
            vec![
                "floating-empty.parquet",
                "floating-missing.parquet",
                "floating-neg-zero.parquet",
                "floating-neg.parquet",
                "floating-null.parquet",
                "floating-one.parquet",
                "floating-pos-zero.parquet",
                "floating-ten.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("float_part").gt(float32_lit(1.5)))?,
            vec!["floating-ten.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("double_part").lt(float64_lit(0.0)))?,
            vec!["floating-neg-zero.parquet", "floating-neg.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &float64_lit(0.0).gt(col("double_part")))?,
            vec!["floating-neg-zero.parquet", "floating-neg.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("float_part").lt(float32_lit(10.0)))?,
            vec![
                "floating-neg-zero.parquet",
                "floating-neg.parquet",
                "floating-one.parquet",
                "floating-pos-zero.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_floating_null_and_zero_semantics()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_floating_partition_characterization_source()?;

        assert_eq!(
            kernel_predicated_file_paths(&source, &col("float_part").is_null())?,
            vec![
                "floating-empty.parquet",
                "floating-missing.parquet",
                "floating-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("double_part").is_not_null())?,
            vec![
                "floating-neg-zero.parquet",
                "floating-neg.parquet",
                "floating-one.parquet",
                "floating-pos-zero.parquet",
                "floating-ten.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("float_part").eq(float32_lit(-0.0)))?,
            vec!["floating-neg-zero.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("float_part").eq(float32_lit(0.0)))?,
            vec!["floating-pos-zero.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("double_part").not_eq(float64_lit(0.0)))?,
            vec![
                "floating-neg-zero.parquet",
                "floating-neg.parquet",
                "floating-one.parquet",
                "floating-ten.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_floating_membership_between_and_composition()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_floating_partition_characterization_source()?;

        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("float_part").in_list(vec![float32_lit(-1.5), float32_lit(1.5)], false),
            )?,
            vec!["floating-neg.parquet", "floating-one.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("double_part").in_list(vec![float64_lit(2.25)], true),
            )?,
            vec![
                "floating-neg-zero.parquet",
                "floating-neg.parquet",
                "floating-pos-zero.parquet",
                "floating-ten.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("float_part").between(float32_lit(-0.0), float32_lit(1.5)),
            )?,
            vec![
                "floating-neg-zero.parquet",
                "floating-one.parquet",
                "floating-pos-zero.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("double_part").not_between(float64_lit(0.0), float64_lit(2.25)),
            )?,
            vec![
                "floating-neg-zero.parquet",
                "floating-neg.parquet",
                "floating-ten.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("float_part")
                    .gt_eq(float32_lit(0.0))
                    .and(col("float_part").lt(float32_lit(10.0))),
            )?,
            vec!["floating-one.parquet", "floating-pos-zero.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("double_part")
                    .eq(float64_lit(-2.25))
                    .or(col("double_part").eq(float64_lit(10.0))),
            )?,
            vec!["floating-neg.parquet", "floating-ten.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &Expr::Not(Box::new(col("float_part").eq(float32_lit(1.5)))),
            )?,
            vec![
                "floating-neg-zero.parquet",
                "floating-neg.parquet",
                "floating-pos-zero.parquet",
                "floating-ten.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_binary_null_empty_and_equality_semantics()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_binary_partition_characterization_source()?;
        let unfiltered_scan = build_projected_delta_scan(&source, None)?;

        assert_eq!(
            kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
            vec![
                "binary-HELLO.parquet",
                "binary-empty.parquet",
                "binary-hello.parquet",
                "binary-missing.parquet",
                "binary-null.parquet",
                "binary-special.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::is_null(binary_partition_column())),
            )?,
            vec![
                "binary-empty.parquet",
                "binary-missing.parquet",
                "binary-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::is_not_null(binary_partition_column())),
            )?,
            vec![
                "binary-HELLO.parquet",
                "binary-hello.parquet",
                "binary-special.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(&source, binary_partition_eq(b"HELLO"))?,
            vec!["binary-HELLO.parquet"]
        );
        assert_eq!(
            kernel_predicate_file_paths(&source, binary_partition_eq(b"hello"))?,
            vec!["binary-hello.parquet"]
        );
        assert_eq!(
            kernel_predicate_file_paths(&source, binary_partition_ne(b"hello"))?,
            vec!["binary-HELLO.parquet", "binary-special.parquet"]
        );
        assert_eq!(
            kernel_predicate_file_paths(&source, binary_partition_eq(&[]))?,
            Vec::<String>::new()
        );
        assert_eq!(
            kernel_predicate_file_paths(&source, binary_partition_ne(&[]))?,
            vec![
                "binary-HELLO.parquet",
                "binary-hello.parquet",
                "binary-special.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_binary_membership_and_composition()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_binary_partition_characterization_source()?;

        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::or(
                    Predicate::eq(binary_partition_column(), binary_partition_value(b"HELLO")),
                    Predicate::eq(binary_partition_column(), binary_partition_value(b"/=%")),
                )),
            )?,
            vec!["binary-HELLO.parquet", "binary-special.parquet"]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::and(
                    Predicate::ne(binary_partition_column(), binary_partition_value(b"hello")),
                    Predicate::ne(binary_partition_column(), binary_partition_value(b"/=%")),
                )),
            )?,
            vec!["binary-HELLO.parquet"]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::and(
                    Predicate::is_not_null(binary_partition_column()),
                    Predicate::ne(binary_partition_column(), binary_partition_value(b"HELLO")),
                )),
            )?,
            vec!["binary-hello.parquet", "binary-special.parquet"]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::or(
                    Predicate::eq(binary_partition_column(), binary_partition_value(b"HELLO")),
                    Predicate::is_null(binary_partition_column()),
                )),
            )?,
            vec![
                "binary-HELLO.parquet",
                "binary-empty.parquet",
                "binary-missing.parquet",
                "binary-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::not(Predicate::eq(
                    binary_partition_column(),
                    binary_partition_value(b"hello"),
                ))),
            )?,
            vec!["binary-HELLO.parquet", "binary-special.parquet"]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_non_utf8_binary_literals_do_not_match()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_binary_partition_characterization_source()?;

        assert_eq!(
            kernel_predicate_file_paths(&source, binary_partition_eq(&[0xDE, 0xAD, 0xBE, 0xEF]))?,
            Vec::<String>::new()
        );
        assert_eq!(
            kernel_predicate_file_paths(&source, binary_partition_ne(&[0xDE, 0xAD, 0xBE, 0xEF]))?,
            vec![
                "binary-HELLO.parquet",
                "binary-hello.parquet",
                "binary-special.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_timestamp_ordering_and_precision()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_timestamp_partition_characterization_source()?;
        let unfiltered_scan = build_projected_delta_scan(&source, None)?;
        let pre_epoch = -1_i64;
        let low = 1_767_225_599_999_999_i64;
        let target = 1_767_225_600_123_456_i64;
        let high = 1_767_225_600_123_457_i64;

        assert_eq!(
            kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
            vec![
                "timestamp-empty.parquet",
                "timestamp-high.parquet",
                "timestamp-low.parquet",
                "timestamp-missing.parquet",
                "timestamp-null.parquet",
                "timestamp-pre-epoch.parquet",
                "timestamp-target.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::lt(
                    timestamp_partition_column(),
                    timestamp_partition_value(target),
                )),
            )?,
            vec!["timestamp-low.parquet", "timestamp-pre-epoch.parquet"]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::ge(
                    timestamp_partition_column(),
                    timestamp_partition_value(target),
                )),
            )?,
            vec!["timestamp-high.parquet", "timestamp-target.parquet"]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::gt(
                    timestamp_partition_value(high),
                    timestamp_partition_column(),
                )),
            )?,
            vec![
                "timestamp-low.parquet",
                "timestamp-pre-epoch.parquet",
                "timestamp-target.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(&source, timestamp_partition_eq(pre_epoch))?,
            vec!["timestamp-pre-epoch.parquet"]
        );
        assert_eq!(
            kernel_predicate_file_paths(&source, timestamp_partition_eq(target))?,
            vec!["timestamp-target.parquet"]
        );
        assert_eq!(
            kernel_predicate_file_paths(&source, timestamp_partition_ne(target))?,
            vec![
                "timestamp-high.parquet",
                "timestamp-low.parquet",
                "timestamp-pre-epoch.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(&source, timestamp_partition_eq(target + 1))?,
            vec!["timestamp-high.parquet"]
        );
        assert_eq!(
            kernel_predicate_file_paths(&source, timestamp_partition_eq(low))?,
            vec!["timestamp-low.parquet"]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_timestamp_null_empty_and_membership()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_timestamp_partition_characterization_source()?;
        let low = 1_767_225_599_999_999_i64;
        let target = 1_767_225_600_123_456_i64;

        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::is_null(timestamp_partition_column())),
            )?,
            vec![
                "timestamp-empty.parquet",
                "timestamp-missing.parquet",
                "timestamp-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::is_not_null(timestamp_partition_column())),
            )?,
            vec![
                "timestamp-high.parquet",
                "timestamp-low.parquet",
                "timestamp-pre-epoch.parquet",
                "timestamp-target.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::or(
                    Predicate::eq(timestamp_partition_column(), timestamp_partition_value(low)),
                    Predicate::eq(
                        timestamp_partition_column(),
                        timestamp_partition_value(target)
                    ),
                )),
            )?,
            vec!["timestamp-low.parquet", "timestamp-target.parquet"]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::and(
                    Predicate::ne(timestamp_partition_column(), timestamp_partition_value(low)),
                    Predicate::ne(
                        timestamp_partition_column(),
                        timestamp_partition_value(target)
                    ),
                )),
            )?,
            vec!["timestamp-high.parquet", "timestamp-pre-epoch.parquet"]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_timestamp_between_and_composition()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_timestamp_partition_characterization_source()?;
        let low = 1_767_225_599_999_999_i64;
        let target = 1_767_225_600_123_456_i64;
        let high = 1_767_225_600_123_457_i64;

        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::and(
                    Predicate::ge(timestamp_partition_column(), timestamp_partition_value(low)),
                    Predicate::le(
                        timestamp_partition_column(),
                        timestamp_partition_value(target)
                    ),
                )),
            )?,
            vec!["timestamp-low.parquet", "timestamp-target.parquet"]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::or(
                    Predicate::lt(timestamp_partition_column(), timestamp_partition_value(low)),
                    Predicate::gt(
                        timestamp_partition_column(),
                        timestamp_partition_value(target)
                    ),
                )),
            )?,
            vec!["timestamp-high.parquet", "timestamp-pre-epoch.parquet"]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::and(
                    Predicate::gt(timestamp_partition_column(), timestamp_partition_value(low)),
                    Predicate::le(
                        timestamp_partition_column(),
                        timestamp_partition_value(high)
                    ),
                )),
            )?,
            vec!["timestamp-high.parquet", "timestamp-target.parquet"]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::or(
                    Predicate::eq(timestamp_partition_column(), timestamp_partition_value(low)),
                    Predicate::is_null(timestamp_partition_column()),
                )),
            )?,
            vec![
                "timestamp-empty.parquet",
                "timestamp-low.parquet",
                "timestamp-missing.parquet",
                "timestamp-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::not(Predicate::eq(
                    timestamp_partition_column(),
                    timestamp_partition_value(target),
                ))),
            )?,
            vec![
                "timestamp-high.parquet",
                "timestamp-low.parquet",
                "timestamp-pre-epoch.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_timestamp_ntz_ordering_and_formatting()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_timestamp_ntz_partition_characterization_source()?;
        let unfiltered_scan = build_projected_delta_scan(&source, None)?;
        let pre_epoch = -1_i64;
        let low = 1_767_225_599_999_999_i64;
        let target = 1_767_225_600_123_456_i64;
        let high = 1_767_225_600_123_457_i64;

        assert_eq!(
            kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
            vec![
                "timestamp-ntz-empty.parquet",
                "timestamp-ntz-high.parquet",
                "timestamp-ntz-low-space.parquet",
                "timestamp-ntz-missing.parquet",
                "timestamp-ntz-null.parquet",
                "timestamp-ntz-pre-epoch.parquet",
                "timestamp-ntz-target-space.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::lt(
                    timestamp_ntz_partition_column(),
                    timestamp_ntz_partition_value(target),
                )),
            )?,
            vec![
                "timestamp-ntz-low-space.parquet",
                "timestamp-ntz-pre-epoch.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::ge(
                    timestamp_ntz_partition_column(),
                    timestamp_ntz_partition_value(target),
                )),
            )?,
            vec![
                "timestamp-ntz-high.parquet",
                "timestamp-ntz-target-space.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::gt(
                    timestamp_ntz_partition_value(high),
                    timestamp_ntz_partition_column(),
                )),
            )?,
            vec![
                "timestamp-ntz-low-space.parquet",
                "timestamp-ntz-pre-epoch.parquet",
                "timestamp-ntz-target-space.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(&source, timestamp_ntz_partition_eq(pre_epoch))?,
            vec!["timestamp-ntz-pre-epoch.parquet"]
        );
        assert_eq!(
            kernel_predicate_file_paths(&source, timestamp_ntz_partition_eq(target))?,
            vec!["timestamp-ntz-target-space.parquet"]
        );
        assert_eq!(
            kernel_predicate_file_paths(&source, timestamp_ntz_partition_ne(target))?,
            vec![
                "timestamp-ntz-high.parquet",
                "timestamp-ntz-low-space.parquet",
                "timestamp-ntz-pre-epoch.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(&source, timestamp_ntz_partition_eq(high))?,
            vec!["timestamp-ntz-high.parquet"]
        );
        assert_eq!(
            kernel_predicate_file_paths(&source, timestamp_ntz_partition_eq(low))?,
            vec!["timestamp-ntz-low-space.parquet"]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_timestamp_ntz_null_empty_and_membership()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_timestamp_ntz_partition_characterization_source()?;
        let low = 1_767_225_599_999_999_i64;
        let target = 1_767_225_600_123_456_i64;

        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::is_null(timestamp_ntz_partition_column())),
            )?,
            vec![
                "timestamp-ntz-empty.parquet",
                "timestamp-ntz-missing.parquet",
                "timestamp-ntz-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::is_not_null(timestamp_ntz_partition_column())),
            )?,
            vec![
                "timestamp-ntz-high.parquet",
                "timestamp-ntz-low-space.parquet",
                "timestamp-ntz-pre-epoch.parquet",
                "timestamp-ntz-target-space.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::or(
                    Predicate::eq(
                        timestamp_ntz_partition_column(),
                        timestamp_ntz_partition_value(low)
                    ),
                    Predicate::eq(
                        timestamp_ntz_partition_column(),
                        timestamp_ntz_partition_value(target)
                    ),
                )),
            )?,
            vec![
                "timestamp-ntz-low-space.parquet",
                "timestamp-ntz-target-space.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::and(
                    Predicate::ne(
                        timestamp_ntz_partition_column(),
                        timestamp_ntz_partition_value(low)
                    ),
                    Predicate::ne(
                        timestamp_ntz_partition_column(),
                        timestamp_ntz_partition_value(target)
                    ),
                )),
            )?,
            vec![
                "timestamp-ntz-high.parquet",
                "timestamp-ntz-pre-epoch.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_timestamp_ntz_between_and_composition()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_timestamp_ntz_partition_characterization_source()?;
        let low = 1_767_225_599_999_999_i64;
        let target = 1_767_225_600_123_456_i64;
        let high = 1_767_225_600_123_457_i64;

        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::and(
                    Predicate::ge(
                        timestamp_ntz_partition_column(),
                        timestamp_ntz_partition_value(low)
                    ),
                    Predicate::le(
                        timestamp_ntz_partition_column(),
                        timestamp_ntz_partition_value(target)
                    ),
                )),
            )?,
            vec![
                "timestamp-ntz-low-space.parquet",
                "timestamp-ntz-target-space.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::or(
                    Predicate::lt(
                        timestamp_ntz_partition_column(),
                        timestamp_ntz_partition_value(low)
                    ),
                    Predicate::gt(
                        timestamp_ntz_partition_column(),
                        timestamp_ntz_partition_value(target)
                    ),
                )),
            )?,
            vec![
                "timestamp-ntz-high.parquet",
                "timestamp-ntz-pre-epoch.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::and(
                    Predicate::gt(
                        timestamp_ntz_partition_column(),
                        timestamp_ntz_partition_value(low)
                    ),
                    Predicate::le(
                        timestamp_ntz_partition_column(),
                        timestamp_ntz_partition_value(high)
                    ),
                )),
            )?,
            vec![
                "timestamp-ntz-high.parquet",
                "timestamp-ntz-target-space.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::or(
                    Predicate::eq(
                        timestamp_ntz_partition_column(),
                        timestamp_ntz_partition_value(low)
                    ),
                    Predicate::is_null(timestamp_ntz_partition_column()),
                )),
            )?,
            vec![
                "timestamp-ntz-empty.parquet",
                "timestamp-ntz-low-space.parquet",
                "timestamp-ntz-missing.parquet",
                "timestamp-ntz-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::not(Predicate::eq(
                    timestamp_ntz_partition_column(),
                    timestamp_ntz_partition_value(target),
                ))),
            )?,
            vec![
                "timestamp-ntz-high.parquet",
                "timestamp-ntz-low-space.parquet",
                "timestamp-ntz-pre-epoch.parquet",
            ]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_invalid_timestamp_metadata()
    -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-invalid-timestamp-characterization",
            TIMESTAMP_PARTITIONED_METADATA_JSON,
            &[
                partitioned_add_json(
                    "timestamp-valid.parquet",
                    r#"{"event_ts":"2026-01-01T00:00:00.123456Z"}"#,
                ),
                partitioned_add_json(
                    "timestamp-invalid.parquet",
                    r#"{"event_ts":"not-a-timestamp"}"#,
                ),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;
        let unfiltered_scan = build_projected_delta_scan(&source, None)?;

        assert_invalid_timestamp_partition_error(kernel_scan_file_paths(
            &unfiltered_scan,
            source.table_uri(),
        ))?;
        assert_invalid_timestamp_partition_error(kernel_predicate_file_paths(
            &source,
            timestamp_partition_eq(1_767_225_600_123_456),
        ))?;

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_invalid_timestamp_ntz_metadata()
    -> Result<(), Box<dyn std::error::Error>> {
        let invalid_text_table = DeltaLogTable::new_with_protocol_metadata_and_adds(
            "kernel-invalid-timestamp-ntz-characterization",
            TIMESTAMP_NTZ_PROTOCOL_JSON,
            TIMESTAMP_NTZ_PARTITIONED_METADATA_JSON,
            &[
                partitioned_add_json(
                    "timestamp-ntz-valid.parquet",
                    r#"{"event_ts_ntz":"2026-01-01 00:00:00.123456"}"#,
                ),
                partitioned_add_json(
                    "timestamp-ntz-invalid.parquet",
                    r#"{"event_ts_ntz":"not-a-timestamp"}"#,
                ),
            ],
        )?;
        let invalid_text_source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: invalid_text_table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;
        let unfiltered_scan = build_projected_delta_scan(&invalid_text_source, None)?;

        assert_invalid_timestamp_ntz_partition_error(kernel_scan_file_paths(
            &unfiltered_scan,
            invalid_text_source.table_uri(),
        ))?;
        assert_invalid_timestamp_ntz_partition_error(kernel_predicate_file_paths(
            &invalid_text_source,
            timestamp_ntz_partition_eq(1_767_225_600_123_456),
        ))?;

        let t_separator_table = DeltaLogTable::new_with_protocol_metadata_and_adds(
            "kernel-t-separator-timestamp-ntz-characterization",
            TIMESTAMP_NTZ_PROTOCOL_JSON,
            TIMESTAMP_NTZ_PARTITIONED_METADATA_JSON,
            &[
                partitioned_add_json(
                    "timestamp-ntz-valid.parquet",
                    r#"{"event_ts_ntz":"2026-01-01 00:00:00.123456"}"#,
                ),
                partitioned_add_json(
                    "timestamp-ntz-t-separator.parquet",
                    r#"{"event_ts_ntz":"2026-01-01T00:00:00.123456"}"#,
                ),
            ],
        )?;
        let t_separator_source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: t_separator_table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;
        let unfiltered_scan = build_projected_delta_scan(&t_separator_source, None)?;

        assert_invalid_timestamp_ntz_partition_error(kernel_scan_file_paths(
            &unfiltered_scan,
            t_separator_source.table_uri(),
        ))?;
        assert_invalid_timestamp_ntz_partition_error(kernel_predicate_file_paths(
            &t_separator_source,
            timestamp_ntz_partition_eq(1_767_225_600_123_456),
        ))?;

        let zone_table = DeltaLogTable::new_with_protocol_metadata_and_adds(
            "kernel-zone-timestamp-ntz-characterization",
            TIMESTAMP_NTZ_PROTOCOL_JSON,
            TIMESTAMP_NTZ_PARTITIONED_METADATA_JSON,
            &[
                partitioned_add_json(
                    "timestamp-ntz-valid.parquet",
                    r#"{"event_ts_ntz":"2026-01-01 00:00:00.123456"}"#,
                ),
                partitioned_add_json(
                    "timestamp-ntz-zone.parquet",
                    r#"{"event_ts_ntz":"2026-01-01T00:00:00.123456Z"}"#,
                ),
            ],
        )?;
        let zone_source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: zone_table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;
        let unfiltered_scan = build_projected_delta_scan(&zone_source, None)?;

        assert_invalid_timestamp_ntz_partition_error(kernel_scan_file_paths(
            &unfiltered_scan,
            zone_source.table_uri(),
        ))?;
        assert_invalid_timestamp_ntz_partition_error(kernel_predicate_file_paths(
            &zone_source,
            timestamp_ntz_partition_eq(1_767_225_600_123_456),
        ))?;

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_invalid_floating_metadata()
    -> Result<(), Box<dyn std::error::Error>> {
        let invalid_text_table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-invalid-floating-characterization",
            FLOATING_PARTITIONED_METADATA_JSON,
            &[
                partitioned_add_json(
                    "floating-valid.parquet",
                    r#"{"float_part":"1.5","double_part":"2.25"}"#,
                ),
                partitioned_add_json(
                    "floating-invalid.parquet",
                    r#"{"float_part":"not-a-float","double_part":"not-a-double"}"#,
                ),
            ],
        )?;
        let invalid_text_source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: invalid_text_table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;
        let unfiltered_scan = build_projected_delta_scan(&invalid_text_source, None)?;

        assert_invalid_floating_partition_error(kernel_scan_file_paths(
            &unfiltered_scan,
            invalid_text_source.table_uri(),
        ))?;
        assert_invalid_floating_partition_error(kernel_predicated_file_paths(
            &invalid_text_source,
            &col("float_part").eq(float32_lit(1.5)),
        ))?;

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_nonfinite_floating_metadata()
    -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-nonfinite-floating-characterization",
            FLOATING_PARTITIONED_METADATA_JSON,
            &[
                partitioned_add_json(
                    "floating-valid.parquet",
                    r#"{"float_part":"1.5","double_part":"2.25"}"#,
                ),
                partitioned_add_json(
                    "floating-nan.parquet",
                    r#"{"float_part":"NaN","double_part":"NaN"}"#,
                ),
                partitioned_add_json(
                    "floating-inf.parquet",
                    r#"{"float_part":"Infinity","double_part":"Infinity"}"#,
                ),
                partitioned_add_json(
                    "floating-neg-inf.parquet",
                    r#"{"float_part":"-Infinity","double_part":"-Infinity"}"#,
                ),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;
        let unfiltered_scan = build_projected_delta_scan(&source, None)?;

        assert_eq!(
            kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
            vec![
                "floating-inf.parquet",
                "floating-nan.parquet",
                "floating-neg-inf.parquet",
                "floating-valid.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("float_part").gt(float32_lit(0.0)))?,
            vec![
                "floating-inf.parquet",
                "floating-nan.parquet",
                "floating-valid.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("double_part").lt(float64_lit(0.0)))?,
            vec!["floating-neg-inf.parquet"]
        );
        assert_eq!(
            kernel_predicate_file_paths(
                &source,
                DeltaKernelPredicate::new(Predicate::eq(
                    Expression::Column(ColumnName::new(["float_part"])),
                    Expression::Literal(Scalar::Float(f32::NAN)),
                )),
            )?,
            vec!["floating-nan.parquet"]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_invalid_date_metadata()
    -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-invalid-date-characterization",
            DATE_PARTITIONED_METADATA_JSON,
            &[
                partitioned_add_json("date-valid.parquet", r#"{"event_date":"2026-01-01"}"#),
                partitioned_add_json("date-invalid.parquet", r#"{"event_date":"not-a-date"}"#),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;
        let unfiltered_scan = build_projected_delta_scan(&source, None)?;

        assert_invalid_date_partition_error(kernel_scan_file_paths(
            &unfiltered_scan,
            source.table_uri(),
        ))?;
        assert_invalid_date_partition_error(kernel_predicated_file_paths(
            &source,
            &col("event_date").eq(date_lit(20_454)),
        ))?;

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_invalid_boolean_metadata()
    -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-invalid-boolean-characterization",
            BOOLEAN_PARTITIONED_METADATA_JSON,
            &[
                partitioned_add_json("boolean-valid.parquet", r#"{"is_current":"true"}"#),
                partitioned_add_json(
                    "boolean-invalid.parquet",
                    r#"{"is_current":"not-a-boolean"}"#,
                ),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;
        let unfiltered_scan = build_projected_delta_scan(&source, None)?;

        assert_invalid_boolean_partition_error(kernel_scan_file_paths(
            &unfiltered_scan,
            source.table_uri(),
        ))?;
        assert_invalid_boolean_partition_error(kernel_predicated_file_paths(
            &source,
            &col("is_current").eq(bool_lit(true)),
        ))?;

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_integer_numeric_ordering()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_integer_partition_characterization_source()?;
        let unfiltered_scan = build_projected_delta_scan(&source, None)?;

        assert_eq!(
            kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
            vec![
                "integer--1.parquet",
                "integer-10.parquet",
                "integer-2.parquet",
                "integer-empty.parquet",
                "integer-missing.parquet",
                "integer-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("long_part").gt(int64_lit(2)))?,
            vec!["integer-10.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("long_part").lt(int64_lit(10)))?,
            vec!["integer--1.parquet", "integer-2.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &int64_lit(10).gt(col("long_part")))?,
            vec!["integer--1.parquet", "integer-2.parquet"]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_integer_null_and_empty_semantics()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_integer_partition_characterization_source()?;

        assert_eq!(
            kernel_predicated_file_paths(&source, &col("long_part").is_null())?,
            vec![
                "integer-empty.parquet",
                "integer-missing.parquet",
                "integer-null.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("long_part").is_not_null())?,
            vec![
                "integer--1.parquet",
                "integer-10.parquet",
                "integer-2.parquet",
            ]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("long_part").eq(int64_lit(2)))?,
            vec!["integer-2.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("long_part").not_eq(int64_lit(2)))?,
            vec!["integer--1.parquet", "integer-10.parquet"]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_integer_membership_between_and_composition()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_table, source) = kernel_integer_partition_characterization_source()?;

        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("long_part").in_list(vec![int64_lit(-1), int64_lit(10)], false),
            )?,
            vec!["integer--1.parquet", "integer-10.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("long_part").in_list(vec![int64_lit(2)], true),
            )?,
            vec!["integer--1.parquet", "integer-10.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("long_part").between(int64_lit(-1), int64_lit(2)),
            )?,
            vec!["integer--1.parquet", "integer-2.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("long_part").not_between(int64_lit(-1), int64_lit(2)),
            )?,
            vec!["integer-10.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("long_part")
                    .gt_eq(int64_lit(-1))
                    .and(col("long_part").lt(int64_lit(10))),
            )?,
            vec!["integer--1.parquet", "integer-2.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &col("long_part")
                    .eq(int64_lit(-1))
                    .or(col("long_part").eq(int64_lit(10))),
            )?,
            vec!["integer--1.parquet", "integer-10.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(
                &source,
                &Expr::Not(Box::new(col("long_part").eq(int64_lit(2)))),
            )?,
            vec!["integer--1.parquet", "integer-10.parquet"]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_integer_width_scalars()
    -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-integer-width-characterization",
            INTEGER_PARTITIONED_METADATA_JSON,
            &[
                partitioned_add_json(
                    "width-byte-min.parquet",
                    r#"{"byte_part":"-128","short_part":"0","int_part":"0","long_part":"0"}"#,
                ),
                partitioned_add_json(
                    "width-short-max.parquet",
                    r#"{"byte_part":"0","short_part":"32767","int_part":"0","long_part":"0"}"#,
                ),
                partitioned_add_json(
                    "width-int-max.parquet",
                    r#"{"byte_part":"0","short_part":"0","int_part":"2147483647","long_part":"0"}"#,
                ),
                partitioned_add_json(
                    "width-long-max.parquet",
                    r#"{"byte_part":"0","short_part":"0","int_part":"0","long_part":"9223372036854775807"}"#,
                ),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;

        assert_eq!(
            kernel_predicated_file_paths(&source, &col("byte_part").eq(int8_lit(i8::MIN)))?,
            vec!["width-byte-min.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("short_part").eq(int16_lit(i16::MAX)))?,
            vec!["width-short-max.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("int_part").eq(int32_lit(i32::MAX)))?,
            vec!["width-int-max.parquet"]
        );
        assert_eq!(
            kernel_predicated_file_paths(&source, &col("long_part").eq(int64_lit(i64::MAX)))?,
            vec!["width-long-max.parquet"]
        );

        Ok(())
    }

    #[test]
    fn kernel_partition_characterization_documents_invalid_integer_metadata()
    -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new_with_metadata_and_adds(
            "kernel-invalid-integer-characterization",
            INTEGER_PARTITIONED_METADATA_JSON,
            &[
                integer_partitioned_add_json("integer-valid.parquet", "7"),
                partitioned_add_json(
                    "integer-invalid.parquet",
                    r#"{"byte_part":"0","short_part":"0","int_part":"0","long_part":"not-an-integer"}"#,
                ),
            ],
        )?;
        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: None,
            storage_options: Default::default(),
        })?;
        let unfiltered_scan = build_projected_delta_scan(&source, None)?;

        assert_invalid_integer_partition_error(kernel_scan_file_paths(
            &unfiltered_scan,
            source.table_uri(),
        ))?;
        assert_invalid_integer_partition_error(kernel_predicated_file_paths(
            &source,
            &col("long_part").eq(int64_lit(7)),
        ))?;

        Ok(())
    }

    #[test]
    fn loads_named_delta_source() -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new("success")?;
        let requested_table_uri = table.path.to_string_lossy().to_string();

        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: requested_table_uri.clone(),
            version: None,
            storage_options: Default::default(),
        })?;

        assert_eq!(source.name(), "orders");
        assert_eq!(source.requested_table_uri(), requested_table_uri);
        assert!(source.table_uri().starts_with("file://"));
        assert_eq!(source.version(), 1);
        assert_eq!(source.loaded_snapshot().version(), 1);

        Ok(())
    }

    #[test]
    fn planned_delta_source_owns_and_isolates_storage_options()
    -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new("storage-options-owned")?;
        let mut caller_options = storage_options(&[
            ("authorization", "caller-original"),
            ("region", "us-west-2"),
        ]);
        let source = load_delta_source(
            DeltaSourceConfig::new("orders", table.path.to_string_lossy())
                .with_storage_options(caller_options.clone()),
        )?;

        caller_options.insert("authorization".to_owned(), "caller-mutated".to_owned());
        caller_options.insert("new-secret".to_owned(), "caller-added".to_owned());

        assert_eq!(
            source
                .storage_options()
                .get("authorization")
                .map(String::as_str),
            Some("caller-original")
        );
        assert_eq!(
            source.storage_options().get("region").map(String::as_str),
            Some("us-west-2")
        );
        assert!(!source.storage_options().contains_key("new-secret"));

        Ok(())
    }

    #[test]
    fn s3_effective_storage_options_merge_aws_env_when_absent() {
        for table_uri in [
            "s3://bucket/root",
            "s3a://bucket/root",
            "https://s3.us-east-1.amazonaws.com/bucket/root",
            "https://bucket.s3.us-east-1.amazonaws.com/root",
            "https://ACCOUNT_ID.r2.cloudflarestorage.com/bucket/root",
        ] {
            let effective = effective_storage_options_for_source_from_env(
                table_uri,
                DeltaStorageOptions::default(),
                env_vars(&[
                    ("AWS_ACCESS_KEY_ID", "env-access"),
                    ("AWS_SECRET_ACCESS_KEY", "env-secret"),
                    ("AWS_REGION", "us-east-1"),
                    ("AWS_CUSTOM_FUTURE_OPTION", "future"),
                    ("NOT_AWS", "ignored"),
                ]),
            );

            assert_eq!(
                effective.get("AWS_ACCESS_KEY_ID").map(String::as_str),
                Some("env-access"),
                "{table_uri}"
            );
            assert_eq!(
                effective.get("AWS_SECRET_ACCESS_KEY").map(String::as_str),
                Some("env-secret"),
                "{table_uri}"
            );
            assert_eq!(
                effective.get("AWS_REGION").map(String::as_str),
                Some("us-east-1"),
                "{table_uri}"
            );
            assert_eq!(
                effective
                    .get("AWS_CUSTOM_FUTURE_OPTION")
                    .map(String::as_str),
                Some("future"),
                "{table_uri}"
            );
            assert!(!effective.contains_key("NOT_AWS"), "{table_uri}");
        }
    }

    #[test]
    fn s3_effective_storage_options_prefer_explicit_caller_options() {
        let caller_options = storage_options(&[
            ("aws_access_key_id", "caller-access"),
            ("AWS_SECRET_ACCESS_KEY", "caller-secret"),
            ("region", "caller-region"),
        ]);
        let effective = effective_storage_options_for_source_from_env(
            "s3://bucket/root",
            caller_options,
            env_vars(&[
                ("AWS_ACCESS_KEY_ID", "env-access"),
                ("AWS_SECRET_ACCESS_KEY", "env-secret"),
                ("AWS_REGION", "env-region"),
                ("AWS_CUSTOM_FUTURE_OPTION", "future"),
            ]),
        );

        assert_eq!(
            effective.get("aws_access_key_id").map(String::as_str),
            Some("caller-access")
        );
        assert_eq!(
            effective.get("AWS_SECRET_ACCESS_KEY").map(String::as_str),
            Some("caller-secret")
        );
        assert_eq!(
            effective.get("region").map(String::as_str),
            Some("caller-region")
        );
        assert_eq!(
            effective
                .get("AWS_CUSTOM_FUTURE_OPTION")
                .map(String::as_str),
            Some("future")
        );
        assert!(!effective.contains_key("AWS_ACCESS_KEY_ID"));
        assert!(!effective.contains_key("AWS_REGION"));
    }

    #[test]
    fn non_s3_effective_storage_options_do_not_merge_aws_env() {
        let caller_options = storage_options(&[("authorization", "caller-token")]);
        let effective = effective_storage_options_for_source_from_env(
            "file:///tmp/table",
            caller_options.clone(),
            env_vars(&[
                ("AWS_ACCESS_KEY_ID", "env-access"),
                ("AWS_SECRET_ACCESS_KEY", "env-secret"),
                ("AWS_REGION", "us-east-1"),
            ]),
        );

        assert_eq!(effective, caller_options);
    }

    #[test]
    fn s3_effective_storage_options_are_resolved_before_source_load() {
        let storage_options = effective_storage_options_for_source_from_env(
            "s3://bucket/root",
            storage_options(&[("region", "caller-region")]),
            env_vars(&[
                ("AWS_ACCESS_KEY_ID", "env-access"),
                ("AWS_SECRET_ACCESS_KEY", "env-secret"),
                ("AWS_REGION", "env-region"),
            ]),
        );

        assert_eq!(
            storage_options.get("AWS_ACCESS_KEY_ID").map(String::as_str),
            Some("env-access")
        );
        assert_eq!(
            storage_options
                .get("AWS_SECRET_ACCESS_KEY")
                .map(String::as_str),
            Some("env-secret")
        );
        assert_eq!(
            storage_options.get("region").map(String::as_str),
            Some("caller-region")
        );
        assert!(!storage_options.contains_key("AWS_REGION"));
    }

    #[test]
    fn s3_storage_options_accept_documented_credential_keys() {
        for (option_key, config_key, expected_value) in [
            (
                "AWS_ACCESS_KEY_ID",
                AmazonS3ConfigKey::AccessKeyId,
                "upper-access",
            ),
            (
                "aws_access_key_id",
                AmazonS3ConfigKey::AccessKeyId,
                "lower-access",
            ),
            (
                "AWS_SECRET_ACCESS_KEY",
                AmazonS3ConfigKey::SecretAccessKey,
                "upper-secret",
            ),
            (
                "aws_secret_access_key",
                AmazonS3ConfigKey::SecretAccessKey,
                "lower-secret",
            ),
            ("AWS_SESSION_TOKEN", AmazonS3ConfigKey::Token, "upper-token"),
            ("aws_session_token", AmazonS3ConfigKey::Token, "lower-token"),
        ] {
            let options = storage_options(&[(option_key, expected_value)]);
            let builder = s3_builder_from_storage_options(&options);

            assert_eq!(
                builder.get_config_value(&config_key).as_deref(),
                Some(expected_value),
                "{option_key}"
            );
        }
    }

    #[test]
    fn s3_storage_options_accept_documented_region_keys() {
        for option_key in ["AWS_REGION", "aws_region", "region"] {
            let options = storage_options(&[(option_key, "explicit-region")]);
            let builder = s3_builder_from_storage_options(&options);

            assert_eq!(
                builder
                    .get_config_value(&AmazonS3ConfigKey::Region)
                    .as_deref(),
                Some("explicit-region"),
                "{option_key}"
            );
        }

        for option_key in ["AWS_DEFAULT_REGION", "aws_default_region"] {
            let options = storage_options(&[(option_key, "fallback-region")]);
            let builder = s3_builder_from_storage_options(&options);

            assert_eq!(
                builder
                    .get_config_value(&AmazonS3ConfigKey::Region)
                    .as_deref(),
                Some("fallback-region"),
                "{option_key}"
            );
        }
    }

    #[test]
    fn s3_default_region_only_applies_without_explicit_region() {
        for default_key in ["AWS_DEFAULT_REGION", "aws_default_region"] {
            for explicit_key in ["AWS_REGION", "aws_region", "region"] {
                let options = storage_options(&[
                    (default_key, "fallback-region"),
                    (explicit_key, "explicit-region"),
                ]);
                let builder = s3_builder_from_storage_options(&options);

                assert_eq!(
                    builder
                        .get_config_value(&AmazonS3ConfigKey::Region)
                        .as_deref(),
                    Some("explicit-region"),
                    "{default_key} with {explicit_key}"
                );
            }
        }
    }

    #[test]
    fn s3_store_construction_accepts_documented_storage_option_keys()
    -> Result<(), Box<dyn std::error::Error>> {
        let cases = [
            (
                "AWS_ACCESS_KEY_ID",
                "AWS_SECRET_ACCESS_KEY",
                "AWS_SESSION_TOKEN",
                "AWS_REGION",
            ),
            (
                "aws_access_key_id",
                "aws_secret_access_key",
                "aws_session_token",
                "aws_region",
            ),
            (
                "aws_access_key_id",
                "aws_secret_access_key",
                "aws_session_token",
                "region",
            ),
            (
                "aws_access_key_id",
                "aws_secret_access_key",
                "aws_session_token",
                "AWS_DEFAULT_REGION",
            ),
            (
                "aws_access_key_id",
                "aws_secret_access_key",
                "aws_session_token",
                "aws_default_region",
            ),
        ];
        let table_url = super::kernel::try_parse_uri("s3://bucket/root")?;

        for (access_key, secret_key, token_key, region_key) in cases {
            let options = storage_options(&[
                (access_key, "access"),
                (secret_key, "secret"),
                (token_key, "token"),
                (region_key, "us-east-1"),
            ]);

            let _store = super::kernel::store_from_url_opts(
                &table_url,
                options
                    .iter()
                    .map(|(key, value)| (key.as_str(), value.as_str())),
            )?;
        }

        Ok(())
    }

    #[test]
    fn source_loading_passes_storage_options_to_kernel_store_construction()
    -> Result<(), Box<dyn std::error::Error>> {
        let scheme = unique_storage_scheme("sourceoptions")?;
        let captured = CapturedStorageOptions::default();
        register_capturing_storage_handler(&scheme, Arc::clone(&captured))?;
        let options =
            storage_options(&[("authorization", "source-token"), ("region", "us-east-1")]);

        let result = load_delta_source(
            DeltaSourceConfig::new("orders", format!("{scheme}://table/root"))
                .with_storage_options(options.clone()),
        );

        assert!(matches!(
            result,
            Err(DeltaFunnelError::DeltaSnapshotLoad { .. })
        ));
        assert_eq!(captured_storage_options(&captured), vec![options]);

        Ok(())
    }

    #[test]
    fn reader_construction_passes_storage_options_to_each_store_construction()
    -> Result<(), Box<dyn std::error::Error>> {
        let scheme = unique_storage_scheme("readeroptions")?;
        let captured = CapturedStorageOptions::default();
        register_capturing_storage_handler(&scheme, Arc::clone(&captured))?;
        let table_uri = format!("{scheme}://table/root/");
        let options = storage_options(&[
            ("authorization", "reader-token"),
            ("endpoint", "http://storage.example"),
        ]);

        let _data_reader = KernelDataFileReader::try_new(KernelDataFileReaderConfig {
            source_name: "orders",
            table_uri: &table_uri,
            snapshot_version: 42,
            storage_options: &options,
        })?;
        let _dv_reader = KernelDeletionVectorReader::try_new(KernelDeletionVectorReaderConfig {
            source_name: "orders",
            table_uri: &table_uri,
            snapshot_version: 42,
            storage_options: &options,
        })?;
        let captured_options = captured_storage_options(&captured);
        assert_eq!(captured_options.len(), 2);
        assert!(captured_options.iter().all(|captured| captured == &options));

        Ok(())
    }

    #[test]
    fn reader_construction_passes_effective_s3_storage_options_to_each_store_construction()
    -> Result<(), Box<dyn std::error::Error>> {
        let scheme = unique_storage_scheme("readereffectiveoptions")?;
        let captured = CapturedStorageOptions::default();
        register_capturing_storage_handler(&scheme, Arc::clone(&captured))?;
        let table_uri = format!("{scheme}://table/root/");
        let effective = effective_storage_options_for_source_from_env(
            "s3://bucket/root",
            storage_options(&[("region", "caller-region")]),
            env_vars(&[
                ("AWS_ACCESS_KEY_ID", "env-access"),
                ("AWS_SECRET_ACCESS_KEY", "env-secret"),
                ("AWS_REGION", "env-region"),
            ]),
        );

        let _data_reader = KernelDataFileReader::try_new(KernelDataFileReaderConfig {
            source_name: "orders",
            table_uri: &table_uri,
            snapshot_version: 42,
            storage_options: &effective,
        })?;
        let _dv_reader = KernelDeletionVectorReader::try_new(KernelDeletionVectorReaderConfig {
            source_name: "orders",
            table_uri: &table_uri,
            snapshot_version: 42,
            storage_options: &effective,
        })?;
        let captured_options = captured_storage_options(&captured);
        assert_eq!(captured_options.len(), 2);
        assert!(
            captured_options
                .iter()
                .all(|captured| captured == &effective)
        );

        Ok(())
    }

    #[test]
    fn storage_option_values_are_not_exposed_when_store_construction_fails()
    -> Result<(), Box<dyn std::error::Error>> {
        for secret_key in [
            "AWS_SECRET_ACCESS_KEY",
            "aws_secret_access_key",
            "AWS_SESSION_TOKEN",
            "aws_session_token",
        ] {
            let scheme = unique_storage_scheme("redactedoptions")?;
            let captured = CapturedStorageOptions::default();
            let secret_value = "super-secret-token-value";
            register_failing_storage_handler(&scheme, Arc::clone(&captured), secret_value)?;
            let options =
                storage_options(&[(secret_key, secret_value), ("AWS_REGION", "us-west-2")]);

            let result = load_delta_source(
                DeltaSourceConfig::new("orders", format!("{scheme}://table/root"))
                    .with_storage_options(options.clone()),
            );
            let error = result
                .err()
                .map(|error| error.to_string())
                .unwrap_or_default();

            assert_eq!(captured_storage_options(&captured), vec![options]);
            assert!(!error.contains(secret_value), "{secret_key}");
            assert!(!error.contains(secret_key), "{secret_key}");
        }

        Ok(())
    }

    #[test]
    fn merged_s3_env_storage_option_values_are_not_exposed_when_store_construction_fails()
    -> Result<(), Box<dyn std::error::Error>> {
        let scheme = unique_storage_scheme("redacteds3envoptions")?;
        let captured = CapturedStorageOptions::default();
        let secret_value = "super-secret-env-token-value";
        register_failing_storage_handler(&scheme, Arc::clone(&captured), secret_value)?;
        let effective = effective_storage_options_for_source_from_env(
            "s3://bucket/root",
            DeltaStorageOptions::default(),
            env_vars(&[
                ("AWS_ACCESS_KEY_ID", "env-access"),
                ("AWS_SECRET_ACCESS_KEY", secret_value),
                ("AWS_REGION", "us-east-1"),
            ]),
        );

        let result = load_delta_source(
            DeltaSourceConfig::new("orders", format!("{scheme}://table/root"))
                .with_storage_options(effective.clone()),
        );
        let error = result
            .err()
            .map(|error| error.to_string())
            .unwrap_or_default();

        assert_eq!(captured_storage_options(&captured), vec![effective]);
        assert!(!error.contains(secret_value));
        assert!(!error.contains("AWS_SECRET_ACCESS_KEY"));

        Ok(())
    }

    #[test]
    fn invalid_recognized_s3_aws_env_values_fail_store_construction() {
        let storage_options = effective_storage_options_for_source_from_env(
            "s3://bucket/root",
            DeltaStorageOptions::default(),
            env_vars(&[
                ("AWS_ALLOW_HTTP", "not-a-bool"),
                ("AWS_REGION", "us-east-1"),
            ]),
        );

        let result = load_delta_table_snapshot("s3://bucket/root", None, &storage_options);

        assert!(matches!(
            result,
            Err(DeltaFunnelError::DeltaSourceEngine { .. })
        ));
    }

    #[test]
    fn loads_named_delta_source_at_fixed_version() -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new("fixed")?;

        let source = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: table.path.to_string_lossy().to_string(),
            version: Some(0),
            storage_options: Default::default(),
        })?;

        assert_eq!(source.version(), 0);

        Ok(())
    }

    #[test]
    fn validates_name_before_table_uri() {
        let result = load_delta_source(DeltaSourceConfig {
            name: "orders.latest".to_owned(),
            table_uri: "missing/path".to_owned(),
            version: None,
            storage_options: Default::default(),
        });

        assert!(matches!(
            result,
            Err(DeltaFunnelError::InvalidSourceName { .. })
        ));
    }

    #[test]
    fn rejects_sql_keyword_name_before_table_uri() {
        let result = load_delta_source(DeltaSourceConfig {
            name: "select".to_owned(),
            table_uri: "missing/path".to_owned(),
            version: None,
            storage_options: Default::default(),
        });

        assert!(matches!(
            result,
            Err(DeltaFunnelError::InvalidSourceName { name, reason })
                if name == "select" && reason == "source names must not be SQL keywords"
        ));
    }

    #[test]
    fn rejects_blank_table_uri_as_invalid_source_uri() {
        let result = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: " \t\n".to_owned(),
            version: None,
            storage_options: Default::default(),
        });

        assert!(matches!(
            result,
            Err(DeltaFunnelError::InvalidSourceUri { reason })
                if reason == "table location must not be empty"
        ));
    }

    #[test]
    fn loads_multiple_named_delta_sources() -> Result<(), Box<dyn std::error::Error>> {
        let orders = DeltaLogTable::new("orders")?;
        let customers = DeltaLogTable::new("customers")?;

        let sources = load_delta_sources([
            DeltaSourceConfig {
                name: "orders".to_owned(),
                table_uri: orders.path.to_string_lossy().to_string(),
                version: None,
                storage_options: Default::default(),
            },
            DeltaSourceConfig {
                name: "customers".to_owned(),
                table_uri: customers.path.to_string_lossy().to_string(),
                version: Some(0),
                storage_options: Default::default(),
            },
        ])?;

        assert_eq!(sources.len(), 2);
        assert_eq!(sources[0].name(), "orders");
        assert_eq!(sources[0].version(), 1);
        assert_eq!(sources[1].name(), "customers");
        assert_eq!(sources[1].version(), 0);

        Ok(())
    }

    #[test]
    fn rejects_duplicate_names_before_table_uri_loading() {
        let result = load_delta_sources([
            DeltaSourceConfig {
                name: "orders".to_owned(),
                table_uri: "missing/orders".to_owned(),
                version: None,
                storage_options: Default::default(),
            },
            DeltaSourceConfig {
                name: "Orders".to_owned(),
                table_uri: "missing/customers".to_owned(),
                version: None,
                storage_options: Default::default(),
            },
        ]);

        assert!(matches!(
            result,
            Err(DeltaFunnelError::DuplicateSourceName { name }) if name == "Orders"
        ));
    }

    #[test]
    fn rejects_invalid_name_before_multi_source_table_uri_loading() {
        let result = load_delta_sources([
            DeltaSourceConfig {
                name: "orders".to_owned(),
                table_uri: "missing/orders".to_owned(),
                version: None,
                storage_options: Default::default(),
            },
            DeltaSourceConfig {
                name: "customers.latest".to_owned(),
                table_uri: "missing/customers".to_owned(),
                version: None,
                storage_options: Default::default(),
            },
        ]);

        assert!(matches!(
            result,
            Err(DeltaFunnelError::InvalidSourceName { name, .. }) if name == "customers.latest"
        ));
    }

    #[test]
    fn source_load_errors_do_not_expose_secret_bearing_uri() {
        let result = load_delta_source(DeltaSourceConfig {
            name: "orders".to_owned(),
            table_uri: "ftp://user:password@example.com/table".to_owned(),
            version: None,
            storage_options: Default::default(),
        });

        let error = result
            .err()
            .map(|error| error.to_string())
            .unwrap_or_default();

        assert!(!error.contains("user"));
        assert!(!error.contains("password"));
        assert!(!error.contains("example.com"));
        assert!(!error.contains("ftp://"));
    }
}