Portable tuple logical type

sdks/python/apache_beam/coders/coder_impl.pxd

@@ -297,6 +297,7 @@ cdef class RowCoderImpl(StreamCoderImpl):
   cdef bint encoding_positions_are_trivial
   cdef list components
   cdef bint has_nullable_fields
+  cdef bint static_encoding
 
   @cython.locals(i=int, nvals=libc.stdint.int64_t, running=int, component_coder=CoderImpl,
                  null_mask=bytes, null_mask_c=char_ptr)

sdks/python/apache_beam/coders/coder_impl.py

@@ -61,6 +61,7 @@
 from apache_beam.coders.avro_record import AvroRecord
 from apache_beam.internal import cloudpickle_pickler
 from apache_beam.internal.cloudpickle import cloudpickle
+from apache_beam.typehints.schemas import _SCHEMA_OPTION_STATIC_ENCODING
 from apache_beam.typehints.schemas import named_tuple_from_schema
 from apache_beam.utils import proto_utils
 from apache_beam.utils import windowed_value
@@ -1955,31 +1956,35 @@ def __init__(self, schema, components):
         for i in self.encoding_positions)
     self.has_nullable_fields = any(
         field.type.nullable for field in self.schema.fields)
+    self.static_encoding = any(
+        opt.name == _SCHEMA_OPTION_STATIC_ENCODING and
+        opt.value.atomic_value.boolean for opt in self.schema.options)
 
   def encode_to_stream(self, value, out, nested):
-    out.write_var_int64(self.num_fields)
     attrs = [getattr(value, name) for name in self.field_names]
 
-    if self.has_nullable_fields:
-      any_nulls = False
-      for attr in attrs:
-        if attr is None:
-          any_nulls = True
-          break
-      if any_nulls:
-        out.write_var_int64((self.num_fields + 7) // 8)
-        # Pack the bits, little-endian, in consecutive bytes.
-        running = 0
-        for i, attr in enumerate(attrs):
-          if i and i % 8 == 0:
-            out.write_byte(running)
-            running = 0
-          running |= (attr is None) << (i % 8)
-        out.write_byte(running)
+    if not self.static_encoding:
+      out.write_var_int64(self.num_fields)
+      if self.has_nullable_fields:
+        any_nulls = False
+        for attr in attrs:
+          if attr is None:
+            any_nulls = True
+            break
+        if any_nulls:
+          out.write_var_int64((self.num_fields + 7) // 8)
+          # Pack the bits, little-endian, in consecutive bytes.
+          running = 0
+          for i, attr in enumerate(attrs):
+            if i and i % 8 == 0:
+              out.write_byte(running)
+              running = 0
+            running |= (attr is None) << (i % 8)
+          out.write_byte(running)
+        else:
+          out.write_byte(0)
       else:
         out.write_byte(0)
-    else:
-      out.write_byte(0)
 
     for i in range(self.num_fields):
       if not self.encoding_positions_are_trivial:
@@ -2021,13 +2026,14 @@ def encode_batch_to_stream(self, columns: Dict[str, np.ndarray], out):
       has_null_bits = np.zeros((n, ), dtype=np.uint8)
 
     for k in range(n):
-      out.write_var_int64(self.num_fields)
-      if has_null_bits[k]:
-        out.write_byte(null_bits_len)
-        for i in range(null_bits_len):
-          out.write_byte(null_bits[k, i])
-      else:
-        out.write_byte(0)
+      if not self.static_encoding:
+        out.write_var_int64(self.num_fields)
+        if has_null_bits[k]:
+          out.write_byte(null_bits_len)
+          for i in range(null_bits_len):
+            out.write_byte(null_bits[k, i])
+        else:
+          out.write_byte(0)
       for i in range(self.num_fields):
         if not self.encoding_positions_are_trivial:
           i = self.encoding_positions_argsort[i]
@@ -2040,11 +2046,15 @@ def encode_batch_to_stream(self, columns: Dict[str, np.ndarray], out):
           cython.cast(RowColumnEncoder, attrs[i]).encode_to_stream(k, out)
 
   def decode_from_stream(self, in_stream, nested):
-    nvals = in_stream.read_var_int64()
-    null_mask_len = in_stream.read_var_int64()
-    if null_mask_len:
-      # pylint: disable=unused-variable
-      null_mask_c = null_mask_py = in_stream.read(null_mask_len)
+    if self.static_encoding:
+      nvals = self.num_fields
+      null_mask_len = 0
+    else:
+      nvals = in_stream.read_var_int64()
+      null_mask_len = in_stream.read_var_int64()
+      if null_mask_len:
+        # pylint: disable=unused-variable
+        null_mask_c = null_mask_py = in_stream.read(null_mask_len)
 
     # Note that if this coder's schema has *fewer* attributes than the encoded
     # value, we just need to ignore the additional values, which will occur
@@ -2078,11 +2088,15 @@ def decode_batch_from_stream(self, dest: Dict[str, np.ndarray], in_stream):
     for k in range(n):
       if in_stream.size() == 0:
         break
-      nvals = in_stream.read_var_int64()
-      null_mask_len = in_stream.read_var_int64()
-      if null_mask_len:
-        # pylint: disable=unused-variable
-        null_mask_c = null_mask = in_stream.read(null_mask_len)
+      if self.static_encoding:
+        nvals = self.num_fields
+        null_mask_len = 0
+      else:
+        nvals = in_stream.read_var_int64()
+        null_mask_len = in_stream.read_var_int64()
+        if null_mask_len:
+          # pylint: disable=unused-variable
+          null_mask_c = null_mask = in_stream.read(null_mask_len)
 
       for i in range(min(self.num_fields, nvals)):
         if not self.encoding_positions_are_trivial:

sdks/python/apache_beam/coders/row_coder.py

@@ -208,5 +208,13 @@ def _create_impl(self):
   def is_deterministic(self):
     return self.representation_coder.is_deterministic()
 
+  def as_deterministic_coder(self, step_label, error_message=None):
+    if self.is_deterministic():
+      return self
+    return LogicalTypeCoder(
+        self.logical_type,
+        self.representation_coder.as_deterministic_coder(
+            step_label, error_message))
+
   def to_type_hint(self):
     return self.logical_type.language_type()

sdks/python/apache_beam/coders/row_coder_test.py

@@ -33,6 +33,8 @@
 from apache_beam.testing.test_pipeline import TestPipeline
 from apache_beam.testing.util import assert_that
 from apache_beam.testing.util import equal_to
+from apache_beam.typehints.schemas import _SCHEMA_OPTION_STATIC_ENCODING
+from apache_beam.typehints.schemas import _static_encoding_option_pb2
 from apache_beam.typehints.schemas import named_tuple_from_schema
 from apache_beam.typehints.schemas import typing_to_runner_api
 from apache_beam.utils.timestamp import Timestamp
@@ -510,6 +512,47 @@ def test_batch_encode_decode(self):
       for field, a in columnar.items():
         assert_array_equal(a[:n], dest[field][:n])
 
+  def test_row_coder_with_tuples(self):
+    class TupleRecord(typing.NamedTuple):
+      key: str
+      fixed_tuple: typing.Tuple[str, int]
+      var_tuple: typing.Tuple[int, ...]
+      homo_tuple: typing.Tuple[str, str]
+
+    coder = RowCoder(typing_to_runner_api(TupleRecord).row_type.schema)
+    record = TupleRecord("k1", ("hello", 42), (1, 2, 3), ("a", "b"))
+    encoded = coder.encode(record)
+    decoded = coder.decode(encoded)
+
+    self.assertEqual(record, decoded)
+    self.assertIsInstance(decoded.fixed_tuple, tuple)
+    self.assertIsInstance(decoded.var_tuple, tuple)
+    self.assertIsInstance(decoded.homo_tuple, tuple)
+    # Verify hashability as dict keys
+    d = {decoded.homo_tuple: "val1", decoded.fixed_tuple: "val2"}
+    self.assertEqual(d[("a", "b")], "val1")
+    self.assertEqual(d[("hello", 42)], "val2")
+
+  def test_static_encoding(self):
+    schema = schema_pb2.Schema(
+        fields=[
+            schema_pb2.Field(
+                name="f_int32",
+                type=schema_pb2.FieldType(atomic_type=schema_pb2.INT32)),
+            schema_pb2.Field(
+                name="f_string",
+                type=schema_pb2.FieldType(atomic_type=schema_pb2.STRING)),
+        ],
+        options=[_static_encoding_option_pb2()])
+    RowType = named_tuple_from_schema(schema)
+    row = RowType(f_int32=42, f_string="hello world!")
+    coder = RowCoder(schema)
+    encoded = coder.encode(row)
+    # VarInt(42) = 1 byte, String("hello world!") = 1 byte len + 12 chars = 13 bytes.
+    # Total = 14 bytes (0 envelope overhead, matching TupleCoder)
+    self.assertEqual(14, len(encoded))
+    self.assertEqual(row, coder.decode(encoded))
+
 
 if __name__ == "__main__":
   logging.getLogger().setLevel(logging.INFO)

sdks/python/apache_beam/transforms/sql_test.py

@@ -57,6 +57,10 @@ def __eq__(self, other):
     "UserTypeRow", [("id", int), ("arb", Aribitrary), ("complex", complex)])
 coders.registry.register_coder(UserTypeRow, coders.RowCoder)
 
+TupleRow = typing.NamedTuple(
+    "TupleRow", [("id", int), ("coords", typing.Tuple[str, int])])
+coders.registry.register_coder(TupleRow, coders.RowCoder)
+
 
 @pytest.mark.xlang_sql_expansion_service
 @unittest.skipIf(
@@ -66,19 +70,12 @@ def __eq__(self, other):
 class SqlTransformTest(unittest.TestCase):
   """Tests that exercise the cross-language SqlTransform (implemented in java).
 
-  Note this test must be executed with pipeline options that run jobs on a local
-  job server. The easiest way to accomplish this is to run the
-  `validatesCrossLanguageRunnerPythonUsingSql` gradle target for a particular
-  job server, which will start the runner and job server for you. For example,
-  `:runners:flink:1.13:job-server:validatesCrossLanguageRunnerPythonUsingSql` to
-  test on Flink 1.13.
-
-  Alternatively, you may be able to iterate faster if you run the tests directly
-  using a runner like `FlinkRunner`, which can start a local Flink cluster and
-  job server for you:
-    $ pip install -e './sdks/python[gcp,test]'
+  To run these tests locally using PrismRunner, build the SQL expansion service
+  and prism binary first:
+    $ ./gradlew :sdks:java:extensions:sql:expansion-service:shadowJar
+    $ ./gradlew :runners:prism:build
     $ pytest apache_beam/transforms/sql_test.py \\
-        --test-pipeline-options="--runner=FlinkRunner"
+        --test-pipeline-options="--runner=PrismRunner"
   """
   _multiprocess_can_split_ = True
 
@@ -229,6 +226,19 @@ def test_sql_ddl_set_option(self):
       # Verify the output matches the query (unaffected by the SET DDL)
       assert_that(out, equal_to([(3, 30)]))
 
+  def test_tuple_field(self):
+    with TestPipeline() as p:
+      out = (
+          p
+          | beam.Create([
+              TupleRow(1, ("foo", 100)),
+              TupleRow(2, ("bar", 200)),
+          ])
+          | SqlTransform(
+              "SELECT t.id, t.coords.f0 AS `name`, t.coords.f1 AS `val`, t.coords "
+              "FROM PCOLLECTION t WHERE t.coords.f1 > 150"))
+      assert_that(out, equal_to([(2, "bar", 200, ("bar", 200))]))
+
 
 if __name__ == "__main__":
   logging.getLogger().setLevel(logging.INFO)

sdks/python/apache_beam/typehints/native_type_compatibility.py

@@ -125,14 +125,16 @@ def _safe_issubclass(derived, parent):
     issubclass(derived, parent), or False if a TypeError was raised.
   """
   try:
-    return issubclass(derived, parent)
+    if issubclass(derived, parent):
+      return True
   except (TypeError, AttributeError):
-    if hasattr(derived, '__origin__'):
-      try:
-        return issubclass(derived.__origin__, parent)
-      except TypeError:
-        pass
-    return False
+    pass
+  if hasattr(derived, '__origin__') and derived.__origin__ is not None:
+    try:
+      return issubclass(derived.__origin__, parent)
+    except (TypeError, AttributeError):
+      pass
+  return False
 
 
 def _match_issubclass(match_against):

sdks/python/apache_beam/typehints/native_type_compatibility_test.py

@@ -41,6 +41,7 @@
 from apache_beam.typehints.native_type_compatibility import convert_typing_to_builtin
 from apache_beam.typehints.native_type_compatibility import is_any
 from apache_beam.typehints.native_type_compatibility import match_dataclass_for_row
+from apache_beam.typehints.native_type_compatibility import _safe_issubclass
 
 _TestNamedTuple = typing.NamedTuple(
     '_TestNamedTuple', [('age', int), ('name', bytes)])
@@ -573,6 +574,19 @@ class NonFrozenDC:
       self.assertEqual(
           compat_version == "2.73.0", match_dataclass_for_row(NonFrozenDC))
 
+  def test_safe_issubclass(self):
+    # In Python <= 3.12, issubclass(types.GenericAlias, tuple) returns False
+    # directly without raising TypeError, whereas in Python >= 3.13 it raises
+    # TypeError. _safe_issubclass inspects __origin__ to ensure types.GenericAlias
+    # (e.g. tuple[...], list[...]) is recognized consistently across Python versions.
+    self.assertTrue(_safe_issubclass(tuple[int, str], tuple))
+    self.assertTrue(_safe_issubclass(tuple[int, str], (str, tuple)))
+    self.assertTrue(_safe_issubclass(list[int], list))
+    self.assertTrue(_safe_issubclass(typing.Tuple[int, str], tuple))
+    self.assertTrue(_safe_issubclass(typing.List[int], list))
+    self.assertFalse(_safe_issubclass(int, tuple))
+    self.assertFalse(_safe_issubclass(typing.Union[int, str], tuple))
+
 
 if __name__ == '__main__':
   unittest.main()

sdks/python/apache_beam/typehints/row_type_test.py

@@ -193,6 +193,18 @@ class MyDataClass:
         row_type.GeneratedClassRowTypeConstraint(
             fields=[('id', int), ('name', str), ('name_hash', int)]))
 
+  def test_tuple_preserved_in_row_after_serialization(self):
+    with TestPipeline() as p:
+      res = (
+          p
+          | beam.Create([{"a": 1, "b": 2, "h": "h1"}])
+          | beam.GroupBy(  # group_by with custom field generates a Beam Row
+              row_field=lambda x: (x["a"], x["b"]))
+          | beam.MapTuple(
+              lambda k, vs: (type(k.row_field), k.row_field)))
+
+      assert_that(res, equal_to([(tuple, (1, 2))]))
+
 
 if __name__ == '__main__':
   unittest.main()