feat: add OptionalFilterGate and optional_filter_mode config

Add a runtime gate that pauses optional filters (filters that are not
needed for correctness, such as hash join and TopK dynamic filters) when
they remove too few rows. The gate is a per-stream state machine
(Evaluate / Paused with exponential backoff) that restarts evaluation
when the filter changes. The gate finds the dynamic filters one time with
`DynamicFilterTracking::classify` and then polls their subscriptions, so
a check does not walk the filter tree. Gates of one plan site share
lock-free pooled statistics, which seed the state of new gates.

Add a crate-private generation tag to `DynamicFilterTracker` (the sum of
the latest observed generations of the dynamic filters, including
complete ones), so the pooled statistics can tag verdicts by filter
version without calling `snapshot_generation`.

Add the `datafusion.execution.optional_filter_mode` (always | adaptive |
pruning_only, default always) and
`datafusion.execution.optional_filter_max_pass_ratio` (default 0.8)
options. No operator uses the gate yet, so behavior does not change.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: adriangb

datafusion/common/src/config.rs

@@ -881,6 +881,61 @@ impl Display for MapKeyDedupPolicy {
     }
 }
 
+/// How DataFusion evaluates optional filters.
+///
+/// Optional filters are filters that are not needed for correctness, such as
+/// dynamic filters pushed down by hash joins and TopK. See
+/// [`ExecutionOptions::optional_filter_mode`].
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
+pub enum OptionalFilterMode {
+    /// Evaluate optional filters like any other pushed-down filter.
+    #[default]
+    Always,
+    /// Pause optional filters that remove too few rows. Try them again at
+    /// intervals to find out if they became selective.
+    Adaptive,
+    /// Use optional filters only for statistics pruning (for example of
+    /// files, row groups and pages). Never evaluate them row by row.
+    PruningOnly,
+}
+
+impl FromStr for OptionalFilterMode {
+    type Err = DataFusionError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s.to_ascii_lowercase().as_str() {
+            "always" => Ok(Self::Always),
+            "adaptive" => Ok(Self::Adaptive),
+            "pruning_only" => Ok(Self::PruningOnly),
+            other => Err(DataFusionError::Configuration(format!(
+                "Invalid optional filter mode: {other}. Expected one of: always, adaptive, pruning_only"
+            ))),
+        }
+    }
+}
+
+impl ConfigField for OptionalFilterMode {
+    fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
+        v.some(key, self, description)
+    }
+
+    fn set(&mut self, _: &str, value: &str) -> Result<()> {
+        *self = OptionalFilterMode::from_str(value)?;
+        Ok(())
+    }
+}
+
+impl Display for OptionalFilterMode {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let str = match self {
+            Self::Always => "always",
+            Self::Adaptive => "adaptive",
+            Self::PruningOnly => "pruning_only",
+        };
+        write!(f, "{str}")
+    }
+}
+
 impl From<SpillCompression> for Option<CompressionType> {
     fn from(c: SpillCompression) -> Self {
         match c {
@@ -1182,6 +1237,26 @@ config_namespace! {
         ///
         /// Disabled by default, set to a number greater than 0 for enabling it.
         pub hash_join_buffering_capacity: usize, default = 0
+
+        /// Controls how DataFusion evaluates filters that are not needed for
+        /// correctness, such as the dynamic filters that hash joins and TopK
+        /// push down into scans. `always` evaluates these filters like any
+        /// other pushed-down filter. `adaptive` pauses these filters when they
+        /// remove too few rows (see
+        /// `datafusion.execution.optional_filter_max_pass_ratio`), and tries
+        /// them again at intervals. `pruning_only` uses these filters only to
+        /// prune files, row groups and pages with statistics, and never
+        /// evaluates them row by row.
+        ///
+        /// This option is most important when
+        /// `datafusion.execution.parquet.pushdown_filters` is true, because then
+        /// the Parquet reader evaluates pushed-down filters row by row.
+        pub optional_filter_mode: OptionalFilterMode, default = OptionalFilterMode::Always
+
+        /// When `datafusion.execution.optional_filter_mode` is `adaptive`, pause
+        /// an optional filter when more than this fraction of the rows pass it.
+        /// Use a value between 0.0 and 1.0.
+        pub optional_filter_max_pass_ratio: f64, default = 0.8
     }
 }
 

datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs

@@ -755,6 +755,13 @@ pub(crate) struct DynamicFilterSubscription {
 }
 
 impl DynamicFilterSubscription {
+    /// The latest generation of the filter that this subscription observed:
+    /// the generation at [`DynamicFilterPhysicalExpr::subscribe`] time, or
+    /// the generation of the last change that [`Self::observe`] reported.
+    pub(crate) fn last_generation(&self) -> u64 {
+        self.last_generation
+    }
+
     /// Observe the latest state of the filter.
     ///
     /// Reports whether the filter's expression advanced since the previous call

datafusion/physical-expr/src/expressions/dynamic_filters/tracker.rs

@@ -69,29 +69,56 @@ impl DynamicFilterTracking {
     /// Walk `predicate` once and classify its dynamic-filter content,
     /// subscribing to every filter that is not yet complete.
     pub fn classify(predicate: &Arc<dyn PhysicalExpr>) -> Self {
+        Self::classify_with_generation_tag(predicate).0
+    }
+
+    /// Same as [`Self::classify`], but also returns the generation tag of
+    /// `predicate` at classification time.
+    ///
+    /// The tag is the wrapping sum of the generations of all dynamic filters
+    /// in `predicate` (0 for a [`Self::Static`] predicate). Two callers that
+    /// classify the same predicate get the same tag if no filter changed
+    /// between the two calls. For [`Self::Watching`], use
+    /// [`DynamicFilterTracker::generation_tag`] after
+    /// [`DynamicFilterTracker::changed`] returns `true` to get the new tag.
+    /// For the other variants the tag never changes.
+    pub(crate) fn classify_with_generation_tag(
+        predicate: &Arc<dyn PhysicalExpr>,
+    ) -> (Self, u64) {
         let mut subscriptions = Vec::new();
+        let mut completed_generations = 0u64;
         let mut found_any = false;
         predicate
             .apply(|expr| {
                 if let Some(filter) = expr.downcast_ref::<DynamicFilterPhysicalExpr>() {
                     found_any = true;
                     // Already-complete filters can never change again, so there
-                    // is no point subscribing to them.
-                    if !filter.is_complete() {
+                    // is no point subscribing to them. Keep their (final)
+                    // generation for the generation tag.
+                    if filter.is_complete() {
+                        completed_generations = completed_generations
+                            .wrapping_add(filter.current_generation());
+                    } else {
                         subscriptions.push(filter.subscribe());
                     }
                 }
                 Ok(TreeNodeRecursion::Continue)
             })
             .expect("traversal closure is infallible");
 
-        if !found_any {
+        let tracker = DynamicFilterTracker {
+            subscriptions,
+            completed_generations,
+        };
+        let generation_tag = tracker.generation_tag();
+        let tracking = if !found_any {
             DynamicFilterTracking::Static
-        } else if subscriptions.is_empty() {
+        } else if tracker.subscriptions.is_empty() {
             DynamicFilterTracking::AllComplete
         } else {
-            DynamicFilterTracking::Watching(DynamicFilterTracker { subscriptions })
-        }
+            DynamicFilterTracking::Watching(tracker)
+        };
+        (tracking, generation_tag)
     }
 
     /// `true` if the predicate contains any dynamic filter (complete or not),
@@ -123,6 +150,10 @@ pub struct DynamicFilterTracker {
     /// Subscriptions to the not-yet-complete dynamic filters. Entries are
     /// dropped as their filters complete, so the set only shrinks.
     subscriptions: Vec<DynamicFilterSubscription>,
+    /// Wrapping sum of the final generations of the complete filters: the
+    /// filters that were complete at classification time and the filters
+    /// whose subscriptions were dropped. See [`Self::generation_tag`].
+    completed_generations: u64,
 }
 
 impl DynamicFilterTracker {
@@ -134,14 +165,41 @@ impl DynamicFilterTracker {
     /// returns `false`.
     pub fn changed(&mut self) -> bool {
         let mut changed = false;
+        let completed_generations = &mut self.completed_generations;
         self.subscriptions.retain_mut(|subscription| {
             let change = subscription.observe();
             changed |= change.changed;
+            if change.complete {
+                // Keep the final generation so the generation tag does not
+                // change when the subscription is dropped.
+                *completed_generations =
+                    completed_generations.wrapping_add(subscription.last_generation());
+            }
             // Keep the subscription only while the filter can still change.
             !change.complete
         });
         changed
     }
+
+    /// A tag that identifies the generations of the watched filters, as
+    /// observed by the last call to [`Self::changed`] (or at classification
+    /// time, before the first call).
+    ///
+    /// The tag is the wrapping sum of the latest observed generation of every
+    /// dynamic filter in the predicate, complete or not. It is equal to the
+    /// tag that [`DynamicFilterTracking::classify_with_generation_tag`]
+    /// returns for the same predicate, if no filter changed since the last
+    /// observation. It does not change when a filter completes without an
+    /// update. A generation only increases, thus in practice two different
+    /// tags mean that a filter changed. Computing it does not walk the
+    /// predicate: it only reads the watched subscriptions.
+    pub(crate) fn generation_tag(&self) -> u64 {
+        self.subscriptions
+            .iter()
+            .fold(self.completed_generations, |tag, subscription| {
+                tag.wrapping_add(subscription.last_generation())
+            })
+    }
 }
 
 #[cfg(test)]
@@ -157,7 +215,7 @@ impl DynamicFilterTracker {
     }
 
     /// `true` once every watched filter has completed and been dropped.
-    fn is_exhausted(&self) -> bool {
+    pub(crate) fn is_exhausted(&self) -> bool {
         self.subscriptions.is_empty()
     }
 }
@@ -328,4 +386,74 @@ mod tests {
         assert!(!tracker.changed());
         assert!(tracker.is_exhausted());
     }
+
+    #[test]
+    fn generation_tag_of_static_predicate_is_zero() {
+        let (tracking, tag) =
+            DynamicFilterTracking::classify_with_generation_tag(&lit(true));
+        assert!(matches!(tracking, DynamicFilterTracking::Static));
+        assert_eq!(tag, 0);
+    }
+
+    #[test]
+    fn generation_tag_follows_observed_updates() {
+        let (predicate, filter) = dynamic_predicate();
+        let (mut tracking, initial_tag) =
+            DynamicFilterTracking::classify_with_generation_tag(&predicate);
+        let tracker = tracking.watcher().unwrap();
+        assert_eq!(tracker.generation_tag(), initial_tag);
+
+        // The tag changes only when the tracker observes the update.
+        filter.update(lit(false)).unwrap();
+        assert_eq!(tracker.generation_tag(), initial_tag);
+        assert!(tracker.changed());
+        let updated_tag = tracker.generation_tag();
+        assert_ne!(updated_tag, initial_tag);
+
+        // A new classification of the same predicate gets the same tag.
+        let (_, tag) = DynamicFilterTracking::classify_with_generation_tag(&predicate);
+        assert_eq!(tag, updated_tag);
+    }
+
+    #[test]
+    fn generation_tag_is_stable_when_filter_completes() {
+        let (predicate, filter) = dynamic_predicate();
+        let (mut tracking, _) =
+            DynamicFilterTracking::classify_with_generation_tag(&predicate);
+        let tracker = tracking.watcher().unwrap();
+
+        filter.update(lit(false)).unwrap();
+        assert!(tracker.changed());
+        let tag = tracker.generation_tag();
+
+        // Completion drops the subscription but keeps its final generation.
+        filter.mark_complete();
+        assert!(!tracker.changed());
+        assert!(tracker.is_exhausted());
+        assert_eq!(tracker.generation_tag(), tag);
+
+        // A predicate classified after completion gets the same tag.
+        let (tracking, complete_tag) =
+            DynamicFilterTracking::classify_with_generation_tag(&predicate);
+        assert!(matches!(tracking, DynamicFilterTracking::AllComplete));
+        assert_eq!(complete_tag, tag);
+    }
+
+    #[test]
+    fn generation_tag_of_coalesced_update_and_complete() {
+        let (predicate, filter) = dynamic_predicate();
+        let (mut tracking, initial_tag) =
+            DynamicFilterTracking::classify_with_generation_tag(&predicate);
+        let tracker = tracking.watcher().unwrap();
+
+        filter.update(lit(false)).unwrap();
+        filter.mark_complete();
+        assert!(tracker.changed());
+        assert!(tracker.is_exhausted());
+        assert_ne!(tracker.generation_tag(), initial_tag);
+
+        let (_, complete_tag) =
+            DynamicFilterTracking::classify_with_generation_tag(&predicate);
+        assert_eq!(tracker.generation_tag(), complete_tag);
+    }
 }

datafusion/physical-expr/src/lib.rs

@@ -36,6 +36,7 @@ pub mod equivalence;
 pub mod expressions;
 pub mod higher_order_function;
 pub mod intervals;
+pub mod optional_filter_gate;
 mod partitioning;
 mod physical_expr;
 pub mod planner;