{"record":{"id":"a1d560bf7f12bf81","repo":"apache/beam","slug":"equality-field-is-not-a-top-level-column-of-schema","errorCode":null,"errorMessage":"Equality field {} is not a top-level column of schema: {}","messagePattern":"Equality field (.+?) is not a top-level column of schema: (.+?)","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/RecordDeltaTaskWriter.java","lineNumber":150,"sourceCode":"      boolean upsert) {\n    this.spec = spec;\n    this.writerFactory = writerFactory;\n    this.fileFactory = fileFactory;\n    this.io = io;\n    this.targetFileSize = targetFileSize;\n    this.deleteSchema = deleteSchema;\n    List<Types.NestedField> pkFields = deleteSchema.columns();\n    this.pkPos = new int[pkFields.size()];\n    // pk should only be in top-level columns\n    List<Types.NestedField> allFields = schema.columns();\n    Map<Integer, Integer> positionById = Maps.newHashMapWithExpectedSize(allFields.size());\n    for (int j = 0; j < allFields.size(); j++) {\n      positionById.put(allFields.get(j).fieldId(), j);\n    }\n    for (int i = 0; i < pkFields.size(); i++) {\n      @Nullable Integer pos = positionById.get(pkFields.get(i).fieldId());\n      if (pos == null) {\n        throw new IllegalStateException(\n            \"Equality field \"\n                + pkFields.get(i).name()\n                + \" is not a top-level column of schema: \"\n                + schema);\n      }\n      this.pkPos[i] = pos;\n    }\n    this.upsert = upsert;\n  }\n\n  /** Routes a record to the {@link PartitionDeltaWriter} responsible for its partition. */\n  abstract PartitionDeltaWriter route(Record row);\n\n  /**\n   * Buffers {@code row} into the current block, flushing the previous block first when {@code\n   * sortKey} starts a new primary key.\n   */\n  public void write(byte[] sortKey, Record row, ValueKind kind) {","sourceCodeStart":132,"sourceCodeEnd":168,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/RecordDeltaTaskWriter.java#L132-L168","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Verify every field name/id in the equality/pk field set exists as a top-level column of the table schema (table.schema().columns())","Re-sync the CDC configuration with the current table schema after any schema evolution (add the column back or update the equality field set)","Ensure you only pass top-level (non-nested) columns as equality fields"],"exampleFix":"// before\nSchema schema = table.schema();\nList<String> pk = List.of(\"id\", \"old_name\"); // old_name no longer exists\n// after\nList<String> pk = schema.columns().stream().map(Types.NestedField::name)\n    .filter(n -> n.equals(\"id\") || n.equals(\"renamed_name\"))\n    .collect(Collectors.toList());","handlingStrategy":"validation","validationCode":"for (Types.NestedField pk : pkFields) {\n  boolean isTopLevel = table.schema().columns().stream()\n      .anyMatch(c -> c.fieldId() == pk.fieldId());\n  if (!isTopLevel) throw new IllegalArgumentException(\"PK field not top-level: \" + pk.name());\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Validate equality fields against table.schema() at pipeline-construction time, before submitting the job","Re-run PK validation after any schema evolution","Never configure nested columns as equality fields"],"tags":["iceberg","cdc","schema-mismatch"],"backgroundTag":"schema-validation-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}