[FLINK-40594][cdc-common] Preserve TIME precision in the pipeline runtime

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/CommonConverter.java

@@ -185,15 +185,17 @@ static DateData convertToDateData(Object obj) {
                 "Cannot convert " + obj + " of type " + obj.getClass() + " to DATE DATA.");
     }
 
-    static TimeData convertToTimeData(Object obj) {
+    static TimeData convertToTimeData(Object obj, int precision) {
+        LocalTime time;
         if (obj instanceof TimeData) {
-            return (TimeData) obj;
-        }
-        if (obj instanceof LocalTime) {
-            return TimeData.fromLocalTime((LocalTime) obj);
-        }
-        throw new RuntimeException(
-                "Cannot convert " + obj + " of type " + obj.getClass() + " to TIME DATA.");
+            time = ((TimeData) obj).toLocalTime();
+        } else if (obj instanceof LocalTime) {
+            time = (LocalTime) obj;
+        } else {
+            throw new RuntimeException(
+                    "Cannot convert " + obj + " of type " + obj.getClass() + " to TIME DATA.");
+        }
+        return TimeData.fromLocalTime(truncateTime(time, precision));
     }
 
     static TimestampData convertToTimestampData(Object obj) {
@@ -342,6 +344,17 @@ static LocalTime convertToLocalTime(Object obj) {
                 "Cannot convert " + obj + " of type " + obj.getClass() + " to LOCAL TIME.");
     }
 
+    static LocalTime truncateTime(LocalTime time, int precision) {
+        if (precision < 0 || precision > 9) {
+            throw new IllegalArgumentException("TIME precision must be between 0 and 9");
+        }
+        int factor = 1;
+        for (int remaining = 9 - precision; remaining > 0; remaining--) {
+            factor *= 10;
+        }
+        return time.withNano(time.getNano() / factor * factor);
+    }
+
     static LocalDateTime convertToLocalDateTime(Object obj) {
         if (obj instanceof LocalDateTime) {
             return (LocalDateTime) obj;

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/InternalObjectConverter.java

@@ -127,7 +127,7 @@ public Function<Object, DateData> visit(DateType dateType) {
 
         @Override
         public Function<Object, TimeData> visit(TimeType timeType) {
-            return CommonConverter::convertToTimeData;
+            return value -> CommonConverter.convertToTimeData(value, timeType.getPrecision());
         }
 
         @Override

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/JavaObjectConverter.java

@@ -17,6 +17,7 @@
 
 package org.apache.flink.cdc.common.converter;
 
+import org.apache.flink.cdc.common.data.TimeData;
 import org.apache.flink.cdc.common.types.ArrayType;
 import org.apache.flink.cdc.common.types.BigIntType;
 import org.apache.flink.cdc.common.types.BinaryType;
@@ -124,7 +125,14 @@ public Function<Object, LocalDate> visit(DateType dateType) {
 
         @Override
         public Function<Object, LocalTime> visit(TimeType timeType) {
-            return CommonConverter::convertToLocalTime;
+            return value -> {
+                LocalTime time = CommonConverter.convertToLocalTime(value);
+                // Preserve the historical pass-through behavior for Java LocalTime values. An
+                // internal TimeData value is normalized to the declared logical precision.
+                return value instanceof TimeData
+                        ? CommonConverter.truncateTime(time, timeType.getPrecision())
+                        : time;
+            };
         }
 
         @Override

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/ArrayData.java

@@ -72,6 +72,16 @@ public interface ArrayData {
     /** Returns the double value at the given position. */
     double getDouble(int pos);
 
+    /**
+     * Returns the time value at the given position using its declared precision.
+     *
+     * <p>The default keeps binary compatibility for implementations that still expose TIME as a
+     * millisecond integer. Precision-aware implementations override it.
+     */
+    default TimeData getTime(int pos, int precision) {
+        return TimeData.fromMillisOfDay(getInt(pos));
+    }
+
     /** Returns the string value at the given position. */
     StringData getString(int pos);
 
@@ -180,9 +190,12 @@ static ElementGetter createElementGetter(DataType elementType) {
                 break;
             case INTEGER:
             case DATE:
-            case TIME_WITHOUT_TIME_ZONE:
                 elementGetter = ArrayData::getInt;
                 break;
+            case TIME_WITHOUT_TIME_ZONE:
+                final int timePrecision = getPrecision(elementType);
+                elementGetter = (array, pos) -> array.getTime(pos, timePrecision);
+                break;
             case BIGINT:
                 elementGetter = ArrayData::getLong;
                 break;

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/GenericArrayData.java

@@ -211,6 +211,16 @@ public double getDouble(int pos) {
         return isPrimitiveArray ? ((double[]) array)[pos] : (double) getObject(pos);
     }
 
+    @Override
+    public TimeData getTime(int pos, int precision) {
+        Object value = getObject(pos);
+        if (value instanceof TimeData) {
+            return (TimeData) value;
+        }
+        // Kept for arrays produced by older callers that used millisecond integers for TIME.
+        return TimeData.fromMillisOfDay((int) value);
+    }
+
     @Override
     public byte[] getBinary(int pos) {
         return (byte[]) getObject(pos);

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/RecordData.java

@@ -61,7 +61,7 @@
  * +--------------------------------+-----------------------------------------+
  * | DATE                           | int (number of days since epoch)        |
  * +--------------------------------+-----------------------------------------+
- * | TIME                           | int (number of milliseconds of the day) |
+ * | TIME                           | {@link TimeData}                        |
  * +--------------------------------+-----------------------------------------+
  * | TIMESTAMP                      | {@link TimestampData}                   |
  * +--------------------------------+-----------------------------------------+
@@ -170,6 +170,16 @@ public interface RecordData {
     /** Returns the Time data at the given position. */
     TimeData getTime(int pos);
 
+    /**
+     * Returns the Time data at the given position using its declared precision.
+     *
+     * <p>The default implementation preserves compatibility with record implementations whose
+     * representation is independent of precision.
+     */
+    default TimeData getTime(int pos, int precision) {
+        return getTime(pos);
+    }
+
     /** Returns the variant value at the given position. */
     Variant getVariant(int pos);
 
@@ -213,7 +223,7 @@ static RecordData.FieldGetter createFieldGetter(DataType fieldType, int fieldPos
                 fieldGetter = record -> record.getDate(fieldPos);
                 break;
             case TIME_WITHOUT_TIME_ZONE:
-                fieldGetter = record -> record.getTime(fieldPos);
+                fieldGetter = record -> record.getTime(fieldPos, getPrecision(fieldType));
                 break;
             case BIGINT:
                 fieldGetter = record -> record.getLong(fieldPos);

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/TimeData.java

@@ -26,31 +26,30 @@
  */
 public class TimeData implements Comparable<TimeData> {
 
-    private static final int SECONDS_TO_MILLIS = 1000;
-    private static final int MILLIS_TO_MICRO = 1000;
-    private static final int MILLIS_TO_NANO = 1_000_000;
+    private static final long SECONDS_TO_NANO = 1_000_000_000L;
+    private static final long MILLIS_TO_NANO = 1_000_000L;
+    private static final long MICRO_TO_NANO = 1_000L;
 
-    private final int millisOfDay;
+    private final long nanoOfDay;
 
-    private TimeData(int millisOfDay) {
-        this.millisOfDay = millisOfDay;
+    private TimeData(long nanoOfDay) {
+        this.nanoOfDay = nanoOfDay;
     }
 
     public static TimeData fromSecondOfDay(int secondOfDay) {
-        return new TimeData(secondOfDay * SECONDS_TO_MILLIS);
+        return new TimeData(secondOfDay * SECONDS_TO_NANO);
     }
 
     public static TimeData fromMillisOfDay(int millisOfDay) {
-        return new TimeData(millisOfDay);
+        return new TimeData(millisOfDay * MILLIS_TO_NANO);
     }
 
     public static TimeData fromMicroOfDay(long microOfDay) {
-        return new TimeData((int) (microOfDay / MILLIS_TO_MICRO));
+        return new TimeData(microOfDay * MICRO_TO_NANO);
     }
 
     public static TimeData fromNanoOfDay(long nanoOfDay) {
-        // millisOfDay should not exceed 86400000, which is safe to fit into INT.
-        return new TimeData((int) (nanoOfDay / MILLIS_TO_NANO));
+        return new TimeData(nanoOfDay);
     }
 
     public static TimeData fromLocalTime(LocalTime localTime) {
@@ -62,11 +61,19 @@ public static TimeData fromIsoLocalTimeString(String timeString) {
     }
 
     public int toMillisOfDay() {
-        return millisOfDay;
+        return (int) (nanoOfDay / MILLIS_TO_NANO);
+    }
+
+    public long toMicroOfDay() {
+        return nanoOfDay / MICRO_TO_NANO;
+    }
+
+    public long toNanoOfDay() {
+        return nanoOfDay;
     }
 
     public LocalTime toLocalTime() {
-        return LocalTime.ofNanoOfDay((long) millisOfDay * MILLIS_TO_NANO);
+        return LocalTime.ofNanoOfDay(nanoOfDay);
     }
 
     public String toString() {
@@ -80,16 +87,16 @@ public final boolean equals(Object o) {
         }
 
         TimeData timeData = (TimeData) o;
-        return millisOfDay == timeData.millisOfDay;
+        return nanoOfDay == timeData.nanoOfDay;
     }
 
     @Override
     public int compareTo(TimeData other) {
-        return Long.compare(millisOfDay, other.millisOfDay);
+        return Long.compare(nanoOfDay, other.nanoOfDay);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(millisOfDay);
+        return Objects.hash(nanoOfDay);
     }
 }

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinaryArrayData.java

@@ -23,6 +23,7 @@
 import org.apache.flink.cdc.common.data.MapData;
 import org.apache.flink.cdc.common.data.RecordData;
 import org.apache.flink.cdc.common.data.StringData;
+import org.apache.flink.cdc.common.data.TimeData;
 import org.apache.flink.cdc.common.data.TimestampData;
 import org.apache.flink.cdc.common.data.ZonedTimestampData;
 import org.apache.flink.cdc.common.types.DataType;
@@ -32,6 +33,7 @@
 
 import java.lang.reflect.Array;
 
+import static org.apache.flink.cdc.common.types.DataTypeChecks.getPrecision;
 import static org.apache.flink.core.memory.MemoryUtils.UNSAFE;
 
 /**
@@ -146,8 +148,9 @@ public static int calculateFixLengthPartSize(DataType type) {
             case INTEGER:
             case FLOAT:
             case DATE:
-            case TIME_WITHOUT_TIME_ZONE:
                 return 4;
+            case TIME_WITHOUT_TIME_ZONE:
+                return getPrecision(type) <= 3 ? 4 : 8;
             default:
                 throw new IllegalArgumentException();
         }
@@ -226,6 +229,21 @@ public int getInt(int pos) {
         return BinarySegmentUtils.getInt(segments, getElementOffset(pos, 4));
     }
 
+    @Override
+    public TimeData getTime(int pos, int precision) {
+        assertIndexIsValid(pos);
+        if (precision <= 3) {
+            return TimeData.fromMillisOfDay(
+                    BinarySegmentUtils.getInt(segments, getElementOffset(pos, 4)));
+        }
+        long encoded = BinarySegmentUtils.getLong(segments, getElementOffset(pos, 8));
+        if (encoded < 0) {
+            return TimeData.fromNanoOfDay(encoded & Long.MAX_VALUE);
+        }
+        throw new IllegalStateException(
+                "High-precision TIME array uses the legacy millisecond binary layout");
+    }
+
     public void setInt(int pos, int value) {
         assertIndexIsValid(pos);
         setNotNullAt(pos);

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinaryRecordData.java

@@ -228,6 +228,21 @@ public DateData getDate(int pos) {
     @Override
     public TimeData getTime(int pos) {
         assertIndexIsValid(pos);
+        return decodeTime(pos);
+    }
+
+    @Override
+    public TimeData getTime(int pos, int precision) {
+        assertIndexIsValid(pos);
+        return decodeTime(pos);
+    }
+
+    private TimeData decodeTime(int pos) {
+        long encoded = getLong(pos);
+        if (encoded < 0) {
+            return TimeData.fromNanoOfDay(encoded & Long.MAX_VALUE);
+        }
+        // Rows written before nanosecond TIME support used the first four bytes of the slot.
         return TimeData.fromMillisOfDay(getInt(pos));
     }
 

flink-cdc-common/src/test/java/org/apache/flink/cdc/common/converter/InternalObjectConverterTest.java

@@ -256,6 +256,12 @@ void testConvertToTime() {
         assertThat(convertToInternal(TimeData.fromNanoOfDay(14419123456789L), DataTypes.TIME(3)))
                 .isInstanceOf(TimeData.class)
                 .hasToString("04:00:19.123");
+        assertThat(convertToInternal(TimeData.fromNanoOfDay(14419123456789L), DataTypes.TIME(6)))
+                .isInstanceOf(TimeData.class)
+                .hasToString("04:00:19.123456");
+        assertThat(convertToInternal(TimeData.fromNanoOfDay(14419123456789L), DataTypes.TIME(9)))
+                .isInstanceOf(TimeData.class)
+                .hasToString("04:00:19.123456789");
         assertThat(convertToInternal(null, DataTypes.TIME())).isNull();
     }
 

flink-cdc-common/src/test/java/org/apache/flink/cdc/common/converter/JavaObjectConverterTest.java

@@ -253,6 +253,12 @@ void testConvertToTime() {
         assertThat(convertToJava(TimeData.fromNanoOfDay(14419123456789L), DataTypes.TIME(3)))
                 .isInstanceOf(LocalTime.class)
                 .hasToString("04:00:19.123");
+        assertThat(convertToJava(TimeData.fromNanoOfDay(14419123456789L), DataTypes.TIME(6)))
+                .isInstanceOf(LocalTime.class)
+                .hasToString("04:00:19.123456");
+        assertThat(convertToJava(TimeData.fromNanoOfDay(14419123456789L), DataTypes.TIME(9)))
+                .isInstanceOf(LocalTime.class)
+                .hasToString("04:00:19.123456789");
         assertThat(convertToJava(null, DataTypes.TIME())).isNull();
     }
 

flink-cdc-common/src/test/java/org/apache/flink/cdc/common/data/TimeDataTest.java

@@ -0,0 +1,53 @@
+/*
+ * 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.flink.cdc.common.data;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalTime;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link TimeData}. */
+class TimeDataTest {
+
+    @Test
+    void preservesNanosecondsAcrossFactoriesAndConversions() {
+        long nanos = 3_723_123_456_789L;
+        TimeData time = TimeData.fromNanoOfDay(nanos);
+
+        assertThat(time.toNanoOfDay()).isEqualTo(nanos);
+        assertThat(time.toMicroOfDay()).isEqualTo(3_723_123_456L);
+        assertThat(time.toMillisOfDay()).isEqualTo(3_723_123);
+        assertThat(time.toLocalTime().toNanoOfDay()).isEqualTo(nanos);
+        assertThat(TimeData.fromMicroOfDay(3_723_123_456L).toNanoOfDay())
+                .isEqualTo(3_723_123_456_000L);
+        assertThat(TimeData.fromLocalTime(LocalTime.of(1, 2, 3, 123_456_789)).toNanoOfDay())
+                .isEqualTo(nanos);
+    }
+
+    @Test
+    void comparisonAndEqualityIncludeSubMillisecondPrecision() {
+        TimeData lower = TimeData.fromNanoOfDay(1_000_000_001L);
+        TimeData higher = TimeData.fromNanoOfDay(1_000_000_999L);
+
+        assertThat(lower).isNotEqualTo(higher);
+        assertThat(lower.compareTo(higher)).isNegative();
+        assertThat(lower.toMillisOfDay()).isEqualTo(higher.toMillisOfDay());
+    }
+}