Fix CombineGlobally with GlobalWindows

sdks/python/apache_beam/transforms/combine_globally_test.py

@@ -0,0 +1,53 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Unit tests for our libraries of combine PTransforms."""
+# pytype: skip-file
+
+import time
+import unittest
+
+import apache_beam as beam
+from apache_beam.testing.test_pipeline import TestPipeline
+from apache_beam.transforms import trigger
+from apache_beam.transforms import window
+from apache_beam.transforms.periodicsequence import PeriodicImpulse
+
+
+class CombineGloballyTest(unittest.TestCase):
+  def test_with_periodic_impulse(self):
+    with TestPipeline() as p:
+      _ = (
+          p
+          | PeriodicImpulse(
+              start_timestamp=time.time(),
+              stop_timestamp=time.time() + 4,
+              fire_interval=1,
+              apply_windowing=False,
+          )
+          | beam.Map(lambda x: ('c', 1))
+          | beam.WindowInto(
+              window.GlobalWindows(),
+              trigger=trigger.Repeatedly(trigger.AfterCount(2)),
+              accumulation_mode=trigger.AccumulationMode.DISCARDING,
+          )
+          | beam.combiners.Count.Globally()
+          | "Print Windows" >> beam.Map(print))
+
+
+if __name__ == '__main__':
+  unittest.main()

sdks/python/apache_beam/transforms/core.py

@@ -2581,7 +2581,7 @@ def add_input_types(transform):
       if pcoll.windowing.windowfn != GlobalWindows():
         raise ValueError(
             "Default values are not yet supported in CombineGlobally() if the "
-            "output  PCollection is not windowed by GlobalWindows. "
+            "output PCollection is not windowed by GlobalWindows. "
             "Instead, use CombineGlobally().without_defaults() to output "
             "an empty PCollection if the input PCollection is empty, "
             "or CombineGlobally().as_singleton_view() to get the default "
@@ -2598,8 +2598,9 @@ def typed(transform):
 
       def inject_default(_, combined):
         if combined:
-          assert len(combined) == 1
-          return combined[0]
+          if len(combined) > 1:
+            _LOGGER.warning('Apply Combined Fn with this list: %s', combined)
+            return combined[-1]
         else:
           try:
             combine_fn.setup(*args, **kwargs)