From 53b092022f0869b5fe38be666460d1186f764ebc Mon Sep 17 00:00:00 2001 From: Eunbin Son Date: Fri, 3 Jul 2026 17:45:06 +0900 Subject: [PATCH] Flink: Fix micros-to-nanos scaling in AvroToRowDataConverters timestamp convertToTimestamp() multiplied the leftover sub-millisecond microseconds by 1_000_000 instead of 1000 before passing them as nanoOfMillisecond to TimestampData.fromEpochMillis(long, int), whose constructor requires that argument in [0, 999999]. Any non-zero leftover microsecond value overflowed that bound and threw IllegalArgumentException on read. Matches the sibling conversions in StructRowData.getTimestamp() and FlinkValueReaders.TimestampMicrosReader.read(), which already scale by 1000. Generated-by: Claude Code --- .../formats/avro/AvroToRowDataConverters.java | 2 +- .../avro/TestAvroToRowDataConverters.java | 71 +++++++++++++++++++ 2 files changed, 72 insertions(+), 1 deletion(-) create mode 100644 flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/formats/avro/TestAvroToRowDataConverters.java 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/formats/avro/TestAvroToRowDataConverters.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/formats/avro/TestAvroToRowDataConverters.java new file mode 100644 index 000000000000..4770960d3e9e --- /dev/null +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/formats/avro/TestAvroToRowDataConverters.java @@ -0,0 +1,71 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.flink.formats.avro; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.apache.avro.Schema; +import org.apache.avro.SchemaBuilder; +import org.apache.avro.generic.GenericData; +import org.apache.avro.generic.GenericRecord; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.types.logical.TimestampType; +import org.junit.jupiter.api.Test; + +public class TestAvroToRowDataConverters { + + private static final long MICROS_PER_MILLI = 1000L; + + // Avro carries timestamps as a plain long; the converter derives precision from the RowType. + private static final Schema RECORD_SCHEMA = + SchemaBuilder.record("test").fields().name("ts").type().longType().noDefault().endRecord(); + + private static RowData convertMicros(long micros) { + RowType rowType = RowType.of(new TimestampType(6)); + GenericRecord record = new GenericData.Record(RECORD_SCHEMA); + record.put(0, micros); + return (RowData) AvroToRowDataConverters.createRowConverter(rowType).convert(record); + } + + @Test + public void microsWithSubMillisecondComponent() { + // micros value with a 123-microsecond leftover that is not aligned to a whole millisecond. + long micros = 1_577_966_400_000_000L + 123L; + + RowData row = convertMicros(micros); + + long expectedMillis = Math.floorDiv(micros, MICROS_PER_MILLI); + int expectedNanos = (int) Math.floorMod(micros, MICROS_PER_MILLI) * 1000; + assertThat(row.getTimestamp(0, 6)) + .isEqualTo(TimestampData.fromEpochMillis(expectedMillis, expectedNanos)); + } + + @Test + public void microsAlignedToMillisecond() { + // micros value aligned to a whole millisecond (no sub-millisecond component). + long micros = 1_577_966_400_000_000L; + + RowData row = convertMicros(micros); + + assertThat(row.getTimestamp(0, 6)) + .isEqualTo(TimestampData.fromEpochMillis(Math.floorDiv(micros, MICROS_PER_MILLI))); + } +}