From a0de74ee4daaa9a67502ef0e53436c6fdbd4bbfd Mon Sep 17 00:00:00 2001 From: JanakiRamChittineni <341781521+raams95@users.noreply.github.com> Date: Tue, 13 Oct 2026 01:33:40 -0500 Subject: [PATCH 1/3] [Python] Avoid ZeroDivisionError in ConcatRangeTracker for zero weights When every SourceBundle passed to a ConcatSource had weight 0, the average used to fill in missing weights was also 0, so computing the cumulative weights divided by zero. Treat zero weights like unknown weights when computing that average. Fixes #21316 --- sdks/python/apache_beam/io/concat_source.py | 4 ++-- sdks/python/apache_beam/io/concat_source_test.py | 11 +++++++++++ 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/sdks/python/apache_beam/io/concat_source.py b/sdks/python/apache_beam/io/concat_source.py index 35ae6fe27317..bd6b7396d5d3 100644 --- a/sdks/python/apache_beam/io/concat_source.py +++ b/sdks/python/apache_beam/io/concat_source.py @@ -135,8 +135,8 @@ def _compute_cumulative_weights(source_bundles): min_diff = 1e-5 # For the computation below, we need weights for all sources. # Substitute average weights for those whose weights are - # unspecified (or 1.0 for everything if none are known). - known = [s.weight for s in source_bundles if s.weight is not None] + # unspecified or zero (or 1.0 for everything if none are known). + known = [s.weight for s in source_bundles if s.weight] avg = sum(known) / len(known) if known else 1.0 weights = [s.weight or avg for s in source_bundles] diff --git a/sdks/python/apache_beam/io/concat_source_test.py b/sdks/python/apache_beam/io/concat_source_test.py index efa24b3975fc..fbda407f2632 100644 --- a/sdks/python/apache_beam/io/concat_source_test.py +++ b/sdks/python/apache_beam/io/concat_source_test.py @@ -171,6 +171,17 @@ def test_position_at_fration(self): self.assertEqual(range_tracker.position_at_fraction(.601), (2, 17)) self.assertEqual(range_tracker.position_at_fraction(1), (3, None)) + def test_zero_weight_bundles(self): + source = ConcatSource([ + iobase.SourceBundle(0, RangeSource(0, 4), None, None), + iobase.SourceBundle(0, RangeSource(4, 8), None, None), + ]) + + range_tracker = source.get_range_tracker() + self.assertEqual(range_tracker.position_at_fraction(.5), (1, 4)) + self.assertEqual( + list(source.read(source.get_range_tracker())), list(range(8))) + def test_empty_source(self): read_all = source_test_utils.read_from_source From 0004d90a123090cd06c38dc9cfd9b3ff86b150ba Mon Sep 17 00:00:00 2001 From: JanakiRamChittineni <341781521+raams95@users.noreply.github.com> Date: Tue, 13 Oct 2026 08:37:27 -0500 Subject: [PATCH 2/3] Add CHANGES.md entry --- CHANGES.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGES.md b/CHANGES.md index fda4c9edc218..724c100953b6 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -104,6 +104,7 @@ * (Python) `Duration` built from float seconds now rounds to the nearest microsecond instead of truncating, which could lose a microsecond ([#40263](https://github.com/apache/beam/issues/40263)). * (Python) `ReadFromText` with `escapechar` no longer skips a delimiter that directly follows an escaped delimiter, which merged two records into one ([#40459](https://github.com/apache/beam/issues/40459)). * (Python) DataFrame API: `agg` with a list that includes `'count'` or `'size'` now labels those rows `count`/`size` instead of `sum` ([#40504](https://github.com/apache/beam/issues/40504)). +* (Python) `ConcatSource` no longer raises `ZeroDivisionError` when every source bundle reports a weight of 0, e.g. bundles smaller than 1 MB ([#21316](https://github.com/apache/beam/issues/21316)). * Fixed X (Java/Python) ([#X](https://github.com/apache/beam/issues/X)). ## Security Fixes From 1db5308825bde7b9620b34a3fa9a74c442e03019 Mon Sep 17 00:00:00 2001 From: JanakiRamChittineni <341781521+raams95@users.noreply.github.com> Date: Tue, 13 Oct 2026 14:39:59 -0500 Subject: [PATCH 3/3] Test zero-weight ConcatSource through a pipeline read --- sdks/python/apache_beam/io/concat_source_test.py | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/sdks/python/apache_beam/io/concat_source_test.py b/sdks/python/apache_beam/io/concat_source_test.py index fbda407f2632..3cd169df5069 100644 --- a/sdks/python/apache_beam/io/concat_source_test.py +++ b/sdks/python/apache_beam/io/concat_source_test.py @@ -171,16 +171,14 @@ def test_position_at_fration(self): self.assertEqual(range_tracker.position_at_fraction(.601), (2, 17)) self.assertEqual(range_tracker.position_at_fraction(1), (3, None)) - def test_zero_weight_bundles(self): + def test_read_zero_weight_bundles_in_pipeline(self): source = ConcatSource([ iobase.SourceBundle(0, RangeSource(0, 4), None, None), iobase.SourceBundle(0, RangeSource(4, 8), None, None), ]) - - range_tracker = source.get_range_tracker() - self.assertEqual(range_tracker.position_at_fraction(.5), (1, 4)) - self.assertEqual( - list(source.read(source.get_range_tracker())), list(range(8))) + with TestPipeline() as pipeline: + pcoll = pipeline | beam.io.Read(source) + assert_that(pcoll, equal_to(list(range(8)))) def test_empty_source(self): read_all = source_test_utils.read_from_source