{"record":{"id":"d2ad64995b235c01","repo":"apache/iceberg","slug":"the-configured-equality-field-column-ids-are-no","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/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java","lineNumber":511,"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":493,"sourceCodeEnd":529,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java#L493-L529","documentation":"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.","triggerScenarios":"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].","commonSituations":"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.","solutions":["Align equality-field-columns with the schema identifier fields, or intentionally keep them and accept the warning","Check table schema identifierFieldIds and update the job's equality-field-columns option to match","If the table identifiers are wrong, fix the schema (ALTER TABLE SET IDENTIFIER FIELDS) instead","Remove equality-field-columns if you want to default to the table's identifier fields","Ignore the warning if the subset choice is deliberate and dedup semantics are understood"],"exampleFix":"// before\nequality-field-columns: user_id  // schema identifierFieldIds: [user_id, updated_at]\n// after\nequality-field-columns: user_id,updated_at  // matches identifierFieldIds","handlingStrategy":"validation","validationCode":"Set<Integer> ids = table.schema().identifierFieldIds();\nList<String> cols = conf.get(\"equality-field-columns\");\nSet<Integer> mapped = cols.stream()\n    .map(c -> table.schema().findField(c).fieldId())\n    .collect(Collectors.toSet());\nif (!mapped.equals(ids)) { /* align before submitting the job */ }","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Keep equality-field-columns in sync with ALTER TABLE SET IDENTIFIER FIELDS changes","Validate column names (case/typos) against the schema at job startup","Default to identifier fields instead of configuring equality fields explicitly","Re-check table schema after evolution before reusing saved job configs"],"tags":["flink","sink","upsert","schema","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"}