Trino fails to read Avro Union type

[altrack]

Hey,

I'm trying to query a Kafka topic with Avro records and getting the following error: [65536] Query failed (#20230618_145535_00362_hh5bx): Incompatible UNION type: '[ { "type" : "record", "name" : "MyRecord", The schema has a union type with multiple record types. Is it something Trino should be able to read, or it's not supported yet?

Thanks

Trino version 418, Driver 419.

[nmarasoiu]

same here.. by mistake we have [string] only just one branch - fails..

[altrack]

Any updates here? Maybe someone aware of a workaround?

[wendigo]

@jklamer is this fixed already?

[hashhar]

union of dis-similar types may not work.

there was https://github.com/trinodb/trino/pull/20233 but it fixes failures when a common super-type can be found for the union.

[altrack]

I tested this with a Trino version after the fix - Driver: 445 and Server: 443, and received the same error of Incompatible UNION type.
Avro doesn't have a concept of a super-type. Custom types in Avro are just "record"s.

Here is the stacktrace of the exception:

java.lang.UnsupportedOperationException: Incompatible UNION type: '[ {
....

    at io.trino.plugin.kafka.schema.confluent.AvroSchemaConverter.convertUnion(AvroSchemaConverter.java:179)
    at io.trino.plugin.kafka.schema.confluent.AvroSchemaConverter.convertUnion(AvroSchemaConverter.java:160)
    at io.trino.plugin.kafka.schema.confluent.AvroSchemaConverter.convert(AvroSchemaConverter.java:137)
    at io.trino.plugin.kafka.schema.confluent.AvroSchemaConverter.convertRecordSchema(AvroSchemaConverter.java:105)
    at io.trino.plugin.kafka.schema.confluent.AvroSchemaConverter.convertAvroSchema(AvroSchemaConverter.java:91)
    at io.trino.plugin.kafka.schema.confluent.AvroSchemaParser.parse(AvroSchemaParser.java:54)
    at io.trino.plugin.kafka.schema.confluent.ConfluentSchemaRegistryTableDescriptionSupplier.getFieldGroup(ConfluentSchemaRegistryTableDescriptionSupplier.java:213)
    at io.trino.plugin.kafka.schema.confluent.ConfluentSchemaRegistryTableDescriptionSupplier.lambda$getTopicDescription$2(ConfluentSchemaRegistryTableDescriptionSupplier.java:202)
    at java.base/java.util.Optional.map(Optional.java:260)
    at io.trino.plugin.kafka.schema.confluent.ConfluentSchemaRegistryTableDescriptionSupplier.getTopicDescription(ConfluentSchemaRegistryTableDescriptionSupplier.java:202)
    at io.trino.plugin.kafka.KafkaMetadata.getTopicDescription(KafkaMetadata.java:265)
    at io.trino.plugin.kafka.KafkaMetadata.getTableHandle(KafkaMetadata.java:92)
    at io.trino.plugin.kafka.KafkaMetadata.getTableHandle(KafkaMetadata.java:63)
    at io.trino.spi.connector.ConnectorMetadata.getTableHandle(ConnectorMetadata.java:141)
    at io.trino.plugin.base.classloader.ClassLoaderSafeConnectorMetadata.getTableHandle(ClassLoaderSafeConnectorMetadata.java:1243)
    at io.trino.tracing.TracingConnectorMetadata.getTableHandle(TracingConnectorMetadata.java:149)
    at io.trino.metadata.MetadataManager.lambda$getTableHandle$5(MetadataManager.java:282)
    at java.base/java.util.Optional.flatMap(Optional.java:289)
    at io.trino.metadata.MetadataManager.getTableHandle(MetadataManager.java:276)
    at io.trino.metadata.MetadataManager.getRedirectionAwareTableHandle(MetadataManager.java:1883)
    at io.trino.metadata.MetadataManager.getRedirectionAwareTableHandle(MetadataManager.java:1875)
    at io.trino.tracing.TracingMetadata.getRedirectionAwareTableHandle(TracingMetadata.java:1475)
    at io.trino.sql.analyzer.StatementAnalyzer$Visitor.getTableHandle(StatementAnalyzer.java:5843)
    at io.trino.sql.analyzer.StatementAnalyzer$Visitor.visitTable(StatementAnalyzer.java:2288)
    at io.trino.sql.analyzer.StatementAnalyzer$Visitor.visitTable(StatementAnalyzer.java:522)
    at io.trino.sql.tree.Table.accept(Table.java:60)
    at io.trino.sql.tree.AstVisitor.process(AstVisitor.java:27)
    at io.trino.sql.analyzer.StatementAnalyzer$Visitor.process(StatementAnalyzer.java:541)
    at io.trino.sql.analyzer.StatementAnalyzer$Visitor.analyzeFrom(StatementAnalyzer.java:4892)
    at io.trino.sql.analyzer.StatementAnalyzer$Visitor.visitQuerySpecification(StatementAnalyzer.java:3085)
    at io.trino.sql.analyzer.StatementAnalyzer$Visitor.visitQuerySpecification(StatementAnalyzer.java:522)
    at io.trino.sql.tree.QuerySpecification.accept(QuerySpecification.java:155)
    at io.trino.sql.tree.AstVisitor.process(AstVisitor.java:27)
    at io.trino.sql.analyzer.StatementAnalyzer$Visitor.process(StatementAnalyzer.java:541)
    at io.trino.sql.analyzer.StatementAnalyzer$Visitor.process(StatementAnalyzer.java:549)
    at io.trino.sql.analyzer.StatementAnalyzer$Visitor.visitQuery(StatementAnalyzer.java:1566)
    at io.trino.sql.analyzer.StatementAnalyzer$Visitor.visitQuery(StatementAnalyzer.java:522)
    at io.trino.sql.tree.Query.accept(Query.java:118)
    at io.trino.sql.tree.AstVisitor.process(AstVisitor.java:27)
    at io.trino.sql.analyzer.StatementAnalyzer$Visitor.process(StatementAnalyzer.java:541)
    at io.trino.sql.analyzer.StatementAnalyzer.analyze(StatementAnalyzer.java:501)
    at io.trino.sql.analyzer.StatementAnalyzer.analyze(StatementAnalyzer.java:490)
    at io.trino.sql.analyzer.Analyzer.analyze(Analyzer.java:97)
    at io.trino.sql.analyzer.Analyzer.analyze(Analyzer.java:86)
    at io.trino.execution.SqlQueryExecution.analyze(SqlQueryExecution.java:281)
    at io.trino.execution.SqlQueryExecution.<init>(SqlQueryExecution.java:214)
    at io.trino.execution.SqlQueryExecution$SqlQueryExecutionFactory.createQueryExecution(SqlQueryExecution.java:878)
    at io.trino.dispatcher.LocalDispatchQueryFactory.lambda$createDispatchQuery$0(LocalDispatchQueryFactory.java:153)
    at io.trino.$gen.Trino_443____20240418_065708_2.call(Unknown Source)
    at com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:131)
    at com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:76)
    at com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:82)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
    at java.base/java.lang.Thread.run(Thread.java:1583)