{"record":{"id":"43b86d6b70995ce0","repo":"apache/iceberg","slug":"invalid-primary-key-s-column-s-is-not-a-phy-43b86d","errorCode":null,"errorMessage":"Invalid primary key '%s'. Column '%s' is not a physical column.","messagePattern":"Invalid primary key '(.+?)'\\. Column '(.+?)' is not a physical column\\.","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":365,"sourceCode":"\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));\n      }\n    }\n  }\n}\n","sourceCodeStart":347,"sourceCodeEnd":381,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java#L347-L381","documentation":"validatePrimaryKey requires every primary key column to be a physical column. If the resolved column is a computed/metadata column (column.isPhysical() false), it throws ValidationException. Iceberg identifiers must map to physical table columns.","triggerScenarios":"toResolvedSchema with a primary key that includes a computed column (e.g. AS ... expression column) or a metadata column, as in PRIMARY KEY (id, computed_ts).","commonSituations":"Including computed/generated columns in keys when porting DDL from other engines; referencing watermark or metadata columns in the key.","solutions":["Remove the computed column from the PRIMARY KEY and key on physical columns only.","If a stable identifier is needed, materialize the computed value into a physical column and use that.","Validate the key columns are physical before building the schema programmatically."],"exampleFix":"// before\nid BIGINT, ts AS CURRENT_TIMESTAMP, PRIMARY KEY (id, ts) NOT ENFORCED\n// after\nid BIGINT, PRIMARY KEY (id) NOT ENFORCED","handlingStrategy":"validation","validationCode":"for (String c : primaryKey.getColumns()) {\n  Column col = schema.getColumn(c).orElse(null);\n  if (col == null || !col.isPhysical()) {\n    throw new IllegalArgumentException(\"PK must reference physical columns, got: \" + c);\n  }\n}","typeGuard":null,"tryCatchPattern":"try {\n  ResolvedSchema rs = FlinkSchemaUtil.toResolvedSchema(schema);\n} catch (ValidationException e) {\n  // message names the non-physical column; remove it from the key\n}","preventionTips":["Never include computed (AS ...) or metadata columns in primary keys.","Materialize needed computed values into physical columns if they must be keyed.","Document that Iceberg identifiers map to physical columns only."],"tags":["flink","schema","primary-key","computed-column"],"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-23T08:17:48.524Z"}