feat!: expose read-side expected stats schemas

kernel/src/checkpoint/mod.rs

@@ -740,9 +740,8 @@ impl CheckpointWriter {
 
         // Get stats schema from table configuration.
         // This already excludes partition columns and applies column mapping.
-        let stats_schema = tc
-            .build_expected_stats_schemas(physical_clustering_columns.as_deref(), None)?
-            .physical;
+        let stats_schema =
+            tc.build_expected_physical_stats_schema(physical_clustering_columns.as_deref(), None)?;
 
         // Build partition schema for partitionValues_parsed (None for non-partitioned tables)
         let partition_schema = tc.build_partition_values_parsed_schema();

kernel/src/scan/data_skipping.rs

@@ -226,7 +226,8 @@ impl DataSkippingFilter {
     /// unlike the scan path which reads pre-parsed `stats_parsed` from transformed batches.
     ///
     /// The stats schema is derived from the predicate's column references via
-    /// [`TableConfiguration::build_expected_stats_schemas`], matching the write side exactly;
+    /// [`TableConfiguration::build_expected_physical_stats_schema`], matching the write side
+    /// exactly;
     /// references outside the table's stats columns fold to NULL (keeping the file). Partition
     /// values are parsed from the raw `add.partitionValues` string map with
     /// [`Expression::map_to_struct`], so predicates over partition columns prune too.
@@ -261,9 +262,8 @@ impl DataSkippingFilter {
             .collect();
         let physical_stats_columns = table_configuration.physical_stats_columns_set(None);
         let physical_stats_schema = table_configuration
-            .build_expected_stats_schemas(None, Some(&predicate_refs))
-            .ok()?
-            .physical;
+            .build_expected_physical_stats_schema(None, Some(&predicate_refs))
+            .ok()?;
         let partition_schema = table_configuration.predicate_partition_schema(&predicate_refs);
 
         // Parse JSON stats from the raw action batch's `add.stats` column, parse partition values

kernel/src/scan/mod.rs

@@ -743,8 +743,7 @@ fn build_physical_stats_output_schema(
                 return Ok(None);
             }
             let stats_schema = table_configuration
-                .build_expected_stats_schemas(Some(requested), Some(requested))?
-                .physical;
+                .build_expected_physical_stats_schema(Some(requested), Some(requested))?;
             Ok(stats_schema_with_data_columns(stats_schema))
         }
     }

kernel/src/scan/state_info.rs

@@ -172,11 +172,11 @@ fn build_data_skipping_schemas(
         resolve_physical_columns(table_configuration, predicate_column_names_logical);
 
     // A stats schema with only `numRecords` and `tightBounds` (the bookkeeping fields
-    // `build_expected_stats_schemas` always emits) has nothing to prune by. Return `None`
+    // `build_expected_physical_stats_schema` 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 `build_expected_stats_schemas` always emits `nullCount` whenever it
+    // `nullCount`, but `build_expected_physical_stats_schema` always emits `nullCount` whenever it
     // emits min/max; this check relies on that implementation property.
     let with_data_cols = |stats_schema: SchemaRef| -> Option<SchemaRef> {
         stats_schema
@@ -188,8 +188,7 @@ fn build_data_skipping_schemas(
     let stats_schema = match (struct_stats, physical_predicate) {
         (StructStats::AllIndexed { .. }, _) => with_data_cols(
             table_configuration
-                .build_expected_stats_schemas(requested_physical_stats_columns, None)?
-                .physical,
+                .build_expected_physical_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.
@@ -198,18 +197,16 @@ fn build_data_skipping_schemas(
                 .unwrap_or_default()
                 .to_vec();
             union_extra_into_filter(&mut filter, &predicate_refs_physical);
-            with_data_cols(
-                table_configuration
-                    .build_expected_stats_schemas(requested_physical_stats_columns, Some(&filter))?
-                    .physical,
-            )
+            with_data_cols(table_configuration.build_expected_physical_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(_, _)) => with_data_cols(
             table_configuration
-                .build_expected_stats_schemas(None, Some(&predicate_refs_physical))?
-                .physical,
+                .build_expected_physical_stats_schema(None, Some(&predicate_refs_physical))?,
         ),
         // No struct stats requested and no predicate: nothing to read or emit, so no stats schema.
         (_, _) => None,

kernel/src/scan/tests.rs

@@ -2270,6 +2270,89 @@ fn scan_builder_tolerates_nonexistent_extra_indexed_column() {
     );
 }
 
+#[rstest]
+#[case::no_column_mapping(None)]
+#[case::name_column_mapping(Some("name"))]
+#[case::id_column_mapping(Some("id"))]
+fn snapshot_expected_stats_schemas_match_scan_output(#[case] column_mapping_mode: Option<&str>) {
+    let table_root = "memory:///expected-stats-schemas/";
+    let store = Arc::new(InMemory::new());
+    let engine = SyncEngine::new_with_store(store);
+    let schema = schema_ref! {
+        nullable "id": LONG,
+        nullable "value": LONG,
+    };
+    let mut create_builder = create_table(table_root, schema, "DefaultEngine")
+        .with_table_properties([("delta.dataSkippingNumIndexedCols", "1")]);
+    if let Some(mode) = column_mapping_mode {
+        create_builder = create_builder.with_table_properties([("delta.columnMapping.mode", mode)]);
+    }
+    create_builder
+        .build(&engine, Box::new(FileSystemCommitter::new()))
+        .unwrap()
+        .commit(&engine)
+        .unwrap()
+        .unwrap_committed();
+
+    let snapshot = Snapshot::builder_for(table_root).build(&engine).unwrap();
+    let extra_indexed_columns = vec![column_name!("value"), column_name!("unresolvable_extra")];
+    let expected = snapshot
+        .expected_stats_schemas(&extra_indexed_columns)
+        .unwrap();
+    let scan = snapshot
+        .scan_builder()
+        .with_stats(StatsOptions::all_struct_with_extra_indexed(
+            extra_indexed_columns,
+        ))
+        .build()
+        .unwrap();
+
+    assert_eq!(
+        scan.physical_stats_output_schema.as_ref(),
+        Some(&expected.physical)
+    );
+
+    let DataType::Struct(logical_min_values) = expected
+        .logical
+        .field(MIN_VALUES)
+        .expect("logical stats should have minValues")
+        .data_type()
+    else {
+        panic!("logical minValues should be a struct");
+    };
+    assert!(logical_min_values.field("id").is_some());
+    assert!(logical_min_values.field("value").is_some());
+    assert!(logical_min_values.field("unresolvable_extra").is_none());
+
+    let DataType::Struct(physical_min_values) = expected
+        .physical
+        .field(MIN_VALUES)
+        .expect("physical stats should have minValues")
+        .data_type()
+    else {
+        panic!("physical minValues should be a struct");
+    };
+    assert_eq!(physical_min_values.num_fields(), 2);
+    if column_mapping_mode.is_some() {
+        assert!(logical_min_values.fields().all(|field| {
+            field
+                .get_config_value(&ColumnMetadataKey::ColumnMappingPhysicalName)
+                .is_none()
+                && field
+                    .get_config_value(&ColumnMetadataKey::ParquetFieldId)
+                    .is_none()
+        }));
+        assert!(physical_min_values.fields().all(|field| {
+            field.name().starts_with("col-")
+                && field
+                    .get_config_value(&ColumnMetadataKey::ParquetFieldId)
+                    .is_none()
+        }));
+    } else {
+        assert_eq!(physical_min_values, logical_min_values);
+    }
+}
+
 /// A [`ParquetHandler`] that returns an empty iterator for every `read_parquet_files` call.
 /// Used to simulate a buggy connector that drops all data for a file.
 struct EmptyParquetHandler;

kernel/src/snapshot/mod.rs

@@ -30,7 +30,9 @@ use crate::metrics::{
 use crate::path::ParsedLogPath;
 use crate::scan::ScanBuilder;
 use crate::schema::SchemaRef;
-use crate::table_configuration::{InCommitTimestampEnablement, TableConfiguration};
+use crate::table_configuration::{
+    ExpectedStatsSchemas, InCommitTimestampEnablement, TableConfiguration,
+};
 use crate::table_features::{physical_to_logical_column_name_and_type, TableFeature};
 use crate::table_properties::TableProperties;
 use crate::transaction::builder::alter_table::AlterTableTransactionBuilder;
@@ -343,6 +345,34 @@ impl Snapshot {
         self.table_configuration.logical_schema()
     }
 
+    /// Returns the expected logical and physical schemas for file statistics.
+    ///
+    /// `extra_indexed_columns` are logical column paths that may have statistics even when they
+    /// fall outside the table's configured indexed-column set. Resolvable extra columns are
+    /// included in both returned schemas; partition columns and unresolvable paths are omitted.
+    /// The physical schema applies the table's column-mapping mode.
+    ///
+    /// Both schemas contain `numRecords` and `tightBounds`. When at least one data column is
+    /// selected, they also contain `nullCount` and, for eligible data types, `minValues` and
+    /// `maxValues`. Nested fields mirror the selected portion of the table schema.
+    ///
+    /// Pass the same extra columns to [`StatsOptions::all_struct_with_extra_indexed`] when building
+    /// a scan that returns structured statistics.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if kernel cannot construct a valid stats schema.
+    ///
+    /// [`StatsOptions::all_struct_with_extra_indexed`]:
+    /// crate::scan::StatsOptions::all_struct_with_extra_indexed
+    pub fn expected_stats_schemas(
+        &self,
+        extra_indexed_columns: &[ColumnName],
+    ) -> DeltaResult<ExpectedStatsSchemas> {
+        self.table_configuration
+            .build_expected_stats_schemas(extra_indexed_columns)
+    }
+
     /// Estimated owned heap size in bytes for this snapshot. Best-effort estimate
     /// for capacity tracking, not authoritative.
     ///