refactor: remove redundant partitioned_by_file_group file scan field

datafusion/catalog-listing/src/table.rs

@@ -30,7 +30,9 @@ use datafusion_common::{
 };
 use datafusion_datasource::file::FileSource;
 use datafusion_datasource::file_groups::FileGroup;
-use datafusion_datasource::file_scan_config::{FileScanConfig, FileScanConfigBuilder};
+use datafusion_datasource::file_scan_config::{
+    FileScanConfig, FileScanConfigBuilder, output_partitioning_from_partition_fields,
+};
 use datafusion_datasource::file_sink_config::{FileOutputMode, FileSinkConfig};
 #[expect(deprecated)]
 use datafusion_datasource::schema_adapter::SchemaAdapterFactory;
@@ -62,7 +64,7 @@ pub struct ListFilesResult {
     pub file_groups: Vec<FileGroup>,
     /// Aggregated statistics for all files.
     pub statistics: Statistics,
-    /// Whether files are grouped by partition values (enables Hash partitioning).
+    /// Whether files are grouped by partition values.
     pub grouped_by_partition: bool,
 }
 
@@ -623,6 +625,15 @@ impl TableProvider for ListingTable {
                 );
             }
             Some(output_partitioning)
+        } else if partitioned_by_file_group {
+            // Files are grouped by partition column values: declare output
+            // partitioning on those columns so the optimizer can skip
+            // repartitioning for aggregates and joins on the partition columns.
+            output_partitioning_from_partition_fields(
+                &self.table_schema,
+                &table_partition_cols.clone().into(),
+                partitioned_file_lists.len(),
+            )
         } else {
             None
         };
@@ -645,7 +656,6 @@ impl TableProvider for ListingTable {
             .with_output_ordering(output_ordering)
             .with_output_partitioning(output_partitioning)
             .with_expr_adapter(self.expr_adapter_factory.clone())
-            .with_partitioned_by_file_group(partitioned_by_file_group)
             .build();
 
         // create the execution plan
@@ -856,8 +866,8 @@ impl ListingTable {
         // Threshold: 0 = disabled, N > 0 = enabled when distinct_keys >= N
         //
         // When enabled, files are grouped by their Hive partition column values, allowing
-        // FileScanConfig to declare Hash partitioning. This enables the optimizer to skip
-        // hash repartitioning for aggregates and joins on partition columns.
+        // FileScanConfig to declare output partitioning. This enables the optimizer to
+        // skip repartitioning for aggregates and joins on partition columns.
         let threshold = ctx.config_options().optimizer.preserve_file_partitions;
 
         let (file_groups, grouped_by_partition) =

datafusion/common/src/config.rs

@@ -1480,7 +1480,7 @@ config_namespace! {
         pub repartition_file_scans: bool, default = true
 
         /// Minimum number of distinct partition values required to group files by their
-        /// Hive partition column values (enabling Hash partitioning declaration).
+        /// Hive partition column values (enabling output partitioning declaration).
         ///
         /// How the option is used:
         ///     - preserve_file_partitions=0: Disable it.

datafusion/datasource/src/file_scan_config/mod.rs

@@ -204,17 +204,6 @@ pub struct FileScanConfig {
     /// would be incorrect if there are filters being applied, thus this should be accessed
     /// via [`FileScanConfig::statistics`].
     pub(crate) statistics: Statistics,
-    /// When true, file_groups are organized by partition column values
-    /// and output_partitioning will return Hash partitioning on partition columns.
-    /// This allows the optimizer to skip hash repartitioning for aggregates and joins
-    /// on partition columns.
-    ///
-    /// If the number of file partitions > target_partitions, the file partitions will be grouped
-    /// in a round-robin fashion such that number of file partitions = target_partitions.
-    ///
-    /// Follow-up: remove this redundant field in favor of
-    /// `output_partitioning`, see <https://github.com/apache/datafusion/issues/23099>.
-    pub partitioned_by_file_group: bool,
     /// Declared physical output partitioning for this scan.
     ///
     /// Expressions are against the full table schema, before scan projection or
@@ -294,7 +283,6 @@ pub struct FileScanConfigBuilder {
     file_compression_type: Option<FileCompressionType>,
     batch_size: Option<usize>,
     expr_adapter_factory: Option<Arc<dyn PhysicalExprAdapterFactory>>,
-    partitioned_by_file_group: bool,
 }
 
 impl FileScanConfigBuilder {
@@ -321,7 +309,6 @@ impl FileScanConfigBuilder {
             constraints: None,
             batch_size: None,
             expr_adapter_factory: None,
-            partitioned_by_file_group: false,
         }
     }
 
@@ -519,18 +506,6 @@ impl FileScanConfigBuilder {
         self
     }
 
-    /// Set whether file groups are organized by partition column values.
-    ///
-    /// When set to true, the output partitioning will be declared as Hash partitioning
-    /// on the partition columns.
-    pub fn with_partitioned_by_file_group(
-        mut self,
-        partitioned_by_file_group: bool,
-    ) -> Self {
-        self.partitioned_by_file_group = partitioned_by_file_group;
-        self
-    }
-
     /// Build the final [`FileScanConfig`] with all the configured settings.
     ///
     /// This method takes ownership of the builder and returns the constructed `FileScanConfig`.
@@ -552,7 +527,6 @@ impl FileScanConfigBuilder {
             file_compression_type,
             batch_size,
             expr_adapter_factory: expr_adapter,
-            partitioned_by_file_group,
         } = self;
 
         let constraints = constraints.unwrap_or_default();
@@ -577,7 +551,6 @@ impl FileScanConfigBuilder {
             batch_size,
             expr_adapter_factory: expr_adapter,
             statistics,
-            partitioned_by_file_group,
             output_partitioning,
         }
     }
@@ -598,12 +571,15 @@ impl From<FileScanConfig> for FileScanConfigBuilder {
             constraints: Some(config.constraints),
             batch_size: config.batch_size,
             expr_adapter_factory: config.expr_adapter_factory,
-            partitioned_by_file_group: config.partitioned_by_file_group,
         }
     }
 }
 
-fn hash_partitioning_from_partition_fields(
+/// Builds output partitioning over `partition_cols` (resolved to their indices in
+/// `schema`) with `partition_count` partitions. Returns `None` when there are no
+/// partition columns. Callers use this to declare the output partitioning of a scan
+/// whose file groups are organized by partition column values.
+pub fn output_partitioning_from_partition_fields(
     schema: &Schema,
     partition_cols: &Fields,
     partition_count: usize,
@@ -765,7 +741,7 @@ impl DataSource for FileScanConfig {
     ) -> Result<Option<Arc<dyn DataSource>>> {
         // When file groups define output partitioning, repartitioning files
         // would invalidate the partition-to-file-group mapping.
-        if self.output_partitioning.is_some() || self.partitioned_by_file_group {
+        if self.output_partitioning.is_some() {
             return Ok(None);
         }
 
@@ -782,10 +758,8 @@ impl DataSource for FileScanConfig {
     /// Returns the output partitioning for this file scan.
     ///
     /// When `output_partitioning` is set, this returns the declared partitioning
-    /// after applying scan projection. When `partitioned_by_file_group` is true,
-    /// this returns `Partitioning::Hash` on the Hive partition columns, allowing
-    /// the optimizer to skip hash repartitioning for aggregates and joins on
-    /// those columns.
+    /// after applying scan projection, allowing the optimizer to skip hash
+    /// repartitioning for aggregates and joins on the partitioning columns.
     ///
     /// If projection or partition count validation fails, this returns
     /// `UnknownPartitioning`.
@@ -801,15 +775,7 @@ impl DataSource for FileScanConfig {
     /// - Idea: Could allow byte-range splitting within partition-aware groups,
     ///   preserving I/O parallelism while maintaining partition semantics.
     fn output_partitioning(&self) -> Partitioning {
-        let Some(output_partitioning) = self.output_partitioning.clone().or_else(|| {
-            self.partitioned_by_file_group.then(|| {
-                hash_partitioning_from_partition_fields(
-                    self.file_source.table_schema().table_schema(),
-                    self.table_partition_cols(),
-                    self.file_groups.len(),
-                )
-            })?
-        }) else {
+        let Some(output_partitioning) = self.output_partitioning.clone() else {
             return Partitioning::UnknownPartitioning(self.file_groups.len());
         };
         if output_partitioning.partition_count() != self.file_groups.len() {
@@ -1140,7 +1106,6 @@ impl DataSource for FileScanConfig {
     ) -> Option<Arc<dyn Any + Send + Sync>> {
         if self.preserve_order
             || self.output_partitioning.is_some()
-            || self.partitioned_by_file_group
             || !config.execution.enable_file_stream_work_stealing
         {
             return None;
@@ -2553,7 +2518,7 @@ mod tests {
             vec![partition_col],
         );
 
-        // partitioned_by_file_group defaults to false
+        // output_partitioning defaults to None
         let partitioning = config.output_partitioning();
         assert!(matches!(partitioning, Partitioning::UnknownPartitioning(_)));
     }
@@ -2613,13 +2578,12 @@ mod tests {
     #[test]
     fn test_output_partitioning_no_partition_columns() {
         let file_schema = aggr_test_schema();
-        let mut config = config_for_projection(
+        let config = config_for_projection(
             Arc::clone(&file_schema),
             None,
             Statistics::new_unknown(&file_schema),
             vec![], // No partition columns
         );
-        config.partitioned_by_file_group = true;
 
         let partitioning = config.output_partitioning();
         assert!(matches!(partitioning, Partitioning::UnknownPartitioning(_)));
@@ -2642,12 +2606,16 @@ mod tests {
             Statistics::new_unknown(&file_schema),
             single_partition_col,
         );
-        config.partitioned_by_file_group = true;
         config.file_groups = vec![
             FileGroup::new(vec![PartitionedFile::new("f1.parquet".to_string(), 1024)]),
             FileGroup::new(vec![PartitionedFile::new("f2.parquet".to_string(), 1024)]),
             FileGroup::new(vec![PartitionedFile::new("f3.parquet".to_string(), 1024)]),
         ];
+        config.output_partitioning = output_partitioning_from_partition_fields(
+            config.file_source.table_schema().table_schema(),
+            config.table_partition_cols(),
+            config.file_groups.len(),
+        );
 
         let partitioning = config.output_partitioning();
         match partitioning {
@@ -2671,11 +2639,15 @@ mod tests {
             Statistics::new_unknown(&file_schema),
             multiple_partition_cols,
         );
-        config.partitioned_by_file_group = true;
         config.file_groups = vec![
             FileGroup::new(vec![PartitionedFile::new("f1.parquet".to_string(), 1024)]),
             FileGroup::new(vec![PartitionedFile::new("f2.parquet".to_string(), 1024)]),
         ];
+        config.output_partitioning = output_partitioning_from_partition_fields(
+            config.file_source.table_schema().table_schema(),
+            config.table_partition_cols(),
+            config.file_groups.len(),
+        );
 
         let partitioning = config.output_partitioning();
         match partitioning {

datafusion/datasource/src/file_stream/mod.rs

@@ -1107,13 +1107,13 @@ mod tests {
         Ok(())
     }
 
-    /// Verifies that `partitioned_by_file_group` disables shared work stealing.
+    /// Verifies that declared output partitioning disables shared work stealing.
     #[tokio::test]
-    async fn morsel_partitioned_by_file_group_keeps_files_local() -> Result<()> {
+    async fn morsel_declared_output_partitioning_keeps_files_local() -> Result<()> {
         // same fixture as `morsel_shared_files_can_be_stolen` but marked as
         // preserve-partitioned
         let test = two_partition_morsel_test()
-            .with_partitioned_by_file_group(true)
+            .with_declared_output_partitioning(true)
             .with_file_stream_events(false);
 
         insta::assert_snapshot!(test.run().await.unwrap(), @r"
@@ -1366,7 +1366,7 @@ mod tests {
         morselizer: MockMorselizer,
         partition_files: BTreeMap<PartitionId, Vec<String>>,
         preserve_order: bool,
-        partitioned_by_file_group: bool,
+        declared_output_partitioning: bool,
         enable_file_stream_work_stealing: bool,
         file_stream_events: bool,
         build_streams_on_first_read: bool,
@@ -1381,7 +1381,7 @@ mod tests {
                 morselizer: MockMorselizer::new(),
                 partition_files: BTreeMap::new(),
                 preserve_order: false,
-                partitioned_by_file_group: false,
+                declared_output_partitioning: false,
                 enable_file_stream_work_stealing: true,
                 file_stream_events: true,
                 build_streams_on_first_read: false,
@@ -1418,13 +1418,13 @@ mod tests {
             self
         }
 
-        /// Marks the test scan as pre-partitioned by file group, which should
-        /// force each stream to keep its own files local.
-        fn with_partitioned_by_file_group(
+        /// Declares the test scan's output partitioning, which should force
+        /// each stream to keep its own files local.
+        fn with_declared_output_partitioning(
             mut self,
-            partitioned_by_file_group: bool,
+            declared_output_partitioning: bool,
         ) -> Self {
-            self.partitioned_by_file_group = partitioned_by_file_group;
+            self.declared_output_partitioning = declared_output_partitioning;
             self
         }
 
@@ -1630,14 +1630,24 @@ mod tests {
                     DataType::Int32,
                     false,
                 )])));
+            // Declaring an output partitioning marks the scan as pre-grouped, which
+            // keeps each stream's files local (disables shared work stealing).
+            let output_partitioning = self.declared_output_partitioning.then(|| {
+                datafusion_physical_expr::Partitioning::Hash(
+                    vec![Arc::new(
+                        datafusion_physical_expr::expressions::Column::new("i", 0),
+                    )],
+                    file_groups.len(),
+                )
+            });
             FileScanConfigBuilder::new(
                 ObjectStoreUrl::parse("test:///").unwrap(),
                 Arc::new(MockSource::new(table_schema)),
             )
             .with_file_groups(file_groups)
             .with_limit(self.limit)
             .with_preserve_order(self.preserve_order)
-            .with_partitioned_by_file_group(self.partitioned_by_file_group)
+            .with_output_partitioning(output_partitioning)
             .build()
         }
     }

datafusion/proto/src/physical_plan/from_proto.rs

@@ -29,7 +29,9 @@ use datafusion_common::{
 };
 use datafusion_datasource::file::FileSource;
 use datafusion_datasource::file_groups::FileGroup;
-use datafusion_datasource::file_scan_config::{FileScanConfig, FileScanConfigBuilder};
+use datafusion_datasource::file_scan_config::{
+    FileScanConfig, FileScanConfigBuilder, output_partitioning_from_partition_fields,
+};
 use datafusion_datasource::file_sink_config::FileSinkConfig;
 use datafusion_datasource::{FileRange, ListingTableUrl, PartitionedFile, TableSchema};
 use datafusion_datasource_csv::file_format::CsvSink;
@@ -558,6 +560,20 @@ 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 {
@@ -586,18 +602,15 @@ pub fn parse_protobuf_file_scan_config(
         file_source
     };
 
-    let mut config_builder = FileScanConfigBuilder::new(object_store_url, file_source)
+    let config = FileScanConfigBuilder::new(object_store_url, file_source)
         .with_file_groups(file_groups)
         .with_constraints(constraints)
         .with_statistics(statistics)
         .with_limit(proto.limit.as_ref().map(|sl| sl.limit as usize))
         .with_output_ordering(output_ordering)
         .with_output_partitioning(output_partitioning)
-        .with_batch_size(proto.batch_size.map(|s| s as usize));
-    if proto.partitioned_by_file_group.unwrap_or(false) {
-        config_builder = config_builder.with_partitioned_by_file_group(true);
-    }
-    let config = config_builder.build();
+        .with_batch_size(proto.batch_size.map(|s| s as usize))
+        .build();
     Ok(config)
 }