[common] Fix negative timestamp decoding in Arrow and Iceberg

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));
         }
     }
 

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));
         }
     }
 

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);

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<InternalRow> 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));
+            }
+        }
+    }
 }

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 =