Compute Dynamic Filters only when a consumer supports them

Author: LiaCastaneda

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

@@ -294,6 +294,16 @@ impl DynamicFilterPhysicalExpr {
             .await;
     }
 
+    /// Check if this dynamic filter is being actively used by any consumers.
+    ///
+    /// Returns `true` if there are references beyond the producer (e.g., the HashJoinExec
+    /// that created the filter). This is useful to avoid computing expensive filter
+    /// expressions when no consumer will actually use them.
+    pub fn is_used(self: &Arc<Self>) -> bool {
+        // Strong count > 1 means at least one consumer is holding a reference beyond the producer.
+        Arc::strong_count(self) > 1
+    }
+
     fn render(
         &self,
         f: &mut std::fmt::Formatter<'_>,
@@ -607,4 +617,47 @@ mod test {
         // wait_complete should return immediately
         dynamic_filter.wait_complete().await;
     }
+
+    #[test]
+    fn test_is_used() {
+        // Create a dynamic filter
+        let filter = Arc::new(DynamicFilterPhysicalExpr::new(
+            vec![],
+            lit(true) as Arc<dyn PhysicalExpr>,
+        ));
+
+        // Initially, only one reference exists (the filter itself)
+        assert!(
+            !filter.is_used(),
+            "Filter should not be used with only one reference"
+        );
+
+        // Simulate a consumer holding a reference (e.g., ParquetExec)
+        let consumer1 = Arc::clone(&filter);
+        assert!(
+            filter.is_used(),
+            "Filter should be used with a consumer reference"
+        );
+
+        // Multiple consumers
+        let consumer2 = Arc::clone(&filter);
+        assert!(
+            filter.is_used(),
+            "Filter should still be used with multiple consumers"
+        );
+
+        // Drop one consumer
+        drop(consumer1);
+        assert!(
+            filter.is_used(),
+            "Filter should still be used with remaining consumer"
+        );
+
+        // Drop all consumers
+        drop(consumer2);
+        assert!(
+            !filter.is_used(),
+            "Filter should not be used after all consumers dropped"
+        );
+    }
 }

datafusion/physical-plan/src/joins/hash_join/exec.rs

@@ -910,7 +910,21 @@ impl ExecutionPlan for HashJoinExec {
              consider using CoalescePartitionsExec or the EnforceDistribution rule"
         );
 
-        let enable_dynamic_filter_pushdown = self.dynamic_filter.is_some();
+        // Only enable dynamic filter pushdown if:
+        // - The session config enables dynamic filter pushdown
+        // - A dynamic filter exists
+        // - At least one consumer is holding a reference to it, this avoids expensive filter
+        //   computation when disabled or when no consumer will use it.
+        let enable_dynamic_filter_pushdown = context
+            .session_config()
+            .options()
+            .optimizer
+            .enable_join_dynamic_filter_pushdown
+            && self
+                .dynamic_filter
+                .as_ref()
+                .map(|df| df.filter.is_used())
+                .unwrap_or(false);
 
         let join_metrics = BuildProbeJoinMetrics::new(partition, &self.metrics);
         let left_fut = match self.mode {