Fix PickleCoder.as_deterministic_coder() raising TypeError

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
 

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

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):