AVRO-4355: [Java] Allow field name swaps by aliases in SchemaCompatibility

lang/java/avro/src/main/java/org/apache/avro/SchemaCompatibility.java

@@ -138,6 +138,59 @@ public static Field lookupWriterField(final Schema writerSchema, final Field rea
     }
   }
 
+  /**
+   * Identifies the writer field that corresponds to the specified field of a
+   * reader record, as {@link Schema#applyAliases} pairs them when reading.
+   *
+   * <p>
+   * 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.
+   * </p>
+   *
+   * @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

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<GenericData.Record>(written.getSchema()).write(written, encoder);
+      encoder.flush();
+      payload = out.toByteArray();
+    }
+    GenericDatumReader<GenericData.Record> reader = new GenericDatumReader<>(written.getSchema(), readSchema);
+    return reader.read(null, DecoderFactory.get().binaryDecoder(payload, null));
+  }
 }