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

When writing upsert/equality-delete rows, FlinkSink compares the user-configured equality field columns against the table schema's identifier field IDs. If they differ, it logs this warning and proceeds using the job-specified equality field columns rather than the table's primary key definition.

Source

Thrown at flink/v2.1/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 the equality field columns with the table's identifier (primary key) fields: .equalityFieldColumns("id, name")
Or update the table schema's identifier field IDs to match the job configuration
If the difference is intentional, ignore the warning and document it - behavior still uses job-specified columns

Example fix

// before
FlinkSink.forRowData(input).equalityFieldColumns("id")...
// after (table PK is id, name)
FlinkSink.forRowData(input).equalityFieldColumns("id, name")...

Defensive patterns

Strategy: validation

Validate before calling

Schema schema = table.schema();
Set<Integer> configuredIds = equalityColumns.stream()
    .map(c -> schema.findField(c).fieldId())
    .collect(Collectors.toSet());
if (!configuredIds.equals(schema.identifierFieldIds())) {
  throw new IllegalArgumentException("equalityFieldColumns must match table PK: "
      + schema.identifierFieldIds());
}

Prevention

Keep table primary key and job equalityFieldColumns in sync (infrastructure-as-code checks)
Re-validate configuration after any ALTER TABLE SET IDENTIFIER FIELDS change
Resolve column names case-sensitively per schema to avoid silent ID mismatches

When it happens

Trigger: FlinkSink builder equalityFieldColumns (or inferred rowtype field IDs) resolve to a set of column IDs that is not exactly equal to table.schema().identifierFieldIds() for an upsert-mode sink.

Common situations: Table's primary key was changed after the Flink job was written; user specified a subset or different columns than the SQL primary key; case/name mismatch resolving columns to IDs.

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