Comments.

Author: claudevdm

sdks/python/apache_beam/coders/coder_impl.pxd

@@ -82,7 +82,7 @@ cdef class FastPrimitivesCoderImpl(StreamCoderImpl):
   cdef object requires_deterministic_step_label
   cdef bint warn_deterministic_fallback
   cdef bint force_use_dill
-  cdef bint skip_use_relative_filepaths
+  cdef bint use_relative_filepaths
 
   @cython.locals(dict_value=dict, int_value=libc.stdint.int64_t,
                  unicode_value=unicode)

sdks/python/apache_beam/coders/coder_impl.py

@@ -379,13 +379,13 @@ def __init__(
       fallback_coder_impl,
       requires_deterministic_step_label=None,
       force_use_dill=False,
-      skip_use_relative_filepaths=False):
+      use_relative_filepaths=True):
     self.fallback_coder_impl = fallback_coder_impl
     self.iterable_coder_impl = IterableCoderImpl(self)
     self.requires_deterministic_step_label = requires_deterministic_step_label
     self.warn_deterministic_fallback = True
     self.force_use_dill = force_use_dill
-    self.skip_use_relative_filepaths = skip_use_relative_filepaths
+    self.use_relative_filepaths = use_relative_filepaths
 
   @staticmethod
   def register_iterable_like_type(t):
@@ -567,7 +567,7 @@ def encode_type(self, t, stream):
           id_generator=None,
           skip_reset_dynamic_type_state=True,
           filepath_interceptor=cloudpickle.get_relative_path)
-      if self.skip_use_relative_filepaths:
+      if not self.use_relative_filepaths:
         config.filepath_interceptor = None
       _pickled_types[t] = cloudpickle_pickler.dumps(t, config=config)
     stream.write(_pickled_types[t], True)

sdks/python/apache_beam/coders/coders.py

@@ -930,21 +930,21 @@ class DeterministicFastPrimitivesCoderV2(FastCoder):
   def __init__(self, coder, step_label, update_compatibility_version=None):
     self._underlying_coder = coder
     self._step_label = step_label
-    self._skip_use_relative_filepaths = False
+    self._use_relative_filepaths = True
     self._version_tag = "v2_69"
     from apache_beam.transforms.util import is_v1_prior_to_v2
     # Versions prior to 2.69.0 did not use relative filepaths.
     if update_compatibility_version and is_v1_prior_to_v2(
         v1=update_compatibility_version, v2="2.69.0"):
       self._version_tag = ""
-      self._skip_use_relative_filepaths = True
+      self._use_relative_filepaths = False
 
   def _create_impl(self):
     return coder_impl.FastPrimitivesCoderImpl(
         self._underlying_coder.get_impl(),
         requires_deterministic_step_label=self._step_label,
         force_use_dill=False,
-        skip_use_relative_filepaths=self._skip_use_relative_filepaths)
+        use_relative_filepaths=self._use_relative_filepaths)
 
   def is_deterministic(self):
     # type: () -> bool
@@ -1024,6 +1024,16 @@ def _should_force_use_dill(update_compat_version):
 
 
 def _update_compatible_deterministic_fast_primitives_coder(coder, step_label):
+  """ Returns the update compatible version of DeterministicFastPrimitivesCoder
+   The differences are in how "special types" e.g. NamedTuples, Dataclasses are
+   deterministically encoded.
+
+   - In SDK version <= 2.67.0 dill is used to encode "special types"
+   - In SDK version 2.68.0 cloudpickle is used to encode "special types" with
+   absolute filepaths in code objects and dynamic functions.
+   - In SDK version 2.69.0 cloudpickle is used to encode "special types" with
+   relative filepaths in code objects and dynamic functions.
+  """
   from apache_beam.coders import typecoders
   update_compat_version = typecoders.registry.update_compatibility_version
   if _should_force_use_dill(update_compat_version):

sdks/python/apache_beam/coders/coders_test_common.py

@@ -252,6 +252,14 @@ def test_memoizing_pickle_coder(self):
       param(compat_version="2.68.0"),
   ])
   def test_deterministic_coder(self, compat_version):
+    """ Test in process determinism for all special deterministic types
+
+    - In SDK version <= 2.67.0 dill is used to encode "special types"
+    - In SDK version 2.68.0 cloudpickle is used to encode "special types" with
+    absolute filepaths in code objects and dynamic functions.
+    - In SDK version 2.69.0 cloudpickle is used to encode "special types" with
+    relative filepaths in code objects and dynamic functions.
+    """
 
     typecoders.registry.update_compatibility_version = compat_version
     coder = coders.FastPrimitivesCoder()
@@ -329,6 +337,15 @@ def test_deterministic_coder(self, compat_version):
       param(compat_version="2.68.0"),
   ])
   def test_deterministic_map_coder_is_update_compatible(self, compat_version):
+    """ Test in process determinism for map coder including when a component
+    coder uses DeterministicFastPrimitivesCoder for "special types".
+
+    - In SDK version <= 2.67.0 dill is used to encode "special types"
+    - In SDK version 2.68.0 cloudpickle is used to encode "special types" with
+    absolute filepaths in code objects and dynamic functions.
+    - In SDK version 2.69.0 cloudpickle is used to encode "special types" with
+    relative file
+    """
     typecoders.registry.update_compatibility_version = compat_version
     values = [{
         MyTypedNamedTuple(i, 'a'): MyTypedNamedTuple('a', i)
@@ -688,7 +705,14 @@ def test_param_windowed_value_coder(self):
   ])
   def test_cross_process_encoding_of_special_types_is_deterministic(
       self, compat_version):
-    """Test cross-process determinism for all special deterministic types"""
+    """Test cross-process determinism for all special deterministic types
+
+    - In SDK version <= 2.67.0 dill is used to encode "special types"
+    - In SDK version 2.68.0 cloudpickle is used to encode "special types" with
+    absolute filepaths in code objects and dynamic functions.
+    - In SDK version 2.69.0 cloudpickle is used to encode "special types" with
+    relative filepaths in code objects and dynamic functions.
+    """
     is_using_dill = compat_version == "2.67.0"
     if is_using_dill:
       pytest.importorskip("dill")
@@ -806,6 +830,8 @@ def run_subprocess():
         continue
 
       if test_name == "named_tuple_simple" and not is_using_dill:
+        # The absense of a compat_version means we are using the most recent
+        # implementation of the coder, which uses relative paths.
         should_have_relative_path = not compat_version
         named_tuple_type = type(decoded1)
         self.assertEqual(