claudevdm commented on code in PR #40081:
URL: https://github.com/apache/beam/pull/40081#discussion_r4018452490
##########
sdks/python/apache_beam/typehints/schemas.py:
##########
@@ -374,7 +379,17 @@ def typing_to_runner_api(self, type_: type) ->
schema_pb2.FieldType:
element_type=schema_pb2.FieldType(
atomic_type=PRIMITIVE_TO_ATOMIC_TYPE[int])))
- elif _safe_issubclass(type_, Sequence) and not _safe_issubclass(type_,
str):
+ elif _safe_issubclass(type_, tuple) and not match_is_named_tuple(type_):
+ arg_types = _get_args(type_)
+ if len(arg_types) == 2 and arg_types[1] is Ellipsis: # Tuple[typ, ...]
Review Comment:
Should we guard with update_compatibility_version? Wire format changes from
ARRAY<T> to fixed_tuple row?
##########
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:
What if value is longer than declared tuple?
--
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]
