refactor: hoist observer guard to caller of publish_finalized_states

The method now always mutates when called and takes the observer as a
required argument; the "is observer installed?" check moves to the caller
in `compute_aggregates`. Removes the "&mut self that only mutates when
observer is set" shape.

Addresses apache#24035 review comment 3738822701.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: avantgardnerio

datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs

@@ -1114,12 +1114,15 @@ pub struct BoundedWindowAggStream {
 }
 
 impl BoundedWindowAggStream {
-    /// Fire the [`WindowStateObserver`] for every partition key whose
-    /// `WindowAggState::is_end` is true.
-    fn publish_finalized_states(&mut self) -> Result<()> {
-        let Some(observer) = &self.state_observer else {
-            return Ok(());
-        };
+    /// Fire `observer` for every partition key whose
+    /// [`WindowAggState::is_end`] is true. Always mutates when called:
+    /// [`Accumulator::state`] requires `&mut`, which propagates up here.
+    /// The caller is responsible for deciding whether to fire (i.e. checking
+    /// whether an observer is installed).
+    fn publish_finalized_states(
+        &mut self,
+        observer: &dyn WindowStateObserver,
+    ) -> Result<()> {
         let Some((first, rest)) = self.window_agg_states.split_first_mut() else {
             return Ok(());
         };
@@ -1222,7 +1225,9 @@ impl BoundedWindowAggStream {
         // already streamed out, so at EOS that call returns `None` and the
         // prune path is skipped — the final partition would otherwise be
         // dropped unobserved.
-        self.publish_finalized_states()?;
+        if let Some(observer) = self.state_observer.clone() {
+            self.publish_finalized_states(observer.as_ref())?;
+        }
 
         let schema = Arc::clone(&self.schema);
         let window_expr_out = self.search_mode.calculate_out_columns(