You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "ShayanGho (via GitHub)" <gi...@apache.org> on 2026/09/27 02:36:22 UTC
[PR] fix: consider input sizes before reusing join partitioning [datafusion]
ShayanGho opened a new pull request, #25787:
URL: https://github.com/apache/datafusion/pull/25787
## Which issue does this PR close?
Closes #25302.
## Rationale for this change
A small range-partitioned input can become the reference for a much larger join input, forcing the larger input to adopt the smaller input's partition boundaries.
## What changes are included in this PR?
Reject a selected reference when another input in the co-partitioned group has a strictly larger PlanSize. This allows the existing hash-partitioning fallback to apply.
Equal or incomparable sizes do not trigger the new rejection.
## What is the testing strategy for this PR?
Added two regression tests:
- A smaller reference triggers hash fallback. This test failed before the fix.
- Equal-sized inputs preserve the reference's range partitioning.
Validation:
- Distribution optimizer tests.
- Extended workspace tests with the required features.
- cargo fmt --all.
- cargo clippy --all-targets --all-features -- -D warnings.
- ./dev/rust_lint.sh.
## Are there any user-facing changes?
Join plans can use hash repartitioning instead of adapting a larger input to a smaller input's existing partitioning.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: consider input sizes before reusing join partitioning [datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #25787:
URL: https://github.com/apache/datafusion/pull/25787#discussion_r4218757869
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs:
##########
@@ -1210,6 +1210,19 @@ fn enforce_distribution_relationships(
}
};
+ // Reject a reference when another child in the co-partitioned group
Review Comment:
When this fires, the larger child is shuffled in both plans (it is unsatisfied), so rejecting the reference never saves a shuffle; it only adds a shuffle of the reference. With strict `>`, that also happens when the sizes are close: at 10M vs 11M rows, 21M rows get shuffled instead of 11M. It also reverses #24766's native-reference rule, and `range_fallback_preserves_target_parallelism` now asserts the opposite of what it was added to guard. Could we reject only when the reference is small, reusing `JoinSelection`'s `CollectLeft` thresholds (`hash_join_single_partition_threshold{,_rows}`)? The issue's 100-row case would still fall back to hash, inputs of similar size would keep a single shuffle, and #24766's test could keep its original assertion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: consider input sizes before reusing join partitioning [datafusion]
Posted by "jayzhan211 (via GitHub) via github" <gi...@datafusion.apache.org>.
jayzhan211 commented on code in PR #25787:
URL: https://github.com/apache/datafusion/pull/25787#discussion_r4492771963
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs:
##########
@@ -1362,6 +1378,24 @@ fn enforce_distribution_relationships(
}
};
+ // Do not adapt a child above the `CollectLeft` thresholds to a reference
+ // below them; fall back to hash partitioning instead. Otherwise keep the
+ // reference, so that it is not shuffled as well.
+ let best_satisfied_child = best_satisfied_child.filter(|(ref_idx, _)| {
Review Comment:
Fine to handle in a follow-up under #25302, since comparing the big side's key min/max to the split points is its own PR. Size alone can't tell whether the ranges suit the big input. With a sort-merge join on 20M rows against an 8-row reference, this gate gives 183 vs 525 ms when all keys land in one range, but 156 vs 114 ms and 175 vs 145 ms when the ranges are already balanced. The slowdown comes from fewer early stops per partition plus hash skew, not from shuffling the reference.
##########
datafusion/core/tests/physical_optimizer/enforce_distribution.rs:
##########
@@ -1976,6 +1993,42 @@ fn range_hash_join_repartitions_unpartitioned_side_to_match_range() -> Result<()
Ok(())
}
+#[test]
+fn range_hash_join_rejects_smaller_reference() -> Result<()> {
Review Comment:
Could you add coverage for the rest of the gate in this PR? Removing the rows-only arm, treating an unknown input as large, or changing `<` to `<=` leaves every test passing, including the slt files, and sort-merge joins aren't tested at all. This table test fails on each of those changes and could replace the three tests here:
```rs
#[test]
fn range_join_reference_size_gate() -> Result<()> {
let optimizer = TestConfig::default().config.optimizer;
let bytes = optimizer.hash_join_single_partition_threshold;
let rows = optimizer.hash_join_single_partition_threshold_rows;
let known = |n: usize| Precision::Inexact(n);
let absent = Precision::Absent;
// (name, other (rows, bytes), reference (rows, bytes), expect hash fallback)
let cases = [
(
"bytes: straddle",
(absent, known(bytes)),
(absent, known(bytes - 1)),
true,
),
(
"bytes: both at or above",
(absent, known(bytes + 1)),
(absent, known(bytes)),
false,
),
(
"bytes: both below",
(absent, known(100)),
(absent, known(100)),
false,
),
(
"rows: straddle",
(known(rows), absent),
(known(rows - 1), absent),
true,
),
(
"rows: both at or above",
(known(rows + 1), absent),
(known(rows), absent),
false,
),
(
"bytes win over rows",
(known(rows * 10), known(100)),
(known(1), known(100)),
false,
),
(
"other unknown",
(absent, absent),
(known(1), known(100)),
false,
),
(
"reference unknown",
(known(rows * 10), known(bytes * 10)),
(absent, absent),
false,
),
];
type MakeJoin = fn(
Arc<dyn ExecutionPlan>,
Arc<dyn ExecutionPlan>,
&JoinOn,
&JoinType,
) -> Arc<dyn ExecutionPlan>;
let joins: [(&str, MakeJoin); 2] = [
("hash", hash_join_exec),
("sort-merge", sort_merge_join_exec),
];
for (join_name, make_join) in joins {
for (name, other, reference, expect_hash) in cases {
let left = parquet_exec_with_partitioning_and_size(
Partitioning::UnknownPartitioning(4),
other.0,
other.1,
);
let right = parquet_exec_with_partitioning_and_size(
range_partitioning("a", [10, 20, 30], SortOptions::default())?,
reference.0,
reference.1,
);
let join_on = vec![(
Arc::new(Column::new_with_schema("a", &left.schema())?) as _,
Arc::new(Column::new_with_schema("a", &right.schema())?) as _,
)];
let join = make_join(left, right, &join_on, &JoinType::Inner);
let plan = TestConfig::default()
.with_query_execution_partitions(4)
.to_plan(join, &DISTRIB_DISTRIB_SORT);
for child in plan.children() {
let is_hash =
matches!(child.output_partitioning(), Partitioning::Hash(_, 4));
let is_range =
matches!(child.output_partitioning(), Partitioning::Range(_));
assert!(
if expect_hash { is_hash } else { is_range },
"{join_name}, {name}: got {:?}",
child.output_partitioning(),
);
}
}
}
Ok(())
}
```
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs:
##########
@@ -1230,6 +1230,21 @@ impl PlanSize {
fn is_known(&self) -> bool {
self.byte_size.is_some() || self.num_rows.is_some()
}
+
+ /// Whether the plan is below the `CollectLeft` thresholds of `JoinSelection`,
+ /// comparing bytes when known and rows otherwise. `None` when the size is unknown.
+ fn is_small(&self, config: &ConfigOptions) -> Option<bool> {
Review Comment:
Optional cleanup: this repeats the bytes-then-rows rule in `supports_collect_by_thresholds`, and that rule has changed before (#20730). A shared helper keeps them in step and also lets the `child_idx != *ref_idx` check go. I ran this with the existing tests and clippy:
```diff
--- a/datafusion/physical-optimizer/src/join_selection.rs
+++ b/datafusion/physical-optimizer/src/join_selection.rs
- if let Some(byte_size) = stats.total_byte_size.get_value() {
- *byte_size < threshold_byte_size
- } else if let Some(num_rows) = stats.num_rows.get_value() {
- *num_rows < threshold_num_rows
- } else {
- false
- }
-}
+ below_collect_thresholds(&stats, threshold_byte_size, threshold_num_rows)
+ .unwrap_or(false)
+}
+
+/// Whether `stats` is below the `CollectLeft` thresholds, comparing bytes when
+/// known and rows otherwise. `None` when the size is unknown.
+pub(crate) fn below_collect_thresholds(
+ stats: &Statistics,
+ threshold_byte_size: usize,
+ threshold_num_rows: usize,
+) -> Option<bool> {
+ if let Some(byte_size) = stats.total_byte_size.get_value() {
+ Some(*byte_size < threshold_byte_size)
+ } else {
+ stats
+ .num_rows
+ .get_value()
+ .map(|num_rows| *num_rows < threshold_num_rows)
+ }
+}
--- a/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs
+++ b/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs
let best_satisfied_child = best_satisfied_child.filter(|(ref_idx, _)| {
- let reference_size =
- PlanSize::from_plan(&children[*ref_idx].context.plan, stats_ctx);
- if reference_size.is_small(config) != Some(true) {
- return true;
- }
-
- !unsatisfied_children.iter().any(|&child_idx| {
- child_idx != *ref_idx
- && PlanSize::from_plan(&children[child_idx].context.plan, stats_ctx)
- .is_small(config)
- == Some(false)
- })
+ let is_small = |idx: usize| {
+ let stats = stats_ctx
+ .compute_arc(&children[idx].context.plan, &StatisticsArgs::new())
+ .ok()?;
+ below_collect_thresholds(
+ &stats,
+ config.optimizer.hash_join_single_partition_threshold,
+ config.optimizer.hash_join_single_partition_threshold_rows,
+ )
+ };
+ is_small(*ref_idx) != Some(true)
+ || !unsatisfied_children
+ .iter()
+ .any(|&child_idx| is_small(child_idx) == Some(false))
});
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: consider input sizes before reusing join partitioning [datafusion]
Posted by "ShayanGho (via GitHub)" <gi...@apache.org>.
ShayanGho commented on PR #25787:
URL: https://github.com/apache/datafusion/pull/25787#issuecomment-5977336439
Hello @jayzhan211 @gene-bordegaray, would either of you have time for this one too? It's a 13-line change in enforce_distribution_relationships for #25302: don't adopt a reference child's partitioning when another input is strictly larger. Merged with main, and it passes alongside the new native-reference test from [apache/datafusion#24766](https://github.com/apache/datafusion/pull/24766).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: consider input sizes before reusing join partitioning [datafusion]
Posted by "ShayanGho (via GitHub)" <gi...@apache.org>.
ShayanGho commented on code in PR #25787:
URL: https://github.com/apache/datafusion/pull/25787#discussion_r4244572231
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs:
##########
@@ -1210,6 +1210,19 @@ fn enforce_distribution_relationships(
}
};
+ // Reject a reference when another child in the co-partitioned group
Review Comment:
Makes sense, done. The reference is now rejected only when it is below the `CollectLeft` thresholds and another input is above them. Inputs on the same side of the thresholds keep the reference, so only one side is shuffled (`range_hash_join_preserves_large_reference` covers the 10M vs 11M case), and `range_fallback_preserves_target_parallelism` is back to its original assertion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: consider input sizes before reusing join partitioning [datafusion]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #25787:
URL: https://github.com/apache/datafusion/pull/25787#issuecomment-6431423736
Marked as draft after 14 days without activity: merge conflicts.
Please address these items and mark **Ready for review** when ready.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: consider input sizes before reusing join partitioning [datafusion]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #25787:
URL: https://github.com/apache/datafusion/pull/25787#issuecomment-5852151697
## [Codecov](https://app.codecov.io/gh/apache/datafusion/pull/25787?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 82.50%. Comparing base ([`acf5c89`](https://app.codecov.io/gh/apache/datafusion/commit/acf5c89e6454469c763c0fb5539e5a294d3439ed?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`55e9ae0`](https://app.codecov.io/gh/apache/datafusion/commit/55e9ae004c12e9b9bc56f63685597d9a6c6941c2?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #25787 +/- ##
=======================================
Coverage 82.50% 82.50%
=======================================
Files 1141 1141
Lines 438825 438834 +9
Branches 438825 438834 +9
=======================================
+ Hits 362050 362059 +9
+ Misses 54892 54888 -4
- Partials 21883 21887 +4
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/datafusion/pull/25787?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org