Pass ParsedSchema to SchemaParser

trinodb · Jan 4, 2021 · 23a4334 · 23a4334
1 parent 2aa8c20
commit 23a4334
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 4 deletions.
diff --git a/...in/trino-kafka/src/main/java/io/trino/plugin/kafka/schema/confluent/AvroSchemaParser.java b/...in/trino-kafka/src/main/java/io/trino/plugin/kafka/schema/confluent/AvroSchemaParser.java
@@ -14,6 +14,8 @@
 package io.trino.plugin.kafka.schema.confluent;
 
 import com.google.common.collect.ImmutableList;
+import io.confluent.kafka.schemaregistry.ParsedSchema;
+import io.confluent.kafka.schemaregistry.avro.AvroSchema;
 import io.trino.decoder.avro.AvroRowDecoderFactory;
 import io.trino.plugin.kafka.KafkaTopicFieldDescription;
 import io.trino.plugin.kafka.KafkaTopicFieldGroup;
@@ -27,6 +29,7 @@
 import java.util.List;
 import java.util.Optional;
 
+import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.collect.Iterables.getOnlyElement;
 import static io.trino.plugin.kafka.schema.confluent.ConfluentSessionProperties.getEmptyFieldStrategy;
@@ -44,9 +47,10 @@ public AvroSchemaParser(TypeManager typeManager)
     }
 
     @Override
-    public KafkaTopicFieldGroup parse(ConnectorSession session, String subject, String rawSchema)
+    public KafkaTopicFieldGroup parse(ConnectorSession session, String subject, ParsedSchema parsedSchema)
     {
-        Schema schema = new Schema.Parser().parse(rawSchema);
+        checkArgument(parsedSchema instanceof AvroSchema, "parsedSchema should be an instance of AvroSchema");
+        Schema schema = ((AvroSchema) parsedSchema).rawSchema();
         AvroSchemaConverter schemaConverter = new AvroSchemaConverter(typeManager, getEmptyFieldStrategy(session));
         List<Type> types = schemaConverter.convertAvroSchema(schema);
         ImmutableList.Builder<KafkaTopicFieldDescription> fieldsBuilder = ImmutableList.builder();

diff --git a/.../trino/plugin/kafka/schema/confluent/ConfluentSchemaRegistryTableDescriptionSupplier.java b/.../trino/plugin/kafka/schema/confluent/ConfluentSchemaRegistryTableDescriptionSupplier.java
@@ -192,7 +192,14 @@ private KafkaTopicFieldGroup getFieldGroup(ConnectorSession session, String subj
         if (schemaParser == null) {
             throw new TrinoException(NOT_SUPPORTED, "Not supported schema: " + schemaMetadata.getSchemaType());
         }
-        return schemaParser.parse(session, subject, schemaMetadata.getSchema());
+        return schemaParser.parse(
+                session,
+                subject,
+                schemaRegistryClient.parseSchema(
+                        schemaMetadata.getSchemaType(),
+                        schemaMetadata.getSchema(),
+                        schemaMetadata.getReferences())
+                        .orElseThrow());
     }
 
     private SchemaMetadata getLatestSchemaMetadata(String subject)

diff --git a/plugin/trino-kafka/src/main/java/io/trino/plugin/kafka/schema/confluent/SchemaParser.java b/plugin/trino-kafka/src/main/java/io/trino/plugin/kafka/schema/confluent/SchemaParser.java
@@ -13,10 +13,11 @@
  */
 package io.trino.plugin.kafka.schema.confluent;
 
+import io.confluent.kafka.schemaregistry.ParsedSchema;
 import io.trino.plugin.kafka.KafkaTopicFieldGroup;
 import io.trino.spi.connector.ConnectorSession;
 
 public interface SchemaParser
 {
-    KafkaTopicFieldGroup parse(ConnectorSession session, String subject, String rawSchema);
+    KafkaTopicFieldGroup parse(ConnectorSession session, String subject, ParsedSchema parsedSchema);
 }