From c9b5a67d5402cdbe8314da9a0bf1dbcfb23ac289 Mon Sep 17 00:00:00 2001 From: Yi Hu Date: Fri, 30 Oct 2026 15:15:08 -0400 Subject: [PATCH 1/2] Skip WindowInto(GlobalWIndows()) elementwise if upstream pcoll is already global window in Write transform `WriteImpl._expand_bounded` previously applied `WindowInto(GlobalWindows())` unconditionally to all input records before writing bundles, even when the input `PCollection` was already using the default global windowing strategy. In typical batch pipelines, this added an unnecessary per-element `WindowInto` pass across the entire dataset (~22% overhead in bounded write benchmarks). Check `windowing.is_default()` at pipeline expansion time and only apply `WindowInto(GlobalWindows())` when the input `PCollection` is not already in the default global window. --- sdks/python/apache_beam/io/iobase.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/sdks/python/apache_beam/io/iobase.py b/sdks/python/apache_beam/io/iobase.py index 34ee781d87a1..e747c56eff5f 100644 --- a/sdks/python/apache_beam/io/iobase.py +++ b/sdks/python/apache_beam/io/iobase.py @@ -1189,9 +1189,10 @@ def _expand_bounded(self, pcoll, min_shards): keyed_pcoll = pcoll | core.Map(lambda x: (None, x)) else: keyed_pcoll = pcoll | core.ParDo(_RoundRobinKeyFn(), count=min_shards) + if not keyed_pcoll.windowing.is_default(): + keyed_pcoll = keyed_pcoll | core.WindowInto(window.GlobalWindows()) write_result_coll = ( keyed_pcoll - | core.WindowInto(window.GlobalWindows()) | core.GroupByKey() | 'WriteBundles' >> core.ParDo( _WriteKeyedBundleDoFn(self.sink), AsSingleton(init_result_coll))) @@ -1200,9 +1201,10 @@ def _expand_bounded(self, pcoll, min_shards): "*** WriteImpl min_shards undef so it's 1, and we write per Bundle") min_shards = 1 + if not pcoll.windowing.is_default(): + pcoll = pcoll | core.WindowInto(window.GlobalWindows()) write_result_coll = ( pcoll - | core.WindowInto(window.GlobalWindows()) | 'WriteBundles' >> core.ParDo( _WriteBundleDoFn(self.sink), AsSingleton(init_result_coll)) | 'Pair' >> core.Map(lambda x: (None, x)) From 6d26e8c1a85a2026f6368dc69198ff61e92a0813 Mon Sep 17 00:00:00 2001 From: Yi Hu Date: Fri, 30 Oct 2026 16:28:16 -0400 Subject: [PATCH 2/2] Fix pcoll.windowing AssertError when pcoll.producer absent --- sdks/python/apache_beam/io/iobase.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/sdks/python/apache_beam/io/iobase.py b/sdks/python/apache_beam/io/iobase.py index e747c56eff5f..fe997b0caa26 100644 --- a/sdks/python/apache_beam/io/iobase.py +++ b/sdks/python/apache_beam/io/iobase.py @@ -1189,7 +1189,9 @@ def _expand_bounded(self, pcoll, min_shards): keyed_pcoll = pcoll | core.Map(lambda x: (None, x)) else: keyed_pcoll = pcoll | core.ParDo(_RoundRobinKeyFn(), count=min_shards) - if not keyed_pcoll.windowing.is_default(): + # pcoll.windowing checks pcoll.producer which may be absent when expand() + # is called directly (e.g. in tests) + if pcoll.producer is None or not keyed_pcoll.windowing.is_default(): keyed_pcoll = keyed_pcoll | core.WindowInto(window.GlobalWindows()) write_result_coll = ( keyed_pcoll @@ -1201,7 +1203,7 @@ def _expand_bounded(self, pcoll, min_shards): "*** WriteImpl min_shards undef so it's 1, and we write per Bundle") min_shards = 1 - if not pcoll.windowing.is_default(): + if pcoll.producer is None or not pcoll.windowing.is_default(): pcoll = pcoll | core.WindowInto(window.GlobalWindows()) write_result_coll = ( pcoll