apache/iceberg · warning

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() maps configured equality field columns to their schema IDs and compares the resulting set with the table schema's identifier field IDs. When they differ, it logs this warning and proceeds using the job-specified equality field columns for upserts. Behavior is intentional fallback, but it can produce different (and possibly unexpected) dedup semantics than identifier fields suggest.

Source

Thrown at flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java:511

    @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

1. Align equality-field-columns with the schema identifier fields, or intentionally keep them and accept the warning
2. Check table schema identifierFieldIds and update the job's equality-field-columns option to match
3. If the table identifiers are wrong, fix the schema (ALTER TABLE SET IDENTIFIER FIELDS) instead
4. Remove equality-field-columns if you want to default to the table's identifier fields
5. Ignore the warning if the subset choice is deliberate and dedup semantics are understood

Example fix

// before
equality-field-columns: user_id  // schema identifierFieldIds: [user_id, updated_at]
// after
equality-field-columns: user_id,updated_at  // matches identifierFieldIds

Set<Integer> ids = table.schema().identifierFieldIds();
List<String> cols = conf.get("equality-field-columns");
Set<Integer> mapped = cols.stream()
    .map(c -> table.schema().findField(c).fieldId())
    .collect(Collectors.toSet());
if (!mapped.equals(ids)) { /* align before submitting the job */ }

When it happens

Trigger: Using FlinkSink with upsert enabled and equality-field-columns set to a column list whose mapped field IDs do not equal schema().identifierFieldIds() — e.g. writing with equality fields [id] while the table schema declares identifierFieldIds [id, version].

Common situations: Table evolved to add identifier fields after the Flink job was configured; typo or case mismatch in equality-field-columns names; copy-pasted equality fields from another table; explicitly choosing a subset of identifier fields for dedup.

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

• Hash distribute rows by equality fields, even though {}=rang
• Unrecognized WRITE_DISTRIBUTION_MODE:
• Unrecognized WRITE_DISTRIBUTION_MODE:
• Fallback to use 'none' distribution mode, because there are
• The configured equality field column IDs {} are not matched

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