Add upgrade compatibility check for potential coder change

Author: Abacn

sdks/python/apache_beam/typehints/native_type_compatibility.py

@@ -193,10 +193,24 @@ def match_dataclass_for_row(user_type):
   if not dataclasses.is_dataclass(user_type):
     return False
 
+  print('check ', user_type)
+
+  # pylint: disable=wrong-import-position
+  try:
+    from apache_beam.options.pipeline_options_context import get_pipeline_options  # pylint: disable=line-too-long
+  except AttributeError:
+    pass
+  else:
+    print('check 3 ', user_type)
+    opts = get_pipeline_options()
+    if opts and opts.is_compat_version_prior_to("2.73.0"):
+      return False
+
+  print('check 2 ', user_type)
+
   is_frozen = user_type.__dataclass_params__.frozen
   # avoid circular import
   try:
-    # pylint: disable=wrong-import-position
     from apache_beam.coders.typecoders import registry as coders_registry
     from apache_beam.coders import RowCoder
   except AttributeError:

sdks/python/apache_beam/typehints/native_type_compatibility_test.py

@@ -533,9 +533,9 @@ class NonFrozenRegisteredDC:
       foo: int
 
     # pylint: disable=wrong-import-position
+    from apache_beam.coders import RowCoder
     from apache_beam.coders import typecoders
     from apache_beam.coders.coders import FastPrimitivesCoder
-    from apache_beam.coders import RowCoder
 
     typecoders.registry.register_coder(FrozenRegisteredDC, RowCoder)
     typecoders.registry.register_coder(