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 rejects a Flink UniqueConstraint (primary key) whose column list contains duplicates, since a key cannot repeat the same column. The duplicate names are collected and included in the ValidationException message. This occurs while converting a Flink schema/ResolvedSchema that includes a primary key.

Source

Thrown at flink/v1.20/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

  1. Remove duplicate column names from the PRIMARY KEY clause / UniqueConstraint column list.
  2. Deduplicate keys programmatically: use a LinkedHashSet on the key columns before constructing the constraint.
  3. Fix the code generator that emits the PRIMARY KEY so each column appears once.

Example fix

// before
PRIMARY KEY (order_id, order_id) NOT ENFORCED
// after
PRIMARY KEY (order_id) NOT ENFORCED
Defensive patterns

Strategy: validation

Validate before calling

Set<String> seen = new HashSet<>();
for (String col : primaryKey.getColumns()) {
  if (!seen.add(col)) throw new IllegalArgumentException("Duplicate PK column: " + col);
}

Type guard

null

Try / catch

try {
  FlinkSchemaUtil.toResolvedSchema(schema, partitionKeys, primaryKey);
} catch (ValidationException e) {
  LOG.error("Primary key invalid: {}", e.getMessage());
  throw e;
}

Prevention

When it happens

Trigger: Defining a Flink table DDL with a PRIMARY KEY clause listing the same column twice (e.g. PRIMARY KEY (id, id) NOT ENFORCED), or building a UniqueConstraint programmatically with a duplicated column list.

Common situations: Generated/templated DDL that concatenates key columns without dedup; hand-written SQL typos; schema evolution tooling merging multiple key definitions.

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/483200096c4ecb3a. Report an issue: GitHub.