Only emit stream values chunks when the output channels have changed (#4774)

2026-09-12 20:57:52 +02:00 · 2025-05-21 13:09:42 -04:00
parent 364508ff20
commit 96e2f68b2d
1 changed files with 8 additions and 3 deletions
@@ -509,9 +509,14 @@ class PregelLoop(LoopProtocol):
                    **read_channels(self.channels, self.stream_keys)
                )
            # produce values output
-            self._emit(
-                "values", map_output_values, self.output_keys, writes, self.channels
-            )
+            if not updated_channels.isdisjoint(
+                (self.output_keys,)
+                if isinstance(self.output_keys, str)
+                else self.output_keys
+            ):
+                self._emit(
+                    "values", map_output_values, self.output_keys, writes, self.channels
+                )
            # clear pending writes
            self.checkpoint_pending_writes.clear()
            # "not skip_done_tasks" only applies to first tick after resuming