You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by jr...@apache.org on 2026/10/19 14:53:34 UTC
(beam) branch master updated: [Python] Avoid ZeroDivisionError in ConcatRangeTracker for zero weights (#40549)
This is an automated email from the ASF dual-hosted git repository.
jrmccluskey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 590abac1145 [Python] Avoid ZeroDivisionError in ConcatRangeTracker for zero weights (#40549)
590abac1145 is described below
commit 590abac114546ec33cedff26e44f37857150eb71
Author: raams95 <ja...@gmail.com>
AuthorDate: Mon Oct 19 09:53:25 2026 -0500
[Python] Avoid ZeroDivisionError in ConcatRangeTracker for zero weights (#40549)
* [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
* Add CHANGES.md entry
* Test zero-weight ConcatSource through a pipeline read
---------
Co-authored-by: JanakiRamChittineni <34...@users.noreply.github.com>
Co-authored-by: Jack McCluskey <34...@users.noreply.github.com>
---
CHANGES.md | 1 +
sdks/python/apache_beam/io/concat_source.py | 4 ++--
sdks/python/apache_beam/io/concat_source_test.py | 9 +++++++++
3 files changed, 12 insertions(+), 2 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index c10b7262209..1ec1dbe8086 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -115,6 +115,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)).
* (Python) DataFrame API: `Series.value_counts(bins=...)` no longer always fails with a `TypeError` ([#40503](https://github.com/apache/beam/issues/40503)).
* Fixed X (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
diff --git a/sdks/python/apache_beam/io/concat_source.py b/sdks/python/apache_beam/io/concat_source.py
index 35ae6fe2731..bd6b7396d5d 100644
--- a/sdks/python/apache_beam/io/concat_source.py
+++ b/sdks/python/apache_beam/io/concat_source.py
@@ -135,8 +135,8 @@ class ConcatRangeTracker(iobase.RangeTracker):
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 efa24b3975f..3cd169df506 100644
--- a/sdks/python/apache_beam/io/concat_source_test.py
+++ b/sdks/python/apache_beam/io/concat_source_test.py
@@ -171,6 +171,15 @@ class ConcatSourceTest(unittest.TestCase):
self.assertEqual(range_tracker.position_at_fraction(.601), (2, 17))
self.assertEqual(range_tracker.position_at_fraction(1), (3, None))
+ 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),
+ ])
+ 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