From a4fe97d3e9ef9ccf76faf504ae3bf87ae0df95a0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=96=B9=E6=99=93=E5=85=B5?= Date: Thu, 10 Sep 2026 20:38:46 +0800 Subject: [PATCH] [common] Fix negative timestamp decoding in Arrow and Iceberg - Reuse microsecond timestamp factories and normalize Arrow nanoseconds. - Cover pre-epoch, zero, and positive Arrow and Iceberg key round trips. --- .../ArrowTimestampLtzColumnVector.java | 5 +- .../ArrowTimestampNtzColumnVector.java | 5 +- .../row/decode/iceberg/IcebergKeyDecoder.java | 7 +- .../row/arrow/ArrowReaderWriterTest.java | 66 +++++++++++++++++++ .../decode/iceberg/IcebergKeyDecoderTest.java | 6 +- 5 files changed, 75 insertions(+), 14 deletions(-) diff --git a/fluss-common/src/main/java/org/apache/fluss/row/arrow/vectors/ArrowTimestampLtzColumnVector.java b/fluss-common/src/main/java/org/apache/fluss/row/arrow/vectors/ArrowTimestampLtzColumnVector.java index 9b870cdf1ff..711aa6e6794 100644 --- a/fluss-common/src/main/java/org/apache/fluss/row/arrow/vectors/ArrowTimestampLtzColumnVector.java +++ b/fluss-common/src/main/java/org/apache/fluss/row/arrow/vectors/ArrowTimestampLtzColumnVector.java @@ -57,10 +57,11 @@ public TimestampLtz getTimestampLtz(int i, int precision) { return TimestampLtz.fromEpochMillis(((TimeStampMilliVector) valueVector).get(i)); } else if (valueVector instanceof TimeStampMicroVector) { long micros = ((TimeStampMicroVector) valueVector).get(i); - return TimestampLtz.fromEpochMillis(micros / 1000, (int) (micros % 1000) * 1000); + return TimestampLtz.fromEpochMicros(micros); } else { long nanos = ((TimeStampNanoVector) valueVector).get(i); - return TimestampLtz.fromEpochMillis(nanos / 1_000_000, (int) (nanos % 1_000_000)); + return TimestampLtz.fromEpochMillis( + Math.floorDiv(nanos, 1_000_000L), (int) Math.floorMod(nanos, 1_000_000L)); } } diff --git a/fluss-common/src/main/java/org/apache/fluss/row/arrow/vectors/ArrowTimestampNtzColumnVector.java b/fluss-common/src/main/java/org/apache/fluss/row/arrow/vectors/ArrowTimestampNtzColumnVector.java index 37bc9f7b9df..60cb6c2b9a4 100644 --- a/fluss-common/src/main/java/org/apache/fluss/row/arrow/vectors/ArrowTimestampNtzColumnVector.java +++ b/fluss-common/src/main/java/org/apache/fluss/row/arrow/vectors/ArrowTimestampNtzColumnVector.java @@ -56,10 +56,11 @@ public TimestampNtz getTimestampNtz(int i, int precision) { return TimestampNtz.fromMillis(((TimeStampMilliVector) valueVector).get(i)); } else if (valueVector instanceof TimeStampMicroVector) { long micros = ((TimeStampMicroVector) valueVector).get(i); - return TimestampNtz.fromMillis(micros / 1000, (int) (micros % 1000) * 1000); + return TimestampNtz.fromMicros(micros); } else { long nanos = ((TimeStampNanoVector) valueVector).get(i); - return TimestampNtz.fromMillis(nanos / 1_000_000, (int) (nanos % 1_000_000)); + return TimestampNtz.fromMillis( + Math.floorDiv(nanos, 1_000_000L), (int) Math.floorMod(nanos, 1_000_000L)); } } diff --git a/fluss-common/src/main/java/org/apache/fluss/row/decode/iceberg/IcebergKeyDecoder.java b/fluss-common/src/main/java/org/apache/fluss/row/decode/iceberg/IcebergKeyDecoder.java index dd746e9866a..5fc2f7257a6 100644 --- a/fluss-common/src/main/java/org/apache/fluss/row/decode/iceberg/IcebergKeyDecoder.java +++ b/fluss-common/src/main/java/org/apache/fluss/row/decode/iceberg/IcebergKeyDecoder.java @@ -92,12 +92,7 @@ private FieldReader createFieldReader(DataType fieldType) { return MemorySegment::getLong; case TIMESTAMP_WITHOUT_TIME_ZONE: - return (segment, offset) -> { - long micros = segment.getLong(offset); - long millis = micros / 1000L; - int nanoOfMillis = (int) ((micros % 1000L) * 1000L); - return TimestampNtz.fromMillis(millis, nanoOfMillis); - }; + return (segment, offset) -> TimestampNtz.fromMicros(segment.getLong(offset)); case DECIMAL: final int decimalPrecision = getPrecision(fieldType); diff --git a/fluss-common/src/test/java/org/apache/fluss/row/arrow/ArrowReaderWriterTest.java b/fluss-common/src/test/java/org/apache/fluss/row/arrow/ArrowReaderWriterTest.java index 3190fa251bf..b3ee59d7d93 100644 --- a/fluss-common/src/test/java/org/apache/fluss/row/arrow/ArrowReaderWriterTest.java +++ b/fluss-common/src/test/java/org/apache/fluss/row/arrow/ArrowReaderWriterTest.java @@ -38,8 +38,11 @@ import org.apache.fluss.utils.ArrowUtils; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; import java.io.IOException; +import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -407,4 +410,67 @@ void testMapWriterWithManyEntries() throws IOException { } } } + + /** + * Tests that timestamp values with a negative micro/nano representation round trip correctly. + */ + @ParameterizedTest + @CsvSource({"false,6", "true,6", "false,9", "true,9"}) + void testNegativeTimestampRoundTrip(boolean ltz, int precision) throws IOException { + // for precision 6, the rows below are -1001us, -1000us, -1us, 0, 1us, 1999us; + // for precision 9, they are -1000001ns, -1000000ns, -1ns, 0, 1ns, 1999999ns. + long[] millis = {-2, -1, -1, 0, 0, 1}; + int[] nanoOfMillis = + precision == 9 + ? new int[] {999_999, 0, 999_999, 0, 1, 999_999} + : new int[] {999_000, 0, 999_000, 0, 1_000, 999_000}; + + DataType timestampType = + ltz ? DataTypes.TIMESTAMP_LTZ(precision) : DataTypes.TIMESTAMP(precision); + RowType rowType = DataTypes.ROW(DataTypes.FIELD("ts", timestampType)); + + List expectedRows = new ArrayList<>(); + try (BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE); + VectorSchemaRoot root = + VectorSchemaRoot.create(ArrowUtils.toArrowSchema(rowType), allocator); + ArrowWriterPool provider = new ArrowWriterPool(allocator); + ArrowWriter writer = + provider.getOrCreateWriter( + 1L, 1, Integer.MAX_VALUE, rowType, NO_COMPRESSION)) { + for (int i = 0; i < millis.length; i++) { + if (ltz) { + expectedRows.add( + GenericRow.of( + TimestampLtz.fromEpochMillis(millis[i], nanoOfMillis[i]))); + } else { + expectedRows.add( + GenericRow.of(TimestampNtz.fromMillis(millis[i], nanoOfMillis[i]))); + } + writer.writeRow(expectedRows.get(i)); + } + + AbstractPagedOutputView pagedOutputView = + new ManagedPagedOutputView(new TestingMemorySegmentPool(10 * 1024)); + + // skip arrow batch header. + int size = + writer.serializeToOutputView( + pagedOutputView, recordBatchHeaderSize(CURRENT_LOG_MAGIC_VALUE)); + int heapMemorySize = Math.max(size, writer.estimatedSizeInBytes()); + MemorySegment segment = MemorySegment.allocateHeapMemory(heapMemorySize); + + assertThat(pagedOutputView.getWrittenSegments().size()).isEqualTo(1); + MemorySegment firstSegment = pagedOutputView.getCurrentSegment(); + firstSegment.copyTo(recordBatchHeaderSize(CURRENT_LOG_MAGIC_VALUE), segment, 0, size); + + ArrowReader reader = + ArrowUtils.createArrowReader(segment, 0, size, root, allocator, rowType); + int rowCount = reader.getRowCount(); + assertThat(rowCount).isEqualTo(expectedRows.size()); + for (int i = 0; i < rowCount; i++) { + ColumnarRow row = reader.read(i); + assertThatRow(row).withSchema(rowType).isEqualTo(expectedRows.get(i)); + } + } + } } diff --git a/fluss-common/src/test/java/org/apache/fluss/row/decode/iceberg/IcebergKeyDecoderTest.java b/fluss-common/src/test/java/org/apache/fluss/row/decode/iceberg/IcebergKeyDecoderTest.java index 49b38437543..71d08c3b83d 100644 --- a/fluss-common/src/test/java/org/apache/fluss/row/decode/iceberg/IcebergKeyDecoderTest.java +++ b/fluss-common/src/test/java/org/apache/fluss/row/decode/iceberg/IcebergKeyDecoderTest.java @@ -126,10 +126,8 @@ void testDecodeTimestamp() { IcebergKeyDecoder decoder = new IcebergKeyDecoder(rowType, Collections.singletonList("ts")); // Iceberg uses microsecond precision, so only test values that are multiples of 1000 nanos - long[] millisValues = {0L, 1000L, 1698235273182L}; - int[] nanosValues = { - 0, 0, 123000, 999000 - }; // Must be multiples of 1000 for microsecond precision + long[] millisValues = {0L, 1000L, 1698235273182L, -1L, -1L, -2L, 1L}; + int[] nanosValues = {0, 0, 123000, 0, 999000, 999000, 999000}; // multiples of 1000 for (int i = 0; i < millisValues.length; i++) { InternalRow original =