Catch decode error in Avro decoder

We should not fail query in case of any problems, during decode phase of tombstone messages
trinodb · May 17, 2023 · 54bb219 · 54bb219
1 parent 9054f64
commit 54bb219
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 8 deletions.
diff --git a/lib/trino-record-decoder/src/main/java/io/trino/decoder/avro/GenericRecordRowDecoder.java b/lib/trino-record-decoder/src/main/java/io/trino/decoder/avro/GenericRecordRowDecoder.java
@@ -46,7 +46,13 @@ public GenericRecordRowDecoder(AvroDeserializer<GenericRecord> deserializer, Set
     @Override
     public Optional<Map<DecoderColumnHandle, FieldValueProvider>> decodeRow(byte[] data)
     {
-        GenericRecord avroRecord = deserializer.deserialize(data);
+        GenericRecord avroRecord;
+        try {
+            avroRecord = deserializer.deserialize(data);
+        }
+        catch (RuntimeException e) {
+            return Optional.empty();
+        }
         return Optional.of(columnDecoders.stream()
                 .collect(toImmutableMap(Map.Entry::getKey, entry -> entry.getValue().decodeField(avroRecord))));
     }

diff --git a/lib/trino-record-decoder/src/main/java/io/trino/decoder/avro/SingleValueRowDecoder.java b/lib/trino-record-decoder/src/main/java/io/trino/decoder/avro/SingleValueRowDecoder.java
@@ -38,7 +38,13 @@ public SingleValueRowDecoder(AvroDeserializer<Object> deserializer, DecoderColum
     @Override
     public Optional<Map<DecoderColumnHandle, FieldValueProvider>> decodeRow(byte[] data)
     {
-        Object avroValue = deserializer.deserialize(data);
+        Object avroValue;
+        try {
+            avroValue = deserializer.deserialize(data);
+        }
+        catch (RuntimeException e) {
+            return Optional.empty();
+        }
         return Optional.of(ImmutableMap.of(column, new AvroColumnDecoder.ObjectValueProvider(avroValue, column.getType(), column.getName())));
     }
 }
diff --git a/lib/trino-record-decoder/src/test/java/io/trino/decoder/avro/TestAvroDecoder.java b/lib/trino-record-decoder/src/test/java/io/trino/decoder/avro/TestAvroDecoder.java
@@ -31,7 +31,6 @@
 import io.trino.spi.type.RowType;
 import io.trino.spi.type.Type;
 import io.trino.spi.type.VarbinaryType;
-import org.apache.avro.AvroTypeException;
 import org.apache.avro.Schema;
 import org.apache.avro.SchemaBuilder;
 import org.apache.avro.SchemaBuilder.FieldAssembler;
@@ -166,7 +165,7 @@ private static Map<DecoderColumnHandle, FieldValueProvider> decodeRow(byte[] avr
     {
         RowDecoder rowDecoder = DECODER_FACTORY.create(dataParams, columns);
         return rowDecoder.decodeRow(avroData)
-                .orElseThrow(AssertionError::new);
+                .orElseThrow(() -> new IllegalStateException("Problems during decode phase"));
     }
 
     private static byte[] buildAvroData(Schema schema, String name, Object value)
@@ -364,10 +363,8 @@ public void testSchemaEvolutionToIncompatibleType()
                 .toString();
 
         assertThatThrownBy(() -> decodeRow(originalIntData, ImmutableSet.of(stringColumnReadingIntData), ImmutableMap.of(DATA_SCHEMA, changedTypeSchema)))
-                .isInstanceOf(TrinoException.class)
-                .hasCauseExactlyInstanceOf(AvroTypeException.class)
-                .hasStackTraceContaining("Found int, expecting string")
-                .hasMessageMatching("Decoding Avro record failed.");
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessageMatching("Problems during decode phase");
     }
 
     @Test