diff --git a/datafusion/optimizer/src/push_down_filter.rs b/datafusion/optimizer/src/push_down_filter.rs index ab3ee7a68bbe2..b95bf3f7df174 100644 --- a/datafusion/optimizer/src/push_down_filter.rs +++ b/datafusion/optimizer/src/push_down_filter.rs @@ -58,7 +58,8 @@ use crate::extract_leaf_expressions::is_pure_extraction_projection; use crate::optimizer::ApplyOrder; use crate::simplify_expressions::{reorder_predicates, simplify_predicates}; use crate::utils::{ - ColumnReference, has_all_column_refs, is_restrict_null_predicate, schema_columns, + ColumnReference, has_all_column_refs, is_repeatable, is_restrict_null_predicate, + schema_columns, }; use crate::{OptimizerConfig, OptimizerRule}; use datafusion_expr::ExpressionPlacement; @@ -642,9 +643,121 @@ fn infer_join_predicates( &mut inferred_predicates, )?; + // Predicates can already be below the join, for example in a derived table + // or a plan supplied by another optimizer. Keep them in place and infer + // additional restrictions on the other input. + if join_type == JoinType::Inner && !join_col_keys.is_empty() { + let left = join_input_predicates( + &join.left, + join_col_keys.iter().map(|(left, _)| (*left).clone()), + )?; + let right = join_input_predicates( + &join.right, + join_col_keys.iter().map(|(_, right)| (*right).clone()), + )?; + let mut from_inputs = InferredPredicates::new(join_type); + infer_join_predicates_impl::( + &join_col_keys, + &left.iter().cloned().collect::>(), + &mut from_inputs, + )?; + infer_join_predicates_impl::( + &join_col_keys, + &right.iter().cloned().collect::>(), + &mut from_inputs, + )?; + inferred_predicates.predicates.extend( + from_inputs.predicates.into_iter().filter(|predicate| { + !left.contains(predicate) && !right.contains(predicate) + }), + ); + } + Ok(inferred_predicates.predicates) } +/// Find predicates guaranteed by an input, expressed in its join-key columns. +/// Only follow filters and column-preserving projections/aliases. Other nodes +/// may change values or introduce rows that do not satisfy a descendant filter. +fn join_input_predicates( + mut input: &LogicalPlan, + columns: impl Iterator, +) -> Result> { + let mut columns = columns + .map(|column| (column.clone(), column)) + .collect::>(); + let mut predicates = IndexSet::new(); + while !columns.is_empty() { + let candidates = match input { + LogicalPlan::Filter(filter) => split_conjunction(&filter.predicate), + LogicalPlan::Projection(projection) => { + let mut projected_columns = HashMap::new(); + for (expr, (qualifier, field)) in + projection.expr.iter().zip(projection.schema.iter()) + { + let output = Column::new(qualifier.cloned(), field.name()); + if let Some(column) = columns.get(&output) + && let Expr::Column(input_column) = unalias(expr) + { + projected_columns.insert(input_column.clone(), column.clone()); + } + } + columns = projected_columns; + input = &projection.input; + continue; + } + LogicalPlan::SubqueryAlias(alias) => { + let mut aliased_columns = HashMap::new(); + for ((qualifier, field), (input_qualifier, input_field)) in + alias.schema.iter().zip(alias.input.schema().iter()) + { + let output = Column::new(qualifier.cloned(), field.name()); + if let Some(column) = columns.get(&output) { + aliased_columns.insert( + Column::new(input_qualifier.cloned(), input_field.name()), + column.clone(), + ); + } + } + columns = aliased_columns; + input = &alias.input; + continue; + } + LogicalPlan::TableScan(scan) if !scan.filters.is_empty() => { + let filters = scan.filters.iter().collect::>(); + let support = scan.source.supports_filters_pushdown(&filters)?; + assert_eq_or_internal_err!(filters.len(), support.len()); + // Inexact scan filters can return false positives. Only exact + // filters establish a fact without a residual Filter node. + filters + .into_iter() + .zip(support) + .filter(|(_, support)| *support == TableProviderFilterPushDown::Exact) + .flat_map(|(predicate, _)| split_conjunction(predicate)) + .collect() + } + _ => break, + }; + let replacements = columns.iter().collect::>(); + for predicate in candidates { + let refs = predicate.column_refs(); + if refs.is_empty() + || !refs.iter().all(|column| columns.contains_key(*column)) + || !is_repeatable(predicate) + { + continue; + } + predicates.insert(replace_col(predicate.clone(), &replacements)?); + } + if let LogicalPlan::Filter(filter) = input { + input = &filter.input; + } else { + break; + } + } + Ok(predicates) +} + /// Inferred predicates collector. /// When the JoinType is not Inner, we need to detect whether the inferred predicate can strictly /// filter out NULL, otherwise ignore it. e.g. @@ -3517,6 +3630,61 @@ mod tests { .build() } + // SQL cannot construct a scan with an inexact filter and no residual Filter. + // Such a scan must not be used as proof that the predicate holds. + #[test] + fn join_input_scan_filter_support() -> Result<()> { + for support in [ + TableProviderFilterPushDown::Exact, + TableProviderFilterPushDown::Inexact, + TableProviderFilterPushDown::Unsupported, + ] { + let left = table_scan_with_pushdown_provider_builder( + support.clone(), + vec![col("a").eq(lit(1i32))], + None, + )? + .alias("l")? + .build()?; + let right = table_scan_with_pushdown_provider_builder( + TableProviderFilterPushDown::Exact, + vec![], + None, + )? + .alias("r")? + .build()?; + let plan = LogicalPlanBuilder::from(left) + .join( + right, + JoinType::Inner, + (vec![Column::from("l.a")], vec![Column::from("r.a")]), + None, + )? + .build()?; + let rewritten = + PushDownFilter::new().rewrite(plan.clone(), &OptimizerContext::new())?; + if support == TableProviderFilterPushDown::Exact { + assert!(rewritten.transformed); + let optimizer = + Optimizer::with_rules(vec![Arc::new(PushDownFilter::new())]); + let optimized = + optimizer.optimize(plan, &OptimizerContext::new(), observe)?; + assert_snapshot!(optimized, @r#" + Inner Join: l.a = r.a + SubqueryAlias: l + TableScan: test, full_filters=[a = Int32(1)] + SubqueryAlias: r + TableScan: test, full_filters=[a = Int32(1)] + "#); + assert_plan_not_transformed!(optimized); + } else { + assert!(!rewritten.transformed); + assert_eq!(rewritten.data, plan); + } + } + Ok(()) + } + #[test] fn filter_with_table_provider_exact() -> Result<()> { let plan = table_scan_with_pushdown_provider(TableProviderFilterPushDown::Exact)?; diff --git a/datafusion/sqllogictest/test_files/join_input_predicates.slt b/datafusion/sqllogictest/test_files/join_input_predicates.slt new file mode 100644 index 0000000000000..be07727e61bea --- /dev/null +++ b/datafusion/sqllogictest/test_files/join_input_predicates.slt @@ -0,0 +1,280 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership +# SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements + +statement ok +CREATE TABLE a (k VARCHAR, value INT); + +statement ok +CREATE TABLE b (k VARCHAR, value INT); + +statement ok +INSERT INTO a VALUES ('k1', 1), ('k2', 2), (NULL, 3); + +statement ok +INSERT INTO b VALUES ('k1', 10), ('k1', 11), ('k2', 20), (NULL, 30); + +statement ok +SET datafusion.explain.logical_plan_only = true; + +# A predicate inside a derived table must reach the other join input. +query TT +EXPLAIN SELECT * FROM (SELECT * FROM a WHERE k = 'k1') a +JOIN b ON a.k = b.k; +---- +logical_plan +01)Inner Join: a.k = b.k +02)--SubqueryAlias: a +03)----Filter: a.k = Utf8View("k1") +04)------TableScan: a projection=[k, value] +05)--Filter: b.k = Utf8View("k1") +06)----TableScan: b projection=[k, value] + +query TITI rowsort +SELECT * FROM (SELECT * FROM a WHERE k = 'k1') a +JOIN b ON a.k = b.k; +---- +k1 1 k1 10 +k1 1 k1 11 + +# Infer from the right input as well. +query TT +EXPLAIN SELECT * FROM a JOIN (SELECT * FROM b WHERE k = 'k1') b ON a.k = b.k; +---- +logical_plan +01)Inner Join: a.k = b.k +02)--Filter: a.k = Utf8View("k1") +03)----TableScan: a projection=[k, value] +04)--SubqueryAlias: b +05)----Filter: b.k = Utf8View("k1") +06)------TableScan: b projection=[k, value] + +query TITI rowsort +SELECT * FROM a JOIN (SELECT * FROM b WHERE k = 'k1') b ON a.k = b.k; +---- +k1 1 k1 10 +k1 1 k1 11 + +# Map a renamed column through a projection and a CTE alias. +query TT +EXPLAIN WITH selected AS (SELECT k AS id, value FROM a WHERE k = 'k1') +SELECT * FROM selected s JOIN b ON s.id = b.k; +---- +logical_plan +01)Inner Join: s.id = b.k +02)--SubqueryAlias: s +03)----SubqueryAlias: selected +04)------Projection: a.k AS id, a.value +05)--------Filter: a.k = Utf8View("k1") +06)----------TableScan: a projection=[k, value] +07)--Filter: b.k = Utf8View("k1") +08)----TableScan: b projection=[k, value] + +query TITI rowsort +WITH selected AS (SELECT k AS id, value FROM a WHERE k = 'k1') +SELECT * FROM selected s JOIN b ON s.id = b.k; +---- +k1 1 k1 10 +k1 1 k1 11 + +# Copy the key conjunct, but not a predicate on another column. +query TT +EXPLAIN SELECT * FROM (SELECT * FROM a WHERE k = 'k1' AND value = 1) a +JOIN b ON a.k = b.k; +---- +logical_plan +01)Inner Join: a.k = b.k +02)--SubqueryAlias: a +03)----Filter: a.k = Utf8View("k1") AND a.value = Int32(1) +04)------TableScan: a projection=[k, value] +05)--Filter: b.k = Utf8View("k1") +06)----TableScan: b projection=[k, value] + +query TITI rowsort +SELECT * FROM (SELECT * FROM a WHERE k = 'k1' AND value = 1) a +JOIN b ON a.k = b.k; +---- +k1 1 k1 10 +k1 1 k1 11 + +# Ranges can be inferred too. +query TT +EXPLAIN SELECT * FROM (SELECT * FROM a WHERE k >= 'k2') a JOIN b ON a.k = b.k; +---- +logical_plan +01)Inner Join: a.k = b.k +02)--SubqueryAlias: a +03)----Filter: a.k >= Utf8View("k2") +04)------TableScan: a projection=[k, value] +05)--Filter: b.k >= Utf8View("k2") +06)----TableScan: b projection=[k, value] + +query TITI rowsort +SELECT * FROM (SELECT * FROM a WHERE k >= 'k2') a JOIN b ON a.k = b.k; +---- +k2 2 k2 20 + +# A mixed predicate is not a restriction on the key alone. +query TT +EXPLAIN SELECT * FROM (SELECT * FROM a WHERE k = 'k1' OR value = 2) a +JOIN b ON a.k = b.k; +---- +logical_plan +01)Inner Join: a.k = b.k +02)--SubqueryAlias: a +03)----Filter: a.k = Utf8View("k1") OR a.value = Int32(2) +04)------TableScan: a projection=[k, value] +05)--TableScan: b projection=[k, value] + +query TITI rowsort +SELECT * FROM (SELECT * FROM a WHERE k = 'k1' OR value = 2) a +JOIN b ON a.k = b.k; +---- +k1 1 k1 10 +k1 1 k1 11 +k2 2 k2 20 + +# Do not translate a predicate through a computed key. +query TT +EXPLAIN SELECT * FROM (SELECT k || 'x' AS k, value FROM a WHERE k = 'k1') a +JOIN b ON a.k = b.k; +---- +logical_plan +01)Inner Join: a.k = b.k +02)--SubqueryAlias: a +03)----Projection: a.k || Utf8View("x") AS k, a.value +04)------Filter: a.k = Utf8View("k1") +05)--------TableScan: a projection=[k, value] +06)--TableScan: b projection=[k, value] + +query TITI rowsort +SELECT * FROM (SELECT k || 'x' AS k, value FROM a WHERE k = 'k1') a +JOIN b ON a.k = b.k; +---- + + +# Do not filter preserved rows using the nullable input of a left join. +query TT +EXPLAIN SELECT * FROM a LEFT JOIN (SELECT * FROM b WHERE k = 'k1') b ON a.k = b.k; +---- +logical_plan +01)Left Join: a.k = b.k +02)--TableScan: a projection=[k, value] +03)--SubqueryAlias: b +04)----Filter: b.k = Utf8View("k1") +05)------TableScan: b projection=[k, value] + +query TITI rowsort +SELECT * FROM a LEFT JOIN (SELECT * FROM b WHERE k = 'k1') b ON a.k = b.k; +---- +NULL 3 NULL NULL +k1 1 k1 10 +k1 1 k1 11 +k2 2 NULL NULL + +# Do not treat a predicate below an aggregate as a predicate on its output. +query TT +EXPLAIN SELECT * FROM (SELECT count(*) AS value FROM a WHERE value = 1) a +JOIN b ON a.value = b.value; +---- +logical_plan +01)Inner Join: a.value = CAST(b.value AS Int64) +02)--SubqueryAlias: a +03)----Projection: count(Int64(1)) AS count(*) AS value +04)------Aggregate: groupBy=[[]], aggr=[[count(Int64(1))]] +05)--------Projection: +06)----------Filter: a.value = Int32(1) +07)------------TableScan: a projection=[value] +08)--TableScan: b projection=[k, value] + +query ITI rowsort +SELECT * FROM (SELECT count(*) AS value FROM a WHERE value = 1) a +JOIN b ON a.value = b.value; +---- + + +# Volatile predicates must not be evaluated independently on both inputs. +query TT +EXPLAIN SELECT * FROM (SELECT * FROM a WHERE k = CAST(random() AS VARCHAR)) a +JOIN b ON a.k = b.k; +---- +logical_plan +01)Inner Join: a.k = b.k +02)--SubqueryAlias: a +03)----Filter: a.k = CAST(random() AS Utf8View) +04)------TableScan: a projection=[k, value] +05)--TableScan: b projection=[k, value] + +# Preserve LIMIT semantics when pushing the inferred predicate into its input. +query TT +EXPLAIN SELECT * FROM (SELECT * FROM a WHERE k = 'k1') a +JOIN (SELECT * FROM b ORDER BY k DESC NULLS LAST LIMIT 1) b ON a.k = b.k; +---- +logical_plan +01)Inner Join: a.k = b.k +02)--SubqueryAlias: a +03)----Filter: a.k = Utf8View("k1") +04)------TableScan: a projection=[k, value] +05)--SubqueryAlias: b +06)----Filter: b.k = Utf8View("k1") +07)------Sort: b.k DESC NULLS LAST, fetch=1 +08)--------TableScan: b projection=[k, value] + +query TITI +SELECT * FROM (SELECT * FROM a WHERE k = 'k1') a +JOIN (SELECT * FROM b ORDER BY k DESC NULLS LAST LIMIT 1) b ON a.k = b.k; +---- + +# Infer both join-key restrictions independently. +query TT +EXPLAIN SELECT * FROM (SELECT * FROM a WHERE k = 'k1' AND value = 1) a +JOIN b ON a.k = b.k AND a.value = b.value; +---- +logical_plan +01)Inner Join: a.k = b.k, a.value = b.value +02)--SubqueryAlias: a +03)----Filter: a.k = Utf8View("k1") AND a.value = Int32(1) +04)------TableScan: a projection=[k, value] +05)--Filter: b.k = Utf8View("k1") AND b.value = Int32(1) +06)----TableScan: b projection=[k, value] + +# Null-safe equality can match NULL keys; preserve those results. +query TT +EXPLAIN SELECT * FROM (SELECT * FROM a WHERE k IS NULL) a +JOIN b ON a.k IS NOT DISTINCT FROM b.k; +---- +logical_plan +01)Inner Join: a.k = b.k +02)--SubqueryAlias: a +03)----Filter: a.k IS NULL +04)------TableScan: a projection=[k, value] +05)--Filter: b.k IS NULL +06)----TableScan: b projection=[k, value] + +query TITI +SELECT * FROM (SELECT * FROM a WHERE k IS NULL) a +JOIN b ON a.k IS NOT DISTINCT FROM b.k; +---- +NULL 3 NULL 30 + +# Applying the rule repeatedly must not accumulate inferred filters. +statement ok +SET datafusion.optimizer.max_passes = 10; + +query TT +EXPLAIN SELECT * FROM (SELECT * FROM a WHERE k = 'k1') a +JOIN b ON a.k = b.k; +---- +logical_plan +01)Inner Join: a.k = b.k +02)--SubqueryAlias: a +03)----Filter: a.k = Utf8View("k1") +04)------TableScan: a projection=[k, value] +05)--Filter: b.k = Utf8View("k1") +06)----TableScan: b projection=[k, value] + +statement ok +SET datafusion.optimizer.max_passes = 3; + +statement ok +SET datafusion.explain.logical_plan_only = false; diff --git a/datafusion/sqllogictest/test_files/range_partitioning.slt b/datafusion/sqllogictest/test_files/range_partitioning.slt index f389e265f84ef..113e1623343b0 100644 --- a/datafusion/sqllogictest/test_files/range_partitioning.slt +++ b/datafusion/sqllogictest/test_files/range_partitioning.slt @@ -1888,7 +1888,8 @@ SELECT range_key, SUM(value) FROM ( # TEST 48: Hash Join Dynamic Filter Pushdown on Compatible Range Inputs # The Parquet-backed probe accepts the dynamic filter. The build has rows only in # partitions 0 and 2, and both chose an IN list, so they merge into one IN list -# with combined bounds instead of a per-partition CASE. +# with combined bounds instead of a per-partition CASE. The probe also receives +# the static IN predicate inferred from the filtered build input. ########## statement ok @@ -1915,7 +1916,7 @@ JOIN range_partitioned p ON b.range_key = p.range_key; physical_plan 01)HashJoinExec: mode=Partitioned, join_type=Inner, on=[(range_key@0, range_key@0)], projection=[range_key@0, value@1, value@3] 02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet, predicate=range_key@0 IN (SET) ([5, 20]), pruning_predicate=range_key_null_count@2 != row_count@3 AND range_key_min@0 <= 5 AND 5 <= range_key_max@1 OR range_key_null_count@2 != row_count@3 AND range_key_min@0 <= 20 AND 20 <= range_key_max@1, required_guarantees=[range_key in (20, 5)] -03)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible +03)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet, predicate=range_key@0 IN (SET) ([5, 20]) AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=range_key_null_count@2 != row_count@3 AND range_key_min@0 <= 5 AND 5 <= range_key_max@1 OR range_key_null_count@2 != row_count@3 AND range_key_min@0 <= 20 AND 20 <= range_key_max@1, required_guarantees=[range_key in (20, 5)] query TT EXPLAIN ANALYZE SELECT b.range_key, b.value, p.value @@ -1929,7 +1930,7 @@ JOIN range_partitioned p ON b.range_key = p.range_key; Plan with Metrics 01)HashJoinExec: mode=Partitionedmetrics=[output_rows=2,] 02)--DataSourceExec: file_type=parquet, predicate=range_key@0 IN (SET) ([5, 20])metrics=[output_rows=2,] -03)--DataSourceExec: file_type=parquet, predicate=DynamicFilter [ range_key@0 >= 5 AND range_key@0 <= 20 AND range_key@0 IN (SET) ([5, 20]) ]metrics=[output_rows=2,] +03)--DataSourceExec: file_type=parquet, predicate=range_key@0 IN (SET) ([5, 20]) AND DynamicFilter [ range_key@0 >= 5 AND range_key@0 <= 20 AND range_key@0 IN (SET) ([5, 20]) ]metrics=[output_rows=2,] query III SELECT b.range_key, b.value, p.value