Fix CombineGlobally with GlobalWindows

sdks/python/apache_beam/transforms/combine_globally_test.py

@@ -0,0 +1,55 @@
+#
+# 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):
+    # this error is expected since the below combination is ill-defined.
+    with self.assertRaises(ValueError):
+      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

@@ -2587,6 +2587,18 @@ def add_input_types(transform):
             "or CombineGlobally().as_singleton_view() to get the default "
             "output of the CombineFn if the input PCollection is empty.")
 
+      # return the error for this ill-defined streaming case
+      if not pcoll.is_bounded and not pcoll.windowing.is_default():
+        raise ValueError(
+            "For unbounded data sources, "
+            "default values are not yet supported in CombineGlobally() if the "
+            "output PCollection is not windowed by GlobalWindows"
+            " with DefaultTrigger. "
+            "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 "
+            "output of the CombineFn if the input PCollection is empty.")
+
       def typed(transform):
         # TODO(robertwb): We should infer this.
         if combined.element_type:
@@ -2598,6 +2610,8 @@ def typed(transform):
 
       def inject_default(_, combined):
         if combined:
+          if len(combined) > 1:
+            _LOGGER.error('Cannot inject default values with: %s', combined)
           assert len(combined) == 1
           return combined[0]
         else: