PartitionedFile statistics grow by one ColumnStatistics per partition column on every protobuf round trip

[adriangb]

Describe the bug

PartitionedFile statistics do not survive a protobuf round trip when the file has partition values: every encode/decode cycle appends one extra ColumnStatistics entry per partition column.

PartitionedFile::statistics covers the full table schema (file columns followed by partition columns). The encode side writes that vector as-is:

https://github.com/apache/datafusion/blob/main/datafusion/proto/src/physical_plan/to_proto.rs#L427-L459

but the decode side rebuilds the file with with_partition_values(...) and then calls with_statistics(...):

https://github.com/apache/datafusion/blob/main/datafusion/proto/src/physical_plan/from_proto.rs#L548-L579

and with_statistics derives and appends a ColumnStatistics for each partition value on top of whatever it is given:

https://github.com/apache/datafusion/blob/main/datafusion/datasource/src/mod.rs#L344-L369

So the partition-column statistics that were already on the wire get appended a second time, and column_statistics.len() grows by partition_values.len() per round trip. Plans that are serialized more than once (a common pattern in distributed engines that re-serialize a plan per stage) keep growing.

To Reproduce

use std::sync::Arc;
use arrow::datatypes::{DataType, Field, Schema};
use datafusion_common::{ScalarValue, Statistics};
use datafusion_datasource::PartitionedFile;
use datafusion_proto::TryFromProto;
use datafusion_proto::protobuf;

let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)]));
let pf = PartitionedFile::new("foo/bar.parquet", 1234)
    .with_partition_values(vec![ScalarValue::from("2024-01-01")])
    .with_statistics(Arc::new(Statistics::new_unknown(&schema)));

// file column + 1 partition column
assert_eq!(pf.statistics.as_ref().unwrap().column_statistics.len(), 2);

let proto = protobuf::PartitionedFile::try_from_proto(&pf).unwrap();
let decoded = PartitionedFile::try_from_proto(&proto).unwrap();

// 3, not 2 — the partition column statistic is appended again
assert_eq!(
    decoded.statistics.as_ref().unwrap().column_statistics.len(),
    2
);

Expected behavior

PartitionedFile statistics round-trip unchanged: decoded.statistics == pf.statistics.

The decode path should set statistics directly (the wire already carries the full-table-schema statistics) rather than routing through with_statistics, which is a builder helper meant for callers supplying file-only statistics.

Additional context

Found while reviewing #23683, but this is on main today and independent of that PR. Same behavior applies to any other caller that hands with_statistics a statistics vector that already includes the partition columns.

[buraksenn]

take