(fluss) branch main updated: [lake/iceberg] Fix integer overflow when converting Iceberg LocalTime to Fluss millis-of-day (#3535)

[yuxia]

This is an automated email from the ASF dual-hosted git repository.

luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git


The following commit(s) were added to refs/heads/main by this push:
     new 2407fe4e1 [lake/iceberg] Fix integer overflow when converting Iceberg 
LocalTime to Fluss millis-of-day (#3535)
2407fe4e1 is described below

commit 2407fe4e1f36dcea9a8b46604e869583b10b23e7
Author: GuoYu <511955...@qq.com>
AuthorDate: Mon Jun 29 13:51:40 2026 +0800

    [lake/iceberg] Fix integer overflow when converting Iceberg LocalTime to 
Fluss millis-of-day (#3535)
---
 .../iceberg/source/IcebergArrayAsFlussArray.java   |  2 +-
 .../iceberg/source/IcebergRecordAsFlussRow.java    |  2 +-
 .../source/IcebergRecordAsFlussRowTest.java        | 31 ++++++++++++++++++++--
 3 files changed, 31 insertions(+), 4 deletions(-)

diff --git 
a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergArrayAsFlussArray.java
 
b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergArrayAsFlussArray.java
index 1b83e8f2b..4686d0d10 100644
--- 
a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergArrayAsFlussArray.java
+++ 
b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergArrayAsFlussArray.java
@@ -79,7 +79,7 @@ public class IcebergArrayAsFlussArray implements 
InternalArray {
             return (int) ((LocalDate) value).toEpochDay();
         }
         if (value instanceof LocalTime) {
-            return (int) ((LocalTime) value).toNanoOfDay() / 1_000_000;
+            return (int) (((LocalTime) value).toNanoOfDay() / 1_000_000);
         }
         return (Integer) value;
     }
diff --git 
a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergRecordAsFlussRow.java
 
b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergRecordAsFlussRow.java
index be990f7c6..808e901d9 100644
--- 
a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergRecordAsFlussRow.java
+++ 
b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergRecordAsFlussRow.java
@@ -94,7 +94,7 @@ public class IcebergRecordAsFlussRow implements InternalRow {
             return (int) ((LocalDate) value).toEpochDay();
         }
         if (value instanceof LocalTime) {
-            return (int) ((LocalTime) value).toNanoOfDay() / 1_000_000;
+            return (int) (((LocalTime) value).toNanoOfDay() / 1_000_000);
         }
         return (Integer) value;
     }
diff --git 
a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/IcebergRecordAsFlussRowTest.java
 
b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/IcebergRecordAsFlussRowTest.java
index ec0cb8c80..e5e142dfb 100644
--- 
a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/IcebergRecordAsFlussRowTest.java
+++ 
b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/IcebergRecordAsFlussRowTest.java
@@ -37,6 +37,7 @@ import java.time.OffsetDateTime;
 import java.time.ZoneOffset;
 import java.util.Arrays;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 import static org.apache.iceberg.types.Types.NestedField.optional;
@@ -203,12 +204,38 @@ class IcebergRecordAsFlussRowTest {
 
         assertThat(icebergRecordAsFlussRow.getInt(15)).isEqualTo((int) 
date.toEpochDay());
         assertThat(icebergRecordAsFlussRow.getInt(16))
-                .isEqualTo((int) time.toNanoOfDay() / 1_000_000);
+                .isEqualTo((int) (time.toNanoOfDay() / 1_000_000));
 
         InternalArray dateArray = icebergRecordAsFlussRow.getArray(17);
         InternalArray timeArray = icebergRecordAsFlussRow.getArray(18);
         assertThat(dateArray.getInt(0)).isEqualTo((int) date.toEpochDay());
-        assertThat(timeArray.getInt(0)).isEqualTo((int) time.toNanoOfDay() / 
1_000_000);
+        assertThat(timeArray.getInt(0)).isEqualTo((int) (time.toNanoOfDay() / 
1_000_000));
+    }
+
+    @Test
+    void testGetIntWithLocalTimeNearEndOfDay() {
+        // 23:59:59.999 -> 86_399_999 ms
+        LocalTime endOfDay = LocalTime.of(23, 59, 59, 999_000_000);
+        long expectedNanos = endOfDay.toNanoOfDay();
+        assertThat(expectedNanos).isGreaterThan(Integer.MAX_VALUE);
+        int expectedMillisOfDay = (int) (expectedNanos / 1_000_000L);
+        assertThat(expectedMillisOfDay).isEqualTo(86_399_999);
+
+        record.setField("id", 1L);
+        record.setField("date_field", LocalDate.of(2020, 6, 15));
+        record.setField("time_field", endOfDay);
+        record.setField("date_array", List.of(LocalDate.of(2020, 6, 15)));
+        record.setField("time_array", List.of(endOfDay));
+        record.setField("__bucket", 1);
+        record.setField("__offset", 100L);
+        record.setField("__timestamp", OffsetDateTime.now(ZoneOffset.UTC));
+
+        icebergRecordAsFlussRow.replaceIcebergRecord(record);
+
+        
assertThat(icebergRecordAsFlussRow.getInt(16)).isEqualTo(expectedMillisOfDay);
+
+        InternalArray timeArray = icebergRecordAsFlussRow.getArray(18);
+        assertThat(timeArray.getInt(0)).isEqualTo(expectedMillisOfDay);
     }
 
     @Test