Repository navigation
Conversation
…er keys No hj query runs a `Partitioned` join on integer keys whose build side is large enough for a hash table or an `ArrayMap` per partition. So the suite does not measure how the build map type changes such a join. This adds two queries over TPC-H `orders`, joined to itself on `o_custkey` in `Partitioned` mode. Q26 builds from the orders of the first 40K customers: dense keys, about 10 rows per key. Q27 builds from customers whose key is a multiple of 37: sparse keys. Both probe 15M rows with about a 2.7% hit rate. The runner now also registers `orders`. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
A `Partitioned` hash join merges its partitions' build maps into one routed `hash_lookup` filter. The routed lookup accepted only hash tables, so a join whose partitions built an `ArrayMap` (dense integer keys) fell back to a `CASE` over one filter per partition. The routed lookup now probes an `ArrayMap` partition with the key values of the rows routed to it, because an `ArrayMap` is keyed by value, not hash. Every non-empty partition can now merge. The `CASE` fallback remains only for canceled partitions. The routed lookup fuzz test now builds an `ArrayMap` for about half of the partitions. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Two hash join dynamic filter tests checked these range routing cases through the `CASE` branch of a canceled partition. The next commit makes a canceled partition reject its rows, which leaves those tests nothing to route. This adds both cases as `RangeRouter` tests, where they belong: a `0.0` split point separates `-0.0` from `0.0`, and each key of a composite key places its NULLs by its own `nulls_first`. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
A `Partitioned` hash join marks a partition canceled when its stream is dropped before it reports its build data. The dynamic filter kept every probe row routed to a canceled partition, because the partition's build keys were unknown. That choice was safe but bought nothing: a stream probes only after its report is delivered, so a canceled stream has emitted no row and never will. Every row routed to it is discarded either way. It also cost something. A canceled partition forced the routed `CASE` fallback instead of the merged lookup. Under `NullEqualsNull` it also made every live partition keep probe-side NULL rows. A canceled partition now counts as empty. The filter drops the flag that tracked cancellation and the `ELSE true` it fed. The `CASE` fallback is now unreachable. The next commit removes it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…ilter The partitioned dynamic filter built a `CASE` over one filter per partition when a partition could not merge into the routed lookup. After the previous two commits every partition merges: an `ArrayMap` is routed, and a canceled partition counts as empty. The `CASE` builder, its per-partition filters, and the single-partition branch had no input left to handle. One non-empty partition already becomes a direct lookup in the merged path. This removes that code and the test assertions that a filter is not a `CASE`. No behavior changes. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…itioned-filter-cancel-as-empty # Conflicts: # datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs # datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs
jayzhan211
requested changes
Nov 6, 2026
jayzhan211
left a comment
jayzhan211
left a comment
Contributor
There was a problem hiding this comment.
@adriangb thanks for this! Treating a canceled partition as empty looks correct to me: I compared results against the filter turned off and against main on queries that cancel partitions, and found no differences.
The one thing that needs to change is the single-partition branch, covered inline. Could you also update the "No behavior changes" line in the PR description once that's settled?
Optional cleanup: a few comments and names still describe the CASE: shared_bounds.rs around lines 225 and 487, exec.rs around line 10021, and the test names partitioned_hash_map_partitions_route_without_case and partitioned_one_real_partition_with_rest_empty_skips_case.
With every partition merged into one routed lookup, a build where only one partition holds rows lost its IN list. The merged IN-list limit counts duplicate rows, so one key with many build rows became a hash_lookup and the probe scan lost bloom filter pruning. This restores the branch that uses that partition's own filter, and adds tests for it and for a canceled partition next to live ones. It also drops the remaining comments and test names that describe the removed CASE, and documents why HashExpr stays. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
jayzhan211
previously approved these changes
Nov 7, 2026
jayzhan211
dismissed
their stale review
November 7, 2026 03:24
Holding this approval until #27091 merges, because this branch includes its commit.
…itioned-filter-cancel-as-empty # Conflicts: # datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs # datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs
Contributor
Author
|
@jayzhan211 #27091 merged now :) |
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
A
Partitionedhash 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
RepartitionExecthen discards them. Keeping them also costs something:CASEfallback instead of the merged lookup. After perf: route probe rows to ArrayMap build partitions in the merged hash join filter #27091 it is the only thing that does.NullEqualsNull, a canceled partition makes every live partition keep probe-side NULL rows.What changes are included in this PR?
Three commits:
test:Two dynamic filter tests checked range routing for-0.0and for composite nullable keys through a canceled partition'sCASEbranch. The commit moves both cases intoRangeRoutertests, before the next commit removes that branch.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.refactor:TheCASEfallback has no input left. The commit removes it, the per-partition filters it used, and the test assertions that a filter is not aCASE. 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.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 remainingCASEwording.This is +155 / −345 lines.
What is the testing strategy for this PR?
RangeRoutertests keep the routing cases that the removed tests covered.partitioned_canceled_and_empty_partitions_reject_every_rowandpartitioned_canceled_partition_does_not_widen_filtercover the new behavior.test_partitioned_dynamic_filter_reports_empty_canceled_partitions, the regression test for bug: TPCH 18 query hangs #21625, still passes.datafusion-physical-plantests, all sqllogictests, andcore_integrationpass.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 parentRightSemijoin 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