feat: add stats_columns to ParquetHandler

kernel/src/engine/default/mod.rs

@@ -216,7 +216,12 @@ impl<E: TaskExecutor> DefaultEngine<E> {
         )?;
         let physical_data = logical_to_physical_expr.evaluate(data)?;
         self.parquet
-            .write_parquet_file(write_context.target_dir(), physical_data, partition_values)
+            .write_parquet_file(
+                write_context.target_dir(),
+                physical_data,
+                partition_values,
+                Some(write_context.stats_columns()),
+            )
             .await
     }
 }

kernel/src/engine/default/parquet.rs

@@ -201,6 +201,7 @@ impl<E: TaskExecutor> DefaultParquetHandler<E> {
         path: &url::Url,
         data: Box<dyn EngineData>,
         partition_values: HashMap<String, String>,
+        _stats_columns: Option<&[String]>,
     ) -> DeltaResult<Box<dyn EngineData>> {
         let parquet_metadata = self.write_parquet(path, data).await?;
         parquet_metadata.as_record_batch(&partition_values)
@@ -294,6 +295,7 @@ impl<E: TaskExecutor> ParquetHandler for DefaultParquetHandler<E> {
     /// - `location` - The full URL path where the Parquet file should be written
     ///   (e.g., `s3://bucket/path/file.parquet`, `file:///path/to/file.parquet`).
     /// - `data` - An iterator of engine data to be written to the Parquet file.
+    /// - `stats_columns` - Optional column names for which statistics should be collected.
     ///
     /// # Returns
     ///
@@ -302,6 +304,7 @@ impl<E: TaskExecutor> ParquetHandler for DefaultParquetHandler<E> {
         &self,
         location: url::Url,
         mut data: Box<dyn Iterator<Item = DeltaResult<Box<dyn EngineData>>> + Send>,
+        _stats_columns: Option<&[String]>,
     ) -> DeltaResult<()> {
         let store = self.store.clone();
 
@@ -776,7 +779,7 @@ mod tests {
         // Test writing through the trait method
         let file_url = Url::parse("memory:///test/data.parquet").unwrap();
         parquet_handler
-            .write_parquet_file(file_url.clone(), data_iter)
+            .write_parquet_file(file_url.clone(), data_iter, None)
             .unwrap();
 
         // Verify we can read the file back
@@ -964,7 +967,7 @@ mod tests {
         // Write the data
         let file_url = Url::parse("memory:///roundtrip/test.parquet").unwrap();
         parquet_handler
-            .write_parquet_file(file_url.clone(), data_iter)
+            .write_parquet_file(file_url.clone(), data_iter, None)
             .unwrap();
 
         // Read it back
@@ -1152,7 +1155,7 @@ mod tests {
 
         // Write the first file
         parquet_handler
-            .write_parquet_file(file_url.clone(), data_iter1)
+            .write_parquet_file(file_url.clone(), data_iter1, None)
             .unwrap();
 
         // Create second data set with different data
@@ -1168,7 +1171,7 @@ mod tests {
 
         // Overwrite with second file (overwrite=true)
         parquet_handler
-            .write_parquet_file(file_url.clone(), data_iter2)
+            .write_parquet_file(file_url.clone(), data_iter2, None)
             .unwrap();
 
         // Read back and verify it contains the second data set
@@ -1231,7 +1234,7 @@ mod tests {
 
         // Write the first file
         parquet_handler
-            .write_parquet_file(file_url.clone(), data_iter1)
+            .write_parquet_file(file_url.clone(), data_iter1, None)
             .unwrap();
 
         // Create second data set
@@ -1247,7 +1250,7 @@ mod tests {
 
         // Write again - should overwrite successfully (new behavior always overwrites)
         parquet_handler
-            .write_parquet_file(file_url.clone(), data_iter2)
+            .write_parquet_file(file_url.clone(), data_iter2, None)
             .unwrap();
 
         // Verify the file was overwritten with the new data

kernel/src/engine/sync/parquet.rs

@@ -79,6 +79,7 @@ impl ParquetHandler for SyncParquetHandler {
     /// - `location` - The full URL path where the Parquet file should be written
     ///   (e.g., `file:///path/to/file.parquet`).
     /// - `data` - An iterator of engine data to be written to the Parquet file.
+    /// - `stats_columns` - Optional column names for which statistics should be collected.
     ///
     /// # Returns
     ///
@@ -87,6 +88,7 @@ impl ParquetHandler for SyncParquetHandler {
         &self,
         location: Url,
         mut data: Box<dyn Iterator<Item = DeltaResult<Box<dyn crate::EngineData>>> + Send>,
+        _stats_columns: Option<&[String]>,
     ) -> DeltaResult<()> {
         // Convert URL to file path
         let path = location
@@ -115,6 +117,7 @@ impl ParquetHandler for SyncParquetHandler {
 
         writer.close()?; // writer must be closed to write footer
 
+        // TODO: Implement stats collection for SyncEngine
         Ok(())
     }
 
@@ -174,7 +177,9 @@ mod tests {
         > = Box::new(std::iter::once(Ok(engine_data)));
 
         // Write the file
-        handler.write_parquet_file(url.clone(), data_iter).unwrap();
+        handler
+            .write_parquet_file(url.clone(), data_iter, None)
+            .unwrap();
 
         // Verify the file exists
         assert!(file_path.exists());
@@ -295,7 +300,9 @@ mod tests {
         > = Box::new(std::iter::once(Ok(engine_data)));
 
         // Write the file
-        handler.write_parquet_file(url.clone(), data_iter).unwrap();
+        handler
+            .write_parquet_file(url.clone(), data_iter, None)
+            .unwrap();
 
         // Verify the file exists
         assert!(file_path.exists());
@@ -370,7 +377,9 @@ mod tests {
         > = Box::new(std::iter::once(Ok(engine_data1)));
 
         // Write the first file
-        handler.write_parquet_file(url.clone(), data_iter1).unwrap();
+        handler
+            .write_parquet_file(url.clone(), data_iter1, None)
+            .unwrap();
         assert!(file_path.exists());
 
         // Create second data set with different data
@@ -386,7 +395,9 @@ mod tests {
         > = Box::new(std::iter::once(Ok(engine_data2)));
 
         // Overwrite with second file (overwrite=true)
-        handler.write_parquet_file(url.clone(), data_iter2).unwrap();
+        handler
+            .write_parquet_file(url.clone(), data_iter2, None)
+            .unwrap();
 
         // Read back and verify it contains the second data set
         let file = File::open(&file_path).unwrap();
@@ -445,7 +456,9 @@ mod tests {
         > = Box::new(std::iter::once(Ok(engine_data1)));
 
         // Write the first file
-        handler.write_parquet_file(url.clone(), data_iter1).unwrap();
+        handler
+            .write_parquet_file(url.clone(), data_iter1, None)
+            .unwrap();
         assert!(file_path.exists());
 
         // Create second data set
@@ -461,7 +474,9 @@ mod tests {
         > = Box::new(std::iter::once(Ok(engine_data2)));
 
         // Write again - should overwrite successfully (new behavior always overwrites)
-        handler.write_parquet_file(url.clone(), data_iter2).unwrap();
+        handler
+            .write_parquet_file(url.clone(), data_iter2, None)
+            .unwrap();
 
         // Verify the file was overwritten with the new data
         let file = File::open(&file_path).unwrap();
@@ -537,7 +552,9 @@ mod tests {
         > = Box::new(batches.into_iter());
 
         // Write the file
-        handler.write_parquet_file(url.clone(), data_iter).unwrap();
+        handler
+            .write_parquet_file(url.clone(), data_iter, None)
+            .unwrap();
 
         // Verify the file exists
         assert!(file_path.exists());

kernel/src/lib.rs

@@ -778,9 +778,10 @@ pub trait ParquetHandler: AsAny {
         predicate: Option<PredicateRef>,
     ) -> DeltaResult<FileDataReadResultIterator>;
 
-    /// Write data to a Parquet file at the specified URL.
+    /// Write data to a Parquet file at the specified URL, collecting statistics.
     ///
-    /// This method writes the provided `data` to a Parquet file at the given `url`.
+    /// This method writes the provided `data` to a Parquet file at the given `url`,
+    /// and collects statistics (min, max, null count) for the specified columns.
     ///
     /// This will overwrite the file if it already exists.
     ///
@@ -789,6 +790,7 @@ pub trait ParquetHandler: AsAny {
     /// - `url` - The full URL path where the Parquet file should be written
     ///   (e.g., `s3://bucket/path/file.parquet`).
     /// - `data` - An iterator of engine data to be written to the Parquet file.
+    /// - `stats_columns` - Optional column names for which statistics should be collected.
     ///
     /// # Returns
     ///
@@ -797,6 +799,7 @@ pub trait ParquetHandler: AsAny {
         &self,
         location: url::Url,
         data: Box<dyn Iterator<Item = DeltaResult<Box<dyn EngineData>>> + Send>,
+        stats_columns: Option<&[String]>,
     ) -> DeltaResult<()>;
 
     /// Read the footer metadata from a Parquet file without reading the data.