static boolean compareSchemaFieldType(Schema.FieldType a, Schema.FieldType b) {
    if (a.getTypeName().equals(b.getTypeName())) {
      if (!a.getTypeName().isLogicalType()) {
        return true;
      } else {
        Schema.LogicalType<?, ?> aLogicalType = checkArgumentNotNull(a.getLogicalType());
        Schema.LogicalType<?, ?> bLogicalType = checkArgumentNotNull(b.getLogicalType());
        return compareSchemaFieldType(aLogicalType.getBaseType(), bLogicalType.getBaseType());
      }
    } else if (a.getTypeName().isLogicalType()) {
      Schema.LogicalType<?, ?> aLogicalType = checkArgumentNotNull(a.getLogicalType());
      return aLogicalType.getBaseType().getTypeName().equals(b.getTypeName());
    } else if (b.getTypeName().isLogicalType()) {
      Schema.LogicalType<?, ?> bLogicalType = checkArgumentNotNull(b.getLogicalType());
      return bLogicalType.getBaseType().getTypeName().equals(a.getTypeName());
    }
    return false;
  }
@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));
  }