Skip to content

Commit 054ce4f

Browse files
kinolaevkinolaev
committed
Flink: Fix timestamp-micros conversion in AvroToRowDataConverters
Signed-off-by: Sergei Nikolaev <kinolaev@gmail.com>
1 parent 7a882c2 commit 054ce4f

6 files changed

Lines changed: 36 additions & 15 deletions

File tree

‎flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/formats/avro/AvroToRowDataConverters.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -233,7 +233,7 @@ private static TimestampData convertToTimestamp(Object object, LogicalType type)
233233
return TimestampData.fromEpochMillis(timeLong);
234234
} else if (precision <= 6) {
235235
return TimestampData.fromEpochMillis(
236-
Math.floorDiv(timeLong, 1000L), (int) Math.floorMod(timeLong, 1000L) * 1_000_000);
236+
Math.floorDiv(timeLong, 1000L), (int) Math.floorMod(timeLong, 1000L) * 1000);
237237
} else {
238238
// Iceberg: Added support for nanoseconds precision (FLINK-39251)
239239
return TimestampData.fromEpochMillis(

‎flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/DataGenerators.java‎

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,7 @@ public static class Primitives implements DataGenerator {
6868
private static final int DAYS_BTW_EPOC_AND_20220110 =
6969
Days.daysBetween(JODA_DATETIME_EPOC, JODA_DATETIME_20220110).getDays();
7070
private static final int HOUR_8_IN_MILLI = (int) TimeUnit.HOURS.toMillis(8);
71+
private static final int MICROS_OF_MILLI_20220110 = 456;
7172

7273
private static final LocalDate JAVA_LOCAL_DATE_20220110 = LocalDate.of(2022, 1, 10);
7374
private static final LocalTime JAVA_LOCAL_TIME_HOUR8 = LocalTime.of(8, 0);
@@ -226,8 +227,10 @@ public GenericRowData generateFlinkRowData() {
226227
HOUR_8_IN_MILLI,
227228
// Although Avro logical type for timestamp fields are in micro seconds,
228229
// AvroToRowDataConverters only looks for long value in milliseconds.
229-
TimestampData.fromEpochMillis(JODA_DATETIME_20220110.getMillis()),
230-
TimestampData.fromEpochMillis(JODA_DATETIME_20220110.getMillis()),
230+
TimestampData.fromEpochMillis(
231+
JODA_DATETIME_20220110.getMillis(), MICROS_OF_MILLI_20220110 * 1000),
232+
TimestampData.fromEpochMillis(
233+
JODA_DATETIME_20220110.getMillis(), MICROS_OF_MILLI_20220110 * 1000),
231234
uuidBytes,
232235
binaryBytes,
233236
DecimalData.fromBigDecimal(BIG_DECIMAL_NEGATIVE, 9, 2),
@@ -253,8 +256,12 @@ public org.apache.avro.generic.GenericRecord generateAvroGenericRecord() {
253256
genericRecord.put("time_field", HOUR_8_IN_MILLI);
254257
// Now that AvroToRowDataConverters correctly supports microseconds,
255258
// we must inject correct microsecond scale values into the Avro data.
256-
genericRecord.put("ts_with_zone_field", JODA_DATETIME_20220110.getMillis() * 1000L);
257-
genericRecord.put("ts_without_zone_field", JODA_DATETIME_20220110.getMillis() * 1000L);
259+
genericRecord.put(
260+
"ts_with_zone_field",
261+
JODA_DATETIME_20220110.getMillis() * 1000L + MICROS_OF_MILLI_20220110);
262+
genericRecord.put(
263+
"ts_without_zone_field",
264+
JODA_DATETIME_20220110.getMillis() * 1000L + MICROS_OF_MILLI_20220110);
258265
genericRecord.put("ts_ns_with_zone_field", ICEBERG_MAX_NANOS_EPOCH);
259266
genericRecord.put("ts_ns_without_zone_field", ICEBERG_MAX_NANOS_EPOCH);
260267

‎flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/formats/avro/AvroToRowDataConverters.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -233,7 +233,7 @@ private static TimestampData convertToTimestamp(Object object, LogicalType type)
233233
return TimestampData.fromEpochMillis(timeLong);
234234
} else if (precision <= 6) {
235235
return TimestampData.fromEpochMillis(
236-
Math.floorDiv(timeLong, 1000L), (int) Math.floorMod(timeLong, 1000L) * 1_000_000);
236+
Math.floorDiv(timeLong, 1000L), (int) Math.floorMod(timeLong, 1000L) * 1000);
237237
} else {
238238
// Iceberg: Added support for nanoseconds precision (FLINK-39251)
239239
return TimestampData.fromEpochMillis(

‎flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/DataGenerators.java‎

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,7 @@ public static class Primitives implements DataGenerator {
6868
private static final int DAYS_BTW_EPOC_AND_20220110 =
6969
Days.daysBetween(JODA_DATETIME_EPOC, JODA_DATETIME_20220110).getDays();
7070
private static final int HOUR_8_IN_MILLI = (int) TimeUnit.HOURS.toMillis(8);
71+
private static final int MICROS_OF_MILLI_20220110 = 456;
7172

7273
private static final LocalDate JAVA_LOCAL_DATE_20220110 = LocalDate.of(2022, 1, 10);
7374
private static final LocalTime JAVA_LOCAL_TIME_HOUR8 = LocalTime.of(8, 0);
@@ -226,8 +227,10 @@ public GenericRowData generateFlinkRowData() {
226227
HOUR_8_IN_MILLI,
227228
// Although Avro logical type for timestamp fields are in micro seconds,
228229
// AvroToRowDataConverters only looks for long value in milliseconds.
229-
TimestampData.fromEpochMillis(JODA_DATETIME_20220110.getMillis()),
230-
TimestampData.fromEpochMillis(JODA_DATETIME_20220110.getMillis()),
230+
TimestampData.fromEpochMillis(
231+
JODA_DATETIME_20220110.getMillis(), MICROS_OF_MILLI_20220110 * 1000),
232+
TimestampData.fromEpochMillis(
233+
JODA_DATETIME_20220110.getMillis(), MICROS_OF_MILLI_20220110 * 1000),
231234
uuidBytes,
232235
binaryBytes,
233236
DecimalData.fromBigDecimal(BIG_DECIMAL_NEGATIVE, 9, 2),
@@ -253,8 +256,12 @@ public org.apache.avro.generic.GenericRecord generateAvroGenericRecord() {
253256
genericRecord.put("time_field", HOUR_8_IN_MILLI);
254257
// Now that AvroToRowDataConverters correctly supports microseconds,
255258
// we must inject correct microsecond scale values into the Avro data.
256-
genericRecord.put("ts_with_zone_field", JODA_DATETIME_20220110.getMillis() * 1000L);
257-
genericRecord.put("ts_without_zone_field", JODA_DATETIME_20220110.getMillis() * 1000L);
259+
genericRecord.put(
260+
"ts_with_zone_field",
261+
JODA_DATETIME_20220110.getMillis() * 1000L + MICROS_OF_MILLI_20220110);
262+
genericRecord.put(
263+
"ts_without_zone_field",
264+
JODA_DATETIME_20220110.getMillis() * 1000L + MICROS_OF_MILLI_20220110);
258265
genericRecord.put("ts_ns_with_zone_field", ICEBERG_MAX_NANOS_EPOCH);
259266
genericRecord.put("ts_ns_without_zone_field", ICEBERG_MAX_NANOS_EPOCH);
260267

‎flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/formats/avro/AvroToRowDataConverters.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -233,7 +233,7 @@ private static TimestampData convertToTimestamp(Object object, LogicalType type)
233233
return TimestampData.fromEpochMillis(timeLong);
234234
} else if (precision <= 6) {
235235
return TimestampData.fromEpochMillis(
236-
Math.floorDiv(timeLong, 1000L), (int) Math.floorMod(timeLong, 1000L) * 1_000_000);
236+
Math.floorDiv(timeLong, 1000L), (int) Math.floorMod(timeLong, 1000L) * 1000);
237237
} else {
238238
// Iceberg: Added support for nanoseconds precision (FLINK-39251)
239239
return TimestampData.fromEpochMillis(

‎flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/DataGenerators.java‎

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,7 @@ public static class Primitives implements DataGenerator {
6868
private static final int DAYS_BTW_EPOC_AND_20220110 =
6969
Days.daysBetween(JODA_DATETIME_EPOC, JODA_DATETIME_20220110).getDays();
7070
private static final int HOUR_8_IN_MILLI = (int) TimeUnit.HOURS.toMillis(8);
71+
private static final int MICROS_OF_MILLI_20220110 = 456;
7172

7273
private static final LocalDate JAVA_LOCAL_DATE_20220110 = LocalDate.of(2022, 1, 10);
7374
private static final LocalTime JAVA_LOCAL_TIME_HOUR8 = LocalTime.of(8, 0);
@@ -226,8 +227,10 @@ public GenericRowData generateFlinkRowData() {
226227
HOUR_8_IN_MILLI,
227228
// Although Avro logical type for timestamp fields are in micro seconds,
228229
// AvroToRowDataConverters only looks for long value in milliseconds.
229-
TimestampData.fromEpochMillis(JODA_DATETIME_20220110.getMillis()),
230-
TimestampData.fromEpochMillis(JODA_DATETIME_20220110.getMillis()),
230+
TimestampData.fromEpochMillis(
231+
JODA_DATETIME_20220110.getMillis(), MICROS_OF_MILLI_20220110 * 1000),
232+
TimestampData.fromEpochMillis(
233+
JODA_DATETIME_20220110.getMillis(), MICROS_OF_MILLI_20220110 * 1000),
231234
uuidBytes,
232235
binaryBytes,
233236
DecimalData.fromBigDecimal(BIG_DECIMAL_NEGATIVE, 9, 2),
@@ -253,8 +256,12 @@ public org.apache.avro.generic.GenericRecord generateAvroGenericRecord() {
253256
genericRecord.put("time_field", HOUR_8_IN_MILLI);
254257
// Now that AvroToRowDataConverters correctly supports microseconds,
255258
// we must inject correct microsecond scale values into the Avro data.
256-
genericRecord.put("ts_with_zone_field", JODA_DATETIME_20220110.getMillis() * 1000L);
257-
genericRecord.put("ts_without_zone_field", JODA_DATETIME_20220110.getMillis() * 1000L);
259+
genericRecord.put(
260+
"ts_with_zone_field",
261+
JODA_DATETIME_20220110.getMillis() * 1000L + MICROS_OF_MILLI_20220110);
262+
genericRecord.put(
263+
"ts_without_zone_field",
264+
JODA_DATETIME_20220110.getMillis() * 1000L + MICROS_OF_MILLI_20220110);
258265
genericRecord.put("ts_ns_with_zone_field", ICEBERG_MAX_NANOS_EPOCH);
259266
genericRecord.put("ts_ns_without_zone_field", ICEBERG_MAX_NANOS_EPOCH);
260267

0 commit comments

Comments
 (0)