github-actions[bot] commented on code in PR #41160:
URL: https://github.com/apache/beam/pull/41160#discussion_r4525800050
##########
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:
**🟡 Medium**: When `self.has_windowed_inputs` is `False`, `window` is never
referenced in the `self.placeholders_for_process` loop below (since
`WindowParam`, `StateParam`, and `TimerParam` all set `self.has_windowed_inputs
= True` in `__init__`) and is only needed on a cache miss when indexing
`self.side_inputs`. Hoisting `window = GlobalWindow()` above `if
self.has_cached_window_args and not additional_args:` causes `GlobalWindow()`
(`__new__` and `__init__`) to be called on every element for non-windowed
`DoFn`s using `PerWindowInvoker` (e.g., `DoFn`s with `TimestampParam`,
`KeyParam`, default args, or globally windowed side inputs). Moving `if not
self.has_windowed_inputs: window = GlobalWindow()` inside the `else:` branch
below (and similarly in `_invoke_process_batch_per_window` at line 1103) avoids
this per-element overhead on the cached fast path.
##########
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:
**🟢 Low**: For consistency with `_invoke_process_per_window` (line 995 and
lines 1025-1035), `_invoke_process_batch_per_window` can also use `window =
windowed_batch.windows[0]` and short-circuit `placeholders_for_process_batch`
checks with `p is ... or`.
```suggestion
window = windowed_batch.windows[0]
```
--
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]
