shehab-ali commented on code in PR #27055:
URL: https://github.com/apache/datafusion/pull/27055#discussion_r4471656774
##########
datafusion/optimizer/src/push_down_filter.rs:
##########
@@ -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::<true, false>(
+ &join_col_keys,
+ &left.iter().cloned().collect::<Vec<_>>(),
+ &mut from_inputs,
+ )?;
+ infer_join_predicates_impl::<false, true>(
+ &join_col_keys,
+ &right.iter().cloned().collect::<Vec<_>>(),
+ &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<Item = Column>,
+) -> Result<IndexSet<Expr>> {
+ let mut columns = columns
+ .map(|column| (column.clone(), column))
+ .collect::<HashMap<_, _>>();
+ 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::<Vec<_>>();
+ 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::<HashMap<_, _>>();
+ for predicate in candidates {
+ let refs = predicate.column_refs();
+ if refs.is_empty()
+ || !refs.iter().all(|column| columns.contains_key(*column))
+ || !is_repeatable(predicate)
Review Comment:
Does `is_repeatable` alone guarantee that a predicate is safe to copy across
join keys? I think datafusion treats `-0.0` and `+0.0` as equal for joins, but
deterministic functions such as `atan2` distinguish them
```
CREATE TABLE a(k DOUBLE);
CRAETE TABLE b(k DOUBLE);
INSERT INTO a VALUES (CAST('-0.0' AS DOUBLE));
INSERT INTO b VALUES (CAST('0.0' AS DOUBLE));
SELECT *
FROM (SELECT * FROM a WHERE atan2(0.0, k) > 1) a
JOIN b ON a.k = b.k;
```
This should return one row: the left predicate is true and the keys match.
The new rule infers `atan2(0.0, b.k) > 1`, which is false for `+0.0`, which
would remov that valid match
probably worth adding a regression test for this case and restricting that
case
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
