Skip to content

Commit bb670fb

Browse files
Phoenix500526alamb
andauthored
refactor(proto): remove legacy scan field (#23445)
## Which issue does this PR close? <!-- We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123. --> - Closes #N/A. ## Rationale for this change <!-- Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed. Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes. --> Since DataFusion doesn't typically guartantee wire format compatibility, I remove the backward compatiblity shim that introduced in PR #23189 FYI: #23189 (comment) ## What changes are included in this PR? <!-- There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR. --> 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. ## Are these changes tested? <!-- We typically require tests for all PRs in order to: 1. Prevent the code from being accidentally broken by subsequent changes 2. Serve as another way to document the expected behavior of the code If tests are not included in your PR, please explain why (for example, are they covered by existing tests)? --> Yes ## Are there any user-facing changes? <!-- If there are user-facing changes then we may require documentation to be updated before approving the PR. --> <!-- If there are any breaking changes to public APIs, please add the `api change` label. --> Yes Signed-off-by: Jiawei Zhao <[email protected]> Co-authored-by: Andrew Lamb <[email protected]>
1 parent 551c592 commit bb670fb

7 files changed

Lines changed: 13 additions & 103 deletions

File tree

‎datafusion/proto-models/proto/datafusion.proto‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1231,7 +1231,9 @@ message FileScanExecConf {
12311231
optional uint64 batch_size = 12;
12321232

12331233
optional ProjectionExprs projection_exprs = 13;
1234-
optional bool partitioned_by_file_group = 14;
1234+
// Was optional bool partitioned_by_file_group = 14.
1235+
reserved 14;
1236+
reserved "partitioned_by_file_group";
12351237
optional Partitioning output_partitioning = 15;
12361238
}
12371239

‎datafusion/proto-models/src/generated/pbjson.rs‎

Lines changed: 0 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -6999,9 +6999,6 @@ impl serde::Serialize for FileScanExecConf {
69996999
if self.projection_exprs.is_some() {
70007000
len += 1;
70017001
}
7002-
if self.partitioned_by_file_group.is_some() {
7003-
len += 1;
7004-
}
70057002
if self.output_partitioning.is_some() {
70067003
len += 1;
70077004
}
@@ -7041,9 +7038,6 @@ impl serde::Serialize for FileScanExecConf {
70417038
if let Some(v) = self.projection_exprs.as_ref() {
70427039
struct_ser.serialize_field("projectionExprs", v)?;
70437040
}
7044-
if let Some(v) = self.partitioned_by_file_group.as_ref() {
7045-
struct_ser.serialize_field("partitionedByFileGroup", v)?;
7046-
}
70477041
if let Some(v) = self.output_partitioning.as_ref() {
70487042
struct_ser.serialize_field("outputPartitioning", v)?;
70497043
}
@@ -7074,8 +7068,6 @@ impl<'de> serde::Deserialize<'de> for FileScanExecConf {
70747068
"batchSize",
70757069
"projection_exprs",
70767070
"projectionExprs",
7077-
"partitioned_by_file_group",
7078-
"partitionedByFileGroup",
70797071
"output_partitioning",
70807072
"outputPartitioning",
70817073
];
@@ -7093,7 +7085,6 @@ impl<'de> serde::Deserialize<'de> for FileScanExecConf {
70937085
Constraints,
70947086
BatchSize,
70957087
ProjectionExprs,
7096-
PartitionedByFileGroup,
70977088
OutputPartitioning,
70987089
}
70997090
impl<'de> serde::Deserialize<'de> for GeneratedField {
@@ -7127,7 +7118,6 @@ impl<'de> serde::Deserialize<'de> for FileScanExecConf {
71277118
"constraints" => Ok(GeneratedField::Constraints),
71287119
"batchSize" | "batch_size" => Ok(GeneratedField::BatchSize),
71297120
"projectionExprs" | "projection_exprs" => Ok(GeneratedField::ProjectionExprs),
7130-
"partitionedByFileGroup" | "partitioned_by_file_group" => Ok(GeneratedField::PartitionedByFileGroup),
71317121
"outputPartitioning" | "output_partitioning" => Ok(GeneratedField::OutputPartitioning),
71327122
_ => Err(serde::de::Error::unknown_field(value, FIELDS)),
71337123
}
@@ -7159,7 +7149,6 @@ impl<'de> serde::Deserialize<'de> for FileScanExecConf {
71597149
let mut constraints__ = None;
71607150
let mut batch_size__ = None;
71617151
let mut projection_exprs__ = None;
7162-
let mut partitioned_by_file_group__ = None;
71637152
let mut output_partitioning__ = None;
71647153
while let Some(k) = map_.next_key()? {
71657154
match k {
@@ -7234,12 +7223,6 @@ impl<'de> serde::Deserialize<'de> for FileScanExecConf {
72347223
}
72357224
projection_exprs__ = map_.next_value()?;
72367225
}
7237-
GeneratedField::PartitionedByFileGroup => {
7238-
if partitioned_by_file_group__.is_some() {
7239-
return Err(serde::de::Error::duplicate_field("partitionedByFileGroup"));
7240-
}
7241-
partitioned_by_file_group__ = map_.next_value()?;
7242-
}
72437226
GeneratedField::OutputPartitioning => {
72447227
if output_partitioning__.is_some() {
72457228
return Err(serde::de::Error::duplicate_field("outputPartitioning"));
@@ -7260,7 +7243,6 @@ impl<'de> serde::Deserialize<'de> for FileScanExecConf {
72607243
constraints: constraints__,
72617244
batch_size: batch_size__,
72627245
projection_exprs: projection_exprs__,
7263-
partitioned_by_file_group: partitioned_by_file_group__,
72647246
output_partitioning: output_partitioning__,
72657247
})
72667248
}

‎datafusion/proto-models/src/generated/prost.rs‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1875,8 +1875,6 @@ pub struct FileScanExecConf {
18751875
pub batch_size: ::core::option::Option<u64>,
18761876
#[prost(message, optional, tag = "13")]
18771877
pub projection_exprs: ::core::option::Option<ProjectionExprs>,
1878-
#[prost(bool, optional, tag = "14")]
1879-
pub partitioned_by_file_group: ::core::option::Option<bool>,
18801878
#[prost(message, optional, tag = "15")]
18811879
pub output_partitioning: ::core::option::Option<Partitioning>,
18821880
}

‎datafusion/proto/src/physical_plan/from_proto.rs‎

Lines changed: 1 addition & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -29,9 +29,7 @@ use datafusion_common::{
2929
};
3030
use datafusion_datasource::file::FileSource;
3131
use datafusion_datasource::file_groups::FileGroup;
32-
use datafusion_datasource::file_scan_config::{
33-
FileScanConfig, FileScanConfigBuilder, output_partitioning_from_partition_fields,
34-
};
32+
use datafusion_datasource::file_scan_config::{FileScanConfig, FileScanConfigBuilder};
3533
use datafusion_datasource::file_sink_config::FileSinkConfig;
3634
use datafusion_datasource::{FileRange, ListingTableUrl, PartitionedFile, TableSchema};
3735
use datafusion_datasource_csv::file_format::CsvSink;
@@ -582,20 +580,6 @@ pub fn parse_protobuf_file_scan_config(
582580
&schema,
583581
proto_converter,
584582
)?;
585-
let output_partitioning = match output_partitioning {
586-
Some(output_partitioning) => Some(output_partitioning),
587-
None if proto.partitioned_by_file_group.unwrap_or(false) => {
588-
// Backward compatibility: older serialized plans used only
589-
// `partitioned_by_file_group` to declare scan output partitioning.
590-
let table_schema = parse_table_schema_from_proto(proto)?;
591-
output_partitioning_from_partition_fields(
592-
&schema,
593-
table_schema.table_partition_cols(),
594-
file_groups.len(),
595-
)
596-
}
597-
None => None,
598-
};
599583

600584
// Parse projection expressions if present and apply to file source
601585
let file_source = if let Some(proto_projection_exprs) = &proto.projection_exprs {

‎datafusion/proto/src/physical_plan/to_proto.rs‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -563,9 +563,6 @@ pub fn serialize_file_scan_config(
563563
constraints: Some(conf.constraints.clone().into()),
564564
batch_size: conf.batch_size.map(|s| s as u64),
565565
projection_exprs,
566-
// Partition grouping is now encoded in `output_partitioning`; this legacy
567-
// wire field is left unset (readers rely on `output_partitioning`).
568-
partitioned_by_file_group: None,
569566
output_partitioning,
570567
})
571568
}

‎datafusion/proto/tests/cases/roundtrip_physical_plan.rs‎

Lines changed: 1 addition & 62 deletions
Original file line numberDiff line numberDiff line change
@@ -133,12 +133,7 @@ use datafusion_proto::bytes::{
133133
physical_plan_from_bytes_with_proto_converter,
134134
physical_plan_to_bytes_with_proto_converter,
135135
};
136-
use datafusion_proto::physical_plan::from_proto::{
137-
parse_protobuf_file_scan_config, parse_table_schema_from_proto,
138-
};
139-
use datafusion_proto::physical_plan::to_proto::{
140-
serialize_file_scan_config, serialize_physical_expr_with_converter,
141-
};
136+
use datafusion_proto::physical_plan::to_proto::serialize_physical_expr_with_converter;
142137
use datafusion_proto::physical_plan::{
143138
AsExecutionPlan, DeduplicatingProtoConverter, DefaultPhysicalExtensionCodec,
144139
DefaultPhysicalProtoConverter, PhysicalExtensionCodec, PhysicalPlanDecodeContext,
@@ -4785,62 +4780,6 @@ fn roundtrip_parquet_exec_output_partitioning() -> Result<()> {
47854780
Ok(())
47864781
}
47874782

4788-
#[test]
4789-
fn parse_legacy_partitioned_by_file_group_as_output_partitioning() -> Result<()> {
4790-
let file_schema =
4791-
Arc::new(Schema::new(vec![Field::new("col", DataType::Utf8, false)]));
4792-
let table_schema = TableSchema::builder(Arc::clone(&file_schema))
4793-
.with_table_partition_cols(vec![Arc::new(Field::new(
4794-
"part",
4795-
DataType::Utf8,
4796-
false,
4797-
))])
4798-
.build();
4799-
let file_source = Arc::new(ParquetSource::new(table_schema));
4800-
let scan_config =
4801-
FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), file_source)
4802-
.with_file_groups(vec![
4803-
FileGroup::new(vec![PartitionedFile::new(
4804-
"/path/to/file1.parquet".to_string(),
4805-
1024,
4806-
)]),
4807-
FileGroup::new(vec![PartitionedFile::new(
4808-
"/path/to/file2.parquet".to_string(),
4809-
1024,
4810-
)]),
4811-
])
4812-
.build();
4813-
4814-
let codec = DefaultPhysicalExtensionCodec {};
4815-
let proto_converter = DefaultPhysicalProtoConverter {};
4816-
let mut proto = serialize_file_scan_config(&scan_config, &codec, &proto_converter)?;
4817-
proto.partitioned_by_file_group = Some(true);
4818-
proto.output_partitioning = None;
4819-
4820-
let ctx = SessionContext::new();
4821-
let task_ctx = ctx.task_ctx();
4822-
let decode_ctx = PhysicalPlanDecodeContext::new(task_ctx.as_ref(), &codec);
4823-
let parsed = parse_protobuf_file_scan_config(
4824-
&proto,
4825-
&decode_ctx,
4826-
&proto_converter,
4827-
Arc::new(ParquetSource::new(parse_table_schema_from_proto(&proto)?)),
4828-
)?;
4829-
4830-
match parsed.output_partitioning {
4831-
Some(Partitioning::Hash(exprs, partition_count)) => {
4832-
assert_eq!(partition_count, 2);
4833-
assert_eq!(exprs.len(), 1);
4834-
let column = exprs[0].downcast_ref::<Column>().unwrap();
4835-
assert_eq!(column.name(), "part");
4836-
assert_eq!(column.index(), 1);
4837-
}
4838-
other => panic!("Expected legacy hash output partitioning, got {other:?}"),
4839-
}
4840-
4841-
Ok(())
4842-
}
4843-
48444783
#[test]
48454784
fn roundtrip_parquet_exec_range_output_partitioning() -> Result<()> {
48464785
let file_schema =

‎docs/source/library-user-guide/upgrading/55.0.0.md‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,12 +73,17 @@ let df = df.fill_null(&ScalarValue::from(0), &[])?;
7373
`FileScanConfigBuilder::with_partitioned_by_file_group(...)` have been removed.
7474
Use `FileScanConfig::output_partitioning` and
7575
`FileScanConfigBuilder::with_output_partitioning(...)` instead.
76+
The corresponding
77+
`datafusion_proto::protobuf::FileScanExecConf::partitioned_by_file_group`
78+
field has also been removed.
7679

7780
**Who is affected:**
7881

7982
- Users who accessed `FileScanConfig::partitioned_by_file_group` directly.
8083
- Users who called
8184
`FileScanConfigBuilder::with_partitioned_by_file_group(true)`.
85+
- Users who constructed or accessed
86+
`datafusion_proto::protobuf::FileScanExecConf::partitioned_by_file_group`.
8287

8388
**Migration guide:**
8489

@@ -108,6 +113,9 @@ otherwise. If you construct the partitioning manually, pass
108113
`Some(Partitioning::Hash(partition_exprs, partition_count))` to
109114
`with_output_partitioning(...)`.
110115

116+
When constructing `FileScanExecConf`, omit `partitioned_by_file_group` and set
117+
`output_partitioning` instead.
118+
111119
### User `SpillFile` traits instead of [`RefCountedTempFile`]
112120

113121
Spill file APIs now use the `datafusion_execution::SpillFile` trait instead of

0 commit comments

Comments
 (0)