From 118fd470f306f87616468ff58060b71f0e761bdb Mon Sep 17 00:00:00 2001 From: Qiwei Huang Date: Sun, 26 Jul 2026 14:25:18 +0800 Subject: [PATCH 1/4] perf: elide redundant HashJoin dynamic-filter membership predicates --- .../physical-plan/src/joins/hash_join/exec.rs | 36 +- .../src/joins/hash_join/shared_bounds.rs | 457 ++++++++++++++++-- .../test_files/push_down_filter_parquet.slt | 20 +- 3 files changed, 448 insertions(+), 65 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 9d9c867c2724b..8e4ae5cca74fd 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -35,7 +35,8 @@ use crate::joins::Map; use crate::joins::array_map::ArrayMap; use crate::joins::hash_join::inlist_builder::build_struct_inlist_values; use crate::joins::hash_join::shared_bounds::{ - ColumnBounds, PartitionBounds, PushdownStrategy, SharedBuildAccumulator, + ColumnBounds, DynamicFilterBuildContext, InListMembership, PartitionBounds, + PushdownStrategy, SharedBuildAccumulator, }; use crate::joins::hash_join::stream::{ BuildSide, BuildSideInitialState, HashJoinStream, HashJoinStreamState, @@ -102,6 +103,8 @@ pub(crate) const HASH_JOIN_SEED: SeededRandomState = SeededRandomState::with_seed(12210250226015887276); const ARRAY_MAP_CREATED_COUNT_METRIC_NAME: &str = "array_map_created_count"; +const DYNAMIC_FILTER_MEMBERSHIP_PREDICATES_ELIDED_METRIC_NAME: &str = + "dynamic_filter_membership_predicates_elided"; #[expect(clippy::too_many_arguments)] fn try_create_array_map( @@ -1404,14 +1407,24 @@ impl ExecutionPlan for HashJoinExec { .map(|(_, right_expr)| Arc::clone(right_expr)) .collect::>(); Some(Arc::clone(df.build_accumulator.get_or_init(|| { + let membership_predicates_elided = MetricBuilder::new( + &self.metrics, + ) + .with_category(MetricCategory::Rows) + .global_counter( + DYNAMIC_FILTER_MEMBERSHIP_PREDICATES_ELIDED_METRIC_NAME, + ); Arc::new(SharedBuildAccumulator::new_from_partition_mode( self.mode, self.left.as_ref(), self.right.as_ref(), - filter, - on_right, - repartition_random_state, - self.null_aware, + DynamicFilterBuildContext::new( + filter, + on_right, + repartition_random_state, + self.null_aware, + membership_predicates_elided, + ), )) }))) }) @@ -2326,7 +2339,7 @@ async fn collect_left_input( { PushdownStrategy::Map(Arc::clone(&map)) } else if let Some(in_list_values) = build_struct_inlist_values(&left_values)? { - PushdownStrategy::InList(in_list_values) + PushdownStrategy::InList(InListMembership::new(in_list_values, map.as_ref())) } else { PushdownStrategy::Map(Arc::clone(&map)) } @@ -6155,6 +6168,17 @@ mod tests { // After the join completes, the dynamic filter should be marked as complete // wait_complete() should return immediately dynamic_filter.wait_complete().await; + assert_eq!( + join.metrics() + .and_then(|metrics| { + metrics.sum_by_name( + DYNAMIC_FILTER_MEMBERSHIP_PREDICATES_ELIDED_METRIC_NAME, + ) + }) + .map(|value| value.as_usize()), + Some(1), + "the contiguous build domain should elide one membership predicate" + ); Ok(()) } diff --git a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs index 1fa06b5c6ca23..fec34918ccc4e 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -23,13 +23,14 @@ use std::sync::Arc; use crate::ExecutionPlan; use crate::ExecutionPlanProperties; -use crate::joins::Map; use crate::joins::PartitionMode; use crate::joins::hash_join::exec::HASH_JOIN_SEED; use crate::joins::hash_join::inlist_builder::build_struct_fields; use crate::joins::hash_join::partitioned_hash_eval::{ HashExpr, HashTableLookupExpr, SeededRandomState, }; +use crate::joins::{ArrayMap, Map}; +use crate::metrics::Count; use arrow::array::ArrayRef; use arrow::datatypes::{DataType, Field, Schema}; use datafusion_common::config::ConfigOptions; @@ -79,21 +80,18 @@ impl PartitionBounds { } } -/// Creates a membership predicate for filter pushdown. -/// -/// If `inlist_values` is provided (for small build sides), creates an InList expression. -/// Otherwise, creates a HashTableLookup expression (for large build sides). +/// Creates a membership predicate using the representation selected while +/// collecting the build side. /// /// Supports both single-column and multi-column joins using struct expressions. fn create_membership_predicate( on_right: &[PhysicalExprRef], - pushdown: PushdownStrategy, + pushdown: &PushdownStrategy, random_state: &SeededRandomState, schema: &Schema, ) -> Result>> { match pushdown { - // Use InList expression for small build sides - PushdownStrategy::InList(in_list_array) => { + PushdownStrategy::InList(membership) => { // Build the expression to compare against let expr = if on_right.len() == 1 { // Single column: col IN (val1, val2, ...) @@ -123,19 +121,17 @@ fn create_membership_predicate( // Use InListExpr::try_new_from_array() to build an InList with static_filter optimization (hash-based lookup) Ok(Some(Arc::new(InListExpr::try_new_from_array( expr, - in_list_array, + Arc::clone(&membership.values), false, schema, )?))) } - // Use hash table lookup for large build sides PushdownStrategy::Map(hash_map) => Ok(Some(Arc::new(HashTableLookupExpr::new( on_right.to_vec(), random_state.clone(), - hash_map, + Arc::clone(hash_map), "hash_lookup".to_string(), )) as Arc)), - // Empty partition - should not create a filter for this PushdownStrategy::Empty => Ok(None), } } @@ -203,6 +199,55 @@ fn combine_membership_and_bounds( } } +fn inclusive_integer_span(bounds: &PartitionBounds) -> Option { + let [column_bounds] = bounds.column_bounds.as_slice() else { + return None; + }; + if column_bounds.min.data_type() != column_bounds.max.data_type() + || column_bounds.min > column_bounds.max + { + return None; + } + let min = ArrayMap::key_to_u64(&column_bounds.min)?; + let max = ArrayMap::key_to_u64(&column_bounds.max)?; + + // ArrayMap uses two's-complement values and wrapping subtraction, which + // also gives the correct span for signed ranges that cross zero. + Some(u128::from(ArrayMap::calculate_range(min, max)) + 1) +} + +/// Returns true when the membership set is provably identical to its +/// single-column integer bounds. +/// +/// `distinct_key_count_lower_bound` is exact for [`ArrayMap`] and is the number of +/// distinct hashes for the regular hash map. The latter is a lower bound on +/// distinct keys. Because the keys are also bounded by the inclusive integer +/// span, equality with the span proves that every value is present even when +/// hash collisions are possible. +fn membership_matches_integer_bounds( + pushdown: &PushdownStrategy, + bounds: &PartitionBounds, + join_key_count: usize, +) -> bool { + if join_key_count != 1 { + return false; + } + let Some(distinct_key_count_lower_bound) = pushdown.distinct_key_count_lower_bound() + else { + return false; + }; + let Some(inclusive_span) = inclusive_integer_span(bounds) else { + return false; + }; + if let PushdownStrategy::InList(membership) = pushdown + && membership.values.data_type() != &bounds.column_bounds[0].min.data_type() + { + return false; + } + + inclusive_span == distinct_key_count_lower_bound as u128 +} + /// Coordinates build-side information collection across multiple partitions /// /// This structure collects information from the build side (hash tables and/or bounds) and @@ -258,19 +303,84 @@ pub(crate) struct SharedBuildAccumulator { /// Null-aware anti join (`NOT IN`). A probe-side NULL must reach the join so its /// three-valued logic can collapse the result, so the pushed filter keeps NULL rows. null_aware: bool, + /// Membership predicates omitted after proving their integer keys cover + /// every value in the corresponding min/max bounds. + membership_predicates_elided: Count, +} + +/// Runtime dependencies used to construct a HashJoin dynamic filter. +pub(super) struct DynamicFilterBuildContext { + filter: Arc, + probe_exprs: Vec, + repartition_random_state: SeededRandomState, + null_aware: bool, + membership_predicates_elided: Count, +} + +impl DynamicFilterBuildContext { + pub(super) fn new( + filter: Arc, + probe_exprs: Vec, + repartition_random_state: SeededRandomState, + null_aware: bool, + membership_predicates_elided: Count, + ) -> Self { + Self { + filter, + probe_exprs, + repartition_random_state, + null_aware, + membership_predicates_elided, + } + } +} + +/// Data required to build an `InListExpr`, together with cardinality evidence +/// derived from the same build-side key set. +#[derive(Clone)] +pub(crate) struct InListMembership { + values: ArrayRef, + /// A lower bound on the number of distinct matchable keys. This is exact + /// for an ArrayMap and may be conservative for a hash map due to collisions. + distinct_key_count_lower_bound: Option, +} + +impl InListMembership { + /// `map` must index the same build-side rows represented by `values`. + pub(super) fn new(values: ArrayRef, map: &Map) -> Self { + let distinct_key_count_lower_bound = map.num_of_distinct_key(); + Self { + // A count larger than the source array cannot describe the same + // build set, so discard the evidence and retain membership. + distinct_key_count_lower_bound: (distinct_key_count_lower_bound + <= values.len()) + .then_some(distinct_key_count_lower_bound), + values, + } + } } /// Strategy for filter pushdown (decided at collection time) #[derive(Clone)] pub(crate) enum PushdownStrategy { - /// Use InList for small build sides (< 128MB) - InList(ArrayRef), - /// Use map lookup for large build sides + /// Use InList when the configured size and cardinality limits allow it. + InList(InListMembership), + /// Reuse the build map when an InList is not selected. Map(Arc), /// There was no data in this partition, do not build a dynamic filter for it Empty, } +impl PushdownStrategy { + fn distinct_key_count_lower_bound(&self) -> Option { + match self { + Self::InList(membership) => membership.distinct_key_count_lower_bound, + Self::Map(map) => Some(map.num_of_distinct_key()), + Self::Empty => None, + } + } +} + /// Build-side data reported by a single partition pub(crate) enum PartitionBuildData { Partitioned { @@ -354,14 +464,11 @@ impl SharedBuildAccumulator { /// We cannot build a partial filter from some partitions - it would incorrectly eliminate /// valid join results. We must wait until we have complete information from ALL /// relevant partitions before updating the dynamic filter. - pub(crate) fn new_from_partition_mode( + pub(super) fn new_from_partition_mode( partition_mode: PartitionMode, left_child: &dyn ExecutionPlan, right_child: &dyn ExecutionPlan, - dynamic_filter: Arc, - on_right: Vec, - repartition_random_state: SeededRandomState, - null_aware: bool, + context: DynamicFilterBuildContext, ) -> Self { // Troubleshooting: If partition counts are incorrect, verify this logic matches // the actual execution pattern in collect_build_side() @@ -398,20 +505,54 @@ impl SharedBuildAccumulator { ), }; + let DynamicFilterBuildContext { + filter, + probe_exprs, + repartition_random_state, + null_aware, + membership_predicates_elided, + } = context; + Self { inner: Mutex::new(AccumulatorState { data: mode_data, completion: CompletionState::Pending, }), completion_notify: Notify::new(), - dynamic_filter, - on_right, + dynamic_filter: filter, + on_right: probe_exprs, repartition_random_state, probe_schema: right_child.schema(), null_aware, + membership_predicates_elided, } } + /// Builds the complete dynamic-filter predicate for one build partition. + fn create_partition_filter( + &self, + partition: &PartitionData, + ) -> Result>> { + let bounds_expr = create_bounds_predicate(&self.on_right, &partition.bounds); + let membership_expr = if membership_matches_integer_bounds( + &partition.pushdown, + &partition.bounds, + self.on_right.len(), + ) { + self.membership_predicates_elided.add(1); + None + } else { + create_membership_predicate( + &self.on_right, + &partition.pushdown, + &HASH_JOIN_SEED, + self.probe_schema.as_ref(), + )? + }; + + Ok(combine_membership_and_bounds(membership_expr, bounds_expr)) + } + /// Report build-side data from a partition /// /// This unified method handles both CollectLeft and Partitioned modes. When all partitions @@ -572,17 +713,8 @@ impl SharedBuildAccumulator { match finalize_input { FinalizeInput::CollectLeft(partition) => match partition { PartitionStatus::Reported(partition_data) => { - let membership_expr = create_membership_predicate( - &self.on_right, - partition_data.pushdown.clone(), - &HASH_JOIN_SEED, - self.probe_schema.as_ref(), - )?; - let bounds_expr = - create_bounds_predicate(&self.on_right, &partition_data.bounds); - if let Some(filter_expr) = - combine_membership_and_bounds(membership_expr, bounds_expr) + self.create_partition_filter(&partition_data)? { self.dynamic_filter .update(self.null_aware_filter(filter_expr))?; @@ -625,21 +757,9 @@ impl SharedBuildAccumulator { empty_partition_ids.push(partition_id); } PartitionStatus::Reported(partition) => { - let membership_expr = create_membership_predicate( - &self.on_right, - partition.pushdown.clone(), - &HASH_JOIN_SEED, - self.probe_schema.as_ref(), - )?; - let bounds_expr = create_bounds_predicate( - &self.on_right, - &partition.bounds, - ); - let then_expr = combine_membership_and_bounds( - membership_expr, - bounds_expr, - ) - .unwrap_or_else(|| lit(true)); + let then_expr = self + .create_partition_filter(partition)? + .unwrap_or_else(|| lit(true)); real_branches.push(( lit(ScalarValue::UInt64(Some(partition_id as u64))), then_expr, @@ -757,6 +877,7 @@ pub(super) fn make_partitioned_accumulator_for_test( repartition_random_state: SeededRandomState::with_seed(1), probe_schema, null_aware: false, + membership_predicates_elided: Count::new(), } } @@ -777,7 +898,7 @@ pub(super) fn completed_partitions_for_test(acc: &SharedBuildAccumulator) -> usi mod tests { use super::*; - use arrow::array::{ArrayRef, Int32Array}; + use arrow::array::{ArrayRef, Int32Array, UInt64Array}; use datafusion_physical_expr::expressions::{Column, Literal}; fn test_on_right() -> Vec { @@ -814,6 +935,7 @@ mod tests { repartition_random_state: SeededRandomState::with_seed(1), probe_schema: test_probe_schema(), null_aware: false, + membership_predicates_elided: Count::new(), } } @@ -841,7 +963,20 @@ mod tests { } fn in_list(values: &[i32]) -> PushdownStrategy { - PushdownStrategy::InList(Arc::new(Int32Array::from(values.to_vec())) as ArrayRef) + let array = Arc::new(Int32Array::from(values.to_vec())) as ArrayRef; + let map = array_map_from_i32(values); + PushdownStrategy::InList(InListMembership::new(array, map.as_ref())) + } + + fn array_map_from_i32(values: &[i32]) -> Arc { + let array = Arc::new(Int32Array::from(values.to_vec())) as ArrayRef; + let min = values.iter().min().copied().unwrap() as u64; + let max = values.iter().max().copied().unwrap() as u64; + Arc::new(Map::ArrayMap(ArrayMap::try_new(&array, min, max).unwrap())) + } + + fn array_map(values: &[i32]) -> PushdownStrategy { + PushdownStrategy::Map(array_map_from_i32(values)) } fn bounds(min: i32, max: i32) -> PartitionBounds { @@ -961,6 +1096,206 @@ mod tests { assert_top_binary_op(&expr, Operator::And); } + #[test] + fn collect_left_contiguous_integer_membership_uses_bounds_only() { + let acc = make_collect_left_accumulator_for_test(); + + acc.build_filter(FinalizeInput::CollectLeft(reported( + in_list(&[1, 2, 3]), + bounds(1, 3), + ))) + .unwrap(); + + let expr = current_expr(&acc); + let bounds = binary_expr(&expr); + assert_eq!(bounds.op(), &Operator::And); + assert!( + bounds.right().downcast_ref::().is_none(), + "a complete integer interval must not retain a redundant InList" + ); + assert_eq!(acc.membership_predicates_elided.value(), 1); + } + + #[test] + fn collect_left_duplicate_contiguous_membership_uses_bounds_only() { + let acc = make_collect_left_accumulator_for_test(); + + acc.build_filter(FinalizeInput::CollectLeft(reported( + in_list(&[1, 2, 2, 3]), + bounds(1, 3), + ))) + .unwrap(); + + let expr = current_expr(&acc); + assert!( + binary_expr(&expr) + .right() + .downcast_ref::() + .is_none() + ); + } + + #[test] + fn collect_left_gapped_integer_membership_is_retained() { + let acc = make_collect_left_accumulator_for_test(); + + acc.build_filter(FinalizeInput::CollectLeft(reported( + in_list(&[1, 3]), + bounds(1, 3), + ))) + .unwrap(); + + let expr = current_expr(&acc); + assert!( + binary_expr(&expr) + .right() + .downcast_ref::() + .is_some() + ); + assert_eq!(acc.membership_predicates_elided.value(), 0); + } + + #[test] + fn collect_left_contiguous_array_map_uses_bounds_only() { + let acc = make_collect_left_accumulator_for_test(); + + acc.build_filter(FinalizeInput::CollectLeft(reported( + array_map(&[1, 2, 3]), + bounds(1, 3), + ))) + .unwrap(); + + let expr = current_expr(&acc); + assert!( + binary_expr(&expr) + .right() + .downcast_ref::() + .is_none() + ); + } + + #[test] + fn collect_left_gapped_array_map_membership_is_retained() { + let acc = make_collect_left_accumulator_for_test(); + + acc.build_filter(FinalizeInput::CollectLeft(reported( + array_map(&[1, 3]), + bounds(1, 3), + ))) + .unwrap(); + + let expr = current_expr(&acc); + assert!( + binary_expr(&expr) + .right() + .downcast_ref::() + .is_some() + ); + } + + #[test] + fn membership_range_proof_accepts_signed_and_unsigned_integers() { + let membership = in_list(&[-2, -1, 0, 1, 2]); + assert!(membership_matches_integer_bounds( + &membership, + &bounds(-2, 2), + 1, + )); + + let values = Arc::new(UInt64Array::from(vec![10, 11, 12])) as ArrayRef; + let map = Arc::new(Map::ArrayMap(ArrayMap::try_new(&values, 10, 12).unwrap())); + let membership = + PushdownStrategy::InList(InListMembership::new(values, map.as_ref())); + assert!(membership_matches_integer_bounds( + &membership, + &PartitionBounds::new(vec![ColumnBounds::new( + ScalarValue::UInt64(Some(10)), + ScalarValue::UInt64(Some(12)), + )]), + 1, + )); + } + + #[test] + fn membership_range_proof_rejects_unsupported_inputs() { + let membership = in_list(&[-2, -1, 0, 1, 2]); + assert!(!membership_matches_integer_bounds( + &membership, + &bounds(-2, 2), + 0, + )); + assert!(!membership_matches_integer_bounds( + &membership, + &bounds(-2, 2), + 2, + )); + assert!(!membership_matches_integer_bounds( + &membership, + &PartitionBounds::new(vec![ + ColumnBounds::new( + ScalarValue::Int32(Some(-2)), + ScalarValue::Int32(Some(2)), + ), + ColumnBounds::new( + ScalarValue::Int32(Some(0)), + ScalarValue::Int32(Some(0)), + ), + ]), + 1, + )); + assert!(!membership_matches_integer_bounds( + &membership, + &PartitionBounds::new(vec![ColumnBounds::new( + ScalarValue::Utf8(Some("a".to_string())), + ScalarValue::Utf8(Some("e".to_string())), + )]), + 1, + )); + assert!(!membership_matches_integer_bounds( + &membership, + &PartitionBounds::new(vec![ColumnBounds::new( + ScalarValue::Int32(Some(-2)), + ScalarValue::Int64(Some(2)), + )]), + 1, + )); + assert!(!membership_matches_integer_bounds( + &membership, + &PartitionBounds::new(vec![ColumnBounds::new( + ScalarValue::Int32(None), + ScalarValue::Int32(Some(2)), + )]), + 1, + )); + assert!(!membership_matches_integer_bounds( + &membership, + &bounds(2, -2), + 1, + )); + + let values = Arc::new(Int32Array::from(vec![1, 3])) as ArrayRef; + let unrelated_map = array_map_from_i32(&[1, 2, 3]); + let inconsistent = PushdownStrategy::InList(InListMembership::new( + values, + unrelated_map.as_ref(), + )); + assert!(!membership_matches_integer_bounds( + &inconsistent, + &bounds(1, 3), + 1, + )); + + let full_i64_domain = PartitionBounds::new(vec![ColumnBounds::new( + ScalarValue::Int64(Some(i64::MIN)), + ScalarValue::Int64(Some(i64::MAX)), + )]); + assert_eq!(inclusive_integer_span(&full_i64_domain), Some(1_u128 << 64)); + assert_ne!( + inclusive_integer_span(&full_i64_domain), + Some(usize::MAX as u128) + ); + } + #[test] fn collect_left_empty_build_data_does_not_update_filter() { let acc = make_collect_left_accumulator_for_test(); @@ -997,6 +1332,30 @@ mod tests { assert!(expr.downcast_ref::().is_none()); } + #[test] + fn partitioned_elides_only_contiguous_membership_predicates() { + let acc = make_partitioned_expr_accumulator_for_test(2); + + acc.build_filter(FinalizeInput::Partitioned(vec![ + reported(in_list(&[1, 2, 3]), bounds(1, 3)), + reported(in_list(&[10, 12]), bounds(10, 12)), + ])) + .unwrap(); + + let expr = current_expr(&acc); + let branches = case_expr(&expr).when_then_expr(); + assert_eq!(branches.len(), 2); + + let contiguous = binary_expr(&branches[0].1); + assert_eq!(contiguous.op(), &Operator::And); + assert!(contiguous.right().downcast_ref::().is_none()); + + let gapped = binary_expr(&branches[1].1); + assert_eq!(gapped.op(), &Operator::And); + assert!(gapped.right().downcast_ref::().is_some()); + assert_eq!(acc.membership_predicates_elided.value(), 1); + } + #[test] fn partitioned_canceled_unknown_partitions_keep_unknown_routes_permissive() { let acc = make_partitioned_expr_accumulator_for_test(2); diff --git a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt index f1e787441d5e1..ace125c63b210 100644 --- a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt +++ b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt @@ -387,7 +387,7 @@ FROM join_probe p INNER JOIN join_build AS build ON p.a = build.a AND p.b = build.b; ---- Plan with Metrics -01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0), (b@1, b@1)], projection=[a@3, b@4, c@2, e@5], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)] +01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0), (b@1, b@1)], projection=[a@3, b@4, c@2, e@5], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, dynamic_filter_membership_predicates_elided=0, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)] 02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/join_build.parquet]]}, projection=[a, b, c], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, predicate_cache_inner_records=0, predicate_cache_records=0, scan_efficiency_ratio=19.58% (196/1.00 K)] 03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/join_probe.parquet]]}, projection=[a, b, e], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}, {c0:ab,c1:bb}]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_null_count@5 != row_count@2 AND b_max@4 >= ba AND b_null_count@5 != row_count@2 AND b_min@6 <= bb, required_guarantees=[], metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=2, pushdown_rows_pruned=2, predicate_cache_inner_records=8, predicate_cache_records=4, scan_efficiency_ratio=22.05% (228/1.03 K)] @@ -472,8 +472,8 @@ INNER JOIN nested_t2 ON nested_t1.a = nested_t2.b INNER JOIN nested_t3 ON nested_t2.c = nested_t3.d; ---- Plan with Metrics -01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(c@3, d@0)], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)] -02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, b@0)], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)] +01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(c@3, d@0)], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, dynamic_filter_membership_predicates_elided=0, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)] +02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, b@0)], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, dynamic_filter_membership_predicates_elided=0, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)] 03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nested_t1.parquet]]}, projection=[a, x], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, predicate_cache_inner_records=0, predicate_cache_records=0, scan_efficiency_ratio=17.37% (132/760)] 04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nested_t2.parquet]]}, projection=[b, c, y], file_type=parquet, predicate=DynamicFilter [ b@0 >= aa AND b@0 <= ab AND b@0 IN (SET) ([aa, ab]) ], dynamic_rg_pruning=eligible, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 >= aa AND b_null_count@1 != row_count@2 AND b_min@3 <= ab AND (b_null_count@1 != row_count@2 AND b_min@3 <= aa AND aa <= b_max@0 OR b_null_count@1 != row_count@2 AND b_min@3 <= ab AND ab <= b_max@0), required_guarantees=[b in (aa, ab)], metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=1 total → 1 matched, page_index_rows_pruned=5 total → 5 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=2, pushdown_rows_pruned=3, predicate_cache_inner_records=5, predicate_cache_records=2, scan_efficiency_ratio=22.46% (234/1.04 K)] 05)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nested_t3.parquet]]}, projection=[d, z], file_type=parquet, predicate=DynamicFilter [ d@0 >= ca AND d@0 <= cb AND hash_lookup ], dynamic_rg_pruning=eligible, pruning_predicate=d_null_count@1 != row_count@2 AND d_max@0 >= ca AND d_null_count@1 != row_count@2 AND d_min@3 <= cb, required_guarantees=[], metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=1 total → 1 matched, page_index_rows_pruned=8 total → 8 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=2, pushdown_rows_pruned=6, predicate_cache_inner_records=8, predicate_cache_records=2, scan_efficiency_ratio=21.45% (172/802)] @@ -604,7 +604,7 @@ LIMIT 2; ---- Plan with Metrics 01)SortExec: TopK(fetch=2), expr=[e@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[e@0 < bb], metrics=[output_rows=2, output_batches=1, row_replacements=2] -02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, d@0)], projection=[e@2], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)] +02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, d@0)], projection=[e@2], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, dynamic_filter_membership_predicates_elided=0, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)] 03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_join_build.parquet]]}, projection=[a], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, predicate_cache_inner_records=0, predicate_cache_records=0, scan_efficiency_ratio=6.39% (64/1.00 K)] 04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_join_probe.parquet]]}, projection=[d, e], file_type=parquet, predicate=DynamicFilter [ d@0 >= aa AND d@0 <= ab AND d@0 IN (SET) ([aa, ab]) ] AND DynamicFilter [ e@1 < bb ], dynamic_rg_pruning=eligible, pruning_predicate=d_null_count@1 != row_count@2 AND d_max@0 >= aa AND d_null_count@1 != row_count@2 AND d_min@3 <= ab AND (d_null_count@1 != row_count@2 AND d_min@3 <= aa AND aa <= d_max@0 OR d_null_count@1 != row_count@2 AND d_min@3 <= ab AND ab <= d_max@0) AND e_null_count@5 != row_count@2 AND e_min@4 < bb, required_guarantees=[d in (aa, ab)], metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=1 total → 1 matched, page_index_rows_pruned=4 total → 4 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=2, pushdown_rows_pruned=2, predicate_cache_inner_records=8, predicate_cache_records=4, scan_efficiency_ratio=14.89% (154/1.03 K)] @@ -739,7 +739,7 @@ INNER JOIN ( ) agg ON b.a = agg.a; ---- Plan with Metrics -01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0)], projection=[a@0, min_value@2], metrics=[output_rows=2, output_batches=2, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=2, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)] +01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0)], projection=[a@0, min_value@2], metrics=[output_rows=2, output_batches=2, array_map_created_count=0, build_input_batches=1, build_input_rows=2, dynamic_filter_membership_predicates_elided=0, input_batches=2, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)] 02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/join_agg_build.parquet]]}, projection=[a], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, predicate_cache_inner_records=0, predicate_cache_records=0, scan_efficiency_ratio=14.45% (64/443)] 03)--ProjectionExec: expr=[a@0 as a, min(join_agg_probe.value)@1 as min_value], metrics=[output_rows=2, output_batches=2] 04)----AggregateExec: mode=FinalPartitioned, gby=[a@0 as a], aggr=[min(join_agg_probe.value)], metrics=[output_rows=2, output_batches=2, spill_count=0, spilled_rows=0] @@ -806,7 +806,7 @@ FROM nulls_build INNER JOIN nulls_probe ON nulls_build.a = nulls_probe.a AND nulls_build.b = nulls_probe.b; ---- Plan with Metrics -01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0), (b@1, b@1)], metrics=[output_rows=1, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=3, input_batches=1, input_rows=1, avg_fanout=100% (1/1), probe_hit_rate=100% (1/1)] +01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0), (b@1, b@1)], metrics=[output_rows=1, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=3, dynamic_filter_membership_predicates_elided=0, input_batches=1, input_rows=1, avg_fanout=100% (1/1), probe_hit_rate=100% (1/1)] 02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nulls_build.parquet]]}, projection=[a, b], file_type=parquet, metrics=[output_rows=3, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, predicate_cache_inner_records=0, predicate_cache_records=0, scan_efficiency_ratio=18.6% (144/774)] 03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nulls_probe.parquet]]}, projection=[a, b, c], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= 1 AND b@1 <= 2 AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:1}, {c0:,c1:2}, {c0:ab,c1:}]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_null_count@5 != row_count@2 AND b_max@4 >= 1 AND b_null_count@5 != row_count@2 AND b_min@6 <= 2, required_guarantees=[], metrics=[output_rows=1, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=1, pushdown_rows_pruned=3, predicate_cache_inner_records=8, predicate_cache_records=2, scan_efficiency_ratio=20.18% (225/1.11 K)] @@ -872,7 +872,7 @@ FROM lj_build LEFT JOIN lj_probe ON lj_build.a = lj_probe.a AND lj_build.b = lj_probe.b; ---- Plan with Metrics -01)HashJoinExec: mode=CollectLeft, join_type=Left, on=[(a@0, a@0), (b@1, b@1)], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=2, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)] +01)HashJoinExec: mode=CollectLeft, join_type=Left, on=[(a@0, a@0), (b@1, b@1)], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, dynamic_filter_membership_predicates_elided=0, input_batches=2, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)] 02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/lj_build.parquet]]}, projection=[a, b, c], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, predicate_cache_inner_records=0, predicate_cache_records=0, scan_efficiency_ratio=19.58% (196/1.00 K)] 03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/lj_probe.parquet]]}, projection=[a, b, e], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}, {c0:ab,c1:bb}]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_null_count@5 != row_count@2 AND b_max@4 >= ba AND b_null_count@5 != row_count@2 AND b_min@6 <= bb, required_guarantees=[], metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=2, pushdown_rows_pruned=2, predicate_cache_inner_records=8, predicate_cache_records=4, scan_efficiency_ratio=22.05% (228/1.03 K)] @@ -888,7 +888,7 @@ WHERE EXISTS ( ); ---- Plan with Metrics -01)HashJoinExec: mode=CollectLeft, join_type=LeftSemi, on=[(a@0, a@0), (b@1, b@1)], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=2, input_rows=4, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)] +01)HashJoinExec: mode=CollectLeft, join_type=LeftSemi, on=[(a@0, a@0), (b@1, b@1)], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, dynamic_filter_membership_predicates_elided=0, input_batches=2, input_rows=4, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)] 02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/lj_build.parquet]]}, projection=[a, b, c], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, predicate_cache_inner_records=0, predicate_cache_records=0, scan_efficiency_ratio=19.58% (196/1.00 K)] 03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/lj_probe.parquet]]}, projection=[a, b], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}, {c0:ab,c1:bb}]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_null_count@5 != row_count@2 AND b_max@4 >= ba AND b_null_count@5 != row_count@2 AND b_min@6 <= bb, required_guarantees=[], metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=2, pushdown_rows_pruned=2, predicate_cache_inner_records=8, predicate_cache_records=4, scan_efficiency_ratio=14.89% (154/1.03 K)] @@ -958,7 +958,7 @@ FROM hl_probe p INNER JOIN hl_build AS build ON p.a = build.a AND p.b = build.b; ---- Plan with Metrics -01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0), (b@1, b@1)], projection=[a@3, b@4, c@2, e@5], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)] +01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0), (b@1, b@1)], projection=[a@3, b@4, c@2, e@5], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, dynamic_filter_membership_predicates_elided=0, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)] 02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/hl_build.parquet]]}, projection=[a, b, c], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, predicate_cache_inner_records=0, predicate_cache_records=0, scan_efficiency_ratio=19.58% (196/1.00 K)] 03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/hl_probe.parquet]]}, projection=[a, b, e], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND hash_lookup ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_null_count@5 != row_count@2 AND b_max@4 >= ba AND b_null_count@5 != row_count@2 AND b_min@6 <= bb, required_guarantees=[], metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=2, pushdown_rows_pruned=2, predicate_cache_inner_records=8, predicate_cache_records=4, scan_efficiency_ratio=22.05% (228/1.03 K)] @@ -1007,7 +1007,7 @@ FROM int_build b INNER JOIN int_probe p ON b.id1 = p.id1 AND b.id2 = p.id2; ---- Plan with Metrics -01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id1@0, id1@0), (id2@1, id2@1)], projection=[id1@0, id2@1, value@2, data@5], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)] +01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id1@0, id1@0), (id2@1, id2@1)], projection=[id1@0, id2@1, value@2, data@5], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, dynamic_filter_membership_predicates_elided=0, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)] 02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/int_build.parquet]]}, projection=[id1, id2, value], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, predicate_cache_inner_records=0, predicate_cache_records=0, scan_efficiency_ratio=18.23% (204/1.12 K)] 03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/int_probe.parquet]]}, projection=[id1, id2, data], file_type=parquet, predicate=DynamicFilter [ id1@0 >= 1 AND id1@0 <= 2 AND id2@1 >= 10 AND id2@1 <= 20 AND hash_lookup ], dynamic_rg_pruning=eligible, pruning_predicate=id1_null_count@1 != row_count@2 AND id1_max@0 >= 1 AND id1_null_count@1 != row_count@2 AND id1_min@3 <= 2 AND id2_null_count@5 != row_count@2 AND id2_max@4 >= 10 AND id2_null_count@5 != row_count@2 AND id2_min@6 <= 20, required_guarantees=[], metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=2, pushdown_rows_pruned=2, predicate_cache_inner_records=8, predicate_cache_records=4, scan_efficiency_ratio=20.67% (221/1.07 K)] From 3e9eda92764acdbdab178994bc76b62091a83e70 Mon Sep 17 00:00:00 2001 From: Qiwei Huang Date: Tue, 28 Jul 2026 17:52:35 +0800 Subject: [PATCH 2/4] test: cover dynamic filter elision edge cases --- .../src/joins/hash_join/shared_bounds.rs | 60 ++++++++++++++++++- 1 file changed, 59 insertions(+), 1 deletion(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs index fec34918ccc4e..bf5e9157daa72 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -759,7 +759,7 @@ impl SharedBuildAccumulator { PartitionStatus::Reported(partition) => { let then_expr = self .create_partition_filter(partition)? - .unwrap_or_else(|| lit(true)); + .expect("a reported non-empty partition must produce a filter"); real_branches.push(( lit(ScalarValue::UInt64(Some(partition_id as u64))), then_expr, @@ -905,6 +905,10 @@ mod tests { vec![Arc::new(Column::new("probe_key", 0))] } + fn invalid_test_on_right() -> Vec { + vec![Arc::new(Column::new("missing_probe_key", 1))] + } + fn test_probe_schema() -> Arc { Arc::new(Schema::new(vec![Field::new( "probe_key", @@ -1267,6 +1271,22 @@ mod tests { )]), 1, )); + assert!(!membership_matches_integer_bounds( + &membership, + &PartitionBounds::new(vec![ColumnBounds::new( + ScalarValue::Int32(Some(-2)), + ScalarValue::Int32(None), + )]), + 1, + )); + assert!(!membership_matches_integer_bounds( + &membership, + &PartitionBounds::new(vec![ColumnBounds::new( + ScalarValue::Int64(Some(-2)), + ScalarValue::Int64(Some(2)), + )]), + 1, + )); assert!(!membership_matches_integer_bounds( &membership, &bounds(2, -2), @@ -1296,6 +1316,44 @@ mod tests { ); } + #[test] + fn membership_expression_errors_propagate_for_both_modes() { + let collect_left = make_accumulator_for_test( + AccumulatedBuildData::CollectLeft { + data: PartitionStatus::Pending, + reported_count: 0, + expected_reports: 1, + }, + invalid_test_on_right(), + ); + assert!( + collect_left + .build_filter(FinalizeInput::CollectLeft(reported( + in_list(&[1, 3]), + no_bounds(), + ))) + .is_err(), + "CollectLeft must propagate membership-expression construction errors" + ); + + let partitioned = make_accumulator_for_test( + AccumulatedBuildData::Partitioned { + partitions: vec![PartitionStatus::Pending], + completed_partitions: 0, + }, + invalid_test_on_right(), + ); + assert!( + partitioned + .build_filter(FinalizeInput::Partitioned(vec![reported( + in_list(&[1, 3]), + no_bounds(), + )])) + .is_err(), + "Partitioned must propagate membership-expression construction errors" + ); + } + #[test] fn collect_left_empty_build_data_does_not_update_filter() { let acc = make_collect_left_accumulator_for_test(); From fd2c5e00f82131cce9cc99c0461a88bc211b0985 Mon Sep 17 00:00:00 2001 From: Qiwei Huang Date: Tue, 28 Jul 2026 21:28:41 +0800 Subject: [PATCH 3/4] refactor: make non-empty dynamic filters explicit --- .../src/joins/hash_join/shared_bounds.rs | 218 +++++++++++------- 1 file changed, 134 insertions(+), 84 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs index bf5e9157daa72..93942bda81617 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -80,18 +80,34 @@ impl PartitionBounds { } } +/// A build-side representation that can produce a membership predicate. +#[derive(Clone, Copy)] +enum MembershipSource<'a> { + InList(&'a InListMembership), + Map(&'a Arc), +} + +impl MembershipSource<'_> { + fn distinct_key_count_lower_bound(self) -> Option { + match self { + Self::InList(membership) => membership.distinct_key_count_lower_bound, + Self::Map(map) => Some(map.num_of_distinct_key()), + } + } +} + /// Creates a membership predicate using the representation selected while /// collecting the build side. /// /// Supports both single-column and multi-column joins using struct expressions. fn create_membership_predicate( on_right: &[PhysicalExprRef], - pushdown: &PushdownStrategy, + source: MembershipSource<'_>, random_state: &SeededRandomState, schema: &Schema, -) -> Result>> { - match pushdown { - PushdownStrategy::InList(membership) => { +) -> Result> { + match source { + MembershipSource::InList(membership) => { // Build the expression to compare against let expr = if on_right.len() == 1 { // Single column: col IN (val1, val2, ...) @@ -119,23 +135,39 @@ fn create_membership_predicate( }; // Use InListExpr::try_new_from_array() to build an InList with static_filter optimization (hash-based lookup) - Ok(Some(Arc::new(InListExpr::try_new_from_array( + Ok(Arc::new(InListExpr::try_new_from_array( expr, Arc::clone(&membership.values), false, schema, - )?))) + )?)) } - PushdownStrategy::Map(hash_map) => Ok(Some(Arc::new(HashTableLookupExpr::new( + MembershipSource::Map(hash_map) => Ok(Arc::new(HashTableLookupExpr::new( on_right.to_vec(), random_state.clone(), Arc::clone(hash_map), "hash_lookup".to_string(), - )) as Arc)), - PushdownStrategy::Empty => Ok(None), + )) as Arc), } } +fn create_column_bounds_predicate( + right_expr: &PhysicalExprRef, + column_bounds: &ColumnBounds, +) -> Arc { + let min_expr = Arc::new(BinaryExpr::new( + Arc::clone(right_expr), + Operator::GtEq, + lit(column_bounds.min.clone()), + )) as Arc; + let max_expr = Arc::new(BinaryExpr::new( + Arc::clone(right_expr), + Operator::LtEq, + lit(column_bounds.max.clone()), + )) as Arc; + Arc::new(BinaryExpr::new(min_expr, Operator::And, max_expr)) as Arc +} + /// Creates a bounds predicate from partition bounds. /// /// Returns `None` if no column bounds are available. @@ -148,20 +180,8 @@ fn create_bounds_predicate( for (col_idx, right_expr) in on_right.iter().enumerate() { if let Some(column_bounds) = bounds.get_column_bounds(col_idx) { - // Create predicate: col >= min AND col <= max - let min_expr = Arc::new(BinaryExpr::new( - Arc::clone(right_expr), - Operator::GtEq, - lit(column_bounds.min.clone()), - )) as Arc; - let max_expr = Arc::new(BinaryExpr::new( - Arc::clone(right_expr), - Operator::LtEq, - lit(column_bounds.max.clone()), - )) as Arc; - let range_expr = Arc::new(BinaryExpr::new(min_expr, Operator::And, max_expr)) - as Arc; - column_predicates.push(range_expr); + column_predicates + .push(create_column_bounds_predicate(right_expr, column_bounds)); } } @@ -180,25 +200,6 @@ fn create_bounds_predicate( } } -/// Combines a membership predicate and a bounds predicate with logical AND. -/// -/// Returns `None` when neither is available; callers decide the fallback (e.g. -/// skip updating the filter vs. emit a `lit(true)` branch inside a CASE). -fn combine_membership_and_bounds( - membership_expr: Option>, - bounds_expr: Option>, -) -> Option> { - match (membership_expr, bounds_expr) { - (Some(membership), Some(bounds)) => { - Some(Arc::new(BinaryExpr::new(bounds, Operator::And, membership)) - as Arc) - } - (Some(membership), None) => Some(membership), - (None, Some(bounds)) => Some(bounds), - (None, None) => None, - } -} - fn inclusive_integer_span(bounds: &PartitionBounds) -> Option { let [column_bounds] = bounds.column_bounds.as_slice() else { return None; @@ -216,36 +217,38 @@ fn inclusive_integer_span(bounds: &PartitionBounds) -> Option { Some(u128::from(ArrayMap::calculate_range(min, max)) + 1) } -/// Returns true when the membership set is provably identical to its -/// single-column integer bounds. +/// Returns the bounds predicate when the membership set is provably identical +/// to its single-column integer bounds. /// /// `distinct_key_count_lower_bound` is exact for [`ArrayMap`] and is the number of /// distinct hashes for the regular hash map. The latter is a lower bound on /// distinct keys. Because the keys are also bounded by the inclusive integer /// span, equality with the span proves that every value is present even when /// hash collisions are possible. -fn membership_matches_integer_bounds( - pushdown: &PushdownStrategy, +fn complete_integer_domain_bounds_predicate( + source: MembershipSource<'_>, bounds: &PartitionBounds, - join_key_count: usize, -) -> bool { - if join_key_count != 1 { - return false; - } - let Some(distinct_key_count_lower_bound) = pushdown.distinct_key_count_lower_bound() - else { - return false; + on_right: &[PhysicalExprRef], +) -> Option> { + let [right_expr] = on_right else { + return None; }; - let Some(inclusive_span) = inclusive_integer_span(bounds) else { - return false; + let [column_bounds] = bounds.column_bounds.as_slice() else { + return None; }; - if let PushdownStrategy::InList(membership) = pushdown - && membership.values.data_type() != &bounds.column_bounds[0].min.data_type() + let distinct_key_count_lower_bound = source.distinct_key_count_lower_bound()?; + let inclusive_span = inclusive_integer_span(bounds)?; + if let MembershipSource::InList(membership) = source + && membership.values.data_type() != &column_bounds.min.data_type() { - return false; + return None; + } + + if inclusive_span != distinct_key_count_lower_bound as u128 { + return None; } - inclusive_span == distinct_key_count_lower_bound as u128 + Some(create_column_bounds_predicate(right_expr, column_bounds)) } /// Coordinates build-side information collection across multiple partitions @@ -372,10 +375,10 @@ pub(crate) enum PushdownStrategy { } impl PushdownStrategy { - fn distinct_key_count_lower_bound(&self) -> Option { + fn membership_source(&self) -> Option> { match self { - Self::InList(membership) => membership.distinct_key_count_lower_bound, - Self::Map(map) => Some(map.num_of_distinct_key()), + Self::InList(membership) => Some(MembershipSource::InList(membership)), + Self::Map(map) => Some(MembershipSource::Map(map)), Self::Empty => None, } } @@ -533,24 +536,46 @@ impl SharedBuildAccumulator { &self, partition: &PartitionData, ) -> Result>> { - let bounds_expr = create_bounds_predicate(&self.on_right, &partition.bounds); - let membership_expr = if membership_matches_integer_bounds( - &partition.pushdown, + let Some(source) = partition.pushdown.membership_source() else { + return Ok(create_bounds_predicate(&self.on_right, &partition.bounds)); + }; + + self.create_non_empty_partition_filter(partition, source) + .map(Some) + } + + /// Builds the predicate for a partition that has membership data. + fn create_non_empty_partition_filter( + &self, + partition: &PartitionData, + source: MembershipSource<'_>, + ) -> Result> { + if let Some(bounds_expr) = complete_integer_domain_bounds_predicate( + source, &partition.bounds, - self.on_right.len(), + &self.on_right, ) { self.membership_predicates_elided.add(1); - None - } else { - create_membership_predicate( - &self.on_right, - &partition.pushdown, - &HASH_JOIN_SEED, - self.probe_schema.as_ref(), - )? - }; + return Ok(bounds_expr); + } - Ok(combine_membership_and_bounds(membership_expr, bounds_expr)) + let membership_expr = create_membership_predicate( + &self.on_right, + source, + &HASH_JOIN_SEED, + self.probe_schema.as_ref(), + )?; + + Ok( + if let Some(bounds_expr) = + create_bounds_predicate(&self.on_right, &partition.bounds) + { + Arc::new(BinaryExpr::new(bounds_expr, Operator::And, membership_expr)) + as Arc + } else { + membership_expr + }, + ) } /// Report build-side data from a partition @@ -751,15 +776,14 @@ impl SharedBuildAccumulator { for (partition_id, partition) in partitions.iter().enumerate() { match partition { - PartitionStatus::Reported(partition) - if matches!(partition.pushdown, PushdownStrategy::Empty) => - { - empty_partition_ids.push(partition_id); - } PartitionStatus::Reported(partition) => { + let Some(source) = partition.pushdown.membership_source() + else { + empty_partition_ids.push(partition_id); + continue; + }; let then_expr = self - .create_partition_filter(partition)? - .expect("a reported non-empty partition must produce a filter"); + .create_non_empty_partition_filter(partition, source)?; real_branches.push(( lit(ScalarValue::UInt64(Some(partition_id as u64))), then_expr, @@ -909,6 +933,22 @@ mod tests { vec![Arc::new(Column::new("missing_probe_key", 1))] } + fn membership_matches_integer_bounds( + pushdown: &PushdownStrategy, + bounds: &PartitionBounds, + join_key_count: usize, + ) -> bool { + let on_right = (0..join_key_count) + .map(|index| Arc::new(Column::new("probe_key", index)) as PhysicalExprRef) + .collect::>(); + pushdown + .membership_source() + .and_then(|source| { + complete_integer_domain_bounds_predicate(source, bounds, &on_right) + }) + .is_some() + } + fn test_probe_schema() -> Arc { Arc::new(Schema::new(vec![Field::new( "probe_key", @@ -1223,6 +1263,16 @@ mod tests { #[test] fn membership_range_proof_rejects_unsupported_inputs() { let membership = in_list(&[-2, -1, 0, 1, 2]); + assert!(!membership_matches_integer_bounds( + &PushdownStrategy::Empty, + &bounds(-2, 2), + 1, + )); + assert!(!membership_matches_integer_bounds( + &membership, + &no_bounds(), + 1, + )); assert!(!membership_matches_integer_bounds( &membership, &bounds(-2, 2), From 5da68518e3325b6f59f60423e008f76aa4a00775 Mon Sep 17 00:00:00 2001 From: Qiwei Huang Date: Wed, 29 Jul 2026 08:27:19 +0800 Subject: [PATCH 4/4] refactor: remove unreachable coverage branches --- .../src/joins/hash_join/shared_bounds.rs | 15 ++++++--------- 1 file changed, 6 insertions(+), 9 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs index 93942bda81617..95c7e1d7cba29 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -200,17 +200,14 @@ fn create_bounds_predicate( } } -fn inclusive_integer_span(bounds: &PartitionBounds) -> Option { - let [column_bounds] = bounds.column_bounds.as_slice() else { - return None; - }; +fn inclusive_integer_span(column_bounds: &ColumnBounds) -> Option { if column_bounds.min.data_type() != column_bounds.max.data_type() || column_bounds.min > column_bounds.max { return None; } - let min = ArrayMap::key_to_u64(&column_bounds.min)?; - let max = ArrayMap::key_to_u64(&column_bounds.max)?; + let (min, max) = ArrayMap::key_to_u64(&column_bounds.min) + .zip(ArrayMap::key_to_u64(&column_bounds.max))?; // ArrayMap uses two's-complement values and wrapping subtraction, which // also gives the correct span for signed ranges that cross zero. @@ -237,7 +234,7 @@ fn complete_integer_domain_bounds_predicate( return None; }; let distinct_key_count_lower_bound = source.distinct_key_count_lower_bound()?; - let inclusive_span = inclusive_integer_span(bounds)?; + let inclusive_span = inclusive_integer_span(column_bounds)?; if let MembershipSource::InList(membership) = source && membership.values.data_type() != &column_bounds.min.data_type() { @@ -1355,10 +1352,10 @@ mod tests { 1, )); - let full_i64_domain = PartitionBounds::new(vec![ColumnBounds::new( + let full_i64_domain = ColumnBounds::new( ScalarValue::Int64(Some(i64::MIN)), ScalarValue::Int64(Some(i64::MAX)), - )]); + ); assert_eq!(inclusive_integer_span(&full_i64_domain), Some(1_u128 << 64)); assert_ne!( inclusive_integer_span(&full_i64_domain),