{"record":{"id":"b115569c2105d166","repo":"apache/iceberg","slug":"the-configured-equality-field-column-ids-are-no-b11556","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.1/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.1/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java#L494-L530","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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"],"exampleFix":"// before\nFlinkSink.forRowData(input).equalityFieldColumns(\"id\")...\n// after (table PK is id, name)\nFlinkSink.forRowData(input).equalityFieldColumns(\"id, name\")...","handlingStrategy":"validation","validationCode":"Schema schema = table.schema();\nSet<Integer> configuredIds = equalityColumns.stream()\n    .map(c -> schema.findField(c).fieldId())\n    .collect(Collectors.toSet());\nif (!configuredIds.equals(schema.identifierFieldIds())) {\n  throw new IllegalArgumentException(\"equalityFieldColumns must match table PK: \"\n      + schema.identifierFieldIds());\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["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"],"tags":["flink","iceberg","upsert","schema"],"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"}