Encode paneinfo with PaneInfoCoder.

Author: Claude

sdks/python/apache_beam/coders/coders.py

@@ -113,7 +113,8 @@
     'WindowedValueCoder',
     'ParamWindowedValueCoder',
     'BigIntegerCoder',
-    'DecimalCoder'
+    'DecimalCoder',
+    'PaneInfoCoder'
 ]
 
 T = TypeVar('T')
@@ -1753,6 +1754,18 @@ def __hash__(self):
     return hash(type(self))
 
 
+class PaneInfoCoder(FastCoder):
+  def _create_impl(self):
+    return coder_impl.PaneInfoCoderImpl()
+
+  def is_deterministic(self):
+    # type: () -> bool
+    return True
+
+  def to_type_hint(self):
+    return windowed_value.PaneInfo
+
+
 class DecimalCoder(FastCoder):
   def _create_impl(self):
     return coder_impl.DecimalCoderImpl()

sdks/python/apache_beam/coders/typecoders.py

@@ -70,9 +70,9 @@ def MakeXyzs(v):
 from typing import Iterable
 from typing import List
 from typing import Type
-
 from apache_beam.coders import coders
 from apache_beam.typehints import typehints
+from apache_beam.utils import windowed_value
 
 __all__ = ['registry']
 
@@ -92,6 +92,7 @@ def register_standard_coders(self, fallback_coder):
     self._register_coder_internal(bytes, coders.BytesCoder)
     self._register_coder_internal(bool, coders.BooleanCoder)
     self._register_coder_internal(str, coders.StrUtf8Coder)
+    self._register_coder_internal(windowed_value.PaneInfo, coders.PaneInfoCoder)
     self._register_coder_internal(typehints.TupleConstraint, coders.TupleCoder)
     self._register_coder_internal(typehints.DictConstraint, coders.MapCoder)
     self._register_coder_internal(