apache/beam · error · RuntimeException

Unsupported Beam logical type

Error message

Unsupported Beam logical type {logicalTypeIdentifier}

What it means

When converting a Beam field type to Iceberg, if the type is a logical type that is not a recognized Timestamp and no direct Iceberg mapping was found, beamFieldTypeToIcebergFieldType throws RuntimeException naming the logical type identifier. It marks Beam logical types the Iceberg writer cannot express.

Solutions

Convert the field to a base Beam type (string, long, double, bytes) before writing to Iceberg.
Implement/serialize the logical type to a primitive representation in a preprocessing step.
Remove or replace the custom logical type field in the schema handed to the Iceberg sink.
If the type should be supported, upgrade Beam or file/patch IcebergUtils to add a mapping.

Example fix

// before
Schema.Field.of("amount", Schema.FieldType.logicalType(new MoneyLogicalType()));

// after
Schema.Field.of("amount", Schema.FieldType.DECIMAL);

Defensive patterns

Strategy: validation

Validate before calling

for (Schema.Field f : beamSchema.getFields()) {
  if (f.getType().getTypeName().isLogicalType()
      && !f.getType().isLogicalType(Timestamp.IDENTIFIER)) {
    throw new IllegalStateException("Unsupported logical type on field: " + f.getName());
  }
}

Try / catch

try {
  IcebergUtils.beamRowToIcebergRecord(schema, row);
} catch (RuntimeException e) {
  LOG.error("Unsupported Beam logical type: {}", e.getMessage());
}

Prevention

Avoid custom logical types in schemas destined for Iceberg sinks.
Serialize complex domain types to primitives upstream.
Restrict sink schemas to primitive + micros Timestamp fields.
Test schema-to-Iceberg conversion in CI.

When it happens

Trigger: Writing a Beam schema containing a custom or unsupported logical type (neither a primitive with a direct mapping nor Timestamp.MICROS) at IcebergUtils.java:236; also reached recursively via list/key/value element types.

Common situations: Custom LogicalType implementations in user pipelines; exotic logical types from third-party Beam IOs feeding an Iceberg sink; schema fields like enums or custom money/duration logical types.

Related errors

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

Appendix: source

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

        Integer scale = Preconditions.checkArgumentNotNull(args.getInt32("scale"));
        return new TypeAndMaxId(--nestedFieldId, Types.DecimalType.of(precision, scale));
      }
      if (logicalType instanceof PassThroughLogicalType) {
        return beamFieldTypeToIcebergFieldType(logicalType.getBaseType(), nestedFieldId);
      }
      String logicalTypeIdentifier = logicalType.getIdentifier();
      @Nullable Type type = BEAM_LOGICAL_TYPES_TO_ICEBERG_TYPES.get(logicalTypeIdentifier);
      if (type == null) {
        if (beamType.isLogicalType(Timestamp.IDENTIFIER)) {
          int precision = checkStateNotNull(logicalType.getArgument());
          if (precision == Timestamp.MICROS.getArgument()) {
            type = Types.TimestampType.withZone();
          } else {
            throw new UnsupportedOperationException(
                "Unsupported Timestamp precision: " + precision);
          }
        } else {
          throw new RuntimeException("Unsupported Beam logical type " + logicalTypeIdentifier);
        }
      }
      return new TypeAndMaxId(--nestedFieldId, type);
    } else if (beamType.getTypeName().isCollectionType()) { // ARRAY or ITERABLE
      Schema.FieldType beamCollectionType =
          Preconditions.checkArgumentNotNull(beamType.getCollectionElementType());

      // nestedFieldId is reserved for the list's collection type.
      // we increment here because further nested fields should use unique ID's
      TypeAndMaxId listInfo =
          beamFieldTypeToIcebergFieldType(beamCollectionType, nestedFieldId + 1);
      Type icebergCollectionType = listInfo.type;

      boolean elementTypeIsNullable =
          Preconditions.checkArgumentNotNull(beamType.getCollectionElementType()).getNullable();

      Type listType =
          elementTypeIsNullable

View on GitHub (pinned to 12126d8942)