This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-23189-0977bb3819edc1875025541bc00dd82262edb22b in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 66b059e041e332293c21bde857865cfdf04f0772 Author: Phoenix <[email protected]> AuthorDate: Fri Jul 10 05:14:48 2026 +0800 refactor: remove redundant partitioned_by_file_group file scan field (#23189) ## Which issue does this PR close? <!-- We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123. --> - Closes #23099 . ## Rationale for this change <!-- Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed. Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes. --> `FileScanConfig` had two overlapping ways to declare a file scan's output partitioning: - `partitioned_by_file_group: bool` — a shorthand meaning "the file groups are organized by Hive partition column values, so the output is Hash-partitioned on those columns", and - `output_partitioning: Option<Partitioning>` — a general, explicit declared partitioning (added in #22657). The bool is just a lazy shorthand for one specific `output_partitioning` value (`Partitioning::Hash` over the partition columns), and every place that consumed it (`output_partitioning()`, `repartitioned()`, `create_sibling_state()`) already checked `output_partitioning.is_some() || partitioned_by_file_group`. Keeping both is redundant and the `ListingTable` builder ended up setting *both*. This PR makes `output_partitioning` the single source of truth. ## What changes are included in this PR? <!-- There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR. --> Following the issue's first option ("Remove `partitioned_by_file_group`"): - Remove `FileScanConfig::partitioned_by_file_group`, the corresponding `FileScanConfigBuilder` field, and the `with_partitioned_by_file_group` builder method. - `ListingTable::scan` now derives the partition-column `Partitioning::Hash` itself (once its file groups are finalized, so the partition count is correct) and passes it through the existing `with_output_partitioning`. The previous `with_output_partitioning(declared)` + `with_partitioned_by_file_group(...)` double-set is collapsed into one branch. - `hash_partitioning_from_partition_fields` is made `pub` so `ListingTable` (a separate crate) can reuse the derivation instead of duplicating the column-index resolution. - proto already round-trips `output_partitioning`, so no behavior is lost: the now-vestigial `partitioned_by_file_group` wire field is left unset on write and ignored on read. The field is kept in the `.proto` definition for backward compatibility. - `output_partitioning()` / `create_sibling_state()` / `repartitioned()` now key solely off `output_partitioning`. ## Are these changes tested? <!-- We typically require tests for all PRs in order to: 1. Prevent the code from being accidentally broken by subsequent changes 2. Serve as another way to document the expected behavior of the code If tests are not included in your PR, please explain why (for example, are they covered by existing tests)? --> Yes — by existing tests, updated to the new single-field model: - `datafusion-datasource`: `test_output_partitioning_with_partition_columns`, `test_output_partitioning_no_partition_columns`, `test_declared_output_partitioning_projects_with_scan`, and the `file_stream` work-stealing test `morsel_partitioned_by_file_group_keeps_files_local` (which verifies that a declared output partitioning keeps each stream's files local). - `datafusion-proto`: `roundtrip_parquet_exec_output_partitioning` (and the other `roundtrip_parquet_exec_*` cases) cover the partitioning round-trip. The old `roundtrip_parquet_exec_partitioned_by_file_group` test exercised the removed API and is dropped, as its coverage is subsumed by the `output_partitioning` round-trip test. All of the above pass, along with `cargo fmt --all --check` and `cargo clippy --all-targets --all-features -- -D warnings` for the affected crates. ## Are there any user-facing changes? <!-- If there are user-facing changes then we may require documentation to be updated before approving the PR. --> <!-- If there are any breaking changes to public APIs, please add the `api change` label. --> Yes — public API changes : - Removed: the public `FileScanConfig::partitioned_by_file_group` field and the `FileScanConfigBuilder::with_partitioned_by_file_group` method. Callers should set `with_output_partitioning(Some(Partitioning::Hash(..)))` instead (or use the now-public `hash_partitioning_from_partition_fields` helper). - Added: `hash_partitioning_from_partition_fields` is now `pub`. Query results, optimizer decisions (e.g. eliding `RepartitionExec`), and the serialized (proto) wire format are unchanged. There is one **display-only** change: EXPLAIN now renders `output_partitioning=Hash(...)` on `DataSourceExec` for partition-grouped scans. The scan already produced that partitioning before (it was derived lazily inside `output_partitioning()`); it is now stored on the `output_partitioning` field and therefore shown. The `repartition_subset_satisfaction` and `preserve_file_partitioning` slt expected plans are updated accordingly. `cargo-semver-checks` will flag the removals as breaking, which is expected for this cleanup. --------- Signed-off-by: Jiawei Zhao <[email protected]> --- datafusion/catalog-listing/src/table.rs | 20 ++++-- datafusion/common/src/config.rs | 2 +- datafusion/datasource/src/file_scan_config/mod.rs | 70 ++++++-------------- datafusion/datasource/src/file_stream/mod.rs | 32 +++++---- datafusion/proto/src/physical_plan/from_proto.rs | 27 ++++++-- datafusion/proto/src/physical_plan/to_proto.rs | 4 +- .../proto/tests/cases/roundtrip_physical_plan.rs | 77 +++++++++++++++++----- .../sqllogictest/test_files/information_schema.slt | 2 +- .../test_files/preserve_file_partitioning.slt | 16 ++--- .../test_files/repartition_subset_satisfaction.slt | 16 ++--- docs/source/library-user-guide/upgrading/55.0.0.md | 41 ++++++++++++ docs/source/user-guide/configs.md | 2 +- 12 files changed, 200 insertions(+), 109 deletions(-) diff --git a/datafusion/catalog-listing/src/table.rs b/datafusion/catalog-listing/src/table.rs index 632b829b16..23c67efa74 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, 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) = diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 454af28c14..b649ecad57 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 b73d100e05..660d0cd7a5 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 <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 { diff --git a/datafusion/datasource/src/file_stream/mod.rs b/datafusion/datasource/src/file_stream/mod.rs index e0641310c2..6daed7c338 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<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,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.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)), @@ -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 53ff4a41d4..19e49f3cb8 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, 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) } diff --git a/datafusion/proto/src/physical_plan/to_proto.rs b/datafusion/proto/src/physical_plan/to_proto.rs index 4614c4f002..e93bb9cc5f 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 8acb891d6a..bb99a1ecbf 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, @@ -4220,43 +4225,81 @@ fn roundtrip_file_scan_config(scan_config: FileScanConfig) -> Result<FileScanCon } #[test] -fn roundtrip_parquet_exec_partitioned_by_file_group() -> Result<()> { +fn roundtrip_parquet_exec_output_partitioning() -> 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 output_partitioning = + Partitioning::Hash(vec![Arc::new(Column::new("col", 0))], 1); 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) + .with_output_partitioning(Some(output_partitioning.clone())) .build(); - assert!(roundtrip_file_scan_config(scan_config)?.partitioned_by_file_group); + assert_eq!( + roundtrip_file_scan_config(scan_config)?.output_partitioning, + Some(output_partitioning) + ); + Ok(()) } #[test] -fn roundtrip_parquet_exec_output_partitioning() -> Result<()> { +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 file_source = Arc::new(ParquetSource::new(Arc::clone(&file_schema))); - let output_partitioning = - Partitioning::Hash(vec![Arc::new(Column::new("col", 0))], 1); + 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/file.parquet".to_string(), - 1024, - )])]) - .with_output_partitioning(Some(output_partitioning.clone())) + .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(); - assert_eq!( - roundtrip_file_scan_config(scan_config)?.output_partitioning, - Some(output_partitioning) - ); + 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(()) } diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index bf45564e26..1adf98f67f 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: Th [...] +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: [...] 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 `fa [...] diff --git a/datafusion/sqllogictest/test_files/preserve_file_partitioning.slt b/datafusion/sqllogictest/test_files/preserve_file_partitioning.slt index 412d606df9..e2dd22cc82 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( [...] 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, pred [...] +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=Has [...] 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 N [...] +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 N [...] 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_ [...] +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_ [...] # 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_partitioni [...] +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 af74280c10..5371ca59be 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 LA [...] +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 LA [...] # 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, [...] +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, [...] # 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 }\" [...] 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 [...] +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 [...] # 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 }\" [...] -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, ti [...] +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, ti [...] # 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 [...] -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 [...] +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 [...] +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 [...] # 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/scr [...] -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 [...] +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/scr [...] +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 [...] # Verify results match with subset satisfaction query TPR rowsort 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 26a26b69b5..181e0e0b7f 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,47 @@ 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, output_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(), +); + +let config = FileScanConfigBuilder::new(object_store_url, source) + .with_file_groups(file_groups) + .with_output_partitioning(output_partitioning) + .build(); +``` + +`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 +`with_output_partitioning(...)`. + ### User `SpillFile` traits instead of [`RefCountedTempFile`] Spill file APIs now use the `datafusion_execution::SpillFile` trait instead of diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index 03340c366d..f6e072b59b 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 execut [...] | 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 in [...] -| 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 [...] +| 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 wh [...] | 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 enviro [...] | 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 e [...] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
