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

Invalid primary key '%s'. Column '%s' 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.

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)

Solutions

  1. Declare the primary key column as NOT NULL in the Flink DDL (e.g. 'id BIGINT NOT NULL').
  2. Ensure the Iceberg table's identifier-field columns are required (non-optional) in the Iceberg schema.
  3. Pick a different, non-nullable column as the primary key.
  4. 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

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.