@VisibleForTesting
  static Optional<String> checkSchemas(
      final LogicalSchema schema,
      final LogicalSchema other
  ) {
    final Optional<String> keyError = checkSchemas(schema.key(), other.key(), "key ")
        .map(msg -> "Key columns must be identical. " + msg);
    if (keyError.isPresent()) {
      return keyError;
    }

    return checkSchemas(schema.columns(), other.columns(), "");
  }
@Test
  public void shouldEnforceNoReorderingKey() {
    // Given:
    final LogicalSchema someSchema = LogicalSchema.builder()
        .keyColumn(ColumnName.of("k0"), SqlTypes.INTEGER)
        .keyColumn(ColumnName.of("k1"), SqlTypes.BIGINT)
        .valueColumn(ColumnName.of("f0"), SqlTypes.STRING)
        .build();

    final LogicalSchema otherSchema = LogicalSchema.builder()
        .keyColumn(ColumnName.of("k1"), SqlTypes.BIGINT)
        .keyColumn(ColumnName.of("k0"), SqlTypes.INTEGER)
        .valueColumn(ColumnName.of("f0"), SqlTypes.STRING)
        .build();

    // When:
    final Optional<String> s = StructuredDataSource.checkSchemas(someSchema, otherSchema);

    // Then:
    assertThat(s.isPresent(), is(true));
    assertThat(s.get(), containsString("The following key columns are changed, missing or reordered: [`k0` INTEGER KEY, `k1` BIGINT KEY]"));
  }