{"record":{"id":"224f737300f82270","repo":"apache/iceberg","slug":"invalid-primary-key-s-a-primary-key-must-not-c-224f73","errorCode":null,"errorMessage":"Invalid primary key '%s'. A primary key must not contain duplicate columns. Found: %s","messagePattern":"Invalid primary key '(.+?)'\\. A primary key must not contain duplicate columns\\. Found: (.+?)","errorType":"validation","errorClass":"org.apache.flink.table.api.ValidationException","httpStatus":null,"severity":"error","filePath":"flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java","lineNumber":349,"sourceCode":"    return new ResolvedSchema(columns, Collections.emptyList(), uniqueConstraint);\n  }\n\n  /**\n   * Copied from\n   * org.apache.flink.table.catalog.DefaultSchemaResolver#validatePrimaryKey(org.apache.flink.table.catalog.UniqueConstraint,\n   * java.util.List)\n   */\n  private static void validatePrimaryKey(UniqueConstraint primaryKey, List<Column> columns) {\n    final Map<String, Column> columnsByNameLookup =\n        columns.stream().collect(Collectors.toMap(Column::getName, Function.identity()));\n\n    final Set<String> duplicateColumns =\n        primaryKey.getColumns().stream()\n            .filter(name -> Collections.frequency(primaryKey.getColumns(), name) > 1)\n            .collect(Collectors.toSet());\n\n    if (!duplicateColumns.isEmpty()) {\n      throw new ValidationException(\n          String.format(\n              \"Invalid primary key '%s'. A primary key must not contain duplicate columns. Found: %s\",\n              primaryKey.getName(), duplicateColumns));\n    }\n\n    for (String columnName : primaryKey.getColumns()) {\n      Column column = columnsByNameLookup.get(columnName);\n      if (column == null) {\n        throw new ValidationException(\n            String.format(\n                \"Invalid primary key '%s'. Column '%s' does not exist.\",\n                primaryKey.getName(), columnName));\n      }\n\n      if (!column.isPhysical()) {\n        throw new ValidationException(\n            String.format(\n                \"Invalid primary key '%s'. Column '%s' is not a physical column.\",","sourceCodeStart":331,"sourceCodeEnd":367,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java#L331-L367","documentation":"FlinkSchemaUtil.validatePrimaryKey validates a Flink UniqueConstraint used as a primary key before converting to a ResolvedSchema. A primary key column list may not contain the same column twice; duplicates are collected and a ValidationException is thrown naming the key and duplicate columns. It guards against malformed PRIMARY KEY declarations.","triggerScenarios":"Calling FlinkSchemaUtil.toResolvedSchema with a schema whose UNIQUE/PRIMARY KEY constraint lists a column more than once, e.g. PRIMARY KEY (id, id) NOT ENFORCED.","commonSituations":"Hand-written DDL duplicating a column in the PRIMARY KEY clause; programmatic schema building that appends key columns from two overlapping sources.","solutions":["Remove the duplicate column names from the PRIMARY KEY clause.","If the key is built programmatically, deduplicate the column list before constructing UniqueConstraint.","Re-run the statement after fixing the key definition."],"exampleFix":"// before\nPRIMARY KEY (user_id, user_id) NOT ENFORCED\n// after\nPRIMARY KEY (user_id) NOT ENFORCED","handlingStrategy":"validation","validationCode":"java.util.List<String> cols = primaryKey.getColumns();\nif (cols.stream().distinct().count() != cols.size()) {\n  throw new IllegalArgumentException(\"Duplicate primary key columns: \" + cols);\n}","typeGuard":null,"tryCatchPattern":"try {\n  ResolvedSchema rs = FlinkSchemaUtil.toResolvedSchema(schema);\n} catch (ValidationException e) {\n  // inspect duplicate key columns in message and fix DDL\n}","preventionTips":["Deduplicate key column lists before constructing UniqueConstraint.","Review hand-written PRIMARY KEY clauses for repeated names.","Add a unit test validating key uniqueness in schema-building code."],"tags":["flink","schema","primary-key","validation"],"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-14T11:17:12.474Z"}