feat(python, rust): add statistics_enabled to ColumnProperties

python/deltalake/table.py

@@ -217,22 +217,25 @@ class ColumnProperties:
     def __init__(
         self,
         dictionary_enabled: Optional[bool] = None,
-        max_statistics_size: Optional[int] = None,
+        statistics_enabled: Optional[Literal["NONE", "CHUNK", "PAGE"]] = None,
         bloom_filter_properties: Optional[BloomFilterProperties] = None,
     ):
         """Create a Column Properties instance for the Rust parquet writer:
 
         Args:
             dictionary_enabled: Enable dictionary encoding for the column.
-            max_statistics_size: Maximum size of statistics for the column.
+            statistics_enabled: Statistics level for the column.
             bloom_filter_properties: Bloom Filter Properties for the column.
         """
         self.dictionary_enabled = dictionary_enabled
-        self.max_statistics_size = max_statistics_size
+        self.statistics_enabled = statistics_enabled
         self.bloom_filter_properties = bloom_filter_properties
 
     def __str__(self) -> str:
-        return f"dictionary_enabled: {self.dictionary_enabled}, max_statistics_size: {self.max_statistics_size}, bloom_filter_properties: {self.bloom_filter_properties}"
+        return (
+            f"dictionary_enabled: {self.dictionary_enabled}, statistics_enabled: {self.statistics_enabled}, "
+            f"bloom_filter_properties: {self.bloom_filter_properties}"
+        )
 
 
 @dataclass(init=True)

python/src/lib.rs

@@ -52,7 +52,7 @@ use deltalake::operations::vacuum::VacuumBuilder;
 use deltalake::operations::{collect_sendable_stream, CustomExecuteHandler};
 use deltalake::parquet::basic::Compression;
 use deltalake::parquet::errors::ParquetError;
-use deltalake::parquet::file::properties::WriterProperties;
+use deltalake::parquet::file::properties::{EnabledStatistics, WriterProperties};
 use deltalake::partitions::PartitionFilter;
 use deltalake::protocol::{DeltaOperation, SaveMode};
 use deltalake::storage::{IORuntime, ObjectStoreRef};
@@ -1566,8 +1566,12 @@ fn set_writer_properties(writer_properties: PyWriterProperties) -> DeltaResult<W
         if let Some(dictionary_enabled) = default_column_properties.dictionary_enabled {
             properties = properties.set_dictionary_enabled(dictionary_enabled);
         }
-        if let Some(max_statistics_size) = default_column_properties.max_statistics_size {
-            properties = properties.set_max_statistics_size(max_statistics_size);
+        if let Some(statistics_enabled) = default_column_properties.statistics_enabled {
+            let enabled_statistics: EnabledStatistics = statistics_enabled
+                .parse()
+                .map_err(|err: String| DeltaTableError::Generic(err))?;
+
+            properties = properties.set_statistics_enabled(enabled_statistics);
         }
         if let Some(bloom_filter_properties) = default_column_properties.bloom_filter_properties {
             if let Some(set_bloom_filter_enabled) = bloom_filter_properties.set_bloom_filter_enabled
@@ -1591,6 +1595,16 @@ fn set_writer_properties(writer_properties: PyWriterProperties) -> DeltaResult<W
                         dictionary_enabled,
                     );
                 }
+                if let Some(statistics_enabled) = column_prop.statistics_enabled {
+                    let enabled_statistics: EnabledStatistics = statistics_enabled
+                        .parse()
+                        .map_err(|err: String| DeltaTableError::Generic(err))?;
+
+                    properties = properties.set_column_statistics_enabled(
+                        column_name.clone().into(),
+                        enabled_statistics,
+                    );
+                }
                 if let Some(bloom_filter_properties) = column_prop.bloom_filter_properties {
                     if let Some(set_bloom_filter_enabled) =
                         bloom_filter_properties.set_bloom_filter_enabled
@@ -1919,7 +1933,7 @@ pub struct BloomFilterProperties {
 #[derive(FromPyObject)]
 pub struct ColumnProperties {
     pub dictionary_enabled: Option<bool>,
-    pub max_statistics_size: Option<usize>,
+    pub statistics_enabled: Option<String>,
     pub bloom_filter_properties: Option<BloomFilterProperties>,
 }
 

python/tests/test_writerproperties.py

@@ -28,14 +28,14 @@ def test_writer_properties_all_filled():
         column_properties={
             "a": ColumnProperties(
                 dictionary_enabled=True,
-                max_statistics_size=40,
+                statistics_enabled="CHUNK",
                 bloom_filter_properties=BloomFilterProperties(
                     set_bloom_filter_enabled=True, fpp=0.2, ndv=30
                 ),
             ),
             "b": ColumnProperties(
                 dictionary_enabled=True,
-                max_statistics_size=400,
+                statistics_enabled="PAGE",
                 bloom_filter_properties=BloomFilterProperties(
                     set_bloom_filter_enabled=False, fpp=0.2, ndv=30
                 ),