{"record":{"id":"1893e1acb73cc02e","repo":"apache/beam","slug":"beam-row-schema-and-iceberg-schema-have-different-sizes-beam","errorCode":null,"errorMessage":"Beam Row schema and Iceberg schema have different sizes.\n\tBeam Row columns: {beamColumns}\n\tIceberg schema columns: {icebergColumns}","messagePattern":"Beam Row schema and Iceberg schema have different sizes\\.\n\tBeam Row columns: (.+?)\n\tIceberg schema columns: (.+?)","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java","lineNumber":340,"sourceCode":"      TypeAndMaxId typeAndMaxId =\n          beamFieldTypeToIcebergFieldType(beamField.getType(), nestedFieldId);\n      Types.NestedField icebergField =\n          Types.NestedField.of(\n              icebergFieldId++,\n              beamField.getType().getNullable(),\n              beamField.getName(),\n              typeAndMaxId.type);\n\n      fields.add(icebergField);\n      nestedFieldId = typeAndMaxId.maxId + 1;\n    }\n    return new org.apache.iceberg.Schema(fields.toArray(new Types.NestedField[fields.size()]));\n  }\n\n  /** Converts a Beam {@link Row} to an Iceberg {@link Record}. */\n  public static Record beamRowToIcebergRecord(org.apache.iceberg.Schema schema, Row row) {\n    if (row.getSchema().getFieldCount() != schema.columns().size()) {\n      throw new IllegalStateException(\n          String.format(\n              \"Beam Row schema and Iceberg schema have different sizes.%n\\tBeam Row columns: %s%n\\tIceberg schema columns: %s\",\n              row.getSchema().getFieldNames(),\n              schema.columns().stream().map(Types.NestedField::name).collect(Collectors.toList())));\n    }\n    return copyRowIntoRecord(GenericRecord.create(schema), row);\n  }\n\n  private static Record copyRowIntoRecord(Record baseRecord, Row value) {\n    Record rec = baseRecord.copy();\n    for (Types.NestedField f : rec.struct().fields()) {\n      copyFieldIntoRecord(rec, f, value);\n    }\n    return rec;\n  }\n\n  private static void copyFieldIntoRecord(Record rec, Types.NestedField field, Row value) {\n    String name = field.name();","sourceCodeStart":322,"sourceCodeEnd":358,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java#L322-L358","documentation":"IcebergUtils.beamRowToIcebergRecord requires the Beam Row's schema to have exactly as many fields as the target Iceberg schema has columns. On mismatch it throws IllegalStateException showing both column lists. Row order and names must align one-to-one for the positional copy into the Iceberg Record.","triggerScenarios":"Calling beamRowToIcebergRecord (directly or via copyFieldIntoRecord) with a Row whose schema field count differs from the passed org.apache.iceberg.Schema column count, at IcebergUtils.java:340.","commonSituations":"Evolving the Beam PCollection schema (adding/dropping fields) without updating the Iceberg table schema (or vice versa); schema drift between producer and table; reusing a stale schema object cached from a previous table version; updateSchema changes not applied before writing.","solutions":["Align the Beam Row schema with the current Iceberg table schema (same fields, same order).","Fetch the latest table schema via loadTable instead of using a cached/stale Iceberg Schema.","Apply the schema evolution (updateSchema) before writing new fields.","Add a pre-write assertion comparing row.getSchema() with the table schema to fail early with a clear message."],"exampleFix":"// before\nRecord rec = IcebergUtils.beamRowToIcebergRecord(staleSchema, row);\n\n// after\norg.apache.iceberg.Schema current = IcebergUtils.beamSchemaToIcebergSchema(row.getSchema());\nRecord rec = IcebergUtils.beamRowToIcebergRecord(current, row);","handlingStrategy":"validation","validationCode":"if (row.getSchema().getFieldCount() != schema.columns().size()) {\n  throw new IllegalStateException(\"Row/Iceberg schema column count mismatch: \"\n      + row.getSchema().getFieldCount() + \" vs \" + schema.columns().size());\n}","typeGuard":null,"tryCatchPattern":"try {\n  Record rec = IcebergUtils.beamRowToIcebergRecord(schema, row);\n} catch (IllegalStateException e) {\n  LOG.error(\"Schema mismatch: {}\", e.getMessage());\n}","preventionTips":["Derive the Row schema and Iceberg schema from a single source of truth.","Refresh the table schema before each write instead of caching.","Run updateSchema/evolution before writing new fields.","Assert schema equality (count and names) in pipeline tests."],"tags":["java","iceberg","schema-mismatch","row"],"backgroundTag":"shape-mismatch","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"}