From 51b3848c4faa3c8a20ddd385579e066b1e18dff5 Mon Sep 17 00:00:00 2001 From: Jiawei Zhao Date: Thu, 25 Jun 2026 23:16:01 +0800 Subject: [PATCH 1/5] refactor: remove redundant partitioned_by_file_group file scan field `FileScanConfig` had two overlapping ways to declare file scan output partitioning: the `partitioned_by_file_group` bool and `output_partitioning`. Collapse them onto `output_partitioning` as the single source of truth. - Remove the `partitioned_by_file_group` field, the builder field, and the `with_partitioned_by_file_group` builder method. - `ListingTable` now derives the partition-column `Partitioning::Hash` once its file groups are finalized and passes it via `with_output_partitioning`; `hash_partitioning_from_partition_fields` is made `pub` for this. - proto already round-trips `output_partitioning`, so the now-vestigial wire bool is left unset on write and ignored on read (the proto field is kept for backward compatibility). Closes #23099. Signed-off-by: Jiawei Zhao --- datafusion/catalog-listing/src/table.rs | 14 +++- .../datasource/src/file_scan_config/mod.rs | 70 ++++++------------- datafusion/datasource/src/file_stream/mod.rs | 12 +++- .../proto/src/physical_plan/from_proto.rs | 9 +-- .../proto/src/physical_plan/to_proto.rs | 4 +- .../tests/cases/roundtrip_physical_plan.rs | 18 ----- 6 files changed, 50 insertions(+), 77 deletions(-) diff --git a/datafusion/catalog-listing/src/table.rs b/datafusion/catalog-listing/src/table.rs index 632b829b161a0..34a90ca606e0e 100644 --- a/datafusion/catalog-listing/src/table.rs +++ b/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, hash_partitioning_from_partition_fields, +}; use datafusion_datasource::file_sink_config::{FileOutputMode, FileSinkConfig}; #[expect(deprecated)] use datafusion_datasource::schema_adapter::SchemaAdapterFactory; @@ -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 Hash + // partitioning on those columns so the optimizer can skip hash + // repartitioning for aggregates and joins on the partition columns. + hash_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 diff --git a/datafusion/datasource/src/file_scan_config/mod.rs b/datafusion/datasource/src/file_scan_config/mod.rs index b73d100e056f4..63e1a0accb92d 100644 --- a/datafusion/datasource/src/file_scan_config/mod.rs +++ b/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 . - 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, batch_size: Option, expr_adapter_factory: Option>, - 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 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 `Partitioning::Hash` 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 hash_partitioning_from_partition_fields( schema: &Schema, partition_cols: &Fields, partition_count: usize, @@ -765,7 +741,7 @@ impl DataSource for FileScanConfig { ) -> Result>> { // 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> { 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 = hash_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 = hash_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 { diff --git a/datafusion/datasource/src/file_stream/mod.rs b/datafusion/datasource/src/file_stream/mod.rs index e0641310c228f..8952e27662625 100644 --- a/datafusion/datasource/src/file_stream/mod.rs +++ b/datafusion/datasource/src/file_stream/mod.rs @@ -1630,6 +1630,16 @@ 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.partitioned_by_file_group.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)), @@ -1637,7 +1647,7 @@ mod tests { .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() } } diff --git a/datafusion/proto/src/physical_plan/from_proto.rs b/datafusion/proto/src/physical_plan/from_proto.rs index 53ff4a41d466e..8f894bfa404ea 100644 --- a/datafusion/proto/src/physical_plan/from_proto.rs +++ b/datafusion/proto/src/physical_plan/from_proto.rs @@ -586,18 +586,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) } diff --git a/datafusion/proto/src/physical_plan/to_proto.rs b/datafusion/proto/src/physical_plan/to_proto.rs index 4614c4f002169..e93bb9cc5fed2 100644 --- a/datafusion/proto/src/physical_plan/to_proto.rs +++ b/datafusion/proto/src/physical_plan/to_proto.rs @@ -562,7 +562,9 @@ pub fn serialize_file_scan_config( constraints: Some(conf.constraints.clone().into()), batch_size: conf.batch_size.map(|s| s as u64), projection_exprs, - partitioned_by_file_group: Some(conf.partitioned_by_file_group), + // 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 8acb891d6a94d..7e8d87fd8c6bb 100644 --- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs +++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs @@ -4219,24 +4219,6 @@ fn roundtrip_file_scan_config(scan_config: FileScanConfig) -> Result Result<()> { - let file_schema = - Arc::new(Schema::new(vec![Field::new("col", DataType::Utf8, false)])); - let file_source = Arc::new(ParquetSource::new(Arc::clone(&file_schema))); - let scan_config = - FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), file_source) - .with_file_groups(vec![FileGroup::new(vec![PartitionedFile::new( - "/path/to/file.parquet".to_string(), - 1024, - )])]) - .with_partitioned_by_file_group(true) - .build(); - - assert!(roundtrip_file_scan_config(scan_config)?.partitioned_by_file_group); - Ok(()) -} - #[test] fn roundtrip_parquet_exec_output_partitioning() -> Result<()> { let file_schema = From a9aafc0ae4cb6222c23c3f12ca16ad1d99feb3aa Mon Sep 17 00:00:00 2001 From: Jiawei Zhao Date: Fri, 26 Jun 2026 10:13:29 +0800 Subject: [PATCH 2/5] test: show output_partitioning in EXPLAIN for partition-grouped scans After collapsing `partitioned_by_file_group` onto `output_partitioning`, the declared Hash partitioning is now stored on the scan and therefore rendered by `DataSourceExec`'s Display. Update the affected sqllogictest expected plans accordingly. Behavior is unchanged; only the EXPLAIN text gains an `output_partitioning=Hash(...)` entry on partition-grouped scans. Signed-off-by: Jiawei Zhao --- .../test_files/preserve_file_partitioning.slt | 16 ++++++++-------- .../repartition_subset_satisfaction.slt | 16 ++++++++-------- 2 files changed, 16 insertions(+), 16 deletions(-) diff --git a/datafusion/sqllogictest/test_files/preserve_file_partitioning.slt b/datafusion/sqllogictest/test_files/preserve_file_partitioning.slt index 412d606df903f..e2dd22cc82bba 100644 --- a/datafusion/sqllogictest/test_files/preserve_file_partitioning.slt +++ b/datafusion/sqllogictest/test_files/preserve_file_partitioning.slt @@ -258,7 +258,7 @@ logical_plan physical_plan 01)ProjectionExec: expr=[f_dkey@0 as f_dkey, count(Int64(1))@1 as count(*), sum(fact_table.value)@2 as sum(fact_table.value)] 02)--AggregateExec: mode=SinglePartitioned, gby=[f_dkey@1 as f_dkey], aggr=[count(Int64(1)), sum(fact_table.value)] -03)----DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[value, f_dkey], file_type=parquet +03)----DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[value, f_dkey], output_partitioning=Hash([f_dkey@1], 3), file_type=parquet # Verify results with optimization match results without optimization query TIR rowsort @@ -320,7 +320,7 @@ physical_plan 01)SortPreservingMergeExec: [f_dkey@0 ASC NULLS LAST] 02)--ProjectionExec: expr=[f_dkey@0 as f_dkey, count(Int64(1))@1 as count(*), avg(fact_table_ordered.value)@2 as avg(fact_table_ordered.value)] 03)----AggregateExec: mode=SinglePartitioned, gby=[f_dkey@1 as f_dkey], aggr=[count(Int64(1)), avg(fact_table_ordered.value)], ordering_mode=Sorted -04)------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[value, f_dkey], output_ordering=[f_dkey@1 ASC NULLS LAST], file_type=parquet +04)------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[value, f_dkey], output_ordering=[f_dkey@1 ASC NULLS LAST], output_partitioning=Hash([f_dkey@1], 3), file_type=parquet query TIR SELECT f_dkey, count(*), avg(value) FROM fact_table_ordered GROUP BY f_dkey ORDER BY f_dkey; @@ -418,7 +418,7 @@ physical_plan 06)----------FilterExec: service@2 = log 07)------------RepartitionExec: partitioning=RoundRobinBatch(3), input_partitions=1 08)--------------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/dimension/data.parquet]]}, projection=[d_dkey, env, service], file_type=parquet, predicate=service@2 = log, pruning_predicate=service_null_count@2 != row_count@3 AND service_min@0 <= log AND log <= service_max@1, required_guarantees=[service in (log)] -09)--------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[value, f_dkey], output_ordering=[f_dkey@1 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible +09)--------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[value, f_dkey], output_ordering=[f_dkey@1 ASC NULLS LAST], output_partitioning=Hash([f_dkey@1], 3), file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible query TTTIR rowsort SELECT f.f_dkey, MAX(d.env), MAX(d.service), count(*), sum(f.value) @@ -493,7 +493,7 @@ logical_plan physical_plan 01)ProjectionExec: expr=[f_dkey@2 as f_dkey, timestamp@0 as timestamp, value@1 as value, row_number() PARTITION BY [fact_table_ordered.f_dkey] ORDER BY [fact_table_ordered.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@3 as rn] 02)--BoundedWindowAggExec: wdw=[row_number() PARTITION BY [fact_table_ordered.f_dkey] ORDER BY [fact_table_ordered.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "row_number() PARTITION BY [fact_table_ordered.f_dkey] ORDER BY [fact_table_ordered.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": UInt64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] -03)----DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], file_type=parquet +03)----DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], output_partitioning=Hash([f_dkey@2], 3), file_type=parquet query TPRI rowsort SELECT f_dkey, timestamp, value, @@ -548,7 +548,7 @@ logical_plan physical_plan 01)ProjectionExec: expr=[f_dkey@0 as f_dkey, count(Int64(1))@1 as count(*), sum(high_cardinality_table.value)@2 as sum(high_cardinality_table.value)] 02)--AggregateExec: mode=SinglePartitioned, gby=[f_dkey@1 as f_dkey], aggr=[count(Int64(1)), sum(high_cardinality_table.value)] -03)----DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/high_cardinality/f_dkey=A/data.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/high_cardinality/f_dkey=D/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/high_cardinality/f_dkey=B/data.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/high_cardinality/f_dkey=E/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/high_cardinality/f_dkey=C/data.parquet]]}, projection=[value, f_dkey], file_type=parquet +03)----DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/high_cardinality/f_dkey=A/data.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/high_cardinality/f_dkey=D/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/high_cardinality/f_dkey=B/data.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/high_cardinality/f_dkey=E/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/high_cardinality/f_dkey=C/data.parquet]]}, projection=[value, f_dkey], output_partitioning=Hash([f_dkey@1], 3), file_type=parquet # Verify results with optimization match results without optimization query TIR rowsort @@ -685,8 +685,8 @@ physical_plan 02)--RepartitionExec: partitioning=Hash([f_dkey@0, env@1], 3), input_partitions=3 03)----AggregateExec: mode=Partial, gby=[f_dkey@1 as f_dkey, env@2 as env], aggr=[sum(f.value)] 04)------HashJoinExec: mode=Partitioned, join_type=Inner, on=[(d_dkey@1, f_dkey@1)], projection=[value@2, f_dkey@3, env@0] -05)--------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/dimension_partitioned/d_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/dimension_partitioned/d_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/dimension_partitioned/d_dkey=C/data.parquet]]}, projection=[env, d_dkey], file_type=parquet -06)--------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[value, f_dkey], file_type=parquet +05)--------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/dimension_partitioned/d_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/dimension_partitioned/d_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/dimension_partitioned/d_dkey=C/data.parquet]]}, projection=[env, d_dkey], output_partitioning=Hash([d_dkey@1], 3), file_type=parquet +06)--------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[value, f_dkey], output_partitioning=Hash([f_dkey@1], 3), file_type=parquet query TTR rowsort SELECT f.f_dkey, d.env, sum(f.value) @@ -722,7 +722,7 @@ logical_plan physical_plan 01)ProjectionExec: expr=[f_dkey@0 as f_dkey, timestamp@1 as timestamp, count(Int64(1))@2 as count(*), avg(fact_table.value)@3 as avg(fact_table.value)] 02)--AggregateExec: mode=SinglePartitioned, gby=[f_dkey@2 as f_dkey, timestamp@0 as timestamp], aggr=[count(Int64(1)), avg(fact_table.value)] -03)----DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], file_type=parquet +03)----DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_partitioning=Hash([f_dkey@2], 3), file_type=parquet query TPIR rowsort SELECT f_dkey, timestamp, diff --git a/datafusion/sqllogictest/test_files/repartition_subset_satisfaction.slt b/datafusion/sqllogictest/test_files/repartition_subset_satisfaction.slt index 043a62314cb5c..5d45c45fe8535 100644 --- a/datafusion/sqllogictest/test_files/repartition_subset_satisfaction.slt +++ b/datafusion/sqllogictest/test_files/repartition_subset_satisfaction.slt @@ -164,7 +164,7 @@ physical_plan 03)----AggregateExec: mode=FinalPartitioned, gby=[f_dkey@0 as f_dkey, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)@1 as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)], aggr=[count(Int64(1)), avg(fact_table_ordered.value)], ordering_mode=Sorted 04)------RepartitionExec: partitioning=Hash([f_dkey@0, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)@1], 3), input_partitions=3, preserve_order=true, sort_exprs=f_dkey@0 ASC NULLS LAST, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)@1 ASC NULLS LAST 05)--------AggregateExec: mode=Partial, gby=[f_dkey@2 as f_dkey, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }, timestamp@0) as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)], aggr=[count(Int64(1)), avg(fact_table_ordered.value)], ordering_mode=Sorted -06)----------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], file_type=parquet +06)----------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], output_partitioning=Hash([f_dkey@2], 3), file_type=parquet # Verify results without subset satisfaction query TPIR rowsort @@ -204,7 +204,7 @@ physical_plan 01)SortPreservingMergeExec: [f_dkey@0 ASC NULLS LAST, time_bin@1 ASC NULLS LAST] 02)--ProjectionExec: expr=[f_dkey@0 as f_dkey, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)@1 as time_bin, count(Int64(1))@2 as count(*), avg(fact_table_ordered.value)@3 as avg(fact_table_ordered.value)] 03)----AggregateExec: mode=SinglePartitioned, gby=[f_dkey@2 as f_dkey, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }, timestamp@0) as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)], aggr=[count(Int64(1)), avg(fact_table_ordered.value)], ordering_mode=Sorted -04)------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], file_type=parquet +04)------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], output_partitioning=Hash([f_dkey@2], 3), file_type=parquet # Verify results match with subset satisfaction query TPIR rowsort @@ -251,7 +251,7 @@ physical_plan 02)--BoundedWindowAggExec: wdw=[row_number() PARTITION BY [fact_table_ordered.f_dkey, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)] ORDER BY [fact_table_ordered.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "row_number() PARTITION BY [fact_table_ordered.f_dkey, date_bin(IntervalMonthDayNano(\"IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }\"),fact_table_ordered.timestamp)] ORDER BY [fact_table_ordered.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": UInt64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] 03)----SortExec: expr=[f_dkey@2 ASC NULLS LAST, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }, timestamp@0) ASC NULLS LAST, timestamp@0 ASC NULLS LAST], preserve_partitioning=[true] 04)------RepartitionExec: partitioning=Hash([f_dkey@2, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }, timestamp@0)], 3), input_partitions=3 -05)--------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], file_type=parquet +05)--------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], output_partitioning=Hash([f_dkey@2], 3), file_type=parquet # Verify results without subset satisfaction query TPRI rowsort @@ -292,7 +292,7 @@ logical_plan physical_plan 01)ProjectionExec: expr=[f_dkey@2 as f_dkey, timestamp@0 as timestamp, value@1 as value, row_number() PARTITION BY [fact_table_ordered.f_dkey, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)] ORDER BY [fact_table_ordered.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@3 as rn] 02)--BoundedWindowAggExec: wdw=[row_number() PARTITION BY [fact_table_ordered.f_dkey, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)] ORDER BY [fact_table_ordered.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "row_number() PARTITION BY [fact_table_ordered.f_dkey, date_bin(IntervalMonthDayNano(\"IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }\"),fact_table_ordered.timestamp)] ORDER BY [fact_table_ordered.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": UInt64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] -03)----DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], file_type=parquet +03)----DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], output_partitioning=Hash([f_dkey@2], 3), file_type=parquet # Verify results match with subset satisfaction query TPRI rowsort @@ -379,8 +379,8 @@ physical_plan 11)--------------------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d_dkey@1, f_dkey@2)], projection=[f_dkey@4, env@0, timestamp@2, value@3] 12)----------------------CoalescePartitionsExec 13)------------------------FilterExec: service@1 = log, projection=[env@0, d_dkey@2] -14)--------------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=A/data.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=D/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=C/data.parquet]]}, projection=[env, service, d_dkey], file_type=parquet, predicate=service@1 = log, pruning_predicate=service_null_count@2 != row_count@3 AND service_min@0 <= log AND log <= service_max@1, required_guarantees=[service in (log)] -15)----------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible +14)--------------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=A/data.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=D/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=C/data.parquet]]}, projection=[env, service, d_dkey], output_partitioning=Hash([d_dkey@2], 3), file_type=parquet, predicate=service@1 = log, pruning_predicate=service_null_count@2 != row_count@3 AND service_min@0 <= log AND log <= service_max@1, required_guarantees=[service in (log)] +15)----------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], output_partitioning=Hash([f_dkey@2], 3), file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible # Verify results without subset satisfaction query TPR rowsort @@ -474,8 +474,8 @@ physical_plan 09)----------------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d_dkey@1, f_dkey@2)], projection=[f_dkey@4, env@0, timestamp@2, value@3] 10)------------------CoalescePartitionsExec 11)--------------------FilterExec: service@1 = log, projection=[env@0, d_dkey@2] -12)----------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=A/data.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=D/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=C/data.parquet]]}, projection=[env, service, d_dkey], file_type=parquet, predicate=service@1 = log, pruning_predicate=service_null_count@2 != row_count@3 AND service_min@0 <= log AND log <= service_max@1, required_guarantees=[service in (log)] -13)------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible +12)----------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=A/data.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=D/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=C/data.parquet]]}, projection=[env, service, d_dkey], output_partitioning=Hash([d_dkey@2], 3), file_type=parquet, predicate=service@1 = log, pruning_predicate=service_null_count@2 != row_count@3 AND service_min@0 <= log AND log <= service_max@1, required_guarantees=[service in (log)] +13)------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], output_partitioning=Hash([f_dkey@2], 3), file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible # Verify results match with subset satisfaction query TPR rowsort From 811389476fef01cbee961efc13e570d4c4cae4f3 Mon Sep 17 00:00:00 2001 From: Jiawei Zhao Date: Fri, 3 Jul 2026 11:03:09 +0800 Subject: [PATCH 3/5] fix(proto): read legacy file scan partitioning Honor the old partitioned_by_file_group protobuf flag when output_partitioning is absent, so previously serialized plans still decode to hash output partitioning. Rename test-only helpers to declared output partitioning wording. Signed-off-by: Jiawei Zhao --- datafusion/datasource/src/file_stream/mod.rs | 22 +++---- .../proto/src/physical_plan/from_proto.rs | 18 +++++- .../tests/cases/roundtrip_physical_plan.rs | 63 ++++++++++++++++++- 3 files changed, 90 insertions(+), 13 deletions(-) diff --git a/datafusion/datasource/src/file_stream/mod.rs b/datafusion/datasource/src/file_stream/mod.rs index 8952e27662625..6daed7c338022 100644 --- a/datafusion/datasource/src/file_stream/mod.rs +++ b/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>, 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 } @@ -1632,7 +1632,7 @@ mod tests { )]))); // 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.partitioned_by_file_group.then(|| { + let output_partitioning = self.declared_output_partitioning.then(|| { datafusion_physical_expr::Partitioning::Hash( vec![Arc::new( datafusion_physical_expr::expressions::Column::new("i", 0), diff --git a/datafusion/proto/src/physical_plan/from_proto.rs b/datafusion/proto/src/physical_plan/from_proto.rs index 8f894bfa404ea..b062e47dcb39e 100644 --- a/datafusion/proto/src/physical_plan/from_proto.rs +++ b/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, hash_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 hash output partitioning. + let table_schema = parse_table_schema_from_proto(proto)?; + hash_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/tests/cases/roundtrip_physical_plan.rs b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs index 7e8d87fd8c6bb..bb99a1ecbf4ed 100644 --- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs +++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs @@ -124,7 +124,12 @@ use datafusion_proto::bytes::{ physical_plan_from_bytes_with_proto_converter, physical_plan_to_bytes_with_proto_converter, }; -use datafusion_proto::physical_plan::to_proto::serialize_physical_expr_with_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::{ AsExecutionPlan, DeduplicatingProtoConverter, DefaultPhysicalExtensionCodec, DefaultPhysicalProtoConverter, PhysicalExtensionCodec, PhysicalPlanDecodeContext, @@ -4243,6 +4248,62 @@ 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 = From f6b19e19b401bb8caf6281209aaa59eab698edaa Mon Sep 17 00:00:00 2001 From: Jiawei Zhao Date: Sat, 4 Jul 2026 22:34:04 +0800 Subject: [PATCH 4/5] docs: document file scan partitioning migration Add an upgrade guide note for the removal of partitioned_by_file_group and its builder API. Point users to output_partitioning and with_output_partitioning. Signed-off-by: Jiawei Zhao --- .../library-user-guide/upgrading/55.0.0.md | 42 +++++++++++++++++++ 1 file changed, 42 insertions(+) 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 f39b34a6402a0..e06cc23a32619 100644 --- a/docs/source/library-user-guide/upgrading/55.0.0.md +++ b/docs/source/library-user-guide/upgrading/55.0.0.md @@ -67,6 +67,48 @@ let df = df.fill_null(&ScalarValue::from(0), &["a", "c"])?; let df = df.fill_null(&ScalarValue::from(0), &[])?; ``` + +### `FileScanConfig::partitioned_by_file_group` removed + +`FileScanConfig::partitioned_by_file_group` and +`FileScanConfigBuilder::with_partitioned_by_file_group(...)` have been removed. +Use `FileScanConfig::output_partitioning` and +`FileScanConfigBuilder::with_output_partitioning(...)` instead. + +**Who is affected:** + +- Users who accessed `FileScanConfig::partitioned_by_file_group` directly. +- Users who called + `FileScanConfigBuilder::with_partitioned_by_file_group(true)`. + +**Migration guide:** + +If your file groups are organized by table partition column values, declare hash +output partitioning over those partition columns: + +```rust,ignore +use datafusion_datasource::file_scan_config::{ + FileScanConfigBuilder, hash_partitioning_from_partition_fields, +}; + +let output_partitioning = hash_partitioning_from_partition_fields( + source.table_schema().table_schema(), + source.table_schema().table_partition_cols(), + file_groups.len(), +); + +let config = FileScanConfigBuilder::new(object_store_url, source) + .with_file_groups(file_groups) + .with_output_partitioning(output_partitioning) + .build(); +``` + +`hash_partitioning_from_partition_fields` returns +`Some(Partitioning::Hash(...))` when partition columns are present and `None` +otherwise. If you construct the partitioning manually, pass +`Some(Partitioning::Hash(partition_exprs, partition_count))` to +`with_output_partitioning(...)`. + ### User `SpillFile` traits instead of [`RefCountedTempFile`] Spill file APIs now use the `datafusion_execution::SpillFile` trait instead of From 9d6db0df6f933a3444512e11cb083d0e64d61aff Mon Sep 17 00:00:00 2001 From: Jiawei Zhao Date: Wed, 8 Jul 2026 20:30:48 +0800 Subject: [PATCH 5/5] refactor: clarify scan partitioning helper Avoid naming the partitioned-file-group helper as hash-specific. This keeps the existing physical representation while making the API and docs less likely to imply literal hash partitioning. Signed-off-by: Jiawei Zhao --- datafusion/catalog-listing/src/table.rs | 14 +++++++------- datafusion/common/src/config.rs | 2 +- datafusion/datasource/src/file_scan_config/mod.rs | 8 ++++---- datafusion/proto/src/physical_plan/from_proto.rs | 6 +++--- .../sqllogictest/test_files/information_schema.slt | 2 +- docs/source/library-user-guide/upgrading/55.0.0.md | 7 +++---- docs/source/user-guide/configs.md | 2 +- 7 files changed, 20 insertions(+), 21 deletions(-) diff --git a/datafusion/catalog-listing/src/table.rs b/datafusion/catalog-listing/src/table.rs index 34a90ca606e0e..23c67efa741e1 100644 --- a/datafusion/catalog-listing/src/table.rs +++ b/datafusion/catalog-listing/src/table.rs @@ -31,7 +31,7 @@ use datafusion_common::{ use datafusion_datasource::file::FileSource; use datafusion_datasource::file_groups::FileGroup; use datafusion_datasource::file_scan_config::{ - FileScanConfig, FileScanConfigBuilder, hash_partitioning_from_partition_fields, + FileScanConfig, FileScanConfigBuilder, output_partitioning_from_partition_fields, }; use datafusion_datasource::file_sink_config::{FileOutputMode, FileSinkConfig}; #[expect(deprecated)] @@ -64,7 +64,7 @@ pub struct ListFilesResult { pub file_groups: Vec, /// 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, } @@ -626,10 +626,10 @@ impl TableProvider for ListingTable { } Some(output_partitioning) } else if partitioned_by_file_group { - // Files are grouped by partition column values: declare Hash - // partitioning on those columns so the optimizer can skip hash + // 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. - hash_partitioning_from_partition_fields( + output_partitioning_from_partition_fields( &self.table_schema, &table_partition_cols.clone().into(), partitioned_file_lists.len(), @@ -866,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) = diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 454af28c14b4a..b649ecad570d2 100644 --- a/datafusion/common/src/config.rs +++ b/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. diff --git a/datafusion/datasource/src/file_scan_config/mod.rs b/datafusion/datasource/src/file_scan_config/mod.rs index 63e1a0accb92d..660d0cd7a5db5 100644 --- a/datafusion/datasource/src/file_scan_config/mod.rs +++ b/datafusion/datasource/src/file_scan_config/mod.rs @@ -575,11 +575,11 @@ impl From for FileScanConfigBuilder { } } -/// Builds `Partitioning::Hash` over `partition_cols` (resolved to their indices in +/// 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 hash_partitioning_from_partition_fields( +pub fn output_partitioning_from_partition_fields( schema: &Schema, partition_cols: &Fields, partition_count: usize, @@ -2611,7 +2611,7 @@ mod tests { FileGroup::new(vec![PartitionedFile::new("f2.parquet".to_string(), 1024)]), FileGroup::new(vec![PartitionedFile::new("f3.parquet".to_string(), 1024)]), ]; - config.output_partitioning = hash_partitioning_from_partition_fields( + config.output_partitioning = output_partitioning_from_partition_fields( config.file_source.table_schema().table_schema(), config.table_partition_cols(), config.file_groups.len(), @@ -2643,7 +2643,7 @@ mod tests { FileGroup::new(vec![PartitionedFile::new("f1.parquet".to_string(), 1024)]), FileGroup::new(vec![PartitionedFile::new("f2.parquet".to_string(), 1024)]), ]; - config.output_partitioning = hash_partitioning_from_partition_fields( + config.output_partitioning = output_partitioning_from_partition_fields( config.file_source.table_schema().table_schema(), config.table_partition_cols(), config.file_groups.len(), diff --git a/datafusion/proto/src/physical_plan/from_proto.rs b/datafusion/proto/src/physical_plan/from_proto.rs index b062e47dcb39e..19e49f3cb8724 100644 --- a/datafusion/proto/src/physical_plan/from_proto.rs +++ b/datafusion/proto/src/physical_plan/from_proto.rs @@ -30,7 +30,7 @@ use datafusion_common::{ use datafusion_datasource::file::FileSource; use datafusion_datasource::file_groups::FileGroup; use datafusion_datasource::file_scan_config::{ - FileScanConfig, FileScanConfigBuilder, hash_partitioning_from_partition_fields, + FileScanConfig, FileScanConfigBuilder, output_partitioning_from_partition_fields, }; use datafusion_datasource::file_sink_config::FileSinkConfig; use datafusion_datasource::{FileRange, ListingTableUrl, PartitionedFile, TableSchema}; @@ -564,9 +564,9 @@ pub fn parse_protobuf_file_scan_config( 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 hash output partitioning. + // `partitioned_by_file_group` to declare scan output partitioning. let table_schema = parse_table_schema_from_proto(proto)?; - hash_partitioning_from_partition_fields( + output_partitioning_from_partition_fields( &schema, table_schema.table_partition_cols(), file_groups.len(), diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index bf45564e26333..1adf98f67ff99 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -489,7 +489,7 @@ datafusion.optimizer.max_passes 3 Number of times that the optimizer will attemp datafusion.optimizer.prefer_existing_sort false When true, DataFusion will opportunistically remove sorts when the data is already sorted, (i.e. setting `preserve_order` to true on `RepartitionExec` and using `SortPreservingMergeExec`) When false, DataFusion will maximize plan parallelism using `RepartitionExec` even if this requires subsequently resorting data using a `SortExec`. datafusion.optimizer.prefer_existing_union false When set to true, the optimizer will not attempt to convert Union to Interleave datafusion.optimizer.prefer_hash_join true When set to true, the physical plan optimizer will prefer HashJoin over SortMergeJoin. HashJoin can work more efficiently than SortMergeJoin but consumes more memory -datafusion.optimizer.preserve_file_partitions 0 Minimum number of distinct partition values required to group files by their Hive partition column values (enabling Hash partitioning declaration). How the option is used: - preserve_file_partitions=0: Disable it. - preserve_file_partitions=1: Always enable it. - preserve_file_partitions=N, actual file partitions=M: Only enable when M >= N. This threshold preserves I/O parallelism when file partitioning is below it. Note: This may reduce parallelism, rooting from the I/O level, if the number of distinct partitions is less than the target_partitions. +datafusion.optimizer.preserve_file_partitions 0 Minimum number of distinct partition values required to group files by their Hive partition column values (enabling output partitioning declaration). How the option is used: - preserve_file_partitions=0: Disable it. - preserve_file_partitions=1: Always enable it. - preserve_file_partitions=N, actual file partitions=M: Only enable when M >= N. This threshold preserves I/O parallelism when file partitioning is below it. Note: This may reduce parallelism, rooting from the I/O level, if the number of distinct partitions is less than the target_partitions. datafusion.optimizer.repartition_aggregations true Should DataFusion repartition data using the aggregate keys to execute aggregates in parallel using the provided `target_partitions` level datafusion.optimizer.repartition_file_min_size 1048576 Minimum total file size in bytes for file-group byte-range splitting to fire. Files (or merged file groups) smaller than this stay as one partition. Lower values produce more, smaller partitions — better at filling `target_partitions` worth of cores when files are modestly sized, at the cost of slightly more per-partition open / metadata-load overhead. datafusion.optimizer.repartition_file_scans true When set to `true`, datasource partitions will be repartitioned to achieve maximum parallelism. This applies to both in-memory partitions and FileSource's file groups (1 group is 1 partition). For FileSources, only Parquet and CSV formats are currently supported. If set to `true` for a FileSource, all files will be repartitioned evenly (i.e., a single large file might be partitioned into smaller chunks) for parallel scanning. If set to `false` for a FileSource, different files will be read in parallel, but repartitioning won't happen within a single file. If set to `true` for an in-memory source, all memtable's partitions will have their batches repartitioned evenly to the desired number of `target_partitions`. Repartitioning can change the total number of partitions and batches per partition, but does not slice the initial record tables provided to the MemTable on creation. 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 e06cc23a32619..8ef414ee8dde8 100644 --- a/docs/source/library-user-guide/upgrading/55.0.0.md +++ b/docs/source/library-user-guide/upgrading/55.0.0.md @@ -67,7 +67,6 @@ let df = df.fill_null(&ScalarValue::from(0), &["a", "c"])?; let df = df.fill_null(&ScalarValue::from(0), &[])?; ``` - ### `FileScanConfig::partitioned_by_file_group` removed `FileScanConfig::partitioned_by_file_group` and @@ -88,10 +87,10 @@ output partitioning over those partition columns: ```rust,ignore use datafusion_datasource::file_scan_config::{ - FileScanConfigBuilder, hash_partitioning_from_partition_fields, + FileScanConfigBuilder, output_partitioning_from_partition_fields, }; -let output_partitioning = hash_partitioning_from_partition_fields( +let output_partitioning = output_partitioning_from_partition_fields( source.table_schema().table_schema(), source.table_schema().table_partition_cols(), file_groups.len(), @@ -103,7 +102,7 @@ let config = FileScanConfigBuilder::new(object_store_url, source) .build(); ``` -`hash_partitioning_from_partition_fields` returns +`output_partitioning_from_partition_fields` returns `Some(Partitioning::Hash(...))` when partition columns are present and `None` otherwise. If you construct the partitioning manually, pass `Some(Partitioning::Hash(partition_exprs, partition_count))` to diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index 03340c366d70f..f6e072b59bceb 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -160,7 +160,7 @@ The following configuration settings are available: | datafusion.optimizer.repartition_joins | true | Should DataFusion repartition data using the join keys to execute joins in parallel using the provided `target_partitions` level | | datafusion.optimizer.allow_symmetric_joins_without_pruning | true | Should DataFusion allow symmetric hash joins for unbounded data sources even when its inputs do not have any ordering or filtering If the flag is not enabled, the SymmetricHashJoin operator will be unable to prune its internal buffers, resulting in certain join types - such as Full, Left, LeftAnti, LeftSemi, Right, RightAnti, and RightSemi - being produced only at the end of the execution. This is not typical in stream processing. Additionally, without proper design for long runner execution, all types of joins may encounter out-of-memory errors. | | datafusion.optimizer.repartition_file_scans | true | When set to `true`, datasource partitions will be repartitioned to achieve maximum parallelism. This applies to both in-memory partitions and FileSource's file groups (1 group is 1 partition). For FileSources, only Parquet and CSV formats are currently supported. If set to `true` for a FileSource, all files will be repartitioned evenly (i.e., a single large file might be partitioned into smaller chunks) for parallel scanning. If set to `false` for a FileSource, different files will be read in parallel, but repartitioning won't happen within a single file. If set to `true` for an in-memory source, all memtable's partitions will have their batches repartitioned evenly to the desired number of `target_partitions`. Repartitioning can change the total number of partitions and batches per partition, but does not slice the initial record tables provided to the MemTable on creation. | -| datafusion.optimizer.preserve_file_partitions | 0 | Minimum number of distinct partition values required to group files by their Hive partition column values (enabling Hash partitioning declaration). How the option is used: - preserve_file_partitions=0: Disable it. - preserve_file_partitions=1: Always enable it. - preserve_file_partitions=N, actual file partitions=M: Only enable when M >= N. This threshold preserves I/O parallelism when file partitioning is below it. Note: This may reduce parallelism, rooting from the I/O level, if the number of distinct partitions is less than the target_partitions. | +| datafusion.optimizer.preserve_file_partitions | 0 | Minimum number of distinct partition values required to group files by their Hive partition column values (enabling output partitioning declaration). How the option is used: - preserve_file_partitions=0: Disable it. - preserve_file_partitions=1: Always enable it. - preserve_file_partitions=N, actual file partitions=M: Only enable when M >= N. This threshold preserves I/O parallelism when file partitioning is below it. Note: This may reduce parallelism, rooting from the I/O level, if the number of distinct partitions is less than the target_partitions. | | datafusion.optimizer.repartition_windows | true | Should DataFusion repartition data using the partitions keys to execute window functions in parallel using the provided `target_partitions` level | | datafusion.optimizer.repartition_sorts | true | Should DataFusion execute sorts in a per-partition fashion and merge afterwards instead of coalescing first and sorting globally. With this flag is enabled, plans in the form below `text "SortExec: [a@0 ASC]", " CoalescePartitionsExec", " RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1", ` would turn into the plan below which performs better in multithreaded environments `text "SortPreservingMergeExec: [a@0 ASC]", " SortExec: [a@0 ASC]", " RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1", ` | | datafusion.optimizer.subset_repartition_threshold | 4 | Partition count threshold for subset satisfaction optimization. When the current partition count is >= this threshold, DataFusion will skip repartitioning if the required partitioning expression is a subset of the current partition expression such as Hash(a) satisfies Hash(a, b). When the current partition count is < this threshold, DataFusion will repartition to increase parallelism even when subset satisfaction applies. Set to 0 to always repartition (disable subset satisfaction optimization). Set to a high value to always use subset satisfaction. Example (subset_repartition_threshold = 4): `text Hash([a]) satisfies Hash([a, b]) because (Hash([a, b]) is subset of Hash([a]) If current partitions (3) < threshold (4), repartition: AggregateExec: mode=FinalPartitioned, gby=[a, b], aggr=[SUM(x)] RepartitionExec: partitioning=Hash([a, b], 8), input_partitions=3 AggregateExec: mode=Partial, gby=[a, b], aggr=[SUM(x)] DataSourceExec: file_groups={...}, output_partitioning=Hash([a], 3) If current partitions (8) >= threshold (4), use subset satisfaction: AggregateExec: mode=SinglePartitioned, gby=[a, b], aggr=[SUM(x)] DataSourceExec: file_groups={...}, output_partitioning=Hash([a], 8) ` |