address nit

apache · Oct 10, 2024 · 86e8271 · 86e8271
1 parent 969263e
commit 86e8271
Showing 1 changed file with 19 additions and 16 deletions.
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java
@@ -433,24 +433,10 @@ public static Row icebergRecordToBeamRow(Schema schema, Record record) {
           rowBuilder.addValue(icebergValue);
           break;
         case DATETIME:
-          long micros;
-          if (icebergValue instanceof OffsetDateTime) {
-            micros = DateTimeUtil.microsFromTimestamptz((OffsetDateTime) icebergValue);
-          } else if (icebergValue instanceof LocalDateTime) {
-            micros = DateTimeUtil.microsFromTimestamp((LocalDateTime) icebergValue);
-          } else if (icebergValue instanceof Long) {
-            micros = (long) icebergValue;
-          } else if (icebergValue instanceof String) {
-            rowBuilder.addValue(DateTime.parse((String) icebergValue));
-            break;
-          } else {
-            throw new UnsupportedOperationException(
-                "Unsupported Iceberg type for Beam type DATETIME: " + icebergValue.getClass());
-          }
-          // Iceberg uses a long for micros
+          // Iceberg uses a long for micros.
           // Beam DATETIME uses joda's DateTime, which only supports millis,
           // so we do lose some precision here
-          rowBuilder.addValue(new DateTime(micros / 1000L));
+          rowBuilder.addValue(getBeamDateTimeValue(icebergValue));
           break;
         case BYTES:
           // Iceberg uses ByteBuffer; Beam uses byte[]
@@ -475,6 +461,23 @@ public static Row icebergRecordToBeamRow(Schema schema, Record record) {
     return rowBuilder.build();
   }
 
+  private static DateTime getBeamDateTimeValue(Object icebergValue) {
+    long micros;
+    if (icebergValue instanceof OffsetDateTime) {
+      micros = DateTimeUtil.microsFromTimestamptz((OffsetDateTime) icebergValue);
+    } else if (icebergValue instanceof LocalDateTime) {
+      micros = DateTimeUtil.microsFromTimestamp((LocalDateTime) icebergValue);
+    } else if (icebergValue instanceof Long) {
+      micros = (long) icebergValue;
+    } else if (icebergValue instanceof String) {
+      return DateTime.parse((String) icebergValue);
+    } else {
+      throw new UnsupportedOperationException(
+          "Unsupported Iceberg type for Beam type DATETIME: " + icebergValue.getClass());
+    }
+    return new DateTime(micros / 1000L);
+  }
+
   private static Object getLogicalTypeValue(Object icebergValue, Schema.FieldType type) {
     if (icebergValue instanceof String) {
       String strValue = (String) icebergValue;