Deepcopy combine_fn in PrecombineFn and PostCombineFn.

[claudevdm]

Deepcopy combine_fn in PrecombineFn and PostCombineFn to avoid sharing state between lifted stages when using cloudpickle.

Previously PrecombineFn and PostCombineFn assign setup, create_accumulator attributes as references the setup, create_accumulator methods on the original combine_fn [1] passed to _CombinePerKeyWithHotKeyFanout.

When using cloudpickler with combiner lifting Precombine is split into ExtractOutputs/Merge stages in the case of FnApiRunner and as a result if these stages run in the same process the state is shared. Similarly when the BundleBasedDirectRunner is used the Precombine is split into PartialGroupByKeyCombiningValues and FinishCombine, both invoking the setup [2][3] and teardown method.

This change ensures a new instance of the combine_fn is used in the lifted stages when using cloudpickler.

[1]

beam/sdks/python/apache_beam/transforms/core.py

Line 3166 in e705e28

setup = combine_fn.setup

[2]

beam/sdks/python/apache_beam/runners/direct/helper_transforms.py

Line 62 in e705e28

self._combine_fn.setup()

[3]

beam/sdks/python/apache_beam/runners/direct/helper_transforms.py

Line 101 in e705e28

self._combine_fn.setup()

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

[tvalentyn]

This looks good to me. I would also test that the failure is no longer reproducible on Dataflow (it might require running a tests similar to existing unit test on Dataflow first, to repro the bug without changes.)

[claudevdm]

fixes #26209

[robertwb]

Thank you!