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

SinkUtil.checkAndGetEqualityFieldIds resolves the user-configured equality field columns to schema field IDs and compares them with the table schema's identifier field IDs. When the sets differ, it logs this warning and proceeds using the job-specified equality field columns. This matters for upsert mode: rows are matched on equality fields, which may silently diverge from the table's declared primary key (identifier fields).

Source

Thrown at flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/SinkUtil.java:74

  private static final Logger LOG = LoggerFactory.getLogger(SinkUtil.class);

  static Set<Integer> checkAndGetEqualityFieldIds(Table table, List<String> equalityFieldColumns) {
    Set<Integer> equalityFieldIds = Sets.newHashSet(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 = Sets.newHashSet(equalityFieldSet);
    }
    return equalityFieldIds;
  }

  static long getMaxCommittedCheckpointId(
      Table table, String flinkJobId, String operatorId, String branch) {
    Snapshot snapshot = table.snapshot(branch);
    long lastCommittedCheckpointId = INITIAL_CHECKPOINT_ID;

    while (snapshot != null) {
      Map<String, String> summary = snapshot.summary();
      String snapshotFlinkJobId = summary.get(FLINK_JOB_ID);

View on GitHub (pinned to 86d9c8fc54)

Solutions

Make the configured equality field columns exactly match the table's identifier field columns.
Remove explicit equality-field-columns so the sink defaults to the schema identifier fields.
If the divergence is intentional, update the Iceberg schema identifiers (or accept the warning) and document that upsert matching uses the job-specified columns.

Example fix

// before
FlinkSink.forRowData(input)
    .setEqualityFieldColumns("order_id")
    ...

// after — match the table's identifier fields
FlinkSink.forRowData(input)
    .setEqualityFieldColumns("order_id", "line_number")
    ...

Defensive patterns

Strategy: validation

Validate before calling

java.util.Set<Integer> configured = table.schema().columns().stream()
    .filter(c -> equalityColumns.contains(c.name()))
    .map(Types.NestedField::fieldId)
    .collect(java.util.stream.Collectors.toSet());
if (!configured.equals(table.schema().identifierFieldIds())) {
  throw new IllegalArgumentException(
      "equality-field-columns must match schema identifier fields: "
          + table.schema().identifierFieldIds());
}

Prevention

Derive equality columns from table.schema().identifierFieldIds() instead of hardcoding.
Re-check equality column config whenever the table's primary key changes.
Validate configured column names resolve to existing schema fields before submitting the job.

When it happens

Trigger: Calling SinkUtil.checkAndGetEqualityFieldIds (used when building Flink upsert sinks) with a table whose schema().identifierFieldIds() does not equal the set of IDs derived from the configured 'equality-field-columns' (e.g. via FlinkSink.Builder.setEqualityFieldColumns or table write.upsert.enabled with a partial column list).

Common situations: Table has an identifier (primary key) defined in Iceberg schema but the Flink job passes a different subset of columns as equality fields; identifiers were added to the table after the job was configured; typo in a configured column name so only some columns resolve.

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/21f57d3cf9de93a1. Report an issue: GitHub.