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 primary key (UniqueConstraint) whose column list contains duplicates. A primary key must uniquely identify rows with a distinct set of columns; duplicates make the constraint invalid, so a ValidationException is thrown naming the duplicated columns.

Source

Thrown at flink/v2.3/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. Deduplicate the primary key column list before constructing the schema/constraint
  2. Fix the DDL so each primary key column appears once
  3. Validate uniqueness of key columns in code that assembles schemas dynamically

Example fix

// before
List<String> keys = Stream.concat(leftKeys, rightKeys).collect(toList());

// after
List<String> keys = Stream.concat(leftKeys, rightKeys).distinct().collect(toList());
Defensive patterns

Strategy: validation

Validate before calling

Set<String> seen = new HashSet<>();
for (String c : keyColumns) { if (!seen.add(c)) throw new IllegalArgumentException("duplicate key column: " + c); }

Try / catch

try { FlinkSchemaUtil.toResolvedSchema(schema); } catch (ValidationException e) { /* inspect primary key definition */ }

Prevention

When it happens

Trigger: Defining a Flink table DDL with PRIMARY KEY (a, a, b) or building a ResolvedSchema whose UniqueConstraint lists the same column twice, then converting via FlinkSchemaUtil.toResolvedSchema.

Common situations: Programmatic schema construction where key columns are appended from multiple sources without dedup; hand-written DDL typos; SQL builders concatenating key lists.

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/74d3e6552fa7f858. Report an issue: GitHub.