From 7e4400a026f2ed1be2f2752e2779b8ef35091308 Mon Sep 17 00:00:00 2001 From: Jiawei Zhao Date: Fri, 10 Jul 2026 19:24:39 +0800 Subject: [PATCH] refactor(proto): remove legacy scan field DataFusion does not guarantee serialized plans across versions. Keeping `partitioned_by_file_group` therefore leaves a dead schema field and decoder path after `output_partitioning` became the source of truth. Reserve the old field number and name to prevent future reuse. Signed-off-by: Jiawei Zhao --- .../proto-models/proto/datafusion.proto | 4 +- .../proto-models/src/generated/pbjson.rs | 18 ------ .../proto-models/src/generated/prost.rs | 2 - .../proto/src/physical_plan/from_proto.rs | 18 +----- .../proto/src/physical_plan/to_proto.rs | 3 - .../tests/cases/roundtrip_physical_plan.rs | 63 +------------------ .../library-user-guide/upgrading/55.0.0.md | 8 +++ 7 files changed, 13 insertions(+), 103 deletions(-) diff --git a/datafusion/proto-models/proto/datafusion.proto b/datafusion/proto-models/proto/datafusion.proto index d68973c44ecbf..e365e23e00c2d 100644 --- a/datafusion/proto-models/proto/datafusion.proto +++ b/datafusion/proto-models/proto/datafusion.proto @@ -1190,7 +1190,9 @@ message FileScanExecConf { optional uint64 batch_size = 12; optional ProjectionExprs projection_exprs = 13; - optional bool partitioned_by_file_group = 14; + // Was optional bool partitioned_by_file_group = 14. + reserved 14; + reserved "partitioned_by_file_group"; optional Partitioning output_partitioning = 15; } diff --git a/datafusion/proto-models/src/generated/pbjson.rs b/datafusion/proto-models/src/generated/pbjson.rs index f8e21030356b0..12458f12621e6 100644 --- a/datafusion/proto-models/src/generated/pbjson.rs +++ b/datafusion/proto-models/src/generated/pbjson.rs @@ -6963,9 +6963,6 @@ impl serde::Serialize for FileScanExecConf { if self.projection_exprs.is_some() { len += 1; } - if self.partitioned_by_file_group.is_some() { - len += 1; - } if self.output_partitioning.is_some() { len += 1; } @@ -7005,9 +7002,6 @@ impl serde::Serialize for FileScanExecConf { if let Some(v) = self.projection_exprs.as_ref() { struct_ser.serialize_field("projectionExprs", v)?; } - if let Some(v) = self.partitioned_by_file_group.as_ref() { - struct_ser.serialize_field("partitionedByFileGroup", v)?; - } if let Some(v) = self.output_partitioning.as_ref() { struct_ser.serialize_field("outputPartitioning", v)?; } @@ -7038,8 +7032,6 @@ impl<'de> serde::Deserialize<'de> for FileScanExecConf { "batchSize", "projection_exprs", "projectionExprs", - "partitioned_by_file_group", - "partitionedByFileGroup", "output_partitioning", "outputPartitioning", ]; @@ -7057,7 +7049,6 @@ impl<'de> serde::Deserialize<'de> for FileScanExecConf { Constraints, BatchSize, ProjectionExprs, - PartitionedByFileGroup, OutputPartitioning, } impl<'de> serde::Deserialize<'de> for GeneratedField { @@ -7091,7 +7082,6 @@ impl<'de> serde::Deserialize<'de> for FileScanExecConf { "constraints" => Ok(GeneratedField::Constraints), "batchSize" | "batch_size" => Ok(GeneratedField::BatchSize), "projectionExprs" | "projection_exprs" => Ok(GeneratedField::ProjectionExprs), - "partitionedByFileGroup" | "partitioned_by_file_group" => Ok(GeneratedField::PartitionedByFileGroup), "outputPartitioning" | "output_partitioning" => Ok(GeneratedField::OutputPartitioning), _ => Err(serde::de::Error::unknown_field(value, FIELDS)), } @@ -7123,7 +7113,6 @@ impl<'de> serde::Deserialize<'de> for FileScanExecConf { let mut constraints__ = None; let mut batch_size__ = None; let mut projection_exprs__ = None; - let mut partitioned_by_file_group__ = None; let mut output_partitioning__ = None; while let Some(k) = map_.next_key()? { match k { @@ -7198,12 +7187,6 @@ impl<'de> serde::Deserialize<'de> for FileScanExecConf { } projection_exprs__ = map_.next_value()?; } - GeneratedField::PartitionedByFileGroup => { - if partitioned_by_file_group__.is_some() { - return Err(serde::de::Error::duplicate_field("partitionedByFileGroup")); - } - partitioned_by_file_group__ = map_.next_value()?; - } GeneratedField::OutputPartitioning => { if output_partitioning__.is_some() { return Err(serde::de::Error::duplicate_field("outputPartitioning")); @@ -7224,7 +7207,6 @@ impl<'de> serde::Deserialize<'de> for FileScanExecConf { constraints: constraints__, batch_size: batch_size__, projection_exprs: projection_exprs__, - partitioned_by_file_group: partitioned_by_file_group__, output_partitioning: output_partitioning__, }) } diff --git a/datafusion/proto-models/src/generated/prost.rs b/datafusion/proto-models/src/generated/prost.rs index 675ead23f4914..f3cc83ccbe9cc 100644 --- a/datafusion/proto-models/src/generated/prost.rs +++ b/datafusion/proto-models/src/generated/prost.rs @@ -1814,8 +1814,6 @@ pub struct FileScanExecConf { pub batch_size: ::core::option::Option, #[prost(message, optional, tag = "13")] pub projection_exprs: ::core::option::Option, - #[prost(bool, optional, tag = "14")] - pub partitioned_by_file_group: ::core::option::Option, #[prost(message, optional, tag = "15")] pub output_partitioning: ::core::option::Option, } diff --git a/datafusion/proto/src/physical_plan/from_proto.rs b/datafusion/proto/src/physical_plan/from_proto.rs index 19e49f3cb8724..8f894bfa404ea 100644 --- a/datafusion/proto/src/physical_plan/from_proto.rs +++ b/datafusion/proto/src/physical_plan/from_proto.rs @@ -29,9 +29,7 @@ use datafusion_common::{ }; use datafusion_datasource::file::FileSource; use datafusion_datasource::file_groups::FileGroup; -use datafusion_datasource::file_scan_config::{ - FileScanConfig, FileScanConfigBuilder, output_partitioning_from_partition_fields, -}; +use datafusion_datasource::file_scan_config::{FileScanConfig, FileScanConfigBuilder}; use datafusion_datasource::file_sink_config::FileSinkConfig; use datafusion_datasource::{FileRange, ListingTableUrl, PartitionedFile, TableSchema}; use datafusion_datasource_csv::file_format::CsvSink; @@ -560,20 +558,6 @@ pub fn parse_protobuf_file_scan_config( &schema, proto_converter, )?; - let output_partitioning = match output_partitioning { - Some(output_partitioning) => Some(output_partitioning), - None if proto.partitioned_by_file_group.unwrap_or(false) => { - // Backward compatibility: older serialized plans used only - // `partitioned_by_file_group` to declare scan output partitioning. - let table_schema = parse_table_schema_from_proto(proto)?; - output_partitioning_from_partition_fields( - &schema, - table_schema.table_partition_cols(), - file_groups.len(), - ) - } - None => None, - }; // Parse projection expressions if present and apply to file source let file_source = if let Some(proto_projection_exprs) = &proto.projection_exprs { diff --git a/datafusion/proto/src/physical_plan/to_proto.rs b/datafusion/proto/src/physical_plan/to_proto.rs index e93bb9cc5fed2..29160d82b0430 100644 --- a/datafusion/proto/src/physical_plan/to_proto.rs +++ b/datafusion/proto/src/physical_plan/to_proto.rs @@ -562,9 +562,6 @@ pub fn serialize_file_scan_config( constraints: Some(conf.constraints.clone().into()), batch_size: conf.batch_size.map(|s| s as u64), projection_exprs, - // Partition grouping is now encoded in `output_partitioning`; this legacy - // wire field is left unset (readers rely on `output_partitioning`). - partitioned_by_file_group: None, output_partitioning, }) } diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs index bb99a1ecbf4ed..7e8d87fd8c6bb 100644 --- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs +++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs @@ -124,12 +124,7 @@ use datafusion_proto::bytes::{ physical_plan_from_bytes_with_proto_converter, physical_plan_to_bytes_with_proto_converter, }; -use datafusion_proto::physical_plan::from_proto::{ - parse_protobuf_file_scan_config, parse_table_schema_from_proto, -}; -use datafusion_proto::physical_plan::to_proto::{ - serialize_file_scan_config, serialize_physical_expr_with_converter, -}; +use datafusion_proto::physical_plan::to_proto::serialize_physical_expr_with_converter; use datafusion_proto::physical_plan::{ AsExecutionPlan, DeduplicatingProtoConverter, DefaultPhysicalExtensionCodec, DefaultPhysicalProtoConverter, PhysicalExtensionCodec, PhysicalPlanDecodeContext, @@ -4248,62 +4243,6 @@ fn roundtrip_parquet_exec_output_partitioning() -> Result<()> { Ok(()) } -#[test] -fn parse_legacy_partitioned_by_file_group_as_output_partitioning() -> Result<()> { - let file_schema = - Arc::new(Schema::new(vec![Field::new("col", DataType::Utf8, false)])); - let table_schema = TableSchema::builder(Arc::clone(&file_schema)) - .with_table_partition_cols(vec![Arc::new(Field::new( - "part", - DataType::Utf8, - false, - ))]) - .build(); - let file_source = Arc::new(ParquetSource::new(table_schema)); - let scan_config = - FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), file_source) - .with_file_groups(vec![ - FileGroup::new(vec![PartitionedFile::new( - "/path/to/file1.parquet".to_string(), - 1024, - )]), - FileGroup::new(vec![PartitionedFile::new( - "/path/to/file2.parquet".to_string(), - 1024, - )]), - ]) - .build(); - - let codec = DefaultPhysicalExtensionCodec {}; - let proto_converter = DefaultPhysicalProtoConverter {}; - let mut proto = serialize_file_scan_config(&scan_config, &codec, &proto_converter)?; - proto.partitioned_by_file_group = Some(true); - proto.output_partitioning = None; - - let ctx = SessionContext::new(); - let task_ctx = ctx.task_ctx(); - let decode_ctx = PhysicalPlanDecodeContext::new(task_ctx.as_ref(), &codec); - let parsed = parse_protobuf_file_scan_config( - &proto, - &decode_ctx, - &proto_converter, - Arc::new(ParquetSource::new(parse_table_schema_from_proto(&proto)?)), - )?; - - match parsed.output_partitioning { - Some(Partitioning::Hash(exprs, partition_count)) => { - assert_eq!(partition_count, 2); - assert_eq!(exprs.len(), 1); - let column = exprs[0].downcast_ref::().unwrap(); - assert_eq!(column.name(), "part"); - assert_eq!(column.index(), 1); - } - other => panic!("Expected legacy hash output partitioning, got {other:?}"), - } - - Ok(()) -} - #[test] fn roundtrip_parquet_exec_range_output_partitioning() -> Result<()> { let file_schema = diff --git a/docs/source/library-user-guide/upgrading/55.0.0.md b/docs/source/library-user-guide/upgrading/55.0.0.md index 181e0e0b7f266..ae50c9f780da3 100644 --- a/docs/source/library-user-guide/upgrading/55.0.0.md +++ b/docs/source/library-user-guide/upgrading/55.0.0.md @@ -73,12 +73,17 @@ let df = df.fill_null(&ScalarValue::from(0), &[])?; `FileScanConfigBuilder::with_partitioned_by_file_group(...)` have been removed. Use `FileScanConfig::output_partitioning` and `FileScanConfigBuilder::with_output_partitioning(...)` instead. +The corresponding +`datafusion_proto::protobuf::FileScanExecConf::partitioned_by_file_group` +field has also been removed. **Who is affected:** - Users who accessed `FileScanConfig::partitioned_by_file_group` directly. - Users who called `FileScanConfigBuilder::with_partitioned_by_file_group(true)`. +- Users who constructed or accessed + `datafusion_proto::protobuf::FileScanExecConf::partitioned_by_file_group`. **Migration guide:** @@ -108,6 +113,9 @@ otherwise. If you construct the partitioning manually, pass `Some(Partitioning::Hash(partition_exprs, partition_count))` to `with_output_partitioning(...)`. +When constructing `FileScanExecConf`, omit `partitioned_by_file_group` and set +`output_partitioning` instead. + ### User `SpillFile` traits instead of [`RefCountedTempFile`] Spill file APIs now use the `datafusion_execution::SpillFile` trait instead of