Abacn commented on code in PR #40081:
URL: https://github.com/apache/beam/pull/40081#discussion_r4047948801
##########
sdks/python/apache_beam/typehints/schemas.py:
##########
@@ -1511,3 +1537,88 @@ def argument(self):
@classmethod
def _from_typing(cls, typ):
return cls()
+
+
+_TUPLE_NAMEDTUPLE_CACHE: Dict[int, type] = {}
+
+
+def _get_tuple_namedtuple(n: int) -> type:
+ cls = _TUPLE_NAMEDTUPLE_CACHE.get(n)
+ if cls is None:
+ cls = NamedTuple(f"_FixedTuple{n}", [(f"f{i}", object) for i in range(n)])
+ _TUPLE_NAMEDTUPLE_CACHE[n] = cls
+ return cls
+
+
+@LogicalType._register_internal
+class FixedTupleLogicalType(NoArgumentLogicalType[tuple, Any]):
+ """Logical type representing fixed-length Python tuples backed by a Row."""
+ def __init__(self, tuple_types: Sequence[type] = ()):
+ self._tuple_types = tuple(tuple_types)
+
+ @classmethod
+ def urn(cls):
+ return FIXED_TUPLE_URN
+
+ def language_type(self=None):
+ if self is None or not self._tuple_types:
+ return tuple
+ return Tuple[self._tuple_types]
+
+ def representation_type(self):
+ if not self._tuple_types:
+ from apache_beam.pvalue import Row
+ return Row
+ fields = [(f"f{i}", t) for i, t in enumerate(self._tuple_types)]
+ options = []
+ st = SchemaTranslation(schema_registry=SCHEMA_REGISTRY)
+ if not any(st.typing_to_runner_api(t).nullable for t in self._tuple_types):
+ options.append((_SCHEMA_OPTION_STATIC_ENCODING, True))
+ return row_type.RowTypeConstraint.from_fields(
+ fields, schema_options=options)
+
+ def to_representation_type(self, value):
+ cls = _get_tuple_namedtuple(len(value))
Review Comment:
If the actual value `(1, "a", "b")` mismatches the schema derived from
typehint `tuple[int, str]`, currently it silently truncates. Changed to check
length.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
