feat(gate): use the measured work of the filter producer as the saving

The gate assumed that each removed row saves `min_saving_ns_per_row`
(20 ns) after the filter. For hash join dynamic filters this is the probe
work of the join, and it is much smaller in star joins with small
dimension tables: 3.5 to 8 ns for each probe row on the TPC-DS SF1
`date_dim` joins (Q65, Q67), 2 ns on Q90, 17 ns on TPC-H Q9. Filters that
cost 3 to 7 ns for each row and remove 80% of the rows thus stayed on,
and cost more than the join work they saved (8-18% slower than
`pruning_only` on the bot).

`RemovedRowWork` (in `filter_stats`) is the work that the producer of a
filter does for each row that the filter removes, as the producer
measures it. Each `DynamicFilterPhysicalExpr` has one, shared by all its
derived filters. The gate uses the smallest measured work of the dynamic
filters in its filter as the saving of a removed row, and
`min_saving_ns_per_row` only until the producer has measured
`MIN_OBSERVED_ROWS` rows (a prior).

PR: #25674

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

Author: adriangb

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

@@ -21,6 +21,7 @@ use std::{fmt::Display, hash::Hash, sync::Arc};
 use tokio::sync::watch;
 
 use crate::PhysicalExpr;
+use crate::filter_stats::RemovedRowWork;
 use arrow::datatypes::{DataType, Schema};
 #[cfg(feature = "proto")]
 use datafusion_common::internal_datafusion_err;
@@ -94,6 +95,10 @@ pub struct DynamicFilterPhysicalExpr {
     /// But this can have overhead in production, so it's only included in our tests.
     data_type: Arc<RwLock<Option<DataType>>>,
     nullable: Arc<RwLock<Option<bool>>>,
+    /// The work that the producer of the filter does for each row that the
+    /// filter removes, see [`Self::removed_row_work`]. Shared by all
+    /// derived filters.
+    removed_row_work: Arc<RemovedRowWork>,
 }
 
 impl std::fmt::Debug for DynamicFilterPhysicalExpr {
@@ -223,9 +228,20 @@ impl DynamicFilterPhysicalExpr {
             state_watch,
             data_type: Arc::new(RwLock::new(None)),
             nullable: Arc::new(RwLock::new(None)),
+            removed_row_work: Arc::default(),
         }
     }
 
+    /// The work that the producer of this filter does for each row that
+    /// the filter removes before it (for example the probe of a hash join),
+    /// as the producer measures it. A consumer that decides if the filter
+    /// is worth its cost uses it as the saving of a removed row (see
+    /// [`OptionalFilterGate`](crate::optional_filter_gate::OptionalFilterGate)).
+    /// The producer records its work in it.
+    pub fn removed_row_work(&self) -> &Arc<RemovedRowWork> {
+        &self.removed_row_work
+    }
+
     fn remap_children(
         children: &[Arc<dyn PhysicalExpr>],
         remapped_children: Option<&Vec<Arc<dyn PhysicalExpr>>>,
@@ -490,6 +506,7 @@ impl DynamicFilterPhysicalExpr {
             state_watch,
             data_type: Arc::new(RwLock::new(None)),
             nullable: Arc::new(RwLock::new(None)),
+            removed_row_work: Arc::default(),
         }
     }
 }
@@ -518,6 +535,7 @@ impl PhysicalExpr for DynamicFilterPhysicalExpr {
             state_watch: self.state_watch.clone(),
             data_type: Arc::clone(&self.data_type),
             nullable: Arc::clone(&self.nullable),
+            removed_row_work: Arc::clone(&self.removed_row_work),
         }))
     }
 
@@ -614,6 +632,7 @@ impl PhysicalExpr for DynamicFilterPhysicalExpr {
             state_watch: _, // Runtime channel, recreated from inner state by from_parts().
             data_type: _,   // Cached test invariant, recomputed from the expression.
             nullable: _,    // Cached test invariant, recomputed from the expression.
+            removed_row_work: _, // Runtime measurement of the producer.
         } = self;
 
         let children = children
@@ -837,6 +856,7 @@ impl DynamicFilterPhysicalExpr {
             state_watch: self.state_watch.clone(),
             data_type: Arc::clone(&self.data_type),
             nullable: Arc::clone(&self.nullable),
+            removed_row_work: Arc::clone(&self.removed_row_work),
         }
     }
 }

datafusion/physical-expr/src/filter_stats.rs

@@ -163,10 +163,63 @@ impl FilterCost {
     }
 }
 
+/// The work, in nanoseconds for each row, that an operator does on the rows
+/// that a filter removes before them, as measured by that operator.
+///
+/// For example, a hash join computes the hashes of the join keys of each
+/// probe row and looks them up in its hash table. A row that the dynamic
+/// filter of the join removes in the scan does not get this work, thus this
+/// is the saving of a removed row. The work that depends on a match (the
+/// output of a matched row) is not in it: the filter does not remove
+/// matched rows. The producer of a dynamic filter measures it and the
+/// consumers of the filter read it (see
+/// [`DynamicFilterPhysicalExpr::removed_row_work`]).
+///
+/// Lock-free.
+///
+/// [`DynamicFilterPhysicalExpr::removed_row_work`]: crate::expressions::DynamicFilterPhysicalExpr::removed_row_work
+#[derive(Debug, Default)]
+pub struct RemovedRowWork {
+    rows: AtomicU64,
+    nanos: AtomicU64,
+}
+
+impl RemovedRowWork {
+    /// Creates an empty measurement.
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Adds `rows` rows and `nanos` nanoseconds of work. The two can be
+    /// recorded separately (for example the rows when a batch arrives and
+    /// the time of each step).
+    pub fn record(&self, rows: u64, nanos: u64) {
+        self.rows.fetch_add(rows, Ordering::Relaxed);
+        self.nanos.fetch_add(nanos, Ordering::Relaxed);
+    }
+
+    /// The work for each row, or `None` before [`MIN_OBSERVED_ROWS`] rows.
+    pub fn ns_per_row(&self) -> Option<f64> {
+        let rows = self.rows.load(Ordering::Relaxed);
+        (rows >= MIN_OBSERVED_ROWS)
+            .then(|| self.nanos.load(Ordering::Relaxed) as f64 / rows as f64)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    #[test]
+    fn removed_row_work_needs_min_observed_rows() {
+        let work = RemovedRowWork::new();
+        work.record(MIN_OBSERVED_ROWS - 1, 0);
+        work.record(0, 3 * (MIN_OBSERVED_ROWS - 1));
+        assert_eq!(work.ns_per_row(), None);
+        work.record(1, 3);
+        assert_eq!(work.ns_per_row(), Some(3.0));
+    }
+
     #[test]
     fn filter_cost_derived_values() {
         let empty = FilterCost::default();

datafusion/physical-expr/src/optional_filter_gate.rs

@@ -73,12 +73,20 @@
 //!    ```text
 //!    cost_ns   = evaluation time + rows_in * measured overhead
 //!    saving_ns = (rows_in - rows_out) * saving_ns_per_row
-//!    saving_ns_per_row = min_saving_ns_per_row + measured saving
+//!    saving_ns_per_row = producer work + measured saving
 //!    ```
 //!
-//!    `min_saving_ns_per_row` comes from the configuration. It is the work
-//!    that a removed row saves after the filter, for example a hash table
-//!    probe in a join. The *measured saving* and the *measured overhead* are
+//!    The *producer work* is the work that a removed row saves after the
+//!    filter, in the operator that produced the filter, for example the
+//!    hash and the hash table lookup of a probe row in a hash join. The
+//!    producer measures it ([`RemovedRowWork`], see
+//!    [`DynamicFilterPhysicalExpr::removed_row_work`]): a hash join with a
+//!    small build side does 2 to 8 ns of work for each probe row (TPC-DS
+//!    SF1 star joins), a join with a large build side much more. Until the
+//!    producer has measured [`MIN_OBSERVED_ROWS`] rows, the gate uses
+//!    `min_saving_ns_per_row` from the configuration (a prior). With more
+//!    than one dynamic filter in the filter, the smallest measured work is
+//!    used. The *measured saving* and the *measured overhead* are
 //!    optional: a consumer that can measure more work that a removed row
 //!    saves (the Parquet scan measures the decode time of the columns that
 //!    the filter does not read), or a fixed cost for each evaluated row in
@@ -146,9 +154,11 @@ use std::time::Duration;
 use datafusion_common::config::ExecutionOptions;
 use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
 
-use crate::expressions::DynamicFilterTracking;
+use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion};
+
+use crate::expressions::{DynamicFilterPhysicalExpr, DynamicFilterTracking};
 use crate::filter_stats::{
-    Clock, FilterCost, MIN_OBSERVED_ROWS, SystemClock, duration_nanos,
+    Clock, FilterCost, MIN_OBSERVED_ROWS, RemovedRowWork, SystemClock, duration_nanos,
 };
 
 /// A running filter is paused by the cost rule only if its cost is larger
@@ -172,7 +182,8 @@ pub struct OptionalFilterGateConfig {
     /// `initial_pause_batches` are used as `initial_pause_batches`.
     pub max_pause_batches: usize,
     /// Work, in nanoseconds, that each row removed by the filter saves after
-    /// the filter, at the least. The gate adds the saving that the consumer
+    /// the filter, until the producer of the filter has measured it (see
+    /// [`RemovedRowWork`]). The gate adds the saving that the consumer
     /// measures (see [`MeasuredRowSaving`]). The gate pauses a filter whose
     /// evaluation time is larger than the saving of the rows that it
     /// removes. Negative values are used as 0.
@@ -421,9 +432,11 @@ pub struct OptionalFilterGate {
     config: OptionalFilterGateConfig,
     /// The clock that consumers use to measure the evaluation time.
     clock: Arc<dyn Clock>,
-    /// The saving that the consumer measures, added to
-    /// `config.min_saving_ns_per_row`.
+    /// The saving that the consumer measures, added to the producer work.
     measured_saving: Option<Arc<MeasuredRowSaving>>,
+    /// The work that the producers of the dynamic filters in `filter` do
+    /// for each removed row, see [`RemovedRowWork`].
+    producer_work: Vec<Arc<RemovedRowWork>>,
     state: GateState,
     /// True while the current window is a probe after a pause. The cost
     /// rule then uses [`RESUME_COST_MARGIN`].
@@ -457,12 +470,22 @@ impl OptionalFilterGate {
     pub fn new(filter: Arc<dyn PhysicalExpr>, config: OptionalFilterGateConfig) -> Self {
         let config = config.normalized();
         let tracking = DynamicFilterTracking::classify(&filter);
+        let mut producer_work = vec![];
+        filter
+            .apply(|expr| {
+                if let Some(dynamic) = expr.downcast_ref::<DynamicFilterPhysicalExpr>() {
+                    producer_work.push(Arc::clone(dynamic.removed_row_work()));
+                }
+                Ok(TreeNodeRecursion::Continue)
+            })
+            .expect("the closure is infallible");
         Self {
             filter,
             tracking,
             config,
             clock: SystemClock::shared(),
             measured_saving: None,
+            producer_work,
             state: GateState::new_window(),
             probing: false,
             backoff: config.initial_pause_batches,
@@ -513,13 +536,26 @@ impl OptionalFilterGate {
     }
 
     /// The work, in nanoseconds, that the gate assumes each removed row
-    /// saves now: the configured minimum plus the measured saving.
+    /// saves now: the work of the producer (measured, or the configured
+    /// `min_saving_ns_per_row` before the measurement) plus the measured
+    /// saving of the consumer.
     pub fn saving_ns_per_row(&self) -> f64 {
         let measured = self
             .measured_saving
             .as_ref()
             .map_or(0.0, |saving| saving.ns_per_row());
-        self.config.min_saving_ns_per_row + measured
+        self.producer_work_ns_per_row() + measured
+    }
+
+    /// The work of the producer for each removed row: the smallest measured
+    /// [`RemovedRowWork`] of the dynamic filters in the filter, or
+    /// `min_saving_ns_per_row` if none is measured yet.
+    fn producer_work_ns_per_row(&self) -> f64 {
+        self.producer_work
+            .iter()
+            .filter_map(|work| work.ns_per_row())
+            .reduce(f64::min)
+            .unwrap_or(self.config.min_saving_ns_per_row)
     }
 
     /// The work, in nanoseconds for each evaluated row, that the gate adds
@@ -1374,6 +1410,32 @@ mod tests {
         assert!(!shared_gate(filter, &shared).is_paused());
     }
 
+    /// The producer of a dynamic filter measures its work for each row that
+    /// the filter removes: once measured, it replaces the configured
+    /// `min_saving_ns_per_row`.
+    #[test]
+    fn producer_work_replaces_configured_saving() {
+        let (dynamic, filter) = dynamic_filter();
+        let mut gate = gate_with(filter);
+        assert_eq!(gate.saving_ns_per_row(), 20.0);
+        // Removes 80% at 5 ns for each row: 5 < 0.8 * 20 * 1.1, it stays on.
+        for _ in 0..4 {
+            assert_eq!(feed_timed(&mut gate, 0.2, 5.0), GateDecision::Evaluate);
+        }
+        assert!(!gate.is_paused());
+
+        // The producer measures 4 ns for each removed row: 0.8 * 4 < 5.
+        let work = dynamic.removed_row_work();
+        work.record(MIN_OBSERVED_ROWS, 4 * MIN_OBSERVED_ROWS);
+        assert_eq!(gate.saving_ns_per_row(), 4.0);
+        assert_eq!(feed_timed(&mut gate, 0.2, 5.0), GateDecision::Evaluate);
+        assert_eq!(feed_timed(&mut gate, 0.2, 5.0), GateDecision::Evaluate);
+        assert!(gate.is_paused());
+
+        // A filter without dynamic filters uses the configuration.
+        assert_eq!(new_gate().saving_ns_per_row(), 20.0);
+    }
+
     /// A clock that moves by a fixed step each time it is read.
     #[derive(Debug)]
     struct SteppingClock {