What causes "Invalid primary key '%s'. Column '%s' is nullable." in apache/iceberg?

Declare the primary key column as NOT NULL in the Flink DDL (e.g. 'id BIGINT NOT NULL').

apache/iceberg · error · org.apache.flink.table.api.ValidationException

Invalid primary key ' '. Column ' ' is nullable.

Error message

Invalid primary key '%s'. Column '%s' is nullable.

What it means

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.

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.

Example fix

// before
CREATE TABLE t (
  id BIGINT,
  name STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (...);

// after
CREATE TABLE t (
  id BIGINT NOT NULL,
  name STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (...);

Defensive patterns

Strategy: validation

Validate before calling

Schema icebergSchema = table.schema();
for (int id : icebergSchema.identifierFieldIds()) {
  Types.NestedField f = icebergSchema.findField(id);
  if (f.isOptional()) {
    throw new IllegalArgumentException(
        "Identifier column '" + f.name() + "' must be required (NOT NULL)");
  }
}

Try / catch

try {
  ResolvedSchema rs = FlinkSchemaUtil.toResolvedSchema(table.schema());
} catch (ValidationException e) {
  if (e.getMessage().contains("is nullable")) {
    // fix DDL: mark key column NOT NULL
  } else {
    throw e;
  }
}

Prevention

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.

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

AI-assisted analysis of apache/iceberg@86d9c8fc54 (2026-09-12). Data as JSON: /api/errors/f3e4fe9525a2505c. Report an issue: GitHub.

Appendix: source

Thrown at flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java:373

    for (String columnName : primaryKey.getColumns()) {
      Column column = columnsByNameLookup.get(columnName);
      if (column == null) {
        throw new ValidationException(
            String.format(
                "Invalid primary key '%s'. Column '%s' does not exist.",
                primaryKey.getName(), columnName));
      }

      if (!column.isPhysical()) {
        throw new ValidationException(
            String.format(
                "Invalid primary key '%s'. Column '%s' is not a physical column.",
                primaryKey.getName(), columnName));
      }

      final LogicalType columnType = column.getDataType().getLogicalType();
      if (columnType.isNullable()) {
        throw new ValidationException(
            String.format(
                "Invalid primary key '%s'. Column '%s' is nullable.",
                primaryKey.getName(), columnName));
      }
    }
  }
}

View on GitHub (pinned to 86d9c8fc54)