From a466c3d061de5bf1828b8c0e897f50d8f5a4e928 Mon Sep 17 00:00:00 2001 From: Sergei Nikolaev Date: Tue, 14 Jul 2026 12:29:06 +0200 Subject: [PATCH] Flink: Fix timestamp-micros conversion in AvroToRowDataConverters Signed-off-by: Sergei Nikolaev --- .../formats/avro/AvroToRowDataConverters.java | 2 +- .../org/apache/iceberg/flink/DataGenerators.java | 15 +++++++++++---- 2 files changed, 12 insertions(+), 5 deletions(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/formats/avro/AvroToRowDataConverters.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/formats/avro/AvroToRowDataConverters.java index 0f70e60a1b9f..350f7742b90b 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/formats/avro/AvroToRowDataConverters.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/formats/avro/AvroToRowDataConverters.java @@ -233,7 +233,7 @@ private static TimestampData convertToTimestamp(Object object, LogicalType type) 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.1/flink/src/test/java/org/apache/iceberg/flink/DataGenerators.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/DataGenerators.java index 795c4fa5a766..72d53bc0385c 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/DataGenerators.java +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/DataGenerators.java @@ -68,6 +68,7 @@ public static class Primitives implements DataGenerator { 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 GenericRowData generateFlinkRowData() { 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 org.apache.avro.generic.GenericRecord generateAvroGenericRecord() { 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);