Skip to content

Commit 3ca3cda

Browse files
committed
perf(hash-join): skip dynamic filter for RightSemi when pushdown_filters=false
Right-Semi joins already implement the exact 'probe key ∈ build set' check as their core operation via the downstream hash lookup. Applying the bounds-based dynamic filter to those probe rows is semantically redundant, and — without arrow-rs RowFilter's lazy-decode amortization (i.e. pushdown_filters=false) — the per-batch PostScanFilter tax exceeds the marginal hash-lookup savings on TPC-H-shaped workloads. The #23701 hj benchmark surfaced this: density=1 RightSemi cases (Q16 25×1.5M, Q17/Q18 100K×60M, Q22 fanout=100 100K×60M) showed 5–35% regression because the bounds filter prunes 0 rows while the filter still evaluates on every probe row. Skip the dynamic filter for RightSemi when pushdown_filters=false. Keep it when pushdown_filters=true so RowFilter's two-phase decode can still amortize the cost on selective Semi joins. RightAnti is already excluded by the existing probe_preserved gate (on_lr_is_preserved reports (true, false) for it), so this new gate is redundant for that type — no separate check needed. Tests: - test_dynamic_filter_pushdown_rejects_right_semi_when_pushdown_off - test_dynamic_filter_pushdown_keeps_right_semi_when_pushdown_on
1 parent 6314f09 commit 3ca3cda

1 file changed

Lines changed: 115 additions & 0 deletions

File tree

  • datafusion/physical-plan/src/joins/hash_join

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

Lines changed: 115 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -892,6 +892,27 @@ impl HashJoinExec {
892892
return false;
893893
}
894894

895+
// Right-Semi / Right-Anti joins already implement the exact
896+
// "probe key ∈ build set" (or its complement) as their core
897+
// operation via the downstream hash lookup. Applying the
898+
// bounds-based dynamic filter to those probe rows is
899+
// semantically redundant, and — without arrow-rs RowFilter's
900+
// lazy-decode amortization (i.e. `pushdown_filters=false`) —
901+
// the per-batch `PostScanFilter` tax typically exceeds the
902+
// marginal hash-lookup savings on TPC-H-shaped workloads
903+
// (see the #23701 `hj` benchmark: the density=1 Semi/Anti
904+
// cases all show 5–35% regression because bounds prune 0 rows
905+
// yet the filter still evaluates on every probe row).
906+
//
907+
// Keep the filter when `pushdown_filters=true` so RowFilter's
908+
// two-phase decode can still amortize the cost on selective
909+
// Semi/Anti joins.
910+
if !config.execution.parquet.pushdown_filters
911+
&& matches!(self.join_type, JoinType::RightSemi | JoinType::RightAnti)
912+
{
913+
return false;
914+
}
915+
895916
true
896917
}
897918

@@ -6837,6 +6858,100 @@ mod tests {
68376858
Ok(())
68386859
}
68396860

6861+
/// `RightSemi` + `pushdown_filters=false` must skip the dynamic
6862+
/// filter — the join's own hash lookup already provides exact
6863+
/// "in build set" filtering, and without `RowFilter`'s lazy-decode
6864+
/// amortization the `PostScanFilter` per-batch evaluation is pure
6865+
/// tax on the probe scan. See #23701 hj benchmark: the density=1
6866+
/// Semi cases all show 5–35% regression because bounds prune 0
6867+
/// rows while the filter still evaluates on every probe row.
6868+
///
6869+
/// (`RightAnti` is already excluded by the general `probe_preserved`
6870+
/// gate — `on_lr_is_preserved` reports `(true, false)` for it — so
6871+
/// this new gate is redundant for that type.)
6872+
#[test]
6873+
fn test_dynamic_filter_pushdown_rejects_right_semi_when_pushdown_off() -> Result<()> {
6874+
let (_, _, on) = build_schema_and_on()?;
6875+
let left = build_table(("a1", &vec![1]), ("b1", &vec![1]), ("c1", &vec![1]));
6876+
let right = build_table(("a2", &vec![1]), ("b1", &vec![1]), ("c2", &vec![1]));
6877+
6878+
let mut session_config = SessionConfig::default();
6879+
session_config
6880+
.options_mut()
6881+
.optimizer
6882+
.enable_join_dynamic_filter_pushdown = true;
6883+
session_config
6884+
.options_mut()
6885+
.execution
6886+
.parquet
6887+
.pushdown_filters = false;
6888+
6889+
let join = HashJoinExec::try_new(
6890+
left,
6891+
right,
6892+
on,
6893+
None,
6894+
&JoinType::RightSemi,
6895+
None,
6896+
PartitionMode::CollectLeft,
6897+
NullEquality::NullEqualsNothing,
6898+
false,
6899+
)?;
6900+
assert!(
6901+
!join.allow_join_dynamic_filter_pushdown(session_config.options()),
6902+
"RightSemi + pushdown_filters=false must skip dynamic filter",
6903+
);
6904+
6905+
Ok(())
6906+
}
6907+
6908+
/// `RightSemi` + `pushdown_filters=true` must **keep** the dynamic
6909+
/// filter — arrow-rs `RowFilter`'s two-phase decode amortizes the
6910+
/// per-row cost by skipping the decode of non-filter columns for
6911+
/// filtered rows, which can pay for the bounds check even when the
6912+
/// join type is Semi.
6913+
///
6914+
/// (`RightAnti` is excluded here because it drops the probe side
6915+
/// entirely from the output — `on_lr_is_preserved` returns
6916+
/// `(true, false)`, so the general `probe_preserved` gate already
6917+
/// disables dynamic filter pushdown for `RightAnti` regardless of
6918+
/// the new `pushdown_filters=false` check.)
6919+
#[test]
6920+
fn test_dynamic_filter_pushdown_keeps_right_semi_when_pushdown_on() -> Result<()> {
6921+
let (_, _, on) = build_schema_and_on()?;
6922+
let left = build_table(("a1", &vec![1]), ("b1", &vec![1]), ("c1", &vec![1]));
6923+
let right = build_table(("a2", &vec![1]), ("b1", &vec![1]), ("c2", &vec![1]));
6924+
6925+
let mut session_config = SessionConfig::default();
6926+
session_config
6927+
.options_mut()
6928+
.optimizer
6929+
.enable_join_dynamic_filter_pushdown = true;
6930+
session_config
6931+
.options_mut()
6932+
.execution
6933+
.parquet
6934+
.pushdown_filters = true;
6935+
6936+
let join = HashJoinExec::try_new(
6937+
left,
6938+
right,
6939+
on,
6940+
None,
6941+
&JoinType::RightSemi,
6942+
None,
6943+
PartitionMode::CollectLeft,
6944+
NullEquality::NullEqualsNothing,
6945+
false,
6946+
)?;
6947+
assert!(
6948+
join.allow_join_dynamic_filter_pushdown(session_config.options()),
6949+
"RightSemi + pushdown_filters=true must keep dynamic filter",
6950+
);
6951+
6952+
Ok(())
6953+
}
6954+
68406955
#[test]
68416956
fn test_with_dynamic_filter_rejects_invalid_columns() -> Result<()> {
68426957
let (_, _, on) = build_schema_and_on()?;

0 commit comments

Comments
 (0)