The configured equality field column IDs {} are not matched

Error message

The configured equality field column IDs {} are not matched with the schema identifier field IDs {}, use job specified equality field columns as the equality fields by default.

What it means

FlinkSink.checkAndGetEqualityFieldIds warns when the equality-field columns explicitly configured for the job (FlinkSink.builder().equalityFieldColumns) resolve to a set of field IDs that differs from the table schema's identifierFieldIds (primary keys). The job-specified columns win, but the mismatch usually signals a stale or divergent table definition.

Source

Thrown at flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java:512

    @VisibleForTesting
    List<Integer> checkAndGetEqualityFieldIds() {
      List<Integer> equalityFieldIds = Lists.newArrayList(table.schema().identifierFieldIds());
      if (equalityFieldColumns != null && !equalityFieldColumns.isEmpty()) {
        Set<Integer> equalityFieldSet =
            Sets.newHashSetWithExpectedSize(equalityFieldColumns.size());
        for (String column : equalityFieldColumns) {
          org.apache.iceberg.types.Types.NestedField field = table.schema().findField(column);
          Preconditions.checkNotNull(
              field,
              "Missing required equality field column '%s' in table schema %s",
              column,
              table.schema());
          equalityFieldSet.add(field.fieldId());
        }

        if (!equalityFieldSet.equals(table.schema().identifierFieldIds())) {
          LOG.warn(
              "The configured equality field column IDs {} are not matched with the schema identifier field IDs"
                  + " {}, use job specified equality field columns as the equality fields by default.",
              equalityFieldSet,
              table.schema().identifierFieldIds());
        }
        equalityFieldIds = Lists.newArrayList(equalityFieldSet);
      }
      return equalityFieldIds;
    }

    private DataStreamSink<Void> appendDummySink(SingleOutputStreamOperator<Void> committerStream) {
      DataStreamSink<Void> resultStream =
          committerStream
              .sinkTo(new DiscardingSink<>())
              .name(operatorName(String.format("IcebergSink %s", this.table.name())))
              .setParallelism(1);
      if (uidPrefix != null) {
        resultStream = resultStream.uid(uidPrefix + "-dummysink");

View on GitHub (pinned to 86d9c8fc54)

Solutions

Align equalityFieldColumns with the table's identifier fields, or remove the explicit setting so the schema identifier fields are used.
Run SELECT * FROM table.metadata_log / inspect the schema to confirm current identifierFieldIds.
Restart the job after updating the sink builder config to match the evolved schema.

Example fix

// before
FlinkSink.forRowData(input)
    .equalityFieldColumns("user_id") // table identifier is now (user_id, country)
    .append();
// after
FlinkSink.forRowData(input)
    .equalityFieldColumns("user_id", "country")
    .append();

Defensive patterns

Strategy: validation

Validate before calling

Set<String> configured = SinkUtil.checkAndGetEqualityFieldIds(table, eqColumns);
if (!configured.equals(table.schema().identifierFieldIds())) {
  throw new IllegalArgumentException("equalityFieldColumns must match identifierFieldIds: " + table.schema().identifierFieldIds());
}

Prevention

Derive equality fields from the table schema instead of hardcoding them
Re-validate sink config after any schema evolution on the table
Use the same source of truth (schema identifier fields) for job and table

When it happens

Trigger: Building a FlinkSink with equalityFieldColumns set on a table whose schema identifier fields were added/changed (e.g. via ALTER TABLE ... SET IDENTIFIER FIELDS or schema evolution) so the resolved ID sets are no longer equal.

Common situations: Table primary keys changed after the Flink job was configured; spelling/case differences in column names resolving to different IDs; copying sink config across tables with different identifier fields.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

AI-assisted analysis of apache/iceberg@86d9c8fc54 (2026-09-12). Data as JSON: /api/errors/c7ff3f4bb2e287ce. Report an issue: GitHub.