Fix PickleCoder.as_deterministic_coder() raising TypeError

[AmirF194]

PickleCoder.as_deterministic_coder() and _MemoizingPickleCoder.as_deterministic_coder() both construct FastPrimitivesCoder with a requires_deterministic keyword argument:

def as_deterministic_coder(self, step_label, error_message=None):
  return FastPrimitivesCoder(self, requires_deterministic=step_label)

FastPrimitivesCoder.__init__(self, fallback_coder=PickleCoder()) has never accepted that argument, so both raise TypeError unconditionally. The bug goes back to commit 9be70c9, which changed PickleCoder.as_deterministic_coder from the working DeterministicFastPrimitivesCoder(self, step_label) to this broken call, in the same diff that added a requires_deterministic kwarg to a different class (coder_impl.FastPrimitivesCoderImpl, the Cython impl, not the coders.FastPrimitivesCoder wrapper referenced here). _MemoizingPickleCoder later copied the already-broken line.

The invariant as_deterministic_coder() is supposed to hold, that it always returns a working deterministic coder rather than raising, is exactly what FastPrimitivesCoder's own as_deterministic_coder already does by delegating to _update_compatible_deterministic_fast_primitives_coder. This PR routes both call sites through that same helper instead.

A reachable consumer: GroupByEncryptedKey.expand() calls coder.as_deterministic_coder(...) inside except ValueError, expecting a non-deterministic key to log a warning. When the coder resolves to PickleCoder/_MemoizingPickleCoder, the uncaught TypeError crashes pipeline construction instead.

Verified:

• PickleCoder().as_deterministic_coder('x') and _MemoizingPickleCoder().as_deterministic_coder('x') raise TypeError on master, confirmed live against the installed apache-beam package (byte-identical coders.py) and against this branch.
• New PickleCoderTest.test_as_deterministic_coder fails on master, passes on this branch, both runs in the same container.
• coders_test.py, typecoders_test.py, and fast_coders_test.py (the standard-coder correctness suite, all classes including FastPrimitivesCoder/DeterministicFastPrimitivesCoder) pass unchanged.
• ruff check and yapf --diff are clean on both changed files.
• Not checked: the Cython (coder_impl) fast paths, since this fix is confined to the plain-Python coders.py wrapper layer and does not touch coder_impl.

Fixes #39942

[github-actions]

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

[AmirF194]

Run Python_Integration PreCommit 3.14

[github-actions]

Assigning reviewers:

R: @claudevdm for label python.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[AmirF194]

remind me after tests pass

[github-actions]

Ok - I'll remind @AmirF194 after tests pass

[github-actions]

Reminder, please take a look at this pr: @claudevdm

[claudevdm]

Returning DeterministicFastPrimitivesCoder will fail at runtime for arbitrary classes.

beam/sdks/python/apache_beam/coders/coder_impl.py

Line 486 in 42663f3

def encode_special_deterministic(self, value, stream):

Maybe add a warning like

PickleCoder was registered for a key type in '%s', but the runner requires a
deterministic key encoding and pickle is not deterministic. Keys in this step
will be encoded with the deterministic fallback coder instead of pickle. That
coder supports primitives, containers, protobuf messages, frozen dataclasses,
NamedTuples, enums, and classes defining both __getstate__ and __setstate__.
Any other key type will fail at encode time. If your key type is not one of
these, register a deterministic custom Coder for it or add a type hint so the
default coder is used.

[AmirF194]

Good point, pushed fa9623c. Added a warning at both call sites (PickleCoder and _MemoizingPickleCoder) using close to your wording, logged before the fallback coder is constructed. Ran the full coders_test.py suite plus a manual check that the warning fires with the right step label; 16 passed.