feat: add stats_columns to ParquetHandler + Transaction stats API

- Add stats_columns parameter to write_parquet_file trait
- Add stats_schema(), stats_columns(), get_clustering_columns() to Transaction
- Add stats_columns to WriteContext
- Update get_write_context() to take engine parameter
- Add clustering column support to expected_stats_schema()

Author: DrakeLin

kernel/examples/write-table/src/main.rs

@@ -94,7 +94,7 @@ async fn try_main() -> DeltaResult<()> {
         .with_data_change(true);
 
     // Write the data using the engine
-    let write_context = Arc::new(txn.get_write_context());
+    let write_context = Arc::new(txn.get_write_context(&engine)?);
     let file_metadata = engine
         .write_parquet(&sample_data, write_context.as_ref(), HashMap::new())
         .await?;

kernel/src/engine/default/parquet.rs

@@ -302,6 +302,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: &[String],
     ) -> DeltaResult<()> {
         let store = self.store.clone();
 
@@ -776,7 +777,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, &[])
             .unwrap();
 
         // Verify we can read the file back
@@ -964,7 +965,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, &[])
             .unwrap();
 
         // Read it back
@@ -1152,7 +1153,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, &[])
             .unwrap();
 
         // Create second data set with different data
@@ -1168,7 +1169,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, &[])
             .unwrap();
 
         // Read back and verify it contains the second data set
@@ -1231,7 +1232,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, &[])
             .unwrap();
 
         // Create second data set
@@ -1247,7 +1248,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, &[])
             .unwrap();
 
         // Verify the file was overwritten with the new data

kernel/src/engine/sync/parquet.rs

@@ -87,6 +87,7 @@ impl ParquetHandler for SyncParquetHandler {
         &self,
         location: Url,
         mut data: Box<dyn Iterator<Item = DeltaResult<Box<dyn crate::EngineData>>> + Send>,
+        _stats_columns: &[String],
     ) -> DeltaResult<()> {
         // Convert URL to file path
         let path = location
@@ -115,6 +116,7 @@ impl ParquetHandler for SyncParquetHandler {
 
         writer.close()?; // writer must be closed to write footer
 
+        // TODO: Implement stats collection for SyncEngine
         Ok(())
     }
 
@@ -174,7 +176,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, &[])
+            .unwrap();
 
         // Verify the file exists
         assert!(file_path.exists());
@@ -295,7 +299,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, &[])
+            .unwrap();
 
         // Verify the file exists
         assert!(file_path.exists());
@@ -370,7 +376,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, &[])
+            .unwrap();
         assert!(file_path.exists());
 
         // Create second data set with different data
@@ -386,7 +394,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, &[])
+            .unwrap();
 
         // Read back and verify it contains the second data set
         let file = File::open(&file_path).unwrap();
@@ -445,7 +455,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, &[])
+            .unwrap();
         assert!(file_path.exists());
 
         // Create second data set
@@ -461,7 +473,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, &[])
+            .unwrap();
 
         // Verify the file was overwritten with the new data
         let file = File::open(&file_path).unwrap();
@@ -537,7 +551,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, &[])
+            .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` - 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: &[String],
     ) -> DeltaResult<()>;
 
     /// Read the footer metadata from a Parquet file without reading the data.