{"record":{"id":"d01f0a796870e751","repo":"apache/iceberg","slug":"the-configured-equality-field-column-ids-are-no-d01f0a","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/SinkUtil.java","lineNumber":74,"sourceCode":"\n  private static final Logger LOG = LoggerFactory.getLogger(SinkUtil.class);\n\n  static Set<Integer> checkAndGetEqualityFieldIds(Table table, List<String> equalityFieldColumns) {\n    Set<Integer> equalityFieldIds = Sets.newHashSet(table.schema().identifierFieldIds());\n    if (equalityFieldColumns != null && !equalityFieldColumns.isEmpty()) {\n      Set<Integer> equalityFieldSet = 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 = Sets.newHashSet(equalityFieldSet);\n    }\n    return equalityFieldIds;\n  }\n\n  static long getMaxCommittedCheckpointId(\n      Table table, String flinkJobId, String operatorId, String branch) {\n    Snapshot snapshot = table.snapshot(branch);\n    long lastCommittedCheckpointId = INITIAL_CHECKPOINT_ID;\n\n    while (snapshot != null) {\n      Map<String, String> summary = snapshot.summary();\n      String snapshotFlinkJobId = summary.get(FLINK_JOB_ID);","sourceCodeStart":56,"sourceCodeEnd":92,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/SinkUtil.java#L56-L92","documentation":"SinkUtil.checkAndGetEqualityFieldIds warns when the equality-field columns configured for the job resolve to field IDs that differ from the table schema's identifierFieldIds (primary keys). The job-specified columns are used as equality fields by default; the mismatch typically means the table's identifier fields evolved after the job config was written.","triggerScenarios":"Calling SinkUtil.checkAndGetEqualityFieldIds (used by sink builders) with equalityFieldColumns set whose resolved ID set does not equal table.schema().identifierFieldIds().","commonSituations":"Primary keys changed via schema evolution after the job was configured; column-name typos/case differences resolving to different IDs; reusing sink configuration across tables.","solutions":["Update equalityFieldColumns to match the table's identifier fields, or drop the explicit setting.","Inspect the current schema's identifierFieldIds to confirm the intended key.","Restart the job with corrected config so equality semantics match the table."],"exampleFix":"// before\n.equalityFieldColumns(\"order_id\") // identifier fields now (order_id, region)\n// after\n.equalityFieldColumns(\"order_id\", \"region\")","handlingStrategy":"validation","validationCode":"Set<Integer> ids = SinkUtil.checkAndGetEqualityFieldIds(table, eqColumns);\nif (!ids.equals(table.schema().identifierFieldIds())) {\n  throw new IllegalArgumentException(\"configured equality fields differ from identifierFieldIds\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Keep equalityFieldColumns in sync with schema identifier fields","Re-check after schema evolution or table migration","Prefer deriving equality fields from the schema"],"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"}