diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 9d9c867c2724b..1f5c9eed0e45e 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -863,14 +863,6 @@ impl HashJoinExec { return false; } - // Bounds and membership filters derived from the build side do not - // account for null-equal matching: a probe-side NULL key evaluates - // such predicates to NULL and would be pruned, even though it can - // match a build-side NULL when nulls compare equal. - if self.null_equality == NullEquality::NullEqualsNull { - return false; - } - // A null-aware anti join emits a build-side NULL only when the probe // is truly empty. The pushed filter can empty the probe by pruning // every row, which would surface that NULL wrongly. A NOT NULL build @@ -1411,6 +1403,7 @@ impl ExecutionPlan for HashJoinExec { filter, on_right, repartition_random_state, + self.null_equality, self.null_aware, )) }))) @@ -6933,7 +6926,7 @@ mod tests { } #[test] - fn test_dynamic_filter_pushdown_rejects_null_equal_join() -> Result<()> { + fn test_dynamic_filter_pushdown_allowed_for_null_equal_join() -> Result<()> { let (_, _, on) = build_schema_and_on()?; let left = build_table(("a1", &vec![1]), ("b1", &vec![1]), ("c1", &vec![1])); let right = build_table(("a2", &vec![1]), ("b1", &vec![1]), ("c2", &vec![1])); @@ -6956,7 +6949,9 @@ mod tests { false, )?; - assert!(!join.allow_join_dynamic_filter_pushdown(session_config.options())); + // Null-equal joins keep dynamic filter pushdown: the pushed predicate carries an + // `IS NULL` disjunct so a probe-side NULL still reaches the join. + assert!(join.allow_join_dynamic_filter_pushdown(session_config.options())); 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..7b58107e93c3f 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -33,7 +33,9 @@ use crate::joins::hash_join::partitioned_hash_eval::{ use arrow::array::ArrayRef; use arrow::datatypes::{DataType, Field, Schema}; use datafusion_common::config::ConfigOptions; -use datafusion_common::{DataFusionError, Result, ScalarValue, SharedResult}; +use datafusion_common::{ + DataFusionError, NullEquality, Result, ScalarValue, SharedResult, +}; use datafusion_expr::Operator; use datafusion_functions::core::r#struct as struct_func; use datafusion_physical_expr::expressions::{ @@ -255,6 +257,9 @@ pub(crate) struct SharedBuildAccumulator { repartition_random_state: SeededRandomState, /// Schema of the probe (right) side for evaluating filter expressions probe_schema: Arc, + /// Null equality of the join. Under `NullEqualsNull` a probe-side NULL can match a + /// build-side NULL, so the pushed filter must keep NULL rows here too. + null_equality: NullEquality, /// 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, @@ -277,10 +282,12 @@ pub(crate) enum PartitionBuildData { partition_id: usize, pushdown: PushdownStrategy, bounds: PartitionBounds, + keys_have_null: bool, }, CollectLeft { pushdown: PushdownStrategy, bounds: PartitionBounds, + keys_have_null: bool, }, } @@ -289,6 +296,9 @@ pub(crate) enum PartitionBuildData { struct PartitionData { bounds: PartitionBounds, pushdown: PushdownStrategy, + /// Whether any build key of this partition is NULL. Decides whether the pushed + /// filter must keep probe-side NULL rows for a null-equal join to match them. + keys_have_null: bool, } /// Build-side data organized by partition mode @@ -354,6 +364,7 @@ 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. + #[expect(clippy::too_many_arguments)] pub(crate) fn new_from_partition_mode( partition_mode: PartitionMode, left_child: &dyn ExecutionPlan, @@ -361,6 +372,7 @@ impl SharedBuildAccumulator { dynamic_filter: Arc, on_right: Vec, repartition_random_state: SeededRandomState, + null_equality: NullEquality, null_aware: bool, ) -> Self { // Troubleshooting: If partition counts are incorrect, verify this logic matches @@ -408,6 +420,7 @@ impl SharedBuildAccumulator { on_right, repartition_random_state, probe_schema: right_child.schema(), + null_equality, null_aware, } } @@ -461,6 +474,7 @@ impl SharedBuildAccumulator { partition_id, pushdown, bounds, + keys_have_null, }, AccumulatedBuildData::Partitioned { partitions, @@ -470,11 +484,18 @@ impl SharedBuildAccumulator { if matches!(partitions[partition_id], PartitionStatus::Pending) { *completed_partitions += 1; } - partitions[partition_id] = - PartitionStatus::Reported(PartitionData { pushdown, bounds }); + partitions[partition_id] = PartitionStatus::Reported(PartitionData { + pushdown, + bounds, + keys_have_null, + }); } ( - PartitionBuildData::CollectLeft { pushdown, bounds }, + PartitionBuildData::CollectLeft { + pushdown, + bounds, + keys_have_null, + }, AccumulatedBuildData::CollectLeft { data, reported_count, @@ -482,7 +503,11 @@ impl SharedBuildAccumulator { }, ) => { if matches!(data, PartitionStatus::Pending) { - *data = PartitionStatus::Reported(PartitionData { pushdown, bounds }); + *data = PartitionStatus::Reported(PartitionData { + pushdown, + bounds, + keys_have_null, + }); } *reported_count += 1; } @@ -584,8 +609,10 @@ impl SharedBuildAccumulator { if let Some(filter_expr) = combine_membership_and_bounds(membership_expr, bounds_expr) { - self.dynamic_filter - .update(self.null_aware_filter(filter_expr))?; + self.dynamic_filter.update(self.preserve_probe_nulls( + filter_expr, + partition_data.keys_have_null, + )?)?; } } PartitionStatus::Pending => { @@ -616,6 +643,7 @@ impl SharedBuildAccumulator { let mut real_branches = Vec::new(); let mut empty_partition_ids = Vec::new(); let mut has_canceled_unknown = false; + let mut keys_have_null = false; for (partition_id, partition) in partitions.iter().enumerate() { match partition { @@ -625,6 +653,7 @@ impl SharedBuildAccumulator { empty_partition_ids.push(partition_id); } PartitionStatus::Reported(partition) => { + keys_have_null |= partition.keys_have_null; let membership_expr = create_membership_predicate( &self.on_right, partition.pushdown.clone(), @@ -647,6 +676,9 @@ impl SharedBuildAccumulator { } PartitionStatus::CanceledUnknown => { has_canceled_unknown = true; + // A canceled partition's build content is unknown, so it + // may hold a NULL key. + keys_have_null = true; } PartitionStatus::Pending => { return datafusion_common::internal_err!( @@ -692,38 +724,59 @@ impl SharedBuildAccumulator { }; self.dynamic_filter - .update(self.null_aware_filter(filter_expr))?; + .update(self.preserve_probe_nulls(filter_expr, keys_have_null)?)?; } } Ok(()) } - /// Wraps a pushdown filter so a null-aware anti join keeps its probe-side NULL rows. + /// Keeps probe rows with a NULL key when the join semantics need them. /// - /// The build-side predicate drops probe rows whose key is NULL, but `NOT IN` three-valued - /// logic needs that NULL to reach the join. OR-ing `probe_key IS NULL` preserves the dynamic - /// filter's selectivity for non-NULL rows while letting the NULL through. - fn null_aware_filter( + /// The build-side predicate drops probe rows whose key is NULL. A null-aware anti join + /// (`NOT IN`) needs that NULL to reach the join so three-valued logic can collapse the + /// result, and a null-equal join needs it to match a build-side NULL. OR-ing `key IS NULL` + /// keeps those rows while preserving the filter's selectivity for the rest; the join refines + /// whatever the widened filter lets through. + fn preserve_probe_nulls( &self, filter_expr: Arc, - ) -> Arc { - if !self.null_aware { - return filter_expr; + build_keys_have_null: bool, + ) -> Result> { + // A null-aware anti join needs every probe NULL no matter what the build holds: one + // probe NULL makes `NOT IN` unknown for every build row. A null-equal join needs probe + // NULLs only to match an actual build-side NULL, so a NULL-free build keeps the filter + // at full selectivity. + let needs_probe_nulls = self.null_aware + || (self.null_equality == NullEquality::NullEqualsNull + && build_keys_have_null); + if !needs_probe_nulls { + return Ok(filter_expr); + } + // Only a key that can actually be NULL needs the disjunct; a NOT NULL key never widens. + // Null-aware joins are single-key; null-equal joins can be multi-key, so OR every nullable + // key. If every key is NOT NULL the filter is left untouched, at full selectivity. + let mut any_key_is_null: Option> = None; + for key in &self.on_right { + // `nullable` fails only when a key is out of sync with the probe schema. That is + // a construction bug, so surface it instead of widening around it. + if !key.nullable(&self.probe_schema)? { + continue; + } + let is_null = + Arc::new(IsNullExpr::new(Arc::clone(key))) as Arc; + any_key_is_null = Some(match any_key_is_null { + Some(acc) => Arc::new(BinaryExpr::new(acc, Operator::Or, is_null)) as _, + None => is_null, + }); } - debug_assert_eq!( - self.on_right.len(), - 1, - "null_aware anti join must have exactly one probe key" - ); - let probe_key_is_null: Arc = - Arc::new(IsNullExpr::new(Arc::clone(&self.on_right[0]))); // Cheap null check first short-circuits before the costlier dynamic filter. - Arc::new(BinaryExpr::new( - probe_key_is_null, - Operator::Or, - filter_expr, - )) + Ok(match any_key_is_null { + Some(any_key_is_null) => { + Arc::new(BinaryExpr::new(any_key_is_null, Operator::Or, filter_expr)) + } + None => filter_expr, + }) } } @@ -756,6 +809,7 @@ pub(super) fn make_partitioned_accumulator_for_test( on_right: vec![], repartition_random_state: SeededRandomState::with_seed(1), probe_schema, + null_equality: NullEquality::NullEqualsNothing, null_aware: false, } } @@ -813,6 +867,7 @@ mod tests { on_right, repartition_random_state: SeededRandomState::with_seed(1), probe_schema: test_probe_schema(), + null_equality: NullEquality::NullEqualsNothing, null_aware: false, } } @@ -856,7 +911,11 @@ mod tests { } fn reported(pushdown: PushdownStrategy, bounds: PartitionBounds) -> PartitionStatus { - PartitionStatus::Reported(PartitionData { pushdown, bounds }) + PartitionStatus::Reported(PartitionData { + pushdown, + bounds, + keys_have_null: false, + }) } fn current_expr(acc: &SharedBuildAccumulator) -> PhysicalExprRef { @@ -1037,6 +1096,7 @@ mod tests { partition_id: 0, pushdown: PushdownStrategy::Empty, bounds: PartitionBounds::new(vec![]), + keys_have_null: false, }, ) .unwrap(); @@ -1073,4 +1133,119 @@ mod tests { assert!(matches!(partitions[0], PartitionStatus::CanceledUnknown)); assert_eq!(completed, 1); } + + fn null_semantics_accumulator( + probe_schema: Arc, + on_right: Vec, + null_equality: NullEquality, + null_aware: bool, + ) -> SharedBuildAccumulator { + SharedBuildAccumulator { + inner: Mutex::new(AccumulatorState { + data: AccumulatedBuildData::Partitioned { + partitions: vec![PartitionStatus::Pending; 1], + completed_partitions: 0, + }, + completion: CompletionState::Pending, + }), + completion_notify: Notify::new(), + dynamic_filter: Arc::new(DynamicFilterPhysicalExpr::new(vec![], lit(true))), + on_right, + repartition_random_state: SeededRandomState::with_seed(1), + probe_schema, + null_equality, + null_aware, + } + } + + fn null_equal_accumulator( + probe_schema: Arc, + on_right: Vec, + ) -> SharedBuildAccumulator { + null_semantics_accumulator( + probe_schema, + on_right, + NullEquality::NullEqualsNull, + false, + ) + } + + #[test] + fn preserve_probe_nulls_only_widens_nullable_keys() { + let probe_schema = Arc::new(Schema::new(vec![ + Field::new("k_nullable", DataType::Int32, true), + Field::new("k_not_null", DataType::Int32, false), + ])); + let on_right: Vec = vec![ + Arc::new(Column::new("k_nullable", 0)), + Arc::new(Column::new("k_not_null", 1)), + ]; + let acc = null_equal_accumulator(probe_schema, on_right); + + // Only the nullable key earns an IS NULL disjunct; the NOT NULL key is left out. + let widened = acc.preserve_probe_nulls(lit(true), true).unwrap(); + assert_eq!(format!("{widened}").matches("IS NULL").count(), 1); + } + + #[test] + fn preserve_probe_nulls_leaves_all_not_null_keys_untouched() { + let probe_schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int32, false), + Field::new("b", DataType::Int32, false), + ])); + let on_right: Vec = + vec![Arc::new(Column::new("a", 0)), Arc::new(Column::new("b", 1))]; + let acc = null_equal_accumulator(probe_schema, on_right); + + // Every key is NOT NULL, so there is nothing to OR in and the filter is returned as-is. + let filter = lit(true); + let result = acc.preserve_probe_nulls(Arc::clone(&filter), true).unwrap(); + assert_eq!(format!("{result}"), format!("{filter}")); + } + + #[test] + fn preserve_probe_nulls_rejects_out_of_sync_key() { + let probe_schema = + Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)])); + // The key's column index points past the probe schema: a construction bug that + // must surface as an error, not get widened around. + let on_right: Vec = vec![Arc::new(Column::new("b", 1))]; + let acc = null_equal_accumulator(probe_schema, on_right); + + assert!(acc.preserve_probe_nulls(lit(true), true).is_err()); + } + + #[test] + fn preserve_probe_nulls_skips_wrap_when_build_has_no_nulls() { + let probe_schema = + Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)])); + let on_right: Vec = vec![Arc::new(Column::new("a", 0))]; + let acc = null_equal_accumulator(probe_schema, on_right); + + // A NULL-free build has nothing for a probe NULL to null-match, so the + // filter keeps its full selectivity. + let filter = lit(true); + let result = acc + .preserve_probe_nulls(Arc::clone(&filter), false) + .unwrap(); + assert_eq!(format!("{result}"), format!("{filter}")); + } + + #[test] + fn preserve_probe_nulls_wraps_null_aware_regardless_of_build() { + let probe_schema = + Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)])); + let on_right: Vec = vec![Arc::new(Column::new("a", 0))]; + let acc = null_semantics_accumulator( + probe_schema, + on_right, + NullEquality::NullEqualsNothing, + true, + ); + + // One probe NULL collapses `NOT IN` for every build row, so the wrap must not + // depend on the build content. + let widened = acc.preserve_probe_nulls(lit(true), false).unwrap(); + assert_eq!(format!("{widened}").matches("IS NULL").count(), 1); + } } diff --git a/datafusion/physical-plan/src/joins/hash_join/stream.rs b/datafusion/physical-plan/src/joins/hash_join/stream.rs index 2aa6e69dff807..686939537e73e 100644 --- a/datafusion/physical-plan/src/joins/hash_join/stream.rs +++ b/datafusion/physical-plan/src/joins/hash_join/stream.rs @@ -559,16 +559,24 @@ impl HashJoinStream { .bounds .clone() .unwrap_or_else(|| PartitionBounds::new(vec![])); + // Arrow tracks null counts per array, so this costs no data scan. + let keys_have_null = left_data + .values() + .iter() + .any(|array| array.null_count() > 0); let build_data = match self.mode { PartitionMode::Partitioned => PartitionBuildData::Partitioned { partition_id: self.partition, pushdown, bounds, + keys_have_null, + }, + PartitionMode::CollectLeft => PartitionBuildData::CollectLeft { + pushdown, + bounds, + keys_have_null, }, - PartitionMode::CollectLeft => { - PartitionBuildData::CollectLeft { pushdown, bounds } - } PartitionMode::Auto => unreachable!( "PartitionMode::Auto should not be present at execution time. This is a bug in DataFusion, please report it!" ), @@ -1075,6 +1083,7 @@ mod tests { partition_id, pushdown: PushdownStrategy::Empty, bounds: PartitionBounds::new(vec![]), + keys_have_null: false, } } diff --git a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt index f1e787441d5e1..6fce96108fd46 100644 --- a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt +++ b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt @@ -1025,10 +1025,10 @@ drop table int_probe; ######## -# Dynamic filters must not be created for null-equal joins (IS NOT DISTINCT -# FROM, INTERSECT): min/max bounds and membership filters derived from the -# build side evaluate to NULL for probe-side NULL keys and would prune rows -# that can null-match a build-side NULL. +# Null-equal joins (IS NOT DISTINCT FROM, INTERSECT) keep dynamic filter pushdown. +# Min/max bounds and membership filters derived from the build side evaluate to NULL +# for a probe-side NULL key, so the pushed predicate carries an `IS NULL` disjunct that +# lets the probe NULL reach the join and null-match a build-side NULL. ######## statement ok @@ -1050,14 +1050,14 @@ SELECT nej_build.id, nej_probe.id FROM nej_build JOIN nej_probe ON nej_build.id 11 11 NULL NULL -# No DynamicFilter predicate may appear on the probe side of a null-equal join +# The probe side now carries a DynamicFilter for a null-equal join (widened with IS NULL at runtime) query TT EXPLAIN SELECT nej_build.id, nej_probe.id FROM nej_build JOIN nej_probe ON nej_build.id IS NOT DISTINCT FROM nej_probe.id ---- physical_plan 01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], NullsEqual: true 02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nej_build.parquet]]}, projection=[id], file_type=parquet -03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nej_probe.parquet]]}, projection=[id], file_type=parquet +03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nej_probe.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible statement ok drop table nej_build; @@ -1066,6 +1066,194 @@ statement ok drop table nej_probe; +# Multi-key null-equal join: the IS NULL disjunct covers every nullable key, so a probe row with a +# NULL in either key still reaches the join and null-matches the build side. +statement ok +COPY (SELECT * FROM (VALUES (1, 10), (2, NULL), (NULL, 30)) v(a, b)) TO 'test_files/scratch/push_down_filter_parquet/mnej_probe.parquet' STORED AS PARQUET; + +statement ok +COPY (SELECT * FROM (VALUES (1, 10), (2, NULL)) v(a, b)) TO 'test_files/scratch/push_down_filter_parquet/mnej_build.parquet' STORED AS PARQUET; + +statement ok +CREATE EXTERNAL TABLE mnej_probe STORED AS PARQUET LOCATION 'test_files/scratch/push_down_filter_parquet/mnej_probe.parquet'; + +statement ok +CREATE EXTERNAL TABLE mnej_build STORED AS PARQUET LOCATION 'test_files/scratch/push_down_filter_parquet/mnej_build.parquet'; + +query IIII rowsort +SELECT mnej_build.a, mnej_build.b, mnej_probe.a, mnej_probe.b FROM mnej_build JOIN mnej_probe ON (mnej_build.a IS NOT DISTINCT FROM mnej_probe.a) AND (mnej_build.b IS NOT DISTINCT FROM mnej_probe.b) +---- +1 10 1 10 +2 NULL 2 NULL + +statement ok +set datafusion.explain.analyze_categories = 'rows'; + +# After execution the populated filter shows the applied predicate: an IS NULL disjunct +# per key ahead of the build-side membership check, because the build holds a NULL. +query TT +EXPLAIN ANALYZE SELECT mnej_build.a, mnej_build.b, mnej_probe.a, mnej_probe.b FROM mnej_build JOIN mnej_probe ON (mnej_build.a IS NOT DISTINCT FROM mnej_probe.a) AND (mnej_build.b IS NOT DISTINCT FROM mnej_probe.b) +---- +Plan with Metrics +01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0), (b@1, b@1)], NullsEqual: true, 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=3, avg_fanout=100% (2/2), probe_hit_rate=66.67% (2/3)] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/mnej_build.parquet]]}, projection=[a, b], 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=16.42% (133/810)] +03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/mnej_probe.parquet]]}, projection=[a, b], file_type=parquet, predicate=DynamicFilter [ a@0 IS NULL OR b@1 IS NULL OR a@0 >= 1 AND a@0 <= 2 AND b@1 >= 10 AND b@1 <= 10 AND struct(a@0, b@1) IN (SET) ([{c0:1,c1:10}, {c0:2,c1:}]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@0 > 0 OR b_null_count@1 > 0 OR a_null_count@0 != row_count@3 AND a_max@2 >= 1 AND a_null_count@0 != row_count@3 AND a_min@4 <= 2 AND b_null_count@1 != row_count@3 AND b_max@5 >= 10 AND b_null_count@1 != row_count@3 AND b_min@6 <= 10, required_guarantees=[], 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=3, pushdown_rows_pruned=0, predicate_cache_inner_records=6, predicate_cache_records=6, scan_efficiency_ratio=18.16% (148/815)] + +statement ok +reset datafusion.explain.analyze_categories; + +statement ok +drop table mnej_build; + +statement ok +drop table mnej_probe; + + +# A NULL-free build has nothing for a probe NULL to null-match, so the pushed filter +# skips the IS NULL widening and keeps its full selectivity. +statement ok +COPY (SELECT * FROM (VALUES (11), (22)) v(id)) TO 'test_files/scratch/push_down_filter_parquet/nnb_build.parquet' STORED AS PARQUET; + +statement ok +COPY (SELECT * FROM (VALUES (11), (33), (NULL)) v(id)) TO 'test_files/scratch/push_down_filter_parquet/nnb_probe.parquet' STORED AS PARQUET; + +statement ok +CREATE EXTERNAL TABLE nnb_build STORED AS PARQUET LOCATION 'test_files/scratch/push_down_filter_parquet/nnb_build.parquet'; + +statement ok +CREATE EXTERNAL TABLE nnb_probe STORED AS PARQUET LOCATION 'test_files/scratch/push_down_filter_parquet/nnb_probe.parquet'; + +query II rowsort +SELECT nnb_build.id, nnb_probe.id FROM nnb_build JOIN nnb_probe ON nnb_build.id IS NOT DISTINCT FROM nnb_probe.id +---- +11 11 + +statement ok +set datafusion.explain.analyze_categories = 'rows'; + +# No IS NULL disjunct in the populated filter: the probe NULL can be pruned safely. +query TT +EXPLAIN ANALYZE SELECT nnb_build.id, nnb_probe.id FROM nnb_build JOIN nnb_probe ON nnb_build.id IS NOT DISTINCT FROM nnb_probe.id +---- +Plan with Metrics +01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], NullsEqual: true, metrics=[output_rows=1, output_batches=1, array_map_created_count=1, build_input_batches=1, build_input_rows=2, 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/nnb_build.parquet]]}, projection=[id], 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=13.71% (68/496)] +03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nnb_probe.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ id@0 >= 11 AND id@0 <= 22 AND id@0 IN (SET) ([11, 22]) ], dynamic_rg_pruning=eligible, pruning_predicate=id_null_count@1 != row_count@2 AND id_max@0 >= 11 AND id_null_count@1 != row_count@2 AND id_min@3 <= 22 AND (id_null_count@1 != row_count@2 AND id_min@3 <= 11 AND 11 <= id_max@0 OR id_null_count@1 != row_count@2 AND id_min@3 <= 22 AND 22 <= id_max@0), required_guarantees=[id in (11, 22)], 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=1 total → 1 matched, page_index_rows_pruned=3 total → 3 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=2, predicate_cache_inner_records=3, predicate_cache_records=1, scan_efficiency_ratio=14.45% (74/512)] + +statement ok +reset datafusion.explain.analyze_categories; + +statement ok +drop table nnb_build; + +statement ok +drop table nnb_probe; + + +# A probe key declared NOT NULL skips the disjunct even when the build holds a NULL: +# no probe row can be NULL, so there is nothing to keep. +statement ok +COPY (SELECT * FROM (VALUES (11), (NULL)) v(id)) TO 'test_files/scratch/push_down_filter_parquet/nnp_build.parquet' STORED AS PARQUET; + +statement ok +COPY (SELECT * FROM (VALUES (11), (33)) v(id)) TO 'test_files/scratch/push_down_filter_parquet/nnp_probe.parquet' STORED AS PARQUET; + +statement ok +CREATE EXTERNAL TABLE nnp_build STORED AS PARQUET LOCATION 'test_files/scratch/push_down_filter_parquet/nnp_build.parquet'; + +statement ok +CREATE EXTERNAL TABLE nnp_probe (id BIGINT NOT NULL) STORED AS PARQUET LOCATION 'test_files/scratch/push_down_filter_parquet/nnp_probe.parquet'; + +# The build NULL matches nothing here: the probe cannot produce a NULL. +query II rowsort +SELECT nnp_build.id, nnp_probe.id FROM nnp_build JOIN nnp_probe ON nnp_build.id IS NOT DISTINCT FROM nnp_probe.id +---- +11 11 + +statement ok +set datafusion.explain.analyze_categories = 'rows'; + +query TT +EXPLAIN ANALYZE SELECT nnp_build.id, nnp_probe.id FROM nnp_build JOIN nnp_probe ON nnp_build.id IS NOT DISTINCT FROM nnp_probe.id +---- +Plan with Metrics +01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], NullsEqual: true, metrics=[output_rows=1, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, 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/nnp_build.parquet]]}, projection=[id], 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=12.92% (65/503)] +03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nnp_probe.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ id@0 >= 11 AND id@0 <= 11 AND id@0 IN (SET) ([11, NULL]) ], dynamic_rg_pruning=eligible, pruning_predicate=id_null_count@1 != row_count@2 AND id_max@0 >= 11 AND id_null_count@1 != row_count@2 AND id_min@3 <= 11 AND (id_null_count@1 != row_count@2 AND id_min@3 <= 11 AND 11 <= id_max@0 OR id_null_count@1 != row_count@2 AND id_min@3 <= NULL AND NULL <= id_max@0), required_guarantees=[id in (11, NULL)], 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=1 total → 1 matched, page_index_rows_pruned=2 total → 2 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=1, predicate_cache_inner_records=2, predicate_cache_records=1, scan_efficiency_ratio=13.71% (68/496)] + +statement ok +reset datafusion.explain.analyze_categories; + +statement ok +drop table nnp_build; + +statement ok +drop table nnp_probe; + + +# Partitioned mode: the per-partition CASE filter gets the same IS NULL widening, so a +# probe NULL routed to a pruning branch still reaches the join. +statement ok +set datafusion.execution.target_partitions = 4; + +statement ok +set datafusion.optimizer.hash_join_single_partition_threshold = 0; + +statement ok +set datafusion.optimizer.hash_join_single_partition_threshold_rows = 0; + +# Two files per side so each scan starts with multiple partitions and the join +# runs real hash routing instead of collapsing to a single branch. +statement ok +COPY (SELECT * FROM (VALUES (11), (22)) v(id)) TO 'test_files/scratch/push_down_filter_parquet/pnej_probe/1.parquet' STORED AS PARQUET; + +statement ok +COPY (SELECT * FROM (VALUES (33), (NULL)) v(id)) TO 'test_files/scratch/push_down_filter_parquet/pnej_probe/2.parquet' STORED AS PARQUET; + +statement ok +COPY (SELECT * FROM (VALUES (11)) v(id)) TO 'test_files/scratch/push_down_filter_parquet/pnej_build/1.parquet' STORED AS PARQUET; + +statement ok +COPY (SELECT * FROM (VALUES (NULL)) v(id)) TO 'test_files/scratch/push_down_filter_parquet/pnej_build/2.parquet' STORED AS PARQUET; + +statement ok +CREATE EXTERNAL TABLE pnej_probe STORED AS PARQUET LOCATION 'test_files/scratch/push_down_filter_parquet/pnej_probe/'; + +statement ok +CREATE EXTERNAL TABLE pnej_build STORED AS PARQUET LOCATION 'test_files/scratch/push_down_filter_parquet/pnej_build/'; + +query TT +EXPLAIN SELECT pnej_build.id, pnej_probe.id FROM pnej_build JOIN pnej_probe ON pnej_build.id IS NOT DISTINCT FROM pnej_probe.id +---- +physical_plan +01)HashJoinExec: mode=Partitioned, join_type=Inner, on=[(id@0, id@0)], NullsEqual: true +02)--RepartitionExec: partitioning=Hash([id@0], 4), input_partitions=2 +03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/pnej_build/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/pnej_build/2.parquet]]}, projection=[id], file_type=parquet +04)--RepartitionExec: partitioning=Hash([id@0], 4), input_partitions=2 +05)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/pnej_probe/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/pnej_probe/2.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible + +query II rowsort +SELECT pnej_build.id, pnej_probe.id FROM pnej_build JOIN pnej_probe ON pnej_build.id IS NOT DISTINCT FROM pnej_probe.id +---- +11 11 +NULL NULL + +statement ok +drop table pnej_build; + +statement ok +drop table pnej_probe; + +statement ok +set datafusion.execution.target_partitions = 4; + +statement ok +RESET datafusion.optimizer.hash_join_single_partition_threshold; + +statement ok +RESET datafusion.optimizer.hash_join_single_partition_threshold_rows; + + ######## # Regression test for build-NULL + emptied-probe interaction in null-aware LeftAnti joins. #