Fix IcebergIO failing to write EnumerationType (proto enum) fields

CHANGES.md

@@ -84,6 +84,7 @@
 * (Java) Fixed the declared schema of the error output of the Kafka write SchemaTransform, which wrapped the error schema a second time and did not match the rows it emits ([#39760](https://github.com/apache/beam/issues/39760)).
 * (Go) Fixed pubsubio importing a `google.golang.org/genproto` package removed in recent releases, which broke builds of Go modules depending on a current `genproto` version ([#40018](https://github.com/apache/beam/issues/40018)).
 * (Java) BigQueryIO now treats a 404 when deleting a temporary table or dataset as success, so a replayed work item whose earlier attempt already deleted it no longer retries forever ([#24997](https://github.com/apache/beam/issues/24997)).
+* (Java) IcebergIO now writes rows containing `EnumerationType` (proto enum) fields as strings, instead of throwing `Unsupported Beam logical type Enum` ([#40299](https://github.com/apache/beam/issues/40299)).
 * Fixed X (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 
 ## Security Fixes

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java

@@ -36,6 +36,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.logicaltypes.EnumerationType;
 import org.apache.beam.sdk.schemas.logicaltypes.FixedPrecisionNumeric;
 import org.apache.beam.sdk.schemas.logicaltypes.MicrosInstant;
 import org.apache.beam.sdk.schemas.logicaltypes.PassThroughLogicalType;
@@ -84,6 +85,7 @@ private IcebergUtils() {}
           .put(SqlTypes.DATETIME.getIdentifier(), Types.TimestampType.withoutZone())
           .put(SqlTypes.UUID.getIdentifier(), Types.UUIDType.get())
           .put(MicrosInstant.IDENTIFIER, Types.TimestampType.withZone())
+          .put(EnumerationType.IDENTIFIER, Types.StringType.get())
           .build();
 
   private static Schema.FieldType icebergTypeToBeamFieldType(
@@ -389,7 +391,14 @@ private static void copyFieldIntoRecord(Record rec, Types.NestedField field, Row
         rec.setField(name, getIcebergTimestampValue(val, ts.shouldAdjustToUTC()));
         break;
       case STRING:
-        Optional.ofNullable(value.getString(name)).ifPresent(v -> rec.setField(name, v));
+        Object beamValue = value.getValue(name);
+        if (beamValue instanceof EnumerationType.Value) { // EnumerationType
+          EnumerationType enumType =
+              value.getSchema().getField(name).getType().getLogicalType(EnumerationType.class);
+          rec.setField(name, enumType.toString((EnumerationType.Value) beamValue));
+        } else { // FieldType.STRING, or a String-backed PassThroughLogicalType
+          Optional.ofNullable((String) beamValue).ifPresent(v -> rec.setField(name, v));
+        }
         break;
       case UUID:
         Optional.ofNullable(value.getBytes(name))

sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergUtilsTest.java

@@ -39,6 +39,7 @@
 import java.util.List;
 import java.util.Map;
 import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.logicaltypes.EnumerationType;
 import org.apache.beam.sdk.schemas.logicaltypes.FixedPrecisionNumeric;
 import org.apache.beam.sdk.schemas.logicaltypes.FixedString;
 import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes;
@@ -289,6 +290,17 @@ public void testDecimal() {
       checkRowValueToRecordValue(Schema.FieldType.DECIMAL, Types.DecimalType.of(6, 3), num);
     }
 
+    @Test
+    public void testEnumeration() {
+      EnumerationType enumType = EnumerationType.create("ONE", "TWO", "THREE");
+
+      checkRowValueToRecordValue(
+          Schema.FieldType.logicalType(enumType),
+          enumType.valueOf("TWO"),
+          Types.StringType.get(),
+          "TWO");
+    }
+
     @Test
     public void testStruct() {
       Schema schema = Schema.builder().addStringField("nested_str").build();
@@ -1151,5 +1163,31 @@ public void testJdbcBeamSchemaToIcebergSchema() {
 
       assertTrue(convertedIcebergSchema.sameSchema(ICEBERG_SCHEMA_JDBC_ALL_TYPES));
     }
+
+    static final EnumerationType TEST_ENUMERATION_TYPE =
+        EnumerationType.create("ONE", "TWO", "THREE");
+
+    static final Schema BEAM_SCHEMA_ENUMERATION =
+        Schema.builder()
+            .addLogicalTypeField("status", TEST_ENUMERATION_TYPE)
+            .addNullableField(
+                "optional_status", Schema.FieldType.logicalType(TEST_ENUMERATION_TYPE))
+            .build();
+
+    static final org.apache.iceberg.Schema ICEBERG_SCHEMA_ENUMERATION =
+        new org.apache.iceberg.Schema(
+            required(1, "status", Types.StringType.get()),
+            optional(2, "optional_status", Types.StringType.get()));
+
+    @Test
+    public void testEnumerationBeamSchemaToIcebergSchema() {
+      // Iceberg has no enum type, so EnumerationType fields are stored as strings. This is a
+      // one-way conversion: converting ICEBERG_SCHEMA_ENUMERATION back with
+      // icebergSchemaToBeamSchema will not recover EnumerationType, only plain STRING fields.
+      org.apache.iceberg.Schema convertedIcebergSchema =
+          IcebergUtils.beamSchemaToIcebergSchema(BEAM_SCHEMA_ENUMERATION);
+
+      assertTrue(convertedIcebergSchema.sameSchema(ICEBERG_SCHEMA_ENUMERATION));
+    }
   }
 }