delta_kernel 0.30.0

//! StateInfo handles the state that we use through log-replay in order to correctly construct all
//! the physical->logical transforms needed for each add file

use std::collections::HashSet;
use std::sync::Arc;

use tracing::{debug, enabled, warn, Level};
use url::Url;

use crate::actions::NULL_COUNT;
use crate::expressions::ColumnName;
use crate::scan::field_classifiers::TransformFieldClassifier;
use crate::scan::transform_spec::{FieldTransformSpec, TransformSpec};
use crate::scan::{PartitionValuesOptions, PhysicalPredicate, StatsOptions, StructStats};
use crate::schema::{DataType, MetadataColumnSpec, SchemaRef, StructType};
use crate::table_configuration::TableConfiguration;
use crate::table_features::{get_any_level_column_physical_name, ColumnMappingMode, TableFeature};
use crate::{KernelError, KernelResult, PredicateRef, StructField};

/// Resolved physical statistics schemas for a scan.
///
/// Both schemas are resolved together during [`StateInfo`] construction. Log replay reads and
/// parses `read`; its final metadata projection emits `output`.
///
/// `read` contains consumer-requested stats and any additional stats needed by the skipping
/// predicate. `output` excludes predicate-only fields. For example, when the consumer requests
/// `name` stats and the predicate is `id > 400`, `read` contains `id` and `name`, while `output`
/// contains only `name`.
#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
pub(crate) struct ResolvedPhysicalStatsSchemas {
    /// Schema Kernel reads and parses before data skipping.
    read: SchemaRef,
    /// Schema Kernel returns to the scan consumer.
    output: Option<SchemaRef>,
}

impl ResolvedPhysicalStatsSchemas {
    /// Resolves the stats schemas and validates that every output field is read.
    fn try_new(read: Option<SchemaRef>, output: Option<SchemaRef>) -> KernelResult<Option<Self>> {
        match (read, output) {
            (None, None) => Ok(None),
            (Some(read), output) => {
                let schemas = Self { read, output };
                schemas.validate()?;
                Ok(Some(schemas))
            }
            (None, Some(_)) => Err(KernelError::internal_error(
                "stats output schema requires a stats read schema",
            )),
        }
    }

    /// Validates that every output field is present and compatible in the read schema.
    pub(crate) fn validate(&self) -> KernelResult<()> {
        if let Some(output) = &self.output {
            validate_stats_output_schema(&self.read, output, "")?;
        }
        Ok(())
    }
}

fn validate_stats_output_schema(
    read: &StructType,
    output: &StructType,
    parent: &str,
) -> KernelResult<()> {
    for output_field in output.fields() {
        let path = if parent.is_empty() {
            output_field.name().to_string()
        } else {
            format!("{parent}.{}", output_field.name())
        };
        let read_field = read.field(output_field.name()).ok_or_else(|| {
            KernelError::internal_error(format!(
                "stats output field '{path}' is missing from the read schema"
            ))
        })?;

        if read_field.is_nullable() != output_field.is_nullable() {
            return Err(KernelError::internal_error(format!(
                "stats output field '{path}' has incompatible nullability in the read schema"
            )));
        }

        match (read_field.data_type(), output_field.data_type()) {
            (DataType::Struct(read), DataType::Struct(output)) => {
                validate_stats_output_schema(read, output, &path)?;
            }
            (read, output) if read == output => {}
            _ => {
                return Err(KernelError::internal_error(format!(
                    "stats output field '{path}' has an incompatible type in the read schema"
                )));
            }
        }
    }
    Ok(())
}

/// All the state needed to process a scan.
#[derive(Debug, Clone)]
pub(crate) struct StateInfo {
    /// The logical schema for this scan
    pub(crate) logical_schema: SchemaRef,
    /// The physical schema to read from parquet files
    pub(crate) physical_schema: SchemaRef,
    /// The physical predicate for data skipping
    pub(crate) physical_predicate: PhysicalPredicate,
    /// Transform specification for converting physical to logical data
    pub(crate) transform_spec: Option<Arc<TransformSpec>>,
    /// The column mapping mode for this scan
    pub(crate) column_mapping_mode: ColumnMappingMode,
    /// Physical statistics schemas resolved while building the scan.
    pub(crate) physical_stats_schemas: Option<ResolvedPhysicalStatsSchemas>,
    /// Physical partition schema with native types for `partitionValues_parsed`. Fields use
    /// physical column names (for column mapping) and are always nullable. Present when the
    /// table has partition columns and either a predicate is provided (narrowed to
    /// predicate-referenced columns, for partition pruning) or the engine requested the typed
    /// struct in scan output (all partition columns).
    pub(crate) physical_partition_schema: Option<SchemaRef>,
    /// Physical leaf paths eligible for data skipping.
    ///
    /// This combines the table's indexed columns with caller-requested columns and gates which
    /// predicate references may use stats. It can be broader than
    /// [`Self::requested_physical_stats_columns`], which preserves the caller's selection for
    /// `stats_parsed` output.
    pub(crate) eligible_physical_stats_columns: HashSet<ColumnName>,
    /// Caller-requested physical stats columns used for data skipping and `stats_parsed` output.
    /// `Columns` resolves names strictly; `All` resolves them best-effort.
    pub(crate) requested_physical_stats_columns: Vec<ColumnName>,
    /// Whether the table is catalog-managed, used to label scan metric events. Converted to a
    /// [`TableType`](crate::metrics::TableType) at event construction.
    pub(crate) is_catalog_managed: bool,
    /// When set, log replay does not build per-file transform expressions:
    /// `parse_partition_values` and `get_transform_expr` are skipped and every
    /// `ScanMetadata::scan_file_transforms` entry is left `None`. `transform_spec` is retained
    /// so the scan can still describe the transform structurally. Set by
    /// [`ScanBuilder::without_row_transforms`](crate::scan::ScanBuilder::without_row_transforms).
    pub(crate) skip_row_transforms: bool,
    /// Table root for normalizing paths. `Some` iff adaptiveMetadata is enabled.
    pub(crate) dv_normalization_root: Option<Url>,
}

/// Validating the metadata columns also extracts information needed to properly construct the full
/// `StateInfo`. We use this struct to group this information so it can be cleanly passed back from
/// `validate_metadata_columns`
#[derive(Default)]
struct MetadataInfo<'a> {
    /// What are the names of the requested metadata fields
    metadata_field_names: HashSet<&'a String>,
    /// The name of the column that's selecting row indexes if that's been requested or None if
    /// they are not requested. We remember this if it's been requested explicitly. this is so
    /// we can reference this column and not re-add it as a requested column if we're _also_
    /// requesting row-ids.
    selected_row_index_col_name: Option<&'a String>,
    /// the materializedRowIdColumnName extracted from the table config if row ids are requested,
    /// or None if they are not requested
    materialized_row_id_column_name: Option<&'a String>,
    /// the materializedRowCommitVersionColumnName extracted from the table config if row commit
    /// versions are requested, or None if they are not requested
    materialized_row_commit_version_column_name: Option<&'a String>,
}

/// This validates that we have sensible metadata columns, and that the requested metadata is
/// supported by the table. Also computes and returns any extra info needed to build the transform
/// for the requested columns.
// Runs in O(supported_number_of_metadata_columns) time since each metadata
// column can appear at most once in the schema
fn validate_metadata_columns<'a>(
    logical_schema: &'a SchemaRef,
    table_configuration: &'a TableConfiguration,
) -> KernelResult<MetadataInfo<'a>> {
    let mut metadata_info = MetadataInfo::default();
    let partition_columns = table_configuration.logical_partition_columns();
    for metadata_column in logical_schema.metadata_columns() {
        // Ensure we don't have a metadata column with same name as a partition column
        if partition_columns.contains(metadata_column.name()) {
            return Err(KernelError::Schema(format!(
                "Metadata column names must not match partition columns: {}",
                metadata_column.name()
            )));
        }
        match metadata_column.get_metadata_column_spec() {
            Some(MetadataColumnSpec::RowIndex) => {
                metadata_info.selected_row_index_col_name = Some(metadata_column.name());
            }
            Some(MetadataColumnSpec::RowId) => {
                if table_configuration.table_properties().enable_row_tracking != Some(true) {
                    return Err(KernelError::unsupported(
                        "Row ids are not enabled on this table",
                    ));
                }
                let row_id_col = table_configuration
                    .metadata()
                    .configuration()
                    .get("delta.rowTracking.materializedRowIdColumnName")
                    .ok_or(KernelError::generic("No delta.rowTracking.materializedRowIdColumnName key found in metadata configuration"))?;
                metadata_info.materialized_row_id_column_name = Some(row_id_col);
            }
            Some(MetadataColumnSpec::RowCommitVersion) => {
                if !table_configuration.is_feature_enabled(&TableFeature::RowTracking) {
                    return Err(KernelError::unsupported(
                        "Row commit versions are not enabled on this table",
                    ));
                }
                let row_commit_version_col = table_configuration
                    .table_properties()
                    .materialized_row_commit_version_column_name
                    .as_ref()
                    .ok_or(KernelError::generic(
                        "No delta.rowTracking.materializedRowCommitVersionColumnName key found in \
                         metadata configuration",
                    ))?;
                metadata_info.materialized_row_commit_version_column_name =
                    Some(row_commit_version_col);
            }
            Some(MetadataColumnSpec::FilePath) => {
                // FilePath metadata column is handled by the parquet reader
            }
            None => {}
        }
        metadata_info
            .metadata_field_names
            .insert(metadata_column.name());
    }
    Ok(metadata_info)
}

/// Builds the physical stats and partition schemas used by scan metadata and data skipping.
///
/// `requested_physical_stats_columns` bypasses the table's indexed set and seeds the scan's stats
/// schema. Predicate references may add other indexed columns.
fn build_data_skipping_schemas(
    stats: &StatsOptions,
    physical_predicate: &PhysicalPredicate,
    predicate_column_names_logical: &[ColumnName],
    requested_physical_stats_columns: Option<&[ColumnName]>,
    table_configuration: &TableConfiguration,
) -> KernelResult<(Option<SchemaRef>, Option<SchemaRef>)> {
    // Narrow the table's typed partition schema to the columns the predicate references. The
    // DataSkippingFilter only needs partition columns that appear in the predicate, and the
    // shared helper forces every field nullable (MapToStruct can yield null for a missing key).
    let predicate_partition_schema = match physical_predicate {
        PhysicalPredicate::Some(pred, _ref_schema) => {
            let refs: Vec<ColumnName> = pred.references().into_iter().cloned().collect();
            table_configuration.predicate_partition_schema(&refs)
        }
        _ => None,
    };

    // `DataSkippingFilter` needs stats for every column its predicate references. Refs without
    // stats fold to NULL and pruning collapses to "keep every file", even when the caller
    // separately requested some other set of columns. Union predicate refs into the stats schema
    // so it serves both. Unresolvable refs (e.g. a predicate typo) are dropped here.
    let predicate_refs_physical =
        resolve_physical_columns(table_configuration, predicate_column_names_logical);

    // A stats schema with only `numRecords` and `tightBounds` (the bookkeeping fields the stats
    // schema builder always emits) has nothing to prune by. Return `None` in that case so the
    // caller skips building a `DataSkippingFilter`. `nullCount` is the per-column stats wrapper,
    // so its presence is the signal that at least one data column survived. The Delta protocol
    // allows `minValues` / `maxValues` without `nullCount`, but the builder always emits
    // `nullCount` whenever it emits min/max; this check relies on that implementation property.
    let build_stats_schema = |required: Option<&[ColumnName]>,
                              requested: Option<&[ColumnName]>|
     -> KernelResult<Option<SchemaRef>> {
        let builder = table_configuration
            .stats_schema_builder()
            .with_required_physical_columns(required)
            .with_requested_physical_columns(requested)
            .with_variant_min_max(stats.variant_min_max);
        #[cfg(feature = "geo-type-in-dev")]
        let builder = builder.with_geometry_min_max(stats.geometry_min_max);
        let stats_schema = builder.build()?;
        Ok(stats_schema
            .field(NULL_COUNT)
            .is_some()
            .then_some(stats_schema))
    };

    let stats_schema = match (&stats.struct_stats, physical_predicate) {
        (StructStats::AllIndexed { .. }, _) => {
            build_stats_schema(requested_physical_stats_columns, None)?
        }
        // Requested columns bypass the indexed set and seed the stats schema; predicate refs join
        // the schema so kernel can still prune.
        (StructStats::Columns { .. }, _) if requested_physical_stats_columns.is_some() => {
            let mut filter = requested_physical_stats_columns
                .unwrap_or_default()
                .to_vec();
            union_extra_into_filter(&mut filter, &predicate_refs_physical);
            build_stats_schema(requested_physical_stats_columns, Some(&filter))?
        }
        // No requested columns, but a predicate is present. Use just the predicate refs so the
        // stats schema is trimmed to what the rewritten predicate needs.
        (_, PhysicalPredicate::Some(_, _)) => {
            build_stats_schema(None, Some(&predicate_refs_physical))?
        }
        // No struct stats requested and no predicate: nothing to read or emit, so no stats schema.
        (_, _) => None,
    };
    Ok((stats_schema, predicate_partition_schema))
}

/// Resolves logical column names best-effort, warning and omitting names that cannot be resolved.
/// Used for predicate references.
fn resolve_physical_columns(
    table_configuration: &TableConfiguration,
    logical: &[ColumnName],
) -> Vec<ColumnName> {
    resolve_physical_columns_with_warnings(table_configuration, logical, true)
}

fn resolve_physical_columns_without_warnings(
    table_configuration: &TableConfiguration,
    logical: &[ColumnName],
) -> Vec<ColumnName> {
    // The output-schema resolver reports unresolved extra-indexed columns once.
    resolve_physical_columns_with_warnings(table_configuration, logical, false)
}

fn resolve_physical_columns_with_warnings(
    table_configuration: &TableConfiguration,
    logical: &[ColumnName],
    warn_on_error: bool,
) -> Vec<ColumnName> {
    let logical_schema = table_configuration.logical_schema();
    let column_mapping_mode = table_configuration.column_mapping_mode();
    logical
        .iter()
        .filter_map(|col| {
            get_any_level_column_physical_name(&logical_schema, col, column_mapping_mode)
                .inspect_err(|e| {
                    if warn_on_error {
                        warn!("Failed to resolve physical name for stats column {col}: {e}");
                    }
                })
                .ok()
        })
        .collect()
}

/// Resolves every logical column name to its physical name.
///
/// Returns an error if any name cannot be resolved. Used for the `requested` columns in
/// [`StructStats::Columns`].
fn resolve_physical_columns_strict(
    table_configuration: &TableConfiguration,
    logical: &[ColumnName],
) -> KernelResult<Vec<ColumnName>> {
    let logical_schema = table_configuration.logical_schema();
    let column_mapping_mode = table_configuration.column_mapping_mode();
    logical
        .iter()
        .map(|col| get_any_level_column_physical_name(&logical_schema, col, column_mapping_mode))
        .collect()
}

/// Adds to `filter` every entry of `extra` not already present.
fn union_extra_into_filter(filter: &mut Vec<ColumnName>, extra: &[ColumnName]) {
    let to_add: Vec<ColumnName> = extra
        .iter()
        .filter(|c| !filter.contains(c))
        .cloned()
        .collect();
    filter.extend(to_add);
}

impl StateInfo {
    /// Create StateInfo with a custom field classifier for different scan types.
    /// Get the state needed to process a scan.
    ///
    /// `logical_read_schema` - The logical schema of the scan output
    /// `table_schema` - The schema against which predicate column references are resolved.
    /// Must contain every column the predicate may legitimately reference (typically the full
    /// table schema, or full CDF-extended schema for CDF scans). Currently, we do not carry
    /// over any metadata columns from the `logical_read_schema` to the `table_schema` (issue
    /// 2633).
    /// `table_configuration` - The TableConfiguration for this table
    /// `predicate` - Optional predicate to filter data during the scan
    /// `stats` - Engine-facing stats options. Drives which stats columns appear in scan
    ///   metadata output and whether the JSON synthesis fallback fires.
    /// `partition_values` - Engine-facing partition value options. Drives whether the typed
    ///   `partitionValues_parsed` column appears in scan metadata output.
    /// `classifier` - The classifier to use for different scan types. Use `()` if not needed
    pub(crate) fn try_new<C: TransformFieldClassifier>(
        logical_read_schema: SchemaRef,
        table_schema: SchemaRef,
        table_configuration: &TableConfiguration,
        predicate: Option<PredicateRef>,
        stats: &StatsOptions,
        partition_values: &PartitionValuesOptions,
        classifier: C,
    ) -> KernelResult<Self> {
        stats.validate()?;
        let partition_columns = table_configuration.logical_partition_columns();
        let column_mapping_mode = table_configuration.column_mapping_mode();
        let mut read_fields = Vec::with_capacity(logical_read_schema.num_fields());
        let mut transform_spec = Vec::with_capacity(logical_read_schema.num_fields());
        let mut last_physical_field: Option<String> = None;

        let metadata_info = validate_metadata_columns(&logical_read_schema, table_configuration)?;

        // Loop over all selected fields and build both the physical schema and transform spec
        for (index, logical_field) in logical_read_schema.fields().enumerate() {
            if let Some(spec) =
                classifier.classify_field(logical_field, index, &last_physical_field)
            {
                // Classifier has handled this field via a transformation, just push it and move on
                transform_spec.push(spec);
            } else if partition_columns.contains(logical_field.name()) {
                // push the transform for this partition column
                transform_spec.push(FieldTransformSpec::MetadataDerivedColumn {
                    field_index: index,
                    insert_after: last_physical_field.clone(),
                });
            } else {
                // Regular field field or a metadata column, figure out which and handle it
                match logical_field.get_metadata_column_spec() {
                    Some(MetadataColumnSpec::RowId) => {
                        let index_column_name = match metadata_info.selected_row_index_col_name {
                            Some(index_column_name) => index_column_name.to_string(),
                            None => {
                                // the index column isn't being explicitly requested, so add it to
                                // `read_fields` so the parquet_reader will generate it, and add a
                                // transform to drop it before returning logical data

                                // ensure we have a column name that isn't already in our schema
                                let index_column_name = (0..)
                                    .map(|i| format!("row_indexes_for_row_id_{i}"))
                                    .find(|name| logical_read_schema.field(name).is_none())
                                    .ok_or(KernelError::generic(
                                        "Couldn't generate row index column name",
                                    ))?;
                                read_fields.push(StructField::create_metadata_column(
                                    &index_column_name,
                                    MetadataColumnSpec::RowIndex,
                                ));
                                transform_spec.push(FieldTransformSpec::StaticDrop {
                                    field_name: index_column_name.clone(),
                                });
                                index_column_name
                            }
                        };
                        let Some(row_id_col_name) = metadata_info.materialized_row_id_column_name
                        else {
                            return Err(KernelError::internal_error(
                                "Should always return a materialized_row_id_column_name if selecting row ids"
                            ));
                        };

                        let row_id_col_name = row_id_col_name.to_string();
                        read_fields.push(StructField::nullable(&row_id_col_name, DataType::LONG));
                        transform_spec.push(FieldTransformSpec::GenerateRowId {
                            field_name: row_id_col_name.clone(),
                            row_index_field_name: index_column_name,
                        });
                        last_physical_field = Some(row_id_col_name);
                    }
                    Some(MetadataColumnSpec::RowCommitVersion) => {
                        let Some(row_commit_version_col_name) =
                            metadata_info.materialized_row_commit_version_column_name
                        else {
                            return Err(KernelError::internal_error(
                                "missing materialized Row Commit Version column name when row \
                                 tracking is enabled",
                            ));
                        };

                        let row_commit_version_col_name = row_commit_version_col_name.to_string();
                        read_fields.push(StructField::nullable(
                            &row_commit_version_col_name,
                            DataType::LONG,
                        ));
                        transform_spec.push(FieldTransformSpec::GenerateRowCommitVersion {
                            field_name: row_commit_version_col_name.clone(),
                        });
                        last_physical_field = Some(row_commit_version_col_name);
                    }
                    Some(MetadataColumnSpec::RowIndex)
                    | Some(MetadataColumnSpec::FilePath)
                    | None => {
                        // note that RowIndex and FilePath are handled in the parquet reader so we
                        // just add them as if they're normal physical
                        // columns
                        let physical_field = logical_field.make_physical(column_mapping_mode)?;
                        debug!("\n\n{logical_field:#?}\nAfter mapping: {physical_field:#?}\n\n");
                        let physical_name = physical_field.name.clone();

                        if !logical_field.is_metadata_column()
                            && metadata_info.metadata_field_names.contains(&physical_name)
                        {
                            return Err(KernelError::Schema(format!(
                                "Metadata column names must not match physical columns, but logical column '{}' has physical name '{}'",
                                logical_field.name(), physical_name,
                            )));
                        }
                        last_physical_field = Some(physical_name);
                        read_fields.push(physical_field);
                    }
                }
            }
        }

        let physical_schema = Arc::new(StructType::try_new(read_fields)?);

        // Logical column names referenced by the predicate. Fed into the stats schema
        // build below and into the dropped-refs observability log.
        let predicate_column_names: Vec<ColumnName> = predicate
            .as_ref()
            .map(|p| p.references().into_iter().cloned().collect())
            .unwrap_or_default();

        // We use table_schema here as predicate can reference columns outside projection.
        let physical_predicate = match predicate {
            Some(pred) => PhysicalPredicate::try_new(&pred, &table_schema, column_mapping_mode)?,
            None => PhysicalPredicate::None,
        };

        // Resolve requested names once for both stats eligibility and schema construction.
        // `Columns` is strict; `AllIndexed` treats extra-indexed names as best-effort hints.
        let requested_physical_stats_columns: Vec<ColumnName> = match &stats.struct_stats {
            StructStats::AllIndexed { extra_indexed } => {
                resolve_physical_columns_without_warnings(table_configuration, extra_indexed)
            }
            StructStats::Columns { requested } => {
                resolve_physical_columns_strict(table_configuration, requested)?
            }
            StructStats::None => Vec::new(),
        };
        let requested_physical_stats_columns_ref = (!requested_physical_stats_columns.is_empty())
            .then_some(requested_physical_stats_columns.as_slice());

        // Stats-eligible column set. Partition columns are excluded; they flow through
        // `partitionValues_parsed` instead.
        let eligible_physical_stats_columns =
            table_configuration.physical_stats_columns_set(requested_physical_stats_columns_ref);
        // Observability: predicate refs outside `eligible_physical_stats_columns` fold to NULL
        // by the gate. Surface the dropped set so an engine operator can see what got folded.
        // The filter walk is bounded by predicate width but still does a physical-name
        // resolution per ref, so gate it on the log level to skip the work when DEBUG is off.
        if enabled!(Level::DEBUG) && matches!(physical_predicate, PhysicalPredicate::Some(_, _)) {
            let dropped: Vec<&ColumnName> = predicate_column_names
                .iter()
                .filter(|c| {
                    get_any_level_column_physical_name(&table_schema, c, column_mapping_mode)
                        .ok()
                        .is_some_and(|physical| {
                            !eligible_physical_stats_columns.contains(&physical)
                        })
                })
                .collect();
            if !dropped.is_empty() {
                debug!(
                    "Checkpoint pushdown: predicate refs to non-stats columns folded to NULL: {:?}",
                    dropped
                );
            }
        }

        // Build partition schema with physical names, used for partition pruning in data
        // skipping and for the engine-facing `partitionValues_parsed` output column. Needed
        // when partition columns exist and either a predicate is present or the engine
        // requested the typed struct in scan output.
        // partition_columns stores logical names (per Delta protocol), so we zip the table's
        // logical and physical schemas (same field ordering, guaranteed by `make_physical`)
        // to match logical names and extract the corresponding physical fields without
        // per-field metadata lookups.
        let has_predicate = !matches!(
            physical_predicate,
            PhysicalPredicate::None | PhysicalPredicate::StaticSkipAll
        );
        let table_partition_schema =
            if (has_predicate || partition_values.parsed_struct) && !partition_columns.is_empty() {
                let partition_fields: Vec<StructField> = table_configuration
                    .logical_schema()
                    .fields()
                    .zip(table_configuration.physical_schema().fields())
                    .filter(|(logical_f, _)| partition_columns.contains(logical_f.name()))
                    .map(|(_, physical_f)| physical_f.clone())
                    .collect();
                if partition_fields.is_empty() {
                    None
                } else {
                    Some(Arc::new(StructType::new_unchecked(partition_fields)))
                }
            } else {
                None
            };

        let (physical_stats_read_schema, predicate_partition_schema) = build_data_skipping_schemas(
            stats,
            &physical_predicate,
            &predicate_column_names,
            requested_physical_stats_columns_ref,
            table_configuration,
        )?;
        let physical_stats_output_schema =
            super::build_stats_output_schemas(table_configuration, stats)?
                .map(|schemas| schemas.physical);
        let physical_stats_schemas = ResolvedPhysicalStatsSchemas::try_new(
            physical_stats_read_schema,
            physical_stats_output_schema,
        )?;

        // When the engine requested the typed struct, emit all partition columns rather than
        // the predicate-narrowed subset. The data skipping filter only references the columns
        // its predicate needs, so the superset is safe for pruning too. All fields are forced
        // nullable: MapToStruct can yield null even for a non-nullable column (a missing key, or
        // an empty string cast to a non-string/binary type), and a non-nullable field would then
        // error.
        let physical_partition_schema = if partition_values.parsed_struct {
            table_partition_schema.map(|tps| {
                let nullable_fields = tps
                    .fields()
                    .map(|f| StructField::nullable(f.name(), f.data_type().clone()));
                Arc::new(StructType::new_unchecked(nullable_fields))
            })
        } else {
            predicate_partition_schema
        };

        let transform_spec =
            if !transform_spec.is_empty() || column_mapping_mode != ColumnMappingMode::None {
                Some(Arc::new(transform_spec))
            } else {
                None
            };

        Ok(StateInfo {
            logical_schema: logical_read_schema,
            physical_schema,
            physical_predicate,
            transform_spec,
            column_mapping_mode,
            physical_stats_schemas,
            physical_partition_schema,
            eligible_physical_stats_columns,
            requested_physical_stats_columns,
            is_catalog_managed: table_configuration.is_catalog_managed(),
            skip_row_transforms: false,
            dv_normalization_root: table_configuration.dv_normalization_root(),
        })
    }

    /// Returns a conservative initial capacity for the dedup `HashSet` in
    /// [`ScanLogReplayProcessor`].
    ///
    /// The exact file count is not available at this point, so the hint is
    /// derived from whether stats are enabled: stats are only computed for
    /// non-trivial tables, so their presence is a reasonable proxy for table
    /// size. Using 4096 vs 512 as the two tiers eliminates the first 12-14
    /// hashbrown doubling events for medium/large tables while staying cheap
    /// for small ones.
    pub(crate) fn dedup_capacity_hint(&self) -> usize {
        if self.physical_stats_schemas.is_some() {
            4096
        } else {
            512
        }
    }

    /// Returns the physical stats schema Kernel reads and parses.
    pub(crate) fn physical_stats_read_schema(&self) -> Option<&SchemaRef> {
        self.physical_stats_schemas
            .as_ref()
            .map(|schemas| &schemas.read)
    }

    /// Returns the physical stats schema exposed to the scan consumer.
    pub(crate) fn physical_stats_output_schema(&self) -> Option<&SchemaRef> {
        self.physical_stats_schemas
            .as_ref()
            .and_then(|schemas| schemas.output.as_ref())
    }
}

#[cfg(test)]
pub(crate) mod tests {
    use std::collections::HashMap;
    use std::sync::Arc;

    use rstest::rstest;

    use super::*;
    use crate::actions::{MAX_VALUES, MIN_VALUES};
    use crate::expressions::{col, column_name, lit, Predicate as Pred};
    #[cfg(feature = "geo-type-in-dev")]
    use crate::scan::build_stats_output_schemas;
    #[cfg(feature = "geo-type-in-dev")]
    use crate::schema::GeometryType;
    use crate::schema::{schema, schema_ref, ColumnMetadataKey, MetadataValue};
    use crate::table_features::TableFeature;
    use crate::unit_test_utils::{
        assert_result_error_with_message, MockProtocolBuilder, MockTableConfigurationBuilder,
    };
    use crate::KernelResult;

    #[test]
    fn stats_schemas_can_both_be_absent() {
        assert!(ResolvedPhysicalStatsSchemas::try_new(None, None)
            .unwrap()
            .is_none());
    }

    #[test]
    fn stats_read_schema_can_exist_without_output() {
        let read = schema_ref! { nullable "id": LONG };
        let schemas = ResolvedPhysicalStatsSchemas::try_new(Some(read.clone()), None)
            .unwrap()
            .unwrap();

        assert_eq!(schemas.read, read);
        assert!(schemas.output.is_none());
    }

    #[test]
    fn stats_output_schema_can_be_a_nested_subset_of_read_schema() {
        let read = schema_ref! {
            nullable "id": LONG,
            nullable "nested": {
                nullable "name": STRING,
                nullable "value": INTEGER,
            },
        };
        let output = schema_ref! {
            nullable "nested": { nullable "name": STRING },
        };

        let schemas =
            ResolvedPhysicalStatsSchemas::try_new(Some(read), Some(output.clone())).unwrap();

        assert_eq!(schemas.unwrap().output, Some(output));
    }

    #[test]
    fn stats_output_schema_requires_read_schema() {
        let output = schema_ref! { nullable "id": LONG };

        assert_result_error_with_message(
            ResolvedPhysicalStatsSchemas::try_new(None, Some(output)),
            "stats output schema requires a stats read schema",
        );
    }

    #[test]
    fn stats_output_fields_must_exist_in_read_schema() {
        let read = schema_ref! { nullable "id": LONG };
        let output = schema_ref! { nullable "name": STRING };

        assert_result_error_with_message(
            ResolvedPhysicalStatsSchemas::try_new(Some(read), Some(output)),
            "stats output field 'name' is missing from the read schema",
        );
    }

    #[test]
    fn stats_output_fields_must_match_read_types() {
        let read = schema_ref! { nullable "id": LONG };
        let output = schema_ref! { nullable "id": STRING };

        assert_result_error_with_message(
            ResolvedPhysicalStatsSchemas::try_new(Some(read), Some(output)),
            "stats output field 'id' has an incompatible type in the read schema",
        );
    }

    // get a state info with no predicate or extra metadata
    pub(crate) fn get_simple_state_info(
        schema: SchemaRef,
        partition_columns: Vec<String>,
    ) -> KernelResult<StateInfo> {
        get_state_info(schema, partition_columns, None, &[], HashMap::new(), vec![])
    }

    /// When features are non-empty, uses protocol (3,7) with explicit feature lists.
    /// When features are empty, uses legacy protocol (2,5).
    pub(crate) fn get_state_info(
        schema: SchemaRef,
        partition_columns: Vec<String>,
        predicate: Option<PredicateRef>,
        features: &[TableFeature],
        metadata_configuration: HashMap<String, String>,
        metadata_cols: Vec<(&str, MetadataColumnSpec)>,
    ) -> KernelResult<StateInfo> {
        get_state_info_with_stats(
            schema,
            partition_columns,
            predicate,
            features,
            metadata_configuration,
            metadata_cols,
            StatsOptions::default(),
        )
    }

    pub(crate) fn get_state_info_with_stats(
        schema: SchemaRef,
        partition_columns: Vec<String>,
        predicate: Option<PredicateRef>,
        features: &[TableFeature],
        metadata_configuration: HashMap<String, String>,
        metadata_cols: Vec<(&str, MetadataColumnSpec)>,
        stats: StatsOptions,
    ) -> KernelResult<StateInfo> {
        get_state_info_with_options(
            schema,
            partition_columns,
            predicate,
            features,
            metadata_configuration,
            metadata_cols,
            stats,
            PartitionValuesOptions::default(),
        )
    }

    #[allow(clippy::too_many_arguments)]
    pub(crate) fn get_state_info_with_options(
        schema: SchemaRef,
        partition_columns: Vec<String>,
        predicate: Option<PredicateRef>,
        features: &[TableFeature],
        metadata_configuration: HashMap<String, String>,
        metadata_cols: Vec<(&str, MetadataColumnSpec)>,
        stats: StatsOptions,
        partition_values: PartitionValuesOptions,
    ) -> KernelResult<StateInfo> {
        let builder = MockTableConfigurationBuilder::new()
            .with_schema(schema.clone())
            .with_partition_columns(partition_columns)
            .with_properties(metadata_configuration)
            .with_table_root("s3://my-table")
            .with_version(1);
        let builder = match features.is_empty() {
            true => builder.with_protocol(MockProtocolBuilder::new().with_versions(2, 5).build()),
            false => {
                builder.with_protocol(MockProtocolBuilder::new().with_features(features).build())
            }
        };
        let table_configuration = builder.try_build()?;

        let mut schema = schema;
        for (name, spec) in metadata_cols.into_iter() {
            schema = Arc::new(
                schema
                    .add_metadata_column(name, spec)
                    .expect("Couldn't add metadata col"),
            );
        }

        StateInfo::try_new(
            schema.clone(),
            table_configuration.logical_schema(),
            &table_configuration,
            predicate,
            &stats,
            &partition_values,
            (),
        )
    }

    pub(crate) fn assert_transform_spec(
        transform_spec: &TransformSpec,
        requested_row_indexes: bool,
        expected_row_id_name: &str,
        expected_row_index_name: &str,
    ) {
        // if we requested row indexes, there's only one transform for the row id col, otherwise the
        // first transform drops the row index column, and the second one adds the row ids
        let expected_transform_count = if requested_row_indexes { 1 } else { 2 };
        let generate_offset = if requested_row_indexes { 0 } else { 1 };

        assert_eq!(transform_spec.len(), expected_transform_count);

        if !requested_row_indexes {
            // ensure we have a drop transform if we didn't request row indexes
            match &transform_spec[0] {
                FieldTransformSpec::StaticDrop { field_name } => {
                    assert_eq!(field_name, expected_row_index_name);
                }
                _ => panic!("Expected StaticDrop transform"),
            }
        }

        match &transform_spec[generate_offset] {
            FieldTransformSpec::GenerateRowId {
                field_name,
                row_index_field_name,
            } => {
                assert_eq!(field_name, expected_row_id_name);
                assert_eq!(row_index_field_name, expected_row_index_name);
            }
            _ => panic!("Expected GenerateRowId transform"),
        }
    }

    #[test]
    fn no_partition_columns() {
        // Test case: No partition columns, no column mapping
        let schema = schema_ref! {
            nullable "id": STRING,
            nullable "value": LONG,
        };

        let state_info = get_simple_state_info(schema.clone(), vec![]).unwrap();

        // Should have no transform spec (no partitions, no column mapping)
        assert!(state_info.transform_spec.is_none());

        // Physical schema should match logical schema
        assert_eq!(state_info.logical_schema, schema);
        assert_eq!(state_info.physical_schema.fields().len(), 2);

        // No predicate
        assert_eq!(state_info.physical_predicate, PhysicalPredicate::None);
    }

    #[test]
    fn with_partition_columns() {
        // Test case: With partition columns
        let schema = schema_ref! {
            nullable "id": STRING,
            nullable "date": DATE, // Partition column
            nullable "value": LONG,
        };

        let state_info = get_simple_state_info(
            schema.clone(),
            vec!["date".to_string()], // date is a partition column
        )
        .unwrap();

        // Should have a transform spec for the partition column
        assert!(state_info.transform_spec.is_some());
        let transform_spec = state_info.transform_spec.as_ref().unwrap();
        assert_eq!(transform_spec.len(), 1);

        // Check the transform spec for the partition column
        match &transform_spec[0] {
            FieldTransformSpec::MetadataDerivedColumn {
                field_index,
                insert_after,
            } => {
                assert_eq!(*field_index, 1); // Index of "date" in logical schema
                assert_eq!(insert_after, &Some("id".to_string())); // After "id" which is physical
            }
            _ => panic!("Expected MetadataDerivedColumn transform"),
        }

        // Physical schema should not include partition column
        assert_eq!(state_info.logical_schema, schema);
        assert_eq!(state_info.physical_schema.fields().len(), 2); // Only id and value
    }

    #[test]
    fn multiple_partition_columns() {
        // Test case: Multiple partition columns interspersed with regular columns
        let schema = schema_ref! {
            nullable "col1": STRING,
            nullable "part1": STRING, // Partition
            nullable "col2": LONG,
            nullable "part2": INTEGER, // Partition
        };

        let state_info = get_simple_state_info(
            schema.clone(),
            vec!["part1".to_string(), "part2".to_string()],
        )
        .unwrap();

        // Should have transforms for both partition columns
        assert!(state_info.transform_spec.is_some());
        let transform_spec = state_info.transform_spec.as_ref().unwrap();
        assert_eq!(transform_spec.len(), 2);

        // Check first partition column transform
        match &transform_spec[0] {
            FieldTransformSpec::MetadataDerivedColumn {
                field_index,
                insert_after,
            } => {
                assert_eq!(*field_index, 1); // Index of "part1"
                assert_eq!(insert_after, &Some("col1".to_string()));
            }
            _ => panic!("Expected MetadataDerivedColumn transform"),
        }

        // Check second partition column transform
        match &transform_spec[1] {
            FieldTransformSpec::MetadataDerivedColumn {
                field_index,
                insert_after,
            } => {
                assert_eq!(*field_index, 3); // Index of "part2"
                assert_eq!(insert_after, &Some("col2".to_string()));
            }
            _ => panic!("Expected MetadataDerivedColumn transform"),
        }

        // Physical schema should only have non-partition columns
        assert_eq!(state_info.physical_schema.fields().len(), 2); // col1 and col2
    }

    #[test]
    fn with_predicate() {
        // Test case: With a valid predicate
        let schema = schema_ref! {
            nullable "id": STRING,
            nullable "value": LONG,
        };

        let predicate = Arc::new(col!("value").gt(lit(10i64)));

        let state_info = get_state_info(
            schema.clone(),
            vec![], // no partition columns
            Some(predicate),
            &[],            // no table features
            HashMap::new(), // no extra metadata
            vec![],         // no metadata
        )
        .unwrap();

        // Should have a physical predicate
        match &state_info.physical_predicate {
            PhysicalPredicate::Some(_pred, schema) => {
                // Physical predicate exists
                assert_eq!(schema.fields().len(), 1); // Only "value" is referenced
            }
            _ => panic!("Expected PhysicalPredicate::Some"),
        }
    }

    #[test]
    fn partition_at_beginning() {
        // Test case: Partition column at the beginning
        let schema = schema_ref! {
            nullable "date": DATE, // Partition column
            nullable "id": STRING,
            nullable "value": LONG,
        };

        let state_info = get_simple_state_info(schema.clone(), vec!["date".to_string()]).unwrap();

        // Should have a transform spec for the partition column
        let transform_spec = state_info.transform_spec.as_ref().unwrap();
        assert_eq!(transform_spec.len(), 1);

        match &transform_spec[0] {
            FieldTransformSpec::MetadataDerivedColumn {
                field_index,
                insert_after,
            } => {
                assert_eq!(*field_index, 0); // Index of "date"
                assert_eq!(insert_after, &None); // No physical field before it, so prepend
            }
            _ => panic!("Expected MetadataDerivedColumn transform"),
        }
    }

    pub(crate) const ROW_TRACKING_FEATURES: &[TableFeature] =
        &[TableFeature::RowTracking, TableFeature::DomainMetadata];

    // TODO(#3248): Add tests for row id.
    #[derive(Clone, Copy, Debug, PartialEq, Eq)]
    pub(crate) enum RowTrackingState {
        Unsupported,
        SupportedNotEnabled,
        Enabled,
        Suspended,
    }

    impl RowTrackingState {
        pub(crate) fn features(self) -> &'static [TableFeature] {
            match self {
                Self::Unsupported => &[],
                Self::SupportedNotEnabled | Self::Enabled | Self::Suspended => {
                    ROW_TRACKING_FEATURES
                }
            }
        }

        pub(crate) fn properties(self) -> HashMap<String, String> {
            let mut properties = get_string_map(&[
                (
                    "delta.rowTracking.materializedRowIdColumnName",
                    "row_id_col",
                ),
                (
                    "delta.rowTracking.materializedRowCommitVersionColumnName",
                    "row_commit_version_col",
                ),
            ]);
            if self == Self::Enabled {
                properties.insert("delta.enableRowTracking".to_string(), "true".to_string());
            }
            if self == Self::Suspended {
                properties.insert("delta.enableRowTracking".to_string(), "false".to_string());
                properties.insert("delta.rowTrackingSuspended".to_string(), "true".to_string());
            }
            properties
        }
    }

    fn get_string_map(slice: &[(&str, &str)]) -> HashMap<String, String> {
        slice
            .iter()
            .map(|(k, v)| (k.to_string(), v.to_string()))
            .collect()
    }

    /// Builds a nullable [`StructField`] carrying column-mapping id + physical name metadata.
    fn cm_field(name: &str, id: i64, physical_name: &str, ty: impl Into<DataType>) -> StructField {
        StructField::nullable(name, ty).with_metadata([
            (
                ColumnMetadataKey::ColumnMappingId.as_ref(),
                MetadataValue::Number(id),
            ),
            (
                ColumnMetadataKey::ColumnMappingPhysicalName.as_ref(),
                MetadataValue::String(physical_name.into()),
            ),
        ])
    }

    #[test]
    fn request_row_ids() {
        let schema = schema_ref! { nullable "id": STRING };

        let state_info = get_state_info(
            schema.clone(),
            vec![],
            None,
            ROW_TRACKING_FEATURES,
            get_string_map(&[
                ("delta.enableRowTracking", "true"),
                (
                    "delta.rowTracking.materializedRowIdColumnName",
                    "some_row_id_col",
                ),
                (
                    "delta.rowTracking.materializedRowCommitVersionColumnName",
                    "some_row_commit_version_col",
                ),
            ]),
            vec![("row_id", MetadataColumnSpec::RowId)],
        )
        .unwrap();

        // Should have a transform spec for the row_id column
        let transform_spec = state_info.transform_spec.as_ref().unwrap();
        assert_transform_spec(
            transform_spec,
            false, // we did not request row indexes
            "some_row_id_col",
            "row_indexes_for_row_id_0",
        );
    }

    #[test]
    fn request_row_commit_versions() {
        let schema = schema_ref! { nullable "id": STRING };
        let state_info = get_state_info(
            schema,
            vec![],
            None,
            ROW_TRACKING_FEATURES,
            get_string_map(&[
                ("delta.enableRowTracking", "true"),
                (
                    "delta.rowTracking.materializedRowIdColumnName",
                    "some_row_id_col",
                ),
                (
                    "delta.rowTracking.materializedRowCommitVersionColumnName",
                    "some_row_commit_version_col",
                ),
            ]),
            vec![("row_commit_version", MetadataColumnSpec::RowCommitVersion)],
        )
        .unwrap();

        assert_eq!(
            state_info
                .physical_schema
                .field("some_row_commit_version_col")
                .map(StructField::data_type),
            Some(&DataType::LONG)
        );
        assert_eq!(
            state_info.transform_spec.as_deref().map(Vec::as_slice),
            Some(
                [FieldTransformSpec::GenerateRowCommitVersion {
                    field_name: "some_row_commit_version_col".to_string(),
                }]
                .as_slice()
            )
        );
    }

    #[test]
    fn request_row_ids_conflicting_row_index_col_name() {
        // "row_indexes_for_row_id_0" conflicts with the first generated name for row indexes
        let schema = schema_ref! { nullable "row_indexes_for_row_id_0": STRING };

        let state_info = get_state_info(
            schema.clone(),
            vec![],
            None,
            ROW_TRACKING_FEATURES,
            get_string_map(&[
                ("delta.enableRowTracking", "true"),
                (
                    "delta.rowTracking.materializedRowIdColumnName",
                    "some_row_id_col",
                ),
                (
                    "delta.rowTracking.materializedRowCommitVersionColumnName",
                    "some_row_commit_version_col",
                ),
            ]),
            vec![("row_id", MetadataColumnSpec::RowId)],
        )
        .unwrap();

        // Should have a transform spec for the row_id column
        let transform_spec = state_info.transform_spec.as_ref().unwrap();
        assert_transform_spec(
            transform_spec,
            false, // we did not request row indexes
            "some_row_id_col",
            "row_indexes_for_row_id_1", // ensure we didn't conflict with the col in the schema
        );
    }

    #[test]
    fn request_row_ids_and_indexes() {
        let schema = schema_ref! { nullable "id": STRING };

        let state_info = get_state_info(
            schema.clone(),
            vec![],
            None,
            ROW_TRACKING_FEATURES,
            get_string_map(&[
                ("delta.enableRowTracking", "true"),
                (
                    "delta.rowTracking.materializedRowIdColumnName",
                    "some_row_id_col",
                ),
                (
                    "delta.rowTracking.materializedRowCommitVersionColumnName",
                    "some_row_commit_version_col",
                ),
            ]),
            vec![
                ("row_id", MetadataColumnSpec::RowId),
                ("row_index", MetadataColumnSpec::RowIndex),
            ],
        )
        .unwrap();

        // Should have a transform spec for the row_id column
        let transform_spec = state_info.transform_spec.as_ref().unwrap();
        assert_transform_spec(
            transform_spec,
            true, // we did request row indexes
            "some_row_id_col",
            "row_index",
        );
    }

    #[test]
    fn invalid_row_id_config() {
        let schema = schema_ref! { nullable "id": STRING };

        // Row IDs requested but row tracking not enabled → error
        let res = get_state_info(
            schema.clone(),
            vec![],
            None,
            &[], // no table features
            HashMap::new(),
            vec![("row_id", MetadataColumnSpec::RowId)],
        );
        assert_result_error_with_message(res, "Unsupported: Row ids are not enabled on this table");

        // Row tracking enabled but missing materializedRowIdColumnName → error
        let res = get_state_info(
            schema,
            vec![],
            None,
            ROW_TRACKING_FEATURES,
            get_string_map(&[("delta.enableRowTracking", "true")]),
            vec![("row_id", MetadataColumnSpec::RowId)],
        );
        assert_result_error_with_message(
            res,
            "Generic delta kernel error: No delta.rowTracking.materializedRowIdColumnName key found in metadata configuration",
        );
    }

    #[rstest]
    #[case::unsupported(RowTrackingState::Unsupported)]
    #[case::supported_not_enabled(RowTrackingState::SupportedNotEnabled)]
    #[case::suspended(RowTrackingState::Suspended)]
    fn request_row_commit_versions_requires_enabled_row_tracking(
        #[case] row_tracking_state: RowTrackingState,
    ) {
        let schema = schema_ref! { nullable "id": STRING };
        let res = get_state_info(
            schema,
            vec![],
            None,
            row_tracking_state.features(),
            row_tracking_state.properties(),
            vec![("row_commit_version", MetadataColumnSpec::RowCommitVersion)],
        );
        assert_result_error_with_message(
            res,
            "Unsupported: Row commit versions are not enabled on this table",
        );
    }

    #[test]
    fn request_row_commit_versions_requires_materialized_column_name() {
        let schema = schema_ref! { nullable "id": STRING };
        let res = get_state_info(
            schema,
            vec![],
            None,
            ROW_TRACKING_FEATURES,
            get_string_map(&[("delta.enableRowTracking", "true")]),
            vec![("row_commit_version", MetadataColumnSpec::RowCommitVersion)],
        );
        assert_result_error_with_message(
            res,
            "No delta.rowTracking.materializedRowCommitVersionColumnName key found in metadata \
             configuration",
        );
    }

    #[test]
    fn metadata_column_matches_partition_column() {
        let table_configuration = MockTableConfigurationBuilder::new()
            .with_schema(schema! {
                nullable "id": STRING,
                nullable "part_col": STRING,
            })
            .with_partition_columns(["part_col"])
            .with_protocol(MockProtocolBuilder::new().with_versions(2, 5).build())
            .with_table_root("s3://my-table")
            .with_version(1)
            .build();

        let read_schema = schema_ref! { nullable "id": STRING };
        let read_schema = Arc::new(
            read_schema
                .add_metadata_column("part_col", MetadataColumnSpec::RowId)
                .expect("Couldn't add metadata col"),
        );
        let res = StateInfo::try_new(
            read_schema,
            table_configuration.logical_schema(),
            &table_configuration,
            None,
            &StatsOptions::default(),
            &PartitionValuesOptions::default(),
            (),
        );
        assert_result_error_with_message(
            res,
            "Schema error: Metadata column names must not match partition columns: part_col",
        );
    }

    #[test]
    fn metadata_column_matches_read_field() {
        let schema = schema_ref! { (cm_field("id", 1, "other", DataType::STRING)) };
        let res = get_state_info(
            schema.clone(),
            vec![],
            None,
            &[], // no table features
            get_string_map(&[("delta.columnMapping.mode", "name")]),
            vec![("other", MetadataColumnSpec::RowIndex)],
        );
        assert_result_error_with_message(
            res,
            "Schema error: Metadata column names must not match physical columns, but logical column 'id' has physical name 'other'"
        );
    }

    #[test]
    fn stats_columns_with_predicate() {
        let schema = schema_ref! {
            nullable "id": STRING,
            nullable "value": LONG,
        };

        let predicate = Arc::new(col!("value").gt(lit(10i64)));

        let state_info = get_state_info_with_stats(
            schema,
            vec![],
            Some(predicate),
            &[], // no table features
            HashMap::new(),
            vec![],
            StatsOptions::all(),
        )
        .unwrap();

        // The stats read schema should be set from expected_stats_schema.
        assert!(
            state_info.physical_stats_read_schema().is_some(),
            "physical_stats_read_schema should be Some when AllColumns is set"
        );
        // physical_predicate should still be active for data skipping
        assert!(
            matches!(state_info.physical_predicate, PhysicalPredicate::Some(..)),
            "physical_predicate should be PhysicalPredicate::Some for data skipping"
        );
    }

    #[test]
    fn stats_columns_with_predicate_merges_columns() {
        // When specific stats_columns are requested alongside a predicate, the stats
        // schema should include both the requested columns and predicate-referenced columns.
        let schema = schema_ref! {
            nullable "id": STRING,
            nullable "value": LONG,
            nullable "extra": LONG,
        };

        let predicate = Arc::new(col!("extra").gt(lit(5i64)));

        let state_info = get_state_info_with_stats(
            schema,
            vec![],
            Some(predicate),
            &[],
            HashMap::new(),
            vec![],
            StatsOptions {
                synthesize_json: true,
                struct_stats: StructStats::Columns {
                    requested: vec![column_name!("value")],
                },
                ..Default::default()
            },
        )
        .unwrap();

        let stats_schema = state_info
            .physical_stats_read_schema()
            .expect("should have physical stats schema");

        let min_values = stats_schema
            .field(MIN_VALUES)
            .expect("should have minValues");
        if let DataType::Struct(inner) = min_values.data_type() {
            assert!(
                inner.field("value").is_some(),
                "minValues should have 'value' (requested)"
            );
            assert!(
                inner.field("extra").is_some(),
                "minValues should have 'extra' (from predicate)"
            );
            assert!(
                inner.field("id").is_none(),
                "minValues should not have 'id' (neither requested nor in predicate)"
            );
        } else {
            panic!("minValues should be a struct");
        }
    }

    #[test]
    fn non_empty_stats_columns_filters_schema() {
        let schema = schema_ref! {
            nullable "id": STRING,
            nullable "value": LONG,
        };

        let state_info = get_state_info_with_stats(
            schema,
            vec![],
            None,
            &[], // no table features
            HashMap::new(),
            vec![],
            StatsOptions {
                synthesize_json: true,
                struct_stats: StructStats::Columns {
                    requested: vec![column_name!("value")],
                },
                ..Default::default()
            },
        )
        .unwrap();

        let stats_schema = state_info
            .physical_stats_read_schema()
            .expect("should have physical stats schema");

        // Check that minValues/maxValues only contain 'value', not 'id'
        let min_values = stats_schema
            .field(MIN_VALUES)
            .expect("should have minValues");
        if let DataType::Struct(inner) = min_values.data_type() {
            assert!(
                inner.field("value").is_some(),
                "minValues should have 'value'"
            );
            assert!(
                inner.field("id").is_none(),
                "minValues should not have 'id'"
            );
        } else {
            panic!("minValues should be a struct");
        }
    }

    #[test]
    fn partition_schema_uses_physical_names_with_column_mapping() {
        // Verify that physical_partition_schema uses physical column names when column
        // mapping is enabled. The logical partition column "date" has physical name
        // "col-date-phys", and the schema should reflect the physical name.
        let schema = schema_ref! {
            (cm_field("id", 1, "col-id-phys", DataType::STRING)),
            (cm_field("date", 2, "col-date-phys", DataType::DATE)),
            (cm_field("value", 3, "col-value-phys", DataType::LONG)),
        };

        let predicate = Arc::new(col!("date").lt(lit(100i32)));

        let state_info = get_state_info(
            schema,
            vec!["date".to_string()],
            Some(predicate),
            &[TableFeature::ColumnMapping],
            get_string_map(&[("delta.columnMapping.mode", "name")]),
            vec![],
        )
        .unwrap();

        // physical_partition_schema should exist and use the physical column name
        let partition_schema = state_info
            .physical_partition_schema
            .as_ref()
            .expect("should have physical_partition_schema with predicate + partition columns");
        assert_eq!(partition_schema.num_fields(), 1);
        let field = partition_schema.fields().next().unwrap();
        assert_eq!(
            field.name(),
            "col-date-phys",
            "partition schema should use physical column name, not logical"
        );
        assert_eq!(field.data_type(), &DataType::DATE);
    }

    /// `with_struct` builds `physical_partition_schema` from all partition columns (independent of
    /// predicate and `StatsOptions::none`), and omits it entirely on non-partitioned tables.
    #[rstest]
    #[case::all_columns_without_predicate(
        vec![
            StructField::nullable("id", DataType::LONG),
            StructField::nullable("region", DataType::STRING),
            StructField::nullable("date", DataType::DATE),
        ],
        vec!["region".to_string(), "date".to_string()],
        StatsOptions::default(),
        Some(vec!["region", "date"]),
    )]
    #[case::survives_stats_none(
        vec![
            StructField::nullable("value", DataType::LONG),
            StructField::nullable("date", DataType::DATE),
        ],
        vec!["date".to_string()],
        StatsOptions::none(),
        Some(vec!["date"]),
    )]
    #[case::omitted_for_non_partitioned_table(
        vec![
            StructField::nullable("id", DataType::LONG),
            StructField::nullable("value", DataType::STRING),
        ],
        vec![],
        StatsOptions::default(),
        None,
    )]
    fn partition_values_with_struct(
        #[case] fields: Vec<StructField>,
        #[case] partition_columns: Vec<String>,
        #[case] stats: StatsOptions,
        #[case] expected_names: Option<Vec<&str>>,
    ) {
        let schema = Arc::new(StructType::new_unchecked(fields));
        let state_info = get_state_info_with_options(
            schema,
            partition_columns,
            None,
            &[],
            HashMap::new(),
            vec![],
            stats,
            PartitionValuesOptions::with_struct(),
        )
        .unwrap();

        match expected_names {
            Some(expected) => {
                let partition_schema = state_info
                    .physical_partition_schema
                    .as_ref()
                    .expect("with_struct should build a partition schema");
                let names: Vec<&str> = partition_schema
                    .fields()
                    .map(|f| f.name().as_str())
                    .collect();
                assert_eq!(names, expected);
                // MapToStruct lookups can return null, so every field must be nullable.
                assert!(partition_schema.fields().all(|f| f.is_nullable()));
            }
            None => assert!(state_info.physical_partition_schema.is_none()),
        }
    }

    #[test]
    fn stats_columns_with_column_mapping_uses_physical_names() {
        let schema = schema_ref! {
            (cm_field("col_a", 1, "phys_a", DataType::LONG)),
            (cm_field("col_b", 2, "phys_b", DataType::LONG)),
            (cm_field("col_c", 3, "phys_c", DataType::LONG)),
        };
        let mut props = HashMap::new();
        props.insert("delta.columnMapping.mode".to_string(), "name".to_string());

        // Request col_a via stats_columns (logical), and reference col_b via predicate (logical).
        // Both must be translated to physical names in the output stats schema.
        let predicate = Arc::new(col!("col_b").gt(lit(5i64)));

        let state_info = get_state_info_with_stats(
            schema,
            vec![],
            Some(predicate),
            &[],
            props,
            vec![],
            StatsOptions {
                synthesize_json: true,
                struct_stats: StructStats::Columns {
                    requested: vec![column_name!("col_a")],
                },
                ..Default::default()
            },
        )
        .unwrap();

        let stats_schema = state_info
            .physical_stats_read_schema()
            .expect("should have physical stats schema");

        assert_stats_leaves(
            stats_schema,
            &["phys_a", "phys_b"],
            &["col_a", "col_b", "phys_c"],
        );
    }

    #[cfg(feature = "geo-type-in-dev")]
    #[rstest]
    #[case::indexed(StatsOptions::all_struct(), None, [true, true, false])]
    #[case::extra(
        StatsOptions::all_struct_with_extra_indexed(vec![column_name!("nested.g")]),
        None,
        [true, true, true]
    )]
    #[case::selected(
        StatsOptions::struct_columns(vec![column_name!("nested.g")]),
        None,
        [false, false, true]
    )]
    #[case::table_columns(StatsOptions::all_struct(), Some("nested.g"), [false, false, true])]
    #[case::table_struct(StatsOptions::all_struct(), Some("nested"), [false, false, true])]
    fn geometry_stats_read_and_output_schemas_preserve_selection_mapping_and_crs(
        #[case] stats: StatsOptions,
        #[case] stats_columns: Option<&str>,
        #[case] included: [bool; 3],
        #[values(
            ColumnMappingMode::None,
            ColumnMappingMode::Name,
            ColumnMappingMode::Id
        )]
        mapping: ColumnMappingMode,
        #[values(false, true)] geometry_min_max: bool,
    ) {
        let geometry = DataType::from(GeometryType::try_new("EPSG:4326").unwrap());
        let nested_geometry = DataType::from(GeometryType::try_new("EPSG:3857").unwrap());
        let schema = schema_ref! {
            (cm_field("part", 1, "phys_part", DataType::LONG)),
            (cm_field("id", 2, "phys_id", DataType::LONG)),
            (cm_field("g", 3, "phys_g", geometry.clone())),
            (cm_field("nested", 4, "phys_nested", schema! {
                (cm_field("g", 5, "phys_nested_g", nested_geometry.clone())),
            })),
        };
        let mut builder = MockTableConfigurationBuilder::new()
            .with_schema(schema.clone())
            .with_partition_columns(["part"])
            .with_column_mapping(mapping)
            .with_properties([("delta.dataSkippingNumIndexedCols", "2")])
            .with_protocol(
                MockProtocolBuilder::new()
                    .with_features([TableFeature::GeospatialType, TableFeature::ColumnMapping])
                    .build(),
            );
        if let Some(columns) = stats_columns {
            builder = builder.with_properties([("delta.dataSkippingStatsColumns", columns)]);
        }
        let config = builder.build();
        let stats = stats.with_geometry_min_max_stats(geometry_min_max);
        let output = build_stats_output_schemas(&config, &stats)
            .unwrap()
            .unwrap();
        let state = StateInfo::try_new(
            schema.clone(),
            schema,
            &config,
            None,
            &stats,
            &PartitionValuesOptions::default(),
            (),
        )
        .unwrap();
        assert_eq!(state.physical_stats_read_schema(), Some(&output.physical));
        assert_eq!(state.physical_stats_output_schema(), Some(&output.physical));

        let columns = [
            (vec!["id"], vec!["phys_id"], DataType::LONG),
            (vec!["g"], vec!["phys_g"], geometry),
            (
                vec!["nested", "g"],
                vec!["phys_nested", "phys_nested_g"],
                nested_geometry,
            ),
        ];
        for (index, (logical, physical, data_type)) in columns.iter().enumerate() {
            for (schema, path) in [
                (&output.logical, logical),
                (
                    &output.physical,
                    if mapping == ColumnMappingMode::None {
                        logical
                    } else {
                        physical
                    },
                ),
            ] {
                for stat_name in [NULL_COUNT, MIN_VALUES, MAX_VALUES] {
                    let path =
                        ColumnName::new(std::iter::once(stat_name).chain(path.iter().copied()));
                    let field = schema.field_at(&path).ok();
                    let present = included[index]
                        && (stat_name == NULL_COUNT || index == 0 || geometry_min_max);
                    assert_eq!(field.is_some(), present, "{path}");
                    if let Some(field) = field {
                        let expected = if stat_name == NULL_COUNT {
                            &DataType::LONG
                        } else {
                            data_type
                        };
                        assert_eq!(field.data_type(), expected, "{path}");
                        assert!(field.is_nullable());
                        assert!(field.metadata.is_empty());
                    }
                }
            }
        }
        assert!(output
            .logical
            .field_at(&column_name!("nullCount.part"))
            .is_err());
        // Scan requests do not change the default schema used by writers and other callers.
        let default_schema = config.stats_schema_builder().build().unwrap();
        assert!(default_schema
            .field_at(&column_name!("minValues.g"))
            .is_err());
        assert!(default_schema
            .field_at(&column_name!("minValues.phys_g"))
            .is_err());
    }

    // === eligible_physical_stats_columns trims the predicate-derived stats schema ===

    /// Flat schema with `n` long columns named `c0..c{n-1}`.
    fn flat_long_schema(n: usize) -> SchemaRef {
        Arc::new(StructType::new_unchecked(
            (0..n)
                .map(|i| StructField::nullable(format!("c{i}"), DataType::LONG))
                .collect::<Vec<_>>(),
        ))
    }

    /// Asserts each `present` top-level leaf is in, and each `absent` leaf is out of, both
    /// `minValues` and `maxValues` of `stats_schema`.
    fn assert_stats_leaves(stats_schema: &SchemaRef, present: &[&str], absent: &[&str]) {
        for stats_field in [MIN_VALUES, MAX_VALUES] {
            let DataType::Struct(inner) = stats_schema
                .field(stats_field)
                .unwrap_or_else(|| panic!("should have {stats_field}"))
                .data_type()
            else {
                panic!("{stats_field} should be a struct");
            };
            for name in present {
                assert!(
                    inner.field(name).is_some(),
                    "{stats_field} expected '{name}'"
                );
            }
            for name in absent {
                assert!(
                    inner.field(name).is_none(),
                    "{stats_field} unexpected '{name}'"
                );
            }
        }
    }

    /// `delta.dataSkippingNumIndexedCols=<n>` configuration map.
    fn num_indexed_cols_config(n: i32) -> HashMap<String, String> {
        let mut m = HashMap::new();
        m.insert(
            "delta.dataSkippingNumIndexedCols".to_string(),
            n.to_string(),
        );
        m
    }

    /// `delta.dataSkippingStatsColumns=<cols joined by ",">` configuration map.
    fn stats_columns_config(cols: &[&str]) -> HashMap<String, String> {
        let mut m = HashMap::new();
        m.insert("delta.dataSkippingStatsColumns".to_string(), cols.join(","));
        m
    }

    /// Both `delta.dataSkippingStatsColumns` and `delta.dataSkippingNumIndexedCols` set
    /// simultaneously. Per the Delta protocol, the explicit list takes precedence over the
    /// cap; this helper exists for tests that exercise that precedence.
    fn both_configs(stats_cols: &[&str], num_indexed: i32) -> HashMap<String, String> {
        let mut m = stats_columns_config(stats_cols);
        m.insert(
            "delta.dataSkippingNumIndexedCols".to_string(),
            num_indexed.to_string(),
        );
        m
    }

    /// `delta.dataSkippingNumIndexedCols` caps the `eligible_physical_stats_columns` set to the
    /// first N leaves.
    #[test]
    fn stats_columns_honors_num_indexed_cols() {
        let schema = flat_long_schema(5);
        let state_info = get_state_info(
            schema,
            vec![],
            None, // no predicate; just check the cached set
            &[],
            num_indexed_cols_config(2),
            vec![],
        )
        .unwrap();
        let cols = HashSet::from_iter([column_name!("c0"), column_name!("c1")]);
        assert_eq!(state_info.eligible_physical_stats_columns, cols);
    }

    /// Predicate on a past-cap column: stats schema goes to `None` (no skipping), but
    /// the physical predicate is retained so engines can still apply it per-row.
    #[test]
    fn predicate_on_past_cap_column_drops_stats_schema() {
        let schema = flat_long_schema(5);
        let predicate = Arc::new(col!("c4").gt(lit(10i64)));
        let state_info = get_state_info(
            schema,
            vec![],
            Some(predicate),
            &[],
            num_indexed_cols_config(2),
            vec![],
        )
        .unwrap();
        assert!(
            state_info.physical_stats_read_schema().is_none(),
            "Predicate on a past-cap column should produce no stats schema, got {:?}",
            state_info.physical_stats_read_schema()
        );
        assert!(
            matches!(state_info.physical_predicate, PhysicalPredicate::Some(_, _)),
            "User predicate must be retained even when stats schema is dropped"
        );
    }

    /// Indexed AND past-cap: indexed leaf survives in the stats schema, past-cap leaf
    /// is dropped.
    #[test]
    fn predicate_on_mixed_indexed_and_past_cap_keeps_indexed_only() {
        let schema = flat_long_schema(5);
        let predicate = Arc::new(Pred::and(
            col!("c0").gt(lit(10i64)),
            col!("c4").gt(lit(10i64)),
        ));
        let state_info = get_state_info(
            schema,
            vec![],
            Some(predicate),
            &[],
            num_indexed_cols_config(2),
            vec![],
        )
        .unwrap();
        let stats_schema = state_info
            .physical_stats_read_schema()
            .expect("should have stats schema (indexed arm survives)");
        // c0 (indexed) survives; c4 (past cap) is dropped.
        assert_stats_leaves(stats_schema, &["c0"], &["c4"]);
    }

    /// `numIndexedCols=2` against `{ a, b, s: { c, d } }` keeps `a, b` and drops the
    /// entire `s` struct, so a predicate on `s.c` produces no stats schema.
    #[test]
    fn predicate_on_nested_past_cap_leaf_drops_parent_struct() {
        let schema = schema_ref! {
            nullable "a": LONG,
            nullable "b": LONG,
            nullable "s": {
                nullable "c": LONG,
                nullable "d": LONG,
            },
        };
        // Predicate only on the past-cap leaf -> stats schema goes empty -> None.
        let predicate = Arc::new(col!("s.c").gt(lit(10i64)));
        let state_info = get_state_info(
            schema,
            vec![],
            Some(predicate),
            &[],
            num_indexed_cols_config(2),
            vec![],
        )
        .unwrap();
        assert!(state_info.physical_stats_read_schema().is_none());
        assert!(!state_info.eligible_physical_stats_columns.is_empty());
        assert!(!state_info
            .eligible_physical_stats_columns
            .contains(&column_name!("s.c")));
    }

    /// `delta.dataSkippingStatsColumns` selects exactly the listed leaves, regardless of
    /// their position relative to the (default) cap.
    #[rstest]
    #[case::single(&["c2"], &["c2"])]
    #[case::sparse_subset(&["c0", "c3"], &["c0", "c3"])]
    #[case::all_listed(&["c0", "c1", "c2", "c3", "c4"], &["c0", "c1", "c2", "c3", "c4"])]
    fn stats_columns_honors_explicit_stats_columns(
        #[case] listed: &[&str],
        #[case] expected: &[&str],
    ) {
        let schema = flat_long_schema(5);
        let state_info = get_state_info(
            schema,
            vec![],
            None,
            &[],
            stats_columns_config(listed),
            vec![],
        )
        .unwrap();
        let expected_cols: HashSet<ColumnName> =
            expected.iter().map(|s| ColumnName::new([*s])).collect();
        assert_eq!(state_info.eligible_physical_stats_columns, expected_cols);
    }

    /// `numIndexedCols=3` against `{ a, b, s: { c, d } }` keeps `a, b, s.c` and drops
    /// `s.d`. A predicate on both `s.c` and `s.d` keeps the `s` struct under
    /// `minValues` / `maxValues` with `c` only.
    #[test]
    fn predicate_on_nested_mixed_keeps_intersection_under_parent_struct() {
        let schema = schema_ref! {
            nullable "a": LONG,
            nullable "b": LONG,
            nullable "s": {
                nullable "c": LONG,
                nullable "d": LONG,
            },
        };
        let predicate = Arc::new(Pred::and(
            col!("s.c").gt(lit(10i64)),
            col!("s.d").gt(lit(10i64)),
        ));
        let state_info = get_state_info(
            schema,
            vec![],
            Some(predicate),
            &[],
            num_indexed_cols_config(3),
            vec![],
        )
        .unwrap();
        let stats_schema = state_info
            .physical_stats_read_schema()
            .expect("indexed arm survives");
        for stats_field in [MIN_VALUES, MAX_VALUES] {
            let DataType::Struct(outer) = stats_schema
                .field(stats_field)
                .unwrap_or_else(|| panic!("should have {stats_field}"))
                .data_type()
            else {
                panic!("{stats_field} should be a struct");
            };
            let DataType::Struct(inner) = outer
                .field("s")
                .unwrap_or_else(|| panic!("{stats_field} should have s"))
                .data_type()
            else {
                panic!("{stats_field}.s should be a struct");
            };
            assert!(
                inner.field("c").is_some(),
                "{stats_field}.s should keep c (indexed)"
            );
            assert!(
                inner.field("d").is_none(),
                "{stats_field}.s should drop d (past cap)"
            );
        }
    }

    /// `dataSkippingStatsColumns` with a parent struct path admits every leaf under that
    /// parent (the trie matches by prefix). `{ a, s: { c, d } }` with the property set to
    /// `"s"` should produce `{ s.c, s.d }` (and exclude `a`, which is not in the list).
    #[test]
    fn stats_columns_admits_all_children_of_nested_parent_in_explicit_list() {
        let schema = schema_ref! {
            nullable "a": LONG,
            nullable "s": {
                nullable "c": LONG,
                nullable "d": LONG,
            },
        };
        let state_info = get_state_info(
            schema,
            vec![],
            None,
            &[],
            stats_columns_config(&["s"]),
            vec![],
        )
        .unwrap();
        let expected = HashSet::from_iter([column_name!("s.c"), column_name!("s.d")]);
        assert_eq!(state_info.eligible_physical_stats_columns, expected);
    }

    /// `dataSkippingStatsColumns` ("A") takes precedence over `dataSkippingNumIndexedCols`
    /// ("B") whether A wants more columns than B allows or fewer.
    #[rstest]
    #[case::a_broader_than_b(&["c0", "c3", "c4"], 2, &["c0", "c3", "c4"])]
    #[case::a_narrower_than_b(&["c0"], 3, &["c0"])]
    fn stats_columns_explicit_list_overrides_num_indexed_cols(
        #[case] listed: &[&str],
        #[case] num_indexed: i32,
        #[case] expected: &[&str],
    ) {
        let schema = flat_long_schema(5);
        let state_info = get_state_info(
            schema,
            vec![],
            None,
            &[],
            both_configs(listed, num_indexed),
            vec![],
        )
        .unwrap();
        let expected_cols: HashSet<ColumnName> =
            expected.iter().map(|s| ColumnName::new([*s])).collect();
        assert_eq!(state_info.eligible_physical_stats_columns, expected_cols);
    }

    #[rstest]
    #[case::all_extra_past_cap(
        StatsOptions::all_struct_with_extra_indexed(vec![column_name!("c4")]),
        num_indexed_cols_config(2),
        &["c0", "c1", "c4"], &["c2", "c3"], &["c0", "c1", "c4"],
    )]
    #[case::all_no_extra(
        StatsOptions::all_struct(),
        num_indexed_cols_config(2),
        &["c0", "c1"], &["c2", "c3", "c4"], &["c0", "c1"],
    )]
    #[case::extra_within_cap_noop(
        StatsOptions::all_struct_with_extra_indexed(vec![column_name!("c0")]),
        num_indexed_cols_config(2),
        &["c0", "c1"], &["c2", "c3", "c4"], &["c0", "c1"],
    )]
    #[case::columns_within_and_past_cap(
        StatsOptions::struct_columns(vec![column_name!("c0"), column_name!("c4")]),
        num_indexed_cols_config(2),
        &["c0", "c4"], &["c1", "c2", "c3"], &["c0", "c1", "c4"],
    )]
    #[case::columns_only_past_cap(
        StatsOptions::struct_columns(vec![column_name!("c4")]),
        num_indexed_cols_config(2),
        &["c4"], &["c0", "c1", "c2", "c3"], &["c0", "c1", "c4"],
    )]
    #[case::extra_with_stats_columns(
        StatsOptions::all_struct_with_extra_indexed(vec![column_name!("c4")]),
        stats_columns_config(&["c0"]),
        &["c0", "c4"], &["c1", "c2", "c3"], &["c0", "c4"],
    )]
    #[case::unresolvable_extra_dropped(
        StatsOptions::all_struct_with_extra_indexed(vec![
            column_name!("c4"),
            column_name!("does_not_exist"),
        ]),
        num_indexed_cols_config(2),
        &["c0", "c1", "c4"], &["c2", "c3"], &["c0", "c1", "c4"],
    )]
    fn extra_indexed_schema_and_gate(
        #[case] stats: StatsOptions,
        #[case] config: HashMap<String, String>,
        #[case] present: &[&str],
        #[case] absent: &[&str],
        #[case] expected_set: &[&str],
    ) {
        let state_info = get_state_info_with_stats(
            flat_long_schema(5),
            vec![],
            None,
            &[],
            config,
            vec![],
            stats,
        )
        .unwrap();
        let stats_schema = state_info
            .physical_stats_read_schema()
            .expect("stats schema present");
        assert_stats_leaves(stats_schema, present, absent);
        let expected: HashSet<ColumnName> =
            expected_set.iter().map(|s| ColumnName::new([*s])).collect();
        assert_eq!(state_info.eligible_physical_stats_columns, expected);
    }

    #[test]
    fn extra_indexed_column_widens_internal_skipping_gate() {
        let state_info = get_state_info_with_stats(
            flat_long_schema(5),
            vec![],
            Some(Arc::new(col!("c4").gt(lit(10i64)))),
            &[],
            num_indexed_cols_config(2),
            vec![],
            StatsOptions::all_struct_with_extra_indexed(vec![column_name!("c4")]),
        )
        .unwrap();
        let stats_schema = state_info
            .physical_stats_read_schema()
            .expect("stats schema present because c4 is extra_indexed");
        assert_stats_leaves(stats_schema, &["c4"], &[]);
        assert!(state_info
            .eligible_physical_stats_columns
            .contains(&column_name!("c4")));
    }

    #[test]
    fn extra_indexed_nested_subfield_past_cap_appears() {
        let schema = schema_ref! {
            nullable "a": LONG,
            nullable "b": LONG,
            nullable "s": {
                nullable "c": LONG,
                nullable "d": LONG,
            },
        };
        let state_info = get_state_info_with_stats(
            schema,
            vec![],
            None,
            &[],
            num_indexed_cols_config(3),
            vec![],
            StatsOptions::all_struct_with_extra_indexed(vec![column_name!("s.d")]),
        )
        .unwrap();
        let stats_schema = state_info
            .physical_stats_read_schema()
            .expect("stats schema present");
        for stats_field in [MIN_VALUES, MAX_VALUES] {
            let DataType::Struct(inner) = stats_schema.field(stats_field).unwrap().data_type()
            else {
                panic!("{stats_field} should be a struct");
            };
            let DataType::Struct(s) = inner.field("s").expect("s present").data_type() else {
                panic!("s should be a struct");
            };
            assert!(s.field("c").is_some(), "s.c (indexed) present");
            assert!(
                s.field("d").is_some(),
                "s.d (extra_indexed, past cap) present"
            );
        }
        assert!(state_info
            .eligible_physical_stats_columns
            .contains(&column_name!("s.d")));
    }

    #[test]
    fn extra_indexed_struct_column_widens_gate_to_leaves() {
        // A struct named as extra_indexed must widen the skipping gate to its leaf paths, not the
        // parent path, so a leaf predicate (matched by exact membership) can still prune.
        let schema = schema_ref! {
            nullable "a": LONG,
            nullable "b": LONG,
            nullable "s": {
                nullable "c": LONG,
                nullable "d": LONG,
            },
        };
        let state_info = get_state_info_with_stats(
            schema,
            vec![],
            None,
            &[],
            num_indexed_cols_config(2),
            vec![],
            StatsOptions::all_struct_with_extra_indexed(vec![column_name!("s")]),
        )
        .unwrap();
        assert!(state_info
            .eligible_physical_stats_columns
            .contains(&column_name!("s.c")));
        assert!(state_info
            .eligible_physical_stats_columns
            .contains(&column_name!("s.d")));
        assert!(
            !state_info
                .eligible_physical_stats_columns
                .contains(&column_name!("s")),
            "the parent path must not stand in for its leaves"
        );
    }

    #[test]
    fn extra_indexed_column_resolves_physical_name_under_column_mapping() {
        let schema = schema_ref! {
            (cm_field("col_a", 1, "phys_a", DataType::LONG)),
            (cm_field("col_b", 2, "phys_b", DataType::LONG)),
            (cm_field("col_c", 3, "phys_c", DataType::LONG)),
        };
        let mut props = HashMap::new();
        props.insert("delta.columnMapping.mode".to_string(), "name".to_string());
        props.insert(
            "delta.dataSkippingNumIndexedCols".to_string(),
            "1".to_string(),
        );
        let state_info = get_state_info_with_stats(
            schema,
            vec![],
            None,
            &[],
            props,
            vec![],
            StatsOptions::all_struct_with_extra_indexed(vec![column_name!("col_c")]),
        )
        .unwrap();
        let stats_schema = state_info
            .physical_stats_read_schema()
            .expect("stats schema present");
        assert_stats_leaves(stats_schema, &["phys_a", "phys_c"], &["col_c", "phys_b"]);
        assert!(state_info
            .eligible_physical_stats_columns
            .contains(&column_name!("phys_c")));
    }
}