{"record":{"id":"f3e4fe9525a2505c","repo":"apache/iceberg","slug":"invalid-primary-key-s-column-s-is-nullable-f3e4fe","errorCode":null,"errorMessage":"Invalid primary key '%s'. Column '%s' is nullable.","messagePattern":"Invalid primary key '(.+?)'\\. Column '(.+?)' is nullable\\.","errorType":"validation","errorClass":"org.apache.flink.table.api.ValidationException","httpStatus":null,"severity":"error","filePath":"flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java","lineNumber":373,"sourceCode":"    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":355,"sourceCodeEnd":381,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java#L355-L381","documentation":"Iceberg's FlinkSchemaUtil validates a Flink UniqueConstraint before building a ResolvedSchema from an Iceberg table schema. A primary key column must be non-nullable (NOT NULL), because Iceberg identifiers cannot reference nullable columns. Flink's LogicalType.isNullable() is true for most declared columns by default, so a key declared without NOT NULL fails validation with a ValidationException.","triggerScenarios":"Calling FlinkSchemaUtil.toResolvedSchema(Schema) on a table whose identifier-field columns map to nullable Flink columns, or defining a Flink table with PRIMARY KEY (col) NOT ENFORCED where col lacks NOT NULL in the DDL.","commonSituations":"CREATE TABLE DDLs like 'PRIMARY KEY (id) NOT ENFORCED' without declaring 'id STRING NOT NULL' (Flink defaults columns to nullable); syncing Iceberg tables into Flink catalogs where the identifier field column is nullable.","solutions":["Declare the primary key column as NOT NULL in the Flink DDL (e.g. 'id BIGINT NOT NULL').","Ensure the Iceberg table's identifier-field columns are required (non-optional) in the Iceberg schema.","Pick a different, non-nullable column as the primary key.","If the table genuinely has no non-nullable column, drop the primary key / identifier requirement instead of forcing one."],"exampleFix":"// before\nCREATE TABLE t (\n  id BIGINT,\n  name STRING,\n  PRIMARY KEY (id) NOT ENFORCED\n) WITH (...);\n\n// after\nCREATE TABLE t (\n  id BIGINT NOT NULL,\n  name STRING,\n  PRIMARY KEY (id) NOT ENFORCED\n) WITH (...);","handlingStrategy":"validation","validationCode":"Schema icebergSchema = table.schema();\nfor (int id : icebergSchema.identifierFieldIds()) {\n  Types.NestedField f = icebergSchema.findField(id);\n  if (f.isOptional()) {\n    throw new IllegalArgumentException(\n        \"Identifier column '\" + f.name() + \"' must be required (NOT NULL)\");\n  }\n}","typeGuard":null,"tryCatchPattern":"try {\n  ResolvedSchema rs = FlinkSchemaUtil.toResolvedSchema(table.schema());\n} catch (ValidationException e) {\n  if (e.getMessage().contains(\"is nullable\")) {\n    // fix DDL: mark key column NOT NULL\n  } else {\n    throw e;\n  }\n}","preventionTips":["Always declare primary key columns with NOT NULL in Flink DDLs.","Keep Iceberg identifier fields required in the Iceberg schema.","Validate schemas with a pre-flight check before creating/updating Iceberg-backed Flink tables."],"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"}