Repository navigation
Handle canceled partitioned hash join dynamic filters lazily - #21666
Merged
adriangb merged 18 commits intoApr 22, 2026
Merged
Conversation
adriangb
marked this pull request as ready for review
April 16, 2026 11:55
Contributor
Author
|
run benchmarks |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing codex/hash-join-empty-partition-reporting (d17d5e4) to 5c653be (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing codex/hash-join-empty-partition-reporting (d17d5e4) to 5c653be (merge-base) diff using: clickbench_partitioned File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing codex/hash-join-empty-partition-reporting (d17d5e4) to 5c653be (merge-base) diff using: tpch File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
Contributor
Author
|
run benchmark tpcds |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing codex/hash-join-empty-partition-reporting (d17d5e4) to 5c653be (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
Contributor
|
query 72 took a bit of a hit here. Query72select i_item_desc
,w_warehouse_name
,d1.d_week_seq
,sum(case when p_promo_sk is null then 1 else 0 end) no_promo
,sum(case when p_promo_sk is not null then 1 else 0 end) promo
,count(*) total_cnt
from catalog_sales
join inventory on (cs_item_sk = inv_item_sk)
join warehouse on (w_warehouse_sk=inv_warehouse_sk)
join item on (i_item_sk = cs_item_sk)
join customer_demographics on (cs_bill_cdemo_sk = cd_demo_sk)
join household_demographics on (cs_bill_hdemo_sk = hd_demo_sk)
join date_dim d1 on (cs_sold_date_sk = d1.d_date_sk)
join date_dim d2 on (inv_date_sk = d2.d_date_sk)
join date_dim d3 on (cs_ship_date_sk = d3.d_date_sk)
left outer join promotion on (cs_promo_sk=p_promo_sk)
left outer join catalog_returns on (cr_item_sk = cs_item_sk and cr_order_number = cs_order_number)
where d1.d_week_seq = d2.d_week_seq
and inv_quantity_on_hand < cs_quantity
and d3.d_date > (d1.d_date + INTERVAL '5 days')
and hd_buy_potential = '1001-5000'
and d1.d_year = 2001
and cd_marital_status = 'M'
group by i_item_desc,w_warehouse_name,d1.d_week_seq
order by total_cnt desc, i_item_desc, w_warehouse_name, d_week_seq
limit 100; |
Contributor
Author
|
Yes. And I think this was another bandaid. But it's closer to the root cause than previous attempts. This has to do with cancellation when multiple joins are involved. TLDR I think what is happening is when you have multiple joins you end up with a tree of operators. One of the joins up higher in the tree hits the new optimization and aborts work, dropping tasks that would have polled downstream joins. But not the downstream join is stuck waiting for all of it's partition tasks to finish even though they never will. I think we were all operating under the assumption that the issue was within a single join operator but really it's an issue any time an upstream operator cancels on a join. I think the real solution is to track when a join build partition task gets dropped and report that to the dynamic filter building so that it doesn't wait for that partition to report. |
Contributor
Author
|
run benchmark tpcds |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing codex/hash-join-empty-partition-reporting (0584854) to 5c653be (merge-base) diff using: tpcds File an issue against this benchmark runner |
adriangb
changed the title
Contributor
There was a problem hiding this comment.
Pull request overview
Fixes a hang in partitioned hash join dynamic filter coordination by allowing partitions that are dropped early to be treated as “canceled” and not block filter finalization (DataFusion issue #21625).
Changes:
- Add cancellation tracking for partitioned build-side reports and treat canceled partitions as
truein the synthesized partitioned dynamic filter. - Mark partitioned
HashJoinStreampartitions as canceled onDropwhen they never reported build data. - Add a regression test covering the early-completing
RightSemiparent join scenario that previously hung.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 3 comments.
| File | Description |
|---|---|
| datafusion/physical-plan/src/joins/hash_join/stream.rs | Track whether build info was reported and report partition cancellation to the coordinator on Drop. |
| datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs | Replace barrier-based coordination with explicit partition status + notify-based completion; synthesize filters that handle canceled partitions. |
| datafusion/physical-plan/src/joins/hash_join/exec.rs | Refactor dynamic-filter accumulator initialization and add a regression test for the cancellation/hang scenario. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| enum CompletionState { | ||
| Pending, | ||
| Finalizing, | ||
| Ready(std::result::Result<(), String>), |
There was a problem hiding this comment.
CompletionState::Ready stores errors as Result<(), String>, which forces later callers to lose the original DataFusionError variant/backtrace/context. Consider storing Result<(), Arc<DataFusionError>> (or datafusion_common::SharedResult<()>) in CompletionState instead, so you can propagate DataFusionError::Shared(...) to all waiters without stringifying.
Suggested change
| Ready(std::result::Result<(), String>), | |
| Ready(std::result::Result<(), Arc<datafusion_common::DataFusionError>>), |
Comment on lines
391
to
+394
| /// # Returns | ||
| /// * `Result<()>` - Ok if successful, Err if filter update failed or mode mismatch | ||
| pub(crate) async fn report_build_data(&self, data: PartitionBuildData) -> Result<()> { | ||
| // Store data in the accumulator | ||
| { | ||
| let finalize_input = { |
There was a problem hiding this comment.
The report_build_data doc comment still says "have reported (barrier wait)", but this method no longer uses tokio::sync::Barrier (it uses Notify/CompletionState). Please update the docs to match the current synchronization mechanism so the comment doesn’t mislead future changes.
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
Contributor
Author
|
run benchmarks |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing codex/hash-join-empty-partition-reporting (7011c5d) to 5c653be (merge-base) diff using: tpch File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing codex/hash-join-empty-partition-reporting (7011c5d) to 5c653be (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing codex/hash-join-empty-partition-reporting (7011c5d) to 5c653be (merge-base) diff using: clickbench_partitioned File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
Contributor
Author
|
Okay I think this addresses the root cause with no performance regression or behavior changes. |
Contributor
Author
|
@RatulDawar could you let me know what you think of this solution? |
Replaces the manual PartitionBuildData construction + report_build_data call + build_reported flag set in collect_build_side with a single transition_after_build_collected method, making it impossible to forget to report build data when transitioning state. Co-Authored-By: Claude Sonnet 4.6 <[email protected]>
Per review feedback, the hoisting of on_right_exprs above the SharedBuildAccumulator init was an unrelated cleanup (same Arc-clone and allocation cost) that distracted from the cancellation fix. Restore the two separate constructions to match main. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
adriangb
force-pushed
the
codex/hash-join-empty-partition-reporting
branch
from
April 19, 2026 15:46
5ad96d9 to
9023c19
Compare
Per review feedback, replace the hand-rolled Mutex<CompletionState> + Notify pair with a watch::Sender<Option<SharedResult<()>>> for the terminal completion broadcast. The mutex still guards the incremental per-partition accumulation (which needs partial writes) and the finalizer-election bool, but wait_for_completion collapses to a single rx.wait_for(|v| v.is_some()).await instead of a manual re-check loop. Net -8 lines and drops the three-state CompletionState enum. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
This reverts commit 4e95802.
Clarify the role of the Notify primitive so the coordination pattern is obvious without reading the wait/finish methods: it wakes parked partitions once the elected finalizer has published CompletionState::Ready. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
kosiew
requested changes
Apr 21, 2026
Contributor
There was a problem hiding this comment.
Thanks for the ping.
I dug into the lifecycle around reporting and cancellation, and there is a subtle timing issue that could reintroduce a hang. I left details inline. I also added a couple of suggestions that could help keep things easier to reason about going forward.
| self.build_waiter = Some(OnceFut::new(async move { | ||
| acc.report_build_data(build_data).await | ||
| })); | ||
| self.build_reported = true; |
Contributor
There was a problem hiding this comment.
I think there is still a race here. build_reported gets flipped as soon as the OnceFut is created, but the future is lazy and only runs once wait_for_partition_bounds_report() actually polls it.
If a parent drops this stream after transition_after_build_collected() returns, but before the waiter is ever polled, Drop will skip report_canceled_partition() even though nothing was delivered to the coordinator. That feels like it recreates the original hang, just in a narrower timing window.
Would it make sense to only mark this as reported after the waiter completes successfully? Alternatively, maybe replace the bool with a small lifecycle state so Drop can still cancel something that was only scheduled but never observed.
| let bounds_expr = | ||
| create_bounds_predicate(&self.on_right, &partition_data.bounds); | ||
|
|
||
| if let Some(filter_expr) = match (membership_expr, bounds_expr) { |
Contributor
There was a problem hiding this comment.
I noticed the “membership + bounds gives final predicate” logic now exists in both the CollectLeft and Partitioned finalize paths, with slightly different control flow.
It might be worth pulling that into a small helper so both branches stay aligned as the dynamic filter logic evolves. That would make future changes a bit safer.
| /// Optional future to signal when build information has been reported by all partitions | ||
| /// and the dynamic filter has been updated | ||
| build_waiter: Option<OnceFut<()>>, | ||
| /// Tracks whether this partition has already reported build information to the coordinator. |
Contributor
There was a problem hiding this comment.
Once the blocking issue above is sorted out, this field might benefit from a stronger type than just bool.
Something like NotReported, ReportScheduled, and ReportDelivered would make the drop semantics much clearer and help avoid another mix-up between "scheduled" and "actually seen by the coordinator".
…lper Per PR apache#21666 review feedback: * Replace `build_reported: bool` with a `BuildReportState` enum (NotReported / ReportScheduled / ReportDelivered). `ReportDelivered` is only set after the build waiter's `get_shared` returns `Ok`, so `Drop` still cancels a partition whose report was merely scheduled but never observed by the coordinator. `store_canceled_partition` guards on `PartitionStatus::Pending`, so a late cancel arriving after the coordinator already saw the report is a harmless no-op. * Extract `combine_membership_and_bounds` so the CollectLeft and Partitioned arms of `SharedBuildAccumulator::build_filter` share a single (membership, bounds) -> predicate combinator. Partitioned keeps its `lit(true)` fallback via `unwrap_or_else`, preserving the existing CASE-branch shape optimizations. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
The `BuildReportState::ReportScheduled` → `Drop` → `report_canceled_partition` path is correct only because `store_canceled_partition` no-ops when the partition is already `Reported`. Previously this invariant was implicit; add unit tests so a future refactor that, say, unconditionally overwrites the partition status can't silently reintroduce the hang. Also pin that `report_canceled_partition` is idempotent — a stray double-drop must not double-count `completed_partitions`. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
kosiew
approved these changes
Apr 22, 2026
kosiew
left a comment
kosiew
left a comment
Contributor
There was a problem hiding this comment.
looks good to me
except a comment clean-up for Barrier.
Contributor
There was a problem hiding this comment.
This should be updated since this PR replaced the Barrier-based coordination.
The synchronization-strategy comment still described a barrier even though this PR replaced it with a per-partition report counter, a CompletionState lifecycle finalized exactly once, and a Notify that wakes parked waiters. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
Contributor
Author
|
Thanks for the review @kosiew ! @gabotechs I did try to use an existing primitive, but I couldn't quite get anything to fit the shape we need. If you or anyone else can improve it a followup is welcome, but I didn't want to hold up unblocking benchmarks. |
This was referenced May 29, 2026
adriangb
added a commit
to adriangb/datafusion
that referenced
this pull request
Nov 4, 2026
The hash join dynamic filter is the global filter of all build partitions: bounds AND an InList or a bloom filter, with a key-range pruning bitmap. The routed filter (`CASE hash_repartition(keys) % n` or a `RangeExpr` router over per-partition `HashTableLookupExpr`s) remained only as the fallback for a canceled partition. A parent cancels a child partition when it drops it early (see apache#21666). The keys of that partition are unknown, so no membership filter can be exact for it. `true` is correct for any consumer of the filter, including rows of the canceled partition, so the fallback is not worth its code. This commit: - Sets the dynamic filter to `true` when any partition is canceled. The coordinator still finalizes, so waiting partitions complete. - Removes the routed `CASE` / `RangeExpr` filter, `HashTableLookupExpr` and its re-exports, the crate-private `RangeExpr::route_partition_ids`, the `HashTableLookupExpr` pruning branch and its proto test. - Replaces `PushdownStrategy::Map(Arc<Map>, Option<Arc<KeyRangeBitmap>>)` with `PushdownStrategy::Large(Option<Arc<KeyRangeBitmap>>)`. The join keeps its own hash map. - Builds the collect-left filter only from the global filter. - Rewrites the `CASE` pruning test with range branches, and drops the unused `rstest` dev-dependency of `datafusion-pruning`. `HashExpr`, `SeededRandomState` and `RangeExpr` stay public with their proto support. Co-Authored-By: Claude Opus 5.5 <[email protected]>
adriangb
added a commit
to adriangb/datafusion
that referenced
this pull request
Nov 10, 2026
The hash join dynamic filter is the global filter of all build partitions: bounds AND an InList or a bloom filter, with a key-range pruning bitmap. The routed filter (`CASE hash_repartition(keys) % n` or a `RangeExpr` router over per-partition `HashTableLookupExpr`s) remained only as the fallback for a canceled partition. A parent cancels a child partition when it drops it early (see apache#21666). The keys of that partition are unknown, so no membership filter can be exact for it. `true` is correct for any consumer of the filter, including rows of the canceled partition, so the fallback is not worth its code. This commit: - Sets the dynamic filter to `true` when any partition is canceled. The coordinator still finalizes, so waiting partitions complete. - Removes the routed `CASE` / `RangeExpr` filter, `HashTableLookupExpr` and its re-exports, the crate-private `RangeExpr::route_partition_ids`, the `HashTableLookupExpr` pruning branch and its proto test. - Replaces `PushdownStrategy::Map(Arc<Map>, Option<Arc<KeyRangeBitmap>>)` with `PushdownStrategy::Large(Option<Arc<KeyRangeBitmap>>)`. The join keeps its own hash map. - Builds the collect-left filter only from the global filter. - Rewrites the `CASE` pruning test with range branches, and drops the unused `rstest` dev-dependency of `datafusion-pruning`. `HashExpr`, `SeededRandomState` and `RangeExpr` stay public with their proto support. Co-Authored-By: Claude Opus 5.5 <[email protected]>
Awesome06
pushed a commit
to Awesome06/datafusion
that referenced
this pull request
Nov 11, 2026
…c filter (apache#27087) ## Which issue does this PR close? - This PR does not close an issue. It is related to apache#21625 and follows up on apache#21666. ## Rationale for this change The partitioned hash join dynamic filter coordinator marks a partition as "canceled" when its stream is dropped before it reports its build data. A reader can assume that this means the whole query was canceled. That assumption is wrong. A parent operator can legally finish early and drop one hash join partition stream while sibling partitions keep running. In apache#21625 (the TPC-H Q18 hang), a `RightSemi` join with an empty build side for some partitions does exactly this. An editor who relies on the wrong reading could reintroduce that hang. ## What changes are included in this PR? Comment-only changes. The docs define a canceled partition by what a reader can observe: a parent operator drops one partition's stream before that stream reads its probe side. They avoid internal terms such as report delivery. - `BuildReportHandle` explains what cancellation means for the join. The other partitions and their shared probe input keep running, a `RepartitionExec` discards the rows routed to the canceled partition, and the canceled stream produces no output. The other partitions stop waiting for the canceled one. - `PartitionStatus::CanceledUnknown` explains only what it means for the dynamic filter. The partition's build keys are unknown, and the filter may keep or reject the rows routed to it, because no stream joins those rows. - `BuildReportState::Canceled` and `SharedBuildAccumulator::report_canceled_partition` point to those two docs. One em-dash in an existing comment is replaced. The last commit rewrites the earlier wording in plain terms, so reviewing the full diff is easier than reviewing commit by commit. ## What is the testing strategy for this PR? The changes are comment-only, so they add no tests. `cargo fmt` and `cargo clippy -p datafusion-physical-plan --all-targets --all-features -- -D warnings` pass. ## Are there any user-facing changes? No. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <[email protected]>
andygrove
pushed a commit
to andygrove/datafusion
that referenced
this pull request
Nov 12, 2026
…he#27092) ## Which issue does this PR close? - This PR does not close an issue. It follows up on apache#21666, which added canceled partitions to the hash join dynamic filter to fix apache#21625. ## Rationale for this change A `Partitioned` hash join cancels one of its partitions when a parent operator drops that partition's stream before it reports its build side. For example, a parent join partition whose own build side is empty drops its probe-side stream unpolled. The dynamic filter keeps every probe row routed to a canceled partition, because that partition's build keys are unknown. Keeping those rows buys nothing. A stream starts to probe only after its report is delivered, so a canceled stream has emitted no row and never will. The shared probe input still reads the rows routed to it, and `RepartitionExec` then discards them. Keeping them also costs something: - A canceled partition forces the routed `CASE` fallback instead of the merged lookup. After apache#27091 it is the only thing that does. - Under `NullEqualsNull`, a canceled partition makes every live partition keep probe-side NULL rows. ## What changes are included in this PR? Three commits: 1. `test:` Two dynamic filter tests checked range routing for `-0.0` and for composite nullable keys through a canceled partition's `CASE` branch. The commit moves both cases into `RangeRouter` tests, before the next commit removes that branch. 2. `fix:` A canceled partition counts as empty, so the filter rejects the rows routed to it. The commit rewrites the tests that asserted the old behavior. 3. `refactor:` The `CASE` fallback has no input left. The commit removes it, the per-partition filters it used, and the test assertions that a filter is not a `CASE`. When exactly one partition holds rows, the filter is still that partition's own IN list or hash lookup, so the probe scan keeps bloom filter pruning. 4. `fix:` Restores the branch for a lone non-empty partition, which the refactor first dropped. Adds a test for it and for a canceled partition next to live ones, and removes the remaining `CASE` wording. This is +155 / −345 lines. ## What is the testing strategy for this PR? - The new `RangeRouter` tests keep the routing cases that the removed tests covered. - `partitioned_canceled_and_empty_partitions_reject_every_row` and `partitioned_canceled_partition_does_not_widen_filter` cover the new behavior. - `test_partitioned_dynamic_filter_reports_empty_canceled_partitions`, the regression test for apache#21625, still passes. - All `datafusion-physical-plan` tests, all sqllogictests, and `core_integration` pass. I also checked the behavior on a query that cancels partitions. In `SELECT x.k FROM (SELECT big.k FROM mid JOIN big ON mid.k = big.k) x WHERE x.k IN (SELECT k FROM tiny)` with 4 partitions, the parent `RightSemi` join cancels two child partitions. The result is the same with and without this PR. ## Are there any user-facing changes? No. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <[email protected]>
philippemnoel
pushed a commit
to paradedb/datafusion
that referenced
this pull request
Nov 14, 2026
…apache#27345) ## Which issue does this PR close? - Related to apache#21625 (already closed by apache#21666). ## Rationale for this change apache#21666 fixed a hang in partitioned hash joins with dynamic filters (first seen as TPC-H Q18 hanging). Its only test is a Rust unit test of the coordinator. No SQL-level test covers the plan shape that triggers the hang, so a planner or execution change could bring it back unnoticed. ## What changes are included in this PR? One new file, `join_dynamic_filter_canceled_partition.slt`. It is a small, deterministic reproducer: - A `Partitioned` inner join sits on top of a `Partitioned` child join on the same key, with no repartition between them. - The parent build side holds only `k = 1`, so 3 of the 4 parent partitions have an empty build side. They complete without polling their probe side, so their child partitions are dropped before they report to the shared dynamic-filter coordinator. - The one non-empty parent partition polls its child partition, which waits for the dropped partitions. Before the fix, it waits forever. The hang does not depend on timing: the hash routing of `k = 1` is fixed, and a child partition reports only when its parent polls it. An `EXPLAIN` assertion locks the plan shape the reproducer needs. The file comments link to the issue and describe the sequence of events. ## What is the testing strategy for this PR? - The new test passes on `main`. - With the fix disabled (the `report_canceled_partition` call in `HashJoinStream::drop` replaced by a no-op), the test hangs. It never completed and was killed by a 500 s timeout. - With `datafusion-cli`, the same query hangs 5 of 5 times at `887595660^` and returns 2 rows 5 of 5 times at `887595660`. Note: a regression shows up as a hang, not a failed assertion, so CI would catch it by the job timeout. ## Are there any user-facing changes? No. Test-only. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5.5 <[email protected]>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Rationale for this change
Partitioned hash join dynamic filters assumed every build-side partition would eventually report build data to the shared coordinator. That assumption breaks when an upstream partitioned operator legally short-circuits and drops a child hash-join partition before it is ever polled far enough to report.
In the original reproducer, a parent
RightSemijoin completes early for partitions whose own build side is empty. That causes child partitioned hash-join streams to be dropped while still waiting to build/report their dynamic-filter contribution. Sibling partitions then wait forever for reports that will never arrive.What changes are included in this PR?
Droptruein the synthesized partitioned filter so they do not block completion or incorrectly filter probe rowsfalseAre these changes tested?
cargo fmt --allcargo test -p datafusion-physical-plan test_partitioned_dynamic_filter_reports_empty_canceled_partitions -- --nocapturecargo test -p datafusion --test core_integration physical_optimizer::filter_pushdown::test_hashjoin_dynamic_filter_pushdown_partitioned -- --nocapturecargo test -p datafusion --test core_integration physical_optimizer::filter_pushdown::test_hashjoin_dynamic_filter_pushdown_partitioned --features force_hash_collisions -- --nocapturetest_partitioned_dynamic_filter_reports_empty_canceled_partitionstimes out on the pre-fix revision and passes on this branchcargo clippy --all-targets --all-features -- -D warningsstill fails on an unrelated existing workspace lint indatafusion/expr/src/logical_plan/plan.rs:3773(clippy::mutable_key_type).Are there any user-facing changes?
No.