{"record":{"id":"c7ff3f4bb2e287ce","repo":"apache/iceberg","slug":"the-configured-equality-field-column-ids-are-no-c7ff3f","errorCode":null,"errorMessage":"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.","messagePattern":"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\\.","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java","lineNumber":512,"sourceCode":"\n    @VisibleForTesting\n    List<Integer> checkAndGetEqualityFieldIds() {\n      List<Integer> equalityFieldIds = Lists.newArrayList(table.schema().identifierFieldIds());\n      if (equalityFieldColumns != null && !equalityFieldColumns.isEmpty()) {\n        Set<Integer> equalityFieldSet =\n            Sets.newHashSetWithExpectedSize(equalityFieldColumns.size());\n        for (String column : equalityFieldColumns) {\n          org.apache.iceberg.types.Types.NestedField field = table.schema().findField(column);\n          Preconditions.checkNotNull(\n              field,\n              \"Missing required equality field column '%s' in table schema %s\",\n              column,\n              table.schema());\n          equalityFieldSet.add(field.fieldId());\n        }\n\n        if (!equalityFieldSet.equals(table.schema().identifierFieldIds())) {\n          LOG.warn(\n              \"The configured equality field column IDs {} are not matched with the schema identifier field IDs\"\n                  + \" {}, use job specified equality field columns as the equality fields by default.\",\n              equalityFieldSet,\n              table.schema().identifierFieldIds());\n        }\n        equalityFieldIds = Lists.newArrayList(equalityFieldSet);\n      }\n      return equalityFieldIds;\n    }\n\n    private DataStreamSink<Void> appendDummySink(SingleOutputStreamOperator<Void> committerStream) {\n      DataStreamSink<Void> resultStream =\n          committerStream\n              .sinkTo(new DiscardingSink<>())\n              .name(operatorName(String.format(\"IcebergSink %s\", this.table.name())))\n              .setParallelism(1);\n      if (uidPrefix != null) {\n        resultStream = resultStream.uid(uidPrefix + \"-dummysink\");","sourceCodeStart":494,"sourceCodeEnd":530,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java#L494-L530","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":"// before\nFlinkSink.forRowData(input)\n    .equalityFieldColumns(\"user_id\") // table identifier is now (user_id, country)\n    .append();\n// after\nFlinkSink.forRowData(input)\n    .equalityFieldColumns(\"user_id\", \"country\")\n    .append();","handlingStrategy":"validation","validationCode":"Set<String> configured = SinkUtil.checkAndGetEqualityFieldIds(table, eqColumns);\nif (!configured.equals(table.schema().identifierFieldIds())) {\n  throw new IllegalArgumentException(\"equalityFieldColumns must match identifierFieldIds: \" + table.schema().identifierFieldIds());\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["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"],"tags":["flink","schema","equality-fields","configuration"],"backgroundTag":"schema-validation-failed","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}