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

Invalid primary key '%s'. A primary key must not contain dup

Error message

Invalid primary key '%s'. A primary key must not contain duplicate columns. Found: %s

What it means

FlinkSchemaUtil.validatePrimaryKey validates a Flink UniqueConstraint used as a primary key before converting to a ResolvedSchema. A primary key column list may not contain the same column twice; duplicates are collected and a ValidationException is thrown naming the key and duplicate columns. It guards against malformed PRIMARY KEY declarations.

Source

Thrown at flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java:349

    return new ResolvedSchema(columns, Collections.emptyList(), uniqueConstraint);
  }

  /**
   * Copied from
   * org.apache.flink.table.catalog.DefaultSchemaResolver#validatePrimaryKey(org.apache.flink.table.catalog.UniqueConstraint,
   * java.util.List)
   */
  private static void validatePrimaryKey(UniqueConstraint primaryKey, List<Column> columns) {
    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.",

View on GitHub (pinned to 86d9c8fc54)

Solutions

Remove the duplicate column names from the PRIMARY KEY clause.
If the key is built programmatically, deduplicate the column list before constructing UniqueConstraint.
Re-run the statement after fixing the key definition.

Example fix

// before
PRIMARY KEY (user_id, user_id) NOT ENFORCED
// after
PRIMARY KEY (user_id) NOT ENFORCED

Defensive patterns

Strategy: validation

Validate before calling

java.util.List<String> cols = primaryKey.getColumns();
if (cols.stream().distinct().count() != cols.size()) {
  throw new IllegalArgumentException("Duplicate primary key columns: " + cols);
}

Try / catch

try {
  ResolvedSchema rs = FlinkSchemaUtil.toResolvedSchema(schema);
} catch (ValidationException e) {
  // inspect duplicate key columns in message and fix DDL
}

Prevention

Deduplicate key column lists before constructing UniqueConstraint.
Review hand-written PRIMARY KEY clauses for repeated names.
Add a unit test validating key uniqueness in schema-building code.

When it happens

Trigger: Calling FlinkSchemaUtil.toResolvedSchema with a schema whose UNIQUE/PRIMARY KEY constraint lists a column more than once, e.g. PRIMARY KEY (id, id) NOT ENFORCED.

Common situations: Hand-written DDL duplicating a column in the PRIMARY KEY clause; programmatic schema building that appends key columns from two overlapping sources.

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/224f737300f82270. Report an issue: GitHub.