Skip to content

Commit 5026ac4

Browse files
gstamatakis95alambclaude
authored
fix: declare Hive partitioned scans as Range (#25279)
## 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 #23436. ## Rationale for this change - Hive grouped scans declare Partitioning::Hash, but no hash ever ran. - The optimizer treats any two Hash partitionings with equal counts as co-partitioned, so it skips the repartition. - Partitioned hash joins then drop rows. Fact {A, B, C} joined to dimension {A, C, D} returns only A, because fact group 1 holds B while dimension group 1 holds C. - Joining a Hive scan to a VALUES list also loses rows, since the planner hash repartitions the other side. ## What changes are included in this PR? - Add FileGroup::group_by_partition_values_with_split_points, returning groups plus the SplitPoints between them. - Merge values into contiguous chunks rather than round robin. Only contiguous chunks can be described by split points. Group counts are unchanged. - Give ListFilesResult a partition_split_points field. - In ListingTable::scan build a logical Range partitioning from the partition columns and the split points and convert it with create_physical_partitioning, the same path a user declared output partitioning takes. No new public helper. - Declare nothing when split_file_groups_by_statistics re-cuts the groups. Comparing counts misses this, since the re-cut can return the same number. - Drop the dynamic filter guard in HashJoinExec since it existed only because Hive groups reported Hash. ## What is the testing strategy for this PR? - TEST 14 and 15 in preserve_file_partitioning.slt cover the two failures above. - TEST 16 self joins the merged high cardinality table. TEST 17 asserts the statistics re-cut declares nothing. - Reverting only table.rs makes TEST 14 return one row instead of two and breaks TEST 12. - Six unit tests route every file through its split points and assert it lands in its own group, covering NULLs, compound keys, a NULL inside a split point, and unordered split points. - Two datafusion-proto round trips cover Dictionary(UInt16, Utf8) and Utf8View split points. ## Are there any user-facing changes? - EXPLAIN prints Range([col ASC], [(v1), (v2)], N) instead of Hash([col], N). - Joins between Hive tables with different values gain a RepartitionExec(Range) rather than returning wrong results. - Partitioned dynamic filter pushdown works again under preserve_file_partitions. - File group membership changes when distinct values exceed target_partitions. Byte balance can shift on date partitions with larger recent days. Set preserve_file_partitions = 0 to restore the old behaviour. - API break: ListFilesResult gains a public field. Add partition_split_points: vec![] or use ... - Providers declaring Hash over value grouped files should declare Range from the split points instead. They already return missing rows today when the other join input is hash repartitioned. --------- Co-authored-by: Andrew Lamb <[email protected]> Co-authored-by: Claude Fable 5.1 <[email protected]>
1 parent c0ec512 commit 5026ac4

7 files changed

Lines changed: 626 additions & 106 deletions

File tree

‎datafusion/catalog-listing/src/table.rs‎

Lines changed: 32 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -25,13 +25,13 @@ use async_trait::async_trait;
2525
use datafusion_catalog::{ScanArgs, ScanResult, Session, TableProvider};
2626
use datafusion_common::stats::{Precision, is_known_empty};
2727
use datafusion_common::{
28-
Constraints, DFSchema, SchemaExt, Statistics, internal_datafusion_err, plan_err,
29-
project_schema,
28+
Constraints, DFSchema, SchemaExt, SplitPoint, Statistics, internal_datafusion_err,
29+
plan_err, project_schema,
3030
};
3131
use datafusion_datasource::file::FileSource;
3232
use datafusion_datasource::file_groups::FileGroup;
3333
use datafusion_datasource::file_scan_config::{
34-
FileScanConfig, FileScanConfigBuilder, output_partitioning_from_partition_fields,
34+
FileScanConfig, FileScanConfigBuilder, output_range_partitioning_from_split_points,
3535
};
3636
use datafusion_datasource::file_sink_config::{FileOutputMode, FileSinkConfig};
3737
#[expect(deprecated)]
@@ -69,6 +69,8 @@ pub struct ListFilesResult {
6969
pub statistics: Statistics,
7070
/// Whether files are grouped by partition values.
7171
pub grouped_by_partition: bool,
72+
/// Boundaries between file groups, empty unless `grouped_by_partition` is true.
73+
pub partition_split_points: Vec<SplitPoint>,
7274
}
7375

7476
/// Built in [`TableProvider`] that reads data from one or more files as a single table.
@@ -668,6 +670,7 @@ impl ListingTable {
668670
file_groups: mut partitioned_file_lists,
669671
statistics,
670672
grouped_by_partition: partitioned_by_file_group,
673+
partition_split_points,
671674
} = self
672675
.list_files_for_scan(state, &partition_filters, statistic_file_limit)
673676
.await?;
@@ -687,6 +690,7 @@ impl ListingTable {
687690
.config_options()
688691
.execution
689692
.split_file_groups_by_statistics;
693+
let mut regrouped_by_statistics = false;
690694
match split_file_groups_by_statistics
691695
.then(|| {
692696
output_ordering.first().map(|output_ordering| {
@@ -704,6 +708,7 @@ impl ListingTable {
704708
Some(Ok(new_groups)) => {
705709
if new_groups.len() <= file_group_count {
706710
partitioned_file_lists = new_groups;
711+
regrouped_by_statistics = true;
707712
} else {
708713
log::debug!(
709714
"attempted to split file groups by statistics, but there were more file groups than target_partitions; falling back to unordered"
@@ -737,26 +742,25 @@ impl ListingTable {
737742
)?
738743
}
739744
};
745+
Some(output_partitioning)
746+
} else if partitioned_by_file_group && !regrouped_by_statistics {
747+
output_range_partitioning_from_split_points(
748+
&self.table_schema,
749+
&table_partition_cols.into(),
750+
partition_split_points,
751+
)?
752+
} else {
753+
None
754+
};
755+
if let Some(output_partitioning) = &output_partitioning {
740756
let partition_count = output_partitioning.partition_count();
741757
if partitioned_file_lists.len() != partition_count {
742758
return plan_err!(
743759
"ListingTable output_partitioning has {partition_count} partitions, but the scan has {} file groups",
744760
partitioned_file_lists.len()
745761
);
746762
}
747-
Some(output_partitioning)
748-
} else if partitioned_by_file_group {
749-
// Files are grouped by partition column values: declare output
750-
// partitioning on those columns so the optimizer can skip
751-
// repartitioning for aggregates and joins on the partition columns.
752-
output_partitioning_from_partition_fields(
753-
&self.table_schema,
754-
&table_partition_cols.clone().into(),
755-
partitioned_file_lists.len(),
756-
)
757-
} else {
758-
None
759-
};
763+
}
760764

761765
let Some(object_store_url) =
762766
self.table_paths.first().map(ListingTableUrl::object_store)
@@ -1005,6 +1009,7 @@ impl ListingTable {
10051009
file_groups: vec![],
10061010
statistics: Statistics::new_unknown(&self.file_schema),
10071011
grouped_by_partition: false,
1012+
partition_split_points: vec![],
10081013
});
10091014
};
10101015
let (file_group, inexact_stats) = self
@@ -1018,28 +1023,31 @@ impl ListingTable {
10181023
// skip repartitioning for aggregates and joins on partition columns.
10191024
let threshold = ctx.config_options().optimizer.preserve_file_partitions;
10201025

1021-
let (file_groups, grouped_by_partition) =
1026+
let (file_groups, grouped_by_partition, partition_split_points) =
10221027
if threshold > 0 && !self.options.table_partition_cols.is_empty() {
1023-
let grouped = file_group.group_by_partition_values(file_group_count);
1028+
let (grouped, split_points) = file_group
1029+
.group_by_partition_values_with_split_points(file_group_count);
10241030
if grouped.len() >= threshold {
1025-
(grouped, true)
1031+
(grouped, true, split_points)
10261032
} else {
10271033
let all_files: Vec<_> =
10281034
grouped.into_iter().flat_map(|g| g.into_inner()).collect();
10291035
(
10301036
FileGroup::new(all_files).split_files(file_group_count),
10311037
false,
1038+
vec![],
10321039
)
10331040
}
10341041
} else {
1035-
(file_group.split_files(file_group_count), false)
1042+
(file_group.split_files(file_group_count), false, vec![])
10361043
};
10371044

10381045
self.list_files_result_from_groups(
10391046
ctx,
10401047
file_groups,
10411048
inexact_stats,
10421049
grouped_by_partition,
1050+
partition_split_points,
10431051
)
10441052
}
10451053

@@ -1067,6 +1075,7 @@ impl ListingTable {
10671075
file_groups: vec![],
10681076
statistics: Statistics::new_unknown(&self.file_schema),
10691077
grouped_by_partition: false,
1078+
partition_split_points: vec![],
10701079
});
10711080
};
10721081
let (file_group, inexact_stats) =
@@ -1078,7 +1087,7 @@ impl ListingTable {
10781087
let file_groups =
10791088
self.filter_declared_file_groups_by_partition_filters(file_groups, filters)?;
10801089

1081-
self.list_files_result_from_groups(ctx, file_groups, inexact_stats, false)
1090+
self.list_files_result_from_groups(ctx, file_groups, inexact_stats, false, vec![])
10821091
}
10831092

10841093
fn filter_declared_file_groups_by_partition_filters(
@@ -1113,6 +1122,7 @@ impl ListingTable {
11131122
file_groups: Vec<FileGroup>,
11141123
inexact_stats: bool,
11151124
grouped_by_partition: bool,
1125+
partition_split_points: Vec<SplitPoint>,
11161126
) -> datafusion_common::Result<ListFilesResult> {
11171127
let (file_groups, stats) = compute_all_files_statistics(
11181128
file_groups,
@@ -1129,6 +1139,7 @@ impl ListingTable {
11291139
file_groups,
11301140
statistics: stats,
11311141
grouped_by_partition,
1142+
partition_split_points,
11321143
})
11331144
}
11341145

0 commit comments

Comments
 (0)