refactor(proto): remove legacy scan field

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;
 }
 

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__,
                 })
             }

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

@@ -1814,8 +1814,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>,
 }

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 {

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,
     })
 }

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::<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 =

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