PartitionedFile::metadata_size_hint serde

datafusion/common/src/scalar/mod.rs

@@ -37,7 +37,7 @@ use crate::cast::{
     as_decimal128_array, as_decimal256_array, as_dictionary_array,
     as_fixed_size_binary_array, as_fixed_size_list_array,
 };
-use crate::error::{DataFusionError, Result, _exec_err, _internal_err, _not_impl_err};
+use crate::error::{_exec_err, _internal_err, _not_impl_err, DataFusionError, Result};
 use crate::hash_utils::create_hashes;
 use crate::utils::SingleRowListArrayBuilder;
 use arrow::array::{

datafusion/proto/proto/datafusion.proto

@@ -1254,6 +1254,7 @@ message PartitionedFile {
   repeated datafusion_common.ScalarValue partition_values = 4;
   FileRange range = 5;
   datafusion_common.Statistics statistics = 6;
+  optional uint64 metadata_size_hint = 7;
 }
 
 message FileRange {

datafusion/proto/src/physical_plan/from_proto.rs

@@ -568,7 +568,7 @@ impl TryFrom<&protobuf::PartitionedFile> for PartitionedFile {
             range: val.range.as_ref().map(|v| v.try_into()).transpose()?,
             statistics: val.statistics.as_ref().map(|v| v.try_into()).transpose()?,
             extensions: None,
-            metadata_size_hint: None,
+            metadata_size_hint: val.metadata_size_hint.map(|v| v as usize),
         })
     }
 }

datafusion/proto/src/physical_plan/to_proto.rs

@@ -450,6 +450,7 @@ impl TryFrom<&PartitionedFile> for protobuf::PartitionedFile {
                 .collect::<Result<Vec<_>, _>>()?,
             range: pf.range.as_ref().map(|r| r.try_into()).transpose()?,
             statistics: pf.statistics.as_ref().map(|s| s.into()),
+            metadata_size_hint: pf.metadata_size_hint.map(|v| v as u64),
         })
     }
 }

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -50,6 +50,7 @@ use datafusion::datasource::physical_plan::{
     wrap_partition_type_in_dict, wrap_partition_value_in_dict, FileScanConfig,
     FileSinkConfig, ParquetSource,
 };
+use datafusion::datasource::source::DataSourceExec;
 use datafusion::execution::FunctionRegistry;
 use datafusion::functions_aggregate::sum::sum_udaf;
 use datafusion::functions_window::nth_value::nth_value_udwf;
@@ -682,6 +683,53 @@ fn roundtrip_sort_preserve_partitioning() -> Result<()> {
     ))
 }
 
+#[test]
+fn roundtrip_partitioned_file_metadata_size_hint() -> Result<()> {
+    let mut file_group =
+        PartitionedFile::new("/path/to/part=0/file.parquet".to_string(), 1024)
+            .with_metadata_size_hint(123);
+
+    let schema = Arc::new(Schema::new(vec![Field::new("col", DataType::Utf8, false)]));
+
+    let source = Arc::new(ParquetSource::default());
+    let scan_config = FileScanConfig {
+        object_store_url: ObjectStoreUrl::local_filesystem(),
+        file_groups: vec![vec![file_group]],
+        constraints: Constraints::empty(),
+        statistics: Statistics::new_unknown(&schema),
+        file_schema: schema,
+        projection: Some(vec![0, 1]),
+        limit: None,
+        table_partition_cols: vec![Field::new(
+            "part".to_string(),
+            wrap_partition_type_in_dict(DataType::Int16),
+            false,
+        )],
+        output_ordering: vec![],
+        file_compression_type: FileCompressionType::UNCOMPRESSED,
+        new_lines_in_values: false,
+        source,
+    };
+
+    let ctx = SessionContext::new();
+    let reconstructed_plan = roundtrip_test_and_return(
+        scan_config.build(),
+        &ctx,
+        &DefaultPhysicalExtensionCodec {},
+    )?;
+    let exec = reconstructed_plan
+        .as_any()
+        .downcast_ref::<DataSourceExec>()
+        .expect("DataSourceExec not found");
+    let scan_config = exec
+        .source()
+        .as_any()
+        .downcast_ref::<FileScanConfig>()
+        .expect("FileScanConfig not found");
+    assert_eq!(scan_config.file_groups[0][0].metadata_size_hint, Some(123));
+    Ok(())
+}
+
 #[test]
 fn roundtrip_coalesce_with_fetch() -> Result<()> {
     let field_a = Field::new("a", DataType::Boolean, false);