shunping commented on code in PR #39662:
URL: https://github.com/apache/beam/pull/39662#discussion_r4213441783
##########
sdks/python/apache_beam/io/gcp/bigquery_schema_tools_test.py:
##########
@@ -512,6 +512,151 @@ def test_type_overrides_can_override_default_types(self):
bq_field_to_type("GEOGRAPHY", "REQUIRED", overrides), bytes)
+class TestBeamSchemaToBqTableSchema(unittest.TestCase):
+ """Tests for beam_schema_to_bq_table_schema, the reverse of
+ generate_user_type_from_bq_schema, used to auto-infer a destination
+ table's schema from a schema'd PCollection (e.g. for FILE_LOADS)."""
+ def test_atomic_types(self):
+ import decimal
+
+ class MyRow(typing.NamedTuple):
+ name: str
+ age: np.int64
+ score: float
+ active: bool
+ raw: bytes
+ amount: decimal.Decimal
+
+ schema = beam.typehints.schemas.schema_from_element_type(MyRow)
+ bq_schema = bigquery_schema_tools.beam_schema_to_bq_table_schema(schema)
+
+ self.assertEqual(
+ bq_schema,
+ {
+ 'fields': [
+ {
+ 'name': 'name', 'type': 'STRING', 'mode': 'REQUIRED'
+ },
+ {
+ 'name': 'age', 'type': 'INT64', 'mode': 'REQUIRED'
+ },
+ {
+ 'name': 'score', 'type': 'FLOAT64', 'mode': 'REQUIRED'
+ },
+ {
+ 'name': 'active', 'type': 'BOOL', 'mode': 'REQUIRED'
+ },
+ {
+ 'name': 'raw', 'type': 'BYTES', 'mode': 'REQUIRED'
+ },
+ {
+ 'name': 'amount', 'type': 'NUMERIC', 'mode': 'REQUIRED'
+ },
+ ]
+ })
+
+ def test_timestamp_type(self):
+ from apache_beam.utils.timestamp import Timestamp
+
+ class MyRow(typing.NamedTuple):
+ when: Timestamp
Review Comment:
Can we also test datetime.date and datetime.time here?
##########
sdks/python/apache_beam/io/gcp/bigquery_schema_tools.py:
##########
@@ -116,6 +116,122 @@ def bq_field_to_type(field, mode, type_overrides=None):
raise ValueError(f"Encountered an unsupported mode: {mode!r}")
+_ATOMIC_TYPE_TO_BQ_TYPE = {
+ schema_pb2.BOOLEAN: 'BOOL',
+ schema_pb2.BYTES: 'BYTES',
+ schema_pb2.STRING: 'STRING',
+ schema_pb2.BYTE: 'INT64',
+ schema_pb2.INT16: 'INT64',
+ schema_pb2.INT32: 'INT64',
+ schema_pb2.INT64: 'INT64',
+ schema_pb2.FLOAT: 'FLOAT64',
+ schema_pb2.DOUBLE: 'FLOAT64',
+}
+
+
+def _logical_type_to_bq_type_map():
+ # Built lazily to avoid a module-level import cycle between
+ # bigquery_schema_tools and typehints.schemas.
+ schemas = apache_beam.typehints.schemas
+ return {
+ schemas.MillisInstant.urn(): 'TIMESTAMP',
+ schemas.MicrosInstant.urn(): 'TIMESTAMP',
+ schemas.Date.urn(): 'DATE',
+ schemas.DecimalLogicalType.urn(): 'NUMERIC',
+ schemas.FixedPrecisionDecimalLogicalType.urn(): 'NUMERIC',
Review Comment:
There are also two new Tuple logical types added recently in
https://github.com/apache/beam/blob/5394822d0176d6ae3c5db0085870c6e3da362acf/sdks/python/apache_beam/typehints/schemas.py#L40-L41.
cc'ed @Abacn
Could we also support and test them?
--
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]
