apache/beam · error · IllegalStateException

Equality field is not a top-level column of schema

Error message

Equality field {} is not a top-level column of schema: {}

What it means

RecordDeltaTaskWriter resolves each primary-key (equality) field to a top-level column position in the Iceberg schema. If a PK field id is absent from the schema's top-level fields, the writer cannot locate the value in the record and throws this IllegalStateException. This guards the CDC collapse logic, which indexes into rows by position.

Solutions

1. Verify every field name/id in the equality/pk field set exists as a top-level column of the table schema (table.schema().columns())
2. Re-sync the CDC configuration with the current table schema after any schema evolution (add the column back or update the equality field set)
3. Ensure you only pass top-level (non-nested) columns as equality fields

Example fix

// before
Schema schema = table.schema();
List<String> pk = List.of("id", "old_name"); // old_name no longer exists
// after
List<String> pk = schema.columns().stream().map(Types.NestedField::name)
    .filter(n -> n.equals("id") || n.equals("renamed_name"))
    .collect(Collectors.toList());

for (Types.NestedField pk : pkFields) {
  boolean isTopLevel = table.schema().columns().stream()
      .anyMatch(c -> c.fieldId() == pk.fieldId());
  if (!isTopLevel) throw new IllegalArgumentException("PK field not top-level: " + pk.name());
}

When it happens

Trigger: Constructing RecordDeltaTaskWriter with an equality/pk field list referencing a field id that is not a top-level column of the table schema — e.g. PK derived from a different schema version, a nested field, or a schema evolved to drop/rename the key column.

Common situations: Schema evolution removed or renamed the primary key column after CDC was configured; equality fields configured at the Iceberg writer level that were never added to the table; using a nested column as an equality field.

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

• Beam Row schema and Iceberg schema have different…
• Expected at least one overlapping task in bidirectional list
• Invalid starting strategy. Valid values are
• Position delete index cardinality exceeds Integer.MAX_VALUE
• RecordDeltaTaskWriter received unsorted input: a record's…

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/a1d560bf7f12bf81. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/RecordDeltaTaskWriter.java:150

      boolean upsert) {
    this.spec = spec;
    this.writerFactory = writerFactory;
    this.fileFactory = fileFactory;
    this.io = io;
    this.targetFileSize = targetFileSize;
    this.deleteSchema = deleteSchema;
    List<Types.NestedField> pkFields = deleteSchema.columns();
    this.pkPos = new int[pkFields.size()];
    // pk should only be in top-level columns
    List<Types.NestedField> allFields = schema.columns();
    Map<Integer, Integer> positionById = Maps.newHashMapWithExpectedSize(allFields.size());
    for (int j = 0; j < allFields.size(); j++) {
      positionById.put(allFields.get(j).fieldId(), j);
    }
    for (int i = 0; i < pkFields.size(); i++) {
      @Nullable Integer pos = positionById.get(pkFields.get(i).fieldId());
      if (pos == null) {
        throw new IllegalStateException(
            "Equality field "
                + pkFields.get(i).name()
                + " is not a top-level column of schema: "
                + schema);
      }
      this.pkPos[i] = pos;
    }
    this.upsert = upsert;
  }

  /** Routes a record to the {@link PartitionDeltaWriter} responsible for its partition. */
  abstract PartitionDeltaWriter route(Record row);

  /**
   * Buffers {@code row} into the current block, flushing the previous block first when {@code
   * sortKey} starts a new primary key.
   */
  public void write(byte[] sortKey, Record row, ValueKind kind) {

View on GitHub (pinned to 12126d8942)