fix: make struct fields nullable in stats schema

crates/core/src/kernel/snapshot/mod.rs

@@ -315,13 +315,7 @@ impl Snapshot {
                 .fields
                 .iter()
                 .enumerate()
-                .filter_map(|(idx, f)| match f.data_type() {
-                    DataType::Map(_) | DataType::Array(_) | &DataType::BINARY => None,
-                    _ if num_indexed_cols < 0 || (idx as i32) < num_indexed_cols => {
-                        Some(StructField::new(f.name(), f.data_type().clone(), true))
-                    }
-                    _ => None,
-                })
+                .filter_map(|(idx, f)| stats_field(idx, num_indexed_cols, f))
                 .collect()
         };
         Ok(StructType::new(vec![
@@ -538,6 +532,31 @@ impl EagerSnapshot {
     }
 }
 
+fn stats_field(idx: usize, num_indexed_cols: i32, field: &StructField) -> Option<StructField> {
+    if !(num_indexed_cols < 0 || (idx as i32) < num_indexed_cols) {
+        return None;
+    }
+    match field.data_type() {
+        DataType::Map(_) | DataType::Array(_) | &DataType::BINARY => None,
+        DataType::Struct(dt_struct) => Some(StructField::new(
+            field.name(),
+            StructType::new(
+                dt_struct
+                    .fields()
+                    .iter()
+                    .flat_map(|f| stats_field(idx, num_indexed_cols, f))
+                    .collect(),
+            ),
+            true,
+        )),
+        DataType::Primitive(_) => Some(StructField::new(
+            field.name(),
+            field.data_type.clone(),
+            true,
+        )),
+    }
+}
+
 fn to_count_field(field: &StructField) -> Option<StructField> {
     match field.data_type() {
         DataType::Map(_) | DataType::Array(_) | &DataType::BINARY => None,

crates/core/tests/read_delta_log_test.rs

@@ -172,3 +172,13 @@ async fn read_delta_table_from_dlt() {
     assert_eq!(table.version(), 1);
     assert!(table.get_schema().is_ok());
 }
+
+#[tokio::test]
+async fn read_delta_table_with_null_stats_in_notnull_struct() {
+    let table =
+        deltalake_core::open_table("../test/tests/data/table_with_null_stats_in_notnull_struct")
+            .await
+            .unwrap();
+    assert_eq!(table.version(), 1);
+    assert!(table.get_schema().is_ok());
+}

crates/test/tests/data/table_with_null_stats_in_notnull_struct/_delta_log/00000000000000000000.json

@@ -0,0 +1,3 @@
+{"commitInfo":{"timestamp":1711560767220,"operation":"CREATE OR REPLACE TABLE","operationParameters":{"isManaged":"false","description":null,"partitionBy":"[]","properties":"{}"},"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{},"engineInfo":"Apache-Spark/3.5.1 Delta-Lake/3.1.0","txnId":"99840af5-4b41-4a0e-b1b8-8d22ae77ccf6"}}
+{"metaData":{"id":"e849a43f-42e1-404e-b13d-224d85eb7573","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"s\",\"type\":{\"type\":\"struct\",\"fields\":[{\"name\":\"l\",\"type\":\"long\",\"nullable\":false,\"metadata\":{}},{\"name\":\"b\",\"type\":\"boolean\",\"nullable\":false,\"metadata\":{}}]},\"nullable\":false,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1711560767211}}
+{"protocol":{"minReaderVersion":1,"minWriterVersion":2}}

crates/test/tests/data/table_with_null_stats_in_notnull_struct/_delta_log/00000000000000000001.json

@@ -0,0 +1,3 @@
+{"commitInfo":{"timestamp":1711560789508,"operation":"WRITE","operationParameters":{"mode":"Append","partitionBy":"[]"},"readVersion":0,"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numFiles":"3","numOutputRows":"2","numOutputBytes":"1978"},"engineInfo":"Apache-Spark/3.5.1 Delta-Lake/3.1.0","txnId":"80eed5be-d309-42dd-91e9-e1215a03e6dd"}}
+{"add":{"path":"part-00004-e7a7f9d9-18cb-45ee-9d64-6c711cd8eb77-c000.snappy.parquet","partitionValues":{},"size":763,"modificationTime":1711560789496,"dataChange":true,"stats":"{\"numRecords\":1,\"minValues\":{\"s\":{\"l\":10}},\"maxValues\":{\"s\":{\"l\":10}},\"nullCount\":{\"s\":{\"l\":0,\"b\":0}}}"}}
+{"add":{"path":"part-00009-ffa42641-8c4e-4f27-875e-bbb68e3bb100-c000.snappy.parquet","partitionValues":{},"size":762,"modificationTime":1711560789496,"dataChange":true,"stats":"{\"numRecords\":1,\"minValues\":{\"s\":{\"l\":20}},\"maxValues\":{\"s\":{\"l\":20}},\"nullCount\":{\"s\":{\"l\":0,\"b\":0}}}"}}