You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "kosiew (via GitHub)" <gi...@apache.org> on 2025/08/22 13:11:06 UTC
[PR] Remove O(n log n) sorting/allocation in HashJoin dynamic filter accumulator by pre-indexing bounds per partition [datafusion]
kosiew opened a new pull request, #17286:
URL: https://github.com/apache/datafusion/pull/17286
## Which issue does this PR close?
* Closes #17280.
## Rationale for this change
The accumulator previously collected build-side partition bounds and then **sorted** them with `sorted_by_key`, which:
* Introduced **extra allocations** and
* Added **O(n log n)** overhead on the number of completed partitions.
Since partitions already have stable IDs, we can **pre-index** bounds by partition ID and avoid sorting entirely. This makes dynamic filter construction **O(n)** with fewer allocations, improves predictability, and eliminates a source of nondeterminism tied to completion order.
## What changes are included in this PR?
* Replaced `PartitionBounds` + `sorted_by_key` with a **preallocated `Vec<Option<Vec<ColumnBounds>>>`** indexed by partition ID.
* Eliminated sorting and the dependency on `itertools`, reducing allocations and algorithmic overhead.
* Updated accumulator logic to:
* **Bounds insertion in O(1)** at the correct index (by partition ID).
* Validate out-of-range partition IDs and return a clear internal error instead of panicking.
* Build the dynamic filter once **all partitions have reported**, ignoring missing partitions.
* Adjusted `create_filter_from_partition_bounds` to iterate the fixed-index vector and construct predicates without any intermediate sorting/allocation.
* Kept/clarified determinism as a by-product: completion order no longer affects the resulting predicate.
## Are these changes tested?
Yes.
* Added an async test `test_hashjoin_dynamic_filter_pushdown_out_of_order` that intentionally reverses completion order of build-side partitions across runs and asserts the resulting dynamic filter predicate string is identical, proving order independence while validating logic.
* Existing join and dynamic filter tests continue to pass.
## Are there any user-facing changes?
No API-breaking changes.
* Internals of dynamic filter construction were optimized for efficiency and determinism.
* Query semantics remain unchanged, but performance improves due to reduced allocations and removal of sorting overhead.
---
--
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] Remove O(n log n) sorting/allocation in HashJoin dynamic filter accumulator by pre-indexing bounds per partition [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on PR #17286:
URL: https://github.com/apache/datafusion/pull/17286#issuecomment-3216216601
I think this change could not improve e2e performance: optimizations should be done on bottlenecks (those steps that are repeating millions of times in a query), this step only get executed 1 time per query, and it's bounded by the number of CPU core, so we'd better choose simplicity over performance.
Besides, sorting small array is super fast thanks to compiler and hardware. Perhaps the original solution is faster.
--
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] Remove O(n log n) sorting/allocation in HashJoin dynamic filter accumulator by pre-indexing bounds per partition [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #17286:
URL: https://github.com/apache/datafusion/pull/17286#discussion_r2294475802
##########
datafusion/physical-plan/src/joins/hash_join.rs:
##########
@@ -115,34 +114,6 @@ impl ColumnBounds {
}
}
-/// Represents the bounds for all join key columns from a single partition.
-/// This contains the min/max values computed from one partition's build-side data.
-#[derive(Debug, Clone)]
-struct PartitionBounds {
Review Comment:
Yes please!
--
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] Remove O(n log n) sorting/allocation in HashJoin dynamic filter accumulator by pre-indexing bounds per partition [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #17286:
URL: https://github.com/apache/datafusion/pull/17286#discussion_r2294234672
##########
datafusion/physical-plan/src/joins/hash_join.rs:
##########
@@ -255,21 +227,21 @@ impl SharedBoundsAccumulator {
/// (col0 >= p1_min0 AND col0 <= p1_max0 AND col1 >= p1_min1 AND col1 <= p1_max1))
fn create_filter_from_partition_bounds(
&self,
- bounds: &[PartitionBounds],
+ bounds: &[Option<Vec<ColumnBounds>>],
) -> Result<Arc<dyn PhysicalExpr>> {
- if bounds.is_empty() {
+ if bounds.iter().all(|b| b.is_none()) {
Review Comment:
This is now more work
##########
datafusion/physical-plan/src/joins/hash_join.rs:
##########
@@ -115,34 +114,6 @@ impl ColumnBounds {
}
}
-/// Represents the bounds for all join key columns from a single partition.
-/// This contains the min/max values computed from one partition's build-side data.
-#[derive(Debug, Clone)]
-struct PartitionBounds {
Review Comment:
I think keeping the struct is helpful from a readability perspective to avoid nested vecs
--
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] Remove O(n log n) sorting/allocation in HashJoin dynamic filter accumulator by pre-indexing bounds per partition [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on PR #17286:
URL: https://github.com/apache/datafusion/pull/17286#issuecomment-3216556630
Thanks @2010YOUY01 , @alamb , @adriangb
for your thoughts on this.
Closing as this [optimisation is not necessary](https://github.com/apache/datafusion/pull/17286#issuecomment-3216216601).
--
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] Remove O(n log n) sorting/allocation in HashJoin dynamic filter accumulator by pre-indexing bounds per partition [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew closed pull request #17286: Remove O(n log n) sorting/allocation in HashJoin dynamic filter accumulator by pre-indexing bounds per partition
URL: https://github.com/apache/datafusion/pull/17286
--
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