### Question:
SchemaComparator implements Comparator<Schema> { @Override public int compare(Schema s1, Schema s2) { int comp = s1.getType().compareTo(s2.getType()); if (comp != 0) { return comp; } if (s1.getLogicalType() == null && s2.getLogicalType() != null) { return -1; } else if (s1.getLogicalType() != null && s2.getLogicalType() == null) { return 1; } else if (s1.getLogicalType() != null) { comp = s1.getLogicalType().compareTo(s2.getLogicalType()); if (comp != 0) { return comp; } } switch (s1.getType()) { case UNION: return compareLists(s1.getUnionSchemas(), s2.getUnionSchemas(), this::compare); case ARRAY: return compare(s1.getComponentSchema(), s2.getComponentSchema()); case MAP: return compareMapSchemas(s1.getMapSchema(), s2.getMapSchema()); case RECORD: return compareRecordSchemas(s1, s2); } return 0; }  @Override int compare(Schema s1, Schema s2);  }### Answer:
@Test public void testTypeComparison() { Set<Schema> schemas = new HashSet<>(Arrays.asList(INT, LONG, FLOAT, DOUBLE, STRING, BOOL, BYTES, NULL, ARRAY_INT, NULLABLE_INT, UNION_NUMERIC, MAP_STRING_INT, RECORD_FLAT)); for (Schema schema1 : schemas) { Assert.assertEquals(0, COMPARATOR.compare(schema1, schema1)); Set<Schema> otherSchemas = new HashSet<>(schemas); otherSchemas.remove(schema1); for (Schema schema2 : otherSchemas) { assertNotEqual(schema1, schema2); } } }
@Test public void testNestedRecord() { Schema middle = Schema.recordOf("middle", Schema.Field.of("int", INT), Schema.Field.of("inner", RECORD_FLAT)); Schema outer1 = Schema.recordOf("outer", Schema.Field.of("middle", middle)); Schema outer2 = Schema.recordOf("outer", Schema.Field.of("middle", RECORD_FLAT)); assertNotEqual(outer1, outer2); Assert.assertEquals(0, COMPARATOR.compare(outer1, outer1)); Assert.assertEquals(0, COMPARATOR.compare(outer2, outer2)); }
@Test public void testSimpleMap() { Assert.assertNotEquals(0, COMPARATOR.compare(Schema.mapOf(STRING, STRING), Schema.mapOf(STRING, INT))); }
@Test public void testNestedMapValues() { Assert.assertNotEquals(0, COMPARATOR.compare( Schema.mapOf(STRING, RECORD_FLAT), Schema.mapOf(STRING, Schema.recordOf("x", Schema.Field.of("x", INT))))); }
@Test public void testSamePhysicalDifferentLogicalTypes() { Assert.assertNotEquals(0, COMPARATOR.compare(Schema.of(Schema.LogicalType.DATE), Schema.of(Schema.LogicalType.TIME_MILLIS))); Assert.assertNotEquals(0, COMPARATOR.compare(Schema.of(Schema.LogicalType.TIME_MICROS), Schema.of(Schema.LogicalType.TIMESTAMP_MILLIS))); }