robinyqiu commented on a change in pull request #14759:
URL: https://github.com/apache/beam/pull/14759#discussion_r629621891
##########
File path:
sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/SchemaUtils.java
##########
@@ -58,13 +60,66 @@
/** Convert DataCatalog schema to Beam schema. */
static Schema fromDataCatalog(com.google.cloud.datacatalog.v1beta1.Schema
dcSchema) {
- return fromColumnsList(dcSchema.getColumnsList());
+ if (dcSchema.hasPhysicalSchema()) {
+ return fromPhysicalSchema(dcSchema.getPhysicalSchema());
+ } else {
+ return fromColumnsList(dcSchema.getColumnsList());
+ }
}
private static Schema fromColumnsList(List<ColumnSchema> columnsMap) {
return
columnsMap.stream().map(SchemaUtils::toBeamField).collect(toSchema());
}
+ private static Schema fromPhysicalSchema(PhysicalSchema physicalSchema) {
+ if (physicalSchema.hasAvro()) {
+ return fromAvroSchema(new
Parser().parse(physicalSchema.getAvro().getText()));
+ } else if (physicalSchema.hasProtobuf()) {
+ return null; // TODO
+ } else {
+ throw new UnsupportedOperationException(
+ "Unsupported Data Catalog physical schema format. "
+ + "Currently we only support Avro and Protobuf.");
+ }
+ }
+
+ private static Schema fromAvroSchema(org.apache.avro.Schema avroSchema) {
Review comment:
Thanks for the pointer, we should use that class eventually. This is
just some quick code that gets the integration test work.
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
