Skip to content

Commit 0a50f31

Browse files
authored
Do not set representsation for pythonsdk_any type when typehint is Any (#38325)
* Do not set representsation for pythonsdk_any type when typehint is Any * fix formatting
1 parent 2e5a3a0 commit 0a50f31

2 files changed

Lines changed: 53 additions & 28 deletions

File tree

sdks/python/apache_beam/typehints/schemas.py

Lines changed: 35 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -258,34 +258,38 @@ def schema_field(
258258
description=description)
259259

260260

261-
def _python_any_schema_pb2():
261+
def _python_any_schema_pb2(has_repr):
262262
# A portable schema matches FastPrimitivesCoder encoded values
263+
if has_repr:
264+
representation = schema_pb2.FieldType(
265+
nullable=False,
266+
row_type=schema_pb2.RowType(
267+
schema=schema_pb2.Schema(
268+
fields=[
269+
schema_pb2.Field(
270+
name=_PYTHON_ANY_FIELD_TYPE_BYTE,
271+
type=schema_pb2.FieldType(
272+
atomic_type=schema_pb2.BYTE, nullable=False)),
273+
schema_pb2.Field(
274+
name=_PYTHON_ANY_FIELD_PAYLOAD,
275+
type=schema_pb2.FieldType(
276+
atomic_type=schema_pb2.BYTES, nullable=False))
277+
],
278+
options=[
279+
schema_pb2.Option(
280+
name=_SCHEMA_OPTION_STATIC_ENCODING,
281+
type=schema_pb2.FieldType(
282+
atomic_type=schema_pb2.BOOLEAN),
283+
value=schema_pb2.FieldValue(
284+
atomic_value=schema_pb2.AtomicTypeValue(
285+
boolean=True)))
286+
]))) if has_repr else None
287+
else:
288+
representation = None
289+
263290
return schema_pb2.FieldType(
264291
logical_type=schema_pb2.LogicalType(
265-
urn=PYTHON_ANY_URN,
266-
representation=schema_pb2.FieldType(
267-
nullable=False,
268-
row_type=schema_pb2.RowType(
269-
schema=schema_pb2.Schema(
270-
fields=[
271-
schema_pb2.Field(
272-
name=_PYTHON_ANY_FIELD_TYPE_BYTE,
273-
type=schema_pb2.FieldType(
274-
atomic_type=schema_pb2.BYTE, nullable=False)),
275-
schema_pb2.Field(
276-
name=_PYTHON_ANY_FIELD_PAYLOAD,
277-
type=schema_pb2.FieldType(
278-
atomic_type=schema_pb2.BYTES, nullable=False))
279-
],
280-
options=[
281-
schema_pb2.Option(
282-
name=_SCHEMA_OPTION_STATIC_ENCODING,
283-
type=schema_pb2.FieldType(
284-
atomic_type=schema_pb2.BOOLEAN),
285-
value=schema_pb2.FieldValue(
286-
atomic_value=schema_pb2.AtomicTypeValue(
287-
boolean=True)))
288-
])))),
292+
urn=PYTHON_ANY_URN, representation=representation),
289293
nullable=True)
290294

291295

@@ -388,14 +392,18 @@ def typing_to_runner_api(self, type_: type) -> schema_pb2.FieldType:
388392
return schema_pb2.FieldType(
389393
array_type=schema_pb2.ArrayType(element_type=element_type))
390394

395+
elif type_ == Any:
396+
return _python_any_schema_pb2(has_repr=False)
397+
391398
try:
392399
if LogicalType.is_known_logical_type(type_):
393400
logical_type = type_
394401
else:
395402
logical_type = LogicalType.from_typing(type_)
396403
except ValueError:
397-
# Unknown type, just treat it like Any
398-
return _python_any_schema_pb2()
404+
# Unknown type, use pythonsdk_any with a representation compatible with
405+
# FastPrimitiveCoder encoded UNKNOWN type
406+
return _python_any_schema_pb2(has_repr=True)
399407
else:
400408
argument_type = None
401409
argument = None

sdks/python/apache_beam/typehints/schemas_test.py

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -582,9 +582,26 @@ def test_proto_survives_typing_roundtrip(self, fieldtype_proto):
582582
fieldtype_proto, schema_registry=SchemaTypeRegistry()),
583583
schema_registry=SchemaTypeRegistry()))
584584

585+
def test_any_maps_to_any(self):
586+
# python_any for typing.Any logical type's representation is delibrately set
587+
# absent to prevent the usage crossing language boundary, as its encoded
588+
# form isn't predictable from foreign SDK.
589+
self.assertEqual(
590+
typing_to_runner_api(Any),
591+
schemas._python_any_schema_pb2(has_repr=False))
592+
585593
def test_unknown_primitive_maps_to_any(self):
586594
self.assertEqual(
587-
typing_to_runner_api(np.uint32), schemas._python_any_schema_pb2())
595+
typing_to_runner_api(np.uint32),
596+
schemas._python_any_schema_pb2(has_repr=True))
597+
598+
def test_unknown_user_type_maps_to_any(self):
599+
class MyUnknownType:
600+
pass
601+
602+
self.assertEqual(
603+
typing_to_runner_api(MyUnknownType),
604+
schemas._python_any_schema_pb2(has_repr=True))
588605

589606
def test_unknown_atomic_raise_valueerror(self):
590607
self.assertRaises(

0 commit comments

Comments
 (0)