github-actions[bot] commented on code in PR #41054:
URL: https://github.com/apache/beam/pull/41054#discussion_r4496951638
##########
sdks/python/apache_beam/transforms/combiners.py:
##########
@@ -787,16 +789,71 @@ def add_input(self, accumulator, element, *args,
**kwargs):
]
+class ToSingleElementCollection(CombinerWithoutDefaults):
+ """Base class for combining a PCollection into a single element.
+
+ This is for combining a PCollection into a single element in the cases where
+ there is no value (and significant overhead) in using a combiner as the
+ output is no smaller than its input.
+ """
+ def expand(self, pcoll):
+ raise NotImplementedError
+
+ def _expand(self, pcoll, post_process):
+ if pcoll.windowing.windowfn == window.GlobalWindows():
Review Comment:
**🟠High**: Checking `pcoll.windowing.windowfn == window.GlobalWindows()`
also matches `PCollection`s in `GlobalWindows` that have non-default triggers
or accumulation modes (for example, `WindowInto(GlobalWindows(),
trigger=Repeatedly(AfterCount(N)),
accumulation_mode=AccumulationMode.DISCARDING)` used with
`.without_defaults()`). Taking the `DoOnce >> Create([None])` side-input branch
in that case replaces the output windowing strategy with default
`GlobalWindows` and only fires once rather than emitting one collection per
trigger pane via `GroupByKey` (which previously worked via
`CombineGlobally(...).without_defaults()`).
Using `pcoll.windowing.is_default()` ensures that `GlobalWindows` with
custom triggers fall through to the `GroupByKey` path when `without_defaults()`
is used.
```suggestion
if pcoll.windowing.is_default():
```
--
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]
