damccorm commented on code in PR #41160:
URL: https://github.com/apache/beam/pull/41160#discussion_r4526453990
##########
sdks/python/apache_beam/runners/common.py:
##########
@@ -990,20 +991,20 @@ def _invoke_process_per_window(
additional_kwargs,
):
# type: (...) -> Optional[SplitResultResidual]
- if self.has_cached_window_args:
+ if self.has_windowed_inputs:
+ window = windowed_value.windows[0]
+ else:
+ window = GlobalWindow()
Review Comment:
Fixed in b7c49de3bc0: moved `if not self.has_windowed_inputs: window =
GlobalWindow()` inside the `else:` branch in both `_invoke_process_per_window`
and `_invoke_process_batch_per_window`.
##########
sdks/python/apache_beam/runners/common.py:
##########
@@ -1094,26 +1095,27 @@ def _invoke_process_batch_per_window(
):
# type: (...) -> Optional[SplitResultResidual]
- if self.has_cached_window_batch_args:
+ if self.has_windowed_inputs:
+ assert isinstance(windowed_batch, HomogeneousWindowedBatch)
+ assert len(windowed_batch.windows) <= 1
+ window, = windowed_batch.windows
Review Comment:
Fixed in b7c49de3bc0: updated `_invoke_process_batch_per_window` to use
`windowed_batch.windows[0]` and identity short-circuiting (`p is ... or`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
