Enable dynamic filter pushdown for LEFT/RIGHT/SEMI/ANTI/Mark joins; surface probe metadata in plans; add join-preservation docs

datafusion/common/src/config.rs

@@ -725,11 +725,43 @@ config_namespace! {
         /// during aggregations, if possible
         pub enable_topk_aggregation: bool, default = true
 
-        /// When set to true attempts to push down dynamic filters generated by operators into the file scan phase.
-        /// For example, for a query such as `SELECT * FROM t ORDER BY timestamp DESC LIMIT 10`, the optimizer
-        /// will attempt to push down the current top 10 timestamps that the TopK operator references into the file scans.
-        /// This means that if we already have 10 timestamps in the year 2025
-        /// any files that only have timestamps in the year 2024 can be skipped / pruned at various stages in the scan.
+        /// When set to true attempts to push down dynamic filters generated by operators
+        /// into the file scan phase. For example, for a query such as
+        /// `SELECT * FROM t ORDER BY timestamp DESC LIMIT 10`, the optimizer will attempt
+        /// to push down the current top 10 timestamps that the TopK operator references
+        /// into the file scans. This means that if we already have 10 timestamps in the
+        /// year 2025 any files that only have timestamps in the year 2024 can be skipped /
+        /// pruned at various stages in the scan.
+        ///
+        /// Dynamic filters are also produced by joins. At runtime, DataFusion applies
+        /// the filter to one input to prune work. `HashJoinExec` builds from its left
+        /// input and probes with its right input, but the dynamic filter target (the
+        /// side we prune) depends on the join type:
+        ///
+        /// | Join type                | Dynamic filter target |
+        /// |--------------------------|-----------------------|
+        /// | `Inner`, `Left`         | Right input            |
+        /// | `Right`                 | Left input             |
+        /// | `LeftSemi`, `LeftAnti`  | Left input             |
+        /// | `RightSemi`, `RightAnti`| Right input            |
+        /// | `LeftMark`              | Right input            |
+        /// | `RightMark`             | Left input             |
+        /// | `Full`                  | Not supported          |
+        ///
+        /// Non-equi join predicates do **not** generate dynamic filters; they require
+        /// range analysis and cross-conjunct reasoning (future work). Composite
+        /// predicates only derive filters from their equi-conjuncts, and rows with
+        /// `NULL` join keys (see [`crate::NullEquality::NullEqualsNothing`]) do not contribute
+        /// filter values. Enabling `optimizer.filter_null_join_keys` can remove such
+        /// rows up front.
+        ///
+        /// Pushdown is effective only when the file source supports predicate pushdown
+        /// (e.g. Parquet) and `execution.parquet.pushdown_filters` is `true`; formats
+        /// without predicate pushdown (CSV/JSON) see no benefit. See the upgrade guide
+        /// for additional details and examples. For example,
+        /// `SELECT * FROM fact LEFT JOIN dim ON fact.id = dim.id WHERE dim.region = 'US'`
+        /// will only read `fact` rows whose `id` values match `dim` rows where
+        /// `region = 'US'`.
         pub enable_dynamic_filter_pushdown: bool, default = true
 
         /// When set to true, the optimizer will insert filters before a join between

datafusion/common/src/join_type.rs

@@ -111,6 +111,63 @@ impl JoinType {
                 | JoinType::RightAnti
         )
     }
+    /// Returns true if the left side of this join preserves its input rows
+    /// for filters applied *after* the join.
+    #[inline]
+    pub const fn preserves_left_for_output_filters(self) -> bool {
+        matches!(
+            self,
+            JoinType::Inner
+                | JoinType::Left
+                | JoinType::LeftSemi
+                | JoinType::LeftAnti
+                | JoinType::LeftMark,
+        )
+    }
+
+    /// Returns true if the right side of this join preserves its input rows
+    /// for filters applied *after* the join.
+    #[inline]
+    pub const fn preserves_right_for_output_filters(self) -> bool {
+        matches!(
+            self,
+            JoinType::Inner
+                | JoinType::Right
+                | JoinType::RightSemi
+                | JoinType::RightAnti
+                | JoinType::RightMark,
+        )
+    }
+
+    /// Returns true if the left side of this join preserves its input rows
+    /// for filters in the join condition (ON-clause filters).
+    #[inline]
+    pub const fn preserves_left_for_on_filters(self) -> bool {
+        matches!(
+            self,
+            JoinType::Inner
+                | JoinType::Right
+                | JoinType::LeftSemi
+                | JoinType::RightSemi
+                | JoinType::RightAnti
+                | JoinType::RightMark,
+        )
+    }
+
+    /// Returns true if the right side of this join preserves its input rows
+    /// for filters in the join condition (ON-clause filters).
+    #[inline]
+    pub const fn preserves_right_for_on_filters(self) -> bool {
+        matches!(
+            self,
+            JoinType::Inner
+                | JoinType::Left
+                | JoinType::LeftSemi
+                | JoinType::RightSemi
+                | JoinType::LeftAnti
+                | JoinType::LeftMark,
+        )
+    }
 }
 
 impl Display for JoinType {