Skip to content

Add options to control hash join dynamic filter pushdown #19932

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
adriangb wants to merge 4 commits into
base: main
Choose a base branch
from
+60 −13
Open

Add options to control hash join dynamic filter pushdown #19932

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
6642819
Add options to control hash join dynamic filter pushdown
adriangb Jan 21, 2026
513aece
fmt
adriangb Jan 21, 2026
99b0103
docs
adriangb Jan 21, 2026
91ae9b0
allow disabling everything except routing
adriangb Jan 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions datafusion/common/src/config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1115,6 +1115,18 @@ config_namespace! {
/// See: <https://trino.io/docs/current/admin/dynamic-filtering.html#dynamic-filter-collection-thresholds>
pub hash_join_inlist_pushdown_max_distinct_values: usize, default = 150

/// When true, pushes down hash table references for membership checks in hash joins
/// when the build side is too large for InList pushdown.
/// When false, no membership filter is created when InList thresholds are exceeded.
Comment on lines +1118 to +1120
Copy link

Copilot AI Jan 21, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The documentation states "When false, no membership filter is created when InList thresholds are exceeded." This could be clearer. Consider rephrasing to: "When false, no hash table reference (Map) is used for membership checks. If the build side exceeds InList thresholds, no membership filter is created (only bounds, if enabled)." This makes it clearer that InList filters can still be created when the build side is small enough, and that this config specifically controls the Map fallback behavior.

Suggested change
/// When true, pushes down hash table references for membership checks in hash joins
/// when the build side is too large for InList pushdown.
/// When false, no membership filter is created when InList thresholds are exceeded.
/// When true, pushes down hash table references (Map) for membership checks in hash joins
/// when the build side is too large for InList pushdown.
/// When false, no hash table reference (Map) is used for membership checks. If the build
/// side exceeds InList thresholds, no membership filter is created (only bounds, if enabled).

Copilot uses AI. Check for mistakes.
/// Default: true
pub hash_join_map_pushdown: bool, default = true
Copy link
Contributor

@Dandandan Dandandan Jan 22, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Perhaps it is also good to differentiate between array map (contains_keys) vs hash map (contains_hashes). The first one should be much cheaper (no hashing / hash probing necessary).

Copy link
Contributor

@Dandandan Dandandan Jan 22, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(i.e. the one introduced in #19411)


/// When true, pushes down min/max bounds for join key columns.
/// This enables statistics-based pruning (e.g., Parquet row group skipping).
/// When false, only membership filters (InList or Map) are pushed down.
/// Default: true
pub hash_join_bounds_pushdown: bool, default = true
Comment on lines +1122 to +1128
Copy link

Copilot AI Jan 21, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The new configuration options hash_join_map_pushdown and hash_join_bounds_pushdown lack test coverage. While the PR description mentions existing hash join tests pass, there are no specific tests that validate the behavior when these options are set to false. Consider adding tests similar to the existing test_hashjoin_hash_table_pushdown_partitioned that verify:

  1. When hash_join_map_pushdown is false and build side exceeds InList thresholds, no membership filter is created (only bounds if enabled)
  2. When hash_join_bounds_pushdown is false, bounds predicates are not included in the filter
  3. When both are false but dynamic filter pushdown is enabled, appropriate behavior occurs
  4. Various combinations of these flags work correctly in both Partitioned and CollectLeft modes

Copilot uses AI. Check for mistakes.

/// The default filter selectivity used by Filter Statistics
/// when an exact selectivity cannot be determined. Valid values are
/// between 0 (no selectivity) and 100 (all rows are selected).
Expand Down
19 changes: 16 additions & 3 deletions datafusion/physical-plan/src/joins/hash_join/exec.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1152,6 +1152,11 @@ impl ExecutionPlan for HashJoinExec {
// Initialize build_accumulator lazily with runtime partition counts (only if enabled)
// Use RepartitionExec's random state (seeds: 0,0,0,0) for partition routing
let repartition_random_state = REPARTITION_RANDOM_STATE;
let bounds_pushdown_enabled = context
.session_config()
.options()
.optimizer
.hash_join_bounds_pushdown;
let build_accumulator = enable_dynamic_filter_pushdown
.then(|| {
self.dynamic_filter.as_ref().map(|df| {
Expand All @@ -1169,6 +1174,7 @@ impl ExecutionPlan for HashJoinExec {
filter,
on_right,
repartition_random_state,
bounds_pushdown_enabled,
))
})))
})
Expand Down Expand Up @@ -1702,7 +1708,8 @@ async fn collect_left_input(
PushdownStrategy::Empty
} else {
// If the build side is small enough we can use IN list pushdown.
// If it's too big we fall back to pushing down a reference to the hash table.
// If it's too big we fall back to pushing down a reference to the hash table
// (if map pushdown is enabled).
// See `PushdownStrategy` for more details.
let estimated_size = left_values
.iter()
Expand All @@ -1716,11 +1723,17 @@ async fn collect_left_input(
.optimizer
.hash_join_inlist_pushdown_max_distinct_values
{
PushdownStrategy::Map(Arc::clone(&map))
if config.optimizer.hash_join_map_pushdown {
PushdownStrategy::Map(Arc::clone(&map))
} else {
PushdownStrategy::Disabled
}
} else if let Some(in_list_values) = build_struct_inlist_values(&left_values)? {
PushdownStrategy::InList(in_list_values)
} else {
} else if config.optimizer.hash_join_map_pushdown {
PushdownStrategy::Map(Arc::clone(&map))
} else {
PushdownStrategy::Disabled
}
};

Expand Down
40 changes: 30 additions & 10 deletions datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,6 +136,8 @@ fn create_membership_predicate(
)) as Arc<dyn PhysicalExpr>)),
// Empty partition - should not create a filter for this
PushdownStrategy::Empty => Ok(None),
// User disabled map pushdown - only bounds (if enabled) will be used
PushdownStrategy::Disabled => Ok(None),
}
}

Expand Down Expand Up @@ -226,6 +228,8 @@ pub(crate) struct SharedBuildAccumulator {
repartition_random_state: SeededRandomState,
/// Schema of the probe (right) side for evaluating filter expressions
probe_schema: Arc<Schema>,
/// Whether to include bounds in the dynamic filter
bounds_pushdown_enabled: bool,
}

/// Strategy for filter pushdown (decided at collection time)
Expand All @@ -237,6 +241,8 @@ pub(crate) enum PushdownStrategy {
Map(Arc<Map>),
/// There was no data in this partition, do not build a dynamic filter for it
Empty,
/// User disabled map pushdown via config, only use bounds (if enabled)
Disabled,
}

/// Build-side data reported by a single partition
Expand Down Expand Up @@ -302,6 +308,7 @@ impl SharedBuildAccumulator {
dynamic_filter: Arc<DynamicFilterPhysicalExpr>,
on_right: Vec<PhysicalExprRef>,
repartition_random_state: SeededRandomState,
bounds_pushdown_enabled: bool,
) -> Self {
// Troubleshooting: If partition counts are incorrect, verify this logic matches
// the actual execution pattern in collect_build_side()
Expand Down Expand Up @@ -342,6 +349,7 @@ impl SharedBuildAccumulator {
on_right,
repartition_random_state,
probe_schema: right_child.schema(),
bounds_pushdown_enabled,
}
}

Expand Down Expand Up @@ -410,11 +418,15 @@ impl SharedBuildAccumulator {
self.probe_schema.as_ref(),
)?;

// Create bounds check expression (if bounds available)
let bounds_expr = create_bounds_predicate(
&self.on_right,
&partition_data.bounds,
);
// Create bounds check expression (if bounds available and enabled)
let bounds_expr = if self.bounds_pushdown_enabled {
create_bounds_predicate(
&self.on_right,
&partition_data.bounds,
)
} else {
None
};

// Combine membership and bounds expressions for multi-layer optimization:
// - Bounds (min/max): Enable statistics-based pruning (Parquet row group/file skipping)
Expand Down Expand Up @@ -490,6 +502,7 @@ impl SharedBuildAccumulator {
as Arc<dyn PhysicalExpr>;

// Create WHEN branches for each partition
let bounds_enabled = self.bounds_pushdown_enabled;
let when_then_branches: Vec<(
Arc<dyn PhysicalExpr>,
Arc<dyn PhysicalExpr>,
Expand All @@ -501,6 +514,9 @@ impl SharedBuildAccumulator {
// Skip empty partitions - they would always return false anyway
match &partition.pushdown {
PushdownStrategy::Empty => None,
// Note: We intentionally keep Disabled partitions even when bounds
// are disabled. This allows measuring the CASE routing overhead
// independently. The (None, None) case returns lit(true).
_ => Some((partition_id, partition)),
}
})
Expand All @@ -520,11 +536,15 @@ impl SharedBuildAccumulator {
self.probe_schema.as_ref(),
)?;

// 2. Create bounds check expression for this partition (if bounds available)
let bounds_expr = create_bounds_predicate(
&self.on_right,
&partition.bounds,
);
// 2. Create bounds check expression for this partition (if bounds available and enabled)
let bounds_expr = if self.bounds_pushdown_enabled {
create_bounds_predicate(
&self.on_right,
&partition.bounds,
)
} else {
None
};

// 3. Combine membership and bounds expressions
let then_expr = match (membership_expr, bounds_expr) {
Expand Down
2 changes: 2 additions & 0 deletions docs/source/user-guide/configs.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -161,6 +161,8 @@ The following configuration settings are available:
| datafusion.optimizer.hash_join_single_partition_threshold_rows | 131072 | The maximum estimated size in rows for one input side of a HashJoin will be collected into a single partition |
| datafusion.optimizer.hash_join_inlist_pushdown_max_size | 131072 | Maximum size in bytes for the build side of a hash join to be pushed down as an InList expression for dynamic filtering. Build sides larger than this will use hash table lookups instead. Set to 0 to always use hash table lookups. InList pushdown can be more efficient for small build sides because it can result in better statistics pruning as well as use any bloom filters present on the scan side. InList expressions are also more transparent and easier to serialize over the network in distributed uses of DataFusion. On the other hand InList pushdown requires making a copy of the data and thus adds some overhead to the build side and uses more memory. This setting is per-partition, so we may end up using `hash_join_inlist_pushdown_max_size` \* `target_partitions` memory. The default is 128kB per partition. This should allow point lookup joins (e.g. joining on a unique primary key) to use InList pushdown in most cases but avoids excessive memory usage or overhead for larger joins. |
| datafusion.optimizer.hash_join_inlist_pushdown_max_distinct_values | 150 | Maximum number of distinct values (rows) in the build side of a hash join to be pushed down as an InList expression for dynamic filtering. Build sides with more rows than this will use hash table lookups instead. Set to 0 to always use hash table lookups. This provides an additional limit beyond `hash_join_inlist_pushdown_max_size` to prevent very large IN lists that might not provide much benefit over hash table lookups. This uses the deduplicated row count once the build side has been evaluated. The default is 150 values per partition. This is inspired by Trino's `max-filter-keys-per-column` setting. See: <https://trino.io/docs/current/admin/dynamic-filtering.html#dynamic-filter-collection-thresholds> |
| datafusion.optimizer.hash_join_map_pushdown | true | When true, pushes down hash table references for membership checks in hash joins when the build side is too large for InList pushdown. When false, no membership filter is created when InList thresholds are exceeded. Default: true |
| datafusion.optimizer.hash_join_bounds_pushdown | true | When true, pushes down min/max bounds for join key columns. This enables statistics-based pruning (e.g., Parquet row group skipping). When false, only membership filters (InList or Map) are pushed down. Default: true |
| datafusion.optimizer.default_filter_selectivity | 20 | The default filter selectivity used by Filter Statistics when an exact selectivity cannot be determined. Valid values are between 0 (no selectivity) and 100 (all rows are selected). |
| datafusion.optimizer.prefer_existing_union | false | When set to true, the optimizer will not attempt to convert Union to Interleave |
| datafusion.optimizer.expand_views_at_output | false | When set to true, if the returned type is a view type then the output will be coerced to a non-view. Coerces `Utf8View` to `LargeUtf8`, and `BinaryView` to `LargeBinary`. |
Expand Down
Loading