apache/beam · error · IllegalStateException

Beam Row schema and Iceberg schema have different…

Error message

Beam Row schema and Iceberg schema have different sizes.
	Beam Row columns: {beamColumns}
	Iceberg schema columns: {icebergColumns}

What it means

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.

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.

Example fix

// before
Record rec = IcebergUtils.beamRowToIcebergRecord(staleSchema, row);

// after
org.apache.iceberg.Schema current = IcebergUtils.beamSchemaToIcebergSchema(row.getSchema());
Record rec = IcebergUtils.beamRowToIcebergRecord(current, row);

Defensive patterns

Strategy: validation

Validate before calling

if (row.getSchema().getFieldCount() != schema.columns().size()) {
  throw new IllegalStateException("Row/Iceberg schema column count mismatch: "
      + row.getSchema().getFieldCount() + " vs " + schema.columns().size());
}

Try / catch

try {
  Record rec = IcebergUtils.beamRowToIcebergRecord(schema, row);
} catch (IllegalStateException e) {
  LOG.error("Schema mismatch: {}", e.getMessage());
}

Prevention

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.

When it happens

Trigger: 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.

Common situations: 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.

Related errors

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

Appendix: source

Thrown at sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java:340

      TypeAndMaxId typeAndMaxId =
          beamFieldTypeToIcebergFieldType(beamField.getType(), nestedFieldId);
      Types.NestedField icebergField =
          Types.NestedField.of(
              icebergFieldId++,
              beamField.getType().getNullable(),
              beamField.getName(),
              typeAndMaxId.type);

      fields.add(icebergField);
      nestedFieldId = typeAndMaxId.maxId + 1;
    }
    return new org.apache.iceberg.Schema(fields.toArray(new Types.NestedField[fields.size()]));
  }

  /** Converts a Beam {@link Row} to an Iceberg {@link Record}. */
  public static Record beamRowToIcebergRecord(org.apache.iceberg.Schema schema, Row row) {
    if (row.getSchema().getFieldCount() != schema.columns().size()) {
      throw new IllegalStateException(
          String.format(
              "Beam Row schema and Iceberg schema have different sizes.%n\tBeam Row columns: %s%n\tIceberg schema columns: %s",
              row.getSchema().getFieldNames(),
              schema.columns().stream().map(Types.NestedField::name).collect(Collectors.toList())));
    }
    return copyRowIntoRecord(GenericRecord.create(schema), row);
  }

  private static Record copyRowIntoRecord(Record baseRecord, Row value) {
    Record rec = baseRecord.copy();
    for (Types.NestedField f : rec.struct().fields()) {
      copyFieldIntoRecord(rec, f, value);
    }
    return rec;
  }

  private static void copyFieldIntoRecord(Record rec, Types.NestedField field, Row value) {
    String name = field.name();

View on GitHub (pinned to 12126d8942)