feat: read parsed-stats from checkpoint (delta-io#1638)

## 🥞 Stacked PR
Use this
[link](https://github.com/delta-io/delta-kernel-rs/pull/1638/files) to
review incremental changes.
-
[**stack/has_compatible_parsed_stats**](delta-io#1638)
[[Files
changed](https://github.com/delta-io/delta-kernel-rs/pull/1638/files)]
-
[stack/output-stat-columns-all](delta-io#1720)
[[Files
changed](https://github.com/delta-io/delta-kernel-rs/pull/1720/files/6f727fd4d2fe1bd7733f98063c83c8624b74370a..71054a500a40f59762098d1dea7073bf161d3570)]

---------
## What changes are proposed in this pull request?

This PR adds infrastructure to detect when checkpoints have compatible
pre-parsed statistics (stats_parsed) that can be used for data skipping
without JSON parsing.

Added CheckpointReadInfo struct containing:

- has_stats_parsed: bool - whether checkpoint has compatible pre-parsed
stats
- checkpoint_read_schema: SchemaRef - schema used to read checkpoint
files

## How was this change tested?
New and existing unit tests

Author: DrakeLin

kernel/src/log_segment.rs

@@ -1170,8 +1170,13 @@ async fn test_create_checkpoint_stream_returns_checkpoint_batches_as_is_if_schem
         None,
         None,
     )?;
-    let mut iter =
-        log_segment.create_checkpoint_stream(&engine, v2_checkpoint_read_schema.clone(), None)?;
+    let checkpoint_result = log_segment.create_checkpoint_stream(
+        &engine,
+        v2_checkpoint_read_schema.clone(),
+        None,
+        None,
+    )?;
+    let mut iter = checkpoint_result.actions;
 
     // Assert that the first batch returned is from reading checkpoint file 1
     let ActionsBatch {
@@ -1235,8 +1240,13 @@ async fn test_create_checkpoint_stream_returns_checkpoint_batches_if_checkpoint_
         None,
         None,
     )?;
-    let mut iter =
-        log_segment.create_checkpoint_stream(&engine, v2_checkpoint_read_schema.clone(), None)?;
+    let checkpoint_result = log_segment.create_checkpoint_stream(
+        &engine,
+        v2_checkpoint_read_schema.clone(),
+        None,
+        None,
+    )?;
+    let mut iter = checkpoint_result.actions;
 
     // Assert the correctness of batches returned
     for expected_sidecar in ["sidecar1.parquet", "sidecar2.parquet"].iter() {
@@ -1295,8 +1305,13 @@ async fn test_create_checkpoint_stream_reads_parquet_checkpoint_batch_without_si
         None,
         None,
     )?;
-    let mut iter =
-        log_segment.create_checkpoint_stream(&engine, v2_checkpoint_read_schema.clone(), None)?;
+    let checkpoint_result = log_segment.create_checkpoint_stream(
+        &engine,
+        v2_checkpoint_read_schema.clone(),
+        None,
+        None,
+    )?;
+    let mut iter = checkpoint_result.actions;
 
     // Assert that the first batch returned is from reading checkpoint file 1
     let ActionsBatch {
@@ -1344,8 +1359,9 @@ async fn test_create_checkpoint_stream_reads_json_checkpoint_batch_without_sidec
         None,
         None,
     )?;
-    let mut iter =
-        log_segment.create_checkpoint_stream(&engine, v2_checkpoint_read_schema, None)?;
+    let checkpoint_result =
+        log_segment.create_checkpoint_stream(&engine, v2_checkpoint_read_schema, None, None)?;
+    let mut iter = checkpoint_result.actions;
 
     // Assert that the first batch returned is from reading checkpoint file 1
     let ActionsBatch {
@@ -1431,8 +1447,13 @@ async fn test_create_checkpoint_stream_reads_checkpoint_file_and_returns_sidecar
         None,
         None,
     )?;
-    let mut iter =
-        log_segment.create_checkpoint_stream(&engine, v2_checkpoint_read_schema.clone(), None)?;
+    let checkpoint_result = log_segment.create_checkpoint_stream(
+        &engine,
+        v2_checkpoint_read_schema.clone(),
+        None,
+        None,
+    )?;
+    let mut iter = checkpoint_result.actions;
 
     // Assert that the first batch returned is from reading checkpoint file 1
     let ActionsBatch {
@@ -2653,6 +2674,18 @@ fn create_checkpoint_schema_with_stats_parsed(min_values_fields: Vec<StructField
     StructType::new_unchecked([StructField::nullable("add", add_schema)])
 }
 
+// Helper to create a stats_schema with proper structure (numRecords, minValues, maxValues)
+fn create_stats_schema(column_fields: Vec<StructField>) -> StructType {
+    StructType::new_unchecked([
+        StructField::nullable("numRecords", DataType::LONG),
+        StructField::nullable(
+            "minValues",
+            StructType::new_unchecked(column_fields.clone()),
+        ),
+        StructField::nullable("maxValues", StructType::new_unchecked(column_fields)),
+    ])
+}
+
 // Helper to create a checkpoint schema without stats_parsed
 fn create_checkpoint_schema_without_stats_parsed() -> StructType {
     use crate::schema::StructType;
@@ -2675,15 +2708,15 @@ fn test_schema_has_compatible_stats_parsed_basic() {
         )]);
 
     // Exact type match should work
-    let stats_schema = StructType::new_unchecked([StructField::nullable("id", DataType::INTEGER)]);
+    let stats_schema = create_stats_schema(vec![StructField::nullable("id", DataType::INTEGER)]);
     assert!(LogSegment::schema_has_compatible_stats_parsed(
         &checkpoint_schema,
         &stats_schema
     ));
 
     // Type widening (int -> long) should work
     let stats_schema_widened =
-        StructType::new_unchecked([StructField::nullable("id", DataType::LONG)]);
+        create_stats_schema(vec![StructField::nullable("id", DataType::LONG)]);
     assert!(LogSegment::schema_has_compatible_stats_parsed(
         &checkpoint_schema,
         &stats_schema_widened
@@ -2703,17 +2736,18 @@ fn test_schema_has_compatible_stats_parsed_basic() {
 
 #[test]
 fn test_schema_has_compatible_stats_parsed_missing_column_ok() {
-    // Checkpoint has "id" column, stats schema needs "other" column (missing in checkpoint is OK)
+    // Checkpoint has "id" column, stats schema needs "other" column
+    // Missing column is acceptable - it will return null when accessed
     let checkpoint_schema =
         create_checkpoint_schema_with_stats_parsed(vec![StructField::nullable(
             "id",
             DataType::INTEGER,
         )]);
 
-    let stats_schema =
-        StructType::new_unchecked([StructField::nullable("other", DataType::INTEGER)]);
+    let stats_schema = create_stats_schema(vec![StructField::nullable("other", DataType::INTEGER)]);
 
-    // Missing column in checkpoint is OK - it will just return NULL
+    // Missing column in checkpoint is OK - it will return null when accessed,
+    // which is acceptable for data skipping (just means we can't skip based on that column)
     assert!(LogSegment::schema_has_compatible_stats_parsed(
         &checkpoint_schema,
         &stats_schema
@@ -2728,7 +2762,7 @@ fn test_schema_has_compatible_stats_parsed_extra_column_ok() {
         StructField::nullable("extra", DataType::STRING),
     ]);
 
-    let stats_schema = StructType::new_unchecked([StructField::nullable("id", DataType::INTEGER)]);
+    let stats_schema = create_stats_schema(vec![StructField::nullable("id", DataType::INTEGER)]);
 
     assert!(LogSegment::schema_has_compatible_stats_parsed(
         &checkpoint_schema,
@@ -2741,7 +2775,7 @@ fn test_schema_has_compatible_stats_parsed_no_stats_parsed() {
     // Checkpoint schema without stats_parsed field
     let checkpoint_schema = create_checkpoint_schema_without_stats_parsed();
 
-    let stats_schema = StructType::new_unchecked([StructField::nullable("id", DataType::INTEGER)]);
+    let stats_schema = create_stats_schema(vec![StructField::nullable("id", DataType::INTEGER)]);
 
     assert!(!LogSegment::schema_has_compatible_stats_parsed(
         &checkpoint_schema,
@@ -2758,7 +2792,7 @@ fn test_schema_has_compatible_stats_parsed_empty_stats_schema() {
             DataType::INTEGER,
         )]);
 
-    let stats_schema = StructType::new_unchecked([]);
+    let stats_schema = create_stats_schema(vec![]);
 
     // If no columns are needed for data skipping, any stats_parsed is compatible
     assert!(LogSegment::schema_has_compatible_stats_parsed(
@@ -2776,7 +2810,7 @@ fn test_schema_has_compatible_stats_parsed_multiple_columns() {
     ]);
 
     // First column matches, second is incompatible
-    let stats_schema = StructType::new_unchecked([
+    let stats_schema = create_stats_schema(vec![
         StructField::nullable("good_col", DataType::LONG),
         StructField::nullable("bad_col", DataType::INTEGER),
     ]);
@@ -2802,7 +2836,7 @@ fn test_schema_has_compatible_stats_parsed_missing_min_max_values() {
 
     let checkpoint_schema = StructType::new_unchecked([StructField::nullable("add", add_schema)]);
 
-    let stats_schema = StructType::new_unchecked([StructField::nullable("id", DataType::INTEGER)]);
+    let stats_schema = create_stats_schema(vec![StructField::nullable("id", DataType::INTEGER)]);
 
     // Should return true - missing minValues/maxValues is handled gracefully with continue
     assert!(LogSegment::schema_has_compatible_stats_parsed(
@@ -2828,7 +2862,7 @@ fn test_schema_has_compatible_stats_parsed_min_values_not_struct() {
 
     let checkpoint_schema = StructType::new_unchecked([StructField::nullable("add", add_schema)]);
 
-    let stats_schema = StructType::new_unchecked([StructField::nullable("id", DataType::INTEGER)]);
+    let stats_schema = create_stats_schema(vec![StructField::nullable("id", DataType::INTEGER)]);
 
     // Should return false - minValues/maxValues must be Struct types
     assert!(!LogSegment::schema_has_compatible_stats_parsed(
@@ -2837,6 +2871,157 @@ fn test_schema_has_compatible_stats_parsed_min_values_not_struct() {
     ));
 }
 
+#[test]
+fn test_schema_has_compatible_stats_parsed_nested_struct() {
+    // Create a nested struct: user: { name: string, age: integer }
+    let user_struct = StructType::new_unchecked([
+        StructField::nullable("name", DataType::STRING),
+        StructField::nullable("age", DataType::INTEGER),
+    ]);
+
+    let checkpoint_schema =
+        create_checkpoint_schema_with_stats_parsed(vec![StructField::nullable(
+            "user",
+            user_struct.clone(),
+        )]);
+
+    // Exact match should work
+    let stats_schema = create_stats_schema(vec![StructField::nullable("user", user_struct)]);
+    assert!(LogSegment::schema_has_compatible_stats_parsed(
+        &checkpoint_schema,
+        &stats_schema
+    ));
+}
+
+#[test]
+fn test_schema_has_compatible_stats_parsed_nested_struct_with_extra_fields() {
+    // Checkpoint has extra nested fields not needed by stats schema
+    let checkpoint_user = StructType::new_unchecked([
+        StructField::nullable("name", DataType::STRING),
+        StructField::nullable("age", DataType::INTEGER),
+        StructField::nullable("extra", DataType::STRING), // extra field
+    ]);
+
+    let checkpoint_schema =
+        create_checkpoint_schema_with_stats_parsed(vec![StructField::nullable(
+            "user",
+            checkpoint_user,
+        )]);
+
+    // Stats schema only needs a subset of fields
+    let stats_user = StructType::new_unchecked([StructField::nullable("name", DataType::STRING)]);
+
+    let stats_schema = create_stats_schema(vec![StructField::nullable("user", stats_user)]);
+
+    // Extra fields in checkpoint nested struct should be OK
+    assert!(LogSegment::schema_has_compatible_stats_parsed(
+        &checkpoint_schema,
+        &stats_schema
+    ));
+}
+
+#[test]
+fn test_schema_has_compatible_stats_parsed_nested_struct_missing_field_ok() {
+    // Checkpoint is missing a nested field that stats schema needs
+    let checkpoint_user =
+        StructType::new_unchecked([StructField::nullable("name", DataType::STRING)]);
+
+    let checkpoint_schema =
+        create_checkpoint_schema_with_stats_parsed(vec![StructField::nullable(
+            "user",
+            checkpoint_user,
+        )]);
+
+    // Stats schema needs more fields than checkpoint has
+    let stats_user = StructType::new_unchecked([
+        StructField::nullable("name", DataType::STRING),
+        StructField::nullable("age", DataType::INTEGER), // missing in checkpoint
+    ]);
+
+    let stats_schema = create_stats_schema(vec![StructField::nullable("user", stats_user)]);
+
+    // Missing nested field is OK - it will return null when accessed
+    assert!(LogSegment::schema_has_compatible_stats_parsed(
+        &checkpoint_schema,
+        &stats_schema
+    ));
+}
+
+#[test]
+fn test_schema_has_compatible_stats_parsed_nested_struct_type_mismatch() {
+    // Checkpoint has incompatible type in nested field
+    let checkpoint_user = StructType::new_unchecked([
+        StructField::nullable("name", DataType::INTEGER), // wrong type!
+    ]);
+
+    let checkpoint_schema =
+        create_checkpoint_schema_with_stats_parsed(vec![StructField::nullable(
+            "user",
+            checkpoint_user,
+        )]);
+
+    let stats_user = StructType::new_unchecked([StructField::nullable("name", DataType::STRING)]);
+
+    let stats_schema = create_stats_schema(vec![StructField::nullable("user", stats_user)]);
+
+    // Type mismatch in nested field should fail
+    assert!(!LogSegment::schema_has_compatible_stats_parsed(
+        &checkpoint_schema,
+        &stats_schema
+    ));
+}
+
+#[test]
+fn test_schema_has_compatible_stats_parsed_deeply_nested() {
+    // Deeply nested: company: { department: { team: { name: string } } }
+    let team = StructType::new_unchecked([StructField::nullable("name", DataType::STRING)]);
+    let department = StructType::new_unchecked([StructField::nullable("team", team.clone())]);
+    let company = StructType::new_unchecked([StructField::nullable("department", department)]);
+
+    let checkpoint_schema =
+        create_checkpoint_schema_with_stats_parsed(vec![StructField::nullable(
+            "company",
+            company.clone(),
+        )]);
+
+    let stats_schema = create_stats_schema(vec![StructField::nullable("company", company)]);
+
+    assert!(LogSegment::schema_has_compatible_stats_parsed(
+        &checkpoint_schema,
+        &stats_schema
+    ));
+}
+
+#[test]
+fn test_schema_has_compatible_stats_parsed_deeply_nested_type_mismatch() {
+    // Type mismatch deep in nested structure
+    let checkpoint_team =
+        StructType::new_unchecked([StructField::nullable("name", DataType::INTEGER)]); // wrong!
+    let checkpoint_dept =
+        StructType::new_unchecked([StructField::nullable("team", checkpoint_team)]);
+    let checkpoint_company =
+        StructType::new_unchecked([StructField::nullable("department", checkpoint_dept)]);
+
+    let checkpoint_schema =
+        create_checkpoint_schema_with_stats_parsed(vec![StructField::nullable(
+            "company",
+            checkpoint_company,
+        )]);
+
+    let stats_team = StructType::new_unchecked([StructField::nullable("name", DataType::STRING)]);
+    let stats_dept = StructType::new_unchecked([StructField::nullable("team", stats_team)]);
+    let stats_company =
+        StructType::new_unchecked([StructField::nullable("department", stats_dept)]);
+
+    let stats_schema = create_stats_schema(vec![StructField::nullable("company", stats_company)]);
+
+    // Type mismatch deep in hierarchy should be detected
+    assert!(!LogSegment::schema_has_compatible_stats_parsed(
+        &checkpoint_schema,
+        &stats_schema
+    ));
+}
+
 // ============================================================================
 // new_with_commit tests
 // ============================================================================

kernel/src/log_segment/tests.rs

@@ -3,11 +3,12 @@
 mod column_filter;
 
 use std::borrow::Cow;
+use std::sync::Arc;
 
 use crate::{
     schema::{
-        ArrayType, ColumnName, DataType, MapType, PrimitiveType, Schema, SchemaTransform,
-        StructField, StructType,
+        ArrayType, ColumnName, DataType, MapType, PrimitiveType, Schema, SchemaRef,
+        SchemaTransform, StructField, StructType,
     },
     table_properties::TableProperties,
     DeltaResult,
@@ -151,6 +152,28 @@ pub(crate) fn stats_column_names(
     columns
 }
 
+/// Creates a stats schema from a referenced schema (columns from predicate).
+/// Returns schema: `{ numRecords, nullCount, minValues, maxValues }`
+///
+/// This is used to build the schema for parsing JSON stats and for reading stats_parsed
+/// from checkpoints.
+pub(crate) fn build_stats_schema(referenced_schema: &StructType) -> Option<SchemaRef> {
+    let stats_schema = NullableStatsTransform
+        .transform_struct(referenced_schema)?
+        .into_owned();
+
+    let nullcount_schema = NullCountStatsTransform
+        .transform_struct(&stats_schema)?
+        .into_owned();
+
+    Some(Arc::new(StructType::new_unchecked([
+        StructField::nullable("numRecords", DataType::LONG),
+        StructField::nullable("nullCount", nullcount_schema),
+        StructField::nullable("minValues", stats_schema.clone()),
+        StructField::nullable("maxValues", stats_schema),
+    ])))
+}
+
 /// Transforms a schema to make all fields nullable.
 /// Used for stats schemas where stats may not be available for all columns.
 pub(crate) struct NullableStatsTransform;