[Bug]: IcebergIO cannot write Beam rows containing EnumerationType (proto enum) fields #40299

What happened?

Writing a Row with an EnumerationType logical type field (e.g. from ProtoSchemaTranslator on a proto message with an enum field) to IcebergIO.writeRows() fails. EnumerationType isn't one of the cases handled here, so schema derivation throws immediately:

java.lang.RuntimeException: Unsupported Beam logical type Enum
      at org.apache.beam.sdk.io.iceberg.IcebergUtils.beamFieldTypeToIcebergFieldType(IcebergUtils.java:236)
      at org.apache.beam.sdk.io.iceberg.IcebergUtils.beamSchemaToIcebergSchema(IcebergUtils.java:323)

Pre-creating the table doesn't help either. copyFieldIntoRecord calls Row.getString()/Row.getInt32() directly, so it fails trying to cast the field's EnumerationType.Value to String:

java.lang.ClassCastException: class org.apache.beam.sdk.schemas.logicaltypes.EnumerationType$Value cannot be cast to class java.lang.String
      at org.apache.beam.sdk.io.iceberg.IcebergUtils.copyFieldIntoRecord(IcebergUtils.java:392)

Avro (AvroUtils.genericFromBeamField) and BigQuery (BigQueryUtils.fromBeamField) already convert EnumerationType to a string on write, so there's a working pattern to follow here.

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Prism Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner