Gaurav598 opened a new pull request, #40496: URL: https://github.com/apache/beam/pull/40496
## Summary - Iterate Dask side-input bags one partition at a time instead of materializing the entire bag locally. - Add regression coverage for lazy partition computation, empty partitions, repeated iteration, and errors in later partitions. ## Problem and root cause `DaskBagWindowedIterator.__iter__` called `list(self.bag)`. Dask computed and collected every partition before yielding the first side-input value, so large iterable side inputs could exhaust local memory. ## Solution Use `Bag.to_delayed()` and compute each partition when iteration reaches it. Values still pass through the existing window conversion. This bounds the iterator's local materialization to one partition at a time. ## Testing - `python -m unittest apache_beam.runners.dask.dask_runner_test -v` — exited successfully. - `python -m pytest -q sdks/python/apache_beam/runners/dask/dask_runner_test.py::DaskBagWindowedIteratorTest` — 3 passed (one configuration warning because pytest-timeout is not installed in the isolated test environment). - Ran the laziness regression test against the original `__iter__` implementation — it failed as expected because both partitions were computed before the first value was yielded. - `yapf --diff` on the two changed files and `git diff --check` — no differences or errors. ## Compatibility and risk The side-input values and window conversion are unchanged. Each partition now incurs a separate Dask computation, which can add scheduler round trips or recompute shared graph dependencies. Views such as `AsList` must still materialize their final list by design; this change removes the iterator's unnecessary whole-bag materialization. Fixes #40282 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
