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

Invalid primary key '%s'. Column '%s' does not exist.

Error message

Invalid primary key '%s'. Column '%s' does not exist.

What it means

validatePrimaryKey verifies each primary key column exists in the resolved schema. If a key column name is not found among the table's columns, a ValidationException 'Invalid primary key ... Column does not exist' is thrown.

Source

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

    final Map<String, Column> columnsByNameLookup =
        columns.stream().collect(Collectors.toMap(Column::getName, Function.identity()));

    final Set<String> duplicateColumns =
        primaryKey.getColumns().stream()
            .filter(name -> Collections.frequency(primaryKey.getColumns(), name) > 1)
            .collect(Collectors.toSet());

    if (!duplicateColumns.isEmpty()) {
      throw new ValidationException(
          String.format(
              "Invalid primary key '%s'. A primary key must not contain duplicate columns. Found: %s",
              primaryKey.getName(), duplicateColumns));
    }

    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. Correct the primary key column names to match columns in the schema
  2. Verify the key columns exist after schema changes (renames/drops)
  3. Derive key columns programmatically from the schema rather than hardcoding
  4. Mind case sensitivity of column names in DDL

Example fix

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

// after
CREATE TABLE t (id BIGINT, name STRING, PRIMARY KEY (id) NOT ENFORCED);
Defensive patterns

Strategy: validation

Validate before calling

Set<String> names = schema.getColumns().stream().map(Column::getName).collect(Collectors.toSet());
keyColumns.forEach(c -> { if (!names.contains(c)) throw new IllegalArgumentException("key column missing: " + c); });

Try / catch

try { FlinkSchemaUtil.toResolvedSchema(schema); } catch (ValidationException e) { /* fix key column names */ }

Prevention

When it happens

Trigger: PRIMARY KEY referencing a misspelled or renamed column; programmatically built key list out of sync with the schema; case-mismatched column names.

Common situations: Schema evolution renamed/dropped a column while the key definition was stale; DDL copied between tables; generated code with hardcoded column names.

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/6c3acbb733b9baf3. Report an issue: GitHub.