{"record":{"id":"abab7b021d33f2ae","repo":"apache/iceberg","slug":"invalid-primary-key-s-column-s-does-not-exi","errorCode":null,"errorMessage":"Invalid primary key '%s'. Column '%s' does not exist.","messagePattern":"Invalid primary key '(.+?)'\\. Column '(.+?)' does not exist\\.","errorType":"validation","errorClass":"org.apache.flink.table.api.ValidationException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java","lineNumber":358,"sourceCode":"    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.\",\n                primaryKey.getName(), columnName));\n      }\n\n      final LogicalType columnType = column.getDataType().getLogicalType();\n      if (columnType.isNullable()) {\n        throw new ValidationException(\n            String.format(\n                \"Invalid primary key '%s'. Column '%s' is nullable.\",\n                primaryKey.getName(), columnName));","sourceCodeStart":340,"sourceCodeEnd":376,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java#L340-L376","documentation":"validatePrimaryKey checks each primary-key column against the table's column lookup; if the named column is absent from the schema, a ValidationException is thrown. The primary key must reference existing (by-name) columns of the Flink schema being converted.","triggerScenarios":"PRIMARY KEY referencing a column that was renamed, dropped, or never defined in the CREATE TABLE; mismatch between the constraint built in code and the columns passed to toResolvedSchema.","commonSituations":"Schema evolution changed column names while the key clause stayed stale; copying DDL between tables; typo in the key column name.","solutions":["Align the PRIMARY KEY column names with the actual columns in the table definition.","Update the key after schema changes (ALTER that renames/drops columns).","Validate programmatically: check each key column exists in the schema's columns before constructing the ResolvedSchema."],"exampleFix":"// before\nCREATE TABLE t (id BIGINT, data STRING, PRIMARY KEY (uid) NOT ENFORCED);\n// after\nCREATE TABLE t (id BIGINT, data STRING, PRIMARY KEY (id) NOT ENFORCED);","handlingStrategy":"validation","validationCode":"for (String col : primaryKey.getColumns()) {\n  if (schema.getColumn(col) == null) {\n    throw new IllegalArgumentException(\"PK column missing in schema: \" + col);\n  }\n}","typeGuard":"null","tryCatchPattern":"try {\n  FlinkSchemaUtil.toResolvedSchema(schema, partitionKeys, primaryKey);\n} catch (ValidationException e) {\n  LOG.error(\"PK references unknown column: {}\", e.getMessage());\n  throw e;\n}","preventionTips":["Update PRIMARY KEY clauses whenever columns are renamed or dropped.","Validate key columns against the schema in tests after every schema change.","Avoid copy-pasting DDL between tables without rechecking key columns."],"tags":["flink","schema","primary-key"],"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"}