You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/07/26 18:16:27 UTC
(datafusion) 01/01: refactor(proto): remove legacy scan field (#23445)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-23445-551c592ca6b6e50ff41b62ca8bff282a28be01e4
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit bb670fbabec111cf74e3ae3ee78d0abec65d7569
Author: Phoenix <Ph...@163.com>
AuthorDate: Mon Jul 27 02:16:04 2026 +0800
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:
https://github.com/apache/datafusion/pull/23189#discussion_r3554855165
## 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 <Ph...@163.com>
Co-authored-by: Andrew Lamb <an...@nerdnetworks.org>
---
datafusion/proto-models/proto/datafusion.proto | 4 +-
datafusion/proto-models/src/generated/pbjson.rs | 18 -------
datafusion/proto-models/src/generated/prost.rs | 2 -
datafusion/proto/src/physical_plan/from_proto.rs | 18 +------
datafusion/proto/src/physical_plan/to_proto.rs | 3 --
.../proto/tests/cases/roundtrip_physical_plan.rs | 63 +---------------------
docs/source/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 205cf89abe..16b1b1532f 100644
--- a/datafusion/proto-models/proto/datafusion.proto
+++ b/datafusion/proto-models/proto/datafusion.proto
@@ -1231,7 +1231,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 d23f8eee5f..c5d7c00301 100644
--- a/datafusion/proto-models/src/generated/pbjson.rs
+++ b/datafusion/proto-models/src/generated/pbjson.rs
@@ -6999,9 +6999,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;
}
@@ -7041,9 +7038,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)?;
}
@@ -7074,8 +7068,6 @@ impl<'de> serde::Deserialize<'de> for FileScanExecConf {
"batchSize",
"projection_exprs",
"projectionExprs",
- "partitioned_by_file_group",
- "partitionedByFileGroup",
"output_partitioning",
"outputPartitioning",
];
@@ -7093,7 +7085,6 @@ impl<'de> serde::Deserialize<'de> for FileScanExecConf {
Constraints,
BatchSize,
ProjectionExprs,
- PartitionedByFileGroup,
OutputPartitioning,
}
impl<'de> serde::Deserialize<'de> for GeneratedField {
@@ -7127,7 +7118,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)),
}
@@ -7159,7 +7149,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 {
@@ -7234,12 +7223,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"));
@@ -7260,7 +7243,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 6baabbf37a..2300f7192f 100644
--- a/datafusion/proto-models/src/generated/prost.rs
+++ b/datafusion/proto-models/src/generated/prost.rs
@@ -1875,8 +1875,6 @@ pub struct FileScanExecConf {
pub batch_size: ::core::option::Option<u64>,
#[prost(message, optional, tag = "13")]
pub projection_exprs: ::core::option::Option<ProjectionExprs>,
- #[prost(bool, optional, tag = "14")]
- pub partitioned_by_file_group: ::core::option::Option<bool>,
#[prost(message, optional, tag = "15")]
pub output_partitioning: ::core::option::Option<Partitioning>,
}
diff --git a/datafusion/proto/src/physical_plan/from_proto.rs b/datafusion/proto/src/physical_plan/from_proto.rs
index f1b324c79d..d5a1e0efac 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;
@@ -582,20 +580,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 515a53c087..e13923dbb9 100644
--- a/datafusion/proto/src/physical_plan/to_proto.rs
+++ b/datafusion/proto/src/physical_plan/to_proto.rs
@@ -563,9 +563,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 3d13ffe16e..864e6d6867 100644
--- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
@@ -133,12 +133,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,
@@ -4785,62 +4780,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::<Column>().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 6ca4d71a78..df268792d8 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
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org