[IcebergIO] Support sorted writes and RANGE distribution mode

CHANGES.md

@@ -67,6 +67,7 @@
 * Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 * (Java) FirestoreIO: added `RpcQosOptions.Builder.withRampupThrottlingDisabled()`, mirroring the existing DatastoreIO option, to opt out of the 500/50/5 write ramp-up ([#27170](https://github.com/apache/beam/issues/27170)).
 * Added support for attributing BigQuery API quota and billing to a specific GCP project (quota project): `quota_project_id` parameter in `ReadFromBigQuery` or `--quota_project_id` pipeline option, honored by the Python BigQuery IO transforms (`WriteToBigQuery` with `FILE_LOADS` or `STREAMING_INSERTS`; not yet `STORAGE_WRITE_API`, which is Java-backed), `--bigQueryQuotaProjectId` pipeline option (Java), and `bigqueryio.WithQuotaProject` read/query option (Go) ([#37431](https://github.com/apache/beam/issues/37431)).
+* [IcebergIO] Added opt-in sorted write support (`withSortedWrites()` / `sort_on_write`) across `NONE`, `HASH`, and `RANGE` distribution modes to write data files in the table's sort order, spilling to local disk via Beam's external sorter when groups exceed memory. Also added `DistributionMode.RANGE`, which dynamically computes non-overlapping quantile key ranges per partition (or accepts a custom `withDistributionFunction(...)`) so each shard writes sorted, non-overlapping file(s) (Java) ([#38406](https://github.com/apache/beam/issues/38406)).
 
 ## New Features / Improvements
 

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

@@ -32,9 +32,11 @@
 import org.apache.beam.sdk.schemas.Schema.FieldType;
 import org.apache.beam.sdk.schemas.logicaltypes.Date;
 import org.apache.beam.sdk.schemas.logicaltypes.DateTime;
+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.Time;
+import org.apache.beam.sdk.schemas.logicaltypes.Timestamp;
 import org.apache.beam.sdk.values.Row;
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.types.Type;
@@ -158,7 +160,8 @@ private interface PositionalGetter<T> {
         BeamRowWrapper nestedWrapper = new BeamRowWrapper(beamSchema, structType);
         return (row, pos) -> nestedWrapper.wrap(row.getRow(pos));
       case LOGICAL_TYPE:
-        if (beamType.isLogicalType(MicrosInstant.IDENTIFIER)) {
+        if (beamType.isLogicalType(MicrosInstant.IDENTIFIER)
+            || beamType.isLogicalType(Timestamp.IDENTIFIER)) {
           return (row, pos) -> {
             Instant instant = checkStateNotNull(row.getLogicalTypeValue(pos, Instant.class));
             return TimeUnit.SECONDS.toMicros(instant.getEpochSecond()) + instant.getNano() / 1000;
@@ -177,6 +180,11 @@ private interface PositionalGetter<T> {
                   checkStateNotNull(row.getLogicalTypeValue(pos, LocalTime.class)));
         } else if (beamType.isLogicalType(FixedPrecisionNumeric.IDENTIFIER)) {
           return (row, pos) -> row.getLogicalTypeValue(pos, BigDecimal.class);
+        } else if (beamType.isLogicalType(EnumerationType.IDENTIFIER)) {
+          EnumerationType enumType = beamType.getLogicalType(EnumerationType.class);
+          return (row, pos) ->
+              enumType.toString(
+                  checkStateNotNull(row.getLogicalTypeValue(pos, EnumerationType.Value.class)));
         } else {
           return null;
         }