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..95c7e1d7cba29 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,34 @@ 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). +/// 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 { - // Use InList expression for small build sides - PushdownStrategy::InList(in_list_array) => { +) -> 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, ...) @@ -121,25 +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, - 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( + MembershipSource::Map(hash_map) => Ok(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), + )) 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. @@ -152,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)); } } @@ -184,23 +200,52 @@ fn create_bounds_predicate( } } -/// Combines a membership predicate and a bounds predicate with logical AND. +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, 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. + Some(u128::from(ArrayMap::calculate_range(min, max)) + 1) +} + +/// Returns the bounds predicate when the membership set is provably identical +/// to its single-column integer bounds. /// -/// 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>, +/// `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 complete_integer_domain_bounds_predicate( + source: MembershipSource<'_>, + bounds: &PartitionBounds, + on_right: &[PhysicalExprRef], ) -> 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, + let [right_expr] = on_right else { + return None; + }; + let [column_bounds] = bounds.column_bounds.as_slice() else { + return None; + }; + let distinct_key_count_lower_bound = source.distinct_key_count_lower_bound()?; + let inclusive_span = inclusive_integer_span(column_bounds)?; + if let MembershipSource::InList(membership) = source + && membership.values.data_type() != &column_bounds.min.data_type() + { + return None; + } + + if inclusive_span != distinct_key_count_lower_bound as u128 { + return None; } + + Some(create_column_bounds_predicate(right_expr, column_bounds)) } /// Coordinates build-side information collection across multiple partitions @@ -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 membership_source(&self) -> Option> { + match self { + Self::InList(membership) => Some(MembershipSource::InList(membership)), + Self::Map(map) => Some(MembershipSource::Map(map)), + 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,76 @@ 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 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, + ) { + self.membership_predicates_elided.add(1); + return Ok(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 /// /// This unified method handles both CollectLeft and Partitioned modes. When all partitions @@ -572,17 +735,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))?; @@ -619,27 +773,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 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 Some(source) = partition.pushdown.membership_source() + else { + empty_partition_ids.push(partition_id); + continue; + }; + let then_expr = self + .create_non_empty_partition_filter(partition, source)?; real_branches.push(( lit(ScalarValue::UInt64(Some(partition_id as u64))), then_expr, @@ -757,6 +898,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,13 +919,33 @@ 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 { vec![Arc::new(Column::new("probe_key", 0))] } + fn invalid_test_on_right() -> Vec { + 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", @@ -814,6 +976,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 +1004,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 +1137,270 @@ 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( + &PushdownStrategy::Empty, + &bounds(-2, 2), + 1, + )); + assert!(!membership_matches_integer_bounds( + &membership, + &no_bounds(), + 1, + )); + 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, + &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), + 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 = 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 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(); @@ -997,6 +1437,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)]