diff --git a/CHANGES.md b/CHANGES.md index 6357802c0eec..ece6a7e7d2be 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -94,6 +94,7 @@ * (Java) KafkaIO dynamic reads no longer require the obsolete `beam_fn_api` experiment ([#29998](https://github.com/apache/beam/issues/29998)). * (Prism) Self-checkpointing splittable DoFns now resume after their requested delay instead of immediately, so polling SDFs no longer busy-spin ([#39848](https://github.com/apache/beam/issues/39848)). * (Java) MongoDbIO read splitting now preserves non-ObjectId `_id` types (e.g. string ids) instead of failing to parse the generated range filters ([#39900](https://github.com/apache/beam/issues/39900)). +* (Python) Fixed `PickleCoder`/`_MemoizingPickleCoder.as_deterministic_coder()` raising `TypeError` instead of returning a working deterministic coder ([#28558](https://github.com/apache/beam/issues/28558)). ## Security Fixes diff --git a/sdks/python/apache_beam/coders/coders.py b/sdks/python/apache_beam/coders/coders.py index 5f22bff5351b..cfdee3b1861c 100644 --- a/sdks/python/apache_beam/coders/coders.py +++ b/sdks/python/apache_beam/coders/coders.py @@ -899,7 +899,8 @@ def _nonhashable_dumps(x): return coder_impl.CallbackCoderImpl(_nonhashable_dumps, pickler.loads) def as_deterministic_coder(self, step_label, error_message=None): - return FastPrimitivesCoder(self, requires_deterministic=step_label) + return _update_compatible_deterministic_fast_primitives_coder( + self, step_label) def to_type_hint(self): return Any @@ -914,7 +915,8 @@ def _create_impl(self): lambda x: dumps(x, protocol), pickle.loads) def as_deterministic_coder(self, step_label, error_message=None): - return FastPrimitivesCoder(self, requires_deterministic=step_label) + return _update_compatible_deterministic_fast_primitives_coder( + self, step_label) def to_type_hint(self): return Any diff --git a/sdks/python/apache_beam/coders/coders_test.py b/sdks/python/apache_beam/coders/coders_test.py index ccd947457ad7..17fc46bd9f5e 100644 --- a/sdks/python/apache_beam/coders/coders_test.py +++ b/sdks/python/apache_beam/coders/coders_test.py @@ -50,6 +50,21 @@ def test_equality(self): self.assertNotEqual(coders.Base64PickleCoder(), coders.PickleCoder()) self.assertNotEqual(coders.Base64PickleCoder(), object()) + def test_as_deterministic_coder(self): + # PickleCoder.as_deterministic_coder used to construct FastPrimitivesCoder + # with a requires_deterministic kwarg that constructor never accepted, + # raising TypeError on every call (see coders.FastPrimitivesCoder). + v = ('a' * 10, 'b' * 90) + deterministic = coders.PickleCoder().as_deterministic_coder('label') + self.assertTrue(deterministic.is_deterministic()) + self.assertEqual(v, deterministic.decode(deterministic.encode(v))) + + memoizing_deterministic = coders._MemoizingPickleCoder( + ).as_deterministic_coder('label') + self.assertTrue(memoizing_deterministic.is_deterministic()) + self.assertEqual( + v, memoizing_deterministic.decode(memoizing_deterministic.encode(v))) + class CodersTest(unittest.TestCase): def test_str_utf8_coder(self):