Skip to content

Commit bf00cbf

Browse files
committed
perf(hash-join): gate union-bounds fast path on pushdown_filters=false
Previously the union-bounds fast path fired whenever enable_hash_join_dynamic_membership_filter=false (the production default). But that also relaxes the filter for pushdown_filters=true, where arrow-rs RowFilter amortizes the per-row CASE hash-routed cost via lazy decode of the remaining columns for filtered rows — the tighter per-partition selectivity is worth paying for there. This commit couples the fast path to the runtime pushdown_filters setting: the SharedBuildAccumulator now mirrors execution.parquet.pushdown_filters and picks the shape based on where the filter will run: * pushdown_filters=false → probe scan applies the filter post-decode (PostScanFilter, Layer 3). Per-row cost dominates because there is no lazy decode to amortize it. Emit the cheap union-of-bounds predicate and skip membership entirely. * pushdown_filters=true → probe scan applies the filter via arrow-rs RowFilter (Layer 2), which skips the decode of the remaining columns for rows that fail the predicate. Emit the historical per-partition CASE hash-routed form; respect the user's opt-in via enable_hash_join_dynamic_membership_filter for membership. Effect matrix: pushdown_filters | enable_membership_filter | Partitioned filter shape -----------------+--------------------------+-------------------------------- false (default) | any | union bounds (fast path) true | false | CASE hash-routed, bounds only true | true | CASE hash-routed + membership Tests: - New helper make_partitioned_accumulator_pushdown_on_and_membership_{on,off}. - New unit test partitioned_pushdown_on_membership_off_still_emits_case_no_membership — pushdown=true must keep CASE even with membership off. - New unit test collect_left_pushdown_off_skips_membership_even_with_gate_on — pushdown=false must skip membership even with opt-in. - Existing test fixtures updated to include pushdown_filters explicitly (existing coverage kept — bounds-only fixtures set pushdown_filters=false to match production default; membership-on fixtures set pushdown_filters=true). Also refreshes projection_pushdown.slt for an unrelated main-branch plan-shape change (#23459 CSE tweak) that surfaced post-rebase.
1 parent c36534b commit bf00cbf

3 files changed

Lines changed: 326 additions & 21 deletions

File tree

‎datafusion/physical-plan/src/joins/hash_join/exec.rs‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1393,6 +1393,18 @@ impl ExecutionPlan for HashJoinExec {
13931393
.options()
13941394
.optimizer
13951395
.enable_hash_join_dynamic_membership_filter;
1396+
// Mirror the probe scan's parquet `pushdown_filters` config into the
1397+
// build coordinator so it can pick the cheap union-of-bounds shape
1398+
// when the probe scan applies the filter post-decode (Layer 3), and
1399+
// the historical per-partition CASE hash-routing when the probe
1400+
// scan applies it via arrow-rs `RowFilter` with lazy decode
1401+
// (Layer 2). See #23701.
1402+
let pushdown_filters = context
1403+
.session_config()
1404+
.options()
1405+
.execution
1406+
.parquet
1407+
.pushdown_filters;
13961408
let build_accumulator = enable_dynamic_filter_pushdown
13971409
.then(|| {
13981410
self.dynamic_filter.as_ref().map(|df| {
@@ -1412,6 +1424,7 @@ impl ExecutionPlan for HashJoinExec {
14121424
repartition_random_state,
14131425
self.null_aware,
14141426
enable_membership_filter,
1427+
pushdown_filters,
14151428
))
14161429
})))
14171430
})

0 commit comments

Comments
 (0)