You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by pv...@apache.org on 2026/07/28 08:25:40 UTC
(iceberg) 02/02: Flink: Backport timestamp-micros conversion fix to 1.20 and 2.0
This is an automated email from the ASF dual-hosted git repository.
pvary pushed a commit to branch 1.11.x
in repository https://gitbox.apache.org/repos/asf/iceberg.git
commit b1c19a28504e8ceee70527828f06cc177d47e34b
Author: Sergei Nikolaev <ki...@gmail.com>
AuthorDate: Wed Jul 15 11:01:50 2026 +0200
Flink: Backport timestamp-micros conversion fix to 1.20 and 2.0
Signed-off-by: Sergei Nikolaev <ki...@gmail.com>
---
.../flink/formats/avro/AvroToRowDataConverters.java | 2 +-
.../java/org/apache/iceberg/flink/DataGenerators.java | 15 +++++++++++----
.../flink/formats/avro/AvroToRowDataConverters.java | 2 +-
.../java/org/apache/iceberg/flink/DataGenerators.java | 15 +++++++++++----
4 files changed, 24 insertions(+), 10 deletions(-)
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/formats/avro/AvroToRowDataConverters.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/formats/avro/AvroToRowDataConverters.java
index 0f70e60a1b..350f7742b9 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/formats/avro/AvroToRowDataConverters.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/formats/avro/AvroToRowDataConverters.java
@@ -233,7 +233,7 @@ public class AvroToRowDataConverters {
return TimestampData.fromEpochMillis(timeLong);
} else if (precision <= 6) {
return TimestampData.fromEpochMillis(
- Math.floorDiv(timeLong, 1000L), (int) Math.floorMod(timeLong, 1000L) * 1_000_000);
+ Math.floorDiv(timeLong, 1000L), (int) Math.floorMod(timeLong, 1000L) * 1000);
} else {
// Iceberg: Added support for nanoseconds precision (FLINK-39251)
return TimestampData.fromEpochMillis(
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/DataGenerators.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/DataGenerators.java
index 795c4fa5a7..72d53bc038 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/DataGenerators.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/DataGenerators.java
@@ -68,6 +68,7 @@ public class DataGenerators {
private static final int DAYS_BTW_EPOC_AND_20220110 =
Days.daysBetween(JODA_DATETIME_EPOC, JODA_DATETIME_20220110).getDays();
private static final int HOUR_8_IN_MILLI = (int) TimeUnit.HOURS.toMillis(8);
+ private static final int MICROS_OF_MILLI_20220110 = 456;
private static final LocalDate JAVA_LOCAL_DATE_20220110 = LocalDate.of(2022, 1, 10);
private static final LocalTime JAVA_LOCAL_TIME_HOUR8 = LocalTime.of(8, 0);
@@ -226,8 +227,10 @@ public class DataGenerators {
HOUR_8_IN_MILLI,
// Although Avro logical type for timestamp fields are in micro seconds,
// AvroToRowDataConverters only looks for long value in milliseconds.
- TimestampData.fromEpochMillis(JODA_DATETIME_20220110.getMillis()),
- TimestampData.fromEpochMillis(JODA_DATETIME_20220110.getMillis()),
+ TimestampData.fromEpochMillis(
+ JODA_DATETIME_20220110.getMillis(), MICROS_OF_MILLI_20220110 * 1000),
+ TimestampData.fromEpochMillis(
+ JODA_DATETIME_20220110.getMillis(), MICROS_OF_MILLI_20220110 * 1000),
uuidBytes,
binaryBytes,
DecimalData.fromBigDecimal(BIG_DECIMAL_NEGATIVE, 9, 2),
@@ -253,8 +256,12 @@ public class DataGenerators {
genericRecord.put("time_field", HOUR_8_IN_MILLI);
// Now that AvroToRowDataConverters correctly supports microseconds,
// we must inject correct microsecond scale values into the Avro data.
- genericRecord.put("ts_with_zone_field", JODA_DATETIME_20220110.getMillis() * 1000L);
- genericRecord.put("ts_without_zone_field", JODA_DATETIME_20220110.getMillis() * 1000L);
+ genericRecord.put(
+ "ts_with_zone_field",
+ JODA_DATETIME_20220110.getMillis() * 1000L + MICROS_OF_MILLI_20220110);
+ genericRecord.put(
+ "ts_without_zone_field",
+ JODA_DATETIME_20220110.getMillis() * 1000L + MICROS_OF_MILLI_20220110);
genericRecord.put("ts_ns_with_zone_field", ICEBERG_MAX_NANOS_EPOCH);
genericRecord.put("ts_ns_without_zone_field", ICEBERG_MAX_NANOS_EPOCH);
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/formats/avro/AvroToRowDataConverters.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/formats/avro/AvroToRowDataConverters.java
index 0f70e60a1b..350f7742b9 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/formats/avro/AvroToRowDataConverters.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/formats/avro/AvroToRowDataConverters.java
@@ -233,7 +233,7 @@ public class AvroToRowDataConverters {
return TimestampData.fromEpochMillis(timeLong);
} else if (precision <= 6) {
return TimestampData.fromEpochMillis(
- Math.floorDiv(timeLong, 1000L), (int) Math.floorMod(timeLong, 1000L) * 1_000_000);
+ Math.floorDiv(timeLong, 1000L), (int) Math.floorMod(timeLong, 1000L) * 1000);
} else {
// Iceberg: Added support for nanoseconds precision (FLINK-39251)
return TimestampData.fromEpochMillis(
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/DataGenerators.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/DataGenerators.java
index 795c4fa5a7..72d53bc038 100644
--- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/DataGenerators.java
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/DataGenerators.java
@@ -68,6 +68,7 @@ public class DataGenerators {
private static final int DAYS_BTW_EPOC_AND_20220110 =
Days.daysBetween(JODA_DATETIME_EPOC, JODA_DATETIME_20220110).getDays();
private static final int HOUR_8_IN_MILLI = (int) TimeUnit.HOURS.toMillis(8);
+ private static final int MICROS_OF_MILLI_20220110 = 456;
private static final LocalDate JAVA_LOCAL_DATE_20220110 = LocalDate.of(2022, 1, 10);
private static final LocalTime JAVA_LOCAL_TIME_HOUR8 = LocalTime.of(8, 0);
@@ -226,8 +227,10 @@ public class DataGenerators {
HOUR_8_IN_MILLI,
// Although Avro logical type for timestamp fields are in micro seconds,
// AvroToRowDataConverters only looks for long value in milliseconds.
- TimestampData.fromEpochMillis(JODA_DATETIME_20220110.getMillis()),
- TimestampData.fromEpochMillis(JODA_DATETIME_20220110.getMillis()),
+ TimestampData.fromEpochMillis(
+ JODA_DATETIME_20220110.getMillis(), MICROS_OF_MILLI_20220110 * 1000),
+ TimestampData.fromEpochMillis(
+ JODA_DATETIME_20220110.getMillis(), MICROS_OF_MILLI_20220110 * 1000),
uuidBytes,
binaryBytes,
DecimalData.fromBigDecimal(BIG_DECIMAL_NEGATIVE, 9, 2),
@@ -253,8 +256,12 @@ public class DataGenerators {
genericRecord.put("time_field", HOUR_8_IN_MILLI);
// Now that AvroToRowDataConverters correctly supports microseconds,
// we must inject correct microsecond scale values into the Avro data.
- genericRecord.put("ts_with_zone_field", JODA_DATETIME_20220110.getMillis() * 1000L);
- genericRecord.put("ts_without_zone_field", JODA_DATETIME_20220110.getMillis() * 1000L);
+ genericRecord.put(
+ "ts_with_zone_field",
+ JODA_DATETIME_20220110.getMillis() * 1000L + MICROS_OF_MILLI_20220110);
+ genericRecord.put(
+ "ts_without_zone_field",
+ JODA_DATETIME_20220110.getMillis() * 1000L + MICROS_OF_MILLI_20220110);
genericRecord.put("ts_ns_with_zone_field", ICEBERG_MAX_NANOS_EPOCH);
genericRecord.put("ts_ns_without_zone_field", ICEBERG_MAX_NANOS_EPOCH);