Add field to DynamicPhysicalExpr to indicate its complete

Author: LiaCastaneda

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

@@ -57,6 +57,8 @@ struct Inner {
     /// This is used for [`PhysicalExpr::snapshot_generation`] to have a cheap check for changes.
     generation: u64,
     expr: Arc<dyn PhysicalExpr>,
+    /// Flag indicating whether all updates have been received and the filter is complete.
+    is_complete: bool,
 }
 
 impl Inner {
@@ -66,6 +68,7 @@ impl Inner {
             // This is not currently used anywhere but it seems useful to have this simple distinction.
             generation: 1,
             expr,
+            is_complete: false,
         }
     }
 
@@ -207,10 +210,26 @@ impl DynamicFilterPhysicalExpr {
         *current = Inner {
             generation: current.generation + 1,
             expr: new_expr,
+            is_complete: current.is_complete,
         };
         Ok(())
     }
 
+    /// Mark this dynamic filter as complete.
+    ///
+    /// This signals that all expected updates have been received.
+    pub fn mark_complete(&self) {
+        let mut current = self.inner.write();
+        current.is_complete = true;
+    }
+
+    /// Check if this dynamic filter is complete.
+    ///
+    /// Returns `true` if all expected updates have been received via [`Self::mark_complete`].
+    pub fn is_complete(&self) -> bool {
+        self.inner.read().is_complete
+    }
+
     fn render(
         &self,
         f: &mut std::fmt::Formatter<'_>,

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

@@ -300,6 +300,7 @@ impl SharedBoundsAccumulator {
                     self.create_filter_from_partition_bounds(&inner.bounds)?;
                 self.dynamic_filter.update(filter_expr)?;
             }
+            self.dynamic_filter.mark_complete();
         }
 
         Ok(())

datafusion/physical-plan/src/topk/mod.rs

@@ -591,10 +591,13 @@ impl TopK {
             common_sort_prefix_converter: _,
             common_sort_prefix: _,
             finished: _,
-            filter: _,
+            filter,
         } = self;
         let _timer = metrics.baseline.elapsed_compute().timer(); // time updated on drop
 
+        // Mark the dynamic filter as complete now that TopK processing is finished.
+        filter.read().expr().mark_complete();
+
         // break into record batches as needed
         let mut batches = vec![];
         if let Some(mut batch) = heap.emit()? {