SchemaUtil { static boolean compareSchemaFieldType(Schema.FieldType a, Schema.FieldType b) { if (a.getTypeName().equals(b.getTypeName())) { return !a.getTypeName().equals(Schema.TypeName.LOGICAL_TYPE) || compareSchemaFieldType( a.getLogicalType().getBaseType(), b.getLogicalType().getBaseType()); } else if (a.getTypeName().isLogicalType()) { return a.getLogicalType().getBaseType().getTypeName().equals(b.getTypeName()); } else if (b.getTypeName().isLogicalType()) { return b.getLogicalType().getBaseType().getTypeName().equals(a.getTypeName()); } return false; }  static boolean compareSchemaField(Schema.Field a, Schema.Field b);  }
@Test public void testSchemaFieldTypeComparator() { assertTrue(SchemaUtil.compareSchemaFieldType(Schema.FieldType.STRING, Schema.FieldType.STRING)); assertFalse(SchemaUtil.compareSchemaFieldType(Schema.FieldType.STRING, Schema.FieldType.INT16)); assertTrue( SchemaUtil.compareSchemaFieldType( LogicalTypes.variableLengthString(JDBCType.VARCHAR, 255), LogicalTypes.variableLengthString(JDBCType.VARCHAR, 255))); assertFalse( SchemaUtil.compareSchemaFieldType( LogicalTypes.variableLengthString(JDBCType.VARCHAR, 255), LogicalTypes.fixedLengthBytes(JDBCType.BIT, 255))); assertTrue( SchemaUtil.compareSchemaFieldType( Schema.FieldType.STRING, LogicalTypes.variableLengthString(JDBCType.VARCHAR, 255))); assertFalse( SchemaUtil.compareSchemaFieldType( Schema.FieldType.INT16, LogicalTypes.variableLengthString(JDBCType.VARCHAR, 255))); assertTrue( SchemaUtil.compareSchemaFieldType( LogicalTypes.variableLengthString(JDBCType.VARCHAR, 255), Schema.FieldType.STRING)); }