+ * As {@link #lookupWriterField(Schema, Field)}, except that an alias of the + * reader field takes precedence over its own name when another field of the + * reader record claims the writer field of that name by an alias. The decoder + * then renames that writer field away, so fields whose names are swapped or + * rotated by aliases each read the writer field their alias names. + *
+ * + * @param writerSchema Schema of the record where to look for the writer field. + * @param readerSchema Schema of the record the reader field belongs to. + * @param readerField Reader field to identify the corresponding writer field + * of. + * @return the writer field, if any does correspond, or null. + */ + static Field lookupWriterField(final Schema writerSchema, final Schema readerSchema, final Field readerField) { + final Field direct = writerSchema.getField(readerField.name()); + if (direct != null && isClaimedByAlias(readerSchema, readerField, direct.name())) { + Field aliased = null; + for (final String readerFieldAliasName : readerField.aliases()) { + final Field writerField = writerSchema.getField(readerFieldAliasName); + if (writerField != null) { + if (aliased != null) { + throw new AvroRuntimeException( + String.format("Reader record field %s matches multiple fields in writer record schema %s", readerField, + writerSchema)); + } + aliased = writerField; + } + } + if (aliased != null) { + return aliased; + } + } + return lookupWriterField(writerSchema, readerField); + } + + /** + * Whether a field of the reader record other than the specified one has an + * alias naming the specified writer field name. + */ + private static boolean isClaimedByAlias(final Schema readerSchema, final Field readerField, final String name) { + for (final Field other : readerSchema.getFields()) { + if (other != readerField && other.aliases().contains(name)) { + return true; + } + } + return false; + } + /** * Reader/writer schema pair that can be used as a key in a hash map. * @@ -398,7 +451,7 @@ private SchemaCompatibilityResult checkReaderWriterRecordFields(final Schema rea // record: for (final Field readerField : reader.getFields()) { location.addFirst(Integer.toString(readerField.pos())); - final Field writerField = lookupWriterField(writer, readerField); + final Field writerField = lookupWriterField(writer, reader, readerField); if (writerField == null) { // Reader field does not correspond to any field in the writer record schema, so // the diff --git a/lang/java/avro/src/test/java/org/apache/avro/TestSchemaCompatibility.java b/lang/java/avro/src/test/java/org/apache/avro/TestSchemaCompatibility.java index 12531e9e226..452f0d319da 100644 --- a/lang/java/avro/src/test/java/org/apache/avro/TestSchemaCompatibility.java +++ b/lang/java/avro/src/test/java/org/apache/avro/TestSchemaCompatibility.java @@ -571,4 +571,80 @@ void checkResolvingDecoder() throws IOException { // This is a new field, and should be null assertNull(loc.get("long_r2")); } + + @Test + void fieldsSwappedByAliasesAreCompatible() throws IOException { + // Each reader field reads the writer field its alias names, as the decoder + // reads them, + // though the writer also has a field of its own name: another field's alias + // renames it away. + Schema writer = parseRecord("{\"name\":\"a\",\"type\":\"int\"},{\"name\":\"b\",\"type\":\"string\"}"); + Schema reader = parseRecord("{\"name\":\"b\",\"type\":\"int\",\"aliases\":[\"a\"]}," + + "{\"name\":\"a\",\"type\":\"string\",\"aliases\":[\"b\"]}"); + assertEquals(SchemaCompatibilityType.COMPATIBLE, checkReaderWriterCompatibility(reader, writer).getType()); + + GenericData.Record written = new GenericData.Record(writer); + written.put("a", 5); + written.put("b", "s"); + GenericData.Record read = decode(written, reader); + assertEquals(5, read.get("b")); + assertEquals(new Utf8("s"), read.get("a")); + } + + @Test + void fieldsRotatedByAliasesAreCompatible() { + Schema writer = parseRecord("{\"name\":\"a\",\"type\":\"int\"},{\"name\":\"b\",\"type\":\"string\"}," + + "{\"name\":\"c\",\"type\":\"long\"}"); + Schema reader = parseRecord("{\"name\":\"b\",\"type\":\"int\",\"aliases\":[\"a\"]}," + + "{\"name\":\"c\",\"type\":\"string\",\"aliases\":[\"b\"]}," + + "{\"name\":\"a\",\"type\":\"long\",\"aliases\":[\"c\"]}"); + assertEquals(SchemaCompatibilityType.COMPATIBLE, checkReaderWriterCompatibility(reader, writer).getType()); + } + + @Test + void fieldsSwappedByAliasesInANestedRecordAreCompatible() { + Schema writer = parseRecord("{\"name\":\"n\",\"type\":{\"type\":\"record\",\"name\":\"N\",\"fields\":[" + + "{\"name\":\"a\",\"type\":\"int\"},{\"name\":\"b\",\"type\":\"string\"}]}}"); + Schema reader = parseRecord("{\"name\":\"n\",\"type\":{\"type\":\"record\",\"name\":\"N\",\"fields\":[" + + "{\"name\":\"b\",\"type\":\"int\",\"aliases\":[\"a\"]}," + + "{\"name\":\"a\",\"type\":\"string\",\"aliases\":[\"b\"]}]}}"); + assertEquals(SchemaCompatibilityType.COMPATIBLE, checkReaderWriterCompatibility(reader, writer).getType()); + } + + @Test + void fieldsSwappedByAliasesAreComparedWithTheFieldsTheirAliasesName() { + // b now reads a, an int, so a string b is a type mismatch. + Schema writer = parseRecord("{\"name\":\"a\",\"type\":\"int\"},{\"name\":\"b\",\"type\":\"string\"}"); + Schema reader = parseRecord("{\"name\":\"b\",\"type\":\"string\",\"aliases\":[\"a\"]}," + + "{\"name\":\"a\",\"type\":\"int\",\"aliases\":[\"b\"]}"); + SchemaPairCompatibility result = checkReaderWriterCompatibility(reader, writer); + assertEquals(SchemaCompatibilityType.INCOMPATIBLE, result.getType()); + assertEquals(SchemaIncompatibilityType.TYPE_MISMATCH, result.getResult().getIncompatibilities().get(0).getType()); + } + + @Test + void aFieldAliasingAnotherWhoseNameNoAliasFreesStillFails() { + // The decoder renames writer a to b beside writer b, a duplicate field: still + // an error. + Schema writer = parseRecord("{\"name\":\"a\",\"type\":\"int\"},{\"name\":\"b\",\"type\":\"int\"}"); + Schema reader = parseRecord( + "{\"name\":\"b\",\"type\":\"int\",\"aliases\":[\"a\"]}," + "{\"name\":\"a\",\"type\":\"int\"}"); + assertThrows(AvroRuntimeException.class, () -> checkReaderWriterCompatibility(reader, writer)); + } + + private static Schema parseRecord(String fields) { + return new Schema.Parser().parse("{\"type\":\"record\",\"name\":\"R\",\"fields\":[" + fields + "]}"); + } + + private static GenericData.Record decode(GenericData.Record written, Schema readSchema) throws IOException { + byte[] payload; + try (ByteArrayOutputStream out = new ByteArrayOutputStream()) { + Encoder encoder = EncoderFactory.get().binaryEncoder(out, null); + new GenericDatumWriter