Flink: Fix long casting issues

Backport of apache#10580 and resolve IDE warning/error wrt `floorMod`/`floorDiv`.
snazy · Jul 4, 2024 · a50f57f · a50f57f
1 parent c7eba34
commit a50f57f
Show file tree

Hide file tree

Showing 5 changed files with 11 additions and 11 deletions.
diff --git a/flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java b/flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java
@@ -422,7 +422,7 @@ public TimestampData read(TimestampData ignored) {
       long value = readLong();
       return TimestampData.fromLocalDateTime(
           Instant.ofEpochSecond(
-                  Math.floorDiv(value, 1000_000), Math.floorMod(value, 1000_000) * 1000)
+                  Math.floorDiv(value, 1000_000L), Math.floorMod(value, 1000_000L) * 1000L)
               .atOffset(ZoneOffset.UTC)
               .toLocalDateTime());
     }
@@ -444,7 +444,7 @@ public TimestampData read(TimestampData ignored) {
       long value = readLong();
       return TimestampData.fromInstant(
           Instant.ofEpochSecond(
-              Math.floorDiv(value, 1000_000), Math.floorMod(value, 1000_000) * 1000));
+              Math.floorDiv(value, 1000_000L), Math.floorMod(value, 1000_000L) * 1000L));
     }
 
     @Override
@@ -517,7 +517,7 @@ private static class LossyMicrosToMillisTimeReader
     @Override
     public Integer read(Integer reuse) {
       // Discard microseconds since Flink uses millisecond unit for TIME type.
-      return (int) Math.floorDiv(column.nextLong(), 1000);
+      return (int) Math.floorDiv(column.nextLong(), 1000L);
     }
   }
 

diff --git a/...c/test/java/org/apache/iceberg/flink/source/assigner/TestWatermarkBasedSplitAssigner.java b/...c/test/java/org/apache/iceberg/flink/source/assigner/TestWatermarkBasedSplitAssigner.java
@@ -123,7 +123,7 @@ protected List<IcebergSourceSplit> createSplits(
                         .mapToObj(
                             fileNum ->
                                 RandomGenericData.generate(
-                                    SCHEMA, 2, splitNum * filesPerSplit + fileNum))
+                                    SCHEMA, 2, (long) splitNum * filesPerSplit + fileNum))
                         .collect(Collectors.toList())))
         .collect(Collectors.toList());
   }

diff --git a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java
@@ -422,7 +422,7 @@ public TimestampData read(TimestampData ignored) {
       long value = readLong();
       return TimestampData.fromLocalDateTime(
           Instant.ofEpochSecond(
-                  Math.floorDiv(value, 1000_000), Math.floorMod(value, 1000_000) * 1000)
+                  Math.floorDiv(value, 1000_000L), Math.floorMod(value, 1000_000L) * 1000L)
               .atOffset(ZoneOffset.UTC)
               .toLocalDateTime());
     }
@@ -444,7 +444,7 @@ public TimestampData read(TimestampData ignored) {
       long value = readLong();
       return TimestampData.fromInstant(
           Instant.ofEpochSecond(
-              Math.floorDiv(value, 1000_000), Math.floorMod(value, 1000_000) * 1000));
+              Math.floorDiv(value, 1000_000L), Math.floorMod(value, 1000_000L) * 1000L));
     }
 
     @Override
@@ -517,7 +517,7 @@ private static class LossyMicrosToMillisTimeReader
     @Override
     public Integer read(Integer reuse) {
       // Discard microseconds since Flink uses millisecond unit for TIME type.
-      return (int) Math.floorDiv(column.nextLong(), 1000);
+      return (int) Math.floorDiv(column.nextLong(), 1000L);
     }
   }
 

diff --git a/...c/test/java/org/apache/iceberg/flink/source/assigner/TestWatermarkBasedSplitAssigner.java b/...c/test/java/org/apache/iceberg/flink/source/assigner/TestWatermarkBasedSplitAssigner.java
@@ -123,7 +123,7 @@ protected List<IcebergSourceSplit> createSplits(
                         .mapToObj(
                             fileNum ->
                                 RandomGenericData.generate(
-                                    SCHEMA, 2, splitNum * filesPerSplit + fileNum))
+                                    SCHEMA, 2, (long) splitNum * filesPerSplit + fileNum))
                         .collect(Collectors.toList())))
         .collect(Collectors.toList());
   }

diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java
@@ -422,7 +422,7 @@ public TimestampData read(TimestampData ignored) {
       long value = readLong();
       return TimestampData.fromLocalDateTime(
           Instant.ofEpochSecond(
-                  Math.floorDiv(value, 1000_000), Math.floorMod(value, 1000_000) * 1000L)
+                  Math.floorDiv(value, 1000_000L), Math.floorMod(value, 1000_000L) * 1000L)
               .atOffset(ZoneOffset.UTC)
               .toLocalDateTime());
     }
@@ -444,7 +444,7 @@ public TimestampData read(TimestampData ignored) {
       long value = readLong();
       return TimestampData.fromInstant(
           Instant.ofEpochSecond(
-              Math.floorDiv(value, 1000_000), Math.floorMod(value, 1000_000) * 1000L));
+              Math.floorDiv(value, 1000_000L), Math.floorMod(value, 1000_000L) * 1000L));
     }
 
     @Override
@@ -517,7 +517,7 @@ private static class LossyMicrosToMillisTimeReader
     @Override
     public Integer read(Integer reuse) {
       // Discard microseconds since Flink uses millisecond unit for TIME type.
-      return (int) Math.floorDiv(column.nextLong(), 1000);
+      return (int) Math.floorDiv(column.nextLong(), 1000L);
     }
   }